keepalive takes turns

4 files changed, 104 insertions, 14 deletions

diff --git a/src/state/peer.rs b/src/state/peer.rs
index 1f08767..b723ccd 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -67,8 +67,8 @@ 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 keepalives. Fixed for the
-    /// pair, so computed once.
+    /// Whether we lead the pair (`swarm::pair`): we send the first keepalive. Fixed for
+    /// the pair, so computed once.
     pub(crate) leads: bool,
 
     pub(crate) standing: Standing,
@@ -87,7 +87,9 @@ pub(crate) struct PeerState {
     pub(crate) last_seen: Option<Instant>,
     /// Our last keepalive to it (a PING or a STATUS).
     pub(crate) last_keepalive_at: Option<Instant>,
-    /// Our last STATUS to it, for whatever reason (`status_refresh`).
+    /// 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
@@ -186,6 +188,7 @@ impl PeerState {
             added_at: Instant::now(),
             last_seen: None,
             last_keepalive_at: None,
+            keepalive_received_at: None,
             last_status_at: None,
             last_sync_rx: None,
             last_resync_at: None,
@@ -302,21 +305,37 @@ impl PeerState {
     }
 
     /// The keepalive an established, quiet peer needs, if any (and records it): a PING,
-    /// or a STATUS if we sent it none for `status_refresh`.
+    /// 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 only one side of a pair needs to send them: the one leading the
-    /// pair (`leads`). Each of its keepalives makes the other side's peer look recently
-    /// heard from, so the other side never sends while it does. If they stop, the other
-    /// side sends 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 keepalives would often cross.
+    /// 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.
+    ///
+    /// 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 quiet_enough = if self.leads {
+        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
@@ -405,6 +424,61 @@ 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() {
diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs
index 41fc219..0d45b32 100644
--- a/src/swarm/pair.rs
+++ b/src/swarm/pair.rs
@@ -1,5 +1,5 @@
-//! Which side of a pair of nodes leads, for duties only one side needs to do (sending
-//! keepalive STATUSes).
+//! Which side of a pair of nodes leads, for what one side needs to start (the first
+//! keepalive; then the two take turns, `PeerState::take_keepalive_due`).
 //!
 //! Both sides decide on their own, from the two keys alone, and agree. The leader is
 //! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 0eaefb9..2e89739 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -11,7 +11,7 @@ use super::epoch::observe_peer_epoch;
 use super::handshake::{
     discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply,
 };
-use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status};
+use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status, status_received};
 use super::lanes::SyncLanes;
 use crate::state::AppState;
 use crate::swarm::sync::SyncEvent;
@@ -149,6 +149,7 @@ pub(crate) async fn handle_message(
             crate::swarm::blob::handle(state, &peer, blob).await;
         }
         transport::ParsedMessage::Status(remote) => {
+            status_received(state, &peer).await;
             let local = on_status(state, &peer, &remote, reactivated).await;
             if let Err(error) = state
                 .transport
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 460b71e..efb8a65 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -63,6 +63,10 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes
     };
     let reactivated = {
         let mut peer_state = peer_state.lock().await;
+        if matches!(message, ParsedMessage::Ping(_)) {
+            // Its keepalive (or a path probe): the next is ours.
+            peer_state.keepalive_received_at = Some(now);
+        }
         if peer_state.unreachable_since.is_some() {
             // Reachable again (`mark_seen` clears it).
             state.topology_changed.notify_one();
@@ -147,6 +151,17 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
     }
 }
 
+/// A STATUS (not a reply) from `peer`: its keepalive, and a STATUS between us
+/// (`Timing::status_refresh`).
+pub(super) async fn status_received(state: &AppState, peer: &str) {
+    if let Some(peer_state) = state.get_peer_state(peer).await {
+        let mut peer_state = peer_state.lock().await;
+        let now = Instant::now();
+        peer_state.keepalive_received_at = Some(now);
+        peer_state.last_status_at = Some(now);
+    }
+}
+
 /// What a peer's status says about it.
 pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) {
     if let Some(peer_state) = state.get_peer_state(peer).await {