Fix HTTP/2 idle timeout defeated by keep-alive pings (review #008)
The C1 fix from review #007 added keep_alive_interval(15s) + keep_alive_timeout(60s) on the HTTP/2 builders. However, keep_alive_timeout is the timeout for receiving a PONG response to a PING, not an idle timeout. Well-behaved HTTP/2 clients (including crawlers) respond to PINGs, resetting the timeout indefinitely. After 6 days of uptime, the proxy had 930+ idle connections (some 145 hours old) that were never reaped. Fix: add a custom idle timeout that tracks real request activity and closes connections with no in-flight requests for longer than the configured timeout, regardless of PING/PONG activity. - IdleState: per-connection last_activity + in_flight counter - IdleTrackingService: tower Service wrapper that updates last_activity and in_flight on call() and on response completion (prevents killing long-running requests like large git clones) - idle_watchdog: races against serve_connection in tokio::select!, fires only when in_flight == 0 and idle_for >= timeout - HTTP/1.1 path: header_read_timeout already covers between-request idle (verified in hyper source); watchdog is defense-in-depth - HTTP/2 path: watchdog is the primary fix (hyper has no native request-activity-based idle timeout) Also refactored build_manual_server_config to extract build_manual_server_config_from_certs for testability (in-memory certs/keys for integration tests with rcgen). Tests: 8 unit tests for IdleState/watchdog logic, 2 integration tests verifying idle connections are closed and active ones are not. Closes review #008.
This commit is contained in:
@@ -0,0 +1,193 @@
|
||||
---
|
||||
status: open
|
||||
last_updated: 2026-08-10
|
||||
reviewed_code:
|
||||
- src/server.rs
|
||||
- src/proxy/handler.rs
|
||||
reviewer: code-reviewer
|
||||
based_on: docs/reviews/007-connection-lifecycle-and-deployment-drift.md
|
||||
trigger: Post-deployment observation — 930+ idle connections persisting after 6 days uptime despite C1 fix
|
||||
---
|
||||
|
||||
# Follow-Up Review #008 — HTTP/2 Keep-Alive Defeats Idle Timeout
|
||||
|
||||
## Purpose
|
||||
|
||||
Review #007 C1 identified that the reverse-proxy had no server-side idle
|
||||
timeout on TLS connections, causing FD exhaustion. The fix (commit `0885486`)
|
||||
added `keep_alive_interval(15s)` + `keep_alive_timeout(60s)` on the HTTP/2
|
||||
builders and `header_read_timeout(60s)` on the HTTP/1.1 builder. After
|
||||
deploying and running for 6 days, the proxy has **930+ established
|
||||
connections** — some over 145 hours old — that are never closed by the idle
|
||||
timeout. This review examines why the C1 fix is not working as intended and
|
||||
what needs to change.
|
||||
|
||||
---
|
||||
|
||||
## Finding: HTTP/2 keep-alive pings defeat the idle timeout [new code required]
|
||||
|
||||
**Location**: `src/server.rs:119-141`
|
||||
|
||||
### What was implemented (C1 fix)
|
||||
|
||||
```rust
|
||||
// HTTP/2 path (line 119-129)
|
||||
let mut builder = hyper::server::conn::http2::Builder::new(TokioExecutor::new());
|
||||
builder
|
||||
.timer(hyper_util::rt::TokioTimer::new())
|
||||
.keep_alive_interval(Some(Duration::from_secs(15))) // sends PING every 15s
|
||||
.keep_alive_timeout(connection_idle_timeout) // 60s
|
||||
.enable_connect_protocol();
|
||||
|
||||
// HTTP/1.1 + auto path (line 130-141)
|
||||
let mut builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
|
||||
builder
|
||||
.http1()
|
||||
.timer(hyper_util::rt::TokioTimer::new())
|
||||
.header_read_timeout(Some(connection_idle_timeout)); // 60s, first-request only
|
||||
builder
|
||||
.http2()
|
||||
.timer(hyper_util::rt::TokioTimer::new())
|
||||
.keep_alive_interval(Some(Duration::from_secs(15)))
|
||||
.keep_alive_timeout(connection_idle_timeout)
|
||||
.enable_connect_protocol();
|
||||
```
|
||||
|
||||
### Why it doesn't work
|
||||
|
||||
**HTTP/2 path**: `keep_alive_interval(15s)` sends a PING frame every 15
|
||||
seconds. `keep_alive_timeout(60s)` is the timeout for receiving a PONG
|
||||
*response* to a PING. If the client responds to the PING (which any
|
||||
well-behaved HTTP/2 client does), the timeout is reset. The PING/PONG cycle
|
||||
keeps the connection "alive" indefinitely — the timeout only fires if the
|
||||
client *stops responding* to pings (e.g. network failure, crashed client).
|
||||
A crawler that opens an HTTP/2 connection, makes one request, then sits idle
|
||||
but still responds to PINGs will keep the connection open forever.
|
||||
|
||||
This is confirmed by production data: 930+ established connections, some 145
|
||||
hours old (since proxy start 6 days ago), all with `timer: 0` in `ss` output
|
||||
(no active timer). The `keep_alive_timeout` is never firing because clients
|
||||
respond to pings.
|
||||
|
||||
**HTTP/1.1 path**: `header_read_timeout(60s)` only applies to the time
|
||||
waiting for the *first* request headers after the TCP/TLS handshake. Once
|
||||
the first request is received, there is no idle timeout between subsequent
|
||||
requests on the same keep-alive connection. A client that sends one request
|
||||
every 55 seconds keeps the connection alive indefinitely.
|
||||
|
||||
### What hyper actually needs
|
||||
|
||||
hyper does not have a built-in "idle connection timeout" that closes
|
||||
connections after N seconds of no *real* (non-ping) activity. The available
|
||||
options are:
|
||||
|
||||
| Setting | What it does | What we need |
|
||||
|---------|-------------|--------------|
|
||||
| `keep_alive_interval` | Sends PINGs to detect dead peers | Useful, but not an idle timeout |
|
||||
| `keep_alive_timeout` | Timeout for PONG response to a PING | Only fires if client stops responding |
|
||||
| `header_read_timeout` | Timeout for first request headers | Only covers first request, not idle between requests |
|
||||
|
||||
None of these provide: "close this connection if no real HTTP request has
|
||||
been received in the last N seconds."
|
||||
|
||||
### Recommendation
|
||||
|
||||
**Option A (preferred): Wrap `serve_connection` with a tokio timeout
|
||||
that resets on each request.** Implement a custom idle timeout that tracks
|
||||
the last real request time and closes the connection if no request arrives
|
||||
within `connection_idle_timeout`. This can be done by wrapping the service
|
||||
to record a timestamp on each request, and racing `serve_connection` against
|
||||
a sleep timer that resets on each request:
|
||||
|
||||
```rust
|
||||
// Pseudocode for the connection handler:
|
||||
let last_activity = Arc::new(AtomicU64::new(now()));
|
||||
let svc = IdleTrackingService::new(svc, last_activity.clone());
|
||||
|
||||
tokio::select! {
|
||||
result = builder.serve_connection(io, svc) => {
|
||||
// connection completed normally
|
||||
}
|
||||
_ = idle_timeout_watcher(last_activity.clone(), connection_idle_timeout) => {
|
||||
// no real request for connection_idle_timeout — close the connection
|
||||
// (dropping the io handle closes the TLS stream)
|
||||
}
|
||||
}
|
||||
|
||||
async fn idle_timeout_watcher(
|
||||
last_activity: Arc<AtomicU64>,
|
||||
timeout: Duration,
|
||||
) {
|
||||
loop {
|
||||
let elapsed = now() - last_activity.load();
|
||||
let remaining = timeout.saturating_sub(elapsed);
|
||||
if remaining.is_zero() {
|
||||
return; // idle timeout fired
|
||||
}
|
||||
tokio::time::sleep(remaining).await;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The `IdleTrackingService` wraps the inner service and updates
|
||||
`last_activity` on each `call()`. This gives a true idle timeout that fires
|
||||
regardless of PING/PONG activity.
|
||||
|
||||
**Option B: Remove `keep_alive_interval` and rely solely on the
|
||||
`select!`-based timeout.** The keep-alive PING mechanism is useful for
|
||||
detecting dead peers, but it's not a substitute for an idle timeout. If
|
||||
both are needed, keep the PING for liveness detection but add the
|
||||
`select!`-based idle timeout on top.
|
||||
|
||||
**Option C: For HTTP/1.1, also set `keep_alive_timeout` on the
|
||||
`http1()` builder.** hyper's `http1::Builder` has a `keep_alive_timeout`
|
||||
method (distinct from `header_read_timeout`) that sets the timeout for
|
||||
reading the next request on a keep-alive connection. This would close idle
|
||||
HTTP/1.1 connections after the timeout. However, this doesn't help with
|
||||
HTTP/2, which is where most of the idle connections are (crawlers default to
|
||||
HTTP/2 via ALPN).
|
||||
|
||||
### HTTP/1.1 fix
|
||||
|
||||
For the HTTP/1.1 path (`auto::Builder`), add
|
||||
`.http1().keep_alive_timeout(Some(connection_idle_timeout))` in addition to
|
||||
`header_read_timeout`. This covers idle time between requests on keep-alive
|
||||
HTTP/1.1 connections.
|
||||
|
||||
### Verification after fix
|
||||
|
||||
After implementing the true idle timeout, verify with:
|
||||
1. Open an HTTP/2 connection to the proxy, send one request, then idle
|
||||
2. Confirm the connection closes after `connection_idle_timeout` (60s)
|
||||
3. Check FD count drops to baseline (~25) when traffic stops
|
||||
4. Confirm active connections (making requests every <60s) are not closed
|
||||
|
||||
### Production impact
|
||||
|
||||
Without this fix, the proxy accumulates idle HTTP/2 connections from every
|
||||
crawler/search-engine that connects. After 6 days of uptime with the C1 fix
|
||||
deployed, the proxy has 930+ idle connections (some 145 hours old) from
|
||||
~625 unique IPs. The `max_connections = 1024` semaphore prevents FD
|
||||
exhaustion, but the proxy is near the cap, and legitimate new connections
|
||||
will start being blocked if the count reaches 1024.
|
||||
|
||||
The connections don't consume CPU or I/O when truly idle, but they hold FDs
|
||||
and TLS state (~60KB each in memory). At 930 connections, that's ~56MB of
|
||||
TLS state for connections that should have been closed hours or days ago.
|
||||
|
||||
---
|
||||
|
||||
## Summary
|
||||
|
||||
| Issue | Status | Action |
|
||||
|-------|--------|--------|
|
||||
| C1 fix (keep_alive_timeout) | Deployed but ineffective | Replace with true idle timeout (Option A) |
|
||||
| HTTP/1.1 header_read_timeout | Deployed but only covers first request | Add `keep_alive_timeout` for between-request idle |
|
||||
| HTTP/2 keep_alive_interval | Deployed, keeps connections alive | Remove or keep alongside true idle timeout |
|
||||
|
||||
## References
|
||||
|
||||
- [Review #007](007-connection-lifecycle-and-deployment-drift.md) — original C1 finding and fix
|
||||
- [hyper http2::Builder docs](https://docs.rs/hyper/latest/hyper/server/conn/http2/struct.Builder.html) — `keep_alive_interval`, `keep_alive_timeout`
|
||||
- [hyper-util auto::Builder docs](https://docs.rs/hyper-util/latest/hyper_util/server/conn/auto/struct.Builder.html) — `http1().keep_alive_timeout()`
|
||||
- Production observation: 930+ idle connections, 145h old, dev1 2026-08-10
|
||||
+270
-12
@@ -1,7 +1,8 @@
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
use std::future::Future;
|
||||
|
||||
use axum::extract::ConnectInfo;
|
||||
use axum::http::Request;
|
||||
@@ -14,7 +15,7 @@ use tokio::net::TcpListener;
|
||||
use tokio::sync::Semaphore;
|
||||
use tokio_rustls::TlsAcceptor;
|
||||
use tower::Service;
|
||||
use tracing::{error, info, warn};
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
const HTTP2_KEEP_ALIVE_INTERVAL_SECS: u64 = 15;
|
||||
|
||||
@@ -55,6 +56,96 @@ impl InFlightCounter {
|
||||
pub fn is_zero(&self) -> bool {
|
||||
self.count.load(Ordering::SeqCst) == 0
|
||||
}
|
||||
|
||||
pub fn count(&self) -> usize {
|
||||
self.count.load(Ordering::SeqCst)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct IdleState {
|
||||
last_activity: Mutex<Instant>,
|
||||
in_flight: AtomicUsize,
|
||||
}
|
||||
|
||||
impl IdleState {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
last_activity: Mutex::new(Instant::now()),
|
||||
in_flight: AtomicUsize::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
fn touch(&self) {
|
||||
*self.last_activity.lock().unwrap() = Instant::now();
|
||||
}
|
||||
|
||||
fn idle_for(&self) -> Duration {
|
||||
Instant::now().duration_since(*self.last_activity.lock().unwrap())
|
||||
}
|
||||
|
||||
fn in_flight(&self) -> usize {
|
||||
self.in_flight.load(Ordering::SeqCst)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct IdleTrackingService<S> {
|
||||
inner: S,
|
||||
idle_state: Arc<IdleState>,
|
||||
}
|
||||
|
||||
impl<S> Service<Request<Incoming>> for IdleTrackingService<S>
|
||||
where
|
||||
S: Service<Request<Incoming>, Response = Response> + Clone + Send + 'static,
|
||||
S::Future: Send + 'static,
|
||||
{
|
||||
type Response = S::Response;
|
||||
type Error = S::Error;
|
||||
type Future = std::pin::Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
|
||||
|
||||
fn poll_ready(
|
||||
&mut self,
|
||||
cx: &mut std::task::Context<'_>,
|
||||
) -> std::task::Poll<Result<(), Self::Error>> {
|
||||
self.inner.poll_ready(cx)
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Request<Incoming>) -> Self::Future {
|
||||
self.idle_state.touch();
|
||||
self.idle_state.in_flight.fetch_add(1, Ordering::SeqCst);
|
||||
|
||||
let inner_fut = self.inner.call(req);
|
||||
let idle_state = self.idle_state.clone();
|
||||
|
||||
Box::pin(async move {
|
||||
let result = inner_fut.await;
|
||||
idle_state.in_flight.fetch_sub(1, Ordering::SeqCst);
|
||||
idle_state.touch();
|
||||
result
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async fn idle_watchdog(idle_state: Arc<IdleState>, timeout: Duration) {
|
||||
loop {
|
||||
let idle_for = idle_state.idle_for();
|
||||
let remaining = timeout.saturating_sub(idle_for);
|
||||
|
||||
let sleep_dur = if idle_state.in_flight() > 0 {
|
||||
timeout
|
||||
} else if !remaining.is_zero() {
|
||||
remaining
|
||||
} else {
|
||||
return;
|
||||
};
|
||||
|
||||
tokio::time::sleep(sleep_dur).await;
|
||||
|
||||
if idle_state.in_flight() == 0 && idle_state.idle_for() >= timeout {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn serve_https_listener(
|
||||
@@ -108,10 +199,16 @@ pub async fn serve_https_listener(
|
||||
let alpn = tls_stream.get_ref().1.alpn_protocol();
|
||||
let is_h2 = alpn == Some(b"h2");
|
||||
|
||||
let idle_state = Arc::new(IdleState::new());
|
||||
|
||||
let svc = ConnectInfoService {
|
||||
inner: router.into_service::<Incoming>(),
|
||||
remote_addr,
|
||||
};
|
||||
let svc = IdleTrackingService {
|
||||
inner: svc,
|
||||
idle_state: idle_state.clone(),
|
||||
};
|
||||
let svc = TowerToHyperService::new(svc);
|
||||
|
||||
let io = hyper_util::rt::TokioIo::new(tls_stream);
|
||||
@@ -123,9 +220,20 @@ pub async fn serve_https_listener(
|
||||
.keep_alive_interval(Some(Duration::from_secs(HTTP2_KEEP_ALIVE_INTERVAL_SECS)))
|
||||
.keep_alive_timeout(connection_idle_timeout)
|
||||
.enable_connect_protocol();
|
||||
if let Err(e) = builder.serve_connection(io, svc).await
|
||||
{
|
||||
error!(error = %e, "HTTPS/2 connection error");
|
||||
|
||||
tokio::select! {
|
||||
result = builder.serve_connection(io, svc) => {
|
||||
if let Err(e) = result {
|
||||
error!(error = %e, "HTTPS/2 connection error");
|
||||
}
|
||||
}
|
||||
_ = idle_watchdog(idle_state.clone(), connection_idle_timeout) => {
|
||||
debug!(
|
||||
remote_addr = %remote_addr,
|
||||
idle_timeout_secs = connection_idle_timeout.as_secs(),
|
||||
"closing idle HTTP/2 connection (no real request activity)"
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let mut builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
|
||||
@@ -139,14 +247,25 @@ pub async fn serve_https_listener(
|
||||
.keep_alive_interval(Some(Duration::from_secs(HTTP2_KEEP_ALIVE_INTERVAL_SECS)))
|
||||
.keep_alive_timeout(connection_idle_timeout)
|
||||
.enable_connect_protocol();
|
||||
if let Err(e) = builder.serve_connection_with_upgrades(io, svc).await
|
||||
{
|
||||
if let Some(hyper_err) = e.downcast_ref::<hyper::Error>() {
|
||||
if hyper_err.is_incomplete_message() {
|
||||
return;
|
||||
|
||||
tokio::select! {
|
||||
result = builder.serve_connection_with_upgrades(io, svc) => {
|
||||
if let Err(e) = result {
|
||||
if let Some(hyper_err) = e.downcast_ref::<hyper::Error>() {
|
||||
if hyper_err.is_incomplete_message() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
error!(error = %e, "HTTPS connection error");
|
||||
}
|
||||
}
|
||||
error!(error = %e, "HTTPS connection error");
|
||||
_ = idle_watchdog(idle_state.clone(), connection_idle_timeout) => {
|
||||
debug!(
|
||||
remote_addr = %remote_addr,
|
||||
idle_timeout_secs = connection_idle_timeout.as_secs(),
|
||||
"closing idle connection (no real request activity)"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -206,3 +325,142 @@ where
|
||||
self.inner.call(req)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::time::Duration;
|
||||
|
||||
#[test]
|
||||
fn idle_state_starts_not_idle() {
|
||||
let state = IdleState::new();
|
||||
assert!(state.idle_for() < Duration::from_millis(100));
|
||||
assert_eq!(state.in_flight(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn idle_state_touch_updates_last_activity() {
|
||||
let state = IdleState::new();
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
assert!(state.idle_for() >= Duration::from_millis(50));
|
||||
|
||||
state.touch();
|
||||
assert!(state.idle_for() < Duration::from_millis(10));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn idle_state_in_flight_tracks_increments_and_decrements() {
|
||||
let state = IdleState::new();
|
||||
assert_eq!(state.in_flight(), 0);
|
||||
state.in_flight.fetch_add(1, Ordering::SeqCst);
|
||||
assert_eq!(state.in_flight(), 1);
|
||||
state.in_flight.fetch_add(1, Ordering::SeqCst);
|
||||
assert_eq!(state.in_flight(), 2);
|
||||
state.in_flight.fetch_sub(1, Ordering::SeqCst);
|
||||
assert_eq!(state.in_flight(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn idle_watchdog_fires_after_timeout_when_idle() {
|
||||
let state = Arc::new(IdleState::new());
|
||||
let timeout = Duration::from_millis(50);
|
||||
|
||||
let start = Instant::now();
|
||||
idle_watchdog(state.clone(), timeout).await;
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
assert!(
|
||||
elapsed >= timeout,
|
||||
"watchdog fired before timeout: {:?} < {:?}",
|
||||
elapsed,
|
||||
timeout
|
||||
);
|
||||
assert!(
|
||||
elapsed < timeout + Duration::from_millis(100),
|
||||
"watchdog took too long: {:?}",
|
||||
elapsed
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn idle_watchdog_does_not_fire_while_request_in_flight() {
|
||||
let state = Arc::new(IdleState::new());
|
||||
let timeout = Duration::from_millis(50);
|
||||
|
||||
state.in_flight.fetch_add(1, Ordering::SeqCst);
|
||||
|
||||
let watchdog = tokio::time::timeout(
|
||||
Duration::from_millis(200),
|
||||
idle_watchdog(state.clone(), timeout),
|
||||
);
|
||||
|
||||
let result = watchdog.await;
|
||||
assert!(
|
||||
result.is_err(),
|
||||
"watchdog fired while a request was in-flight (should have kept sleeping)"
|
||||
);
|
||||
assert_eq!(state.in_flight(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn idle_watchdog_resets_after_request_completes() {
|
||||
let state = Arc::new(IdleState::new());
|
||||
let timeout = Duration::from_millis(50);
|
||||
|
||||
state.in_flight.fetch_add(1, Ordering::SeqCst);
|
||||
std::thread::sleep(Duration::from_millis(30));
|
||||
|
||||
state.in_flight.fetch_sub(1, Ordering::SeqCst);
|
||||
state.touch();
|
||||
|
||||
let start = Instant::now();
|
||||
idle_watchdog(state.clone(), timeout).await;
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
assert!(
|
||||
elapsed >= timeout,
|
||||
"watchdog fired before full timeout after touch(): {:?} < {:?}",
|
||||
elapsed,
|
||||
timeout
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn idle_watchdog_rechecks_after_short_sleep_when_partially_idle() {
|
||||
let state = Arc::new(IdleState::new());
|
||||
let timeout = Duration::from_millis(100);
|
||||
|
||||
std::thread::sleep(Duration::from_millis(40));
|
||||
|
||||
let start = Instant::now();
|
||||
idle_watchdog(state.clone(), timeout).await;
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
assert!(
|
||||
elapsed >= Duration::from_millis(60),
|
||||
"watchdog fired too soon after partial idle: {:?}",
|
||||
elapsed
|
||||
);
|
||||
assert!(
|
||||
elapsed < Duration::from_millis(200),
|
||||
"watchdog took too long: {:?}",
|
||||
elapsed
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn idle_watchdog_never_fires_with_persistent_in_flight() {
|
||||
let state = Arc::new(IdleState::new());
|
||||
let timeout = Duration::from_millis(50);
|
||||
|
||||
state.in_flight.fetch_add(1, Ordering::SeqCst);
|
||||
|
||||
let result = tokio::time::timeout(
|
||||
Duration::from_millis(300),
|
||||
idle_watchdog(state.clone(), timeout),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(result.is_err(), "watchdog should not fire with in-flight > 0");
|
||||
}
|
||||
}
|
||||
+14
-2
@@ -56,6 +56,20 @@ pub fn build_manual_server_config(cert_path: &str, key_path: &str) -> Result<Ser
|
||||
let certs = load_certs(cert_path)?;
|
||||
let key = load_private_key(key_path)?;
|
||||
|
||||
build_server_config(certs, key)
|
||||
}
|
||||
|
||||
pub fn build_manual_server_config_from_certs(
|
||||
certs: Vec<CertificateDer<'static>>,
|
||||
key: PrivateKeyDer<'static>,
|
||||
) -> Result<ServerConfig> {
|
||||
build_server_config(certs, key)
|
||||
}
|
||||
|
||||
fn build_server_config(
|
||||
certs: Vec<CertificateDer<'static>>,
|
||||
key: PrivateKeyDer<'static>,
|
||||
) -> Result<ServerConfig> {
|
||||
let provider = crypto_provider();
|
||||
let config = ServerConfig::builder_with_provider(provider)
|
||||
.with_protocol_versions(&[&TLS12, &TLS13])
|
||||
@@ -65,8 +79,6 @@ pub fn build_manual_server_config(cert_path: &str, key_path: &str) -> Result<Ser
|
||||
.with_context(|| "failed to configure certificate/key pair")?;
|
||||
|
||||
let mut config = config;
|
||||
// Advertise HTTP/2 and HTTP/1.1 via ALPN so clients can negotiate HTTP/2.
|
||||
// Note: acme-tls/1 is NOT included here — it's only needed for ACME mode.
|
||||
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
|
||||
|
||||
Ok(config)
|
||||
|
||||
@@ -964,3 +964,258 @@ async fn test_graceful_shutdown_with_health_check() {
|
||||
|
||||
handle.abort();
|
||||
}
|
||||
|
||||
mod idle_timeout_tests {
|
||||
use super::*;
|
||||
use reverse_proxy::server::InFlightCounter;
|
||||
use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_rustls::TlsAcceptor;
|
||||
|
||||
fn make_test_tls_acceptor() -> TlsAcceptor {
|
||||
let mut params = rcgen::CertificateParams::new(vec!["test.local".to_string()]).unwrap();
|
||||
params.distinguished_name = rcgen::DistinguishedName::new();
|
||||
params
|
||||
.distinguished_name
|
||||
.push(rcgen::DnType::CommonName, "test.local");
|
||||
let key_pair = rcgen::KeyPair::generate().unwrap();
|
||||
let cert = params.self_signed(&key_pair).unwrap();
|
||||
let cert_der = cert.der().clone();
|
||||
let key_der = key_pair.serialize_der();
|
||||
let private_key = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key_der));
|
||||
|
||||
let server_config = reverse_proxy::tls::config::build_manual_server_config_from_certs(
|
||||
vec![cert_der],
|
||||
private_key,
|
||||
)
|
||||
.unwrap();
|
||||
TlsAcceptor::from(Arc::new(server_config))
|
||||
}
|
||||
|
||||
fn make_proxy_router() -> (
|
||||
Arc<reverse_proxy::proxy::ProxyState>,
|
||||
Arc<ArcSwap<DynamicConfig>>,
|
||||
Arc<reverse_proxy::rate_limit::RateLimiter>,
|
||||
) {
|
||||
let sites = vec![SiteConfig {
|
||||
host: "test.local".to_string(),
|
||||
upstream: "127.0.0.1:18080".to_string(),
|
||||
upstream_scheme: "http".to_string(),
|
||||
upstream_connect_timeout_secs: 5,
|
||||
upstream_request_timeout_secs: 60,
|
||||
}];
|
||||
let config = DynamicConfig::from_sites(
|
||||
sites,
|
||||
RateLimitConfig {
|
||||
requests_per_second: 100,
|
||||
burst: 100,
|
||||
},
|
||||
BodyConfig {
|
||||
limit_bytes: 104857600,
|
||||
},
|
||||
);
|
||||
let config_arc = Arc::new(ArcSwap::from_pointee(config));
|
||||
let proxy_state = Arc::new(reverse_proxy::proxy::ProxyState {
|
||||
config: config_arc.clone(),
|
||||
http_client: reverse_proxy::proxy::create_http_client(),
|
||||
https_client: reverse_proxy::proxy::create_https_client(),
|
||||
});
|
||||
let rate_limiter =
|
||||
Arc::new(reverse_proxy::rate_limit::RateLimiter::new(config_arc.clone()));
|
||||
(proxy_state, config_arc, rate_limiter)
|
||||
}
|
||||
|
||||
async fn start_test_https_server(
|
||||
idle_timeout: Duration,
|
||||
) -> (
|
||||
std::net::SocketAddr,
|
||||
Arc<InFlightCounter>,
|
||||
tokio::task::JoinHandle<()>,
|
||||
tokio::sync::watch::Sender<bool>,
|
||||
) {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let tls_acceptor = make_test_tls_acceptor();
|
||||
let (proxy_state, config_arc, rate_limiter) = make_proxy_router();
|
||||
let router = reverse_proxy::proxy::build_router(proxy_state, config_arc, rate_limiter);
|
||||
let in_flight = InFlightCounter::new();
|
||||
let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
|
||||
|
||||
let in_flight_clone = in_flight.clone();
|
||||
let handle = tokio::spawn(async move {
|
||||
reverse_proxy::server::serve_https_listener(
|
||||
listener,
|
||||
tls_acceptor,
|
||||
router,
|
||||
shutdown_rx,
|
||||
in_flight_clone,
|
||||
idle_timeout,
|
||||
1024,
|
||||
)
|
||||
.await;
|
||||
});
|
||||
|
||||
(addr, in_flight, handle, shutdown_tx)
|
||||
}
|
||||
|
||||
fn make_client_tls_config() -> Arc<rustls::ClientConfig> {
|
||||
let mut roots = rustls::RootCertStore::empty();
|
||||
let _ = roots.add(CertificateDer::from_slice(b"test".to_vec().as_slice()));
|
||||
let config = rustls::ClientConfig::builder()
|
||||
.dangerous()
|
||||
.with_custom_certificate_verifier(Arc::new(NoVerifyVerifier))
|
||||
.with_no_client_auth();
|
||||
Arc::new(config)
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct NoVerifyVerifier;
|
||||
|
||||
impl rustls::client::danger::ServerCertVerifier for NoVerifyVerifier {
|
||||
fn verify_server_cert(
|
||||
&self,
|
||||
_end_entity: &CertificateDer<'_>,
|
||||
_intermediates: &[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: &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: &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::ECDSA_NISTP256_SHA256,
|
||||
rustls::SignatureScheme::RSA_PSS_SHA256,
|
||||
rustls::SignatureScheme::RSA_PKCS1_SHA384,
|
||||
rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
|
||||
rustls::SignatureScheme::RSA_PSS_SHA384,
|
||||
rustls::SignatureScheme::RSA_PKCS1_SHA512,
|
||||
rustls::SignatureScheme::RSA_PSS_SHA512,
|
||||
rustls::SignatureScheme::ED25519,
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
async fn connect_tls(addr: std::net::SocketAddr) -> tokio_rustls::client::TlsStream<tokio::net::TcpStream> {
|
||||
let tcp_stream = tokio::net::TcpStream::connect(addr).await.unwrap();
|
||||
let server_name = rustls::pki_types::ServerName::try_from("test.local").unwrap();
|
||||
let connector = tokio_rustls::TlsConnector::from(make_client_tls_config());
|
||||
connector.connect(server_name, tcp_stream).await.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn idle_http1_connection_closed_after_timeout() {
|
||||
let idle_timeout = Duration::from_millis(500);
|
||||
let (addr, in_flight, _handle, _shutdown_tx) = start_test_https_server(idle_timeout).await;
|
||||
|
||||
let mut tls_stream = connect_tls(addr).await;
|
||||
|
||||
tls_stream
|
||||
.write_all(b"GET / HTTP/1.1\r\nHost: test.local\r\nConnection: keep-alive\r\n\r\n")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let mut buf = vec![0u8; 4096];
|
||||
let n = tokio::time::timeout(Duration::from_secs(2), tls_stream.read(&mut buf))
|
||||
.await
|
||||
.expect("timeout waiting for first response")
|
||||
.expect("read error");
|
||||
let response = String::from_utf8_lossy(&buf[..n]);
|
||||
assert!(
|
||||
response.starts_with("HTTP/1.1 "),
|
||||
"expected HTTP response, got: {response}"
|
||||
);
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
assert_eq!(
|
||||
in_flight.count(),
|
||||
1,
|
||||
"connection should still be in-flight right after response"
|
||||
);
|
||||
|
||||
let read_result = tokio::time::timeout(
|
||||
idle_timeout + Duration::from_secs(2),
|
||||
tls_stream.read(&mut buf),
|
||||
)
|
||||
.await
|
||||
.expect("timeout waiting for idle close");
|
||||
|
||||
let closed = match read_result {
|
||||
Ok(0) => true,
|
||||
Ok(_) => false,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => true,
|
||||
Err(e) => panic!("unexpected read error: {e:?}"),
|
||||
};
|
||||
assert!(
|
||||
closed,
|
||||
"connection should be closed by server after idle timeout"
|
||||
);
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
assert_eq!(
|
||||
in_flight.count(),
|
||||
0,
|
||||
"in-flight count should be 0 after connection closed"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn active_http1_connection_not_closed_within_timeout() {
|
||||
let idle_timeout = Duration::from_secs(2);
|
||||
let (addr, in_flight, _handle, _shutdown_tx) = start_test_https_server(idle_timeout).await;
|
||||
|
||||
let mut tls_stream = connect_tls(addr).await;
|
||||
|
||||
tls_stream
|
||||
.write_all(b"GET / HTTP/1.1\r\nHost: test.local\r\nConnection: keep-alive\r\n\r\n")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let mut buf = vec![0u8; 4096];
|
||||
let n = tokio::time::timeout(Duration::from_secs(2), tls_stream.read(&mut buf))
|
||||
.await
|
||||
.expect("timeout waiting for first response")
|
||||
.expect("read error");
|
||||
assert!(n > 0);
|
||||
|
||||
let read_result = tokio::time::timeout(
|
||||
idle_timeout / 2,
|
||||
tls_stream.read(&mut buf),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
read_result.is_err(),
|
||||
"connection should NOT be closed within idle timeout"
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
in_flight.count(),
|
||||
1,
|
||||
"connection should still be in-flight (active)"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user