From 1e6e0469c648875a919e73864b5f8737cb946be4 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 13 Oct 2026 20:51:10 -0600 Subject: [PATCH 1/3] Shrink the per-connection task future from 42.8 KB to 8.2 KB `handle`'s future is the tokio task each accepted connection keeps for its whole life. Two things in `forward` made it 42.8 KB: - The HTTP/1 header-rewriting future (17.4 KB) was awaited inline, so every connection carried its state although only header-rewriting HTTP/1 routes run it. It is now awaited through `Box::pin`, allocated only on those routes. - `copy` was captured into `body` and then moved again by `tokio::pin!`, and rustc keeps the captured and the pinned copy as separate fields. It is now pinned once where it is created. No behavior change: the same pinned future is polled by the same `select!` against the half-close bound. A unit test reads the future's size from its type and fails past 12 KiB; it fails on the previous code (42,872 B). Dispatch-Task: symphony-shrink-connection-future Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DxQT3d8CgJNgAjYp8Pc5ww --- CLAUDE.md | 1 + src/proxy_conn.rs | 40 +++++++++++++++++++++++++++++++++++----- 2 files changed, 36 insertions(+), 5 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index ed6df6e..79fadf0 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 (~8 KiB), so it multiplies by the connection count. Await a large, rarely-taken future through `Box::pin` (the HTTP/1 rewriter is ~17 KiB), and pin a future where it is created rather than moving it into another async block and pinning it there — rustc keeps both copies. `proxy_conn::tests::the_per_connection_future_stays_within_budget` fails past 12 KiB; `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/src/proxy_conn.rs b/src/proxy_conn.rs index cc32e13..a32eaae 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -398,7 +398,9 @@ 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 where it is created: moving it into `body` and pinning it there would keep both the + // captured copy and the pinned one in the connection's future. + 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 +414,23 @@ where ) .await } else { - crate::http_proxy::proxy_http1_rewriting(&mut client, &mut upstream, &rewrites).await + // Boxed so its ~17 KiB of state is allocated only on header-rewriting routes, not + // carried inline 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,28 @@ mod tests { (max_of(&from_client), max_of(&from_upstream)) } + /// Read from the type alone, so the guard below needs no socket, runtime or `ConnContext`. + fn future_size(_: &impl FnOnce(A, B, C) -> F) -> usize { + std::mem::size_of::() + } + + /// `handle`'s future is the per-connection task, resident for the connection's whole life, so + /// at 100k parked connections each KiB of it is ~100 MB of RSS. It is ~8 KiB when every large, + /// rarely-taken branch is boxed and nothing is stored twice; inlining the 17 KiB HTTP/1 + /// rewriter alone takes it to 25 KiB. The 4 KiB of headroom absorbs a new field or a + /// dependency's growth, not a buffer- or future-sized regression. The test profile's layout is + /// within 64 bytes of release. + #[test] + fn the_per_connection_future_stays_within_budget() { + const BUDGET: usize = 12 * 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 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 From ad9016b79bb36b0f6ca96be7486b3e7028af60ef Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 13 Oct 2026 21:09:27 -0600 Subject: [PATCH 2/3] Keep the TLS stream in one slot of the connection future: 8.2 KB to 3.3 KB The accepted 1.2 KB TlsStream was held five times while a TLS-terminated connection was parked: the handshake result, the `tls_stream` binding (kept because it was borrowed), and in `proxy_via_tls` the async fn argument, its working copy and the CountingStream. Only the last was live. Matching the handshake result by reference and passing `&mut TlsStream` on leaves one. The listeners also spawn `handle(..)` / `handle_http(..)` directly instead of an `async move` block that held their arguments a second time (-128 B per task cell). `handle`'s future is now 3,344 B in release; the task cell is 3,456 B, down from main's 43,016 B in two allocations (tokio boxes futures over 16 KiB). The size guard tightens to 4 KiB, so putting the TLS stream back by value fails it. Two liveness cases cover the half-close bound on a TLS-terminated route that rewrites HTTP headers, enabled and disabled. Dispatch-Task: symphony-shrink-connection-future Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DxQT3d8CgJNgAjYp8Pc5ww --- CLAUDE.md | 2 +- __test__/liveness.spec.ts | 84 ++++++++++++++++++++++++++++++++++++--- src/http_listener.rs | 5 +-- src/listener.rs | 5 +-- src/proxy_conn.rs | 23 ++++++----- 5 files changed, 94 insertions(+), 25 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 79fadf0..72610e3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -238,7 +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 (~8 KiB), so it multiplies by the connection count. Await a large, rarely-taken future through `Box::pin` (the HTTP/1 rewriter is ~17 KiB), and pin a future where it is created rather than moving it into another async block and pinning it there — rustc keeps both copies. `proxy_conn::tests::the_per_connection_future_stays_within_budget` fails past 12 KiB; `RUSTC_BOOTSTRAP=1 RUSTFLAGS=-Zprint-type-sizes cargo build --release --lib` shows which field grew. +- **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; `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..f6d8528 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,73 @@ describe('dead connection reaping', () => { client.destroy(); }); + const cert = generateSelfSignedCert(SNI); + + /** + * A terminated `protocol: 'http'` route with XFF, so the copy phase is the HTTP/1 header + * rewriter rather than the plain copy. The upstream answers one request and closes, which + * shuts down the write half to the client; the client stays silent and never sends its FIN. + */ + 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) => { + let body = ''; + client.on('data', (c) => (body += c)); + client.on('end', () => resolve(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', 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', 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 a32eaae..d414ec6 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -240,7 +240,10 @@ 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 +304,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc, + client: &mut tokio_rustls::server::TlsStream, dest: &Destination, sf: SourceForwarding<'_>, ctx: &ConnContext, @@ -807,25 +810,23 @@ mod tests { (max_of(&from_client), max_of(&from_upstream)) } - /// Read from the type alone, so the guard below needs no socket, runtime or `ConnContext`. fn future_size(_: &impl FnOnce(A, B, C) -> F) -> usize { std::mem::size_of::() } - /// `handle`'s future is the per-connection task, resident for the connection's whole life, so - /// at 100k parked connections each KiB of it is ~100 MB of RSS. It is ~8 KiB when every large, - /// rarely-taken branch is boxed and nothing is stored twice; inlining the 17 KiB HTTP/1 - /// rewriter alone takes it to 25 KiB. The 4 KiB of headroom absorbs a new field or a - /// dependency's growth, not a buffer- or future-sized regression. The test profile's layout is - /// within 64 bytes of release. + /// `handle`'s future is the connection's task, resident for its whole life, so each KiB here + /// is ~100 MB at 100k connections. It stays small only while large, rarely-taken futures are + /// boxed and the TLS stream is passed by reference; undoing either adds 1.2 to 17 KiB. The + /// budget leaves ~700 bytes for new fields and dependency growth. A small future stored twice + /// can still fit under it. #[test] fn the_per_connection_future_stays_within_budget() { - const BUDGET: usize = 12 * 1024; + 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 stored twice (RUSTC_BOOTSTRAP=1 RUSTFLAGS=-Zprint-type-sizes)" + future held inline or a value stored twice (RUSTC_BOOTSTRAP=1 RUSTFLAGS=-Zprint-type-sizes)" ); } From 312469f7fe8cfd9d6d77bec31b902c480f5ad8c3 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 13 Oct 2026 21:21:45 -0600 Subject: [PATCH 3/3] Bound the rewriting liveness cases; trim the new comments The response wait in the new liveness cases resolved only on `end`, so a failed handshake or a reset (`error` then `close`, no `end`) would hang the suite. It now rejects on `close` with what arrived, and both cases carry a 10 s timeout. The layout comments drop their size figures, and CLAUDE.md notes that the size guard runs in the debug profile. Dispatch-Task: symphony-shrink-connection-future Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DxQT3d8CgJNgAjYp8Pc5ww --- CLAUDE.md | 2 +- __test__/liveness.spec.ts | 13 +++++-------- src/proxy_conn.rs | 19 +++++++------------ 3 files changed, 13 insertions(+), 21 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 72610e3..7883aa4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -238,7 +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; `RUSTC_BOOTSTRAP=1 RUSTFLAGS=-Zprint-type-sizes cargo build --release --lib` shows which field grew. +- **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 f6d8528..44dd986 100644 --- a/__test__/liveness.spec.ts +++ b/__test__/liveness.spec.ts @@ -207,11 +207,7 @@ describe('dead connection reaping', () => { const cert = generateSelfSignedCert(SNI); - /** - * A terminated `protocol: 'http'` route with XFF, so the copy phase is the HTTP/1 header - * rewriter rather than the plain copy. The upstream answers one request and closes, which - * shuts down the write half to the client; the client stays silent and never sends its FIN. - */ + /** 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 }> { @@ -240,16 +236,17 @@ describe('dead connection reaping', () => { } as tls.ConnectionOptions); client.on('error', () => {}); sockets.push(client); - const response = await new Promise((resolve) => { + 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', async () => { + 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$/); @@ -260,7 +257,7 @@ describe('dead connection reaping', () => { assert.equal(listener.halfClosedConnections, 0); }); - it('holds a half-closed rewriting connection when the bound is disabled', async () => { + 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$/); diff --git a/src/proxy_conn.rs b/src/proxy_conn.rs index d414ec6..23ca83d 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -240,9 +240,8 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc { // Route by negotiated protocol: h2 connections go to the route's @@ -401,8 +400,7 @@ 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); - // Pinned where it is created: moving it into `body` and pinning it there would keep both the - // captured copy and the pinned one in the connection's future. + // 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); @@ -417,8 +415,7 @@ where ) .await } else { - // Boxed so its ~17 KiB of state is allocated only on header-rewriting routes, not - // carried inline in every connection's future. + // Boxed: its state would otherwise sit in every connection's future. Box::pin(crate::http_proxy::proxy_http1_rewriting( &mut client, &mut upstream, @@ -814,11 +811,9 @@ mod tests { std::mem::size_of::() } - /// `handle`'s future is the connection's task, resident for its whole life, so each KiB here - /// is ~100 MB at 100k connections. It stays small only while large, rarely-taken futures are - /// boxed and the TLS stream is passed by reference; undoing either adds 1.2 to 17 KiB. The - /// budget leaves ~700 bytes for new fields and dependency growth. A small future stored twice - /// can still fit under it. + /// `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;