diff --git a/CHANGELOG.md b/CHANGELOG.md index 49ad3a8..0ac3d81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `pool_inbound` and `pool_outbound` counters on the TCP and Tor + transport stats (`TcpStats`, `TorStats`). Per-direction accounting + is updated at every pool-insert and receive-loop-exit site, plus on + transport stop and on send-failure-driven removal. Surfaces through + `TcpStatsSnapshot` and `TorStatsSnapshot` for `show_transports`. - [`PR-REVIEW.md`](PR-REVIEW.md) — the 13-criteria PR review checklist the maintainer runs against every incoming PR, published at the repo root so contributors can run the same pass on their own change @@ -60,6 +65,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- TCP and Tor `max_inbound_connections` admission cap is now compared + against the per-direction inbound count (`pool_inbound`) rather than + the combined pool size. Outbound connect-on-send connections share + the same pool data structure but no longer consume slots against the + operator-facing inbound cap. The configuration field name and + operator semantics are preserved; only the cap-check comparison and + accounting change. Operators with mixed outbound + inbound + deployments no longer see legitimate inbound peers rejected once + outbound connections fill the pool past the configured cap. - Outbound connection initiation now honors the `node.limits.max_peers` cap that was previously only checked on inbound msg1 admission. Four paths gated: auto-reconnect retries (`process_pending_retries`), diff --git a/src/transport/tcp/mod.rs b/src/transport/tcp/mod.rs index 9a325cf..1e8f71b 100644 --- a/src/transport/tcp/mod.rs +++ b/src/transport/tcp/mod.rs @@ -52,6 +52,17 @@ use tracing::{debug, info, trace, warn}; // Connection Pool // ============================================================================ +/// Direction of a pooled connection, used to drive separate +/// `pool_inbound` / `pool_outbound` accounting for the +/// `max_inbound_connections` admission cap. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Direction { + /// Inbound — accepted by the listener. + Inbound, + /// Outbound — initiated by connect-on-send or background connect. + Outbound, +} + /// State for a single TCP connection to a peer. struct TcpConnection { /// Write half of the split stream. @@ -64,6 +75,8 @@ struct TcpConnection { /// When the connection was established. #[allow(dead_code)] established_at: Instant, + /// Direction of the connection — drives pool-inbound/outbound accounting. + direction: Direction, } /// Shared connection pool. @@ -241,14 +254,22 @@ impl TcpTransport { } drop(connecting); - // Close all established connections + // Close all established connections. The receive-loop cleanup + // would normally decrement pool_inbound / pool_outbound, but + // aborting the task skips that path; decrement explicitly here + // using the direction we stored on the connection record. let mut pool = self.pool.lock().await; for (addr, conn) in pool.drain() { conn.recv_task.abort(); let _ = conn.recv_task.await; + match conn.direction { + Direction::Inbound => self.stats.record_pool_inbound_removed(), + Direction::Outbound => self.stats.record_pool_outbound_removed(), + } debug!( transport_id = %self.transport_id, remote_addr = %addr, + direction = ?conn.direction, "TCP connection closed (transport stopping)" ); } @@ -326,6 +347,10 @@ impl TcpTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); + match conn.direction { + Direction::Inbound => self.stats.record_pool_inbound_removed(), + Direction::Outbound => self.stats.record_pool_outbound_removed(), + } } Err(TransportError::SendFailed(format!("{}", e))) } @@ -394,6 +419,7 @@ impl TcpTransport { pool, mtu, recv_stats, + Direction::Outbound, ) .await; }); @@ -403,12 +429,14 @@ impl TcpTransport { recv_task, mtu: mss_mtu, established_at: Instant::now(), + direction: Direction::Outbound, }; let mut pool = self.pool.lock().await; pool.insert(addr.clone(), conn); self.stats.record_connection_established(); + self.stats.record_pool_outbound_added(); debug!( transport_id = %self.transport_id, @@ -428,9 +456,14 @@ impl TcpTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); + match conn.direction { + Direction::Inbound => self.stats.record_pool_inbound_removed(), + Direction::Outbound => self.stats.record_pool_outbound_removed(), + } debug!( transport_id = %self.transport_id, remote_addr = %addr, + direction = ?conn.direction, "TCP connection closed (close_connection)" ); } @@ -635,6 +668,7 @@ impl TcpTransport { pool, mss_mtu, recv_stats, + Direction::Outbound, ) .await; }); @@ -644,6 +678,7 @@ impl TcpTransport { recv_task, mtu: mss_mtu, established_at: Instant::now(), + direction: Direction::Outbound, }; // Use try_lock since we're in a sync context and the pool @@ -651,6 +686,7 @@ impl TcpTransport { if let Ok(mut pool) = self.pool.try_lock() { pool.insert(addr.clone(), conn); self.stats.record_connection_established(); + self.stats.record_pool_outbound_added(); debug!( transport_id = %self.transport_id, remote_addr = %addr, @@ -759,19 +795,19 @@ async fn accept_loop( loop { match listener.accept().await { Ok((stream, peer_addr)) => { - // Check connection limit - { - let pool_guard = pool.lock().await; - if pool_guard.len() >= max_inbound { - stats.record_connection_rejected(); - warn!( - transport_id = %transport_id, - peer_addr = %peer_addr, - max = max_inbound, - "Rejecting inbound TCP connection (max_inbound_connections reached)" - ); - continue; - } + // Check inbound connection cap. Counts only inbound (accepted) + // connections currently held in the pool; outbound (connect-on-send) + // connections live in the same pool but are not subject to the + // operator-facing inbound cap. + if stats.pool_inbound_count() >= max_inbound as u64 { + stats.record_connection_rejected(); + warn!( + transport_id = %transport_id, + peer_addr = %peer_addr, + max = max_inbound, + "Rejecting inbound TCP connection (max_inbound_connections reached)" + ); + continue; } // Configure socket options @@ -838,6 +874,7 @@ async fn accept_loop( recv_pool, conn_mtu, recv_stats, + Direction::Inbound, ) .await; }); @@ -847,12 +884,14 @@ async fn accept_loop( recv_task, mtu: conn_mtu, established_at: Instant::now(), + direction: Direction::Inbound, }; let mut pool_guard = pool.lock().await; pool_guard.insert(remote_addr.clone(), conn); stats.record_connection_accepted(); + stats.record_pool_inbound_added(); debug!( transport_id = %transport_id, @@ -880,7 +919,11 @@ async fn accept_loop( /// /// Reads complete FMP packets using the stream reader, delivers them to /// the node via the packet channel. On error or EOF, removes the -/// connection from the pool and exits. +/// connection from the pool and exits. `direction` is captured here so +/// the cleanup path can decrement the correct `pool_inbound` / +/// `pool_outbound` counter regardless of whether the matching pool +/// entry survived to be removed. +#[allow(clippy::too_many_arguments)] async fn tcp_receive_loop( mut reader: tokio::net::tcp::OwnedReadHalf, transport_id: TransportId, @@ -889,6 +932,7 @@ async fn tcp_receive_loop( pool: ConnectionPool, mtu: u16, stats: Arc, + direction: Direction, ) { debug!( transport_id = %transport_id, @@ -932,13 +976,24 @@ async fn tcp_receive_loop( } } - // Clean up: remove ourselves from the pool + // Clean up: remove ourselves from the pool, then decrement the + // direction-specific pool counter. Decrement is conditional on the + // entry actually being removed so a double-cleanup never drives + // the counter below zero. let mut pool_guard = pool.lock().await; - pool_guard.remove(&remote_addr); + let removed = pool_guard.remove(&remote_addr).is_some(); + drop(pool_guard); + if removed { + match direction { + Direction::Inbound => stats.record_pool_inbound_removed(), + Direction::Outbound => stats.record_pool_outbound_removed(), + } + } debug!( transport_id = %transport_id, remote_addr = %remote_addr, + direction = ?direction, "TCP receive loop stopped" ); } diff --git a/src/transport/tcp/stats.rs b/src/transport/tcp/stats.rs index 0434038..86785f8 100644 --- a/src/transport/tcp/stats.rs +++ b/src/transport/tcp/stats.rs @@ -21,6 +21,12 @@ 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, } impl TcpStats { @@ -39,6 +45,8 @@ 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), } } @@ -94,6 +102,31 @@ impl TcpStats { self.connect_refused.fetch_add(1, Ordering::Relaxed); } + /// Increment the inbound pool count (called on accept). + pub fn record_pool_inbound_added(&self) { + self.pool_inbound.fetch_add(1, Ordering::Relaxed); + } + + /// 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); + } + + /// 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); + } + + /// 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); + } + + /// Load the current inbound pool count for the admission gate. + pub fn pool_inbound_count(&self) -> u64 { + self.pool_inbound.load(Ordering::Relaxed) + } + /// Take a snapshot of all counters. pub fn snapshot(&self) -> TcpStatsSnapshot { TcpStatsSnapshot { @@ -109,6 +142,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), } } } @@ -134,4 +169,6 @@ pub struct TcpStatsSnapshot { pub connections_rejected: u64, pub connect_timeouts: u64, pub connect_refused: u64, + pub pool_inbound: u64, + pub pool_outbound: u64, } diff --git a/src/transport/tor/mod.rs b/src/transport/tor/mod.rs index dd7355f..a665efb 100644 --- a/src/transport/tor/mod.rs +++ b/src/transport/tor/mod.rs @@ -111,6 +111,17 @@ fn parse_tor_addr(addr: &TransportAddr) -> Result { // Connection Pool // ============================================================================ +/// Direction of a pooled Tor connection, used to drive separate +/// `pool_inbound` / `pool_outbound` accounting for the +/// `max_inbound_connections` admission cap on the onion-service side. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Direction { + /// Inbound — accepted by the onion service listener. + Inbound, + /// Outbound — initiated via SOCKS5 connect to a remote onion. + Outbound, +} + /// State for a single Tor connection to a peer. struct TorConnection { /// Write half of the split stream. @@ -123,6 +134,8 @@ struct TorConnection { /// When the connection was established. #[allow(dead_code)] established_at: Instant, + /// Direction of the connection — drives pool-inbound/outbound accounting. + direction: Direction, } /// Shared connection pool. @@ -516,14 +529,22 @@ impl TorTransport { } drop(connecting); - // Close all connections + // Close all connections. The receive-loop cleanup would + // normally decrement pool_inbound / pool_outbound, but + // aborting the task skips that path; decrement explicitly + // here using the direction we stored on the connection record. let mut pool = self.pool.lock().await; for (addr, conn) in pool.drain() { conn.recv_task.abort(); let _ = conn.recv_task.await; + match conn.direction { + Direction::Inbound => self.stats.record_pool_inbound_removed(), + Direction::Outbound => self.stats.record_pool_outbound_removed(), + } debug!( transport_id = %self.transport_id, remote_addr = %addr, + direction = ?conn.direction, "Tor connection closed (transport stopping)" ); } @@ -690,6 +711,10 @@ impl TorTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); + match conn.direction { + Direction::Inbound => self.stats.record_pool_inbound_removed(), + Direction::Outbound => self.stats.record_pool_outbound_removed(), + } } Err(TransportError::SendFailed(format!("{}", e))) } @@ -802,6 +827,7 @@ impl TorTransport { pool, mtu, recv_stats, + Direction::Outbound, ) .await; }); @@ -811,12 +837,14 @@ impl TorTransport { recv_task, mtu, established_at: Instant::now(), + direction: Direction::Outbound, }; let mut pool = self.pool.lock().await; pool.insert(addr.clone(), conn); self.stats.record_connection_established(); + self.stats.record_pool_outbound_added(); info!( transport_id = %self.transport_id, @@ -1023,6 +1051,7 @@ impl TorTransport { pool, mtu, recv_stats, + Direction::Outbound, ) .await; }); @@ -1032,6 +1061,7 @@ impl TorTransport { recv_task, mtu, established_at: Instant::now(), + direction: Direction::Outbound, }; // Use try_lock since we're in a sync context and the pool @@ -1039,6 +1069,7 @@ impl TorTransport { if let Ok(mut pool) = self.pool.try_lock() { pool.insert(addr.clone(), conn); self.stats.record_connection_established(); + self.stats.record_pool_outbound_added(); debug!( transport_id = %self.transport_id, remote_addr = %addr, @@ -1060,6 +1091,10 @@ impl TorTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); + match conn.direction { + Direction::Inbound => self.stats.record_pool_inbound_removed(), + Direction::Outbound => self.stats.record_pool_outbound_removed(), + } debug!( transport_id = %self.transport_id, remote_addr = %addr, @@ -1125,7 +1160,10 @@ impl Transport for TorTransport { /// /// Reads complete FMP packets using the stream reader, delivers them to /// the node via the packet channel. On error or EOF, removes the -/// connection from the pool and exits. +/// connection from the pool and exits. `direction` is captured so the +/// cleanup path can decrement the correct `pool_inbound` / +/// `pool_outbound` counter. +#[allow(clippy::too_many_arguments)] async fn tor_receive_loop( mut reader: tokio::net::tcp::OwnedReadHalf, transport_id: TransportId, @@ -1134,6 +1172,7 @@ async fn tor_receive_loop( pool: ConnectionPool, mtu: u16, stats: Arc, + direction: Direction, ) { debug!( transport_id = %transport_id, @@ -1176,13 +1215,24 @@ async fn tor_receive_loop( } } - // Clean up: remove ourselves from the pool + // Clean up: remove ourselves from the pool, then decrement the + // direction-specific pool counter. Decrement is conditional on the + // entry actually being removed so a double-cleanup never drives + // the counter below zero. let mut pool_guard = pool.lock().await; - pool_guard.remove(&remote_addr); + let removed = pool_guard.remove(&remote_addr).is_some(); + drop(pool_guard); + if removed { + match direction { + Direction::Inbound => stats.record_pool_inbound_removed(), + Direction::Outbound => stats.record_pool_outbound_removed(), + } + } debug!( transport_id = %transport_id, remote_addr = %remote_addr, + direction = ?direction, "Tor receive loop stopped" ); } @@ -1254,12 +1304,11 @@ async fn tor_accept_loop( } }; - // Check inbound connection limit - let current_count = { - let pool_guard = pool.lock().await; - pool_guard.len() - }; - if current_count >= max_inbound { + // Check inbound connection cap. Counts only inbound (accepted) + // connections currently held in the pool; outbound (SOCKS5-connect) + // connections live in the same pool but are not subject to the + // operator-facing inbound cap. + if stats.pool_inbound_count() >= max_inbound as u64 { stats.record_connection_rejected(); debug!( transport_id = %transport_id, @@ -1321,6 +1370,7 @@ async fn tor_accept_loop( recv_pool, mtu, recv_stats, + Direction::Inbound, ) .await; }); @@ -1330,6 +1380,7 @@ async fn tor_accept_loop( recv_task, mtu, established_at: Instant::now(), + direction: Direction::Inbound, }; { @@ -1338,6 +1389,7 @@ async fn tor_accept_loop( } stats.record_connection_accepted(); + stats.record_pool_inbound_added(); debug!( transport_id = %transport_id, diff --git a/src/transport/tor/stats.rs b/src/transport/tor/stats.rs index 4a6c8d9..1838713 100644 --- a/src/transport/tor/stats.rs +++ b/src/transport/tor/stats.rs @@ -23,6 +23,12 @@ 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, } impl TorStats { @@ -43,6 +49,8 @@ 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), } } @@ -108,6 +116,31 @@ impl TorStats { self.control_errors.fetch_add(1, Ordering::Relaxed); } + /// Increment the inbound pool count (called on accept). + pub fn record_pool_inbound_added(&self) { + self.pool_inbound.fetch_add(1, Ordering::Relaxed); + } + + /// 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); + } + + /// 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); + } + + /// 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); + } + + /// Load the current inbound pool count for the admission gate. + pub fn pool_inbound_count(&self) -> u64 { + self.pool_inbound.load(Ordering::Relaxed) + } + /// Take a snapshot of all counters. pub fn snapshot(&self) -> TorStatsSnapshot { TorStatsSnapshot { @@ -125,6 +158,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), } } } @@ -152,4 +187,6 @@ pub struct TorStatsSnapshot { pub connections_accepted: u64, pub connections_rejected: u64, pub control_errors: u64, + pub pool_inbound: u64, + pub pool_outbound: u64, }