feat(net): expose transport statistics

crates/net/src/chunk.rs

@@ -2,12 +2,15 @@
 
 //! Chunk-stream transport: the per-connection task that pumps chunk subscriptions and deliveries.
 
+use std::sync::Arc;
+
 use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe};
 use tokio::sync::mpsc::{Sender, UnboundedReceiver, UnboundedSender};
 use tracing::{debug, warn};
 
 use crate::codec::{MAX_CHUNK_FRAME_LEN, read_frame, write_frame};
 use crate::runtime::ServerEvent;
+use crate::stats::NetCounters;
 
 /// A synchronous handle the simulation loop uses to hand [`ChunkMessage`]s to a connection's chunk-stream task.
 ///
@@ -123,6 +126,7 @@ pub(crate) async fn client_chunk_task(
     connection: quinn::Connection,
     mut subscribe: UnboundedReceiver<ChunkSubscribe>,
     deliveries: Sender<ChunkMessage>,
+    counters: &Arc<NetCounters>,
 ) {
     // The client opens the chunk stream after the handshake; the server accepts it, mirroring the control-stream convention.
     let (mut send, mut recv) = match connection.open_bi().await {
@@ -138,6 +142,11 @@ pub(crate) async fn client_chunk_task(
         loop {
             match read_frame::<ChunkMessage>(&mut recv, MAX_CHUNK_FRAME_LEN).await {
                 Ok(message) => {
+                    // Counted on arrival rather than on delivery to the UI, so the figure reflects what the transport received even while the UI thread is backpressuring below.
+                    match message {
+                        ChunkMessage::Chunk { .. } => counters.record_chunk(),
+                        ChunkMessage::Drop { .. } => counters.record_drop(),
+                    }
                     // `send` awaits when the delivery channel is full: the task suspends (yielding the runtime thread so the connection keeps ACKing) until the UI drains a slot, and until then reads no further frames, which backpressures the server via QUIC stream flow control. An error means the UI dropped its receiver, so the session ends.
                     if deliveries.send(message).await.is_err() {
                         break;
@@ -159,6 +168,7 @@ pub(crate) async fn client_chunk_task(
                 warn!(%error, "failed to write chunk subscribe; ending chunk stream");
                 break;
             }
+            counters.record_subscribe();
         }
     };
 

crates/net/src/lib.rs

@@ -12,11 +12,13 @@ pub mod endpoint;
 pub mod error;
 pub mod handshake;
 pub mod runtime;
+pub mod stats;
 
 pub use chunk::{ChunkSink, ChunkSubscriber};
 pub use runtime::{
     ChunkStream, ClientLink, ConnectOutcome, NetworkServer, ServerEvent, connect_in_background,
 };
+pub use stats::NetStats;
 
 /// Default UDP port the server binds and the client connects to when none is configured.
 // TODO: make the bind address and port configurable through server/client configuration.

crates/net/src/runtime.rs

@@ -3,6 +3,7 @@
 //! Threaded `tokio` runtime bridge between the async network and the synchronous simulation.
 
 use std::net::SocketAddr;
+use std::sync::{Arc, OnceLock};
 use std::thread;
 
 use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe};
@@ -13,6 +14,7 @@ use crate::chunk::{ChunkSink, ChunkSubscriber, chunk_stream_task, client_chunk_t
 use crate::endpoint::{client_endpoint, server_endpoint};
 use crate::error::NetError;
 use crate::handshake::{ServerConnection, accept_connection, connect};
+use crate::stats::{self, NetCounters, NetStats};
 
 /// Channel receiver delivering the outcome of a background client connect: the negotiated [`HandshakeAck`] on success, or a human-readable error string on failure.
 pub type ConnectOutcome = crossbeam_channel::Receiver<Result<HandshakeAck, String>>;
@@ -34,6 +36,20 @@ pub struct ClientLink {
     pub subscribe: ChunkSubscriber,
     /// Receives chunk deliveries from the server, drained non-blocking each frame.
     pub chunks: ChunkStream,
+    /// The live QUIC connection, published by the network thread once the handshake completes. Held privately so the `quinn` types stay inside this crate; the UI thread reads through [`ClientLink::stats`].
+    connection: Arc<OnceLock<quinn::Connection>>,
+    /// Application-level message counters shared with the chunk task.
+    counters: Arc<NetCounters>,
+}
+
+impl ClientLink {
+    /// Snapshots the connection's transport statistics.
+    ///
+    /// Safe to call before the handshake completes; the result then reports the disconnected state rather than failing.
+    #[must_use]
+    pub fn stats(&self) -> NetStats {
+        stats::snapshot(self.connection.get(), &self.counters)
+    }
 }
 
 /// An event surfaced by the network thread to the synchronous server loop.
@@ -257,6 +273,11 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli
     let (outcome_tx, outcome_rx) = crossbeam_channel::bounded(1);
     // Retained so a failure to spawn the thread can still be reported to the caller.
     let spawn_err_tx = outcome_tx.clone();
+    // Published by the network thread once the handshake succeeds, so the UI thread can read `quinn`'s own connection statistics without owning the connection.
+    let connection = Arc::new(OnceLock::new());
+    let task_connection = Arc::clone(&connection);
+    let counters = Arc::new(NetCounters::default());
+    let task_counters = Arc::clone(&counters);
     // Subscription updates flow UI -> network (sync send, async recv); chunk deliveries flow network -> UI (async send, sync try_recv).
     let (subscribe_tx, subscribe_rx) = tokio::sync::mpsc::unbounded_channel::<ChunkSubscribe>();
     // The delivery channel is bounded so a slow (e.g. debug-build) UI thread applies backpressure to the network task instead of letting undelivered chunks accumulate without limit.
@@ -292,8 +313,16 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli
                         if outcome_tx.send(Ok(connected.ack.clone())).is_err() {
                             return;
                         }
+                        // The cell is written exactly once, here; a failure would mean a second handshake on one link, which cannot occur.
+                        let _ = task_connection.set(connected.connection.clone());
                         // Pump the chunk stream on this thread until the UI drops its handles or the server closes the connection.
-                        client_chunk_task(connected.connection, subscribe_rx, chunks_tx).await;
+                        client_chunk_task(
+                            connected.connection,
+                            subscribe_rx,
+                            chunks_tx,
+                            &task_counters,
+                        )
+                        .await;
                         warn!("server connection closed");
                     }
                     Err(error) => {
@@ -311,5 +340,7 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli
         handshake: outcome_rx,
         subscribe: ChunkSubscriber::new(subscribe_tx),
         chunks: chunks_rx,
+        connection,
+        counters,
     }
 }

crates/net/src/stats.rs

@@ -0,0 +1,103 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//! Transport statistics for a client connection.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+/// Application-level message counters, incremented by the client's chunk task and read by the UI thread.
+///
+/// Held behind an [`Arc`] and mutated with relaxed atomics: each counter is independent, nothing else is ordered against them, and a reader that observes a slightly stale value is reporting a diagnostic figure, not making a decision.
+#[derive(Debug, Default)]
+pub(crate) struct NetCounters {
+    /// Chunk deliveries received from the server.
+    chunks_received: AtomicU64,
+    /// Chunk drop notifications received from the server.
+    drops_received: AtomicU64,
+    /// Subscription updates written to the server.
+    subscribes_sent: AtomicU64,
+}
+
+impl NetCounters {
+    /// Records one received chunk delivery.
+    pub(crate) fn record_chunk(&self) {
+        self.chunks_received.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Records one received chunk drop notification.
+    pub(crate) fn record_drop(&self) {
+        self.drops_received.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Records one subscription update written to the server.
+    pub(crate) fn record_subscribe(&self) {
+        self.subscribes_sent.fetch_add(1, Ordering::Relaxed);
+    }
+}
+
+/// A snapshot of one client connection's transport state.
+///
+/// Every field is a value copied at the moment of the call. A snapshot taken before the handshake completes reports `connected == false` and zeroes throughout, which is a meaningful state rather than missing data.
+#[derive(Copy, Clone, Debug, Default, PartialEq)]
+pub struct NetStats {
+    /// Whether a QUIC connection is currently established.
+    pub connected: bool,
+    /// Bytes carried in UDP datagrams sent on this connection, including QUIC framing and retransmissions.
+    pub bytes_sent: u64,
+    /// Bytes carried in UDP datagrams received on this connection, including QUIC framing.
+    pub bytes_received: u64,
+    /// UDP datagrams sent on this connection.
+    pub datagrams_sent: u64,
+    /// UDP datagrams received on this connection.
+    pub datagrams_received: u64,
+    /// Chunk deliveries received since the connection was established.
+    pub chunks_received: u64,
+    /// Chunk drop notifications received since the connection was established.
+    pub drops_received: u64,
+    /// Subscription updates written to the server since the connection was established.
+    pub subscribes_sent: u64,
+    /// The QUIC stack's current round-trip-time estimate, in milliseconds.
+    pub rtt_ms: f32,
+    /// Packets the congestion controller has declared lost on the current path.
+    pub lost_packets: u64,
+    /// Current congestion window, in bytes: the ceiling on data in flight.
+    pub congestion_window: u64,
+    /// Largest UDP payload the current path is known to carry, in bytes, as discovered by path MTU probing.
+    pub path_mtu: u16,
+}
+
+/// Reads a connection's transport statistics, or reports the disconnected state.
+///
+/// `connection` is [`None`] until the handshake completes and after the connection closes.
+pub(crate) fn snapshot(
+    connection: Option<&quinn::Connection>,
+    counters: &Arc<NetCounters>,
+) -> NetStats {
+    let mut stats = NetStats {
+        chunks_received: counters.chunks_received.load(Ordering::Relaxed),
+        drops_received: counters.drops_received.load(Ordering::Relaxed),
+        subscribes_sent: counters.subscribes_sent.load(Ordering::Relaxed),
+        ..NetStats::default()
+    };
+
+    let Some(connection) = connection else {
+        return stats;
+    };
+
+    let quic = connection.stats();
+    // A connection handle outlives the connection itself; a close reason is how a torn-down connection distinguishes itself from a live one, and its final counters stay readable either way.
+    stats.connected = connection.close_reason().is_none();
+    stats.bytes_sent = quic.udp_tx.bytes;
+    stats.bytes_received = quic.udp_rx.bytes;
+    stats.datagrams_sent = quic.udp_tx.datagrams;
+    stats.datagrams_received = quic.udp_rx.datagrams;
+    stats.rtt_ms = quic.path.rtt.as_secs_f32() * 1000.0;
+    stats.lost_packets = quic.path.lost_packets;
+    stats.congestion_window = quic.path.cwnd;
+    stats.path_mtu = quic.path.current_mtu;
+    stats
+}
+
+#[cfg(test)]
+#[path = "tests/stats.rs"]
+mod tests;

crates/net/src/tests/stats.rs

@@ -0,0 +1,31 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//! Unit tests for the transport statistics snapshot.
+
+use super::*;
+
+#[test]
+fn snapshot_without_a_connection_reports_the_disconnected_state() {
+    let counters = Arc::new(NetCounters::default());
+    let stats = snapshot(None, &counters);
+
+    assert!(!stats.connected);
+    assert_eq!(stats, NetStats::default());
+}
+
+#[test]
+fn counters_are_reported_before_a_connection_exists() {
+    let counters = Arc::new(NetCounters::default());
+    counters.record_chunk();
+    counters.record_chunk();
+    counters.record_drop();
+    counters.record_subscribe();
+
+    let stats = snapshot(None, &counters);
+    assert_eq!(stats.chunks_received, 2);
+    assert_eq!(stats.drops_received, 1);
+    assert_eq!(stats.subscribes_sent, 1);
+    // Transport figures stay zero: they come from the QUIC stack, which has nothing to report yet.
+    assert_eq!(stats.bytes_sent, 0);
+    assert_eq!(stats.bytes_received, 0);
+}