move keepalive and firewall to paths

17 files changed, 541 insertions, 240 deletions

diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 58faa9c..8919ca6 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -886,7 +886,10 @@ paths:
     put:
       tags: [Debug]
       summary: Replace the firewall rules
-      description: Drops messages from (`inbound`) or to (`outbound`) these peers, like a lossy network.
+      description: |
+        Drops messages from (`inbound`) or to (`outbound`) these peers (node IDs), like a
+        lossy network: everything, at every address they proved themselves at, including
+        WHO/ME and PING/PONG (applied in the paths layer).
       requestBody:
         required: true
         content:
@@ -1225,7 +1228,7 @@ components:
           description: WHO and ME, counted under the address they went to or came from (so only in the totals).
         paths:
           $ref: '#/components/schemas/Counter'
-          description: PING and PONG, counted under the address (so only in the totals).
+          description: PING and PONG; the paths layer's own PINGs counted under the address (so only in the totals).
 
     Traffic:
       type: object
diff --git a/src/node.rs b/src/node.rs
index 8e4ace7..97dec79 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -23,7 +23,7 @@ use crate::swarm::state_file::StateFile;
 use crate::swarm::sync::SyncEvent;
 use crate::swarm::{self, peer, sync};
 use crate::transport::Transport;
-use crate::transport::firewall::{Firewall, FirewallTransport};
+use crate::transport::firewall::Firewall;
 use crate::transport::identity::{IdentityConfig, IdentityTransport};
 use crate::transport::traffic::{CountingTransport, Traffic};
 
@@ -77,6 +77,7 @@ impl Node {
                 config.transport,
                 identity,
                 Arc::clone(&traffic),
+                Some(Arc::clone(&firewall)),
             )) as Arc<dyn Transport>,
             None => config.transport,
         };
@@ -91,13 +92,11 @@ impl Node {
                 discovered_swarm_id: None,
             })),
             peer_states: Arc::new(RwLock::new(HashMap::new())),
-            // Counted outside the firewall: what it drops outbound counts as sent (it
-            // plays a lossy network), what it drops inbound never arrived. Both above
-            // the identity layer, so by node ID.
-            transport: Arc::new(CountingTransport::new(
-                Arc::new(FirewallTransport::new(transport, Arc::clone(&firewall))),
-                Arc::clone(&traffic),
-            )),
+            // Counted above the firewall (in the paths layer, under the identity layer):
+            // what it drops outbound counts as sent (it plays a lossy network), what it
+            // drops inbound never arrived. Without the identity layer (tests) there is
+            // no firewall.
+            transport: Arc::new(CountingTransport::new(transport, Arc::clone(&traffic))),
             firewall,
             traffic,
             stats: Default::default(),
diff --git a/src/state/host.rs b/src/state/host.rs
index 441139e..6eab803 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -161,14 +161,24 @@ impl Host for AppState {
     fn resume(&self) -> impl Future<Output = ()> + Send {
         let state = self.clone();
         async move {
-            let mut lifecycle = state.lifecycle.write().await;
-            if let NodeLifecycle::Joining {
-                discovered_swarm_id: Some(swarm_id),
-            } = &*lifecycle
-            {
-                let swarm_id = swarm_id.clone();
-                debug!(swarm_id = %swarm_id, "Node resumed");
-                *lifecycle = NodeLifecycle::Ready { swarm_id };
+            let resumed = {
+                let mut lifecycle = state.lifecycle.write().await;
+                let swarm_id = match &*lifecycle {
+                    NodeLifecycle::Joining {
+                        discovered_swarm_id: Some(swarm_id),
+                    } => Some(swarm_id.clone()),
+                    _ => None,
+                };
+                if let Some(swarm_id) = &swarm_id {
+                    debug!(swarm_id = %swarm_id, "Node resumed");
+                    *lifecycle = NodeLifecycle::Ready {
+                        swarm_id: swarm_id.clone(),
+                    };
+                }
+                swarm_id.is_some()
+            };
+            if resumed {
+                crate::swarm::peer::announce(&state);
             }
         }
     }
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 20f92c5..d8291a7 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -24,7 +24,7 @@ mod read;
 mod session;
 mod timing;
 
-pub(crate) use peer::{HelloState, Incidents, Liveness, PeerState, Standing};
+pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing};
 pub(crate) use read::{ReadSource, StaleView};
 pub(crate) use session::SyncSession;
 pub(crate) use timing::Timing;
diff --git a/src/state/peer.rs b/src/state/peer.rs
index c9356ea..1f08767 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 keepalive STATUSes. Fixed for
-    /// the pair, so computed once.
+    /// Whether we lead the pair (`swarm::pair`): we send the keepalives. Fixed for the
+    /// pair, so computed once.
     pub(crate) leads: bool,
 
     pub(crate) standing: Standing,
@@ -85,6 +85,9 @@ pub(crate) struct PeerState {
     pub(crate) liveness: Liveness,
     pub(crate) added_at: Instant,
     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`).
     pub(crate) last_status_at: Option<Instant>,
 
     /// Last time a sync message from this peer was applied (survives resets: the
@@ -133,7 +136,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 STATUS_REPLY answers a STATUS.
+/// 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 +161,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,6 +185,7 @@ impl PeerState {
             liveness: Liveness::Active,
             added_at: Instant::now(),
             last_seen: None,
+            last_keepalive_at: None,
             last_status_at: None,
             last_sync_rx: None,
             last_resync_at: None,
@@ -287,16 +301,21 @@ impl PeerState {
         true
     }
 
-    /// True (and records the time) 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 we sent it none for `status_refresh`.
     ///
-    /// One 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 STATUSes
-    /// 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
-    /// STATUSes would often cross.
-    pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
+    /// 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.
+    pub(crate) fn take_keepalive_due(
+        &mut self,
+        timing: &Timing,
+        now: Instant,
+    ) -> Option<Keepalive> {
         let quiet_enough = if self.leads {
             timing.keepalive
         } else {
@@ -307,16 +326,24 @@ impl PeerState {
             && now.saturating_duration_since(self.added_at) >= timing.keepalive
             && self.quiet_for(now) >= quiet_enough
             && self
-                .last_status_at
+                .last_keepalive_at
                 .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
-
-        if due {
-            self.last_status_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 STATUS/STATUS_REPLY. 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 +405,33 @@ impl PeerState {
 mod tests {
     use super::*;
 
+    /// 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]
diff --git a/src/state/timing.rs b/src/state/timing.rs
index 6ce5bef..9055ed6 100644
--- a/src/state/timing.rs
+++ b/src/state/timing.rs
@@ -18,6 +18,10 @@ const PROBE_MAX_FACTOR: u32 = 6;
 #[derive(Clone, Copy, Debug)]
 pub(crate) struct Timing {
     pub(crate) keepalive: Duration,
+    /// How often a quiet pair's keepalive is a STATUS rather than a PING
+    /// (`PeerState::take_keepalive_due`): its state (heads, readiness, epoch) for the heal
+    /// and against lost announcements. Changes go out as they happen anyway.
+    pub(crate) status_refresh: Duration,
     pub(crate) dormant_after: Duration,
     pub(crate) resync_after: Duration,
     pub(crate) dormant_probe: Duration,
@@ -43,8 +47,9 @@ impl Timing {
     ///
     /// Each node keeps about `k` neighbours (`--neighbours`) whatever the swarm's size,
     /// and only one side of a pair sends them, so idle, a node sends about `k` messages and
-    /// receives about `k` (STATUSes and STATUS_REPLYs) per keepalive interval (more on a stable
-    /// member others prefer). The interval trades that against how soon a silent peer is
+    /// receives about `k` per keepalive interval (more on a stable member others prefer):
+    /// PINGs and PONGs (a few bytes), every third a STATUS and STATUS_REPLY (heads,
+    /// readiness, epoch) instead. The interval trades that against how soon a silent peer is
     /// noticed (`dormant_after`, 4 intervals) and a gap closes (within an interval).
     /// Recommended:
     ///
@@ -68,7 +73,10 @@ impl Timing {
     ) -> Self {
         Self {
             keepalive,
-            dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost STATUSes in a row
+            // Every third keepalive; with `resync_after` at two keepalives, a heal needs
+            // two STATUSes that differ: within about two refreshes.
+            status_refresh: keepalive * 3,
+            dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost keepalives in a row
             // >= 2 keepalives, so a second exchange confirms the difference; and >= twice
             // the first sync reply timeout, so a lost message's sender resends it first.
             resync_after: (keepalive * 2).max(sync_reply_timeout * 2),
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 7c4c444..a9ab1b2 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -119,6 +119,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
             if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() {
                 state.mark_ready(swarm_id.clone()).await?;
                 info!(swarm_id = %swarm_id, "Node is ready");
+                crate::swarm::peer::announce(state);
                 return Ok(());
             }
         }
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 {
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs
index 01d10ee..466fc98 100644
--- a/src/swarm/rtt.rs
+++ b/src/swarm/rtt.rs
@@ -1,9 +1,11 @@
-//! Round-trip times to peers, measured from traffic we send anyway: STATUS → STATUS_REPLY and
-//! HELLO → HELLO_REPLY. No probes of our own.
+//! Round-trip times to peers, measured from traffic we send anyway: the keepalive PING
+//! → PONG (`swarm::peer::keepalive`), and HELLO → HELLO_REPLY.
 //!
-//! A sample only counts when exactly one request was outstanding: with two, we can't tell
-//! which one the answer belongs to (an answer to the older one would look too fast).
-//! Samples are smoothed (exponential moving average, weight 1/4) to ride out jitter.
+//! A PONG answers the PING with its nonce, so it is attributed exactly; a PONG to an
+//! older PING (one a newer one replaced) is no sample. A HELLO_REPLY only counts when
+//! exactly one HELLO was outstanding: with two, we can't tell which one the answer
+//! belongs to (an answer to the older one would look too fast). Samples are smoothed
+//! (exponential moving average, weight 1/4) to ride out jitter.
 //!
 //! What we use is the *effective* round-trip time: the smoothed value, raised to the age
 //! of the oldest request still unanswered (the round trip takes at least that long), so
@@ -18,11 +20,20 @@ use tokio::time::Instant;
 
 #[derive(Debug, Default)]
 pub(crate) struct RttTable {
-    statuses: HashMap<String, Outstanding>,
+    pings: HashMap<String, Ping>,
     hellos: HashMap<String, Outstanding>,
     smoothed: HashMap<String, Sample>,
 }
 
+/// The PING out to a peer.
+#[derive(Debug, Clone, Copy)]
+struct Ping {
+    nonce: u64,
+    sent: Instant,
+    /// When the oldest PING still unanswered went out (an earlier one, replaced).
+    waiting_since: Instant,
+}
+
 #[derive(Debug, Clone, Copy)]
 struct Sample {
     rtt: Duration,
@@ -36,13 +47,26 @@ struct Outstanding {
 }
 
 impl RttTable {
-    pub(crate) fn status_sent(&mut self, peer: &str, now: Instant) {
-        Self::sent(&mut self.statuses, peer, now);
+    pub(crate) fn ping_sent(&mut self, peer: &str, nonce: u64, now: Instant) {
+        let waiting_since = self
+            .pings
+            .get(peer)
+            .map_or(now, |earlier| earlier.waiting_since);
+        self.pings.insert(
+            peer.to_owned(),
+            Ping {
+                nonce,
+                sent: now,
+                waiting_since,
+            },
+        );
     }
 
-    pub(crate) fn status_reply_received(&mut self, peer: &str, now: Instant) {
-        if let Some(sample) = Self::answered(&mut self.statuses, peer, now) {
-            self.record(peer, sample, now);
+    pub(crate) fn pong_received(&mut self, peer: &str, nonce: u64, now: Instant) {
+        if self.pings.get(peer).is_some_and(|ping| ping.nonce == nonce)
+            && let Some(ping) = self.pings.remove(peer)
+        {
+            self.record(peer, now.saturating_duration_since(ping.sent), now);
         }
     }
 
@@ -68,12 +92,15 @@ impl RttTable {
             .smoothed
             .get(peer)
             .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?;
-        let waiting = [self.statuses.get(peer), self.hellos.get(peer)]
-            .into_iter()
-            .flatten()
-            .map(|outstanding| now.saturating_duration_since(outstanding.sent))
-            .max()
-            .unwrap_or_default();
+        let waiting = [
+            self.pings.get(peer).map(|ping| ping.waiting_since),
+            self.hellos.get(peer).map(|outstanding| outstanding.sent),
+        ]
+        .into_iter()
+        .flatten()
+        .map(|sent| now.saturating_duration_since(sent))
+        .max()
+        .unwrap_or_default();
         Some(sample.rtt.max(waiting))
     }
 
@@ -128,31 +155,47 @@ mod tests {
         let mut table = RttTable::default();
         let start = Instant::now();
         let at = |ms| start + Duration::from_millis(ms);
-        table.status_sent("a", start);
-        table.status_reply_received("a", at(100));
+        table.ping_sent("a", 1, start);
+        table.pong_received("a", 1, at(100));
         assert_eq!(
             table.effective("a", at(100), FRESH),
             Some(Duration::from_millis(100))
         );
 
-        table.status_sent("a", at(100));
-        table.status_reply_received("a", at(300));
+        table.ping_sent("a", 2, at(100));
+        table.pong_received("a", 2, at(300));
         assert_eq!(
             table.effective("a", at(300), FRESH),
             Some(Duration::from_millis(125))
         );
 
-        // Two STATUSes out: the STATUS_REPLY can't be attributed.
-        table.status_sent("a", at(300));
-        table.status_sent("a", at(400));
-        table.status_reply_received("a", at(450));
+        // A newer PING replaced it: a PONG to the older one is no sample.
+        table.ping_sent("a", 3, at(300));
+        table.ping_sent("a", 4, at(400));
+        table.pong_received("a", 3, at(410));
         assert_eq!(
-            table.effective("a", at(450), FRESH),
+            table.effective("a", at(410), FRESH),
             Some(Duration::from_millis(125))
         );
+        table.pong_received("a", 4, at(500));
+        assert_eq!(
+            table.effective("a", at(500), FRESH),
+            Some(Duration::from_micros(118_750)) // (125 * 3 + 100) / 4
+        );
+
+        // Two HELLOs out: the HELLO_REPLY can't be attributed.
+        table.hello_sent("c", start);
+        table.hello_reply_received("c", at(100));
+        table.hello_sent("c", at(100));
+        table.hello_sent("c", at(200));
+        table.hello_reply_received("c", at(210));
+        assert_eq!(
+            table.effective("c", at(210), FRESH),
+            Some(Duration::from_millis(100))
+        );
 
-        // An unsolicited STATUS_REPLY changes nothing.
-        table.status_reply_received("b", start);
+        // An unsolicited PONG changes nothing.
+        table.pong_received("b", 9, start);
         assert_eq!(table.effective("b", start, FRESH), None);
     }
 
@@ -160,12 +203,13 @@ mod tests {
     fn unanswered_requests_raise_it_and_old_samples_expire() {
         let mut table = RttTable::default();
         let start = Instant::now();
-        table.status_sent("a", start);
-        table.status_reply_received("a", start + Duration::from_millis(50));
+        table.ping_sent("a", 1, start);
+        table.pong_received("a", 1, start + Duration::from_millis(50));
 
-        // A STATUS unanswered for 40 s: at least 40 s.
+        // PINGs unanswered for 40 s: at least 40 s, from the oldest.
         let sent = start + Duration::from_secs(60);
-        table.status_sent("a", sent);
+        table.ping_sent("a", 2, sent);
+        table.ping_sent("a", 3, sent + Duration::from_secs(30));
         let later = sent + Duration::from_secs(40);
         assert_eq!(
             table.effective("a", later, FRESH),
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs
index 846ae1b..7b0ec0d 100644
--- a/src/transport/firewall.rs
+++ b/src/transport/firewall.rs
@@ -1,13 +1,14 @@
+//! The debug firewall (`/debug/firewall`, with `--debug-api`): drops messages from and
+//! to peers, by node ID, to test partitions on a real swarm. It works at the bottom of
+//! the node, in the paths layer (`transport::paths::Gate`), so it drops everything a
+//! partition would: WHO and ME, PINGs and their PONGs, as well as the node's messages.
+
 use std::collections::HashSet;
-use std::sync::{Arc, RwLock, RwLockReadGuard};
+use std::sync::{RwLock, RwLockReadGuard};
 
-use anyhow::Result;
-use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 
 use crate::locks::RwLockExt;
-use crate::transport::EncodedMessage;
-use crate::transport::{MessageHandler, ParsedMessage, Transport};
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub(crate) struct FirewallRules {
@@ -46,99 +47,3 @@ impl Firewall {
         self.read().outbound.contains(peer)
     }
 }
-
-pub(crate) struct FirewallTransport {
-    inner: Arc<dyn Transport>,
-    firewall: Arc<Firewall>,
-}
-
-impl FirewallTransport {
-    pub(crate) fn new(inner: Arc<dyn Transport>, firewall: Arc<Firewall>) -> Self {
-        Self { inner, firewall }
-    }
-}
-
-#[async_trait]
-impl Transport for FirewallTransport {
-    fn local_id(&self) -> &str {
-        self.inner.local_id()
-    }
-
-    fn is_peer_id(&self, id: &str) -> bool {
-        self.inner.is_peer_id(id)
-    }
-
-    fn delivery_limit(&self) -> std::time::Duration {
-        self.inner.delivery_limit()
-    }
-
-    fn local_addresses(&self) -> Vec<String> {
-        self.inner.local_addresses()
-    }
-
-    fn addresses_of(&self, peer: &str) -> Vec<String> {
-        self.inner.addresses_of(peer)
-    }
-
-    fn learn_addresses(&self, peer: &str, addresses: &[String]) {
-        self.inner.learn_addresses(peer, addresses)
-    }
-
-    fn can_reach(&self, peer: &str) -> bool {
-        self.inner.can_reach(peer)
-    }
-
-    fn discovered_seeds(&self) -> Vec<String> {
-        self.inner.discovered_seeds()
-    }
-
-    fn transports(&self) -> Vec<String> {
-        self.inner.transports()
-    }
-
-    fn destination(&self, peer: &str) -> Option<crate::transport::Destination> {
-        self.inner.destination(peer)
-    }
-
-    fn connection(&self, target: &str) -> Option<crate::transport::Connection> {
-        self.inner.connection(target)
-    }
-
-    fn stats(&self) -> crate::transport::stats::TransportStats {
-        self.inner.stats()
-    }
-
-    async fn resolve_seed(&self, seed: &str) -> Result<String> {
-        self.inner.resolve_seed(seed).await
-    }
-
-    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
-        self.inner.encode(peer, message)
-    }
-
-    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
-        if self.firewall.blocks_outbound(&message.peer) {
-            tracing::trace!(peer = %message.peer, "Firewall dropped outbound message");
-            return Ok(());
-        }
-        self.inner.send_encoded(message).await
-    }
-
-    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
-        let firewall = Arc::clone(&self.firewall);
-
-        self.inner
-            .start_consumer(Box::new(move |sender, message, payload_len| {
-                if firewall.blocks_inbound(&sender) {
-                    tracing::trace!(peer = %sender, "Firewall dropped inbound message");
-                    return Box::pin(async { Ok(()) });
-                }
-                handler(sender, message, payload_len)
-            }))
-            .await
-    }
-
-    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
-        self.inner.wait_for_exit().await
-    }
-}
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 5857fb0..153dacc 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -37,7 +37,8 @@ use tracing::{debug, info};
 
 use crate::identity::{self, NodeKey};
 use crate::locks::MutexExt;
-use crate::transport::paths::{PathHandler, Paths, scheme_of};
+use crate::transport::firewall::Firewall;
+use crate::transport::paths::{Gate, PathHandler, Paths, scheme_of};
 use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats};
 use crate::transport::traffic::Traffic;
 use crate::transport::{
@@ -96,6 +97,8 @@ pub(crate) struct IdentityTransport {
 
 struct Shared {
     paths: Arc<Paths>,
+    /// The debug firewall, by node ID; paths applies it (`Gate`).
+    firewall: Option<Arc<Firewall>>,
     key: Arc<NodeKey>,
     node_id: String,
     /// WHO and ME are counted here, under the address (`traffic`).
@@ -208,27 +211,54 @@ impl Book {
 
 impl IdentityTransport {
     /// Over `transport`, whose addresses have `config.scheme`, and `config.more`.
+    /// `firewall`: the debug firewall, if any, which paths applies by address.
     pub(crate) fn new(
         transport: Arc<dyn Transport>,
         config: IdentityConfig,
         traffic: Arc<Traffic>,
+        firewall: Option<Arc<Firewall>>,
     ) -> Self {
         let node_id = config.key.id();
-        let paths = Arc::new(Paths::new(
-            std::iter::once((config.scheme, transport))
-                .chain(config.more)
-                .collect(),
-            Some(Arc::clone(&traffic)),
-        ));
-        Self {
-            shared: Arc::new(Shared {
-                paths,
+        let shared = Arc::new_cyclic(|shared: &std::sync::Weak<Shared>| {
+            let gate: std::sync::Weak<dyn Gate> = shared.clone();
+            Shared {
+                paths: Arc::new(Paths::new(
+                    std::iter::once((config.scheme, transport))
+                        .chain(config.more)
+                        .collect(),
+                    Some(Arc::clone(&traffic)),
+                    firewall.is_some().then_some(gate),
+                )),
+                firewall,
                 key: config.key,
                 node_id,
                 traffic,
                 book: Mutex::default(),
-            }),
-        }
+            }
+        });
+        Self { shared }
+    }
+}
+
+/// The firewall's rules are by node ID: an address is blocked if the node proven at it
+/// is. Addresses no node proved itself at aren't (nothing could name them).
+impl Gate for Shared {
+    fn blocks_inbound(&self, address: &str) -> bool {
+        self.blocks(address, Firewall::blocks_inbound)
+    }
+
+    fn blocks_outbound(&self, address: &str) -> bool {
+        self.blocks(address, Firewall::blocks_outbound)
+    }
+}
+
+impl Shared {
+    fn blocks(&self, address: &str, rule: fn(&Firewall, &str) -> bool) -> bool {
+        let Some(firewall) = &self.firewall else {
+            return false;
+        };
+        let node = self.lock().at.get(address).map(|(node, _)| node.clone());
+        node.is_some_and(|node| rule(firewall, &node))
     }
 }
 
@@ -461,6 +491,11 @@ impl Transport for IdentityTransport {
         })
     }
 
+    /// The paths layer answers them (`Paths::receive`).
+    fn answers_pings(&self) -> bool {
+        true
+    }
+
     /// The identity layer's, and each transport's.
     fn stats(&self) -> TransportStats {
         let mut stats = self.shared.paths.stats();
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 147e560..f60a9f6 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -649,6 +649,13 @@ pub(crate) trait Transport: Send + Sync {
         None
     }
 
+    /// Whether the transport answers PINGs with PONGs itself (the paths layer,
+    /// `transport::paths`); if not (a bare transport, in tests), the node does. By
+    /// default not.
+    fn answers_pings(&self) -> bool {
+        false
+    }
+
     /// Figures about the transport itself (`stats`). By default none.
     fn stats(&self) -> stats::TransportStats {
         stats::TransportStats::default()
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 30aab8d..dda2d87 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -17,16 +17,24 @@
 //!
 //! Paths measures the addresses of such nodes itself: every `PROBE_EVERY`, each address
 //! it had a choice of within `ACTIVE_FOR` gets a PING with a random nonce, which the
-//! other side answers with a PONG at once, over the same path. Answers give each
+//! other side's paths answers at once with a PONG, over the same path. Answers give each
 //! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a
 //! send that failed within `DOWN_FOR`, make a path down. `rank` puts paths that are up
 //! first, the one in use first among them, and the rest by round-trip time; it changes
 //! the one in use when another is faster by `SWITCH_GAIN` (a quarter) and at least
 //! `SWITCH_MIN` (both measured), so it doesn't flap. Until both are measured, the
-//! configured order decides. PING and PONG end here: the identity layer never sees them.
+//! configured order decides.
+//!
+//! PINGs and PONGs then go up as other messages do: they tell the node the peer is
+//! alive. The node's keepalive is a PING too (`swarm::peer::keepalive`), which paths
+//! sends like any message, and notes as the probe of the path it went over.
+//!
+//! The debug firewall (`transport::firewall`) works here, below everything else, through
+//! a `Gate`: messages from an address it blocks are dropped as they arrive, and sends to
+//! one are dropped silently, as by a lossy network.
 
 use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, Weak};
 use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow};
@@ -36,7 +44,14 @@ use tracing::{debug, info};
 use crate::locks::MutexExt;
 use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats};
 use crate::transport::traffic::Traffic;
-use crate::transport::{Connection, EncodedMessage, MessageKind, ParsedMessage, Transport};
+use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport};
+
+/// Whether to drop what comes from or goes to an address (the debug firewall,
+/// `transport::firewall`; the identity layer knows which node is where).
+pub(crate) trait Gate: Send + Sync {
+    fn blocks_inbound(&self, address: &str) -> bool;
+    fn blocks_outbound(&self, address: &str) -> bool;
+}
 
 /// Called with the sender's address, the message and its `payload_len`.
 pub(crate) type PathHandler = dyn Fn(String, ParsedMessage, usize) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>
@@ -250,6 +265,16 @@ impl Table {
         due
     }
 
+    /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that
+    /// path, if paths measures it.
+    fn pinged(&mut self, address: &str, nonce: u64, now: Instant) {
+        if let Some(path) = self.paths.get_mut(address) {
+            path.probe = Some((nonce, now));
+            path.last_probe = Some(now);
+            self.counts.probes_sent += 1;
+        }
+    }
+
     /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out.
     fn pong(&mut self, address: &str, nonce: u64, now: Instant) -> Option<Duration> {
         let path = self.paths.get_mut(address)?;
@@ -298,18 +323,21 @@ impl Table {
 pub(crate) struct Paths {
     /// In order of preference.
     links: Vec<Link>,
-    /// PINGs and PONGs are counted here, under the address (`traffic`).
+    /// Its own PINGs and PONGs are counted here, under the address (`traffic`).
     traffic: Option<Arc<Traffic>>,
+    /// The debug firewall, if any.
+    gate: Option<Weak<dyn Gate>>,
     /// A std mutex: never held across an await.
     table: Mutex<Table>,
 }
 
 impl Paths {
-    /// Over `transports`, by scheme, in order of preference; PINGs and PONGs counted in
-    /// `traffic`.
+    /// Over `transports`, by scheme, in order of preference; its own PINGs and PONGs
+    /// counted in `traffic`, the firewall through `gate`.
     pub(crate) fn new(
         transports: Vec<(&'static str, Arc<dyn Transport>)>,
         traffic: Option<Arc<Traffic>>,
+        gate: Option<Weak<dyn Gate>>,
     ) -> Self {
         Self {
             links: transports
@@ -317,10 +345,15 @@ impl Paths {
                 .map(|(scheme, inner)| Link { scheme, inner })
                 .collect(),
             traffic,
+            gate,
             table: Mutex::default(),
         }
     }
 
+    fn gate(&self) -> Option<Arc<dyn Gate>> {
+        self.gate.as_ref()?.upgrade()
+    }
+
     /// The transport `address` is on, and the transport identity at it
     /// (`Transport::locate`, which notes where that is if the address says); None if no
     /// transport of ours has its scheme, or it isn't one of that transport's.
@@ -519,6 +552,9 @@ impl Paths {
                 Ok(()) => {
                     let mut table = self.table.lock_or_recover();
                     table.sent(address);
+                    if let Some(ParsedMessage::Ping(nonce)) = message.message.as_deref() {
+                        table.pinged(address, *nonce, Instant::now());
+                    }
                     if !failures.is_empty() {
                         table.counts.failovers += 1;
                         drop(table);
@@ -545,8 +581,15 @@ impl Paths {
         }
     }
 
-    /// Sends `message` to its `address`.
+    /// Sends `message` to its `address` (dropped silently if the firewall blocks it).
     pub(crate) async fn send(&self, message: &EncodedMessage) -> Result<()> {
+        if self
+            .gate()
+            .is_some_and(|gate| gate.blocks_outbound(&message.address))
+        {
+            tracing::trace!(to = %message.address, "Firewall dropped outbound message");
+            return Ok(());
+        }
         let (link, target) = self
             .split(&message.address)
             .with_context(|| format!("no transport for {}", message.address))?;
@@ -583,7 +626,8 @@ impl Paths {
         stats
     }
 
-    /// Sends a PING or PONG with `nonce` to `address`, counted (`traffic`).
+    /// Sends a PING (or with `pong`, a PONG) with `nonce` to `address`, counted
+    /// (`traffic`).
     async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> {
         let message = if pong {
             ParsedMessage::Pong(nonce)
@@ -622,8 +666,8 @@ impl Paths {
         }
     }
 
-    /// A message from `address`: a PING is answered, a PONG noted; anything else goes to
-    /// `handler`.
+    /// A message from `address`, handed on to `handler`: a PING answered and a PONG
+    /// noted first; dropped if the firewall blocks it.
     async fn receive(
         self: Arc<Self>,
         address: String,
@@ -631,29 +675,31 @@ impl Paths {
         payload_len: usize,
         handler: &PathHandler,
     ) -> Result<()> {
+        if self
+            .gate()
+            .is_some_and(|gate| gate.blocks_inbound(&address))
+        {
+            tracing::trace!(from = %address, "Firewall dropped inbound message");
+            return Ok(());
+        }
         match message {
             ParsedMessage::Ping(nonce) => {
-                if let Some(traffic) = &self.traffic {
-                    traffic.received(&address, MessageKind::Ping, payload_len);
-                }
+                let paths = Arc::clone(&self);
+                let to = address.clone();
                 tokio::spawn(async move {
-                    if let Err(error) = self.send_probe(&address, true, nonce).await {
-                        debug!(to = %address, error = %format!("{error:#}"), "Couldn't answer a PING");
+                    if let Err(error) = paths.send_probe(&to, true, nonce).await {
+                        debug!(to = %to, error = %format!("{error:#}"), "Couldn't answer a PING");
                     }
                 });
-                Ok(())
             }
             ParsedMessage::Pong(nonce) => {
-                if let Some(traffic) = &self.traffic {
-                    traffic.received(&address, MessageKind::Pong, payload_len);
-                }
                 self.table
                     .lock_or_recover()
                     .pong(&address, nonce, Instant::now());
-                Ok(())
             }
-            message => handler(address, message, payload_len).await,
+            _ => {}
         }
+        handler(address, message, payload_len).await
     }
 
     /// Every transport's consumer, each message handed to `handler` under the sender's
@@ -712,7 +758,7 @@ impl Paths {
 }
 
 /// A PING's nonce: random, so a PONG can't be made up for a PING not seen.
-fn random_nonce() -> u64 {
+pub(crate) fn random_nonce() -> u64 {
     let mut bytes = [0u8; 8];
     if getrandom::fill(&mut bytes).is_err() {
         // No randomness: a nonce still tells answers apart, if not unguessably.
@@ -764,6 +810,7 @@ mod tests {
                     ("other", second.transport(id) as Arc<dyn Transport>),
                 ],
                 None,
+                None,
             ))
         };
         (on_both("a"), on_both("b"))
@@ -896,6 +943,99 @@ mod tests {
         assert!(table.paths.is_empty() && table.chosen.is_empty());
     }
 
+    /// The node's keepalive PING on a path is its probe: paths doesn't ping it too, and
+    /// the PONG measures it.
+    #[test]
+    fn keepalive_pings_are_probes() {
+        let mut table = Table::default();
+        let now = Instant::now();
+        let mut nonce = 100;
+        let mut next = || {
+            nonce += 1;
+            nonce
+        };
+        table.rank(paths(&["libp2p:x", "mycelium:x"]), now);
+        table.pinged("libp2p:x", 7, now);
+        let due = table.due(now, &mut next);
+        assert_eq!(due.len(), 1);
+        assert_eq!(due[0].0, "mycelium:x");
+        assert!(
+            table
+                .pong("libp2p:x", 7, now + Duration::from_millis(20))
+                .is_some()
+        );
+        assert_eq!(table.counts.probes_sent, 2);
+        // A peer with one path isn't measured.
+        table.pinged("libp2p:z", 8, now);
+        assert!(!table.paths.contains_key("libp2p:z"));
+    }
+
+    /// Blocks one address, both ways.
+    struct Blocking(&'static str);
+
+    impl Gate for Blocking {
+        fn blocks_inbound(&self, address: &str) -> bool {
+            address == self.0
+        }
+        fn blocks_outbound(&self, address: &str) -> bool {
+            address == self.0
+        }
+    }
+
+    /// The firewall drops what comes from a blocked address before anything sees it
+    /// (a PING isn't answered), and sends to one silently.
+    #[tokio::test]
+    async fn the_firewall_drops_both_ways() {
+        let net = SimNet::new(4);
+        let gate: Arc<dyn Gate> = Arc::new(Blocking("sim:a"));
+        let b = Arc::new(Paths::new(
+            vec![("sim", net.transport("b") as Arc<dyn Transport>)],
+            None,
+            Some(Arc::downgrade(&gate)),
+        ));
+        let a = Arc::new(Paths::new(
+            vec![("sim", net.transport("a") as Arc<dyn Transport>)],
+            None,
+            None,
+        ));
+        let received = |paths: &Arc<Paths>| {
+            let (sender, received) = tokio::sync::mpsc::unbounded_channel();
+            let handler: Arc<PathHandler> = Arc::new(move |from, message, _| {
+                let sender = sender.clone();
+                Box::pin(async move {
+                    let _ = sender.send((from, message.kind()));
+                    Ok(())
+                })
+            });
+            let paths = Arc::clone(paths);
+            tokio::spawn(async move { paths.start_consumer(handler).await });
+            received
+        };
+        let (mut at_a, mut at_b) = (received(&a), received(&b));
+
+        // a's PING never reaches b, so nothing answers it.
+        let ping = a.encode("sim:b", &ParsedMessage::Ping(5)).unwrap();
+        a.send(&ping).await.unwrap();
+        // b's send to a "works", and goes nowhere.
+        let status = b.encode("sim:a", &some_status()).unwrap();
+        b.send(&status).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        assert!(at_b.try_recv().is_err());
+        assert!(at_a.try_recv().is_err());
+
+        // Without the gate: a PING is answered by b's paths, and goes on up there.
+        drop(gate);
+        a.send(&ping).await.unwrap();
+        assert_eq!(
+            at_b.recv().await.unwrap(),
+            ("sim:a".to_owned(), crate::transport::MessageKind::Ping)
+        );
+        assert_eq!(
+            at_a.recv().await.unwrap(),
+            ("sim:b".to_owned(), crate::transport::MessageKind::Pong)
+        );
+    }
+
     /// A transport whose sends all fail.
     struct Broken;
 
@@ -948,10 +1088,12 @@ mod tests {
                 ("sim", net.transport("a") as Arc<dyn Transport>),
             ],
             None,
+            None,
         );
         let b = Arc::new(Paths::new(
             vec![("sim", net.transport("b") as Arc<dyn Transport>)],
             None,
+            None,
         ));
         let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
         let handler: Arc<PathHandler> = Arc::new(move |from, _, _| {
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 9b5f9c9..b070023 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -5,8 +5,8 @@
 //! (for Mycelium: the topic, or the base64 of its HTTP API). Counters only grow (from
 //! the start of the process); rates are for whoever reads them.
 //!
-//! A message dropped by the debug firewall still counts as sent (the firewall plays a
-//! lossy network) but not as received.
+//! A message dropped by the debug firewall (in the paths layer, below) still counts as
+//! sent (the firewall plays a lossy network) but not as received.
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex, MutexGuard};
@@ -188,6 +188,10 @@ impl Transport for CountingTransport {
         self.inner.stats()
     }
 
+    fn answers_pings(&self) -> bool {
+        self.inner.answers_pings()
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }