Retransmit messages

src/crdt.rs

@@ -590,33 +590,51 @@ impl Crdt {
     }
     fn run_window_request(
         window: &Window,
+        me: &ReplicatedData,
         from: &ReplicatedData,
         ix: u64,
         blob_recycler: &BlobRecycler,
     ) -> Option<SharedBlob> {
         let pos = (ix as usize) % window.read().unwrap().len();
         if let Some(blob) = &window.read().unwrap()[pos] {
-            let rblob = blob.read().unwrap();
-            let blob_ix = rblob.get_index().expect("run_window_request get_index");
+            let mut wblob = blob.write().unwrap();
+            let blob_ix = wblob.get_index().expect("run_window_request get_index");
             if blob_ix == ix {
+                let num_retransmits = wblob.meta.num_retransmits;
+                wblob.meta.num_retransmits += 1;
+                // Setting the sender id to the requester id
+                // prevents the requester from retransmitting this response
+                // to other peers
+                let mut sender_id = from.id;
+
+                // Allow retransmission of this response if the node
+                // is the leader and the number of repair requests equals
+                // a power of two
+                if me.current_leader_id == me.id
+                    && (num_retransmits == 0 || num_retransmits.is_power_of_two())
+                {
+                    sender_id = me.id
+                }
+
                 let out = blob_recycler.allocate();
+
                 // copy to avoid doing IO inside the lock
                 {
                     let mut outblob = out.write().unwrap();
-                    let sz = rblob.meta.size;
+                    let sz = wblob.meta.size;
                     outblob.meta.size = sz;
-                    outblob.data[..sz].copy_from_slice(&rblob.data[..sz]);
+                    outblob.data[..sz].copy_from_slice(&wblob.data[..sz]);
                     outblob.meta.set_addr(&from.repair_addr);
-                    //TODO, set the sender id to the requester so we dont retransmit
-                    //come up with a cleaner solution for this when sender signatures are checked
-                    outblob.set_id(from.id).expect("blob set_id");
+                    outblob.set_id(sender_id).expect("blob set_id");
                 }
+
                 return Some(out);
             }
         } else {
             assert!(window.read().unwrap()[pos].is_none());
             info!("failed RequestWindowIndex {} {}", ix, from.repair_addr);
         }
+
         None
     }
 
@@ -683,7 +701,7 @@ impl Crdt {
                     me.repair_addr
                 );
                 assert_ne!(from.repair_addr, me.repair_addr);
-                Self::run_window_request(&window, &from, ix, blob_recycler)
+                Self::run_window_request(&window, &me, &from, ix, blob_recycler)
             }
             Err(_) => {
                 warn!("deserialize crdt packet failed");
@@ -1092,18 +1110,55 @@ mod tests {
             "127.0.0.1:1238".parse().unwrap(),
         );
         let recycler = BlobRecycler::default();
-        let rv = Crdt::run_window_request(&window, &me, 0, &recycler);
+        let rv = Crdt::run_window_request(&window, &me, &me, 0, &recycler);
         assert!(rv.is_none());
         let out = recycler.allocate();
         out.write().unwrap().meta.size = 200;
         window.write().unwrap()[0] = Some(out);
-        let rv = Crdt::run_window_request(&window, &me, 0, &recycler);
+        let rv = Crdt::run_window_request(&window, &me, &me, 0, &recycler);
         assert!(rv.is_some());
         let v = rv.unwrap();
         //test we copied the blob
         assert_eq!(v.read().unwrap().meta.size, 200);
         let len = window.read().unwrap().len() as u64;
-        let rv = Crdt::run_window_request(&window, &me, len, &recycler);
+        let rv = Crdt::run_window_request(&window, &me, &me, len, &recycler);
         assert!(rv.is_none());
     }
+
+    /// test window requests respond with the right blob, and do not overrun
+    #[test]
+    fn run_window_request_with_backoff() {
+        let window = default_window();
+
+        let mut me = ReplicatedData::new_leader(&"127.0.0.1:1234".parse().unwrap());
+        me.current_leader_id = me.id;
+
+        let mock_peer = ReplicatedData::new_leader(&"127.0.0.1:1234".parse().unwrap());
+
+        let recycler = BlobRecycler::default();
+
+        // Simulate handling a repair request from mock_peer
+        let rv = Crdt::run_window_request(&window, &me, &mock_peer, 0, &recycler);
+        assert!(rv.is_none());
+        let blob = recycler.allocate();
+        let blob_size = 200;
+        blob.write().unwrap().meta.size = blob_size;
+        window.write().unwrap()[0] = Some(blob);
+
+        let num_requests: u32 = 64;
+        for i in 0..num_requests {
+            let shared_blob =
+                Crdt::run_window_request(&window, &me, &mock_peer, 0, &recycler).unwrap();
+            let blob = shared_blob.read().unwrap();
+            // Test we copied the blob
+            assert_eq!(blob.meta.size, blob_size);
+
+            let id = if i == 0 || i.is_power_of_two() {
+                me.id
+            } else {
+                mock_peer.id
+            };
+            assert_eq!(blob.get_id().unwrap(), id);
+        }
+    }
 }

src/packet.rs

@@ -29,6 +29,7 @@ pub const NUM_BLOBS: usize = (NUM_PACKETS * PACKET_DATA_SIZE) / BLOB_SIZE;
 #[repr(C)]
 pub struct Meta {
     pub size: usize,
+    pub num_retransmits: u64,
     pub addr: [u16; 8],
     pub port: u16,
     pub v6: bool,