diff --git a/CLAUDE.md b/CLAUDE.md index ed6df6e..7883aa4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -238,6 +238,7 @@ Tests bind on random high ports (`port: 0`) to avoid conflicts. Suspended-route ## Non-obvious gotchas +- **The per-connection future is per-connection memory**: `handle`'s future is the tokio task allocated at accept and resident for the connection's life (~3.4 KiB), so it multiplies by the connection count. Await a large, rarely-taken future through `Box::pin` (the HTTP/1 rewriter is ~17 KiB). Moving a large value onward inside the future — into an async fn's arguments, out of a `match` scrutinee, into another async block — can leave its old slot reserved for the rest of the connection (an async fn keeps its arguments beside its working copy; a borrowed or partly-moved local keeps its slot until its scope ends). So the 1.2 KiB `TlsStream` is matched and passed by reference, `copy` is pinned where it is created, and the listeners spawn `handle(..)` itself rather than an `async move` wrapper around it. `proxy_conn::tests::the_per_connection_future_stays_within_budget` fails past 4 KiB under `cargo test` (the debug layout tracks release within a few dozen bytes); `RUSTC_BOOTSTRAP=1 RUSTFLAGS=-Zprint-type-sizes cargo build --release --lib` shows which field grew. - **`copy_bidirectional` half-close**: it returns only once *both* directions have finished — an error or RST on either side ends the whole copy, but a clean EOF on one side does not. That asymmetry is why `liveness.rs` exists. The `ActiveGuard` drop handles both clean close and error paths. - **Keepalive is set on the listening socket, not on accepted ones**: Linux copies `SOCK_KEEPOPEN` and the keepalive timings into each accepted socket and arms its timer (`tcp_create_openreq_child`), so `make_reuseport_socket` configures it once per worker instead of the accept path paying three `setsockopt` calls per connection. `liveness::tests::accepted_sockets_carry_the_configured_keepalive` is the guard: if a platform stops inheriting them, every connection silently loses dead-peer detection, which no round-trip test would notice. - **The half-close bound arms on the client half's `poll_shutdown` completing, not on the upstream's read EOF**: those are different instants, and only the first one means the FIN is out. A `TlsStream` accepts writes into rustls' unbounded buffer and drains them during shutdown, so a clock started at the EOF runs against a flush still in progress and truncates the tail for a slow client. `liveness::tests::a_pending_client_shutdown_does_not_arm_the_bound` pins the ordering. A *client* EOF shuts down the write half to the *upstream* and arms nothing — bounding that would cut a client that legitimately `shutdown(SHUT_WR)`s after its request and then waits on a slow upstream. diff --git a/__test__/liveness.spec.ts b/__test__/liveness.spec.ts index 40af21d..44dd986 100644 --- a/__test__/liveness.spec.ts +++ b/__test__/liveness.spec.ts @@ -11,9 +11,10 @@ import assert from 'node:assert/strict'; import { after, describe, it } from 'node:test'; import * as net from 'node:net'; +import * as tls from 'node:tls'; import { SymphonyProxy } from '../ts/index.js'; -import type { ProxyConfig } from '../ts/types.js'; -import { clientHello, getFreePort, sleep } from './util.js'; +import type { ProxyConfig, RouteConfig } from '../ts/types.js'; +import { clientHello, generateSelfSignedCert, getFreePort, sleep } from './util.js'; const HALF_CLOSE_MS = 300; const SNI = 'liveness.test'; @@ -47,10 +48,11 @@ describe('dead connection reaping', () => { await Promise.all(servers.map((s) => new Promise((r) => s.close(() => r(undefined))))); }); - /** A passthrough proxy in front of `onConnection`, so a client FIN is a plain TCP FIN. */ + /** A proxy in front of `onConnection`; passthrough unless `route` overrides it, so a client FIN is a plain TCP FIN. */ async function startProxy( onConnection: (socket: net.Socket) => void, - overrides: Partial = {} + overrides: Partial = {}, + route: Partial = {} ): Promise<{ proxy: SymphonyProxy; port: number }> { // allowHalfOpen so the upstream models a server that keeps working after the client // half-closes; Node's default would answer that FIN with its own and end the response. @@ -66,7 +68,12 @@ describe('dead connection reaping', () => { const proxy = new SymphonyProxy({ listeners: [{ host: '127.0.0.1', port }], routes: [ - { sni: SNI, upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: upstreamPort }], terminateTls: false }, + { + sni: SNI, + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: upstreamPort }], + terminateTls: false, + ...route, + }, ], halfCloseTimeoutMs: HALF_CLOSE_MS, ...overrides, @@ -198,6 +205,70 @@ describe('dead connection reaping', () => { client.destroy(); }); + const cert = generateSelfSignedCert(SNI); + + /** A TLS-terminated `protocol: 'http'` route with XFF, so the connection runs the HTTP/1 header rewriter. */ + async function rewritingRoundTrip( + halfCloseTimeoutMs: number + ): Promise<{ proxy: SymphonyProxy; client: tls.TLSSocket; request: string; response: string }> { + let request = ''; + const { proxy, port } = await startProxy( + (socket) => + socket.on('data', (chunk) => { + request += chunk; + if (request.includes('\r\n\r\n')) socket.end('HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello'); + }), + { halfCloseTimeoutMs }, + { + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + protocol: 'http', + sourceAddressHeader: 'xForwardedFor', + } + ); + // allowHalfOpen is a tls.connect option, though @types/node 20 does not declare it. + const client = tls.connect({ + host: '127.0.0.1', + port, + servername: SNI, + rejectUnauthorized: false, + allowHalfOpen: true, + } as tls.ConnectionOptions); + client.on('error', () => {}); + sockets.push(client); + const response = await new Promise((resolve, reject) => { + let body = ''; + client.on('data', (c) => (body += c)); + client.on('end', () => resolve(body)); + client.on('close', () => reject(new Error(`closed before the response ended: '${body}'`))); + client.on('secureConnect', () => client.write(`GET / HTTP/1.1\r\nHost: ${SNI}\r\n\r\n`)); + }); + return { proxy, client, request, response }; + } + + it('reclaims a half-closed connection on a route that rewrites HTTP headers', { timeout: 10_000 }, async () => { + const { proxy, request, response } = await rewritingRoundTrip(HALF_CLOSE_MS); + assert.match(request, /\r\nX-Forwarded-For: 127\.0\.0\.1\r\n/i, 'the request went through the rewriter'); + assert.match(response, /\r\n\r\nhello$/); + + await waitFor(() => proxy.metrics().listeners[0].activeConnections === 0); + const listener = proxy.metrics().listeners[0]; + assert.equal(reasonCount(listener.errorsByReason, 'half_closed'), 1); + assert.equal(listener.halfClosedConnections, 0); + }); + + it('holds a half-closed rewriting connection when the bound is disabled', { timeout: 10_000 }, async () => { + const { proxy, client, response } = await rewritingRoundTrip(0); + assert.match(response, /\r\n\r\nhello$/); + + await waitFor(() => proxy.metrics().listeners[0].halfClosedConnections === 1); + await sleep(HALF_CLOSE_MS * 4); + assert.equal(proxy.metrics().listeners[0].activeConnections, 1); + client.destroy(); + await waitFor(() => proxy.metrics().listeners[0].activeConnections === 0); + assert.equal(proxy.metrics().listeners[0].halfClosedConnections, 0); + }); + it('rejects a keepalive schedule that cannot be honoured', async () => { const port = await getFreePort(); const config = (tcpKeepalive: ProxyConfig['tcpKeepalive']): ProxyConfig => ({ diff --git a/src/http_listener.rs b/src/http_listener.rs index c3a0b78..af0d4aa 100644 --- a/src/http_listener.rs +++ b/src/http_listener.rs @@ -100,10 +100,7 @@ async fn accept_loop( if let Some(cfg) = &ctx.keepalive { arm_accepted(&stream, cfg); } - let ctx2 = ctx.clone(); - tokio::spawn(async move { - handle_http(stream, peer_addr, ctx2).await; - }); + tokio::spawn(handle_http(stream, peer_addr, ctx.clone())); } Err(e) => { tracing::error!("accept error on {}: {e}", ctx.listener_addr); diff --git a/src/listener.rs b/src/listener.rs index fe73375..f0c2590 100644 --- a/src/listener.rs +++ b/src/listener.rs @@ -77,10 +77,7 @@ async fn accept_loop( if let Some(cfg) = &ctx.keepalive { arm_accepted(&stream, cfg); } - let ctx2 = ctx.clone(); - tokio::spawn(async move { - handle(stream, peer_addr, ctx2).await; - }); + tokio::spawn(handle(stream, peer_addr, ctx.clone())); } Err(e) => { tracing::error!("accept error on {}: {e}", ctx.listener_addr); diff --git a/src/proxy_conn.rs b/src/proxy_conn.rs index cc32e13..23ca83d 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -240,7 +240,9 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc { // Route by negotiated protocol: h2 connections go to the route's // h2-marked upstreams (e.g. Harper's `-h2.sock` mirror) when present. @@ -301,7 +303,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc, + client: &mut tokio_rustls::server::TlsStream, dest: &Destination, sf: SourceForwarding<'_>, ctx: &ConnContext, @@ -398,7 +400,8 @@ where // of a per-connection cache line per read chunk; do not "optimise" it away without also // deciding to lose that metric. let watch = HalfCloseWatch::new(&ctx.listener_metrics); - let copy = async { + // Pinned here, not inside `body`: moving it into `body` first would store it twice. + let copy = std::pin::pin!(async { write_connection_prefix(upstream, sf).await?; let mut client = Watched::client(client, &watch); let mut upstream = Watched::upstream(upstream, &watch); @@ -412,17 +415,22 @@ where ) .await } else { - crate::http_proxy::proxy_http1_rewriting(&mut client, &mut upstream, &rewrites).await + // Boxed: its state would otherwise sit in every connection's future. + Box::pin(crate::http_proxy::proxy_http1_rewriting( + &mut client, + &mut upstream, + &rewrites, + )) + .await } - }; + }); let body = async { if ctx.half_close_timeout.is_zero() { copy.await.map_err(classify_copy_failure) } else { - tokio::pin!(copy); tokio::select! { - result = &mut copy => result.map_err(classify_copy_failure), + result = copy => result.map_err(classify_copy_failure), _ = watch.expired(ctx.half_close_timeout) => Err(ErrorKind::HalfClosed), } } @@ -799,6 +807,24 @@ mod tests { (max_of(&from_client), max_of(&from_upstream)) } + fn future_size(_: &impl FnOnce(A, B, C) -> F) -> usize { + std::mem::size_of::() + } + + /// `handle`'s future is held for each connection's whole life. The budget fails if a large, + /// rarely-taken future is held inline or the TLS stream is stored by value again, and leaves + /// room for new fields; a small future stored twice can still fit. + #[test] + fn the_per_connection_future_stays_within_budget() { + const BUDGET: usize = 4 * 1024; + let size = future_size(&handle); + assert!( + size <= BUDGET, + "handle's future is {size} bytes, over the {BUDGET}-byte budget: look for a large \ + future held inline or a value stored twice (RUSTC_BOOTSTRAP=1 RUSTFLAGS=-Zprint-type-sizes)" + ); + } + #[tokio::test] async fn each_configured_size_reaches_its_own_direction() { // Deliberately asymmetric, and not the default, so a config that is accepted-then-ignored