fix(transport): sample RTT from ping/pong keep-alive cycle

## Problem

`TransportMetrics::record_rtt_sample` was only reachable from
`record_transfer_completed`, which fires on stream-transfer completion.
Connections that exchange only keep-alive / control / small-contract
traffic for hours never complete a stream transfer, so they contributed
zero RTT samples. The dashboard / telemetry RTT statistics
(`min_rtt_us`, `max_rtt_us`, `avg_rtt_us`) under-represented quiet
connections. Same class of undercounting as the per-peer SENT/RECV bug
fixed in #3996.

## Approach

The ping/pong keep-alive cycle already records each Ping's send
timestamp in `pending_pings` (sequence -> nanos). The Pong handler
already removed the matching entry but discarded the timestamp. Capture
it, compute the round-trip against the same `time_source`, and record an
RTT sample. The keep-alive cycle runs on every connection regardless of
stream traffic, so RTT statistics now stay populated for quiet,
long-lived connections.

- `record_rtt_sample` widened from private to `pub(crate)` so the
  connection layer can call it (it was already the shared sink for the
  stream-completion path).
- The `rtt_us > 0` guard mirrors the existing stream-completion path.
- The write lock on `pending_pings` is released before recording the
  sample.

This is observation-only: it does not feed congestion control (the
shadow-RTT system in #4074 remains the path for that).

## Testing

New regression test `pong_records_rtt_sample_from_keepalive_cycle`
drives an encrypted Pong through the real `recv()` path after seeding a
pending ping, and asserts the global RTT sample counter advances and the
ping is consumed. Verified it FAILS without the fix (records zero
samples) and PASSES with it. Full transport lib suite (607 tests) green;
`cargo clippy --locked -- -D warnings` clean.

Closes #4000

[AI-assisted - Claude]

Author: sanity

crates/core/src/transport/metrics.rs

@@ -255,7 +255,13 @@ impl TransportMetrics {
     }
 
     /// Record an RTT sample in microseconds.
-    fn record_rtt_sample(&self, rtt_us: u64) {
+    ///
+    /// Called both from [`Self::record_transfer_completed`] (LEDBAT
+    /// `base_delay` at stream completion) and from the ping/pong keep-alive
+    /// cycle in `peer_connection.rs` (round-trip of a keep-alive Ping). The
+    /// keep-alive path is what keeps RTT statistics populated for quiet,
+    /// long-lived connections that rarely complete a stream transfer (#4000).
+    pub(crate) fn record_rtt_sample(&self, rtt_us: u64) {
         self.rtt_sum_us
             .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
                 Some(v.saturating_add(rtt_us))
@@ -290,6 +296,16 @@ impl TransportMetrics {
         self.cumulative_bytes_sent.load(Ordering::Relaxed)
     }
 
+    /// Number of RTT samples accumulated in the current snapshot window.
+    ///
+    /// Reset to 0 by `take_snapshot`. Exposed so tests can assert that a
+    /// code path (e.g. the ping/pong keep-alive cycle) actually recorded a
+    /// sample without consuming the snapshot.
+    #[cfg(test)]
+    pub(crate) fn rtt_sample_count(&self) -> u32 {
+        self.rtt_samples.load(Ordering::Relaxed)
+    }
+
     /// Record a completed inbound stream transfer.
     ///
     /// Aggregates stream-payload bytes into the per-snapshot `bytes_received`

crates/core/src/transport/peer_connection.rs

@@ -972,16 +972,42 @@ impl<S: super::Socket, T: TimeSource> PeerConnection<S, T> {
                                 pong_sequence = sequence,
                                 "Received Pong, confirming bidirectional liveness"
                             );
-                            // Remove the corresponding ping from pending set
-                            let mut pending = self.pending_pings.write();
-                            if pending.remove(sequence).is_some() {
-                                tracing::trace!(
-                                    target: "freenet_core::transport::keepalive_received",
-                                    remote = ?self.remote_conn.remote_addr,
-                                    pong_sequence = sequence,
-                                    remaining_pending = pending.len(),
-                                    "Removed acknowledged ping from pending set"
-                                );
+                            // Remove the corresponding ping from pending set and,
+                            // if found, sample the round-trip time. The keep-alive
+                            // cycle runs on every connection regardless of stream
+                            // traffic, so this keeps RTT statistics populated for
+                            // quiet, long-lived connections that rarely (or never)
+                            // complete a stream transfer (#4000). Without it,
+                            // RTT was only sampled at stream completion and quiet
+                            // connections contributed zero samples.
+                            let removed_send_nanos = {
+                                let mut pending = self.pending_pings.write();
+                                let removed = pending.remove(sequence);
+                                if removed.is_some() {
+                                    tracing::trace!(
+                                        target: "freenet_core::transport::keepalive_received",
+                                        remote = ?self.remote_conn.remote_addr,
+                                        pong_sequence = sequence,
+                                        remaining_pending = pending.len(),
+                                        "Removed acknowledged ping from pending set"
+                                    );
+                                }
+                                removed
+                            };
+                            if let Some(send_nanos) = removed_send_nanos {
+                                // The ping timestamp was recorded with this same
+                                // `time_source` (the keep-alive task clones it), so
+                                // the subtraction is over one monotonic clock.
+                                // `saturating_sub` is defensive against any skew
+                                // yielding a 0 sample rather than underflowing.
+                                let rtt_nanos = self
+                                    .time_source
+                                    .now_nanos()
+                                    .saturating_sub(send_nanos);
+                                let rtt_us = rtt_nanos / 1_000;
+                                if rtt_us > 0 {
+                                    super::TRANSPORT_METRICS.record_rtt_sample(rtt_us);
+                                }
                             }
                         }
                         SymmetricMessagePayload::AckConnection { .. } | SymmetricMessagePayload::ShortMessage { .. } | SymmetricMessagePayload::StreamFragment { .. } => {}
@@ -3032,6 +3058,136 @@ mod tests {
         crate::transport::TRANSPORT_METRICS.remove_peer(remote_addr);
     }
 
+    /// Regression test for #4000: a Pong received for a pending keep-alive
+    /// Ping must record an RTT sample, so quiet connections that never
+    /// complete a stream transfer still contribute to the RTT statistics.
+    ///
+    /// Before the fix, the Pong handler removed the matching ping timestamp
+    /// from `pending_pings` and discarded it — `record_rtt_sample` was only
+    /// reachable from `record_transfer_completed` (stream completion), so a
+    /// connection exchanging only keep-alive traffic recorded zero RTT
+    /// samples. This test seeds a pending ping, advances the (mock) clock,
+    /// feeds an encrypted Pong, and asserts the global RTT sample counter
+    /// moved by at least one.
+    #[tokio::test]
+    async fn pong_records_rtt_sample_from_keepalive_cycle() {
+        use crate::transport::crypto::TransportKeypair;
+        use crate::transport::packet_data::PacketData;
+        use crate::transport::symmetric_message::SymmetricMessagePayload;
+        use crate::util::time_source::SharedMockTimeSource;
+        use bytes::Bytes;
+
+        let time_source = SharedMockTimeSource::new();
+        let (inbound_tx, inbound_rx) = mpsc::channel(16);
+        let remote_addr = SocketAddr::new(Ipv4Addr::new(10, 99, 99, 2).into(), 50002);
+
+        let mut key = [0u8; 16];
+        crate::config::GlobalRng::fill_bytes(&mut key);
+        let cipher = Aes128Gcm::new(&key.into());
+        let keypair = TransportKeypair::new();
+
+        let sent_tracker = Arc::new(parking_lot::Mutex::new(
+            SentPacketTracker::new_with_time_source(time_source.clone()),
+        ));
+        let congestion_controller =
+            crate::transport::congestion_control::CongestionControlConfig::default()
+                .build_arc_with_time_source(time_source.clone());
+        let token_bucket = Arc::new(TokenBucket::new_with_time_source(
+            10_000,
+            10_000_000,
+            time_source.clone(),
+        ));
+        let socket = Arc::new(TestSocket::new(
+            mpsc::channel::<(SocketAddr, Arc<[u8]>)>(16).0,
+        ));
+
+        let rolling_rtt_stats = crate::transport::rolling_rtt_stats::RollingRttStatsHandle::new(
+            remote_addr,
+            time_source.clone(),
+        );
+        let remote_conn = RemoteConnection {
+            outbound_symmetric_key: cipher.clone(),
+            remote_addr,
+            sent_tracker,
+            last_packet_id: Arc::new(AtomicU32::new(0)),
+            inbound_packet_recv: inbound_rx,
+            inbound_symmetric_key: cipher.clone(),
+            inbound_symmetric_key_bytes: key,
+            my_address: None,
+            transport_secret_key: keypair.secret,
+            congestion_controller,
+            token_bucket,
+            socket,
+            global_bandwidth: None,
+            rolling_rtt_stats,
+            time_source: time_source.clone(),
+        };
+
+        // Build the encrypted Pong (sequence 7) and a trailing ShortMessage.
+        // The Pong handler does not make `recv()` return (it produces no
+        // message), so the ShortMessage gives the loop a value to yield once
+        // the Pong has been processed.
+        let pong_seq = 7u64;
+        let pong = SymmetricMessage::serialize_msg_to_packet_data(
+            1,
+            SymmetricMessagePayload::Pong { sequence: pong_seq },
+            &cipher,
+            vec![],
+        )
+        .expect("encrypt pong");
+        let pong_packet =
+            PacketData::<crate::transport::packet_data::UnknownEncryption>::from_buf(pong.data());
+
+        let short = SymmetricMessage::serialize_msg_to_packet_data(
+            2,
+            SymmetricMessagePayload::ShortMessage {
+                payload: Bytes::from_static(b"done"),
+            },
+            &cipher,
+            vec![],
+        )
+        .expect("encrypt short");
+        let short_packet =
+            PacketData::<crate::transport::packet_data::UnknownEncryption>::from_buf(short.data());
+
+        let mut conn = PeerConnection::new(remote_conn);
+
+        // Advance past zero before seeding so the ping's send timestamp is
+        // strictly positive. The recv loop's stale-ping cleanup retains only
+        // pings with `sent_at_nanos > now_nanos - idle_timeout`; at small mock
+        // times that threshold saturates to 0, so a ping stamped at exactly 0
+        // would be pruned before the Pong arrives. Stamp at 10 ms, then
+        // advance another 50 ms so the round-trip is a deterministic 50 ms.
+        time_source.advance_time(Duration::from_millis(10));
+        let send_nanos = time_source.now_nanos();
+        conn.pending_pings.write().insert(pong_seq, send_nanos);
+        time_source.advance_time(Duration::from_millis(50));
+
+        let samples_before = crate::transport::TRANSPORT_METRICS.rtt_sample_count();
+
+        inbound_tx.send(pong_packet).await.expect("send pong");
+        inbound_tx.send(short_packet).await.expect("send short");
+
+        let _msg = conn.recv().await.expect("recv");
+
+        let samples_after = crate::transport::TRANSPORT_METRICS.rtt_sample_count();
+        assert!(
+            samples_after >= samples_before + 1,
+            "Pong for a pending ping must record at least one RTT sample \
+             (before={samples_before}, after={samples_after}). \
+             The keep-alive RTT path (#4000) is not wired up."
+        );
+
+        // The matching ping must have been consumed so it isn't sampled twice
+        // or treated as still-unanswered by the keep-alive backoff.
+        assert!(
+            !conn.pending_pings.read().contains_key(&pong_seq),
+            "Pong must remove the matching pending ping"
+        );
+
+        crate::transport::TRANSPORT_METRICS.remove_peer(remote_addr);
+    }
+
     /// Regression test for #4079.
     ///
     /// Reproduces both leak scenarios called out in the issue: