1 files changed, 148 insertions, 20 deletions

diff --git a/src/state/peer.rs b/src/state/peer.rs
index 8086f63..b723ccd 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -67,7 +67,7 @@ const HELLO_RETRY_MAX: Duration = Duration::from_secs(30);
 
 #[derive(Clone)]
 pub(crate) struct PeerState {
-    /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for
+    /// Whether we lead the pair (`swarm::pair`): we send the first keepalive. Fixed for
     /// the pair, so computed once.
     pub(crate) leads: bool,
 
@@ -85,7 +85,12 @@ pub(crate) struct PeerState {
     pub(crate) liveness: Liveness,
     pub(crate) added_at: Instant,
     pub(crate) last_seen: Option<Instant>,
-    pub(crate) last_ping_at: Option<Instant>,
+    /// Our last keepalive to it (a PING or a STATUS).
+    pub(crate) last_keepalive_at: Option<Instant>,
+    /// Its last keepalive to us (a PING or a STATUS): after it, the next is ours.
+    pub(crate) keepalive_received_at: Option<Instant>,
+    /// The last STATUS between us, either way (`status_refresh`).
+    pub(crate) last_status_at: Option<Instant>,
 
     /// Last time a sync message from this peer was applied (survives resets: the
     /// keepalive heal looks at it).
@@ -133,7 +138,8 @@ pub(crate) struct PeerState {
 }
 
 /// 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.
+/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING,
+/// every STATUS_REPLY a STATUS.
 #[derive(Debug, Default, Clone, Copy, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct Incidents {
@@ -157,6 +163,15 @@ impl Incidents {
     }
 }
 
+/// What a keepalive due is (`PeerState::take_keepalive_due`).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum Keepalive {
+    /// A PING: the peer's transport answers it with a PONG.
+    Ping,
+    /// A STATUS: our state, answered with the peer's.
+    Status,
+}
+
 /// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart).
 const UNREACHABLE_AFTER_HELLOS: u32 = 3;
 
@@ -172,7 +187,9 @@ impl PeerState {
             liveness: Liveness::Active,
             added_at: Instant::now(),
             last_seen: None,
-            last_ping_at: None,
+            last_keepalive_at: None,
+            keepalive_received_at: None,
+            last_status_at: None,
             last_sync_rx: None,
             last_resync_at: None,
             diverged_since: None,
@@ -287,17 +304,38 @@ impl PeerState {
         true
     }
 
-    /// True (and records the ping) if an established, quiet peer needs a keepalive.
+    /// The keepalive an established, quiet peer needs, if any (and records it): a PING,
+    /// or a STATUS if none went between us for `status_refresh`.
+    ///
+    /// A PING and its PONG (or a STATUS and its STATUS_REPLY) tell both sides that the
+    /// other is alive, so one per interval does for the pair. The two sides take turns:
+    /// after a keepalive from the peer, the next is ours (the one leading the pair,
+    /// `leads`, sends the first), so both measure the round trip and each sends every
+    /// other interval. On our turn we send once the pair was quiet for an interval; on its
+    /// turn only after half an interval more, in case it doesn't (a keepalive or its
+    /// answer was lost). Without turns, both sides would get due at about the same time
+    /// (each one's quiet time starts with the same exchange) and their keepalives would
+    /// often cross.
     ///
-    /// One PING and its PONG tell both sides that the other is alive, so only one side of
-    /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs
-    /// makes the other side's peer look recently heard from, so the other side never
-    /// pings while it does. If the pings stop, the other side pings once it has heard
-    /// nothing for half an interval more. Without that, both sides would get due at about
-    /// the same time (each one's quiet time starts with the same exchange) and their
-    /// PINGs would often cross.
-    pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
-        let quiet_enough = if self.leads {
+    /// When they do cross anyway (both sent before seeing the other's: the peer's came
+    /// within half an interval after ours), each side would take the next turn for its
+    /// own and they would keep crossing; then the next turn is the leader's.
+    pub(crate) fn take_keepalive_due(
+        &mut self,
+        timing: &Timing,
+        now: Instant,
+    ) -> Option<Keepalive> {
+        let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) {
+            (None, None) => self.leads,
+            (None, Some(_)) => true,
+            (Some(_), None) => false,
+            (Some(sent), Some(received)) if received > sent => {
+                let crossed = received.saturating_duration_since(sent) < timing.keepalive / 2;
+                !crossed || self.leads
+            }
+            (Some(_), Some(_)) => false,
+        };
+        let quiet_enough = if our_turn {
             timing.keepalive
         } else {
             timing.keepalive + timing.keepalive / 2
@@ -307,16 +345,24 @@ impl PeerState {
             && now.saturating_duration_since(self.added_at) >= timing.keepalive
             && self.quiet_for(now) >= quiet_enough
             && self
-                .last_ping_at
+                .last_keepalive_at
                 .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
-
-        if due {
-            self.last_ping_at = Some(now);
+        if !due {
+            return None;
         }
-        due
+        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
+        })
     }
 
-    /// Self-healing check on PING/PONG. Returns true if a re-handshake should start.
+    /// Self-healing check on a STATUS or STATUS_REPLY. Returns true if a re-handshake
+    /// should start.
     pub(crate) fn needs_heal(
         &mut self,
         local: &transport::HeadsSummary,
@@ -378,6 +424,88 @@ impl PeerState {
 mod tests {
     use super::*;
 
+    /// The two sides of a pair take turns: each sends every other interval.
+    #[test]
+    fn keepalives_take_turns() {
+        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
+        let mut a = PeerState::new(true, &timing);
+        let mut b = PeerState::new(false, &timing);
+        a.hello_state = HelloState::Established;
+        b.hello_state = HelloState::Established;
+        b.added_at = a.added_at;
+        let mut senders = Vec::new();
+        let mut now = a.added_at;
+        for _ in 0..40 {
+            now += Duration::from_secs(5);
+            // Whoever is due sends; the other one gets it at once, and both hear from
+            // each other (the keepalive, its answer).
+            if a.take_keepalive_due(&timing, now).is_some() {
+                senders.push("a");
+                b.keepalive_received_at = Some(now);
+            } else if b.take_keepalive_due(&timing, now).is_some() {
+                senders.push("b");
+                a.keepalive_received_at = Some(now);
+            } else {
+                continue;
+            }
+            a.mark_seen(now);
+            b.mark_seen(now);
+        }
+        // 200 s: one keepalive a minute, a first, then taking turns.
+        assert_eq!(senders, ["a", "b", "a"]);
+    }
+
+    /// Keepalives that crossed (both sides sent before seeing the other's) leave the next
+    /// turn to the leader, rather than both taking it and crossing again.
+    #[test]
+    fn crossed_keepalives_go_back_to_taking_turns() {
+        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
+        let mut a = PeerState::new(true, &timing);
+        let mut b = PeerState::new(false, &timing);
+        let start = a.added_at;
+        b.added_at = start;
+        for peer in [&mut a, &mut b] {
+            peer.hello_state = HelloState::Established;
+        }
+        // Both sent at 100 s and got the other's a moment later.
+        let sent = start + Duration::from_secs(100);
+        for peer in [&mut a, &mut b] {
+            peer.last_keepalive_at = Some(sent);
+            peer.keepalive_received_at = Some(sent + Duration::from_millis(50));
+            peer.mark_seen(sent + Duration::from_millis(50));
+        }
+        let next = sent + timing.keepalive + Duration::from_secs(1);
+        assert!(a.take_keepalive_due(&timing, next).is_some()); // The leader's turn.
+        assert!(b.take_keepalive_due(&timing, next).is_none());
+    }
+
+    /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet.
+    #[test]
+    fn keepalives_are_pings_and_now_and_then_a_status() {
+        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
+        let mut peer = PeerState::new(true, &timing);
+        peer.hello_state = HelloState::Established;
+        let start = peer.added_at;
+        assert_eq!(peer.take_keepalive_due(&timing, start), None); // Too new.
+        let mut kinds = Vec::new();
+        for interval in 1..=6 {
+            let now = start + timing.keepalive * interval;
+            let due = peer.take_keepalive_due(&timing, now);
+            if due == Some(Keepalive::Status) {
+                peer.last_status_at = Some(now);
+            }
+            kinds.extend(due);
+            // Not again within the interval.
+            assert_eq!(peer.take_keepalive_due(&timing, now), None);
+        }
+        use Keepalive::{Ping, Status};
+        assert_eq!(kinds, [Status, Ping, Ping, Status, Ping, Ping]);
+        // Heard from just now: nothing due.
+        let now = start + timing.keepalive * 7;
+        peer.mark_seen(now);
+        assert_eq!(peer.take_keepalive_due(&timing, now), None);
+    }
+
     /// R41: unreachability lasts until the peer is heard from, not for a set time, and
     /// probes back off meanwhile.
     #[test]