Skip to content

Overlay V2: fix flooding issues that occasionally cause instability #5403

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
marta-lokhova wants to merge 3 commits into
base: overlay-v2-Aug3-rebased
Choose a base branch
from
Open

Overlay V2: fix flooding issues that occasionally cause instability #5403

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0cae17f
Chunk GETDATA messages at the TxDemandVector XDR bound
marta-lokhova Aug 3, 2026
b933857
Properly pipe ledger_seq to tx set cache to avoid occasional prematur…
marta-lokhova Aug 4, 2026
c87a7fd
Address review comments
marta-lokhova Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 50 additions & 1 deletion overlay/src/flood/inv_messages.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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>> {

@SirTyson SirTyson Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we get rid of this function? I think its only called in tests and it seems like a footgun given that there's no reason not to call the more robust chunked version.

@marta-lokhova marta-lokhova Aug 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good point, removed

let hashes = self
.hashes
Expand All @@ -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 {
Expand Down Expand Up @@ -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(&[]);
Expand Down
98 changes: 59 additions & 39 deletions overlay/src/libp2p_overlay.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 },
Expand All @@ -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 },
Expand All @@ -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],
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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!(
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -1638,8 +1651,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;
Expand All @@ -1649,10 +1662,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);
}
}
});
}
Expand Down Expand Up @@ -1906,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
Expand All @@ -1918,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
}
};

Expand All @@ -1929,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 {}: {}",
Expand Down Expand Up @@ -2030,27 +2046,31 @@ 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 {}",
hashes.len(),
peer
);
let getdata = GetData { hashes };
let encoded = match getdata.encode() {
Ok(encoded) => encoded,
let chunks = match getdata.encode_chunked() {

@SirTyson SirTyson Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Outside of the lower level unit test, do we have any tests to make sure that multiple messages actually land correctly if we're in the chunking case?

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 {

@SirTyson SirTyson Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think there's a potential issue here with timeouts and timestamps. We update sent_at for every request before doing a lot of the work, like encoding chunks, getting TX stream lock, and actually flushing them. I think we're treating queue delay/pressure on the local node as a peer delay with this timeout.

If we're stalling on message outbound, we can potentially snowball our own issues. For example, if it takes 800 ms to flush these messages from the queue, we only give the peer 200 ms to respond before we consider the peer timed out when really we're the ones being slow.

Also, not a rust expert so idk what's up, but AI flagged this .await in the for loop as suspicious. Per our AI overlords:

Awaiting each write here serializes sends across all peers, and write_framed has no timeout — one backpressured peer stalls retries to every other peer (and the INV flushing at the top of this loop). Since all the retry timestamps were already reset above, hashes for unrelated peers can age past the 1s timeout before their demand is even sent, triggering duplicate retries. Could we send to each peer in its own task (chunks staying sequential per peer), with a write timeout? Sketch:

  for (peer, hashes) in per_peer {
      let state = Arc::clone(&state);
      tokio::spawn(async move {
          // needs encode_chunked to return (bytes, hashes) per chunk
          for (encoded, chunk_hashes) in encode_chunks(&hashes) {
              match timeout(WRITE_TIMEOUT,
                  try_send_to_existing_stream(&state, peer, StreamType::Tx, &encoded)).await
              {
                  Ok(Ok(())) => {
                      // 1s clock starts when the demand hits the wire
                      let mut pending = state.pending_getdata.write().await;
                      for h in &chunk_hashes {
                          if let Some(req) = pending.get_mut(h) {
                              req.mark_sent(); // sent_at = now
                          }
                      }
                  }
                  _ => break, // keep dispatch stamp; retries ~1s later
              }
          }
      });
  }

@marta-lokhova marta-lokhova Aug 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think there's a potential issue here with timeouts and timestamps. We update sent_at for every request before doing a lot of the work, like encoding chunks, getting TX stream lock, and actually flushing them. I think we're treating queue delay/pressure on the local node as a peer delay with this timeout.

that's a good observation, though it isn't any different from what happens in core already tbh. it's not great to snowball like this, and in production this is typically solved with better load shedding/prioritization discipline. Whether we need to solve this in v2 prototype is... questionable i think. this behavior manifests itself when nodes are slow due to high load. in the context of simulations, we probably want to get overloaded and fail the simulation anyways.

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);
}
}
}
}
Expand Down Expand Up @@ -2542,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);
Expand Down Expand Up @@ -3085,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
Expand Down Expand Up @@ -3120,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;
Expand Down Expand Up @@ -3315,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);
Expand Down Expand Up @@ -3373,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
Expand Down Expand Up @@ -3428,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);
Expand Down Expand Up @@ -3881,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
Expand Down Expand Up @@ -3925,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
Expand Down
Loading
Loading