From 98b49df1e7631f99b0ad780fb0f01924cebd0756 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 21:15:56 +0100 Subject: large messages over slow links are deltas --- src/swarm/blob.rs | 23 +++++++++++++++++++++++ src/swarm/sync.rs | 24 +++++++++++++++++------- 2 files changed, 40 insertions(+), 7 deletions(-) (limited to 'src/swarm') diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 3f66d50..ce9a2bd 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -87,6 +87,10 @@ pub(crate) struct Blobs { /// lost message costs too much to send again whole, and a slow link may not deliver it /// within the transport's delivery limit. const DELTA_THRESHOLD_CHUNKS: usize = 16; +/// The longest a sync message should take whole at the throughput measured to its peer +/// (`delta_threshold_for`): well within what a transport allows a message before its send +/// fails (10 s, plus its size at 1 MiB/s, on libp2p). +const WHOLE_WITHIN: Duration = Duration::from_secs(5); /// The largest sync message accepted: the threshold of a sender with the largest chunks /// it may have configured. Anything larger comes from a node that doesn't keep to it. @@ -115,6 +119,25 @@ impl Blobs { DELTA_THRESHOLD_CHUNKS * self.chunk_size as usize } + /// [`delta_threshold`](Self::delta_threshold) for one peer: lower where a message + /// that large wouldn't make it whole. `rate`: the measured throughput to it, bytes + /// per second (`swarm::throughput`), at which a message whole should take at most + /// `WHOLE_WITHIN`; `failed`: the largest message to it that didn't make it lately + /// (`PeerState::failed_whole`), of which messages half as large go as deltas, if it + /// was larger than two chunks (smaller ones are lost for other reasons than their + /// size). Never below a chunk: a delta then is no smaller per message. + pub(crate) fn delta_threshold_for(&self, rate: Option, failed: Option) -> usize { + let chunk = self.chunk_size as usize; + let mut threshold = self.delta_threshold(); + if let Some(rate) = rate.filter(|rate| rate.is_finite() && *rate > 0.0) { + threshold = threshold.min((rate * WHOLE_WITHIN.as_secs_f64()) as usize); + } + if let Some(failed) = failed.filter(|failed| *failed > 2 * chunk) { + threshold = threshold.min(failed / 2); + } + threshold.max(chunk) + } + /// Starts downloading a delta offered by `peer`; false if one from it is running. pub(crate) fn start_delta(&self, peer: &str) -> bool { self.deltas.lock_or_recover().insert(peer.to_owned()) diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index c8564c2..b881609 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -128,7 +128,16 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let ready = state.is_ready().await; // Before this message is counted. let bytes_before = SyncBytes::of(&state.traffic, peer); - let delta_threshold = state.blobs.delta_threshold(); + // Lower towards a peer a message that large wouldn't reach whole: a slow link + // takes a delta chunk by chunk, where a message whole would fail and be sent + // again, whole, every time. + let rate = state + .throughput + .lock_or_recover() + .get(peer) + .map(|estimate| estimate.bytes_per_sec); + let failed = peer_state_mutex.lock().await.failed_whole(Instant::now()); + let delta_threshold = state.blobs.delta_threshold_for(rate, failed); // The session the message comes from, for its send's outcome. let (session, generated) = tokio::task::spawn_blocking( @@ -252,12 +261,13 @@ async fn send_generated( Ok(()) } Err(error) => { - let Some(delay) = peer_state_mutex - .lock() - .await - .sync - .send_failed(session, Instant::now()) - else { + let delay = { + let mut peer_state = peer_state_mutex.lock().await; + let now = Instant::now(); + peer_state.whole_failed(encoded.payload_len, now); + peer_state.sync.send_failed(session, now) + }; + let Some(delay) = delay else { // The new session may not have opened yet. debug!("Sync message refused, but a reset already discarded it"); state.request_sync(peer); -- cgit v1.3.1