1 files changed, 116 insertions, 24 deletions
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index ce2951b..c8a93eb 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -1,12 +1,19 @@
-//! Keepalives (PING / PONG) and the self-healing re-handshake when they show our
-//! documents differ while sync is idle.
+//! Keepalives and the self-healing re-handshake.
+//!
+//! A quiet pair's keepalive (`PeerState::take_keepalive_due`) is a PING, answered with a
+//! PONG by the peer's paths layer (`transport::paths`), over the same path: it tells both
+//! sides the other is alive, and measures the path (`swarm::rtt`). Every `Timing::status_refresh` it is a
+//! STATUS instead, answered with a STATUS_REPLY: each side's state (heads, readiness,
+//! epoch), which also go out as they change (relay announcements, the epoch after a
+//! switch, readiness: `announce`). When a STATUS shows our documents differ while sync
+//! is idle, the pair re-handshakes (`heal_if_diverged`).
 
 use std::sync::atomic::Ordering;
 use std::time::Duration;
 
 use anyhow::Result;
 use tokio::time::Instant;
-use tracing::warn;
+use tracing::{debug, info, warn};
 
 use super::handshake::ensure_handshake;
 use crate::state::{AppState, Liveness, Standing};
@@ -14,37 +21,111 @@ use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 use crate::transport::{self, ParsedMessage};
 
-/// Sends `peer` a PING with our keepalive, noting the time for its round trip.
+/// Sends `peer` a STATUS with our status.
+pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
+    let status = local_status(state, peer).await;
+    if let Some(peer_state) = state.get_peer_state(peer).await {
+        peer_state.lock().await.last_status_at = Some(Instant::now());
+    }
+    state
+        .transport
+        .send_message(peer, &ParsedMessage::Status(status))
+        .await
+}
+
+/// Sends `peer` a PING (the paths layer measures the path with it).
 pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
-    let keepalive = local_keepalive(state, peer).await;
-    state.lock_rtt().ping_sent(peer, Instant::now());
+    let nonce = crate::transport::paths::random_nonce();
     state
         .transport
-        .send_message(peer, &ParsedMessage::Ping(keepalive))
+        .send_message(peer, &ParsedMessage::Ping(nonce))
         .await
 }
 
-/// How long shutting down waits for the last PINGs (`farewell`).
+/// A PING or PONG from `peer`: it is alive, and after a PONG the paths layer has a
+/// new round-trip time to it (`swarm::rtt`). A transport without the paths layer (`Transport::answers_pings`: in tests)
+/// leaves answering a PING to us. Only peers we know already: a PING makes none.
+pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMessage) {
+    let now = Instant::now();
+    match message {
+        ParsedMessage::Ping(nonce) if !state.transport.answers_pings() => {
+            let pong = ParsedMessage::Pong(*nonce);
+            if let Err(error) = state.transport.send_message(peer, &pong).await {
+                debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
+            }
+        }
+        ParsedMessage::Pong(_) => {
+            if let Some(rtt) = state.transport.rtt(peer) {
+                state.lock_rtt().measured(peer, rtt, now);
+            }
+        }
+        _ => {}
+    }
+    let Some(peer_state) = state.get_peer_state(peer).await else {
+        return;
+    };
+    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();
+        }
+        peer_state.mark_seen(now)
+    };
+    crate::swarm::source::heard_from(state, peer);
+    if reactivated {
+        info!(peer = %peer, "Dormant peer is active again; re-handshaking");
+        ensure_handshake(state, peer).await;
+    }
+}
+
+/// How long shutting down waits for the last STATUSes (`farewell`).
 const FAREWELL_TIMEOUT: Duration = Duration::from_secs(2);
 
-/// At shutdown: one last PING to each neighbour, saying we aren't ready. A neighbour
-/// catching up from us picks another source now, instead of once we go dormant. Bounded
-/// in time: our transport may be gone already.
-pub(crate) async fn farewell(state: &AppState) {
-    state.stopping.store(true, Ordering::Release);
-    let mut pings = tokio::task::JoinSet::new();
+/// A STATUS to each neighbour, at most `within` (all sent: none).
+async fn status_to_neighbours(state: &AppState, within: Option<Duration>) {
+    let mut statuses = tokio::task::JoinSet::new();
     for (peer, _) in state.peers_snapshot().await {
         if topology::is_neighbour(state, &peer) {
             let state = state.clone();
-            pings.spawn(async move { send_ping(&state, &peer).await });
+            statuses.spawn(async move {
+                if let Err(error) = send_status(&state, &peer).await {
+                    debug!(peer = %peer, error = %format!("{error:#}"), "STATUS failed");
+                }
+            });
         }
     }
-    let all_sent = async { while pings.join_next().await.is_some() {} };
-    let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).await;
+    let all_sent = async { while statuses.join_next().await.is_some() {} };
+    match within {
+        Some(within) => {
+            let _ = tokio::time::timeout(within, all_sent).await;
+        }
+        None => all_sent.await,
+    }
+}
+
+/// Our readiness changed (we became ready, or ready again after a rejoin): a STATUS
+/// to each neighbour now, rather than with the next refresh. A peer opens sync towards
+/// us only once it knows we are ready.
+pub(crate) fn announce(state: &AppState) {
+    let state = state.clone();
+    tokio::spawn(async move { status_to_neighbours(&state, None).await });
+}
+
+/// At shutdown: one last STATUS to each neighbour, saying we aren't ready. A neighbour
+/// catching up from us picks another source now, instead of once we go dormant. Bounded
+/// in time: our transport may be gone already.
+pub(crate) async fn farewell(state: &AppState) {
+    state.stopping.store(true, Ordering::Release);
+    status_to_neighbours(state, Some(FAREWELL_TIMEOUT)).await;
 }
 
-/// Our keepalive for `peer`.
-pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
+/// Our status for `peer`.
+pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Status {
     // Heads and epoch are published together (`Replica::stamps`), so they always match,
     // and reading them doesn't wait for a sync message being applied.
     let (heads, epoch, epoch_seq) = {
@@ -64,7 +145,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
         }
         None => false,
     };
-    transport::Keepalive {
+    transport::Status {
         heads,
         epoch,
         epoch_seq,
@@ -73,8 +154,19 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
     }
 }
 
-/// What a peer's keepalive says about it.
-pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) {
+/// 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 {
         let mut peer_state = peer_state.lock().await;
         let became_ready = remote.ready && peer_state.reports_ready != Some(true);
@@ -103,8 +195,8 @@ pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transp
 pub(super) async fn heal_if_diverged(
     state: &AppState,
     peer: &str,
-    local: &transport::Keepalive,
-    remote: &transport::Keepalive,
+    local: &transport::Status,
+    remote: &transport::Status,
 ) {
     // Heads of documents from different epochs are never equal and never converge.
     // And a peer that isn't our neighbour isn't ours to keep in step.