transport: cap max_inbound_connections on inbound count, not combined pool

Add per-direction pool_inbound/pool_outbound counters to TcpStats and
TorStats, updated at every pool-insert, receive-loop-exit, transport-stop,
and send-failure removal site. Compare the max_inbound_connections cap
against pool_inbound rather than the combined pool length, so outbound
connect-on-send connections no longer consume the operator-facing inbound
budget. The configuration field name and operator semantics are preserved;
only the cap-check comparison and accounting change.

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`),

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
+                // Check inbound connection cap. Counts only inbound (accepted)
-                {
+                // connections currently held in the pool; outbound (connect-on-send)
-                    let pool_guard = pool.lock().await;
+                // connections live in the same pool but are not subject to the
-                    if pool_guard.len() >= max_inbound {
+                // operator-facing inbound cap.
-                        stats.record_connection_rejected();
+                if stats.pool_inbound_count() >= max_inbound as u64 {
-                        warn!(
+                    stats.record_connection_rejected();
-                            transport_id = %transport_id,
+                    warn!(
-                            peer_addr = %peer_addr,
+                        transport_id = %transport_id,
-                            max = max_inbound,
+                        peer_addr = %peer_addr,
-                            "Rejecting inbound TCP connection (max_inbound_connections reached)"
+                        max = max_inbound,
-                        );
+                        "Rejecting inbound TCP connection (max_inbound_connections reached)"
-                        continue;
+                    );
-                    }
+                    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<TcpStats>,
+    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"
     );
 }

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,
 }

src/transport/tor/mod.rs

@@ -111,6 +111,17 @@ fn parse_tor_addr(addr: &TransportAddr) -> Result<TorAddr, TransportError> {
 // 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<TorStats>,
+    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
+        // Check inbound connection cap. Counts only inbound (accepted)
-        let current_count = {
+        // connections currently held in the pool; outbound (SOCKS5-connect)
-            let pool_guard = pool.lock().await;
+        // connections live in the same pool but are not subject to the
-            pool_guard.len()
+        // operator-facing inbound cap.
-        };
+        if stats.pool_inbound_count() >= max_inbound as u64 {
-        if current_count >= max_inbound {
             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,

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,
 }