Compare commits

..

13 Commits

Author SHA1 Message Date
e69de246e9 v25.8.5
Some checks failed
Default (tags) / security (push) Successful in 43s
Default (tags) / test (push) Failing after 4m1s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-02-26 21:31:38 +00:00
5126049ae6 fix(release): bump patch version (no source changes) 2026-02-26 21:31:38 +00:00
8db621657f fix(proxy): close connection buildup vectors in HTTP idle, WebSocket, socket relay, and TLS forwarding paths
- Add HTTP keep-alive idle timeout (60s default) with periodic watchdog that
  skips active requests (panic-safe via RAII ActiveRequestGuard)
- Make WebSocket inactivity/max-lifetime timeouts configurable from ConnectionConfig
  instead of hardcoded 1h/24h
- Replace bare copy_bidirectional in socket handler relay with timeout+cancel-aware
  split forwarding (inactivity, max lifetime, graceful shutdown)
- Add CancellationToken to forward_bidirectional_split_with_timeouts so TLS-terminated
  TCP connections respond to graceful shutdown
- Fix graceful_stop to actually abort listener tasks that exceed the shutdown deadline
  (previously they detached and ran forever)
- Add 10s metadata parsing timeout on TS socket-handler-server to prevent stuck sockets
2026-02-26 21:29:19 +00:00
ef060d5e79 v25.8.4
Some checks failed
Default (tags) / security (push) Successful in 40s
Default (tags) / test (push) Failing after 4m1s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-02-26 17:32:35 +00:00
cd7f3f7f75 fix(proxy): adjust default proxy timeouts and keep-alive behavior to shorter, more consistent values 2026-02-26 17:32:35 +00:00
8df18728d4 v25.8.3
Some checks failed
Default (tags) / security (push) Successful in 29s
Default (tags) / test (push) Failing after 4m2s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-02-26 17:01:57 +00:00
bedecc6b6b fix(smartproxy): no code or dependency changes detected; no version bump required 2026-02-26 17:01:57 +00:00
b5f166bc92 v25.8.2
Some checks failed
Default (tags) / security (push) Successful in 31s
Default (tags) / test (push) Failing after 4m1s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-02-26 16:58:30 +00:00
94266222fe fix(connection): improve connection handling and timeouts 2026-02-26 16:58:30 +00:00
697d51a9d4 v25.8.1
Some checks failed
Default (tags) / security (push) Successful in 42s
Default (tags) / test (push) Failing after 4m4s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-02-25 00:12:41 +00:00
7e5fe2bec3 fix(allocator): switch global allocator from tikv-jemallocator to mimalloc 2026-02-25 00:12:41 +00:00
f592bf627f v25.8.0
Some checks failed
Default (tags) / security (push) Successful in 42s
Default (tags) / test (push) Failing after 4m1s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-02-24 23:22:49 +00:00
6114a00fb8 feat(rustproxy): use tikv-jemallocator as the global allocator to reduce glibc fragmentation and slow RSS growth; add allocator dependency and enable it in rustproxy, update lockfile, and run tsrust before tests 2026-02-24 23:22:49 +00:00
15 changed files with 353 additions and 82 deletions

View File

@@ -1,5 +1,54 @@
# Changelog # Changelog
## 2026-02-26 - 25.8.5 - fix(release)
bump patch version (no source changes)
- No changes detected in git diff
- Current version: 25.8.4
- Recommend patch bump to 25.8.5 to record release without code changes
## 2026-02-26 - 25.8.4 - fix(proxy)
adjust default proxy timeouts and keep-alive behavior to shorter, more consistent values
- Increase connection timeout default from 30,000ms to 60,000ms (30s -> 60s).
- Reduce socket timeout default from 3,600,000ms to 60,000ms (1h -> 60s).
- Reduce max connection lifetime default from 86,400,000ms to 3,600,000ms (24h -> 1h).
- Change inactivity timeout default from 14,400,000ms to 75,000ms (4h -> 75s).
- Update keep-alive defaults: keepAliveTreatment 'extended' -> 'standard', keepAliveInactivityMultiplier 6 -> 4, extendedKeepAliveLifetime 604800000 -> 3,600,000ms (7d -> 1h).
- Apply these consistent default values across Rust crates (rustproxy-config, rustproxy-passthrough) and the TypeScript smart-proxy implementation.
- Update unit test expectations to match the new defaults.
## 2026-02-26 - 25.8.3 - fix(smartproxy)
no code or dependency changes detected; no version bump required
- No files changed in the provided diff (No changes).
- package.json version remains 25.8.2.
- No dependency or source updates detected; skip release.
## 2026-02-26 - 25.8.2 - fix(connection)
improve connection handling and timeouts
- Flush logs on process beforeExit and avoid calling process.exit in SIGINT/SIGTERM handlers to preserve host graceful shutdown
- Store protocol entries with a createdAt timestamp in ProtocolDetector and remove stale entries older than 30s to prevent leaked state from abandoned handshakes or port scanners
- Add backend connect timeout (30s) and idle timeouts (5 minutes) for dynamic forwards; destroy sockets on timeout and emit logs for timeout events
## 2026-02-25 - 25.8.1 - fix(allocator)
switch global allocator from tikv-jemallocator to mimalloc
- Replaced tikv-jemallocator with mimalloc in rust/Cargo.toml workspace dependencies.
- Updated rust/crates/rustproxy/Cargo.toml to use mimalloc as a workspace dependency.
- Updated rust/Cargo.lock: added mimalloc and libmimalloc-sys entries and removed tikv-jemallocator and tikv-jemalloc-sys entries.
- Changed the global allocator in crates/rustproxy/src/main.rs from tikv_jemallocator::Jemalloc to mimalloc::MiMalloc.
- Impact: runtime memory allocator is changed which may affect memory usage and performance; no public API changes but recommend testing memory/performance in deployments.
## 2026-02-24 - 25.8.0 - feat(rustproxy)
use tikv-jemallocator as the global allocator to reduce glibc fragmentation and slow RSS growth; add allocator dependency and enable it in rustproxy, update lockfile, and run tsrust before tests
- Added tikv-jemallocator dependency to rust/Cargo.toml and rust/crates/rustproxy/Cargo.toml
- Enabled tikv_jemallocator as the global allocator in rust/crates/rustproxy/src/main.rs
- Updated rust/Cargo.lock with tikv-jemallocator and tikv-jemalloc-sys entries
- Modified package.json test script to run tsrust before tstest
## 2026-02-24 - 25.7.10 - fix(rustproxy) ## 2026-02-24 - 25.7.10 - fix(rustproxy)
Use cooperative cancellation for background tasks, prune stale caches and metric entries, and switch tests to dynamic port allocation to avoid port conflicts Use cooperative cancellation for background tasks, prune stale caches and metric entries, and switch tests to dynamic port allocation to avoid port conflicts

View File

@@ -1,6 +1,6 @@
{ {
"name": "@push.rocks/smartproxy", "name": "@push.rocks/smartproxy",
"version": "25.7.10", "version": "25.8.5",
"private": false, "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.", "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", "main": "dist_ts/index.js",
@@ -9,7 +9,7 @@
"author": "Lossless GmbH", "author": "Lossless GmbH",
"license": "MIT", "license": "MIT",
"scripts": { "scripts": {
"test": "(tstest test/**/test*.ts --verbose --timeout 60 --logfile)", "test": "(tsrust) && (tstest test/**/test*.ts --verbose --timeout 60 --logfile)",
"build": "(tsbuild tsfolders --allowimplicitany) && (tsrust)", "build": "(tsbuild tsfolders --allowimplicitany) && (tsrust)",
"format": "(gitzone format)", "format": "(gitzone format)",
"buildDocs": "tsdoc" "buildDocs": "tsdoc"

20
rust/Cargo.lock generated
View File

@@ -612,6 +612,16 @@ version = "0.2.180"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc" checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc"
[[package]]
name = "libmimalloc-sys"
version = "0.1.44"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "667f4fec20f29dfc6bc7357c582d91796c169ad7e2fce709468aefeb2c099870"
dependencies = [
"cc",
"libc",
]
[[package]] [[package]]
name = "lock_api" name = "lock_api"
version = "0.4.14" version = "0.4.14"
@@ -642,6 +652,15 @@ version = "2.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
[[package]]
name = "mimalloc"
version = "0.1.48"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1ee66a4b64c74f4ef288bcbb9192ad9c3feaad75193129ac8509af543894fd8"
dependencies = [
"libmimalloc-sys",
]
[[package]] [[package]]
name = "mio" name = "mio"
version = "1.1.1" version = "1.1.1"
@@ -924,6 +943,7 @@ dependencies = [
"http-body-util", "http-body-util",
"hyper", "hyper",
"hyper-util", "hyper-util",
"mimalloc",
"rcgen", "rcgen",
"rustls", "rustls",
"rustproxy-config", "rustproxy-config",

View File

@@ -91,6 +91,9 @@ libc = "0.2"
# Socket-level options (keepalive, etc.) # Socket-level options (keepalive, etc.)
socket2 = { version = "0.5", features = ["all"] } socket2 = { version = "0.5", features = ["all"] }
# mimalloc allocator (prevents glibc fragmentation / slow RSS growth)
mimalloc = "0.1"
# Internal crates # Internal crates
rustproxy-config = { path = "crates/rustproxy-config" } rustproxy-config = { path = "crates/rustproxy-config" }
rustproxy-routing = { path = "crates/rustproxy-routing" } rustproxy-routing = { path = "crates/rustproxy-routing" }

View File

@@ -298,7 +298,7 @@ impl RustProxyOptions {
/// Get the effective connection timeout in milliseconds. /// Get the effective connection timeout in milliseconds.
pub fn effective_connection_timeout(&self) -> u64 { pub fn effective_connection_timeout(&self) -> u64 {
self.connection_timeout.unwrap_or(30_000) self.connection_timeout.unwrap_or(60_000)
} }
/// Get the effective initial data timeout in milliseconds. /// Get the effective initial data timeout in milliseconds.
@@ -308,12 +308,12 @@ impl RustProxyOptions {
/// Get the effective socket timeout in milliseconds. /// Get the effective socket timeout in milliseconds.
pub fn effective_socket_timeout(&self) -> u64 { pub fn effective_socket_timeout(&self) -> u64 {
self.socket_timeout.unwrap_or(3_600_000) self.socket_timeout.unwrap_or(60_000)
} }
/// Get the effective max connection lifetime in milliseconds. /// Get the effective max connection lifetime in milliseconds.
pub fn effective_max_connection_lifetime(&self) -> u64 { pub fn effective_max_connection_lifetime(&self) -> u64 {
self.max_connection_lifetime.unwrap_or(86_400_000) self.max_connection_lifetime.unwrap_or(3_600_000)
} }
/// Get all unique ports that routes listen on. /// Get all unique ports that routes listen on.
@@ -377,10 +377,10 @@ mod tests {
#[test] #[test]
fn test_default_timeouts() { fn test_default_timeouts() {
let options = RustProxyOptions::default(); let options = RustProxyOptions::default();
assert_eq!(options.effective_connection_timeout(), 30_000); assert_eq!(options.effective_connection_timeout(), 60_000);
assert_eq!(options.effective_initial_data_timeout(), 60_000); assert_eq!(options.effective_initial_data_timeout(), 60_000);
assert_eq!(options.effective_socket_timeout(), 3_600_000); assert_eq!(options.effective_socket_timeout(), 60_000);
assert_eq!(options.effective_max_connection_lifetime(), 86_400_000); assert_eq!(options.effective_max_connection_lifetime(), 3_600_000);
} }
#[test] #[test]

View File

@@ -34,12 +34,35 @@ use crate::upstream_selector::UpstreamSelector;
/// Default upstream connect timeout (30 seconds). /// Default upstream connect timeout (30 seconds).
const DEFAULT_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); const DEFAULT_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
/// Default HTTP keep-alive idle timeout (60 seconds).
/// If no new request arrives within this duration, the connection is closed.
const DEFAULT_HTTP_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
/// Default WebSocket inactivity timeout (1 hour). /// Default WebSocket inactivity timeout (1 hour).
const DEFAULT_WS_INACTIVITY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3600); const DEFAULT_WS_INACTIVITY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3600);
/// Default WebSocket max lifetime (24 hours). /// Default WebSocket max lifetime (24 hours).
const DEFAULT_WS_MAX_LIFETIME: std::time::Duration = std::time::Duration::from_secs(86400); const DEFAULT_WS_MAX_LIFETIME: std::time::Duration = std::time::Duration::from_secs(86400);
/// RAII guard that decrements the active request counter on drop.
/// Ensures the counter is correct even if the request handler panics.
struct ActiveRequestGuard {
counter: Arc<AtomicU64>,
}
impl ActiveRequestGuard {
fn new(counter: Arc<AtomicU64>) -> Self {
counter.fetch_add(1, Ordering::Relaxed);
Self { counter }
}
}
impl Drop for ActiveRequestGuard {
fn drop(&mut self) {
self.counter.fetch_sub(1, Ordering::Relaxed);
}
}
/// Backend stream that can be either plain TCP or TLS-wrapped. /// Backend stream that can be either plain TCP or TLS-wrapped.
/// Used for `terminate-and-reencrypt` mode where the backend requires TLS. /// Used for `terminate-and-reencrypt` mode where the backend requires TLS.
pub(crate) enum BackendStream { pub(crate) enum BackendStream {
@@ -125,6 +148,12 @@ pub struct HttpProxyService {
backend_tls_config: Arc<rustls::ClientConfig>, backend_tls_config: Arc<rustls::ClientConfig>,
/// Backend connection pool for reusing keep-alive connections. /// Backend connection pool for reusing keep-alive connections.
connection_pool: Arc<crate::connection_pool::ConnectionPool>, connection_pool: Arc<crate::connection_pool::ConnectionPool>,
/// HTTP keep-alive idle timeout: close connection if no new request arrives within this duration.
http_idle_timeout: std::time::Duration,
/// WebSocket inactivity timeout (no data in either direction).
ws_inactivity_timeout: std::time::Duration,
/// WebSocket maximum connection lifetime.
ws_max_lifetime: std::time::Duration,
} }
impl HttpProxyService { impl HttpProxyService {
@@ -139,6 +168,9 @@ impl HttpProxyService {
regex_cache: DashMap::new(), regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(), backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()), connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
http_idle_timeout: DEFAULT_HTTP_IDLE_TIMEOUT,
ws_inactivity_timeout: DEFAULT_WS_INACTIVITY_TIMEOUT,
ws_max_lifetime: DEFAULT_WS_MAX_LIFETIME,
} }
} }
@@ -158,9 +190,25 @@ impl HttpProxyService {
regex_cache: DashMap::new(), regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(), backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()), connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
http_idle_timeout: DEFAULT_HTTP_IDLE_TIMEOUT,
ws_inactivity_timeout: DEFAULT_WS_INACTIVITY_TIMEOUT,
ws_max_lifetime: DEFAULT_WS_MAX_LIFETIME,
} }
} }
/// Set the HTTP keep-alive idle timeout, WebSocket inactivity timeout, and
/// WebSocket max lifetime from connection config values.
pub fn set_connection_timeouts(
&mut self,
http_idle_timeout: std::time::Duration,
ws_inactivity_timeout: std::time::Duration,
ws_max_lifetime: std::time::Duration,
) {
self.http_idle_timeout = http_idle_timeout;
self.ws_inactivity_timeout = ws_inactivity_timeout;
self.ws_max_lifetime = ws_max_lifetime;
}
/// Set the shared backend TLS config (enables session resumption). /// Set the shared backend TLS config (enables session resumption).
/// Call this after construction to inject the shared config from tls_handler. /// Call this after construction to inject the shared config from tls_handler.
pub fn set_backend_tls_config(&mut self, config: Arc<rustls::ClientConfig>) { pub fn set_backend_tls_config(&mut self, config: Arc<rustls::ClientConfig>) {
@@ -192,6 +240,10 @@ impl HttpProxyService {
/// based on ALPN negotiation (TLS) or connection preface (h2c). /// based on ALPN negotiation (TLS) or connection preface (h2c).
/// Supports HTTP/1.1 upgrades (WebSocket) and HTTP/2 CONNECT. /// Supports HTTP/1.1 upgrades (WebSocket) and HTTP/2 CONNECT.
/// Responds to graceful shutdown via the cancel token. /// Responds to graceful shutdown via the cancel token.
///
/// An idle watchdog closes the connection if no new HTTP request arrives
/// within `http_idle_timeout` (default 60s). This prevents keep-alive
/// connections from accumulating indefinitely.
pub async fn handle_io<I>( pub async fn handle_io<I>(
self: Arc<Self>, self: Arc<Self>,
stream: I, stream: I,
@@ -204,13 +256,34 @@ impl HttpProxyService {
{ {
let io = TokioIo::new(stream); let io = TokioIo::new(stream);
// Capture timeouts before `self` is moved into the service closure.
let idle_timeout = self.http_idle_timeout;
// Activity tracker: updated at the START and END of each request.
// The idle watchdog checks this to determine if the connection is idle
// (no request in progress and none started recently).
let last_activity = Arc::new(AtomicU64::new(0));
let active_requests = Arc::new(AtomicU64::new(0));
let start = std::time::Instant::now();
let la_inner = Arc::clone(&last_activity);
let ar_inner = Arc::clone(&active_requests);
let cancel_inner = cancel.clone(); let cancel_inner = cancel.clone();
let service = hyper::service::service_fn(move |req: Request<Incoming>| { let service = hyper::service::service_fn(move |req: Request<Incoming>| {
// Mark request start — RAII guard decrements on drop (panic-safe)
la_inner.store(start.elapsed().as_millis() as u64, Ordering::Relaxed);
let req_guard = ActiveRequestGuard::new(Arc::clone(&ar_inner));
let svc = Arc::clone(&self); let svc = Arc::clone(&self);
let peer = peer_addr; let peer = peer_addr;
let cn = cancel_inner.clone(); let cn = cancel_inner.clone();
let la = Arc::clone(&la_inner);
let st = start;
async move { async move {
svc.handle_request(req, peer, port, cn).await let result = svc.handle_request(req, peer, port, cn).await;
// Mark request end — update activity timestamp before guard drops
la.store(st.elapsed().as_millis() as u64, Ordering::Relaxed);
drop(req_guard); // Explicitly drop to decrement active_requests
result
} }
}); });
@@ -221,7 +294,7 @@ impl HttpProxyService {
// Pin on the heap — auto::UpgradeableConnection is !Unpin // Pin on the heap — auto::UpgradeableConnection is !Unpin
let mut conn = Box::pin(conn); let mut conn = Box::pin(conn);
// Use select to support graceful shutdown via cancellation token // Use select to support graceful shutdown, cancellation, and idle timeout
tokio::select! { tokio::select! {
result = conn.as_mut() => { result = conn.as_mut() => {
if let Err(e) = result { if let Err(e) = result {
@@ -235,6 +308,37 @@ impl HttpProxyService {
debug!("HTTP connection error during shutdown from {}: {}", peer_addr, e); debug!("HTTP connection error during shutdown from {}: {}", peer_addr, e);
} }
} }
_ = async {
// Idle watchdog: check every 5s whether the connection has been idle
// (no active requests AND no activity for idle_timeout).
// This avoids killing long-running requests or upgraded connections.
let check_interval = std::time::Duration::from_secs(5);
let mut last_seen = 0u64;
loop {
tokio::time::sleep(check_interval).await;
// Never close while a request is in progress
if active_requests.load(Ordering::Relaxed) > 0 {
last_seen = last_activity.load(Ordering::Relaxed);
continue;
}
let current = last_activity.load(Ordering::Relaxed);
if current == last_seen {
// No new activity since last check
let elapsed_since_activity = start.elapsed().as_millis() as u64 - current;
if elapsed_since_activity >= idle_timeout.as_millis() as u64 {
return;
}
}
last_seen = current;
}
} => {
debug!("HTTP connection idle timeout ({}s) from {}", idle_timeout.as_secs(), peer_addr);
conn.as_mut().graceful_shutdown();
// Give any in-flight work 5s to drain after graceful shutdown
let _ = tokio::time::timeout(std::time::Duration::from_secs(5), conn).await;
}
} }
} }
@@ -1022,6 +1126,8 @@ impl HttpProxyService {
let source_ip_owned = source_ip.to_string(); let source_ip_owned = source_ip.to_string();
let upstream_selector = self.upstream_selector.clone(); let upstream_selector = self.upstream_selector.clone();
let upstream_key_owned = upstream_key.to_string(); let upstream_key_owned = upstream_key.to_string();
let ws_inactivity_timeout = self.ws_inactivity_timeout;
let ws_max_lifetime = self.ws_max_lifetime;
tokio::spawn(async move { tokio::spawn(async move {
let client_upgraded = match on_client_upgrade.await { let client_upgraded = match on_client_upgrade.await {
@@ -1084,8 +1190,8 @@ impl HttpProxyService {
let la_watch = Arc::clone(&last_activity); let la_watch = Arc::clone(&last_activity);
let c2u_handle = c2u.abort_handle(); let c2u_handle = c2u.abort_handle();
let u2c_handle = u2c.abort_handle(); let u2c_handle = u2c.abort_handle();
let inactivity_timeout = DEFAULT_WS_INACTIVITY_TIMEOUT; let inactivity_timeout = ws_inactivity_timeout;
let max_lifetime = DEFAULT_WS_MAX_LIFETIME; let max_lifetime = ws_max_lifetime;
let watchdog = tokio::spawn(async move { let watchdog = tokio::spawn(async move {
let check_interval = std::time::Duration::from_secs(5); let check_interval = std::time::Duration::from_secs(5);
@@ -1391,6 +1497,9 @@ impl Default for HttpProxyService {
regex_cache: DashMap::new(), regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(), backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()), connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
http_idle_timeout: DEFAULT_HTTP_IDLE_TIMEOUT,
ws_inactivity_timeout: DEFAULT_WS_INACTIVITY_TIMEOUT,
ws_max_lifetime: DEFAULT_WS_MAX_LIFETIME,
} }
} }
} }

View File

@@ -118,10 +118,10 @@ pub struct ConnectionConfig {
impl Default for ConnectionConfig { impl Default for ConnectionConfig {
fn default() -> Self { fn default() -> Self {
Self { Self {
connection_timeout_ms: 30_000, connection_timeout_ms: 60_000,
initial_data_timeout_ms: 60_000, initial_data_timeout_ms: 60_000,
socket_timeout_ms: 3_600_000, socket_timeout_ms: 60_000,
max_connection_lifetime_ms: 86_400_000, max_connection_lifetime_ms: 3_600_000,
graceful_shutdown_timeout_ms: 30_000, graceful_shutdown_timeout_ms: 30_000,
max_connections_per_ip: None, max_connections_per_ip: None,
connection_rate_limit_per_minute: None, connection_rate_limit_per_minute: None,
@@ -174,6 +174,11 @@ impl TcpListenerManager {
std::time::Duration::from_millis(conn_config.connection_timeout_ms), std::time::Duration::from_millis(conn_config.connection_timeout_ms),
); );
http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config()); http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config());
http_proxy_svc.set_connection_timeouts(
std::time::Duration::from_millis(conn_config.socket_timeout_ms),
std::time::Duration::from_millis(conn_config.socket_timeout_ms),
std::time::Duration::from_millis(conn_config.max_connection_lifetime_ms),
);
let http_proxy = Arc::new(http_proxy_svc); let http_proxy = Arc::new(http_proxy_svc);
let conn_tracker = Arc::new(ConnectionTracker::new( let conn_tracker = Arc::new(ConnectionTracker::new(
conn_config.max_connections_per_ip, conn_config.max_connections_per_ip,
@@ -204,6 +209,11 @@ impl TcpListenerManager {
std::time::Duration::from_millis(conn_config.connection_timeout_ms), std::time::Duration::from_millis(conn_config.connection_timeout_ms),
); );
http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config()); http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config());
http_proxy_svc.set_connection_timeouts(
std::time::Duration::from_millis(conn_config.socket_timeout_ms),
std::time::Duration::from_millis(conn_config.socket_timeout_ms),
std::time::Duration::from_millis(conn_config.max_connection_lifetime_ms),
);
let http_proxy = Arc::new(http_proxy_svc); let http_proxy = Arc::new(http_proxy_svc);
let conn_tracker = Arc::new(ConnectionTracker::new( let conn_tracker = Arc::new(ConnectionTracker::new(
conn_config.max_connections_per_ip, conn_config.max_connections_per_ip,
@@ -232,6 +242,22 @@ impl TcpListenerManager {
config.connection_rate_limit_per_minute, config.connection_rate_limit_per_minute,
)); ));
self.conn_semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_connections as usize)); self.conn_semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_connections as usize));
// Rebuild http_proxy with updated timeouts
let rm = self.route_manager.load_full();
let mut http_proxy_svc = HttpProxyService::with_connect_timeout(
rm,
Arc::clone(&self.metrics),
std::time::Duration::from_millis(config.connection_timeout_ms),
);
http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config());
http_proxy_svc.set_connection_timeouts(
std::time::Duration::from_millis(config.socket_timeout_ms),
std::time::Duration::from_millis(config.socket_timeout_ms),
std::time::Duration::from_millis(config.max_connection_lifetime_ms),
);
self.http_proxy = Arc::new(http_proxy_svc);
self.conn_config = Arc::new(config); self.conn_config = Arc::new(config);
} }
@@ -336,13 +362,15 @@ impl TcpListenerManager {
for (port, handle) in self.listeners.drain() { for (port, handle) in self.listeners.drain() {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
let abort_handle = handle.abort_handle();
if remaining.is_zero() { if remaining.is_zero() {
handle.abort(); abort_handle.abort();
warn!("Force-stopped listener on port {} (timeout exceeded)", port); warn!("Force-stopped listener on port {} (timeout exceeded)", port);
} else { } else {
match tokio::time::timeout(remaining, handle).await { match tokio::time::timeout(remaining, handle).await {
Ok(_) => info!("Listener on port {} stopped gracefully", port), Ok(_) => info!("Listener on port {} stopped gracefully", port),
Err(_) => { Err(_) => {
abort_handle.abort();
warn!("Listener on port {} did not stop in time, aborting", port); warn!("Listener on port {} did not stop in time, aborting", port);
} }
} }
@@ -791,7 +819,8 @@ impl TcpListenerManager {
stream, n, port, peer_addr, stream, n, port, peer_addr,
&route_match, domain.as_deref(), is_tls, &route_match, domain.as_deref(), is_tls,
&relay_socket_path, &relay_socket_path,
&metrics, route_id, Arc::clone(&metrics), route_id,
&conn_config, cancel.clone(),
).await; ).await;
} else { } else {
debug!("Socket-handler route matched but no relay path configured"); debug!("Socket-handler route matched but no relay path configured");
@@ -964,7 +993,7 @@ impl TcpListenerManager {
let (_bytes_in, _bytes_out) = Self::forward_bidirectional_split_with_timeouts( let (_bytes_in, _bytes_out) = Self::forward_bidirectional_split_with_timeouts(
tls_read, tls_write, backend_read, backend_write, tls_read, tls_write, backend_read, backend_write,
inactivity_timeout, max_lifetime, inactivity_timeout, max_lifetime, cancel.clone(),
Some(forwarder::ForwardMetricsCtx { Some(forwarder::ForwardMetricsCtx {
collector: Arc::clone(&metrics), collector: Arc::clone(&metrics),
route_id: route_id.map(|s| s.to_string()), route_id: route_id.map(|s| s.to_string()),
@@ -1023,7 +1052,7 @@ impl TcpListenerManager {
Self::handle_tls_reencrypt_tunnel( Self::handle_tls_reencrypt_tunnel(
buf_stream, &target_host, target_port, buf_stream, &target_host, target_port,
peer_addr, Arc::clone(&metrics), route_id, peer_addr, Arc::clone(&metrics), route_id,
&conn_config, &conn_config, cancel.clone(),
).await?; ).await?;
} }
Ok(()) Ok(())
@@ -1100,8 +1129,10 @@ impl TcpListenerManager {
domain: Option<&str>, domain: Option<&str>,
is_tls: bool, is_tls: bool,
relay_path: &str, relay_path: &str,
metrics: &MetricsCollector, metrics: Arc<MetricsCollector>,
route_id: Option<&str>, route_id: Option<&str>,
conn_config: &ConnectionConfig,
cancel: CancellationToken,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixStream; use tokio::net::UnixStream;
@@ -1141,27 +1172,34 @@ impl TcpListenerManager {
// Forward initial data to the Unix socket // Forward initial data to the Unix socket
unix_stream.write_all(&initial_buf).await?; unix_stream.write_all(&initial_buf).await?;
// Bidirectional relay between TCP client and Unix socket handler // Bidirectional relay with inactivity timeout, max lifetime, and cancellation.
// Split both streams and use the same watchdog pattern as other forwarding paths.
let initial_len = initial_buf.len() as u64; let initial_len = initial_buf.len() as u64;
match tokio::io::copy_bidirectional(&mut stream, &mut unix_stream).await { let inactivity_timeout = std::time::Duration::from_millis(conn_config.socket_timeout_ms);
Ok((c2s, s2c)) => { let max_lifetime = std::time::Duration::from_millis(conn_config.max_connection_lifetime_ms);
// Include initial data bytes that were forwarded before copy_bidirectional
let total_in = c2s + initial_len; let (tcp_read, tcp_write) = stream.into_split();
debug!("Socket handler relay complete for {}: {} bytes in, {} bytes out", let (unix_read, unix_write) = unix_stream.into_split();
route_key, total_in, s2c);
let ip = peer_addr.ip().to_string(); let ip_str = peer_addr.ip().to_string();
metrics.record_bytes(total_in, s2c, route_id, Some(&ip)); let (_bytes_in, _bytes_out) = Self::forward_bidirectional_split_with_timeouts(
} tcp_read, tcp_write, unix_read, unix_write,
Err(e) => { inactivity_timeout, max_lifetime, cancel,
// Still record the initial data even on error Some(forwarder::ForwardMetricsCtx {
if initial_len > 0 { collector: Arc::clone(&metrics),
let ip = peer_addr.ip().to_string(); route_id: route_id.map(|s| s.to_string()),
metrics.record_bytes(initial_len, 0, route_id, Some(&ip)); source_ip: Some(ip_str.clone()),
} }),
debug!("Socket handler relay ended for {}: {}", route_key, e); ).await;
}
// Include the initial data that was forwarded before the bidirectional relay
if initial_len > 0 {
metrics.record_bytes(initial_len, 0, route_id, Some(&ip_str));
} }
debug!("Socket handler relay complete for {}: {} bytes in, {} bytes out",
route_key, _bytes_in + initial_len, _bytes_out);
Ok(()) Ok(())
} }
@@ -1176,6 +1214,7 @@ impl TcpListenerManager {
metrics: Arc<MetricsCollector>, metrics: Arc<MetricsCollector>,
route_id: Option<&str>, route_id: Option<&str>,
conn_config: &ConnectionConfig, conn_config: &ConnectionConfig,
cancel: CancellationToken,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// Connect to backend over TLS with timeout // Connect to backend over TLS with timeout
let backend_tls = match tokio::time::timeout( let backend_tls = match tokio::time::timeout(
@@ -1220,7 +1259,7 @@ impl TcpListenerManager {
let (_bytes_in, _bytes_out) = Self::forward_bidirectional_split_with_timeouts( let (_bytes_in, _bytes_out) = Self::forward_bidirectional_split_with_timeouts(
client_read, client_write, backend_read, backend_write, client_read, client_write, backend_read, backend_write,
inactivity_timeout, max_lifetime, inactivity_timeout, max_lifetime, cancel,
Some(forwarder::ForwardMetricsCtx { Some(forwarder::ForwardMetricsCtx {
collector: metrics, collector: metrics,
route_id: route_id.map(|s| s.to_string()), route_id: route_id.map(|s| s.to_string()),
@@ -1295,6 +1334,7 @@ impl TcpListenerManager {
mut backend_write: W2, mut backend_write: W2,
inactivity_timeout: std::time::Duration, inactivity_timeout: std::time::Duration,
max_lifetime: std::time::Duration, max_lifetime: std::time::Duration,
cancel: CancellationToken,
metrics: Option<forwarder::ForwardMetricsCtx>, metrics: Option<forwarder::ForwardMetricsCtx>,
) -> (u64, u64) ) -> (u64, u64)
where where
@@ -1362,7 +1402,7 @@ impl TcpListenerManager {
total total
}); });
// Watchdog task: check for inactivity and max lifetime // Watchdog task: check for inactivity, max lifetime, and cancellation
let la_watch = Arc::clone(&last_activity); let la_watch = Arc::clone(&last_activity);
let c2b_handle = c2b.abort_handle(); let c2b_handle = c2b.abort_handle();
let b2c_handle = b2c.abort_handle(); let b2c_handle = b2c.abort_handle();
@@ -1370,29 +1410,37 @@ impl TcpListenerManager {
let check_interval = std::time::Duration::from_secs(5); let check_interval = std::time::Duration::from_secs(5);
let mut last_seen = 0u64; let mut last_seen = 0u64;
loop { loop {
tokio::time::sleep(check_interval).await; tokio::select! {
_ = cancel.cancelled() => {
// Check max lifetime debug!("Split-stream connection cancelled by shutdown");
if start.elapsed() >= max_lifetime {
debug!("Connection exceeded max lifetime, closing");
c2b_handle.abort();
b2c_handle.abort();
break;
}
// Check inactivity
let current = la_watch.load(Ordering::Relaxed);
if current == last_seen {
// No activity since last check
let elapsed_since_activity = start.elapsed().as_millis() as u64 - current;
if elapsed_since_activity >= inactivity_timeout.as_millis() as u64 {
debug!("Connection inactive for {}ms, closing", elapsed_since_activity);
c2b_handle.abort(); c2b_handle.abort();
b2c_handle.abort(); b2c_handle.abort();
break; break;
} }
_ = tokio::time::sleep(check_interval) => {
// Check max lifetime
if start.elapsed() >= max_lifetime {
debug!("Connection exceeded max lifetime, closing");
c2b_handle.abort();
b2c_handle.abort();
break;
}
// Check inactivity
let current = la_watch.load(Ordering::Relaxed);
if current == last_seen {
// No activity since last check
let elapsed_since_activity = start.elapsed().as_millis() as u64 - current;
if elapsed_since_activity >= inactivity_timeout.as_millis() as u64 {
debug!("Connection inactive for {}ms, closing", elapsed_since_activity);
c2b_handle.abort();
b2c_handle.abort();
break;
}
}
last_seen = current;
}
} }
last_seen = current;
} }
}); });

View File

@@ -39,6 +39,7 @@ hyper = { workspace = true }
hyper-util = { workspace = true } hyper-util = { workspace = true }
http-body-util = { workspace = true } http-body-util = { workspace = true }
bytes = { workspace = true } bytes = { workspace = true }
mimalloc = { workspace = true }
[dev-dependencies] [dev-dependencies]
rcgen = { workspace = true } rcgen = { workspace = true }

View File

@@ -1,3 +1,6 @@
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
use clap::Parser; use clap::Parser;
use tracing_subscriber::EnvFilter; use tracing_subscriber::EnvFilter;
use anyhow::Result; use anyhow::Result;

View File

@@ -3,6 +3,6 @@
*/ */
export const commitinfo = { export const commitinfo = {
name: '@push.rocks/smartproxy', name: '@push.rocks/smartproxy',
version: '25.7.10', version: '25.8.5',
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.' 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.'
} }

View File

@@ -354,17 +354,17 @@ export class LogDeduplicator {
// Global instance for connection-related log deduplication // Global instance for connection-related log deduplication
export const connectionLogDeduplicator = new LogDeduplicator(5000); // 5 second batches export const connectionLogDeduplicator = new LogDeduplicator(5000); // 5 second batches
// Ensure logs are flushed on process exit // Ensure logs are flushed on process exit.
// Only use beforeExit — do NOT call process.exit() from SIGINT/SIGTERM handlers
// as that kills the host process's graceful shutdown (e.g., dcrouter connection draining).
process.on('beforeExit', () => { process.on('beforeExit', () => {
connectionLogDeduplicator.flushAll(); connectionLogDeduplicator.flushAll();
}); });
process.on('SIGINT', () => { process.on('SIGINT', () => {
connectionLogDeduplicator.cleanup(); connectionLogDeduplicator.cleanup();
process.exit(0);
}); });
process.on('SIGTERM', () => { process.on('SIGTERM', () => {
connectionLogDeduplicator.cleanup(); connectionLogDeduplicator.cleanup();
process.exit(0);
}); });

View File

@@ -18,8 +18,8 @@ export class ProtocolDetector {
private fragmentManager: DetectionFragmentManager; private fragmentManager: DetectionFragmentManager;
private tlsDetector: TlsDetector; private tlsDetector: TlsDetector;
private httpDetector: HttpDetector; private httpDetector: HttpDetector;
private connectionProtocols: Map<string, 'tls' | 'http'> = new Map(); private connectionProtocols: Map<string, { protocol: 'tls' | 'http'; createdAt: number }> = new Map();
constructor() { constructor() {
this.fragmentManager = new DetectionFragmentManager(); this.fragmentManager = new DetectionFragmentManager();
this.tlsDetector = new TlsDetector(); this.tlsDetector = new TlsDetector();
@@ -124,8 +124,9 @@ export class ProtocolDetector {
const connectionId = DetectionFragmentManager.createConnectionId(context); const connectionId = DetectionFragmentManager.createConnectionId(context);
// Check if we already know the protocol for this connection // Check if we already know the protocol for this connection
const knownProtocol = this.connectionProtocols.get(connectionId); const knownEntry = this.connectionProtocols.get(connectionId);
const knownProtocol = knownEntry?.protocol;
if (knownProtocol === 'http') { if (knownProtocol === 'http') {
const result = this.httpDetector.detectWithContext(buffer, context, options); const result = this.httpDetector.detectWithContext(buffer, context, options);
if (result) { if (result) {
@@ -163,7 +164,7 @@ export class ProtocolDetector {
if (!knownProtocol) { if (!knownProtocol) {
// First peek to determine protocol type // First peek to determine protocol type
if (this.tlsDetector.canHandle(buffer)) { if (this.tlsDetector.canHandle(buffer)) {
this.connectionProtocols.set(connectionId, 'tls'); this.connectionProtocols.set(connectionId, { protocol: 'tls', createdAt: Date.now() });
// Handle TLS with fragment accumulation // Handle TLS with fragment accumulation
const handler = this.fragmentManager.getHandler('tls'); const handler = this.fragmentManager.getHandler('tls');
const fragmentResult = handler.addFragment(connectionId, buffer); const fragmentResult = handler.addFragment(connectionId, buffer);
@@ -189,7 +190,7 @@ export class ProtocolDetector {
} }
if (this.httpDetector.canHandle(buffer)) { if (this.httpDetector.canHandle(buffer)) {
this.connectionProtocols.set(connectionId, 'http'); this.connectionProtocols.set(connectionId, { protocol: 'http', createdAt: Date.now() });
const result = this.httpDetector.detectWithContext(buffer, context, options); const result = this.httpDetector.detectWithContext(buffer, context, options);
if (result) { if (result) {
if (result.isComplete) { if (result.isComplete) {
@@ -221,6 +222,14 @@ export class ProtocolDetector {
private cleanupInstance(): void { private cleanupInstance(): void {
this.fragmentManager.cleanup(); this.fragmentManager.cleanup();
// Remove stale connectionProtocols entries (abandoned handshakes, port scanners)
const maxAge = 30_000; // 30 seconds
const now = Date.now();
for (const [id, entry] of this.connectionProtocols) {
if (now - entry.createdAt > maxAge) {
this.connectionProtocols.delete(id);
}
}
} }
/** /**
@@ -242,8 +251,7 @@ export class ProtocolDetector {
* @param _maxAge Maximum age in milliseconds (default: 30 seconds) * @param _maxAge Maximum age in milliseconds (default: 30 seconds)
*/ */
static cleanupConnections(_maxAge: number = 30000): void { static cleanupConnections(_maxAge: number = 30000): void {
// Cleanup is now handled internally by the fragment manager this.getInstance().cleanupInstance();
this.getInstance().fragmentManager.cleanup();
} }
/** /**

View File

@@ -112,12 +112,12 @@ export interface ISmartProxyOptions {
maxVersion?: string; maxVersion?: string;
// Timeout settings // Timeout settings
connectionTimeout?: number; // Timeout for establishing connection to backend (ms), default: 30000 (30s) connectionTimeout?: number; // Timeout for establishing connection to backend (ms), default: 60000 (60s)
initialDataTimeout?: number; // Timeout for initial data/SNI (ms), default: 60000 (60s) initialDataTimeout?: number; // Timeout for initial data/SNI (ms), default: 60000 (60s)
socketTimeout?: number; // Socket inactivity timeout (ms), default: 3600000 (1h) socketTimeout?: number; // Socket inactivity timeout (ms), default: 60000 (60s)
inactivityCheckInterval?: number; // How often to check for inactive connections (ms), default: 60000 (60s) inactivityCheckInterval?: number; // How often to check for inactive connections (ms), default: 60000 (60s)
maxConnectionLifetime?: number; // Default max connection lifetime (ms), default: 86400000 (24h) maxConnectionLifetime?: number; // Max connection lifetime (ms), default: 3600000 (1h)
inactivityTimeout?: number; // Inactivity timeout (ms), default: 14400000 (4h) inactivityTimeout?: number; // Inactivity timeout (ms), default: 75000 (75s)
gracefulShutdownTimeout?: number; // (ms) maximum time to wait for connections to close during shutdown gracefulShutdownTimeout?: number; // (ms) maximum time to wait for connections to close during shutdown

View File

@@ -47,16 +47,16 @@ export class SmartProxy extends plugins.EventEmitter {
// Apply defaults // Apply defaults
this.settings = { this.settings = {
...settingsArg, ...settingsArg,
initialDataTimeout: settingsArg.initialDataTimeout || 120000, initialDataTimeout: settingsArg.initialDataTimeout || 60_000,
socketTimeout: settingsArg.socketTimeout || 3600000, socketTimeout: settingsArg.socketTimeout || 60_000,
maxConnectionLifetime: settingsArg.maxConnectionLifetime || 86400000, maxConnectionLifetime: settingsArg.maxConnectionLifetime || 3_600_000,
inactivityTimeout: settingsArg.inactivityTimeout || 14400000, inactivityTimeout: settingsArg.inactivityTimeout || 75_000,
gracefulShutdownTimeout: settingsArg.gracefulShutdownTimeout || 30000, gracefulShutdownTimeout: settingsArg.gracefulShutdownTimeout || 30_000,
maxConnectionsPerIP: settingsArg.maxConnectionsPerIP || 100, maxConnectionsPerIP: settingsArg.maxConnectionsPerIP || 100,
connectionRateLimitPerMinute: settingsArg.connectionRateLimitPerMinute || 300, connectionRateLimitPerMinute: settingsArg.connectionRateLimitPerMinute || 300,
keepAliveTreatment: settingsArg.keepAliveTreatment || 'extended', keepAliveTreatment: settingsArg.keepAliveTreatment || 'standard',
keepAliveInactivityMultiplier: settingsArg.keepAliveInactivityMultiplier || 6, keepAliveInactivityMultiplier: settingsArg.keepAliveInactivityMultiplier || 4,
extendedKeepAliveLifetime: settingsArg.extendedKeepAliveLifetime || 7 * 24 * 60 * 60 * 1000, extendedKeepAliveLifetime: settingsArg.extendedKeepAliveLifetime || 3_600_000,
}; };
// Normalize ACME options // Normalize ACME options

View File

@@ -92,6 +92,16 @@ export class SocketHandlerServer {
let metadataBuffer = ''; let metadataBuffer = '';
let metadataParsed = false; let metadataParsed = false;
// 10s timeout for metadata parsing phase — if Rust connects but never
// sends the JSON metadata line, don't hold the socket open indefinitely.
socket.setTimeout(10_000);
socket.on('timeout', () => {
if (!metadataParsed) {
logger.log('warn', 'Socket handler metadata timeout, closing', { component: 'socket-handler-server' });
socket.destroy();
}
});
const onData = (chunk: Buffer) => { const onData = (chunk: Buffer) => {
if (metadataParsed) return; if (metadataParsed) return;
@@ -108,6 +118,7 @@ export class SocketHandlerServer {
} }
metadataParsed = true; metadataParsed = true;
socket.setTimeout(0); // Clear metadata timeout
socket.removeListener('data', onData); socket.removeListener('data', onData);
socket.pause(); // Prevent data loss between handler removal and pipe setup socket.pause(); // Prevent data loss between handler removal and pipe setup
@@ -254,11 +265,30 @@ export class SocketHandlerServer {
// Connect to the resolved target // Connect to the resolved target
const backend = plugins.net.connect(port, host, () => { const backend = plugins.net.connect(port, host, () => {
// Connection established — set idle timeout on both sides (5 min)
socket.setTimeout(300_000);
backend.setTimeout(300_000);
// Pipe bidirectionally // Pipe bidirectionally
socket.pipe(backend); socket.pipe(backend);
backend.pipe(socket); backend.pipe(socket);
}); });
// Connect timeout: if backend doesn't connect within 30s, destroy both
backend.setTimeout(30_000);
backend.on('timeout', () => {
logger.log('warn', `Dynamic forward timeout to ${host}:${port}`, { component: 'socket-handler-server' });
backend.destroy();
socket.destroy();
});
socket.on('timeout', () => {
logger.log('debug', `Dynamic forward client idle timeout`, { component: 'socket-handler-server' });
socket.destroy();
backend.destroy();
});
backend.on('error', (err) => { backend.on('error', (err) => {
logger.log('error', `Dynamic forward backend error: ${err.message}`, { component: 'socket-handler-server' }); logger.log('error', `Dynamic forward backend error: ${err.message}`, { component: 'socket-handler-server' });
socket.destroy(); socket.destroy();