update sync protocol and ordering algo:

sync protocol: every 4 seconds the node broadcast last_term msg, once the other nodes received that msg they will compare it to their last term if the last_term is less than self_last_term they will send back a vector which contain msgs with a term greater than last_term

ordering algorithm: for ordering algorithm: on receiving new private message,it will check if the same term exist in the buffers if so it will check the timestamp difference if it's more than 3 minute then it will ignore the received msg

bin/ircd/src/buffers.rs

@@ -12,6 +12,7 @@ use crate::Privmsg;
 pub const SIZE_OF_MSGS_BUFFER: usize = 4095;
 pub const SIZE_OF_MSG_IDSS_BUFFER: usize = 65536;
 pub const LIFETIME_FOR_ORPHAN: i64 = 600;
+pub const TERM_MAX_TIME_DIFFERENCE: i64 = 180;
 
 pub type InvSeenIds = Arc<Mutex<RingBuffer<u64>>>;
 pub type SeenIds = Mutex<RingBuffer<u64>>;
@@ -131,7 +132,16 @@ impl PrivmsgsBuffer {
 
     pub fn push(&mut self, privmsg: &Privmsg) {
         match privmsg.term.cmp(&(self.last_term() + 1)) {
-            Ordering::Equal | Ordering::Less => self.buffer.push(privmsg.clone()),
+            Ordering::Equal => self.buffer.push(privmsg.clone()),
+            Ordering::Less => {
+                if let Some(msg) = self.get_msg_by_term(privmsg.term) {
+                    if (msg.timestamp - privmsg.timestamp) <= TERM_MAX_TIME_DIFFERENCE {
+                        self.buffer.push(privmsg.clone());
+                    }
+                } else {
+                    self.buffer.push(privmsg.clone());
+                }
+            }
             Ordering::Greater => self.orphans.push(Orphan::new(privmsg)),
         }
         self.update();
@@ -141,6 +151,10 @@ impl PrivmsgsBuffer {
         self.buffer.iter()
     }
 
+    pub fn get_msg_by_term(&self, term: u64) -> Option<Privmsg> {
+        self.buffer.iter().find(|p| p.term == term).cloned()
+    }
+
     pub fn len(&self) -> usize {
         self.buffer.len()
     }
@@ -156,6 +170,10 @@ impl PrivmsgsBuffer {
         }
     }
 
+    pub fn fetch_msgs(&self, term: u64) -> Vec<Privmsg> {
+        self.buffer.iter().take_while(|p| p.term >= term).cloned().collect()
+    }
+
     fn update(&mut self) {
         self.sort_orphans();
         self.update_orphans();
@@ -190,10 +208,20 @@ impl PrivmsgsBuffer {
             }
 
             match privmsg.term.cmp(&(self.last_term() + 1)) {
-                Ordering::Equal | Ordering::Less => {
+                Ordering::Equal => {
                     self.buffer.push(privmsg.clone());
                     self.orphans.remove(orphan);
                 }
+                Ordering::Less => {
+                    if let Some(msg) = self.get_msg_by_term(privmsg.term) {
+                        if (msg.timestamp - privmsg.timestamp) <= TERM_MAX_TIME_DIFFERENCE {
+                            self.buffer.push(privmsg.clone());
+                        }
+                    } else {
+                        self.buffer.push(privmsg.clone());
+                    }
+                    self.orphans.remove(orphan);
+                }
                 Ordering::Greater => {}
             }
         }

bin/ircd/src/main.rs

@@ -11,6 +11,7 @@ use structopt_toml::StructOptToml;
 
 use darkfi::{
     async_daemonize, net,
+    net::P2pPtr,
     rpc::server::listen_and_serve,
     system::{Subscriber, SubscriberPtr},
     util::{
@@ -18,6 +19,7 @@ use darkfi::{
         expand_path,
         file::save_json_file,
         path::get_config_path,
+        sleep,
     },
     Result,
 };
@@ -34,11 +36,14 @@ use crate::{
     buffers::{create_buffers, Buffers, RingBuffer, SIZE_OF_MSG_IDSS_BUFFER},
     irc::IrcServer,
     privmsg::Privmsg,
-    protocol_privmsg::ProtocolPrivmsg,
+    protocol_privmsg::{LastTerm, ProtocolPrivmsg},
     rpc::JsonRpcInterface,
     settings::{Args, ChannelInfo, CONFIG_FILE, CONFIG_FILE_CONTENTS},
 };
 
+const TIMEOUT_FOR_RESEND: u64 = 240;
+const SEND_LAST_TERM_MSG: u64 = 4;
+
 #[derive(serde::Serialize)]
 struct KeyPair {
     private_key: String,
@@ -51,6 +56,25 @@ impl fmt::Display for KeyPair {
     }
 }
 
+async fn resend_unread_msgs(p2p: P2pPtr, buffers: Buffers) -> Result<()> {
+    loop {
+        sleep(TIMEOUT_FOR_RESEND).await;
+
+        for msg in buffers.unread_msgs.lock().await.msgs.values() {
+            p2p.broadcast(msg.clone()).await?;
+        }
+    }
+}
+
+async fn send_last_term(p2p: P2pPtr, buffers: Buffers) -> Result<()> {
+    loop {
+        sleep(SEND_LAST_TERM_MSG).await;
+
+        let term = buffers.privmsgs.lock().await.last_term();
+        p2p.broadcast(LastTerm { term }).await?;
+    }
+}
+
 struct Ircd {
     notify_clients: SubscriberPtr<Privmsg>,
 }
@@ -129,8 +153,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     //
     // P2p setup
     //
-    let mut net_settings = settings.net.clone();
-    net_settings.app_version = Some(option_env!("CARGO_PKG_VERSION").unwrap_or("").to_string());
+    let net_settings = settings.net.clone();
     let (p2p_send_channel, p2p_recv_channel) = async_channel::unbounded::<Privmsg>();
 
     let p2p = net::P2p::new(net_settings.into()).await;
@@ -157,6 +180,12 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let executor_cloned = executor.clone();
     executor_cloned.spawn(p2p.clone().run(executor.clone())).detach();
 
+    //
+    // Sync tasks
+    //
+    executor.spawn(resend_unread_msgs(p2p.clone(), buffers.clone())).detach();
+    executor.spawn(send_last_term(p2p.clone(), buffers.clone())).detach();
+
     //
     // RPC interface
     //

bin/ircd/src/protocol_privmsg.rs

@@ -1,4 +1,5 @@
 use async_std::sync::Arc;
+use std::cmp::Ordering;
 
 use async_executor::Executor;
 use async_trait::async_trait;
@@ -8,10 +9,7 @@ use rand::{rngs::OsRng, RngCore};
 
 use darkfi::{
     net,
-    util::{
-        serial::{SerialDecodable, SerialEncodable},
-        sleep,
-    },
+    util::serial::{SerialDecodable, SerialEncodable},
     Result,
 };
 
@@ -21,8 +19,7 @@ use crate::{
 };
 
 const MAX_CONFIRM: u8 = 4;
-const SLEEP_TIME_FOR_RESEND: u64 = 1200;
-const UNREAD_MSG_EXPIRE_TIME: i64 = 259200;
+const UNREAD_MSG_EXPIRE_TIME: i64 = 18000;
 
 #[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
 struct Inv {
@@ -30,6 +27,11 @@ struct Inv {
     invs: Vec<InvObject>,
 }
 
+#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
+pub struct LastTerm {
+    pub term: u64,
+}
+
 impl Inv {
     fn new(invs: Vec<InvObject>) -> Self {
         let id = OsRng.next_u64();
@@ -57,6 +59,7 @@ pub struct ProtocolPrivmsg {
     msg_sub: net::MessageSubscription<Privmsg>,
     inv_sub: net::MessageSubscription<Inv>,
     getdata_sub: net::MessageSubscription<GetData>,
+    last_term_sub: net::MessageSubscription<LastTerm>,
     p2p: net::P2pPtr,
     channel: net::ChannelPtr,
     inv_ids: InvSeenIds,
@@ -79,16 +82,23 @@ impl ProtocolPrivmsg {
         let msg_sub =
             channel.clone().subscribe_msg::<Privmsg>().await.expect("Missing Privmsg dispatcher!");
 
+        let inv_sub = channel.subscribe_msg::<Inv>().await.expect("Missing Inv dispatcher!");
+
         let getdata_sub =
             channel.clone().subscribe_msg::<GetData>().await.expect("Missing GetData dispatcher!");
 
-        let inv_sub = channel.subscribe_msg::<Inv>().await.expect("Missing Inv dispatcher!");
+        let last_term_sub = channel
+            .clone()
+            .subscribe_msg::<LastTerm>()
+            .await
+            .expect("Missing LastTerm dispatcher!");
 
         Arc::new(Self {
             notify,
             msg_sub,
             inv_sub,
             getdata_sub,
+            last_term_sub,
             jobsman: net::ProtocolJobsManager::new("ProtocolPrivmsg", channel.clone()),
             p2p,
             channel,
@@ -158,6 +168,28 @@ impl ProtocolPrivmsg {
         }
     }
 
+    async fn handle_receive_last_term(self: Arc<Self>) -> Result<()> {
+        debug!(target: "ircd", "ProtocolPrivmsg::handle_receive_last_term() [START]");
+        loop {
+            let last_term = self.last_term_sub.receive().await?;
+            let last_term = last_term.term;
+
+            self.update_unread_msgs().await?;
+
+            let privmsgs = self.buffers.privmsgs.lock().await;
+            let self_last_term = privmsgs.last_term();
+
+            match self_last_term.cmp(&last_term) {
+                Ordering::Less => {
+                    for msg in privmsgs.fetch_msgs(last_term) {
+                        self.channel.send(msg).await?;
+                    }
+                }
+                Ordering::Greater | Ordering::Equal => continue,
+            }
+        }
+    }
+
     async fn handle_receive_getdata(self: Arc<Self>) -> Result<()> {
         debug!(target: "ircd", "ProtocolPrivmsg::handle_receive_getdata() [START]");
         loop {
@@ -197,17 +229,6 @@ impl ProtocolPrivmsg {
         self.notify.send(msg.clone()).await?;
         Ok(())
     }
-
-    async fn resend_loop(self: Arc<Self>) -> Result<()> {
-        sleep(SLEEP_TIME_FOR_RESEND).await;
-
-        self.update_unread_msgs().await?;
-
-        for msg in self.buffers.unread_msgs.lock().await.msgs.values() {
-            self.channel.send(msg.clone()).await?;
-        }
-        Ok(())
-    }
 }
 
 #[async_trait]
@@ -228,7 +249,7 @@ impl net::ProtocolBase for ProtocolPrivmsg {
         self.jobsman.clone().spawn(self.clone().handle_receive_msg(), executor.clone()).await;
         self.jobsman.clone().spawn(self.clone().handle_receive_inv(), executor.clone()).await;
         self.jobsman.clone().spawn(self.clone().handle_receive_getdata(), executor.clone()).await;
-        self.jobsman.clone().spawn(self.clone().resend_loop(), executor.clone()).await;
+        self.jobsman.clone().spawn(self.clone().handle_receive_last_term(), executor.clone()).await;
         debug!(target: "ircd", "ProtocolPrivmsg::start() [END]");
         Ok(())
     }
@@ -255,3 +276,9 @@ impl net::Message for GetData {
         "getdata"
     }
 }
+
+impl net::Message for LastTerm {
+    fn name() -> &'static str {
+        "last_term"
+    }
+}