v25.9.3

changelog.md

@@ -1,5 +1,32 @@
 # Changelog
 
+## 2026-03-11 - 25.9.3 - fix(rustproxy-http)
+Evict stale HTTP/2 pooled senders and retry bodyless requests with fresh backend connections to avoid 502s
+
+- Introduce MAX_H2_AGE (120s) and evict HTTP/2 senders older than this or closed
+- Check MAX_H2_AGE on checkout and during background eviction to prevent reuse of stale h2 connections
+- Add connection_pool.remove_h2() to explicitly remove dead H2 senders from the pool
+- When a pooled H2 request returns a 502 and the original request had an empty body, retry using a fresh H2 connection (retry_h2_with_fresh_connection)
+- On H2 auto-detect failures, retry as HTTP/1.1 for bodyless requests via forward_h1_empty_body; return 502 for requests with bodies
+- Evict dead H2 senders on backend request failures in reconnect_backend so subsequent attempts create fresh connections
+
+## 2026-03-08 - 25.9.2 - fix(protocol-cache)
+Include requested_host in protocol detection cache key to avoid cache oscillation when multiple frontend domains share the same backend
+
+- Add ProtocolCacheKey.requested_host: Option<String> to distinguish cache entries by incoming request Host/:authority
+- Update protocol cache lookups/inserts in proxy_service to populate requested_host
+- Enhance debug logging to show requested_host on cache hits
+- Fixes repeated ALPN probing / cache oscillation when different frontend domains share a backend with differing HTTP/2 support
+
+## 2026-03-03 - 25.9.1 - fix(rustproxy)
+Cancel connections for routes removed/disabled by adding per-route cancellation tokens and make RouteManager swappable (ArcSwap) for runtime updates
+
+- Add per-route CancellationToken map (DashMap) to TcpListenerManager and call token.cancel() when routes are removed (invalidate_removed_routes)
+- Propagate Arc<ArcSwap<RouteManager>> into HttpProxyService and passthrough listener so the route manager can be hot-swapped without restarting listeners
+- Use per-route child cancellation tokens in accept/connection handling and forwarders to terminate existing connections when a route is removed
+- Prune HTTP proxy caches and retain/cleanup per-route tokens when routes are active/removed
+- Update test.test.sni-requirement.node.ts to allocate unique free ports via findFreePorts to avoid port conflicts during tests
+
 ## 2026-03-03 - 25.9.0 - feat(rustproxy-http)
 add HTTP/2 auto-detection via ALPN with TTL-backed protocol cache and h1-only/h2 ALPN client configs
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartproxy",
-  "version": "25.9.0",
+  "version": "25.9.3",
   "private": false,
   "description": "A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.",
   "main": "dist_ts/index.js",

rust/crates/rustproxy-http/src/connection_pool.rs

@@ -18,6 +18,9 @@ const MAX_IDLE_PER_KEY: usize = 16;
 const IDLE_TIMEOUT: Duration = Duration::from_secs(90);
 /// Background eviction interval.
 const EVICTION_INTERVAL: Duration = Duration::from_secs(30);
+/// Maximum age for pooled HTTP/2 connections before proactive eviction.
+/// Prevents staleness from backends that close idle connections (e.g. nginx GOAWAY).
+const MAX_H2_AGE: Duration = Duration::from_secs(120);
 
 /// Identifies a unique backend endpoint.
 #[derive(Clone, Debug, Hash, Eq, PartialEq)]
@@ -37,7 +40,6 @@ struct IdleH1 {
 /// A pooled HTTP/2 sender (multiplexed, Clone-able).
 struct PooledH2 {
     sender: http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
-    #[allow(dead_code)] // Reserved for future age-based eviction
     created_at: Instant,
 }
 
@@ -116,8 +118,8 @@ impl ConnectionPool {
         let entry = self.h2_pool.get(key)?;
         let pooled = entry.value();
 
-        // Check if the h2 connection is still alive
+        // Check if the h2 connection is still alive and not too old
-        if pooled.sender.is_closed() {
+        if pooled.sender.is_closed() || pooled.created_at.elapsed() >= MAX_H2_AGE {
             drop(entry);
             self.h2_pool.remove(key);
             return None;
@@ -130,6 +132,12 @@ impl ConnectionPool {
         None
     }
 
+    /// Remove a dead HTTP/2 sender from the pool.
+    /// Called when `send_request` fails to prevent subsequent requests from reusing the stale sender.
+    pub fn remove_h2(&self, key: &PoolKey) {
+        self.h2_pool.remove(key);
+    }
+
     /// Register an HTTP/2 sender in the pool. Since h2 is multiplexed,
     /// only one sender per key is stored (it's Clone-able).
     pub fn register_h2(&self, key: PoolKey, sender: http2::SendRequest<BoxBody<Bytes, hyper::Error>>) {
@@ -165,10 +173,10 @@ impl ConnectionPool {
                 h1_pool.remove(&key);
             }
 
-            // Evict dead H2 connections
+            // Evict dead or aged-out H2 connections
             let mut dead_h2 = Vec::new();
             for entry in h2_pool.iter() {
-                if entry.value().sender.is_closed() {
+                if entry.value().sender.is_closed() || entry.value().created_at.elapsed() >= MAX_H2_AGE {
                     dead_h2.push(entry.key().clone());
                 }
             }

rust/crates/rustproxy-http/src/protocol_cache.rs

@@ -1,7 +1,8 @@
 //! Bounded, TTL-based protocol detection cache for HTTP/2 auto-detection.
 //!
-//! Caches the ALPN-negotiated protocol (H1 or H2) per backend endpoint (host:port).
+//! Caches the ALPN-negotiated protocol (H1 or H2) per backend endpoint and requested
-//! Prevents repeated ALPN probes for backends whose protocol is already known.
+//! domain (host:port + requested_host). This prevents cache oscillation when multiple
+//! frontend domains share the same backend but differ in HTTP/2 support.
 
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -27,11 +28,14 @@ pub enum DetectedProtocol {
     H2,
 }
 
-/// Key for the protocol cache: (host, port).
+/// Key for the protocol cache: (host, port, requested_host).
 #[derive(Clone, Debug, Hash, Eq, PartialEq)]
 pub struct ProtocolCacheKey {
     pub host: String,
     pub port: u16,
+    /// The incoming request's domain (Host header / :authority).
+    /// Distinguishes protocol detection when multiple domains share the same backend.
+    pub requested_host: Option<String>,
 }
 
 /// A cached protocol detection result with a timestamp.
@@ -73,7 +77,7 @@ impl ProtocolCache {
     pub fn get(&self, key: &ProtocolCacheKey) -> Option<DetectedProtocol> {
         let entry = self.cache.get(key)?;
         if entry.detected_at.elapsed() < PROTOCOL_CACHE_TTL {
-            debug!("Protocol cache hit: {:?} for {}:{}", entry.protocol, key.host, key.port);
+            debug!("Protocol cache hit: {:?} for {}:{} (requested: {:?})", entry.protocol, key.host, key.port, key.requested_host);
             Some(entry.protocol)
         } else {
             // Expired — remove and return None to trigger re-probe

rust/crates/rustproxy-http/src/proxy_service.rs

@@ -8,8 +8,10 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 
+use arc_swap::ArcSwap;
 use bytes::Bytes;
 use dashmap::DashMap;
+use http_body::Body as HttpBody;
 use http_body_util::{BodyExt, Full, combinators::BoxBody};
 use hyper::body::Incoming;
 use hyper::{Request, Response, StatusCode};
@@ -133,7 +135,7 @@ async fn connect_tls_backend(
 
 /// HTTP proxy service that processes HTTP traffic.
 pub struct HttpProxyService {
-    route_manager: Arc<RouteManager>,
+    route_manager: Arc<ArcSwap<RouteManager>>,
     metrics: Arc<MetricsCollector>,
     upstream_selector: UpstreamSelector,
     /// Timeout for connecting to upstream backends.
@@ -161,7 +163,7 @@ pub struct HttpProxyService {
 }
 
 impl HttpProxyService {
-    pub fn new(route_manager: Arc<RouteManager>, metrics: Arc<MetricsCollector>) -> Self {
+    pub fn new(route_manager: Arc<ArcSwap<RouteManager>>, metrics: Arc<MetricsCollector>) -> Self {
         Self {
             route_manager,
             metrics,
@@ -182,7 +184,7 @@ impl HttpProxyService {
 
     /// Create with a custom connect timeout.
     pub fn with_connect_timeout(
-        route_manager: Arc<RouteManager>,
+        route_manager: Arc<ArcSwap<RouteManager>>,
         metrics: Arc<MetricsCollector>,
         connect_timeout: std::time::Duration,
     ) -> Self {
@@ -405,7 +407,8 @@ impl HttpProxyService {
             protocol: Some("http"),
         };
 
-        let route_match = match self.route_manager.find_route(&ctx) {
+        let current_rm = self.route_manager.load();
+        let route_match = match current_rm.find_route(&ctx) {
             Some(rm) => rm,
             None => {
                 debug!("No route matched for HTTP request to {:?}{}", host, path);
@@ -591,6 +594,7 @@ impl HttpProxyService {
                     let cache_key = crate::protocol_cache::ProtocolCacheKey {
                         host: upstream.host.clone(),
                         port: upstream.port,
+                        requested_host: host.clone(),
                     };
                     match self.protocol_cache.get(&cache_key) {
                         Some(crate::protocol_cache::DetectedProtocol::H2) => (true, false),
@@ -648,6 +652,7 @@ impl HttpProxyService {
                         let cache_key = crate::protocol_cache::ProtocolCacheKey {
                             host: upstream.host.clone(),
                             port: upstream.port,
+                            requested_host: host.clone(),
                         };
                         let detected = if is_h2 {
                             crate::protocol_cache::DetectedProtocol::H2
@@ -719,6 +724,7 @@ impl HttpProxyService {
                 self.forward_h2_with_fallback(
                     io, parts, body, upstream_headers, &upstream_path,
                     &upstream, route_match.route, route_id, &ip_str, &final_pool_key,
+                    host.clone(),
                 ).await
             } else {
                 // Explicit H2 mode: hard-fail on handshake error (preserved behavior)
@@ -867,10 +873,12 @@ impl HttpProxyService {
         // Register for multiplexed reuse
         self.connection_pool.register_h2(pool_key.clone(), sender.clone());
 
-        self.forward_h2_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip).await
+        self.forward_h2_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip, Some(pool_key)).await
     }
 
     /// Forward request using an existing (pooled) HTTP/2 sender.
+    /// If the pooled sender is stale (GOAWAY, connection closed), evicts it and retries
+    /// with a fresh connection for bodyless requests (GET/HEAD/DELETE).
     async fn forward_h2_pooled(
         &self,
         sender: hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
@@ -881,9 +889,129 @@ impl HttpProxyService {
         route: &rustproxy_config::RouteConfig,
         route_id: Option<&str>,
         source_ip: &str,
-        _pool_key: &crate::connection_pool::PoolKey,
+        pool_key: &crate::connection_pool::PoolKey,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
-        self.forward_h2_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip).await
+        // Save retry state for bodyless requests (cheap: Method is an enum, HeaderMap clones Arc-backed Bytes)
+        let retry_state = if body.is_end_stream() {
+            Some((parts.method.clone(), upstream_headers.clone()))
+        } else {
+            None
+        };
+
+        let result = self.forward_h2_with_sender(
+            sender, parts, body, upstream_headers, upstream_path,
+            route, route_id, source_ip, Some(pool_key),
+        ).await;
+
+        // If the request failed (502) and we can retry with an empty body, do so
+        let is_502 = matches!(&result, Ok(resp) if resp.status() == StatusCode::BAD_GATEWAY);
+        if is_502 {
+            if let Some((method, headers)) = retry_state {
+                warn!("Stale pooled H2 sender for {}:{}, retrying with fresh connection",
+                    pool_key.host, pool_key.port);
+                return self.retry_h2_with_fresh_connection(
+                    method, headers, upstream_path,
+                    pool_key, route, route_id, source_ip,
+                ).await;
+            }
+        }
+        result
+    }
+
+    /// Retry an H2 request with a fresh backend connection and empty body.
+    /// Used when a pooled sender was stale (GOAWAY/closed) and the original body was empty.
+    async fn retry_h2_with_fresh_connection(
+        &self,
+        method: hyper::Method,
+        upstream_headers: hyper::HeaderMap,
+        upstream_path: &str,
+        pool_key: &crate::connection_pool::PoolKey,
+        route: &rustproxy_config::RouteConfig,
+        route_id: Option<&str>,
+        source_ip: &str,
+    ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+        // Establish fresh backend connection
+        let backend = if pool_key.use_tls {
+            match tokio::time::timeout(
+                self.connect_timeout,
+                connect_tls_backend(&self.backend_tls_config, &pool_key.host, pool_key.port),
+            ).await {
+                Ok(Ok(tls)) => BackendStream::Tls(tls),
+                Ok(Err(e)) => {
+                    error!("H2 retry: TLS connect failed for {}:{}: {}", pool_key.host, pool_key.port, e);
+                    return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable on H2 retry"));
+                }
+                Err(_) => {
+                    error!("H2 retry: TLS connect timeout for {}:{}", pool_key.host, pool_key.port);
+                    return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend timeout on H2 retry"));
+                }
+            }
+        } else {
+            match tokio::time::timeout(
+                self.connect_timeout,
+                TcpStream::connect(format!("{}:{}", pool_key.host, pool_key.port)),
+            ).await {
+                Ok(Ok(s)) => {
+                    s.set_nodelay(true).ok();
+                    BackendStream::Plain(s)
+                }
+                Ok(Err(e)) => {
+                    error!("H2 retry: connect failed for {}:{}: {}", pool_key.host, pool_key.port, e);
+                    return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable on H2 retry"));
+                }
+                Err(_) => {
+                    error!("H2 retry: connect timeout for {}:{}", pool_key.host, pool_key.port);
+                    return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend timeout on H2 retry"));
+                }
+            }
+        };
+
+        let io = TokioIo::new(backend);
+        let exec = hyper_util::rt::TokioExecutor::new();
+        let (mut sender, conn): (
+            hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
+            hyper::client::conn::http2::Connection<TokioIo<BackendStream>, BoxBody<Bytes, hyper::Error>, hyper_util::rt::TokioExecutor>,
+        ) = match hyper::client::conn::http2::handshake(exec, io).await {
+            Ok(h) => h,
+            Err(e) => {
+                error!("H2 retry: handshake failed for {}:{}: {}", pool_key.host, pool_key.port, e);
+                return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 retry handshake failed"));
+            }
+        };
+
+        tokio::spawn(async move {
+            if let Err(e) = conn.await {
+                debug!("H2 retry: upstream connection error: {}", e);
+            }
+        });
+
+        // Register fresh sender in pool for future requests
+        self.connection_pool.register_h2(pool_key.clone(), sender.clone());
+
+        // Build request with empty body
+        let mut upstream_req = Request::builder()
+            .method(method)
+            .uri(upstream_path);
+
+        if let Some(headers) = upstream_req.headers_mut() {
+            *headers = upstream_headers;
+        }
+
+        let empty_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(
+            http_body_util::Empty::new().map_err(|never| match never {})
+        );
+        let upstream_req = upstream_req.body(empty_body).unwrap();
+
+        match sender.send_request(upstream_req).await {
+            Ok(resp) => {
+                self.build_streaming_response(resp, route, route_id, source_ip).await
+            }
+            Err(e) => {
+                error!("H2 retry: request failed for {}:{}: {}", pool_key.host, pool_key.port, e);
+                self.connection_pool.remove_h2(pool_key);
+                Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 request failed on retry"))
+            }
+        }
     }
 
     /// Forward via HTTP/2 with fallback to HTTP/1.1 (auto-detect mode).
@@ -891,8 +1019,8 @@ impl HttpProxyService {
     /// Handles two failure scenarios:
     /// 1. H2 handshake fails → reconnects and falls back to H1 (body not consumed yet).
     /// 2. H2 handshake "succeeds" but request fails (backend advertises h2 via ALPN but
-    ///    doesn't actually speak h2) → updates cache to H1. The request body is consumed
+    ///    doesn't actually speak h2) → updates cache to H1, retries as H1 for bodyless
-    ///    so this request fails, but all subsequent requests will correctly use H1.
+    ///    requests, or returns 502 for requests with bodies.
     async fn forward_h2_with_fallback(
         &self,
         io: TokioIo<BackendStream>,
@@ -905,6 +1033,7 @@ impl HttpProxyService {
         route_id: Option<&str>,
         source_ip: &str,
         pool_key: &crate::connection_pool::PoolKey,
+        requested_host: Option<String>,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
         let exec = hyper_util::rt::TokioExecutor::new();
         let handshake_result: Result<(
@@ -920,6 +1049,13 @@ impl HttpProxyService {
                     }
                 });
 
+                // Save retry state before consuming parts/body (for bodyless requests like GET)
+                let retry_state = if body.is_end_stream() {
+                    Some((parts.method.clone(), upstream_headers.clone()))
+                } else {
+                    None
+                };
+
                 // Build and send the h2 request inline (don't register in pool yet —
                 // we need to verify the request actually succeeds first, because some
                 // backends advertise h2 via ALPN but don't speak the h2 binary protocol).
@@ -950,18 +1086,40 @@ impl HttpProxyService {
                     Err(e) => {
                         // H2 request failed — backend advertises h2 via ALPN but doesn't
                         // actually speak it. Update cache so future requests use H1.
-                        // The request body is consumed so this request can't be retried,
-                        // but all subsequent requests will correctly use H1.
                         warn!(
-                            "Auto-detect: H2 request failed for {}:{}, updating cache to H1: {}",
+                            "Auto-detect: H2 request failed for {}:{}, falling back to H1: {}",
                             upstream.host, upstream.port, e
                         );
                         let cache_key = crate::protocol_cache::ProtocolCacheKey {
                             host: upstream.host.clone(),
                             port: upstream.port,
+                            requested_host: requested_host.clone(),
                         };
                         self.protocol_cache.insert(cache_key, crate::protocol_cache::DetectedProtocol::H1);
-                        Ok(error_response(StatusCode::BAD_GATEWAY, "Backend protocol mismatch, retrying with H1"))
+
+                        // Retry as H1 for bodyless requests; return 502 for requests with bodies
+                        if let Some((method, headers)) = retry_state {
+                            match self.reconnect_backend(upstream).await {
+                                Some(fallback_backend) => {
+                                    let h1_pool_key = crate::connection_pool::PoolKey {
+                                        host: upstream.host.clone(),
+                                        port: upstream.port,
+                                        use_tls: upstream.use_tls,
+                                        h2: false,
+                                    };
+                                    let fallback_io = TokioIo::new(fallback_backend);
+                                    self.forward_h1_empty_body(
+                                        fallback_io, method, headers, upstream_path,
+                                        route, route_id, source_ip, &h1_pool_key,
+                                    ).await
+                                }
+                                None => {
+                                    Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable after H2 fallback"))
+                                }
+                            }
+                        } else {
+                            Ok(error_response(StatusCode::BAD_GATEWAY, "Backend protocol mismatch"))
+                        }
                     }
                 }
             }
@@ -977,6 +1135,7 @@ impl HttpProxyService {
                 let cache_key = crate::protocol_cache::ProtocolCacheKey {
                     host: upstream.host.clone(),
                     port: upstream.port,
+                    requested_host: requested_host.clone(),
                 };
                 self.protocol_cache.insert(cache_key, crate::protocol_cache::DetectedProtocol::H1);
 
@@ -1003,6 +1162,64 @@ impl HttpProxyService {
         }
     }
 
+    /// Forward a request with an empty body via HTTP/1.1.
+    /// Used when retrying after a failed H2 attempt where the original body was consumed.
+    async fn forward_h1_empty_body(
+        &self,
+        io: TokioIo<BackendStream>,
+        method: hyper::Method,
+        upstream_headers: hyper::HeaderMap,
+        upstream_path: &str,
+        route: &rustproxy_config::RouteConfig,
+        route_id: Option<&str>,
+        source_ip: &str,
+        pool_key: &crate::connection_pool::PoolKey,
+    ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+        let (mut sender, conn): (
+            hyper::client::conn::http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
+            hyper::client::conn::http1::Connection<TokioIo<BackendStream>, BoxBody<Bytes, hyper::Error>>,
+        ) = match hyper::client::conn::http1::handshake(io).await {
+            Ok(h) => h,
+            Err(e) => {
+                error!("H1 fallback: handshake failed: {}", e);
+                return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H1 fallback handshake failed"));
+            }
+        };
+
+        tokio::spawn(async move {
+            if let Err(e) = conn.await {
+                debug!("H1 fallback: upstream connection error: {}", e);
+            }
+        });
+
+        let mut upstream_req = Request::builder()
+            .method(method)
+            .uri(upstream_path)
+            .version(hyper::Version::HTTP_11);
+
+        if let Some(headers) = upstream_req.headers_mut() {
+            *headers = upstream_headers;
+        }
+
+        let empty_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(
+            http_body_util::Empty::new().map_err(|never| match never {})
+        );
+        let upstream_req = upstream_req.body(empty_body).unwrap();
+
+        let upstream_response = match sender.send_request(upstream_req).await {
+            Ok(resp) => resp,
+            Err(e) => {
+                error!("H1 fallback: request failed: {}", e);
+                return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H1 fallback request failed"));
+            }
+        };
+
+        // Return sender to pool for keep-alive reuse
+        self.connection_pool.checkin_h1(pool_key.clone(), sender);
+
+        self.build_streaming_response(upstream_response, route, route_id, source_ip).await
+    }
+
     /// Reconnect to a backend (used for H2→H1 fallback).
     async fn reconnect_backend(
         &self,
@@ -1058,6 +1275,7 @@ impl HttpProxyService {
         route: &rustproxy_config::RouteConfig,
         route_id: Option<&str>,
         source_ip: &str,
+        pool_key: Option<&crate::connection_pool::PoolKey>,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
         let mut upstream_req = Request::builder()
             .method(parts.method)
@@ -1083,6 +1301,10 @@ impl HttpProxyService {
             Ok(resp) => resp,
             Err(e) => {
                 error!("HTTP/2 upstream request failed: {}", e);
+                // Evict the dead sender so subsequent requests get fresh connections
+                if let Some(key) = pool_key {
+                    self.connection_pool.remove_h2(key);
+                }
                 return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 request failed"));
             }
         };
@@ -1759,7 +1981,7 @@ impl rustls::client::danger::ServerCertVerifier for InsecureBackendVerifier {
 impl Default for HttpProxyService {
     fn default() -> Self {
         Self {
-            route_manager: Arc::new(RouteManager::new(vec![])),
+            route_manager: Arc::new(ArcSwap::from(Arc::new(RouteManager::new(vec![])))),
             metrics: Arc::new(MetricsCollector::new()),
             upstream_selector: UpstreamSelector::new(),
             connect_timeout: DEFAULT_CONNECT_TIMEOUT,

rust/crates/rustproxy-passthrough/src/tcp_listener.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 use arc_swap::ArcSwap;
+use dashmap::DashMap;
 use tokio::net::TcpListener;
 use tokio_rustls::TlsAcceptor;
 use tokio_util::sync::CancellationToken;
@@ -162,14 +163,18 @@ pub struct TcpListenerManager {
     socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>,
     /// Global connection semaphore — limits total simultaneous connections.
     conn_semaphore: Arc<tokio::sync::Semaphore>,
+    /// Per-route cancellation tokens (child of cancel_token).
+    /// When a route is removed, its token is cancelled, terminating all connections on that route.
+    route_cancels: Arc<DashMap<String, CancellationToken>>,
 }
 
 impl TcpListenerManager {
     pub fn new(route_manager: Arc<RouteManager>) -> Self {
         let metrics = Arc::new(MetricsCollector::new());
         let conn_config = ConnectionConfig::default();
+        let route_manager_swap = Arc::new(ArcSwap::from(route_manager));
         let mut http_proxy_svc = HttpProxyService::with_connect_timeout(
-            Arc::clone(&route_manager),
+            Arc::clone(&route_manager_swap),
             Arc::clone(&metrics),
             std::time::Duration::from_millis(conn_config.connection_timeout_ms),
         );
@@ -188,7 +193,7 @@ impl TcpListenerManager {
         let max_conns = conn_config.max_connections as usize;
         Self {
             listeners: HashMap::new(),
-            route_manager: Arc::new(ArcSwap::from(route_manager)),
+            route_manager: route_manager_swap,
             metrics,
             tls_configs: Arc::new(ArcSwap::from(Arc::new(HashMap::new()))),
             shared_tls_acceptor: Arc::new(ArcSwap::from(Arc::new(None))),
@@ -198,14 +203,16 @@ impl TcpListenerManager {
             cancel_token: CancellationToken::new(),
             socket_handler_relay: Arc::new(std::sync::RwLock::new(None)),
             conn_semaphore: Arc::new(tokio::sync::Semaphore::new(max_conns)),
+            route_cancels: Arc::new(DashMap::new()),
         }
     }
 
     /// Create with a metrics collector.
     pub fn with_metrics(route_manager: Arc<RouteManager>, metrics: Arc<MetricsCollector>) -> Self {
         let conn_config = ConnectionConfig::default();
+        let route_manager_swap = Arc::new(ArcSwap::from(route_manager));
         let mut http_proxy_svc = HttpProxyService::with_connect_timeout(
-            Arc::clone(&route_manager),
+            Arc::clone(&route_manager_swap),
             Arc::clone(&metrics),
             std::time::Duration::from_millis(conn_config.connection_timeout_ms),
         );
@@ -224,7 +231,7 @@ impl TcpListenerManager {
         let max_conns = conn_config.max_connections as usize;
         Self {
             listeners: HashMap::new(),
-            route_manager: Arc::new(ArcSwap::from(route_manager)),
+            route_manager: route_manager_swap,
             metrics,
             tls_configs: Arc::new(ArcSwap::from(Arc::new(HashMap::new()))),
             shared_tls_acceptor: Arc::new(ArcSwap::from(Arc::new(None))),
@@ -234,6 +241,7 @@ impl TcpListenerManager {
             cancel_token: CancellationToken::new(),
             socket_handler_relay: Arc::new(std::sync::RwLock::new(None)),
             conn_semaphore: Arc::new(tokio::sync::Semaphore::new(max_conns)),
+            route_cancels: Arc::new(DashMap::new()),
         }
     }
 
@@ -245,10 +253,9 @@ impl TcpListenerManager {
         ));
         self.conn_semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_connections as usize));
 
-        // Rebuild http_proxy with updated timeouts
+        // Rebuild http_proxy with updated timeouts (shares the same ArcSwap<RouteManager>)
-        let rm = self.route_manager.load_full();
         let mut http_proxy_svc = HttpProxyService::with_connect_timeout(
-            rm,
+            Arc::clone(&self.route_manager),
             Arc::clone(&self.metrics),
             std::time::Duration::from_millis(config.connection_timeout_ms),
         );
@@ -317,12 +324,13 @@ impl TcpListenerManager {
         let cancel = self.cancel_token.clone();
         let relay = Arc::clone(&self.socket_handler_relay);
         let semaphore = Arc::clone(&self.conn_semaphore);
+        let route_cancels = Arc::clone(&self.route_cancels);
 
         let handle = tokio::spawn(async move {
             Self::accept_loop(
                 listener, port, route_manager_swap, metrics, tls_configs,
                 shared_tls_acceptor, http_proxy, conn_config, conn_tracker, cancel, relay,
-                semaphore,
+                semaphore, route_cancels,
             ).await;
         });
 
@@ -401,6 +409,20 @@ impl TcpListenerManager {
         self.route_manager.store(route_manager);
     }
 
+    /// Cancel connections on routes that no longer exist in the active set.
+    /// Existing connections on removed routes are terminated via their per-route CancellationToken.
+    pub fn invalidate_removed_routes(&self, active_route_ids: &std::collections::HashSet<String>) {
+        self.route_cancels.retain(|id, token| {
+            if active_route_ids.contains(id) {
+                true
+            } else {
+                info!("Cancelling connections for removed route '{}'", id);
+                token.cancel();
+                false // remove cancelled token from map
+            }
+        });
+    }
+
     /// Prune HTTP proxy caches for route IDs that are no longer active.
     pub fn prune_http_proxy_caches(&self, active_route_ids: &std::collections::HashSet<String>) {
         self.http_proxy.prune_stale_routes(active_route_ids);
@@ -430,6 +452,7 @@ impl TcpListenerManager {
         cancel: CancellationToken,
         socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>,
         conn_semaphore: Arc<tokio::sync::Semaphore>,
+        route_cancels: Arc<DashMap<String, CancellationToken>>,
     ) {
         loop {
             tokio::select! {
@@ -484,6 +507,7 @@ impl TcpListenerManager {
                             let ct = Arc::clone(&conn_tracker);
                             let cn = cancel.clone();
                             let sr = Arc::clone(&socket_handler_relay);
+                            let rc = Arc::clone(&route_cancels);
                             debug!("Accepted connection from {} on port {}", peer_addr, port);
 
                             tokio::spawn(async move {
@@ -492,7 +516,7 @@ impl TcpListenerManager {
                                 // RAII guard ensures connection_closed is called on all paths
                                 let _ct_guard = ConnectionTrackerGuard::new(ct, ip);
                                 let result = Self::handle_connection(
-                                    stream, port, peer_addr, rm, m, tc, sa, hp, cc, cn, sr,
+                                    stream, port, peer_addr, rm, m, tc, sa, hp, cc, cn, sr, rc,
                                 ).await;
                                 if let Err(e) = result {
                                     debug!("Connection error from {}: {}", peer_addr, e);
@@ -522,6 +546,7 @@ impl TcpListenerManager {
         conn_config: Arc<ConnectionConfig>,
         cancel: CancellationToken,
         socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>,
+        route_cancels: Arc<DashMap<String, CancellationToken>>,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         use tokio::io::AsyncReadExt;
 
@@ -626,6 +651,14 @@ impl TcpListenerManager {
                         let target_port = target.port.resolve(port);
                         let route_id = quick_match.route.id.as_deref();
 
+                        // Resolve per-route cancel token (child of global cancel)
+                        let conn_cancel = match route_id {
+                            Some(id) => route_cancels.entry(id.to_string())
+                                .or_insert_with(|| cancel.child_token())
+                                .clone(),
+                            None => cancel.clone(),
+                        };
+
                         // Check route-level IP security
                         if let Some(ref security) = quick_match.route.security {
                             if !rustproxy_http::request_filter::RequestFilter::check_ip_security(
@@ -680,7 +713,7 @@ impl TcpListenerManager {
 
                             let (_bytes_in, _bytes_out) = forwarder::forward_bidirectional_with_timeouts(
                                 stream, backend_w, None,
-                                inactivity_timeout, max_lifetime, cancel,
+                                inactivity_timeout, max_lifetime, conn_cancel,
                                 Some(forwarder::ForwardMetricsCtx {
                                     collector: Arc::clone(&metrics),
                                     route_id: route_id.map(|s| s.to_string()),
@@ -690,7 +723,7 @@ impl TcpListenerManager {
                         } else {
                             let (_bytes_in, _bytes_out) = forwarder::forward_bidirectional_with_timeouts(
                                 stream, backend, None,
-                                inactivity_timeout, max_lifetime, cancel,
+                                inactivity_timeout, max_lifetime, conn_cancel,
                                 Some(forwarder::ForwardMetricsCtx {
                                     collector: Arc::clone(&metrics),
                                     route_id: route_id.map(|s| s.to_string()),
@@ -795,6 +828,16 @@ impl TcpListenerManager {
 
         let route_id = route_match.route.id.as_deref();
 
+        // Resolve per-route cancel token (child of global cancel).
+        // When this route is removed via updateRoutes, the token is cancelled,
+        // terminating all connections on this route.
+        let cancel = match route_id {
+            Some(id) => route_cancels.entry(id.to_string())
+                .or_insert_with(|| cancel.child_token())
+                .clone(),
+            None => cancel,
+        };
+
         // Check route-level IP security for passthrough connections
         if let Some(ref security) = route_match.route.security {
             if !rustproxy_http::request_filter::RequestFilter::check_ip_security(

rust/crates/rustproxy/src/lib.rs

@@ -610,6 +610,8 @@ impl RustProxy {
         // Update listener manager
         if let Some(ref mut listener) = self.listener_manager {
             listener.update_route_manager(Arc::clone(&new_manager));
+            // Cancel connections on routes that were removed or disabled
+            listener.invalidate_removed_routes(&active_route_ids);
             // Prune HTTP proxy caches (rate limiters, regex cache, round-robin counters)
             listener.prune_http_proxy_caches(&active_route_ids);
 

test/test.sni-requirement.node.ts

@@ -7,10 +7,15 @@
 import { expect, tap } from '@git.zone/tstest/tapbundle';
 import { SmartProxy } from '../ts/proxies/smart-proxy/index.js';
 import type { IRouteConfig } from '../ts/proxies/smart-proxy/models/route-types.js';
+import { findFreePorts } from './helpers/port-allocator.js';
 
-// Use unique high ports for each test to avoid conflicts
+let testPorts: number[];
-let testPort = 20000;
+let portIndex = 0;
-const getNextPort = () => testPort++;
+const getNextPort = () => testPorts[portIndex++];
+
+tap.test('setup - allocate ports', async () => {
+  testPorts = await findFreePorts(16);
+});
 
 // --------------------------------- Single Route, No Domain Restriction ---------------------------------
 

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartproxy',
-  version: '25.9.0',
+  version: '25.9.3',
   description: 'A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.'
 }