From a7cc9ffa7de30717f00eb8cf5f21e002a8d8f9af Mon Sep 17 00:00:00 2001
From: marta-lokhova <marta@stellar.org>
Date: Mon, 3 Aug 2026 15:19:35 -0700
Subject: [PATCH 1/6] Chunk GETDATA messages at the TxDemandVector XDR bound

The GETDATA retry path batched every timed-out hash for a peer into a
single FloodDemand. TxDemandVector caps at TX_DEMAND_VECTOR_MAX_SIZE
(1000) hashes, so under fetch storms the encode failed with 'xdr value
max length exceeded' and the entire retry batch was silently dropped,
feeding the 30s give-up spiral seen in the 15-node stress test.

Add GetData::encode_chunked() which splits hashes into XDR-legal
messages, and use it on both the initial and retry send paths.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 overlay/src/flood/inv_messages.rs    | 51 +++++++++++++++++++++++++++-
 overlay/src/libp2p_overlay.rs        | 34 +++++++++++--------
 src/overlay/test/OverlayIPCTests.cpp | 11 +++---
 3 files changed, 77 insertions(+), 19 deletions(-)

diff --git a/overlay/src/flood/inv_messages.rs b/overlay/src/flood/inv_messages.rs
index 80db3c8541..77371e4b1c 100644
--- a/overlay/src/flood/inv_messages.rs
+++ b/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;
@@ -78,6 +78,9 @@ impl GetData {
     }
 
     /// Encode as a `StellarMessage::FloodDemand` XDR.
+    ///
+    /// Fails if there are more than `TX_DEMAND_VECTOR_MAX_SIZE` hashes; use
+    /// [`GetData::encode_chunked`] when the hash count is unbounded.
     pub fn encode(&self) -> io::Result<Vec<u8>> {
         let hashes = self
             .hashes
@@ -89,6 +92,22 @@ impl GetData {
             .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`).
+    pub fn encode_chunked(&self) -> io::Result<Vec<Vec<u8>>> {
+        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)?;
+                StellarMessage::FloodDemand(FloodDemand { tx_hashes })
+                    .to_xdr(Limits::none())
+                    .map_err(to_invalid_data)
+            })
+            .collect()
+    }
 }
 
 impl Default for GetData {
@@ -199,6 +218,36 @@ mod tests {
         }
     }
 
+    #[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);
+        }
+
+        // Single-message encode must reject an oversized demand vector.
+        assert!(gd.encode().is_err());
+
+        // Chunked encode must split it into decodable messages that
+        // round-trip every hash in order.
+        let chunks = gd.encode_chunked().unwrap();
+        assert_eq!(chunks.len(), 3);
+        let mut decoded_hashes = Vec::new();
+        for chunk in &chunks {
+            match TxStreamMessage::decode(chunk).unwrap() {
+                TxStreamMessage::GetData(decoded) => {
+                    assert!(decoded.hashes.len() <= max);
+                    decoded_hashes.extend(decoded.hashes);
+                }
+                _ => panic!("Expected GetData"),
+            }
+        }
+        assert_eq!(decoded_hashes, gd.hashes);
+    }
+
     #[test]
     fn test_decode_empty_message_fails() {
         let result = TxStreamMessage::decode(&[]);
diff --git a/overlay/src/libp2p_overlay.rs b/overlay/src/libp2p_overlay.rs
index dd679a98a4..c43d9aa9cf 100644
--- a/overlay/src/libp2p_overlay.rs
+++ b/overlay/src/libp2p_overlay.rs
@@ -1638,8 +1638,8 @@ async fn handle_inv_batch(state: &Arc<SharedState>, peer_id: &PeerId, batch: Inv
         for hash in to_request {
             getdata.push(hash);
         }
-        let encoded = match getdata.encode() {
-            Ok(encoded) => encoded,
+        let encoded_chunks = match getdata.encode_chunked() {
+            Ok(chunks) => chunks,
             Err(e) => {
                 warn!("Failed to encode GETDATA for {}: {}", peer_id, e);
                 return;
@@ -1649,10 +1649,12 @@ async fn handle_inv_batch(state: &Arc<SharedState>, peer_id: &PeerId, batch: Inv
         let state_clone = Arc::clone(state);
         let peer_clone = *peer_id;
         tokio::spawn(async move {
-            if let Err(e) =
-                send_to_peer_stream(&state_clone, peer_clone, StreamType::Tx, &encoded).await
-            {
-                warn!("Failed to send GETDATA to {}: {}", peer_clone, e);
+            for encoded in encoded_chunks {
+                if let Err(e) =
+                    send_to_peer_stream(&state_clone, peer_clone, StreamType::Tx, &encoded).await
+                {
+                    warn!("Failed to send GETDATA to {}: {}", peer_clone, e);
+                }
             }
         });
     }
@@ -2030,7 +2032,9 @@ async fn inv_getdata_housekeeping_task(state: Arc<SharedState>) {
                 }
             }
 
-            // Send one batched GETDATA per peer
+            // Send batched GETDATA per peer, chunked to the XDR demand-vector
+            // bound (a retry round can accumulate far more than one message's
+            // worth of hashes)
             for (peer, hashes) in per_peer {
                 debug!(
                     "GETDATA_RETRY: Retrying {} TXs to peer {}",
@@ -2038,19 +2042,21 @@ async fn inv_getdata_housekeeping_task(state: Arc<SharedState>) {
                     peer
                 );
                 let getdata = GetData { hashes };
-                let encoded = match getdata.encode() {
-                    Ok(encoded) => encoded,
+                let chunks = match getdata.encode_chunked() {
+                    Ok(chunks) => chunks,
                     Err(e) => {
                         warn!("Failed to encode GETDATA retry to {}: {}", peer, e);
                         continue;
                     }
                 };
 
-                if let Err(e) =
-                    try_send_to_existing_stream(&state, peer.clone(), StreamType::Tx, &encoded)
-                        .await
-                {
-                    warn!("Failed to send GETDATA retry to {}: {:?}", peer, e);
+                for encoded in chunks {
+                    if let Err(e) =
+                        try_send_to_existing_stream(&state, peer.clone(), StreamType::Tx, &encoded)
+                            .await
+                    {
+                        warn!("Failed to send GETDATA retry to {}: {:?}", peer, e);
+                    }
                 }
             }
         }
diff --git a/src/overlay/test/OverlayIPCTests.cpp b/src/overlay/test/OverlayIPCTests.cpp
index 573e64ae8a..41b22ca347 100644
--- a/src/overlay/test/OverlayIPCTests.cpp
+++ b/src/overlay/test/OverlayIPCTests.cpp
@@ -1238,9 +1238,12 @@ TEST_CASE("Rust overlay 15-node 2000 TPS stress test", "[overlay-ipc-large]")
             }
         }
 
-        // High throughput configuration
-        cfg.GENESIS_TEST_ACCOUNT_COUNT = 30000;
-        cfg.TESTING_UPGRADE_MAX_TX_SET_SIZE = 15000;
+        // High throughput configuration. One genesis account per transaction
+        // so every tx has a unique source: the mempool is fee-ordered and
+        // seqnum-oblivious, so multiple txs chained on one account can be
+        // sampled out of order and trimmed as invalid at nomination.
+        cfg.GENESIS_TEST_ACCOUNT_COUNT = totalTxs;
+        cfg.TESTING_UPGRADE_MAX_TX_SET_SIZE = 30000;
 
         auto node = simulation->addNode(keys[i], qSet, &cfg);
         nodes.push_back(node);
@@ -1308,7 +1311,7 @@ TEST_CASE("Rust overlay 15-node 2000 TPS stress test", "[overlay-ipc-large]")
                        .NewMeter({"loadgen", "run", "complete"}, "run")
                        .count() == 1;
         },
-        500 * simulation->getExpectedLedgerCloseTime(), false);
+        20 * simulation->getExpectedLedgerCloseTime(), false);
 
     auto endTime = std::chrono::steady_clock::now();
     auto durationMs = std::chrono::duration_cast<std::chrono::milliseconds>(

From cf19b2a5780ea1d923529cd58cecc677bf619ade Mon Sep 17 00:00:00 2001
From: marta-lokhova <marta@stellar.org>
Date: Tue, 4 Aug 2026 12:10:24 -0700
Subject: [PATCH 2/6] Properly pipe ledger_seq to tx set cache to avoid
 occasional premature eviction

---
 overlay/src/libp2p_overlay.rs        | 64 +++++++++++++++++-----------
 overlay/src/main.rs                  | 37 +++++++++-------
 src/herder/HerderImpl.cpp            |  3 +-
 src/herder/PendingEnvelopes.cpp      |  5 ++-
 src/overlay/OverlayIPC.cpp           | 25 +++++++----
 src/overlay/OverlayIPC.h             |  7 ++-
 src/overlay/RustOverlayManager.cpp   |  9 ++--
 src/overlay/RustOverlayManager.h     | 13 +++---
 src/overlay/test/OverlayIPCTests.cpp |  2 +-
 9 files changed, 103 insertions(+), 62 deletions(-)

diff --git a/overlay/src/libp2p_overlay.rs b/overlay/src/libp2p_overlay.rs
index c43d9aa9cf..1df476e7e8 100644
--- a/overlay/src/libp2p_overlay.rs
+++ b/overlay/src/libp2p_overlay.rs
@@ -57,6 +57,9 @@ pub enum OverlayEvent {
         envelope: Vec<u8>,
         txset_hashes: Vec<[u8; 32]>,
         from: PeerId,
+        /// Slot the envelope is for (statement.slot_index), used to stamp
+        /// tx sets fetched on behalf of this envelope.
+        slot: u32,
     },
     /// Received TX from peer
     TxReceived { tx: Arc<ValidatedTx>, from: PeerId },
@@ -65,6 +68,9 @@ pub enum OverlayEvent {
         hash: [u8; 32],
         data: Vec<u8>,
         from: PeerId,
+        /// Slot the set was requested for; None if the response was
+        /// unsolicited (no pending request recorded).
+        slot: Option<u32>,
     },
     /// Peer is requesting a TX set (need to look up and respond)
     TxSetRequested { hash: [u8; 32], from: PeerId },
@@ -84,7 +90,7 @@ pub enum OverlayCommand {
     /// Broadcast a validated TX to all peers
     BroadcastTx(Arc<ValidatedTx>),
     /// Request TX set from a peer (picks best peer)
-    FetchTxSet { hash: [u8; 32] },
+    FetchTxSet { hash: [u8; 32], slot: u32 },
     /// Send TX set to a specific peer (response to their request)
     SendTxSet {
         hash: [u8; 32],
@@ -182,8 +188,12 @@ impl OverlayHandle {
         }
     }
 
-    pub async fn fetch_txset(&self, hash: [u8; 32]) {
-        if let Err(e) = self.cmd_tx.send(OverlayCommand::FetchTxSet { hash }).await {
+    pub async fn fetch_txset(&self, hash: [u8; 32], slot: u32) {
+        if let Err(e) = self
+            .cmd_tx
+            .send(OverlayCommand::FetchTxSet { hash, slot })
+            .await
+        {
             warn!(
                 "Overlay command channel closed, failed to send FetchTxSet: {}",
                 e
@@ -303,7 +313,8 @@ struct SharedState {
     /// TX set sources: which peer has which TX set (learned from SCP messages)
     txset_sources: RwLock<lru::LruCache<[u8; 32], PeerId>>,
     /// Pending TX set requests: hash -> (peer, request_time) to avoid duplicate fetches and track latency
-    pending_txset_requests: RwLock<HashMap<[u8; 32], (PeerId, Instant)>>,
+    /// hash -> (peer asked, request time, slot the set is for)
+    pending_txset_requests: RwLock<HashMap<[u8; 32], (PeerId, Instant, u32)>>,
     /// Event sender for non-TX events (SCP, TxSet - critical path, unbounded)
     event_tx: mpsc::UnboundedSender<OverlayEvent>,
     /// Bounded TX event sender (backpressure - drops allowed)
@@ -517,8 +528,8 @@ impl StellarOverlay {
                         OverlayCommand::BroadcastTx(tx) => {
                             self.broadcast_tx(tx).await;
                         }
-                        OverlayCommand::FetchTxSet { hash } => {
-                            self.fetch_txset(hash).await;
+                        OverlayCommand::FetchTxSet { hash, slot } => {
+                            self.fetch_txset(hash, slot).await;
                         }
                         OverlayCommand::SendTxSet { hash, data, to } => {
                             self.send_txset_response(to, hash, data).await;
@@ -680,7 +691,7 @@ impl StellarOverlay {
                     {
                         let mut pending = self.state.pending_txset_requests.write().await;
                         let before_len = pending.len();
-                        pending.retain(|_hash, (p, _)| p != &peer_id);
+                        pending.retain(|_hash, (p, _, _)| p != &peer_id);
                         let removed = before_len - pending.len();
                         if removed > 0 {
                             info!(
@@ -886,11 +897,11 @@ impl StellarOverlay {
     }
 
     /// Fetch TX set from a peer - preferring the peer who sent us the SCP message referencing it
-    async fn fetch_txset(&mut self, hash: [u8; 32]) {
+    async fn fetch_txset(&mut self, hash: [u8; 32], slot: u32) {
         // Check if we're already fetching this TxSet from a connected peer (dedup)
         {
             let pending = self.state.pending_txset_requests.read().await;
-            if let Some((pending_peer, _)) = pending.get(&hash) {
+            if let Some((pending_peer, _, _)) = pending.get(&hash) {
                 // Check if that peer is still connected
                 let streams = self.state.peer_streams.read().await;
                 if streams.contains_key(pending_peer) {
@@ -964,7 +975,7 @@ impl StellarOverlay {
             .pending_txset_requests
             .write()
             .await
-            .insert(hash, (peer.clone(), Instant::now()));
+            .insert(hash, (peer.clone(), Instant::now(), slot));
 
         let request = crate::xdr::frame_get_tx_set(hash);
 
@@ -1494,10 +1505,12 @@ async fn handle_inbound_scp_streams(mut incoming: IncomingStreams, state: Arc<Sh
                         // from the single decode above and forward to Core.
                         let txset_hashes =
                             crate::xdr::extract_txset_hashes_from_envelope(&scp_envelope);
+                        let slot = scp_envelope.statement.slot_index as u32;
                         if let Err(e) = state.event_tx.send(OverlayEvent::ScpReceived {
                             envelope: envelope_bytes.to_vec(),
                             txset_hashes,
                             from: peer_id.clone(),
+                            slot,
                         }) {
                             warn!("Failed to forward SCP event from {}: {}", peer_id, e);
                         }
@@ -1908,9 +1921,9 @@ async fn handle_inbound_txset_streams(mut incoming: IncomingStreams, state: Arc<
                                 let hash = crate::xdr::sha256_hash(&txset_data);
 
                                 // Clear pending request flag and measure fetch latency
-                                let was_pending = {
+                                let slot = {
                                     let mut pending = state.pending_txset_requests.write().await;
-                                    if let Some((_, request_time)) = pending.remove(&hash) {
+                                    if let Some((_, request_time, slot)) = pending.remove(&hash) {
                                         let fetch_us = request_time.elapsed().as_micros() as u64;
                                         state
                                             .metrics
@@ -1920,9 +1933,9 @@ async fn handle_inbound_txset_streams(mut incoming: IncomingStreams, state: Arc<
                                             .metrics
                                             .fetch_txset_count
                                             .fetch_add(1, Ordering::Relaxed);
-                                        true
+                                        Some(slot)
                                     } else {
-                                        false
+                                        None
                                     }
                                 };
 
@@ -1931,12 +1944,13 @@ async fn handle_inbound_txset_streams(mut incoming: IncomingStreams, state: Arc<
                                     &hash[..4],
                                     txset_data.len(),
                                     peer_id,
-                                    was_pending
+                                    slot.is_some()
                                 );
                                 if let Err(e) = state.event_tx.send(OverlayEvent::TxSetReceived {
                                     hash,
                                     data: txset_data,
                                     from: peer_id,
+                                    slot,
                                 }) {
                                     warn!(
                                         "Failed to forward TxSetReceived event from {}: {}",
@@ -2548,7 +2562,7 @@ mod tests {
 
         // Node2 requests a TxSet by hash
         let (requested_hash, txset_data) = test_txset_xdr(0x42);
-        handle2.fetch_txset(requested_hash).await;
+        handle2.fetch_txset(requested_hash, 1).await;
 
         // Node1 should receive TxSetRequested event
         let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
@@ -3091,7 +3105,7 @@ async fn test_txset_source_tracking() {
     tokio::time::sleep(Duration::from_millis(100)).await;
 
     // Now try to fetch - since fake_peer isn't connected, it should fall back
-    handle2.fetch_txset(test_hash).await;
+    handle2.fetch_txset(test_hash, 1).await;
     tokio::time::sleep(Duration::from_millis(100)).await;
 
     // Clean up
@@ -3126,7 +3140,7 @@ async fn test_txset_fetch_flow() {
 
     // overlay2 requests a TX set that overlay1 doesn't have
     let test_hash: [u8; 32] = [0xCD; 32];
-    handle2.fetch_txset(test_hash).await;
+    handle2.fetch_txset(test_hash, 1).await;
 
     // overlay1 should receive the request (as TxSetRequested event)
     tokio::time::sleep(Duration::from_millis(200)).await;
@@ -3321,7 +3335,7 @@ async fn test_txset_request_and_response() {
     // Node2 requests a TX set
     let (requested_hash, txset_data) = test_txset_xdr(0x77);
 
-    handle2.fetch_txset(requested_hash).await;
+    handle2.fetch_txset(requested_hash, 1).await;
 
     // Node1 receives request and responds
     let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
@@ -3379,7 +3393,7 @@ async fn test_txset_fetch_no_peers() {
 
     // Request TX set with no peers connected
     let requested_hash: [u8; 32] = [0x88; 32];
-    handle.fetch_txset(requested_hash).await;
+    handle.fetch_txset(requested_hash, 1).await;
 
     // Should not crash or hang - just no response
     // Wait briefly to ensure no panic
@@ -3434,9 +3448,9 @@ async fn test_txset_multiple_concurrent_requests() {
     let hash2: [u8; 32] = [0x22; 32];
     let hash3: [u8; 32] = [0x33; 32];
 
-    handle2.fetch_txset(hash1).await;
-    handle2.fetch_txset(hash2).await;
-    handle2.fetch_txset(hash3).await;
+    handle2.fetch_txset(hash1, 1).await;
+    handle2.fetch_txset(hash2, 1).await;
+    handle2.fetch_txset(hash3, 1).await;
 
     // Node1 should receive all 3 requests
     let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
@@ -3887,7 +3901,7 @@ async fn test_pending_txset_cleanup_on_disconnect() {
 
     // Request TxSet - this tests that pending_txset_requests correctly stores (hash, peer)
     let (txset_hash, txset_data) = test_txset_xdr(0x42);
-    handle1.fetch_txset(txset_hash).await;
+    handle1.fetch_txset(txset_hash, 1).await;
     tokio::time::sleep(Duration::from_millis(100)).await;
 
     // Verify node2 received the request
@@ -3931,7 +3945,7 @@ async fn test_pending_txset_cleanup_on_disconnect() {
     assert!(got_response, "Node1 should receive TxSet response");
 
     // Request the same TxSet again - should NOT be skipped since pending was cleared
-    handle1.fetch_txset(txset_hash).await;
+    handle1.fetch_txset(txset_hash, 1).await;
     tokio::time::sleep(Duration::from_millis(100)).await;
 
     // Verify node2 receives the second request
diff --git a/overlay/src/main.rs b/overlay/src/main.rs
index c29a257134..9c69641857 100644
--- a/overlay/src/main.rs
+++ b/overlay/src/main.rs
@@ -657,6 +657,7 @@ impl App {
                 envelope,
                 txset_hashes,
                 from,
+                slot,
             } => {
                 // Copy first 4 bytes for logging identification
                 let mut id_bytes = [0u8; 4];
@@ -696,7 +697,7 @@ impl App {
                                     &txhash[..4],
                                     from_peer
                                 );
-                                handle.fetch_txset(*txhash).await;
+                                handle.fetch_txset(*txhash, slot).await;
                             }
                         }
                     });
@@ -724,7 +725,12 @@ impl App {
                 );
                 self.overlay_handle.submit_tx(tx);
             }
-            LibP2pOverlayEvent::TxSetReceived { hash, data, from } => {
+            LibP2pOverlayEvent::TxSetReceived {
+                hash,
+                data,
+                from,
+                slot,
+            } => {
                 // `data` was strict-decoded and its content hash verified in the
                 // reader task, so we cache and forward it as-is.
                 info!(
@@ -736,9 +742,11 @@ impl App {
 
                 // IMPORTANT: Cache the TxSet FIRST, before pushing to Core
                 // This ensures the TxSet is available when SCP processing resumes
+                // Stamp with the slot the set was requested for so eviction is
+                // exact; for an unsolicited set fall back to the next slot.
                 cache_tx_set_xdr(
                     &mut self.tx_set_cache,
-                    self.current_ledger_seq,
+                    slot.unwrap_or(self.current_ledger_seq + 1),
                     hash,
                     data.clone(),
                 );
@@ -993,13 +1001,15 @@ impl App {
 
             MessageType::RequestTxSet => {
                 // Request TX set by hash - check local cache first, then fetch from peers via libp2p
-                if msg.payload.len() < 32 {
+                // Payload: [hash:32][slotSeq:4]
+                if msg.payload.len() < 36 {
                     warn!("RequestTxSet payload too short");
                     return true;
                 }
 
                 let mut hash = [0u8; 32];
                 hash.copy_from_slice(&msg.payload[0..32]);
+                let slot = u32::from_le_bytes(msg.payload[32..36].try_into().unwrap());
 
                 // First check local cache
                 if let Some(xdr) = get_cached_tx_set_xdr(&self.tx_set_cache, &hash) {
@@ -1020,22 +1030,23 @@ impl App {
                     );
                     let handle = self.libp2p_handle.clone();
                     tokio::spawn(async move {
-                        handle.fetch_txset(hash).await;
+                        handle.fetch_txset(hash, slot).await;
                     });
                 }
             }
 
             MessageType::CacheTxSet => {
                 // Core built a TX set locally and wants us to cache it for peer requests
-                // Payload: [hash:32][txSetXDR...]
-                if msg.payload.len() < 33 {
+                // Payload: [hash:32][slotSeq:4][txSetXDR...]
+                if msg.payload.len() < 37 {
                     warn!("CacheTxSet payload too short");
                     return true;
                 }
 
                 let mut hash = [0u8; 32];
                 hash.copy_from_slice(&msg.payload[0..32]);
-                let tx_set_xdr = &msg.payload[32..];
+                let slot = u32::from_le_bytes(msg.payload[32..36].try_into().unwrap());
+                let tx_set_xdr = &msg.payload[36..];
 
                 // Core is trusted for encoding, so we skip decoding. We still
                 // guard the content hash cheaply: caching bytes under a hash
@@ -1050,17 +1061,13 @@ impl App {
                 }
 
                 info!(
-                    "TXSET_CACHE: Caching locally-built TX set {:02x?}... ({} bytes)",
+                    "TXSET_CACHE: Caching locally-built TX set {:02x?}... for slot {} ({} bytes)",
                     &hash[..4],
+                    slot,
                     tx_set_xdr.len()
                 );
 
-                cache_tx_set_xdr(
-                    &mut self.tx_set_cache,
-                    self.current_ledger_seq,
-                    hash,
-                    tx_set_xdr.to_vec(),
-                );
+                cache_tx_set_xdr(&mut self.tx_set_cache, slot, hash, tx_set_xdr.to_vec());
             }
 
             MessageType::SubmitTx => {
diff --git a/src/herder/HerderImpl.cpp b/src/herder/HerderImpl.cpp
index 29f8401667..b1c83d745d 100644
--- a/src/herder/HerderImpl.cpp
+++ b/src/herder/HerderImpl.cpp
@@ -1677,7 +1677,8 @@ HerderImpl::triggerNextLedger(uint32_t ledgerSeqToTrigger,
         GeneralizedTransactionSet xdrTxSet;
         proposedSet->toXDR(xdrTxSet);
         auto xdrBytes = xdr::xdr_to_opaque(xdrTxSet);
-        mApp.getOverlayManager().cacheTxSet(txSetHash, xdrBytes);
+        mApp.getOverlayManager().cacheTxSet(txSetHash, xdrBytes,
+                                            lcl.header.ledgerSeq + 1);
     }
 
     lcl = mLedgerManager.getLastClosedLedgerHeader();
diff --git a/src/herder/PendingEnvelopes.cpp b/src/herder/PendingEnvelopes.cpp
index 74e8dece4d..c662bd55ce 100644
--- a/src/herder/PendingEnvelopes.cpp
+++ b/src/herder/PendingEnvelopes.cpp
@@ -683,7 +683,10 @@ PendingEnvelopes::startFetch(SCPEnvelope const& envelope)
             auto& vec = mPendingTxSetFetches[h2];
             vec.push_back(envelope);
             mTxSetFetchStartTimes.emplace(h2, mApp.getClock().now());
-            mApp.getOverlayManager().requestTxSet(h2); // Only once!
+            mApp.getOverlayManager().requestTxSet(
+                h2,
+                static_cast<uint32_t>(
+                    envelope.statement.slotIndex)); // Only once!
         }
     }
 
diff --git a/src/overlay/OverlayIPC.cpp b/src/overlay/OverlayIPC.cpp
index a3b6331112..9547d4a0ea 100644
--- a/src/overlay/OverlayIPC.cpp
+++ b/src/overlay/OverlayIPC.cpp
@@ -624,7 +624,7 @@ OverlayIPC::submitTransaction(TransactionEnvelope const& tx, int64_t fee,
 }
 
 void
-OverlayIPC::requestTxSet(Hash const& hash)
+OverlayIPC::requestTxSet(Hash const& hash, uint32_t slotIndex)
 {
     if (!mChannel || !mChannel->isConnected())
     {
@@ -633,16 +633,21 @@ OverlayIPC::requestTxSet(Hash const& hash)
 
     IPCMessage msg;
     msg.type = IPCMessageType::REQUEST_TX_SET;
-    msg.payload.resize(32);
+    // Payload: [hash:32][slotSeq:4]. The slot the set is for; used by the
+    // Rust overlay to stamp the cache entry for exact age-based eviction.
+    msg.payload.resize(36);
     std::memcpy(msg.payload.data(), hash.data(), 32);
+    std::memcpy(msg.payload.data() + 32, &slotIndex, 4);
 
-    CLOG_DEBUG(Overlay, "Requesting TX set {}", hexAbbrev(hash));
+    CLOG_DEBUG(Overlay, "Requesting TX set {} for slot {}", hexAbbrev(hash),
+               slotIndex);
     std::lock_guard<std::mutex> lock(mSendMutex);
     mChannel->send(msg);
 }
 
 void
-OverlayIPC::cacheTxSet(Hash const& hash, std::vector<uint8_t> const& xdr)
+OverlayIPC::cacheTxSet(Hash const& hash, std::vector<uint8_t> const& xdr,
+                       uint32_t slotIndex)
 {
     if (!mChannel || !mChannel->isConnected())
     {
@@ -651,12 +656,16 @@ OverlayIPC::cacheTxSet(Hash const& hash, std::vector<uint8_t> const& xdr)
 
     IPCMessage msg;
     msg.type = IPCMessageType::CACHE_TX_SET;
-    msg.payload.resize(32 + xdr.size());
+    // Payload: [hash:32][slotSeq:4][txSetXDR...]. The slot the set is for;
+    // used by the Rust overlay to stamp the cache entry for exact age-based
+    // eviction.
+    msg.payload.resize(36 + xdr.size());
     std::memcpy(msg.payload.data(), hash.data(), 32);
-    std::memcpy(msg.payload.data() + 32, xdr.data(), xdr.size());
+    std::memcpy(msg.payload.data() + 32, &slotIndex, 4);
+    std::memcpy(msg.payload.data() + 36, xdr.data(), xdr.size());
 
-    CLOG_DEBUG(Overlay, "Caching TX set {} ({} bytes)", hexAbbrev(hash),
-               xdr.size());
+    CLOG_DEBUG(Overlay, "Caching TX set {} for slot {} ({} bytes)",
+               hexAbbrev(hash), slotIndex, xdr.size());
     std::lock_guard<std::mutex> lock(mSendMutex);
     mChannel->send(msg);
 }
diff --git a/src/overlay/OverlayIPC.h b/src/overlay/OverlayIPC.h
index 5f58e6360c..62c433cb3a 100644
--- a/src/overlay/OverlayIPC.h
+++ b/src/overlay/OverlayIPC.h
@@ -151,8 +151,9 @@ class OverlayIPC
      * TxSetReceivedCallback when available.
      *
      * @param hash The TX set hash to request
+     * @param slotIndex The slot the set is for (stamps the cache entry)
      */
-    void requestTxSet(Hash const& hash);
+    void requestTxSet(Hash const& hash, uint32_t slotIndex);
 
     /**
      * Cache a locally-built TX set in the Rust overlay.
@@ -163,8 +164,10 @@ class OverlayIPC
      *
      * @param hash The TX set hash
      * @param xdr The serialized TX set XDR
+     * @param slotIndex The slot the set is for (stamps the cache entry)
      */
-    void cacheTxSet(Hash const& hash, std::vector<uint8_t> const& xdr);
+    void cacheTxSet(Hash const& hash, std::vector<uint8_t> const& xdr,
+                    uint32_t slotIndex);
 
     /// Set callback for received SCP envelopes
     void setOnSCPReceived(SCPReceivedCallback cb);
diff --git a/src/overlay/RustOverlayManager.cpp b/src/overlay/RustOverlayManager.cpp
index 39b5e3e664..94cde991fb 100644
--- a/src/overlay/RustOverlayManager.cpp
+++ b/src/overlay/RustOverlayManager.cpp
@@ -166,21 +166,22 @@ RustOverlayManager::notifyTxSetExternalized(Hash const& txSetHash,
 }
 
 void
-RustOverlayManager::requestTxSet(Hash const& txSetHash)
+RustOverlayManager::requestTxSet(Hash const& txSetHash, uint32_t slotIndex)
 {
     if (mOverlayIPC && !mShuttingDown)
     {
-        mOverlayIPC->requestTxSet(txSetHash);
+        mOverlayIPC->requestTxSet(txSetHash, slotIndex);
     }
 }
 
 void
 RustOverlayManager::cacheTxSet(Hash const& txSetHash,
-                               std::vector<uint8_t> const& xdr)
+                               std::vector<uint8_t> const& xdr,
+                               uint32_t slotIndex)
 {
     if (mOverlayIPC && !mShuttingDown)
     {
-        mOverlayIPC->cacheTxSet(txSetHash, xdr);
+        mOverlayIPC->cacheTxSet(txSetHash, xdr, slotIndex);
     }
 }
 
diff --git a/src/overlay/RustOverlayManager.h b/src/overlay/RustOverlayManager.h
index 647a27e8be..252789cde9 100644
--- a/src/overlay/RustOverlayManager.h
+++ b/src/overlay/RustOverlayManager.h
@@ -50,11 +50,14 @@ class RustOverlayManager
     void notifyTxSetExternalized(Hash const& txSetHash,
                                  std::vector<Hash> const& txHashes);
 
-    // Request TX set from peers (via Rust overlay, async)
-    void requestTxSet(Hash const& txSetHash);
-
-    // Cache a locally-built TX set in Rust overlay
-    void cacheTxSet(Hash const& txSetHash, std::vector<uint8_t> const& xdr);
+    // Request TX set from peers (via Rust overlay, async). slotIndex is the
+    // slot the set is for, used to stamp the Rust-side cache entry.
+    void requestTxSet(Hash const& txSetHash, uint32_t slotIndex);
+
+    // Cache a locally-built TX set in Rust overlay. slotIndex is the slot the
+    // set is for, used to stamp the Rust-side cache entry.
+    void cacheTxSet(Hash const& txSetHash, std::vector<uint8_t> const& xdr,
+                    uint32_t slotIndex);
 
     // Get top transactions from Rust overlay's mempool for TX set building.
     // Blocks until the overlay responds, shuts down, or disconnects.
diff --git a/src/overlay/test/OverlayIPCTests.cpp b/src/overlay/test/OverlayIPCTests.cpp
index 41b22ca347..c22a0c5676 100644
--- a/src/overlay/test/OverlayIPCTests.cpp
+++ b/src/overlay/test/OverlayIPCTests.cpp
@@ -1311,7 +1311,7 @@ TEST_CASE("Rust overlay 15-node 2000 TPS stress test", "[overlay-ipc-large]")
                        .NewMeter({"loadgen", "run", "complete"}, "run")
                        .count() == 1;
         },
-        20 * simulation->getExpectedLedgerCloseTime(), false);
+        50 * simulation->getExpectedLedgerCloseTime(), false);
 
     auto endTime = std::chrono::steady_clock::now();
     auto durationMs = std::chrono::duration_cast<std::chrono::milliseconds>(

From ee9e0c280466afcd26be362745c50c752b10995a Mon Sep 17 00:00:00 2001
From: marta-lokhova <marta@stellar.org>
Date: Fri, 7 Aug 2026 12:55:27 -0700
Subject: [PATCH 3/6] Address review comments

---
 overlay/src/flood/inv_messages.rs     |  49 +++----
 overlay/src/flood/pending_requests.rs |  39 ++++-
 overlay/src/ipc/mod.rs                |   2 +-
 overlay/src/ipc/transport.rs          |  18 ++-
 overlay/src/libp2p_overlay.rs         | 203 ++++++++++++++++++++++++--
 overlay/src/main.rs                   | 190 ++++++++++++++++++++++++
 overlay/src/xdr.rs                    |   2 +-
 7 files changed, 450 insertions(+), 53 deletions(-)

diff --git a/overlay/src/flood/inv_messages.rs b/overlay/src/flood/inv_messages.rs
index 77371e4b1c..197dac956f 100644
--- a/overlay/src/flood/inv_messages.rs
+++ b/overlay/src/flood/inv_messages.rs
@@ -77,34 +77,23 @@ impl GetData {
         self.hashes.push(hash);
     }
 
-    /// Encode as a `StellarMessage::FloodDemand` XDR.
-    ///
-    /// Fails if there are more than `TX_DEMAND_VECTOR_MAX_SIZE` hashes; use
-    /// [`GetData::encode_chunked`] when the hash count is unbounded.
-    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`).
-    pub fn encode_chunked(&self) -> io::Result<Vec<Vec<u8>>> {
+    ///
+    /// 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)?;
-                StellarMessage::FloodDemand(FloodDemand { tx_hashes })
+                let encoded = StellarMessage::FloodDemand(FloodDemand { tx_hashes })
                     .to_xdr(Limits::none())
-                    .map_err(to_invalid_data)
+                    .map_err(to_invalid_data)?;
+                Ok((encoded, chunk.to_vec()))
             })
             .collect()
     }
@@ -209,8 +198,11 @@ 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 {
@@ -228,24 +220,27 @@ mod tests {
             gd.push(hash);
         }
 
-        // Single-message encode must reject an oversized demand vector.
-        assert!(gd.encode().is_err());
-
         // Chunked encode must split it into decodable messages that
-        // round-trip every hash in order.
+        // 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();
-        for chunk in &chunks {
-            match TxStreamMessage::decode(chunk).unwrap() {
+        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]
diff --git a/overlay/src/flood/pending_requests.rs b/overlay/src/flood/pending_requests.rs
index e96bc3f4b1..481bd67a93 100644
--- a/overlay/src/flood/pending_requests.rs
+++ b/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();
diff --git a/overlay/src/ipc/mod.rs b/overlay/src/ipc/mod.rs
index 12d9267d4a..f4023d456b 100644
--- a/overlay/src/ipc/mod.rs
+++ b/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;
diff --git a/overlay/src/ipc/transport.rs b/overlay/src/ipc/transport.rs
index c12203c37b..dd68896eb6 100644
--- a/overlay/src/ipc/transport.rs
+++ b/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];
diff --git a/overlay/src/libp2p_overlay.rs b/overlay/src/libp2p_overlay.rs
index 1df476e7e8..65fa57e105 100644
--- a/overlay/src/libp2p_overlay.rs
+++ b/overlay/src/libp2p_overlay.rs
@@ -90,7 +90,10 @@ pub enum OverlayCommand {
     /// Broadcast a validated TX to all peers
     BroadcastTx(Arc<ValidatedTx>),
     /// Request TX set from a peer (picks best peer)
-    FetchTxSet { hash: [u8; 32], slot: u32 },
+    FetchTxSet {
+        hash: [u8; 32],
+        slot: u32,
+    },
     /// Send TX set to a specific peer (response to their request)
     SendTxSet {
         hash: [u8; 32],
@@ -98,15 +101,26 @@ pub enum OverlayCommand {
         to: PeerId,
     },
     /// Record that a peer has a specific TX set (learned from SCP message)
-    RecordTxSetSource { hash: [u8; 32], peer: PeerId },
+    RecordTxSetSource {
+        hash: [u8; 32],
+        peer: PeerId,
+    },
     /// Connect to a peer by address (bootstrap — PeerId unknown)
     Dial(Multiaddr),
     /// Connect to a known peer by PeerId (reconnect — deduplicates automatically)
-    DialPeer { peer_id: PeerId, addr: Multiaddr },
+    DialPeer {
+        peer_id: PeerId,
+        addr: Multiaddr,
+    },
     /// Request SCP state from all peers
-    RequestScpState { ledger_seq: u32 },
+    RequestScpState {
+        ledger_seq: u32,
+    },
     /// Send SCP envelope to a specific peer
-    SendScpToPeer { peer_id: PeerId, envelope: Vec<u8> },
+    SendScpToPeer {
+        peer_id: PeerId,
+        envelope: Vec<u8>,
+    },
     /// Shutdown
     Shutdown,
     /// Query the number of connected peers (responds via oneshot)
@@ -291,6 +305,7 @@ impl OverlayHandle {
         rx.await.unwrap_or(0)
     }
 
+
     /// Ping the event loop and wait for response - for testing responsiveness
     #[cfg(test)]
     pub async fn ping(&self) -> Result<(), tokio::sync::oneshot::error::RecvError> {
@@ -1204,6 +1219,21 @@ async fn try_send_to_existing_stream(
     write_framed(stream, data).await
 }
 
+/// Restart the per-peer GETDATA timeout clock for hashes whose demand chunk
+/// was just written to `peer`'s stream. Skips hashes that completed or were
+/// re-assigned to a different peer while the write was in flight —
+/// re-stamping those would silently delay their next retry.
+async fn mark_getdata_sent(state: &SharedState, peer: &PeerId, hashes: &[[u8; 32]]) {
+    let mut pending = state.pending_getdata.write().await;
+    for hash in hashes {
+        if let Some(req) = pending.get_mut(hash) {
+            if req.peer == *peer {
+                req.mark_sent();
+            }
+        }
+    }
+}
+
 /// Send message to a specific peer's stream, reopening if needed
 async fn send_to_peer_stream(
     state: &SharedState,
@@ -1662,11 +1692,16 @@ async fn handle_inv_batch(state: &Arc<SharedState>, peer_id: &PeerId, batch: Inv
         let state_clone = Arc::clone(state);
         let peer_clone = *peer_id;
         tokio::spawn(async move {
-            for encoded in encoded_chunks {
+            for (encoded, chunk_hashes) in encoded_chunks {
                 if let Err(e) =
                     send_to_peer_stream(&state_clone, peer_clone, StreamType::Tx, &encoded).await
                 {
                     warn!("Failed to send GETDATA to {}: {}", peer_clone, e);
+                } else {
+                    // The pending entries were stamped at insert time above;
+                    // restart the peer's response window now that the demand
+                    // is actually on the wire.
+                    mark_getdata_sent(&state_clone, &peer_clone, &chunk_hashes).await;
                 }
             }
         });
@@ -2048,7 +2083,12 @@ async fn inv_getdata_housekeeping_task(state: Arc<SharedState>) {
 
             // Send batched GETDATA per peer, chunked to the XDR demand-vector
             // bound (a retry round can accumulate far more than one message's
-            // worth of hashes)
+            // worth of hashes). Each peer gets its own task so one slow or
+            // backpressured peer can't stall retries to the others or delay
+            // the next housekeeping tick. `retry()` above stamped dispatch
+            // time (which keeps subsequent ticks from re-dispatching these
+            // hashes while the send is in flight); the peer's 1s response
+            // window only starts once its chunk is actually on the wire.
             for (peer, hashes) in per_peer {
                 debug!(
                     "GETDATA_RETRY: Retrying {} TXs to peer {}",
@@ -2064,14 +2104,30 @@ async fn inv_getdata_housekeeping_task(state: Arc<SharedState>) {
                     }
                 };
 
-                for encoded in chunks {
-                    if let Err(e) =
-                        try_send_to_existing_stream(&state, peer.clone(), StreamType::Tx, &encoded)
-                            .await
-                    {
-                        warn!("Failed to send GETDATA retry to {}: {:?}", peer, e);
+                let task_state = Arc::clone(&state);
+                tokio::spawn(async move {
+                    for (encoded, chunk_hashes) in chunks {
+                        match try_send_to_existing_stream(
+                            &task_state,
+                            peer,
+                            StreamType::Tx,
+                            &encoded,
+                        )
+                        .await
+                        {
+                            Ok(()) => {
+                                mark_getdata_sent(&task_state, &peer, &chunk_hashes).await;
+                            }
+                            Err(e) => {
+                                // Keep the dispatch stamp: the next
+                                // housekeeping round retries these hashes
+                                // (to the next peer) about a second from now.
+                                warn!("Failed to send GETDATA retry to {}: {:?}", peer, e);
+                                break;
+                            }
+                        }
                     }
-                }
+                });
             }
         }
     }
@@ -3494,11 +3550,13 @@ async fn test_scp_state_request_on_connection() {
     let (handle2, mut events2, _tx_events2, overlay2) =
         create_overlay(keypair2, Arc::new(OverlayMetrics::new())).unwrap();
 
-    let listen_port = 19801;
+    // NB: unique per test — 19801/19802 are already taken by a test above,
+    // and clashing listen ports makes parallel runs flaky.
+    let listen_port = 19901;
     tokio::spawn(async move { overlay1.run("127.0.0.1", listen_port).await });
     tokio::time::sleep(Duration::from_millis(100)).await;
 
-    tokio::spawn(async move { overlay2.run("127.0.0.1", 19802).await });
+    tokio::spawn(async move { overlay2.run("127.0.0.1", 19902).await });
     tokio::time::sleep(Duration::from_millis(100)).await;
 
     // Connect node2 to node1
@@ -4050,6 +4108,119 @@ async fn test_inv_getdata_tx_propagation() {
     let _ = tokio::time::timeout(Duration::from_secs(1), overlay2_task).await;
 }
 
+/// Regression test for the GETDATA fetch-storm bug: a retry round with more
+/// hashes than `TX_DEMAND_VECTOR_MAX_SIZE` must be split into multiple
+/// FloodDemand messages that all land and get processed by the peer.
+///
+/// Before chunking, the oversized demand vector failed XDR encoding
+/// ("xdr value max length exceeded") and the entire retry batch was silently
+/// dropped, feeding the 30s give-up spiral. With this test, that failure mode
+/// shows up as node1 processing zero demanded hashes.
+#[tokio::test]
+async fn test_getdata_retry_chunked_across_multiple_messages() {
+    use stellar_xdr::curr::TX_DEMAND_VECTOR_MAX_SIZE;
+
+    let keypair1 = Keypair::generate_ed25519();
+    let keypair2 = Keypair::generate_ed25519();
+
+    let metrics1 = Arc::new(OverlayMetrics::new());
+    let (handle1, _events1, _tx_events1, overlay1) =
+        create_overlay(keypair1.clone(), Arc::clone(&metrics1)).unwrap();
+    let (handle2, _events2, _tx_events2, overlay2) =
+        create_overlay(keypair2, Arc::new(OverlayMetrics::new())).unwrap();
+
+    let peer1_id = PeerId::from_public_key(&keypair1.public());
+    // Reach into node2's shared state to stage the retry round directly.
+    let state2 = Arc::clone(&overlay2.state);
+
+    // NB: keep unique across all tests in this crate — clashing listen ports
+    // make fully-parallel runs flaky.
+    let listen_port = 24101;
+    let overlay1_task = tokio::spawn(async move {
+        overlay1.run("127.0.0.1", listen_port).await;
+    });
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let overlay2_task = tokio::spawn(async move {
+        overlay2.run("127.0.0.1", listen_port + 1).await;
+    });
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let addr: Multiaddr = format!(
+        "/ip4/127.0.0.1/udp/{}/quic-v1/p2p/{}",
+        listen_port, peer1_id
+    )
+    .parse()
+    .unwrap();
+    handle2.dial(addr).await;
+
+    // The retry path only writes to an already-open TX stream. Broadcast TXs
+    // from node2 until its outbound TX stream to node1 is open (the INV send
+    // opens it on demand). Polling instead of fixed sleeps keeps this robust
+    // when the test suite runs fully parallel and dials are slow.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    let mut seq = 1i64;
+    loop {
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "Node2's TX stream to node1 never opened"
+        );
+        handle2.broadcast_tx(test_tx(seq)).await;
+        seq += 1;
+        tokio::time::sleep(Duration::from_millis(200)).await;
+
+        let streams = state2.peer_streams.read().await;
+        if let Some(peer_streams) = streams.get(&peer1_id) {
+            if peer_streams.tx.lock().await.is_some() {
+                break;
+            }
+        }
+    }
+
+    // Stage more pending GETDATA requests than fit in one FloodDemand, all
+    // already timed out, with node1 as their (only) source. The housekeeping
+    // task must retry them to node1 as multiple chunks.
+    let n = TX_DEMAND_VECTOR_MAX_SIZE as usize * 2 + 100;
+    {
+        let mut tracker = state2.inv_tracker.write().await;
+        let mut pending = state2.pending_getdata.write().await;
+        for i in 0..n {
+            let mut hash = [0u8; 32];
+            hash[..8].copy_from_slice(&(i as u64).to_be_bytes());
+            tracker.record_source(hash, peer1_id);
+            pending.insert(hash, peer1_id);
+            pending.get_mut(&hash).unwrap().sent_at = Instant::now() - Duration::from_secs(2);
+        }
+    }
+
+    // Node1 has none of these TXs, so every demanded hash it decodes and
+    // processes increments flood_unfulfilled_unknown. All n hashes arriving
+    // proves every chunk landed and was parsed.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    let mut processed = 0;
+    while tokio::time::Instant::now() < deadline {
+        processed = metrics1.flood_unfulfilled_unknown.load(Ordering::Relaxed) as usize;
+        if processed >= n {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    // >= rather than ==: the staged requests never complete (node1 doesn't
+    // have the TXs), so node2 legitimately starts another retry round after
+    // 1s, which may re-deliver some hashes before we sample the counter.
+    assert!(
+        processed >= n,
+        "Node1 should process all {} demanded hashes across multiple GETDATA chunks, got {}",
+        n,
+        processed
+    );
+
+    handle1.shutdown().await;
+    handle2.shutdown().await;
+    let _ = tokio::time::timeout(Duration::from_secs(1), overlay1_task).await;
+    let _ = tokio::time::timeout(Duration::from_secs(1), overlay2_task).await;
+}
+
 /// Test INV/GETDATA protocol: TX relay through 3 nodes (A→B→C)
 #[tokio::test]
 async fn test_inv_getdata_three_node_relay() {
diff --git a/overlay/src/main.rs b/overlay/src/main.rs
index 9c69641857..b3fecd0e8f 100644
--- a/overlay/src/main.rs
+++ b/overlay/src/main.rs
@@ -1891,4 +1891,194 @@ mod tests {
         let stripped = strip_p2p_suffix(&bare);
         assert_eq!(stripped, bare);
     }
+
+    // --- App::handle_core_message / handle_libp2p_event tests ---
+
+    use std::os::unix::net::UnixStream as StdUnixStream;
+    use stellar_overlay::ipc::MessageCodec;
+
+    /// Build an App wired to an in-process socket pair, without touching the
+    /// network. Returns the core-side stream for driving and observing IPC.
+    /// The libp2p overlay object is dropped (not run), so cache-miss fetches
+    /// just log a warning — these tests only exercise the cache paths.
+    fn test_app() -> (App, StdUnixStream) {
+        let (overlay_side, core_side) = StdUnixStream::pair().unwrap();
+        let core_ipc = CoreIpc::from_stream(overlay_side).unwrap();
+
+        let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
+        let mempool_manager = Overlay::new(cmd_rx);
+        tokio::spawn(async move {
+            let _ = mempool_manager.run().await;
+        });
+        let overlay_handle = OverlayHandle::new(cmd_tx);
+
+        let metrics = Arc::new(OverlayMetrics::new());
+        let (libp2p_handle, libp2p_events, tx_events, _overlay) =
+            create_overlay(Libp2pKeypair::generate_ed25519(), Arc::clone(&metrics)).unwrap();
+
+        let app = App {
+            core_ipc,
+            overlay_handle,
+            tx_set_cache: TxSetCache::new(100),
+            current_ledger_seq: 0,
+            libp2p_handle,
+            libp2p_events,
+            tx_events,
+            pending_scp_state_requests: Arc::new(RwLock::new(HashMap::new())),
+            next_scp_request_id: Arc::new(AtomicU64::new(1)),
+            local_addrs: Arc::new(RwLock::new(HashSet::new())),
+            configured_peers: Arc::new(RwLock::new(ConfiguredPeers {
+                addrs: Vec::new(),
+                listen_port: 11625,
+                resolved: HashMap::new(),
+            })),
+            known_peers: Arc::new(RwLock::new(HashMap::new())),
+            peer_hostnames: Arc::new(RwLock::new(HashMap::new())),
+            metrics,
+        };
+        (app, core_side)
+    }
+
+    /// A minimal valid GeneralizedTransactionSet whose content hash matches,
+    /// so it passes the CacheTxSet hash guard.
+    fn test_txset_xdr(seed: u8) -> ([u8; 32], Vec<u8>) {
+        use stellar_xdr::curr::{GeneralizedTransactionSet, Hash};
+
+        let mut tx_set = GeneralizedTransactionSet::default();
+        let GeneralizedTransactionSet::V1(v1) = &mut tx_set;
+        v1.previous_ledger_hash = Hash([seed; 32]);
+        let bytes = tx_set.to_xdr(Limits::none()).unwrap();
+        let hash = xdr::sha256_hash(&bytes);
+        (hash, bytes)
+    }
+
+    fn request_tx_set_payload(hash: &[u8; 32], slot: u32) -> Vec<u8> {
+        let mut payload = hash.to_vec();
+        payload.extend_from_slice(&slot.to_le_bytes());
+        payload
+    }
+
+    fn ledger_closed_payload(seq: u32) -> Vec<u8> {
+        let mut payload = seq.to_le_bytes().to_vec();
+        payload.extend_from_slice(&[0u8; 32]);
+        payload
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_request_tx_set_rejects_legacy_32_byte_payload() {
+        let (mut app, mut core) = test_app();
+        let (hash, xdr_bytes) = test_txset_xdr(7);
+
+        // Cache the set, so if the handler wrongly accepted the legacy
+        // format it would respond with TxSetAvailable below.
+        cache_tx_set_xdr(&mut app.tx_set_cache, 1, hash, xdr_bytes);
+
+        // Pre-slotSeq payload: [hash:32] only. The protocol is now
+        // [hash:32][slotSeq:4]; the short payload must be dropped.
+        let handled = app
+            .handle_core_message(Message::new(MessageType::RequestTxSet, hash.to_vec()))
+            .await;
+        assert!(
+            handled,
+            "short payload should be dropped, not kill the loop"
+        );
+
+        core.set_read_timeout(Some(Duration::from_millis(300)))
+            .unwrap();
+        assert!(
+            MessageCodec::read(&mut core).is_err(),
+            "no response expected for a legacy 32-byte RequestTxSet payload"
+        );
+    }
+
+    /// Regression test for premature tx set eviction: a set Core caches for a
+    /// future slot must be stamped with that slot, not with the overlay's
+    /// current ledger view. Before the fix it was stamped with
+    /// current_ledger_seq (0 here) and evicted on the next LedgerClosed,
+    /// making the set unfetchable exactly when SCP needed it.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_cache_tx_set_slot_stamp_prevents_premature_eviction() {
+        let (mut app, mut core) = test_app();
+        let (hash, xdr_bytes) = test_txset_xdr(9);
+        let slot: u32 = 100;
+
+        // Core caches the set it built for slot 100 while the overlay still
+        // thinks the current ledger is 0.
+        // CacheTxSet payload: [hash:32][slotSeq:4][txSetXDR...]
+        let mut payload = request_tx_set_payload(&hash, slot);
+        payload.extend_from_slice(&xdr_bytes);
+        assert!(
+            app.handle_core_message(Message::new(MessageType::CacheTxSet, payload))
+                .await
+        );
+
+        // Ledger 99 closes; eviction drops sets stamped before slot 87. The
+        // slot-100 entry must survive (pre-fix it was stamped 0 and died).
+        assert!(
+            app.handle_core_message(Message::new(
+                MessageType::LedgerClosed,
+                ledger_closed_payload(99)
+            ))
+            .await
+        );
+
+        // Core asks for the set: it must come back from the local cache.
+        assert!(
+            app.handle_core_message(Message::new(
+                MessageType::RequestTxSet,
+                request_tx_set_payload(&hash, slot)
+            ))
+            .await
+        );
+
+        core.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+        let resp = MessageCodec::read(&mut core).unwrap();
+        assert_eq!(resp.msg_type, MessageType::TxSetAvailable);
+        assert_eq!(&resp.payload[0..32], &hash[..]);
+        assert_eq!(&resp.payload[32..], &xdr_bytes[..]);
+    }
+
+    /// Same property for sets fetched from peers: a TxSetReceived event is
+    /// cached under the slot the set was requested for, so it survives
+    /// eviction until that slot is actually past.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_txset_received_cached_under_requested_slot() {
+        let (mut app, mut core) = test_app();
+        let (hash, xdr_bytes) = test_txset_xdr(11);
+
+        app.handle_libp2p_event(LibP2pOverlayEvent::TxSetReceived {
+            hash,
+            data: xdr_bytes.clone(),
+            from: PeerId::random(),
+            slot: Some(100),
+        })
+        .await;
+
+        // Receiving the set pushes it straight to Core; drain that message.
+        core.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+        let pushed = MessageCodec::read(&mut core).unwrap();
+        assert_eq!(pushed.msg_type, MessageType::TxSetAvailable);
+
+        // Ledger 99 closes (evicts sets stamped before 87); the entry was
+        // stamped with the requested slot 100 and must survive.
+        assert!(
+            app.handle_core_message(Message::new(
+                MessageType::LedgerClosed,
+                ledger_closed_payload(99)
+            ))
+            .await
+        );
+
+        assert!(
+            app.handle_core_message(Message::new(
+                MessageType::RequestTxSet,
+                request_tx_set_payload(&hash, 100)
+            ))
+            .await
+        );
+        let resp = MessageCodec::read(&mut core).unwrap();
+        assert_eq!(resp.msg_type, MessageType::TxSetAvailable);
+        assert_eq!(&resp.payload[0..32], &hash[..]);
+        assert_eq!(&resp.payload[32..], &xdr_bytes[..]);
+    }
 }
diff --git a/overlay/src/xdr.rs b/overlay/src/xdr.rs
index 57127dd296..763a443b52 100644
--- a/overlay/src/xdr.rs
+++ b/overlay/src/xdr.rs
@@ -36,7 +36,7 @@ impl From<xdr::Error> for XdrError {
     }
 }
 
-pub(crate) fn sha256_hash(data: &[u8]) -> [u8; 32] {
+pub fn sha256_hash(data: &[u8]) -> [u8; 32] {
     let mut hasher = Sha256::new();
     hasher.update(data);
     let result = hasher.finalize();

From c3242efeb808bae464e75db584bda97f44da6fe5 Mon Sep 17 00:00:00 2001
From: marta-lokhova <marta@stellar.org>
Date: Wed, 12 Aug 2026 14:07:56 -0700
Subject: [PATCH 4/6] Fix off by one error in peer re-connect logic

---
 overlay/src/main.rs | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/overlay/src/main.rs b/overlay/src/main.rs
index b3fecd0e8f..e4386246ee 100644
--- a/overlay/src/main.rs
+++ b/overlay/src/main.rs
@@ -546,7 +546,11 @@ impl App {
                     let cp = self.configured_peers.read().await;
                     let addrs = cp.addrs.clone();
                     let listen_port = cp.listen_port;
-                    let expected_peers = addrs.len().saturating_sub(1); // exclude self
+                    // Expect a connection for every configured address. Self is
+                    // not normally in the list; if it is, the extra safety-net
+                    // tick is harmless (self-dials and connected peers are
+                    // skipped when re-dialing).
+                    let expected_peers = addrs.len();
                     // Build set of hostnames that have a known PeerId — these
                     // are handled by PeerId-based dials and must NOT be raw-dialed.
                     let hostnames_with_known_peer: HashSet<String> = {

From f25ecc1e45691e72138784d51aed8da4dd9b6e96 Mon Sep 17 00:00:00 2001
From: marta-lokhova <marta@stellar.org>
Date: Wed, 12 Aug 2026 14:14:57 -0700
Subject: [PATCH 5/6] Linter

---
 overlay/src/libp2p_overlay.rs | 25 +++++--------------------
 1 file changed, 5 insertions(+), 20 deletions(-)

diff --git a/overlay/src/libp2p_overlay.rs b/overlay/src/libp2p_overlay.rs
index 65fa57e105..7dd1b0c423 100644
--- a/overlay/src/libp2p_overlay.rs
+++ b/overlay/src/libp2p_overlay.rs
@@ -90,10 +90,7 @@ pub enum OverlayCommand {
     /// Broadcast a validated TX to all peers
     BroadcastTx(Arc<ValidatedTx>),
     /// Request TX set from a peer (picks best peer)
-    FetchTxSet {
-        hash: [u8; 32],
-        slot: u32,
-    },
+    FetchTxSet { hash: [u8; 32], slot: u32 },
     /// Send TX set to a specific peer (response to their request)
     SendTxSet {
         hash: [u8; 32],
@@ -101,26 +98,15 @@ pub enum OverlayCommand {
         to: PeerId,
     },
     /// Record that a peer has a specific TX set (learned from SCP message)
-    RecordTxSetSource {
-        hash: [u8; 32],
-        peer: PeerId,
-    },
+    RecordTxSetSource { hash: [u8; 32], peer: PeerId },
     /// Connect to a peer by address (bootstrap — PeerId unknown)
     Dial(Multiaddr),
     /// Connect to a known peer by PeerId (reconnect — deduplicates automatically)
-    DialPeer {
-        peer_id: PeerId,
-        addr: Multiaddr,
-    },
+    DialPeer { peer_id: PeerId, addr: Multiaddr },
     /// Request SCP state from all peers
-    RequestScpState {
-        ledger_seq: u32,
-    },
+    RequestScpState { ledger_seq: u32 },
     /// Send SCP envelope to a specific peer
-    SendScpToPeer {
-        peer_id: PeerId,
-        envelope: Vec<u8>,
-    },
+    SendScpToPeer { peer_id: PeerId, envelope: Vec<u8> },
     /// Shutdown
     Shutdown,
     /// Query the number of connected peers (responds via oneshot)
@@ -305,7 +291,6 @@ impl OverlayHandle {
         rx.await.unwrap_or(0)
     }
 
-
     /// Ping the event loop and wait for response - for testing responsiveness
     #[cfg(test)]
     pub async fn ping(&self) -> Result<(), tokio::sync::oneshot::error::RecvError> {

From 6b6197d17321efbe8148a5c1d9412146d1fb6b9a Mon Sep 17 00:00:00 2001
From: marta-lokhova <marta@stellar.org>
Date: Wed, 12 Aug 2026 15:53:15 -0700
Subject: [PATCH 6/6] Fix flaky tests

---
 overlay/src/libp2p_overlay.rs | 40 +++++++++++++++++++++++++----------
 1 file changed, 29 insertions(+), 11 deletions(-)

diff --git a/overlay/src/libp2p_overlay.rs b/overlay/src/libp2p_overlay.rs
index 7dd1b0c423..f5e4396e02 100644
--- a/overlay/src/libp2p_overlay.rs
+++ b/overlay/src/libp2p_overlay.rs
@@ -2394,11 +2394,15 @@ mod tests {
             tx_total_time
         );
 
-        // Also verify TX flood took meaningful time (not instant)
+        // Also verify the flood dominated the delivery window. If streams were
+        // blocked, SCP latency would be comparable to the flood duration
+        // (ratio near 1), so requiring a wide margin keeps the test meaningful
+        // without an absolute wall-clock bound that fast CI machines beat.
         assert!(
-            tx_total_time > Duration::from_millis(50),
-            "TX flood should take measurable time ({:?}), otherwise test is invalid",
-            tx_total_time
+            tx_total_time > scp_latency * 5,
+            "TX flood ({:?}) should dwarf SCP latency ({:?}), otherwise test is invalid",
+            tx_total_time,
+            scp_latency
         );
 
         handle1.shutdown().await;
@@ -2781,7 +2785,7 @@ mod tests {
         let keypair_b = Keypair::generate_ed25519();
         let keypair_c = Keypair::generate_ed25519();
 
-        let (handle_a, _events_a, _tx_events_a, overlay_a) =
+        let (handle_a, mut events_a, _tx_events_a, overlay_a) =
             create_overlay(keypair_a, Arc::new(OverlayMetrics::new())).unwrap();
         let (handle_b, mut events_b, _tx_events_b, overlay_b) =
             create_overlay(keypair_b, Arc::new(OverlayMetrics::new())).unwrap();
@@ -2810,8 +2814,22 @@ mod tests {
         handle_b.dial(addr_a.clone()).await;
         handle_c.dial(addr_a).await;
 
-        // Wait for connections to establish
-        tokio::time::sleep(Duration::from_millis(500)).await;
+        // Wait until A has seen both connections: broadcast only reaches peers
+        // A already knows about, so a fixed sleep flakes when the test suite
+        // runs fully parallel and connection setup is slow.
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+        let mut a_connections = 0;
+        while a_connections < 2 {
+            assert!(
+                tokio::time::Instant::now() < deadline,
+                "A should see connections from both B and C (saw {})",
+                a_connections
+            );
+            match tokio::time::timeout(Duration::from_millis(100), events_a.recv()).await {
+                Ok(Some(OverlayEvent::PeerConnected { .. })) => a_connections += 1,
+                _ => {}
+            }
+        }
 
         // Drain connection events
         while events_b.try_recv().is_ok() {}
@@ -3535,13 +3553,13 @@ async fn test_scp_state_request_on_connection() {
     let (handle2, mut events2, _tx_events2, overlay2) =
         create_overlay(keypair2, Arc::new(OverlayMetrics::new())).unwrap();
 
-    // NB: unique per test — 19801/19802 are already taken by a test above,
-    // and clashing listen ports makes parallel runs flaky.
-    let listen_port = 19901;
+    // NB: unique per test — clashing listen ports break parallel runs: the
+    // loser of the bind race isn't reachable and its test times out.
+    let listen_port = 21401;
     tokio::spawn(async move { overlay1.run("127.0.0.1", listen_port).await });
     tokio::time::sleep(Duration::from_millis(100)).await;
 
-    tokio::spawn(async move { overlay2.run("127.0.0.1", 19902).await });
+    tokio::spawn(async move { overlay2.run("127.0.0.1", 21402).await });
     tokio::time::sleep(Duration::from_millis(100)).await;
 
     // Connect node2 to node1