Overlay V2: fix flooding issues that occasionally cause instability

[marta-lokhova]

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.

Also properly attach ledger_seq to cached tx sets to avoid premature eviction in case of overload.

[SirTyson]

I think there's one potential issue with the way we do peer timeouts (plus one concurrency issue that I don't quite understand that the AI flagged, take it or leave it).

I also think there's a couple places where we could have better tests, but given that this is a dev branch, I also don't care if you ignore those comments either.

[SirTyson]

Can we add a test to hit the flakey flow (or at least hit this branch at all)? I see we have test_request_tx_set_flow but I think that's a little too low level. I'm still kinda new to rust so I might be missing something, but it looks like we still write the 32 byte payload instead of the new payload with slotSeq.

[SirTyson]

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.

[SirTyson]

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?

[SirTyson]

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
              }
          }
      });
  }