Compare commits

..

9 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
11 changed files with 308 additions and 81 deletions

View File

@@ -1,5 +1,37 @@
# 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) ## 2026-02-25 - 25.8.1 - fix(allocator)
switch global allocator from tikv-jemallocator to mimalloc switch global allocator from tikv-jemallocator to mimalloc

View File

@@ -1,6 +1,6 @@
{ {
"name": "@push.rocks/smartproxy", "name": "@push.rocks/smartproxy",
"version": "25.8.1", "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",

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 {
collector: Arc::clone(&metrics),
route_id: route_id.map(|s| s.to_string()),
source_ip: Some(ip_str.clone()),
}),
).await;
// Include the initial data that was forwarded before the bidirectional relay
if initial_len > 0 { if initial_len > 0 {
let ip = peer_addr.ip().to_string(); metrics.record_bytes(initial_len, 0, route_id, Some(&ip_str));
metrics.record_bytes(initial_len, 0, route_id, Some(&ip));
}
debug!("Socket handler relay ended for {}: {}", route_key, e);
}
} }
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,8 +1410,14 @@ 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() => {
debug!("Split-stream connection cancelled by shutdown");
c2b_handle.abort();
b2c_handle.abort();
break;
}
_ = tokio::time::sleep(check_interval) => {
// Check max lifetime // Check max lifetime
if start.elapsed() >= max_lifetime { if start.elapsed() >= max_lifetime {
debug!("Connection exceeded max lifetime, closing"); debug!("Connection exceeded max lifetime, closing");
@@ -1394,6 +1440,8 @@ impl TcpListenerManager {
} }
last_seen = current; last_seen = current;
} }
}
}
}); });
let bytes_in = c2b.await.unwrap_or(0); let bytes_in = c2b.await.unwrap_or(0);

View File

@@ -3,6 +3,6 @@
*/ */
export const commitinfo = { export const commitinfo = {
name: '@push.rocks/smartproxy', name: '@push.rocks/smartproxy',
version: '25.8.1', 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,7 +18,7 @@ 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();
@@ -124,7 +124,8 @@ 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);
@@ -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();