1 files changed, 160 insertions, 22 deletions

diff --git a/src/state/peer.rs b/src/state/peer.rs
index 923c6d2..063b72b 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -4,9 +4,10 @@ use std::time::Duration;
 
 use tokio::time::Instant;
 
-use super::{SyncSession, Timing};
+use super::Timing;
 use crate::backoff::Backoff;
 use crate::document::replica::Epoch;
+use crate::document::sync::SyncSession;
 use crate::transport;
 
 /// How far a peer got into the swarm, as far as we're concerned.
@@ -137,6 +138,10 @@ pub(crate) struct PeerState {
     /// How many heads its most recent keepalive listed (None if it sent a digest).
     pub(crate) reported_heads: Option<usize>,
 
+    /// It said farewell (`say_farewell`): what else that run of it still sends while it
+    /// stops doesn't make it active again; a message from another run does (`returned`).
+    pub(crate) left: bool,
+
     /// HELLOs sent in the current handshake attempt without hearing anything back.
     pub(crate) hello_attempts: u32,
     /// Since when we consider the peer unreachable (three unanswered HELLOs, or
@@ -146,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.
@@ -212,9 +226,11 @@ impl PeerState {
             reports_ready: None,
             syncs_with_us: None,
             reported_heads: None,
+            left: false,
             hello_attempts: 0,
             unreachable_since: None,
             incidents: Incidents::default(),
+            whole_failed: None,
         }
     }
 
@@ -233,20 +249,43 @@ 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))
     }
 
-    /// Records inbound activity. Returns true if the peer was dormant.
+    /// Records inbound activity. Returns true if the peer was dormant. From a peer that
+    /// said farewell (`left`), only that it was heard from.
     pub(crate) fn mark_seen(&mut self, now: Instant) -> bool {
         self.last_seen = Some(now);
+        if self.left {
+            return false;
+        }
         self.hello_attempts = 0;
         self.unreachable_since = None;
         self.probe_backoff.reset();
@@ -275,11 +314,32 @@ impl PeerState {
         }
     }
 
+    /// Whether we sync with it on `epoch`: an active member on that epoch (what our
+    /// STATUS tells it as `syncs_with_you`).
+    pub(crate) fn syncs_on(&self, epoch: Epoch) -> bool {
+        !self.liveness.is_dormant()
+            && matches!(self.standing, Standing::Member { .. })
+            && self.epoch == Some(epoch)
+    }
+
     /// Whether to leave the peer out when choosing neighbours (`swarm::topology`).
     pub(crate) fn is_unreachable(&self) -> bool {
         self.unreachable_since.is_some()
     }
 
+    /// What choosing neighbours reads of this peer (`swarm::topology::refresh`): whether
+    /// it is unreachable, whether it was ever heard from, whether it is a member, and
+    /// whether it syncs with us. A change to any may change the neighbours
+    /// (`swarm::topology::change_peer`).
+    pub(crate) fn neighbour_inputs(&self) -> impl PartialEq + use<> {
+        (
+            self.is_unreachable(),
+            self.last_seen.is_some(),
+            matches!(self.standing, Standing::Member { .. }),
+            self.syncs_with_us,
+        )
+    }
+
     /// The wait before the next HELLO: a probe's if unreachable.
     pub(crate) fn next_hello_wait(&mut self) -> Duration {
         if self.is_unreachable() {
@@ -304,9 +364,32 @@ impl PeerState {
         if self.liveness.is_dormant() || self.quiet_for(now) < timing.dormant_after {
             return false;
         }
+        self.incidents.dormancies += 1;
+        self.go_dormant(now);
+        true
+    }
+
+    /// The peer said farewell (it is shutting down): Dormant now, without waiting for
+    /// its silence, and not counted as an incident. Returns true on the transition.
+    pub(crate) fn say_farewell(&mut self, now: Instant) -> bool {
+        self.left = true;
+        if self.liveness.is_dormant() {
+            return false;
+        }
+        self.go_dormant(now);
+        true
+    }
+
+    /// A message from `run` of the peer (a HELLO's or HELLO_REPLY's): a run other than
+    /// the one that said farewell is back (`left`).
+    pub(crate) fn returned(&mut self, run: crate::transport::RunId) {
+        if self.run_id != Some(run) {
+            self.left = false;
+        }
+    }
 
+    fn go_dormant(&mut self, now: Instant) {
         self.liveness = Liveness::Dormant { since: now };
-        self.incidents.dormancies += 1;
         self.unreachable_since.get_or_insert(now);
         self.reset_sync();
 
@@ -315,7 +398,6 @@ impl PeerState {
                 next_attempt: now + self.probe_backoff.next(),
             };
         }
-        true
     }
 
     /// The keepalive an established, quiet peer needs, if any (and records it): a PING,
@@ -339,6 +421,26 @@ impl PeerState {
         timing: &Timing,
         now: Instant,
     ) -> Option<Keepalive> {
+        if self.keepalive_due_at(timing).is_none_or(|at| now < at) {
+            return None;
+        }
+        self.last_keepalive_at = Some(now);
+        let status = self
+            .last_status_at
+            .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
+        Some(if status {
+            Keepalive::Status
+        } else {
+            Keepalive::Ping
+        })
+    }
+
+    /// When the next keepalive is due (`take_keepalive_due`), if any is: none to a
+    /// dormant peer, nor before the handshake.
+    pub(crate) fn keepalive_due_at(&self, timing: &Timing) -> Option<Instant> {
+        if self.liveness.is_dormant() || self.hello_state != HelloState::Established {
+            return None;
+        }
         let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) {
             (None, None) => self.leads,
             (None, Some(_)) => true,
@@ -354,27 +456,44 @@ impl PeerState {
         } else {
             timing.keepalive + timing.keepalive / 2
         };
-        let due = !self.liveness.is_dormant()
-            && self.hello_state == HelloState::Established
-            && now.saturating_duration_since(self.added_at) >= timing.keepalive
-            && self.quiet_for(now) >= quiet_enough
-            && self
-                .last_keepalive_at
-                .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
-        if !due {
-            return None;
-        }
-        self.last_keepalive_at = Some(now);
-        let status = self
-            .last_status_at
-            .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
-        Some(if status {
-            Keepalive::Status
-        } else {
-            Keepalive::Ping
+        let quiet_since = self.last_seen.unwrap_or(self.added_at);
+        let at = (self.added_at + timing.keepalive).max(quiet_since + quiet_enough);
+        Some(match self.last_keepalive_at {
+            Some(sent) => at.max(sent + timing.keepalive),
+            None => at,
         })
     }
 
+    /// When the peer worker next has something to do for this peer
+    /// (`swarm::peer::run_peer_worker`): a keepalive (to a `neighbour`), dormancy (of a
+    /// neighbour, once `ready`), a HELLO retry, an overdue sync answer. None if nothing
+    /// is due until something changes.
+    pub(crate) fn next_due(
+        &self,
+        timing: &Timing,
+        neighbour: bool,
+        ready: bool,
+    ) -> Option<Instant> {
+        let dormancy = (ready && neighbour && !self.liveness.is_dormant())
+            .then(|| self.last_seen.unwrap_or(self.added_at) + timing.dormant_after);
+        let keepalive = if neighbour {
+            self.keepalive_due_at(timing)
+        } else {
+            None
+        };
+        let hello = match self.hello_state {
+            HelloState::AwaitingReply { next_attempt } => Some(next_attempt),
+            _ => None,
+        };
+        let reply = (!self.liveness.is_dormant() && self.hello_state == HelloState::Established)
+            .then(|| self.sync.reply.as_ref().map(|wait| wait.deadline))
+            .flatten();
+        [dormancy, keepalive, hello, reply]
+            .into_iter()
+            .flatten()
+            .min()
+    }
+
     /// Self-healing check on a STATUS or STATUS_REPLY. Returns true if a re-handshake
     /// should start.
     pub(crate) fn needs_heal(
@@ -454,6 +573,25 @@ mod tests {
         assert!(silent.liveness.is_dormant());
     }
 
+    /// After its farewell, a peer stays dormant whatever its stopping run still sends;
+    /// its next run makes it active again.
+    #[test]
+    fn dormant_after_farewell_until_back() {
+        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
+        let mut peer = PeerState::new(true, &timing);
+        let now = peer.added_at;
+        peer.run_id = Some([1; 16]);
+        peer.mark_seen(now);
+        assert!(peer.say_farewell(now));
+        assert!(!peer.mark_seen(now), "a straggler from the same run");
+        assert!(peer.liveness.is_dormant() && peer.is_unreachable());
+        peer.returned([1; 16]);
+        assert!(!peer.mark_seen(now), "the same run again");
+        peer.returned([2; 16]);
+        assert!(peer.mark_seen(now), "its next run");
+        assert!(!peer.liveness.is_dormant() && !peer.is_unreachable());
+    }
+
     /// The two sides of a pair take turns: each sends every other interval.
     #[test]
     fn keepalives_take_turns() {