push.rocks/smartproxy
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add TCP keepalive options and connection pooling for improved performance

Browse Source 
- Added `socket2` dependency for socket options.
- Introduced `keep_alive`, `keep_alive_initial_delay_ms`, and `max_connections` fields in `ConnectionConfig`.
- Implemented TCP keepalive settings in `TcpListenerManager` for both client and backend connections.
- Created a new `ConnectionPool` for managing idle HTTP/1.1 and HTTP/2 connections to reduce overhead.
- Enhanced TLS configuration to support ALPN for HTTP/2.
- Added performance tests for connection pooling, stability, and concurrent connections.
This commit is contained in:
Jürgen Kunz
2026-02-20 18:16:09 +00:00
parent 0f6752b9a7
commit 9521f2e044
14 changed files with 1058 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
rust/Cargo.lock generated
View File

@@ -509,7 +509,7 @@ dependencies = [
"hyper",
"libc",
"pin-project-lite",
"socket2",
"socket2 0.6.2",
"tokio",
"tower-service",
"tracing",
@@ -971,6 +971,7 @@ dependencies = [
"rustproxy-metrics",
"rustproxy-routing",
"rustproxy-security",
"socket2 0.5.10",
"thiserror 2.0.18",
"tokio",
"tokio-rustls",
@@ -1019,6 +1020,7 @@ dependencies = [
"rustproxy-routing",
"serde",
"serde_json",
"socket2 0.5.10",
"thiserror 2.0.18",
"tokio",
"tokio-rustls",
@@ -1204,6 +1206,16 @@ version = "1.15.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
[[package]]
name = "socket2"
version = "0.5.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678"
dependencies = [
"libc",
"windows-sys 0.52.0",
]
[[package]]
name = "socket2"
version = "0.6.2"
@@ -1329,7 +1341,7 @@ dependencies = [
"parking_lot",
"pin-project-lite",
"signal-hook-registry",
"socket2",
"socket2 0.6.2",
"tokio-macros",
"windows-sys 0.61.2",
]

3
rust/Cargo.toml
View File

@@ -88,6 +88,9 @@ async-trait = "0.1"
# libc for uid checks
libc = "0.2"
# Socket-level options (keepalive, etc.)
socket2 = { version = "0.5", features = ["all"] }
# Internal crates
rustproxy-config = { path = "crates/rustproxy-config" }
rustproxy-routing = { path = "crates/rustproxy-routing" }

5
rust/crates/rustproxy-config/src/proxy_options.rs
View File

@@ -208,6 +208,10 @@ pub struct RustProxyOptions {
#[serde(skip_serializing_if = "Option::is_none")]
pub connection_rate_limit_per_minute: Option<u64>,
/// Global maximum simultaneous connections (default: 100000)
#[serde(skip_serializing_if = "Option::is_none")]
pub max_connections: Option<u64>,
// ─── Keep-Alive Settings ─────────────────────────────────────────
/// How to treat keep-alive connections
@@ -272,6 +276,7 @@ impl Default for RustProxyOptions {
enable_randomized_timeouts: None,
max_connections_per_ip: None,
connection_rate_limit_per_minute: None,
max_connections: None,
keep_alive_treatment: None,
keep_alive_inactivity_multiplier: None,
extended_keep_alive_lifetime: None,

1
rust/crates/rustproxy-http/Cargo.toml
View File

@@ -26,3 +26,4 @@ anyhow = { workspace = true }
arc-swap = { workspace = true }
dashmap = { workspace = true }
tokio-util = { workspace = true }
socket2 = { workspace = true }

188
rust/crates/rustproxy-http/src/connection_pool.rs Normal file
View File

@@ -0,0 +1,188 @@
//! Backend connection pool for HTTP/1.1 and HTTP/2.
//!
//! Reuses idle keep-alive connections to avoid per-request TCP+TLS handshakes.
//! HTTP/2 connections are multiplexed (clone the sender for each request).
use std::sync::Arc;
use std::time::{Duration, Instant};
use bytes::Bytes;
use dashmap::DashMap;
use http_body_util::combinators::BoxBody;
use hyper::client::conn::{http1, http2};
use tracing::debug;
/// Maximum idle connections per backend key.
const MAX_IDLE_PER_KEY: usize = 16;
/// Default idle timeout — connections not used within this window are evicted.
const IDLE_TIMEOUT: Duration = Duration::from_secs(90);
/// Background eviction interval.
const EVICTION_INTERVAL: Duration = Duration::from_secs(30);
/// Identifies a unique backend endpoint.
#[derive(Clone, Debug, Hash, Eq, PartialEq)]
pub struct PoolKey {
pub host: String,
pub port: u16,
pub use_tls: bool,
pub h2: bool,
}
/// An idle HTTP/1.1 sender with a timestamp for eviction.
struct IdleH1 {
sender: http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
idle_since: Instant,
}
/// 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,
}
/// Backend connection pool.
pub struct ConnectionPool {
/// HTTP/1.1 idle connections indexed by backend key.
h1_pool: Arc<DashMap<PoolKey, Vec<IdleH1>>>,
/// HTTP/2 multiplexed connections indexed by backend key.
h2_pool: Arc<DashMap<PoolKey, PooledH2>>,
/// Handle for the background eviction task.
eviction_handle: Option<tokio::task::JoinHandle<()>>,
}
impl ConnectionPool {
/// Create a new pool and start the background eviction task.
pub fn new() -> Self {
let h1_pool: Arc<DashMap<PoolKey, Vec<IdleH1>>> = Arc::new(DashMap::new());
let h2_pool: Arc<DashMap<PoolKey, PooledH2>> = Arc::new(DashMap::new());
let h1_clone = Arc::clone(&h1_pool);
let h2_clone = Arc::clone(&h2_pool);
let eviction_handle = tokio::spawn(async move {
Self::eviction_loop(h1_clone, h2_clone).await;
});
Self {
h1_pool,
h2_pool,
eviction_handle: Some(eviction_handle),
}
}
/// Try to check out an idle HTTP/1.1 sender for the given key.
/// Returns `None` if no usable idle connection exists.
pub fn checkout_h1(&self, key: &PoolKey) -> Option<http1::SendRequest<BoxBody<Bytes, hyper::Error>>> {
let mut entry = self.h1_pool.get_mut(key)?;
let idles = entry.value_mut();
while let Some(idle) = idles.pop() {
// Check if the connection is still alive and ready
if idle.idle_since.elapsed() < IDLE_TIMEOUT && idle.sender.is_ready() && !idle.sender.is_closed() {
debug!("Pool hit (h1): {}:{}", key.host, key.port);
return Some(idle.sender);
}
// Stale or closed — drop it
}
// Clean up empty entry
if idles.is_empty() {
drop(entry);
self.h1_pool.remove(key);
}
None
}
/// Return an HTTP/1.1 sender to the pool after the response body has been prepared.
/// The caller should NOT call this if the sender is closed or not ready.
pub fn checkin_h1(&self, key: PoolKey, sender: http1::SendRequest<BoxBody<Bytes, hyper::Error>>) {
if sender.is_closed() || !sender.is_ready() {
return; // Don't pool broken connections
}
let mut entry = self.h1_pool.entry(key).or_insert_with(Vec::new);
if entry.value().len() < MAX_IDLE_PER_KEY {
entry.value_mut().push(IdleH1 {
sender,
idle_since: Instant::now(),
});
}
// If at capacity, just drop the sender
}
/// Try to get a cloned HTTP/2 sender for the given key.
/// HTTP/2 senders are Clone-able (multiplexed), so we clone rather than remove.
pub fn checkout_h2(&self, key: &PoolKey) -> Option<http2::SendRequest<BoxBody<Bytes, hyper::Error>>> {
let entry = self.h2_pool.get(key)?;
let pooled = entry.value();
// Check if the h2 connection is still alive
if pooled.sender.is_closed() {
drop(entry);
self.h2_pool.remove(key);
return None;
}
if pooled.sender.is_ready() {
debug!("Pool hit (h2): {}:{}", key.host, key.port);
return Some(pooled.sender.clone());
}
None
}
/// 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>>) {
if sender.is_closed() {
return;
}
self.h2_pool.insert(key, PooledH2 {
sender,
created_at: Instant::now(),
});
}
/// Background eviction loop — runs every EVICTION_INTERVAL to remove stale connections.
async fn eviction_loop(
h1_pool: Arc<DashMap<PoolKey, Vec<IdleH1>>>,
h2_pool: Arc<DashMap<PoolKey, PooledH2>>,
) {
let mut interval = tokio::time::interval(EVICTION_INTERVAL);
loop {
interval.tick().await;
// Evict stale H1 connections
let mut empty_keys = Vec::new();
for mut entry in h1_pool.iter_mut() {
entry.value_mut().retain(|idle| {
idle.idle_since.elapsed() < IDLE_TIMEOUT && !idle.sender.is_closed()
});
if entry.value().is_empty() {
empty_keys.push(entry.key().clone());
}
}
for key in empty_keys {
h1_pool.remove(&key);
}
// Evict dead H2 connections
let mut dead_h2 = Vec::new();
for entry in h2_pool.iter() {
if entry.value().sender.is_closed() {
dead_h2.push(entry.key().clone());
}
}
for key in dead_h2 {
h2_pool.remove(&key);
}
}
}
}
impl Drop for ConnectionPool {
fn drop(&mut self) {
if let Some(handle) = self.eviction_handle.take() {
handle.abort();
}
}
}

2
rust/crates/rustproxy-http/src/lib.rs
View File

@@ -3,6 +3,7 @@
//! Hyper-based HTTP proxy service for RustProxy.
//! Handles HTTP request parsing, route-based forwarding, and response filtering.
pub mod connection_pool;
pub mod counting_body;
pub mod proxy_service;
pub mod request_filter;
@@ -10,6 +11,7 @@ pub mod response_filter;
pub mod template;
pub mod upstream_selector;
pub use connection_pool::*;
pub use counting_body::*;
pub use proxy_service::*;
pub use template::*;

300
rust/crates/rustproxy-http/src/proxy_service.rs
View File

@@ -87,21 +87,20 @@ impl tokio::io::AsyncWrite for BackendStream {
}
}
/// Connect to a backend over TLS. Uses InsecureVerifier for internal backends
/// with self-signed certs (same pattern as tls_handler::connect_tls).
/// Connect to a backend over TLS using the shared backend TLS config
/// (from tls_handler). Session resumption is automatic.
async fn connect_tls_backend(
backend_tls_config: &Arc<rustls::ClientConfig>,
host: &str,
port: u16,
) -> Result<tokio_rustls::client::TlsStream<TcpStream>, Box<dyn std::error::Error + Send + Sync>> {
let _ = rustls::crypto::ring::default_provider().install_default();
let config = rustls::ClientConfig::builder()
.dangerous()
.with_custom_certificate_verifier(Arc::new(InsecureBackendVerifier))
.with_no_client_auth();
let connector = tokio_rustls::TlsConnector::from(Arc::new(config));
let connector = tokio_rustls::TlsConnector::from(Arc::clone(backend_tls_config));
let stream = TcpStream::connect(format!("{}:{}", host, port)).await?;
stream.set_nodelay(true)?;
// Apply keepalive with 60s default
let _ = socket2::SockRef::from(&stream).set_tcp_keepalive(
&socket2::TcpKeepalive::new().with_time(std::time::Duration::from_secs(60))
);
let server_name = rustls::pki_types::ServerName::try_from(host.to_string())?;
let tls_stream = connector.connect(server_name, stream).await?;
@@ -109,56 +108,6 @@ async fn connect_tls_backend(
Ok(tls_stream)
}
/// Insecure certificate verifier for backend TLS connections.
/// Internal backends may use self-signed certs.
#[derive(Debug)]
struct InsecureBackendVerifier;
impl rustls::client::danger::ServerCertVerifier for InsecureBackendVerifier {
fn verify_server_cert(
&self,
_end_entity: &rustls::pki_types::CertificateDer<'_>,
_intermediates: &[rustls::pki_types::CertificateDer<'_>],
_server_name: &rustls::pki_types::ServerName<'_>,
_ocsp_response: &[u8],
_now: rustls::pki_types::UnixTime,
) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
_message: &[u8],
_cert: &rustls::pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn verify_tls13_signature(
&self,
_message: &[u8],
_cert: &rustls::pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
vec![
rustls::SignatureScheme::RSA_PKCS1_SHA256,
rustls::SignatureScheme::RSA_PKCS1_SHA384,
rustls::SignatureScheme::RSA_PKCS1_SHA512,
rustls::SignatureScheme::ECDSA_NISTP256_SHA256,
rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
rustls::SignatureScheme::ED25519,
rustls::SignatureScheme::RSA_PSS_SHA256,
rustls::SignatureScheme::RSA_PSS_SHA384,
rustls::SignatureScheme::RSA_PSS_SHA512,
]
}
}
/// HTTP proxy service that processes HTTP traffic.
pub struct HttpProxyService {
route_manager: Arc<RouteManager>,
@@ -172,6 +121,10 @@ pub struct HttpProxyService {
request_counter: AtomicU64,
/// Cache of compiled URL rewrite regexes (keyed by pattern string).
regex_cache: DashMap<String, Regex>,
/// Shared backend TLS config for session resumption across connections.
backend_tls_config: Arc<rustls::ClientConfig>,
/// Backend connection pool for reusing keep-alive connections.
connection_pool: Arc<crate::connection_pool::ConnectionPool>,
}
impl HttpProxyService {
@@ -184,6 +137,8 @@ impl HttpProxyService {
route_rate_limiters: Arc::new(DashMap::new()),
request_counter: AtomicU64::new(0),
regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
}
}
@@ -201,9 +156,17 @@ impl HttpProxyService {
route_rate_limiters: Arc::new(DashMap::new()),
request_counter: AtomicU64::new(0),
regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
}
}
/// Set the shared backend TLS config (enables session resumption).
/// Call this after construction to inject the shared config from tls_handler.
pub fn set_backend_tls_config(&mut self, config: Arc<rustls::ClientConfig>) {
self.backend_tls_config = config;
}
/// Handle an incoming HTTP connection on a plain TCP stream.
pub async fn handle_connection(
self: Arc<Self>,
@@ -217,8 +180,10 @@ impl HttpProxyService {
/// Handle an incoming HTTP connection on any IO type (plain TCP or TLS-terminated).
///
/// Uses HTTP/1.1 with upgrade support. Responds to graceful shutdown via the
/// cancel token — in-flight requests complete, but no new requests are accepted.
/// Uses `hyper_util::server::conn::auto::Builder` to auto-detect h1 vs h2
/// based on ALPN negotiation (TLS) or connection preface (h2c).
/// Supports HTTP/1.1 upgrades (WebSocket) and HTTP/2 CONNECT.
/// Responds to graceful shutdown via the cancel token.
pub async fn handle_io<I>(
self: Arc<Self>,
stream: I,
@@ -241,24 +206,23 @@ impl HttpProxyService {
}
});
// Use http1::Builder with upgrades for WebSocket support
let mut conn = hyper::server::conn::http1::Builder::new()
.keep_alive(true)
.serve_connection(io, service)
.with_upgrades();
// Auto-detect h1 vs h2 based on ALPN / connection preface.
// serve_connection_with_upgrades supports h1 Upgrade (WebSocket) and h2 CONNECT.
let builder = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
let conn = builder.serve_connection_with_upgrades(io, service);
// Pin on the heap — auto::UpgradeableConnection is !Unpin
let mut conn = Box::pin(conn);
// Use select to support graceful shutdown via cancellation token
let conn_pin = std::pin::Pin::new(&mut conn);
tokio::select! {
result = conn_pin => {
result = conn.as_mut() => {
if let Err(e) = result {
debug!("HTTP connection error from {}: {}", peer_addr, e);
}
}
_ = cancel.cancelled() => {
// Graceful shutdown: let in-flight request finish, stop accepting new ones
let conn_pin = std::pin::Pin::new(&mut conn);
conn_pin.graceful_shutdown();
conn.as_mut().graceful_shutdown();
if let Err(e) = conn.await {
debug!("HTTP connection error during shutdown from {}: {}", peer_addr, e);
}
@@ -478,11 +442,32 @@ impl HttpProxyService {
}
}
// Connect to upstream with timeout (TLS if upstream.use_tls is set)
// --- Connection pooling: try reusing an existing connection first ---
let pool_key = crate::connection_pool::PoolKey {
host: upstream.host.clone(),
port: upstream.port,
use_tls: upstream.use_tls,
h2: use_h2,
};
// Try pooled connection first (H2 only — H2 senders are Clone and multiplexed,
// so checkout doesn't consume request parts. For H1, we try pool inside forward_h1.)
if use_h2 {
if let Some(sender) = self.connection_pool.checkout_h2(&pool_key) {
let result = self.forward_h2_pooled(
sender, parts, body, upstream_headers, &upstream_path,
route_match.route, route_id, &ip_str, &pool_key,
).await;
self.upstream_selector.connection_ended(&upstream_key);
return result;
}
}
// Fresh connection path
let backend = if upstream.use_tls {
match tokio::time::timeout(
self.connect_timeout,
connect_tls_backend(&upstream.host, upstream.port),
connect_tls_backend(&self.backend_tls_config, &upstream.host, upstream.port),
).await {
Ok(Ok(tls)) => BackendStream::Tls(tls),
Ok(Err(e)) => {
@@ -503,6 +488,9 @@ impl HttpProxyService {
).await {
Ok(Ok(s)) => {
s.set_nodelay(true).ok();
let _ = socket2::SockRef::from(&s).set_tcp_keepalive(
&socket2::TcpKeepalive::new().with_time(std::time::Duration::from_secs(60))
);
BackendStream::Plain(s)
}
Ok(Err(e)) => {
@@ -521,17 +509,16 @@ impl HttpProxyService {
let io = TokioIo::new(backend);
let result = if use_h2 {
// HTTP/2 backend
self.forward_h2(io, parts, body, upstream_headers, &upstream_path, &upstream, route_match.route, route_id, &ip_str).await
self.forward_h2(io, parts, body, upstream_headers, &upstream_path, &upstream, route_match.route, route_id, &ip_str, &pool_key).await
} else {
// HTTP/1.1 backend (default)
self.forward_h1(io, parts, body, upstream_headers, &upstream_path, &upstream, route_match.route, route_id, &ip_str).await
self.forward_h1(io, parts, body, upstream_headers, &upstream_path, &upstream, route_match.route, route_id, &ip_str, &pool_key).await
};
self.upstream_selector.connection_ended(&upstream_key);
result
}
/// Forward request to backend via HTTP/1.1 with body streaming.
/// Tries a pooled connection first; if unavailable, uses the fresh IO connection.
async fn forward_h1(
&self,
io: TokioIo<BackendStream>,
@@ -543,8 +530,21 @@ impl HttpProxyService {
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) = match hyper::client::conn::http1::handshake(io).await {
// Try pooled H1 connection first — avoids TCP+TLS handshake
if let Some(pooled_sender) = self.connection_pool.checkout_h1(pool_key) {
return self.forward_h1_with_sender(
pooled_sender, parts, body, upstream_headers, upstream_path,
route, route_id, source_ip, pool_key,
).await;
}
// Fresh connection: explicitly type the handshake with BoxBody for uniform pool type
let (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!("Upstream handshake failed: {}", e);
@@ -558,6 +558,22 @@ impl HttpProxyService {
}
});
self.forward_h1_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip, pool_key).await
}
/// Common H1 forwarding logic used by both fresh and pooled paths.
async fn forward_h1_with_sender(
&self,
mut sender: hyper::client::conn::http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
parts: hyper::http::request::Parts,
body: Incoming,
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 upstream_req = Request::builder()
.method(parts.method)
.uri(upstream_path)
@@ -567,7 +583,7 @@ impl HttpProxyService {
*headers = upstream_headers;
}
// Wrap the request body in CountingBody to track bytes_in
// Wrap the request body in CountingBody then box it for the uniform pool type
let counting_req_body = CountingBody::new(
body,
Arc::clone(&self.metrics),
@@ -575,9 +591,9 @@ impl HttpProxyService {
Some(source_ip.to_string()),
Direction::In,
);
let boxed_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_req_body);
// Stream the request body through to upstream
let upstream_req = upstream_req.body(counting_req_body).unwrap();
let upstream_req = upstream_req.body(boxed_body).unwrap();
let upstream_response = match sender.send_request(upstream_req).await {
Ok(resp) => resp,
@@ -587,10 +603,14 @@ impl HttpProxyService {
}
};
// Return sender to pool (body streams lazily, sender is reusable once response head is received)
self.connection_pool.checkin_h1(pool_key.clone(), sender);
self.build_streaming_response(upstream_response, route, route_id, source_ip).await
}
/// Forward request to backend via HTTP/2 with body streaming.
/// Forward request to backend via HTTP/2 with body streaming (fresh connection).
/// Registers the h2 sender in the pool for future multiplexed requests.
async fn forward_h2(
&self,
io: TokioIo<BackendStream>,
@@ -602,9 +622,14 @@ impl HttpProxyService {
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 exec = hyper_util::rt::TokioExecutor::new();
let (mut sender, conn) = match hyper::client::conn::http2::handshake(exec, io).await {
// Explicitly type the handshake with BoxBody for uniform pool type
let (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!("HTTP/2 upstream handshake failed: {}", e);
@@ -618,6 +643,40 @@ 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
}
/// Forward request using an existing (pooled) HTTP/2 sender.
async fn forward_h2_pooled(
&self,
sender: hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
parts: hyper::http::request::Parts,
body: Incoming,
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> {
self.forward_h2_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip).await
}
/// Common H2 forwarding logic used by both fresh and pooled paths.
async fn forward_h2_with_sender(
&self,
mut sender: hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
parts: hyper::http::request::Parts,
body: Incoming,
upstream_headers: hyper::HeaderMap,
upstream_path: &str,
route: &rustproxy_config::RouteConfig,
route_id: Option<&str>,
source_ip: &str,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
let mut upstream_req = Request::builder()
.method(parts.method)
.uri(upstream_path);
@@ -626,7 +685,7 @@ impl HttpProxyService {
*headers = upstream_headers;
}
// Wrap the request body in CountingBody to track bytes_in
// Wrap the request body in CountingBody then box it for the uniform pool type
let counting_req_body = CountingBody::new(
body,
Arc::clone(&self.metrics),
@@ -634,9 +693,9 @@ impl HttpProxyService {
Some(source_ip.to_string()),
Direction::In,
);
let boxed_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_req_body);
// Stream the request body through to upstream
let upstream_req = upstream_req.body(counting_req_body).unwrap();
let upstream_req = upstream_req.body(boxed_body).unwrap();
let upstream_response = match sender.send_request(upstream_req).await {
Ok(resp) => resp,
@@ -723,7 +782,7 @@ impl HttpProxyService {
let mut upstream_stream: BackendStream = if upstream.use_tls {
match tokio::time::timeout(
self.connect_timeout,
connect_tls_backend(&upstream.host, upstream.port),
connect_tls_backend(&self.backend_tls_config, &upstream.host, upstream.port),
).await {
Ok(Ok(tls)) => BackendStream::Tls(tls),
Ok(Err(e)) => {
@@ -744,6 +803,9 @@ impl HttpProxyService {
).await {
Ok(Ok(s)) => {
s.set_nodelay(true).ok();
let _ = socket2::SockRef::from(&s).set_tcp_keepalive(
&socket2::TcpKeepalive::new().with_time(std::time::Duration::from_secs(60))
);
BackendStream::Plain(s)
}
Ok(Err(e)) => {
@@ -1221,6 +1283,70 @@ fn guess_content_type(path: &std::path::Path) -> &'static str {
}
}
impl HttpProxyService {
/// Build a default backend TLS config with InsecureVerifier.
/// Used as fallback when no shared config is injected from tls_handler.
fn default_backend_tls_config() -> Arc<rustls::ClientConfig> {
let _ = rustls::crypto::ring::default_provider().install_default();
let config = rustls::ClientConfig::builder()
.dangerous()
.with_custom_certificate_verifier(Arc::new(InsecureBackendVerifier))
.with_no_client_auth();
Arc::new(config)
}
}
/// Insecure certificate verifier for backend TLS connections (fallback only).
/// The production path uses the shared config from tls_handler which has the same
/// behavior but with session resumption across all outbound connections.
#[derive(Debug)]
struct InsecureBackendVerifier;
impl rustls::client::danger::ServerCertVerifier for InsecureBackendVerifier {
fn verify_server_cert(
&self,
_end_entity: &rustls::pki_types::CertificateDer<'_>,
_intermediates: &[rustls::pki_types::CertificateDer<'_>],
_server_name: &rustls::pki_types::ServerName<'_>,
_ocsp_response: &[u8],
_now: rustls::pki_types::UnixTime,
) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
_message: &[u8],
_cert: &rustls::pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn verify_tls13_signature(
&self,
_message: &[u8],
_cert: &rustls::pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
vec![
rustls::SignatureScheme::RSA_PKCS1_SHA256,
rustls::SignatureScheme::RSA_PKCS1_SHA384,
rustls::SignatureScheme::RSA_PKCS1_SHA512,
rustls::SignatureScheme::ECDSA_NISTP256_SHA256,
rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
rustls::SignatureScheme::ED25519,
rustls::SignatureScheme::RSA_PSS_SHA256,
rustls::SignatureScheme::RSA_PSS_SHA384,
rustls::SignatureScheme::RSA_PSS_SHA512,
]
}
}
impl Default for HttpProxyService {
fn default() -> Self {
Self {
@@ -1231,6 +1357,8 @@ impl Default for HttpProxyService {
route_rate_limiters: Arc::new(DashMap::new()),
request_counter: AtomicU64::new(0),
regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
}
}
}

1
rust/crates/rustproxy-passthrough/Cargo.toml
View File

@@ -23,3 +23,4 @@ rustls-pemfile = { workspace = true }
tokio-util = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
socket2 = { workspace = true }

2
rust/crates/rustproxy-passthrough/src/lib.rs
View File

@@ -11,6 +11,7 @@ pub mod tls_handler;
pub mod connection_record;
pub mod connection_tracker;
pub mod socket_relay;
pub mod socket_opts;
pub use tcp_listener::*;
pub use sni_parser::*;
@@ -20,3 +21,4 @@ pub use tls_handler::*;
pub use connection_record::*;
pub use connection_tracker::*;
pub use socket_relay::*;
pub use socket_opts::*;

19
rust/crates/rustproxy-passthrough/src/socket_opts.rs Normal file
View File

@@ -0,0 +1,19 @@
//! Socket-level options for TCP streams (keepalive, etc.).
//!
//! Uses `socket2::SockRef::from()` to borrow the raw fd without ownership transfer.
use std::io;
use std::time::Duration;
use tokio::net::TcpStream;
/// Apply TCP keepalive to a connected socket.
///
/// Enables SO_KEEPALIVE and sets the initial probe delay.
/// On Linux, also sets the interval between probes to the same value.
pub fn apply_keepalive(stream: &TcpStream, delay: Duration) -> io::Result<()> {
let sock_ref = socket2::SockRef::from(stream);
let ka = socket2::TcpKeepalive::new().with_time(delay);
#[cfg(target_os = "linux")]
let ka = ka.with_interval(delay);
sock_ref.set_tcp_keepalive(&ka)
}

85
rust/crates/rustproxy-passthrough/src/tcp_listener.rs
View File

@@ -15,6 +15,7 @@ use crate::sni_parser;
use crate::forwarder;
use crate::tls_handler;
use crate::connection_tracker::ConnectionTracker;
use crate::socket_opts;
/// RAII guard that decrements the active connection metric on drop.
/// Ensures connection_closed is called on ALL exit paths — normal, error, or panic.
@@ -87,6 +88,12 @@ pub struct ConnectionConfig {
/// Trusted IPs that may send PROXY protocol headers.
/// When non-empty, only connections from these IPs will have PROXY headers parsed.
pub proxy_ips: Vec<std::net::IpAddr>,
/// Enable TCP keepalive on sockets (default: true)
pub keep_alive: bool,
/// Initial delay before first keepalive probe (ms, default: 60000)
pub keep_alive_initial_delay_ms: u64,
/// Global maximum simultaneous connections (default: 100000)
pub max_connections: u64,
}
impl Default for ConnectionConfig {
@@ -105,6 +112,9 @@ impl Default for ConnectionConfig {
accept_proxy_protocol: false,
send_proxy_protocol: false,
proxy_ips: Vec::new(),
keep_alive: true,
keep_alive_initial_delay_ms: 60_000,
max_connections: 100_000,
}
}
}
@@ -131,21 +141,26 @@ pub struct TcpListenerManager {
cancel_token: CancellationToken,
/// Path to Unix domain socket for relaying socket-handler connections to TypeScript.
socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>,
/// Global connection semaphore — limits total simultaneous connections.
conn_semaphore: Arc<tokio::sync::Semaphore>,
}
impl TcpListenerManager {
pub fn new(route_manager: Arc<RouteManager>) -> Self {
let metrics = Arc::new(MetricsCollector::new());
let conn_config = ConnectionConfig::default();
let http_proxy = Arc::new(HttpProxyService::with_connect_timeout(
let mut http_proxy_svc = HttpProxyService::with_connect_timeout(
Arc::clone(&route_manager),
Arc::clone(&metrics),
std::time::Duration::from_millis(conn_config.connection_timeout_ms),
));
);
http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config());
let http_proxy = Arc::new(http_proxy_svc);
let conn_tracker = Arc::new(ConnectionTracker::new(
conn_config.max_connections_per_ip,
conn_config.connection_rate_limit_per_minute,
));
let max_conns = conn_config.max_connections as usize;
Self {
listeners: HashMap::new(),
route_manager: Arc::new(ArcSwap::from(route_manager)),
@@ -157,21 +172,25 @@ impl TcpListenerManager {
conn_tracker,
cancel_token: CancellationToken::new(),
socket_handler_relay: Arc::new(std::sync::RwLock::new(None)),
conn_semaphore: Arc::new(tokio::sync::Semaphore::new(max_conns)),
}
}
/// Create with a metrics collector.
pub fn with_metrics(route_manager: Arc<RouteManager>, metrics: Arc<MetricsCollector>) -> Self {
let conn_config = ConnectionConfig::default();
let http_proxy = Arc::new(HttpProxyService::with_connect_timeout(
let mut http_proxy_svc = HttpProxyService::with_connect_timeout(
Arc::clone(&route_manager),
Arc::clone(&metrics),
std::time::Duration::from_millis(conn_config.connection_timeout_ms),
));
);
http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config());
let http_proxy = Arc::new(http_proxy_svc);
let conn_tracker = Arc::new(ConnectionTracker::new(
conn_config.max_connections_per_ip,
conn_config.connection_rate_limit_per_minute,
));
let max_conns = conn_config.max_connections as usize;
Self {
listeners: HashMap::new(),
route_manager: Arc::new(ArcSwap::from(route_manager)),
@@ -183,6 +202,7 @@ impl TcpListenerManager {
conn_tracker,
cancel_token: CancellationToken::new(),
socket_handler_relay: Arc::new(std::sync::RwLock::new(None)),
conn_semaphore: Arc::new(tokio::sync::Semaphore::new(max_conns)),
}
}
@@ -192,6 +212,7 @@ impl TcpListenerManager {
config.max_connections_per_ip,
config.connection_rate_limit_per_minute,
));
self.conn_semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_connections as usize));
self.conn_config = Arc::new(config);
}
@@ -247,11 +268,13 @@ impl TcpListenerManager {
let conn_tracker = Arc::clone(&self.conn_tracker);
let cancel = self.cancel_token.clone();
let relay = Arc::clone(&self.socket_handler_relay);
let semaphore = Arc::clone(&self.conn_semaphore);
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,
).await;
});
@@ -346,6 +369,7 @@ impl TcpListenerManager {
conn_tracker: Arc<ConnectionTracker>,
cancel: CancellationToken,
socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>,
conn_semaphore: Arc<tokio::sync::Semaphore>,
) {
loop {
tokio::select! {
@@ -358,10 +382,31 @@ impl TcpListenerManager {
Ok((stream, peer_addr)) => {
let ip = peer_addr.ip();
// Global connection limit — acquire semaphore permit with timeout
let permit = match tokio::time::timeout(
std::time::Duration::from_secs(5),
conn_semaphore.clone().acquire_owned(),
).await {
Ok(Ok(permit)) => permit,
Ok(Err(_)) => {
// Semaphore closed — shouldn't happen, but be safe
debug!("Connection semaphore closed, dropping connection from {}", peer_addr);
drop(stream);
continue;
}
Err(_) => {
// Timeout — global limit reached
debug!("Global connection limit reached, dropping connection from {}", peer_addr);
drop(stream);
continue;
}
};
// Check per-IP limits and rate limiting
if !conn_tracker.try_accept(&ip) {
debug!("Rejected connection from {} (per-IP limit or rate limit)", peer_addr);
drop(stream);
drop(permit);
continue;
}
@@ -382,6 +427,8 @@ impl TcpListenerManager {
debug!("Accepted connection from {} on port {}", peer_addr, port);
tokio::spawn(async move {
// Move permit into the task — auto-releases on drop
let _permit = permit;
let result = Self::handle_connection(
stream, port, peer_addr, rm, m, tc, sa, hp, cc, cn, sr,
).await;
@@ -418,6 +465,12 @@ impl TcpListenerManager {
use tokio::io::AsyncReadExt;
stream.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&stream, delay) {
debug!("Failed to set keepalive on client socket: {}", e);
}
}
// --- PROXY protocol: must happen BEFORE ip_str and fast path ---
// Only parse PROXY headers from trusted proxy IPs (security).
@@ -543,6 +596,12 @@ impl TcpListenerManager {
Err(_) => return Err("Backend connection timeout".into()),
};
backend.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&backend, delay) {
debug!("Failed to set keepalive on backend socket: {}", e);
}
}
// Send PROXY protocol header if configured
let should_send_proxy = conn_config.send_proxy_protocol
@@ -778,6 +837,12 @@ impl TcpListenerManager {
Err(_) => return Err("Backend connection timeout".into()),
};
backend.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&backend, delay) {
debug!("Failed to set keepalive on backend socket: {}", e);
}
}
// Send PROXY protocol header if configured
if let Some(ref header) = proxy_header {
@@ -857,6 +922,12 @@ impl TcpListenerManager {
Err(_) => return Err("Backend connection timeout".into()),
};
backend.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&backend, delay) {
debug!("Failed to set keepalive on backend socket: {}", e);
}
}
let (tls_read, tls_write) = tokio::io::split(buf_stream);
let (backend_read, backend_write) = tokio::io::split(backend);
@@ -944,6 +1015,12 @@ impl TcpListenerManager {
Err(_) => return Err("Backend connection timeout".into()),
};
backend.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&backend, delay) {
debug!("Failed to set keepalive on backend socket: {}", e);
}
}
// Send PROXY protocol header if configured
if let Some(ref header) = proxy_header {

43
rust/crates/rustproxy-passthrough/src/tls_handler.rs
View File

@@ -1,6 +1,6 @@
use std::collections::HashMap;
use std::io::BufReader;
use std::sync::Arc;
use std::sync::{Arc, OnceLock};
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use rustls::server::ResolvesServerCert;
@@ -84,13 +84,16 @@ pub fn build_shared_tls_acceptor(resolver: CertResolver) -> Result<TlsAcceptor,
.with_no_client_auth()
.with_cert_resolver(Arc::new(resolver));
// ALPN: advertise h2 and http/1.1 for client-facing HTTP/2 support
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
// Shared session cache — enables session ID resumption across connections
config.session_storage = rustls::server::ServerSessionMemoryCache::new(4096);
// Session ticket resumption (12-hour lifetime, Chacha20Poly1305 encrypted)
config.ticketer = rustls::crypto::ring::Ticketer::new()
.map_err(|e| format!("Ticketer: {}", e))?;
info!("Built shared TLS config with session cache (4096) and ticket support");
info!("Built shared TLS config with session cache (4096), ticket support, and ALPN h2+http/1.1");
Ok(TlsAcceptor::from(Arc::new(config)))
}
@@ -122,6 +125,9 @@ pub fn build_tls_acceptor_with_config(
.with_single_cert(certs, key)?
};
// ALPN: advertise h2 and http/1.1 for client-facing HTTP/2 support
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
// Apply session timeout if configured
if let Some(route_tls) = tls_config {
if let Some(timeout_secs) = route_tls.session_timeout {
@@ -179,21 +185,40 @@ pub async fn accept_tls(
Ok(tls_stream)
}
/// Get or create a shared backend TLS `ClientConfig`.
///
/// Uses `OnceLock` to ensure only one config is created across the entire process.
/// The built-in rustls `Resumption` (session tickets + session IDs) is enabled
/// by default, so all outbound backend connections share the same session cache.
static SHARED_CLIENT_CONFIG: OnceLock<Arc<rustls::ClientConfig>> = OnceLock::new();
pub fn shared_backend_tls_config() -> Arc<rustls::ClientConfig> {
SHARED_CLIENT_CONFIG.get_or_init(|| {
ensure_crypto_provider();
let config = rustls::ClientConfig::builder()
.dangerous()
.with_custom_certificate_verifier(Arc::new(InsecureVerifier))
.with_no_client_auth();
info!("Built shared backend TLS client config with session resumption");
Arc::new(config)
}).clone()
}
/// Connect to a backend with TLS (for terminate-and-reencrypt mode).
/// Uses the shared backend TLS config for session resumption.
pub async fn connect_tls(
host: &str,
port: u16,
) -> Result<tokio_rustls::client::TlsStream<TcpStream>, Box<dyn std::error::Error + Send + Sync>> {
ensure_crypto_provider();
let config = rustls::ClientConfig::builder()
.dangerous()
.with_custom_certificate_verifier(Arc::new(InsecureVerifier))
.with_no_client_auth();
let connector = TlsConnector::from(Arc::new(config));
let config = shared_backend_tls_config();
let connector = TlsConnector::from(config);
let stream = TcpStream::connect(format!("{}:{}", host, port)).await?;
stream.set_nodelay(true)?;
// Apply keepalive with 60s default (tls_handler doesn't have ConnectionConfig access)
if let Err(e) = crate::socket_opts::apply_keepalive(&stream, std::time::Duration::from_secs(60)) {
debug!("Failed to set keepalive on backend TLS socket: {}", e);
}
let server_name = rustls::pki_types::ServerName::try_from(host.to_string())?;
let tls_stream = connector.connect(server_name, stream).await?;

3
rust/crates/rustproxy/src/lib.rs
View File

@@ -221,6 +221,9 @@ impl RustProxy {
.iter()
.filter_map(|s| s.parse::<std::net::IpAddr>().ok())
.collect(),
keep_alive: options.keep_alive.unwrap_or(true),
keep_alive_initial_delay_ms: options.keep_alive_initial_delay.unwrap_or(60_000),
max_connections: options.max_connections.unwrap_or(100_000),
}
}