aboutsummaryrefslogtreecommitdiffstats
path: root/src/state/peer.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/state/peer.rs')
-rw-r--r--src/state/peer.rs29
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))
}