large messages over slow links are deltas

4 files changed, 85 insertions, 14 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 2a24007..2119eda 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -95,6 +95,9 @@ pub(crate) struct SyncReplyWait {
     pub(crate) started: (Instant, SyncBytes),
     /// Changes first sent in the unanswered messages.
     pub(crate) sent: Vec<crate::document::ChangeHash>,
+    /// The largest of them, in bytes: if they go unanswered, messages that large may
+    /// not make it to the peer whole (`PeerState::whole_failed`).
+    pub(crate) largest: usize,
 }
 
 impl SyncSession {
@@ -143,12 +146,13 @@ impl SyncSession {
             .difference(&sent_before)
             .copied()
             .collect();
+        let payload = message.map(|message| SyncPayload(message.encode().into()));
         if in_step {
             self.reply_settled();
-        } else if message.is_some() {
-            self.expect_reply(sent.clone(), now, bytes);
+        } else if let Some(payload) = &payload {
+            self.expect_reply(sent.clone(), now, bytes, payload.0.len());
         }
-        if message.is_some() {
+        if payload.is_some() {
             // Marked together with the generation, so no second message is generated
             // while this one is handed over.
             self.sending = Some(Sending {
@@ -156,7 +160,7 @@ impl SyncSession {
                 regenerate: false,
             });
         }
-        message.map(|message| SyncPayload(message.encode().into()))
+        payload
     }
 
     /// Applies a sync message from the peer to `doc`. Doesn't settle the awaited
@@ -303,22 +307,27 @@ impl SyncSession {
         Some(delay)
     }
 
-    /// A sync message went out; `sent` are the changes it carried for the first time.
-    /// `bytes` are the sync bytes exchanged with the peer before it.
+    /// A sync message of `len` bytes went out; `sent` are the changes it carried for the
+    /// first time. `bytes` are the sync bytes exchanged with the peer before it.
     pub(crate) fn expect_reply(
         &mut self,
         sent: Vec<crate::document::ChangeHash>,
         now: Instant,
         bytes: SyncBytes,
+        len: usize,
     ) {
         match &mut self.reply {
             // Still waiting for an earlier one: keep its deadline, remember these too.
-            Some(wait) => wait.sent.extend(sent),
+            Some(wait) => {
+                wait.sent.extend(sent);
+                wait.largest = wait.largest.max(len);
+            }
             None => {
                 self.reply = Some(SyncReplyWait {
                     deadline: now + self.reply_backoff.current(),
                     started: (now, bytes),
                     sent,
+                    largest: len,
                 });
             }
         }
diff --git a/src/state/peer.rs b/src/state/peer.rs
index cb42dc1..063b72b 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -151,8 +151,17 @@ pub(crate) struct PeerState {
 
     /// What went wrong with this peer so far (counted only, for /peers and /admin/stats).
     pub(crate) incidents: Incidents,
+
+    /// The largest sync message to it that didn't make it (its send failed, or it went
+    /// unanswered), in bytes, and when: messages about that large go as deltas for a
+    /// while (`swarm::blob::Blobs::delta_threshold_for`), which a slow link takes chunk
+    /// by chunk.
+    whole_failed: Option<(usize, Instant)>,
 }
 
+/// How long a sync message that didn't make it keeps others that large from going whole.
+const WHOLE_FAILED_KEPT: Duration = Duration::from_secs(3600);
+
 /// Trouble with a peer, counted since it was first listed. Lost keepalives and refused
 /// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING,
 /// every STATUS_REPLY a STATUS.
@@ -221,6 +230,7 @@ impl PeerState {
             hello_attempts: 0,
             unreachable_since: None,
             incidents: Incidents::default(),
+            whole_failed: None,
         }
     }
 
@@ -239,13 +249,32 @@ impl PeerState {
         if self.liveness.is_dormant() || self.hello_state != HelloState::Established {
             return None;
         }
+        let largest = self.sync.reply.as_ref().map(|wait| wait.largest);
         let overdue = self.sync.take_overdue_reply(now);
         if overdue.is_some() {
             self.incidents.reply_timeouts += 1;
+            if let Some(largest) = largest {
+                self.whole_failed(largest, now);
+            }
         }
         overdue
     }
 
+    /// A sync message of `bytes` to it didn't make it (`whole_failed`).
+    pub(crate) fn whole_failed(&mut self, bytes: usize, now: Instant) {
+        let kept = self
+            .whole_failed
+            .filter(|(failed, at)| *failed >= bytes && now < *at + WHOLE_FAILED_KEPT);
+        self.whole_failed = kept.or(Some((bytes, now)));
+    }
+
+    /// The largest sync message to it that didn't make it lately, in bytes.
+    pub(crate) fn failed_whole(&self, now: Instant) -> Option<usize> {
+        self.whole_failed
+            .filter(|(_, at)| now < *at + WHOLE_FAILED_KEPT)
+            .map(|(bytes, _)| bytes)
+    }
+
     fn quiet_for(&self, now: Instant) -> Duration {
         now.saturating_duration_since(self.last_seen.unwrap_or(self.added_at))
     }
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<f64>, failed: Option<usize>) -> 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);