large messages over slow links are deltas

1 files changed, 29 insertions, 0 deletions

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