Overlay V2: fix flooding issues that occasionally cause instability

overlay/src/flood/inv_messages.rs

@@ -6,7 +6,7 @@ use std::io;
 use std::sync::Arc;
 use stellar_xdr::curr::{
     FloodAdvert, FloodDemand, Hash, Limits, ReadXdr, StellarMessage, TxAdvertVector,
-    TxDemandVector, WriteXdr,
+    TxDemandVector, WriteXdr, TX_DEMAND_VECTOR_MAX_SIZE,
 };
 
 use crate::wire::ValidatedTx;
@@ -77,17 +77,25 @@ impl GetData {
         self.hashes.push(hash);
     }
 
-    /// Encode as a `StellarMessage::FloodDemand` XDR.
-    pub fn encode(&self) -> io::Result<Vec<u8>> {
-        let hashes = self
-            .hashes
-            .iter()
-            .map(|hash| Hash(*hash))
-            .collect::<Vec<_>>();
-        let tx_hashes = TxDemandVector::try_from(hashes).map_err(to_invalid_data)?;
-        StellarMessage::FloodDemand(FloodDemand { tx_hashes })
-            .to_xdr(Limits::none())
-            .map_err(to_invalid_data)
+    /// Encode as one or more `StellarMessage::FloodDemand` XDR messages,
+    /// splitting the hashes so no message exceeds the `TxDemandVector` XDR
+    /// bound (`TX_DEMAND_VECTOR_MAX_SIZE`).
+    ///
+    /// Each chunk is returned with the hashes it carries, so callers can
+    /// track per-hash delivery (e.g. stamp request timestamps only after the
+    /// chunk actually went out on the wire).
+    pub fn encode_chunked(&self) -> io::Result<Vec<(Vec<u8>, Vec<[u8; 32]>)>> {
+        self.hashes
+            .chunks(TX_DEMAND_VECTOR_MAX_SIZE as usize)
+            .map(|chunk| {
+                let hashes = chunk.iter().map(|hash| Hash(*hash)).collect::<Vec<_>>();
+                let tx_hashes = TxDemandVector::try_from(hashes).map_err(to_invalid_data)?;
+                let encoded = StellarMessage::FloodDemand(FloodDemand { tx_hashes })
+                    .to_xdr(Limits::none())
+                    .map_err(to_invalid_data)?;
+                Ok((encoded, chunk.to_vec()))
+            })
+            .collect()
     }
 }
 
@@ -190,15 +198,51 @@ mod tests {
     fn test_tx_stream_message_getdata() {
         let mut gd = GetData::new();
         gd.push([0xFF; 32]);
-        let encoded = gd.encode().unwrap();
-        let decoded = TxStreamMessage::decode(&encoded).unwrap();
+        let chunks = gd.encode_chunked().unwrap();
+        assert_eq!(chunks.len(), 1);
+        let (encoded, chunk_hashes) = &chunks[0];
+        assert_eq!(chunk_hashes, &gd.hashes);
+        let decoded = TxStreamMessage::decode(encoded).unwrap();
         if let TxStreamMessage::GetData(decoded_gd) = decoded {
             assert_eq!(gd, decoded_gd);
         } else {
             panic!("Expected GetData");
         }
     }
 
+    #[test]
+    fn test_getdata_encode_chunked_splits_at_xdr_bound() {
+        let max = TX_DEMAND_VECTOR_MAX_SIZE as usize;
+        let mut gd = GetData::new();
+        for i in 0..(max * 2 + 5) {
+            let mut hash = [0u8; 32];
+            hash[..8].copy_from_slice(&(i as u64).to_be_bytes());
+            gd.push(hash);
+        }
+
+        // Chunked encode must split it into decodable messages that
+        // round-trip every hash in order, and report the hashes carried by
+        // each chunk so callers can track per-chunk delivery.
+        let chunks = gd.encode_chunked().unwrap();
+        assert_eq!(chunks.len(), 3);
+        let mut decoded_hashes = Vec::new();
+        let mut reported_hashes = Vec::new();
+        for (encoded, chunk_hashes) in &chunks {
+            match TxStreamMessage::decode(encoded).unwrap() {
+                TxStreamMessage::GetData(decoded) => {
+                    assert!(decoded.hashes.len() <= max);
+                    // The reported hashes must match the encoded content.
+                    assert_eq!(&decoded.hashes, chunk_hashes);
+                    decoded_hashes.extend(decoded.hashes);
+                }
+                _ => panic!("Expected GetData"),
+            }
+            reported_hashes.extend(chunk_hashes.iter().copied());
+        }
+        assert_eq!(decoded_hashes, gd.hashes);
+        assert_eq!(reported_hashes, gd.hashes);
+    }
+
     #[test]
     fn test_decode_empty_message_fails() {
         let result = TxStreamMessage::decode(&[]);

overlay/src/flood/pending_requests.rs

@@ -49,12 +49,26 @@ impl PendingRequest {
         self.first_sent_at.elapsed() >= total_timeout
     }
 
-    /// Update for retry to a new peer
+    /// Update for retry to a new peer.
+    ///
+    /// Stamps `sent_at` at dispatch time so the housekeeping loop doesn't
+    /// re-dispatch the hash while the send is still in flight; callers must
+    /// call [`PendingRequest::mark_sent`] once the demand actually reaches
+    /// the wire so the peer's response window doesn't include local queueing
+    /// delay.
     pub fn retry(&mut self, new_peer: PeerId) {
         self.peer = new_peer;
         self.sent_at = Instant::now();
         self.attempts += 1;
     }
+
+    /// Restart the per-peer timeout clock: the demand was actually written to
+    /// the peer's stream, so the peer's response window starts now (not at
+    /// dispatch, which may precede the write by encoding/lock/backpressure
+    /// delays).
+    pub fn mark_sent(&mut self) {
+        self.sent_at = Instant::now();
+    }
 }
 
 /// Tracks pending GETDATA requests
@@ -174,6 +188,29 @@ mod tests {
         assert_eq!(req.attempts, 2);
     }
 
+    #[test]
+    fn test_mark_sent_restarts_peer_timeout_only() {
+        let peer = PeerId::random();
+        let mut req = PendingRequest::new(peer);
+
+        // Simulate dispatch happening long before the actual write: the
+        // request looks timed out even though the demand never reached the
+        // peer yet.
+        req.sent_at = Instant::now() - Duration::from_secs(2);
+        let first_sent_at = req.first_sent_at;
+        assert!(req.is_timed_out(Duration::from_secs(1)));
+
+        req.mark_sent();
+
+        // The per-peer response window restarts at the actual send...
+        assert!(!req.is_timed_out(Duration::from_secs(1)));
+        // ...but nothing else changes: total-timeout clock, target peer, and
+        // attempt count are untouched.
+        assert_eq!(req.first_sent_at, first_sent_at);
+        assert_eq!(req.peer, peer);
+        assert_eq!(req.attempts, 1);
+    }
+
     #[test]
     fn test_pending_requests_insert_remove() {
         let mut pending = PendingRequests::new();

overlay/src/ipc/mod.rs

@@ -3,5 +3,5 @@
 mod messages;
 mod transport;
 
-pub use messages::{Message, MessageType};
+pub use messages::{Message, MessageCodec, MessageType};
 pub use transport::CoreIpc;

overlay/src/ipc/transport.rs

@@ -604,14 +604,13 @@ mod tests {
 
         let mut core = core_side;
 
-        // Core requests a TX set
+        // Core requests a TX set. Payload: [hash:32][slotSeq:4]
         let tx_set_hash = [0x42; 32];
+        let slot: u32 = 1234;
+        let mut payload = tx_set_hash.to_vec();
+        payload.extend_from_slice(&slot.to_le_bytes());
 
-        MessageCodec::write(
-            &mut core,
-            &Message::new(MessageType::RequestTxSet, tx_set_hash.to_vec()),
-        )
-        .unwrap();
+        MessageCodec::write(&mut core, &Message::new(MessageType::RequestTxSet, payload)).unwrap();
 
         // Overlay receives request
         let mut receiver = ipc.receiver;
@@ -621,7 +620,12 @@ mod tests {
             .unwrap();
 
         assert_eq!(received.msg_type, MessageType::RequestTxSet);
-        assert_eq!(received.payload.len(), 32);
+        assert_eq!(received.payload.len(), 36);
+        assert_eq!(&received.payload[0..32], &tx_set_hash[..]);
+        assert_eq!(
+            u32::from_le_bytes(received.payload[32..36].try_into().unwrap()),
+            slot
+        );
 
         // Overlay responds with TxSetAvailable
         let tx_set_data = vec![1, 2, 3, 4, 5, 6, 7, 8];