From a7dfe476635cb513ef810c9e23dc92ca615f7455 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 11 Jul 2026 04:27:50 +0000 Subject: [PATCH 1/3] transport: derive Serialize for EthernetStatsSnapshot EthernetStatsSnapshot derived only Clone/Debug/Default, unlike every other transport's *StatsSnapshot which also derives Serialize. That gap forced a hand-rolled serde_json::json!{} arm in TransportHandle:: transport_stats() that re-listed all ten fields by hand. Add the Serialize derive and collapse the arm to the same one-line serde_json::to_value(...) form the other transports use. The emitted JSON keys and values are unchanged. --- src/transport/ethernet/stats.rs | 4 +++- src/transport/mod.rs | 14 +------------- 2 files changed, 4 insertions(+), 14 deletions(-) diff --git a/src/transport/ethernet/stats.rs b/src/transport/ethernet/stats.rs index c20492f..0346232 100644 --- a/src/transport/ethernet/stats.rs +++ b/src/transport/ethernet/stats.rs @@ -2,6 +2,8 @@ use portable_atomic::{AtomicU64, Ordering}; +use serde::Serialize; + /// Statistics for an Ethernet transport instance. /// /// Uses atomic counters for lock-free updates from the receive loop @@ -92,7 +94,7 @@ impl Default for EthernetStats { } /// Point-in-time snapshot of Ethernet stats (non-atomic, copyable). -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Default, Serialize)] pub struct EthernetStatsSnapshot { pub frames_sent: u64, pub frames_recv: u64, diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 5128ce8..dadc520 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -1022,19 +1022,7 @@ impl TransportHandle { } #[cfg(unix)] TransportHandle::Ethernet(t) => { - let snap = t.stats().snapshot(); - serde_json::json!({ - "frames_sent": snap.frames_sent, - "frames_recv": snap.frames_recv, - "bytes_sent": snap.bytes_sent, - "bytes_recv": snap.bytes_recv, - "send_errors": snap.send_errors, - "recv_errors": snap.recv_errors, - "beacons_sent": snap.beacons_sent, - "beacons_recv": snap.beacons_recv, - "frames_too_short": snap.frames_too_short, - "frames_too_long": snap.frames_too_long, - }) + serde_json::to_value(t.stats().snapshot()).unwrap_or_default() } TransportHandle::Tcp(t) => { serde_json::to_value(t.stats().snapshot()).unwrap_or_default() From 1aacdfa08605aaac5caaff6f101656647397bf4b Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 11 Jul 2026 04:30:19 +0000 Subject: [PATCH 2/3] transport: share connection-pool counters via PoolCounters TcpStats and TorStats each carried an identical pair of pool_inbound/ pool_outbound atomics plus the same five record_pool_* / pool_inbound_count methods over them. Move the counter logic into a shared PoolCounters struct (new transport/stats_common.rs) embedded as a 'pool' field in both. The public record_pool_* methods stay as thin delegators so all call sites and the flat pool_inbound/pool_outbound snapshot fields are unchanged; only the duplicated atomic bookkeeping is now single-sourced. --- src/transport/mod.rs | 3 ++ src/transport/stats_common.rs | 56 +++++++++++++++++++++++++++++++++++ src/transport/tcp/stats.rs | 28 ++++++++---------- src/transport/tor/stats.rs | 29 +++++++++--------- 4 files changed, 86 insertions(+), 30 deletions(-) create mode 100644 src/transport/stats_common.rs diff --git a/src/transport/mod.rs b/src/transport/mod.rs index dadc520..9dacc17 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -34,6 +34,9 @@ use tor::TorTransport; use tor::control::TorMonitoringInfo; use udp::UdpTransport; +mod stats_common; +pub(crate) use stats_common::PoolCounters; + mod types; pub use types::*; diff --git a/src/transport/stats_common.rs b/src/transport/stats_common.rs new file mode 100644 index 0000000..c5e20c8 --- /dev/null +++ b/src/transport/stats_common.rs @@ -0,0 +1,56 @@ +//! Statistics helpers shared across transports. +//! +//! Counter logic that is identical across multiple transports lives here +//! rather than being copied into each transport's `stats.rs`. + +use portable_atomic::{AtomicU64, Ordering}; + +/// Inbound/outbound connection-pool occupancy counters. +/// +/// The connection-oriented transports (TCP, Tor) each track how many inbound +/// and outbound connections are currently held in their pool; the inbound +/// count drives the `max_inbound_connections` admission gate. The counter +/// logic is identical, so it is defined once here and embedded in each +/// transport's stats struct. +#[derive(Default)] +pub struct PoolCounters { + inbound: AtomicU64, + outbound: AtomicU64, +} + +impl PoolCounters { + /// Create counters at zero. + pub fn new() -> Self { + Self::default() + } + + /// Increment the inbound pool count (called on accept). + pub fn record_inbound_added(&self) { + self.inbound.fetch_add(1, Ordering::Relaxed); + } + + /// Decrement the inbound pool count (called on inbound receive-loop exit). + pub fn record_inbound_removed(&self) { + self.inbound.fetch_sub(1, Ordering::Relaxed); + } + + /// Increment the outbound pool count (called on connect-on-send / promote). + pub fn record_outbound_added(&self) { + self.outbound.fetch_add(1, Ordering::Relaxed); + } + + /// Decrement the outbound pool count (called on outbound receive-loop exit). + pub fn record_outbound_removed(&self) { + self.outbound.fetch_sub(1, Ordering::Relaxed); + } + + /// Load the current inbound pool count for the admission gate. + pub fn inbound_count(&self) -> u64 { + self.inbound.load(Ordering::Relaxed) + } + + /// Load the current outbound pool count. + pub fn outbound_count(&self) -> u64 { + self.outbound.load(Ordering::Relaxed) + } +} diff --git a/src/transport/tcp/stats.rs b/src/transport/tcp/stats.rs index 86785f8..62340e4 100644 --- a/src/transport/tcp/stats.rs +++ b/src/transport/tcp/stats.rs @@ -4,6 +4,8 @@ use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; +use crate::transport::PoolCounters; + /// Statistics for a TCP transport instance. /// /// Uses atomic counters for lock-free updates from per-connection @@ -21,12 +23,9 @@ pub struct TcpStats { pub connections_rejected: AtomicU64, pub connect_timeouts: AtomicU64, pub connect_refused: AtomicU64, - /// Current number of inbound (accepted) connections held in the pool. - /// Drives the `max_inbound_connections` admission check. - pub pool_inbound: AtomicU64, - /// Current number of outbound (connect-on-send / promoted) connections - /// held in the pool. Independent of the inbound cap. - pub pool_outbound: AtomicU64, + /// Inbound/outbound connection-pool occupancy. Inbound drives the + /// `max_inbound_connections` admission check. + pub pool: PoolCounters, } impl TcpStats { @@ -45,8 +44,7 @@ impl TcpStats { connections_rejected: AtomicU64::new(0), connect_timeouts: AtomicU64::new(0), connect_refused: AtomicU64::new(0), - pool_inbound: AtomicU64::new(0), - pool_outbound: AtomicU64::new(0), + pool: PoolCounters::new(), } } @@ -104,27 +102,27 @@ impl TcpStats { /// Increment the inbound pool count (called on accept). pub fn record_pool_inbound_added(&self) { - self.pool_inbound.fetch_add(1, Ordering::Relaxed); + self.pool.record_inbound_added(); } /// Decrement the inbound pool count (called on inbound receive-loop exit). pub fn record_pool_inbound_removed(&self) { - self.pool_inbound.fetch_sub(1, Ordering::Relaxed); + self.pool.record_inbound_removed(); } /// Increment the outbound pool count (called on connect-on-send / promote). pub fn record_pool_outbound_added(&self) { - self.pool_outbound.fetch_add(1, Ordering::Relaxed); + self.pool.record_outbound_added(); } /// Decrement the outbound pool count (called on outbound receive-loop exit). pub fn record_pool_outbound_removed(&self) { - self.pool_outbound.fetch_sub(1, Ordering::Relaxed); + self.pool.record_outbound_removed(); } /// Load the current inbound pool count for the admission gate. pub fn pool_inbound_count(&self) -> u64 { - self.pool_inbound.load(Ordering::Relaxed) + self.pool.inbound_count() } /// Take a snapshot of all counters. @@ -142,8 +140,8 @@ impl TcpStats { connections_rejected: self.connections_rejected.load(Ordering::Relaxed), connect_timeouts: self.connect_timeouts.load(Ordering::Relaxed), connect_refused: self.connect_refused.load(Ordering::Relaxed), - pool_inbound: self.pool_inbound.load(Ordering::Relaxed), - pool_outbound: self.pool_outbound.load(Ordering::Relaxed), + pool_inbound: self.pool.inbound_count(), + pool_outbound: self.pool.outbound_count(), } } } diff --git a/src/transport/tor/stats.rs b/src/transport/tor/stats.rs index 1838713..87a1790 100644 --- a/src/transport/tor/stats.rs +++ b/src/transport/tor/stats.rs @@ -4,6 +4,8 @@ use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; +use crate::transport::PoolCounters; + /// Statistics for a Tor transport instance. /// /// Uses atomic counters for lock-free updates from per-connection @@ -23,12 +25,10 @@ pub struct TorStats { pub connections_accepted: AtomicU64, pub connections_rejected: AtomicU64, pub control_errors: AtomicU64, - /// Current number of inbound (accepted via onion service) connections - /// held in the pool. Drives the `max_inbound_connections` admission check. - pub pool_inbound: AtomicU64, - /// Current number of outbound (SOCKS5 connect) connections held in - /// the pool. Independent of the inbound cap. - pub pool_outbound: AtomicU64, + /// Inbound (accepted via onion service) / outbound (SOCKS5 connect) + /// connection-pool occupancy. Inbound drives the + /// `max_inbound_connections` admission check. + pub pool: PoolCounters, } impl TorStats { @@ -49,8 +49,7 @@ impl TorStats { connections_accepted: AtomicU64::new(0), connections_rejected: AtomicU64::new(0), control_errors: AtomicU64::new(0), - pool_inbound: AtomicU64::new(0), - pool_outbound: AtomicU64::new(0), + pool: PoolCounters::new(), } } @@ -118,27 +117,27 @@ impl TorStats { /// Increment the inbound pool count (called on accept). pub fn record_pool_inbound_added(&self) { - self.pool_inbound.fetch_add(1, Ordering::Relaxed); + self.pool.record_inbound_added(); } /// Decrement the inbound pool count (called on inbound receive-loop exit). pub fn record_pool_inbound_removed(&self) { - self.pool_inbound.fetch_sub(1, Ordering::Relaxed); + self.pool.record_inbound_removed(); } /// Increment the outbound pool count (called on SOCKS5-connect promote). pub fn record_pool_outbound_added(&self) { - self.pool_outbound.fetch_add(1, Ordering::Relaxed); + self.pool.record_outbound_added(); } /// Decrement the outbound pool count (called on outbound receive-loop exit). pub fn record_pool_outbound_removed(&self) { - self.pool_outbound.fetch_sub(1, Ordering::Relaxed); + self.pool.record_outbound_removed(); } /// Load the current inbound pool count for the admission gate. pub fn pool_inbound_count(&self) -> u64 { - self.pool_inbound.load(Ordering::Relaxed) + self.pool.inbound_count() } /// Take a snapshot of all counters. @@ -158,8 +157,8 @@ impl TorStats { connections_accepted: self.connections_accepted.load(Ordering::Relaxed), connections_rejected: self.connections_rejected.load(Ordering::Relaxed), control_errors: self.control_errors.load(Ordering::Relaxed), - pool_inbound: self.pool_inbound.load(Ordering::Relaxed), - pool_outbound: self.pool_outbound.load(Ordering::Relaxed), + pool_inbound: self.pool.inbound_count(), + pool_outbound: self.pool.outbound_count(), } } } From f2e6b8befbb984e01847eefce835dfafe1a4c6d1 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 11 Jul 2026 04:31:56 +0000 Subject: [PATCH 3/3] transport: promote FMP stream framing to transport::framing The FMP frame-boundary reader lived in transport/tcp/stream.rs, but the Tor and Nym transports both reached across module boundaries to 'use crate::transport::tcp::stream::read_fmp_packet', a layering smell: the reader is a shared stream-framing utility, not a TCP-private one. Move the file to transport/framing.rs and repoint the tcp/tor/nym use sites, removing the tor->tcp and nym->tcp dependencies. Behavior unchanged. --- src/transport/{tcp/stream.rs => framing.rs} | 4 ++-- src/transport/mod.rs | 2 ++ src/transport/nym/mod.rs | 4 ++-- src/transport/tcp/mod.rs | 3 +-- src/transport/tor/mod.rs | 6 +++--- 5 files changed, 10 insertions(+), 9 deletions(-) rename src/transport/{tcp/stream.rs => framing.rs} (99%) diff --git a/src/transport/tcp/stream.rs b/src/transport/framing.rs similarity index 99% rename from src/transport/tcp/stream.rs rename to src/transport/framing.rs index 4323fdc..4fe0f24 100644 --- a/src/transport/tcp/stream.rs +++ b/src/transport/framing.rs @@ -3,8 +3,8 @@ //! Recovers FIPS packet boundaries from a TCP byte stream using the //! existing 4-byte FMP common prefix `[ver+phase:1][flags:1][payload_len:2 LE]`. //! -//! This module is deliberately separate from the TCP transport so it can -//! be reused by the future Tor transport. +//! This module is deliberately separate from any single transport so it can +//! be shared by the stream-oriented transports (TCP, Tor, Nym). use tokio::io::{AsyncRead, AsyncReadExt}; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 9dacc17..1176215 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -34,6 +34,8 @@ use tor::TorTransport; use tor::control::TorMonitoringInfo; use udp::UdpTransport; +pub(crate) mod framing; + mod stats_common; pub(crate) use stats_common::PoolCounters; diff --git a/src/transport/nym/mod.rs b/src/transport/nym/mod.rs index 4996664..053677f 100644 --- a/src/transport/nym/mod.rs +++ b/src/transport/nym/mod.rs @@ -9,7 +9,7 @@ //! //! Outbound-only: connects to remote TCP peers through the local //! nym-socks5-client SOCKS5 proxy. Like the Tor transport, reuses FMP -//! stream framing from `tcp::stream` and follows the same connection +//! stream framing from `transport::framing` and follows the same connection //! pool pattern. No inbound service is supported. pub mod stats; @@ -22,7 +22,7 @@ use super::{ TransportError, TransportId, TransportState, TransportType, }; use crate::config::NymConfig; -use crate::transport::tcp::stream::read_fmp_packet; +use crate::transport::framing::read_fmp_packet; use stats::NymStats; use futures::FutureExt; diff --git a/src/transport/tcp/mod.rs b/src/transport/tcp/mod.rs index 350c879..b1975b8 100644 --- a/src/transport/tcp/mod.rs +++ b/src/transport/tcp/mod.rs @@ -23,7 +23,6 @@ //! TCP stream and the receiver uses phase-dependent size computation. pub mod stats; -pub mod stream; use super::resolve_socket_addr; use super::{ @@ -31,8 +30,8 @@ use super::{ TransportError, TransportId, TransportState, TransportType, }; use crate::config::TcpConfig; +use crate::transport::framing::read_fmp_packet; use stats::TcpStats; -use stream::read_fmp_packet; use futures::FutureExt; use socket2::TcpKeepalive; diff --git a/src/transport/tor/mod.rs b/src/transport/tor/mod.rs index 5c2b8a9..e183632 100644 --- a/src/transport/tor/mod.rs +++ b/src/transport/tor/mod.rs @@ -14,8 +14,8 @@ //! ## Architecture //! //! Like TCP, each peer has its own connection. The transport reuses FMP -//! stream framing from `tcp::stream` and follows the same connection pool -//! pattern as the TCP transport. Inbound connections arrive via a local +//! stream framing from `transport::framing` and follows the same connection +//! pool pattern as the TCP transport. Inbound connections arrive via a local //! TCP listener that the Tor daemon forwards onion service traffic to. pub mod control; @@ -31,7 +31,7 @@ use super::{ TransportError, TransportId, TransportState, TransportType, }; use crate::config::TorConfig; -use crate::transport::tcp::stream::read_fmp_packet; +use crate::transport::framing::read_fmp_packet; use control::{ControlAuth, TorControlClient, TorMonitoringInfo}; use stats::TorStats;