4 files changed, 119 insertions, 30 deletions

diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 92a1a5e..0eaefb9 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_status};
+use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status};
 use super::lanes::SyncLanes;
 use crate::state::AppState;
 use crate::swarm::sync::SyncEvent;
@@ -30,6 +30,11 @@ pub(crate) async fn handle_message(
         debug!("Ignoring message from self");
         return Ok(());
     }
+    // PING and PONG only say the peer is alive (`keepalive::note_alive`).
+    if matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) {
+        note_alive(state, &peer, &message).await;
+        return Ok(());
+    }
     // WHO and ME end at the identity layer (`transport::identity`).
     let Some((epoch, epoch_seq)) = message.epoch_stamp() else {
         return Ok(());
@@ -154,10 +159,6 @@ pub(crate) async fn handle_message(
             }
         }
         transport::ParsedMessage::StatusReply(remote) => {
-            // It answers our STATUS: a round-trip sample.
-            state
-                .lock_rtt()
-                .status_reply_received(&peer, Instant::now());
             on_status(state, &peer, &remote, reactivated).await;
         }
         transport::ParsedMessage::Who(_)
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 810c377..460b71e 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -1,12 +1,19 @@
-//! Keepalives (STATUS / STATUS_REPLY) 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 gives the round trip (`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,33 +21,100 @@ use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 use crate::transport::{self, ParsedMessage};
 
-/// Sends `peer` a STATUS with our status, 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;
-    state.lock_rtt().status_sent(peer, Instant::now());
+    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, noting it for its round trip.
+pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
+    let nonce = crate::transport::paths::random_nonce();
+    state.lock_rtt().ping_sent(peer, nonce, Instant::now());
+    state
+        .transport
+        .send_message(peer, &ParsedMessage::Ping(nonce))
+        .await
+}
+
+/// A PING or PONG from `peer`: it is alive, and a PONG may answer our PING (its round
+/// trip). 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(nonce) => state.lock_rtt().pong_received(peer, *nonce, now),
+        _ => {}
+    }
+    let Some(peer_state) = state.get_peer_state(peer).await else {
+        return;
+    };
+    let reactivated = {
+        let mut peer_state = peer_state.lock().await;
+        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 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);
+/// 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();
-            statuses.spawn(async move { send_status(&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 statuses.join_next().await.is_some() {} };
-    let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).await;
+    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 status for `peer`.
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index 415cddf..326f3a9 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -2,7 +2,8 @@
 //! swarm-wide forget records, and dispatch of inbound messages.
 //!
 //! * `handshake`: discovering peers, HELLO / HELLO_REPLY, membership.
-//! * `keepalive`: STATUS / STATUS_REPLY and the self-healing re-handshake.
+//! * `keepalive`: PING / PONG and STATUS / STATUS_REPLY, and the self-healing
+//!   re-handshake.
 //! * `epoch`: the epochs peers advertise.
 //! * `admit`: admission with a swarm key.
 //! * `forget`: swarm-wide forget records.
@@ -24,6 +25,6 @@ pub(crate) use handshake::{
     apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once,
 };
 pub(crate) use inbound::handle_message;
-pub(crate) use keepalive::{farewell, send_status};
+pub(crate) use keepalive::{announce, farewell, send_status};
 pub(crate) use lanes::SyncLanes;
 pub(crate) use worker::run_peer_worker;
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index c68e2ea..f3d0e84 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -10,9 +10,9 @@ use tracing::{debug, info, warn};
 
 use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
 use super::handshake::retry_hello_if_due;
-use super::keepalive::send_status;
+use super::keepalive::{send_ping, send_status};
 use crate::locks::MutexExt;
-use crate::state::AppState;
+use crate::state::{AppState, Keepalive};
 use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 
@@ -50,16 +50,20 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             // Only neighbours are kept alive and judged; others we only answer.
             let neighbour = topology::is_neighbour(&state, peer);
 
-            let (became_dormant, ping_due, unanswered) = {
+            let (became_dormant, keepalive, unanswered) = {
                 let mut peer_state = peer_state_mutex.lock().await;
                 let became_dormant =
                     ready && neighbour && peer_state.check_dormancy(&state.timing, now);
-                let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now);
+                let keepalive = if neighbour {
+                    peer_state.take_keepalive_due(&state.timing, now)
+                } else {
+                    None
+                };
                 if !neighbour {
                     peer_state.stop_hello_retries();
                 }
                 let unanswered = peer_state.take_overdue_sync_reply(now);
-                (became_dormant, ping_due, unanswered)
+                (became_dormant, keepalive, unanswered)
             };
 
             if let Some(changes) = unanswered {
@@ -80,12 +84,21 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                 );
             }
 
-            let probe_due =
-                !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now());
-            if (ping_due || probe_due)
-                && let Err(error) = send_status(&state, peer).await
-            {
-                debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive STATUS failed");
+            // The catch-up source gets STATUSes: we need its readiness, and whether it
+            // syncs with us.
+            let keepalive = match keepalive {
+                None if crate::swarm::source::probe_due(&state, peer, Instant::now()) => {
+                    Some(Keepalive::Status)
+                }
+                keepalive => keepalive,
+            };
+            let sent = match keepalive {
+                Some(Keepalive::Ping) => send_ping(&state, peer).await,
+                Some(Keepalive::Status) => send_status(&state, peer).await,
+                None => Ok(()),
+            };
+            if let Err(error) = sent {
+                debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive failed");
             }
 
             if let Err(error) = retry_hello_if_due(&state, peer).await {