Cut parked per-connection memory 59% by shrinking each connection's task

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.

__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<ProxyConfig> = {}
+		overrides: Partial<ProxyConfig> = {},
+		route: Partial<RouteConfig> = {}
 	): 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<string>((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 => ({

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);

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);

src/proxy_conn.rs

@@ -240,7 +240,9 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc<ConnConte
 	let upstream_result = if effective_route.terminate_tls {
 		if let Some(tls_cfg) = effective_route.tls_config {
 			let acceptor = TlsAcceptor::from(tls_cfg);
-			match timeout(hs_timeout, acceptor.accept(stream)).await {
+			// By reference: binding or passing the stream by value would leave a dead copy of it
+			// in the connection's future.
+			match &mut timeout(hs_timeout, acceptor.accept(stream)).await {
 				Ok(Ok(tls_stream)) => {
 					// 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<ConnConte
 // ── Proxy helpers ─────────────────────────────────────────────────────────────
 
 async fn proxy_via_tls(
-	client: tokio_rustls::server::TlsStream<TcpStream>,
+	client: &mut tokio_rustls::server::TlsStream<TcpStream>,
 	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<A, B, C, F: std::future::Future>(_: &impl FnOnce(A, B, C) -> F) -> usize {
+		std::mem::size_of::<F>()
+	}
+
+	/// `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