TLS session resumption for resolved suspended routes; reclaim expired IP-affinity entries

.github/workflows/CI.yml

@@ -44,7 +44,7 @@ jobs:
           restore-keys: lint-cargo-
 
       - name: cargo clippy
-        run: cargo clippy --all-targets
+        run: cargo clippy --all-targets -- -D warnings
 
       - name: cargo fmt
         run: cargo fmt --check

CLAUDE.md

@@ -135,17 +135,18 @@ Each suspended connection gets a `tokio::sync::oneshot::channel`. The sender is
 ### Per-route Arc<ServerConfig> deduplication (TlsConfigCache)
 Routes that share the same cert+mTLS combination share a single `Arc<ServerConfig>` allocation. The cache key is `(sha256(cert_pem + key_pem), sha256(mtls_ca_pem), http2)`. Important for deployments where many routes share a wildcard cert.
 
-The cache is owned by `SymphonyProxyWrap` and **threaded through every `build_route_table`** rather than created per build. That is what makes TLS session resumption survive a config reload, and it is the non-obvious part: `build_server_config` gives each `ServerConfig` its own `session_storage` (TLS 1.2 session IDs) and its own `Ticketer` (TLS 1.3 ticket keys), so minting a fresh config for an *unchanged* cert silently invalidates every ticket already issued. Nothing errors — clients just quietly fall back to full handshakes. Since a route add/remove or an on-disk cert renewal rebuilds the whole table, a per-build cache meant clients almost never resumed. Keying on the cert bytes gives the right lifetime for free: session state lives exactly as long as the cert it was issued under, and a rotation retires it.
+The cache is owned by `SymphonyProxyWrap` and **threaded through every `build_route_table`** (and every `resolveConnection`) rather than created per build. That is what makes TLS session resumption survive a config reload, and it is the non-obvious part: `build_server_config` gives each `ServerConfig` its own `session_storage` (TLS 1.2 session IDs) and its own `Ticketer` (TLS 1.3 ticket keys), so minting a fresh config for an *unchanged* cert silently invalidates every ticket already issued. Nothing errors — clients just quietly fall back to full handshakes. Since a route add/remove or an on-disk cert renewal rebuilds the whole table, a per-build cache meant clients almost never resumed. Keying on the cert bytes gives the right lifetime for free: session state lives exactly as long as the cert it was issued under, and a rotation retires it.
 
-Three properties keep that honest:
+Four properties keep that honest:
 
 - **What survives a sweep is decided by the refcount, not the mark set.** `retain_used()` keeps an entry when it was marked during the build *or* when `Arc::strong_count > 1`, and the second clause is the load-bearing one: the carry-forward branch in `build_route_table` (mid-rotation cert/key mismatch) reuses the previous table's `Arc` without going through `get_or_build`, so a live route can be unmarked. Since the table is swapped in before the sweep runs, a committed table's configs always have `strong_count >= 2`. `clear_used()` at the top of a build evicts on the same rule inverted (`strong_count == 1` — cache-only), which is what bounds a reload that fails validation over and over.
 - **The sweep is the caller's, at the commit point.** `retain_used()` (mark-and-sweep over the keys touched during a build) runs in `proxy.rs` *after* the new table is swapped in — never inside `build_route_table`. `updateConfig` is all-or-nothing: a route build can succeed and then be discarded when the protection half fails validation, and a sweep driven by that build's marks names none of the live table's configs. The refcount clause above spares them regardless, so this is the belt to that suspenders — but it is what keeps the placement from depending on it. Over-retaining for one generation is the safe direction; under-retaining costs live session state.
+- **Configs nothing references are bounded without a reload.** `suspended.rs::build_resolved_route` draws from the same cache, but no table holds a resolved config, so it lives by refcount alone: kept while a connection holds it, otherwise dropped by the next build's `clear_used()` (even one `updateConfig` then rejects) or when an insert pushes the cache past `evict_at`. That threshold is `max(256, 2 × the configs still referenced after the last eviction)`. Without it, a proxy that never reloads keeps a config — with a session store preallocated for 1024 entries — for every distinct cert JS has ever resolved; the doubling keeps the scan amortised O(1) when the route table alone holds more than 256. A resolution carrying the same material as a table route *is* that route's entry, which is safe because the cache key covers every input to what a resumed session may skip (cert, key, client-auth CA and requirement, ALPN).
 - **Ticket keys stay per-config, not process-global.** A process-wide ticketer would let a ticket minted under one tenant's cert resume against another tenant's route. Sharing across configs is the obvious "optimisation" here and it is the wrong one.
 
-Scope: this is one proxy's *route table* cache, and it lives as long as that `SymphonyProxy` does. A reload that only changes routes keeps it; anything that changes a listener signature — including an on-disk rotation of a listener `defaultCert`/mTLS CA, since `server.ts` computes `listenerSig` over the *resolved* listeners — recreates the proxy, and every route on that port starts over with fresh ticketers. Hoisting the cache to process scope would fix that (it is keyed on cert bytes, so it cannot leak a ticket across tenants) and is not done here. `suspended.rs::build_resolved_route` is outside it too: it builds from a throwaway cache per resolution, so a JS-resolved suspended connection re-parses its cert and gets a fresh ticketer — those clients never resume.
+Scope: the cache belongs to one proxy and lives as long as that `SymphonyProxy` does. A reload that only changes routes keeps it; anything that changes a listener signature — including an on-disk rotation of a listener `defaultCert`/mTLS CA, since `server.ts` computes `listenerSig` over the *resolved* listeners — recreates the proxy, and every route on that port starts over with fresh ticketers. Hoisting the cache to process scope would fix that (it is keyed on cert bytes, so it cannot leak a ticket across tenants) and is not done here.
 
-Two further things are deliberately *not* solved: rustls' ring `Ticketer` rotates keys on its own ~6h schedule (independent of reloads), and ticket keys are per-process, so during host-manager's overlapping-process upgrade window clients that land on the new process fall back to a full handshake until they hold one of its tickets. Both degrade to a full handshake, never to an error. `__test__/session-resumption.spec.ts` covers all of this end-to-end (TLS 1.2 and 1.3, reload, and rotation).
+Two further things are deliberately *not* solved: rustls' ring `Ticketer` rotates keys on its own ~6h schedule (independent of reloads), and ticket keys are per-process, so during host-manager's overlapping-process upgrade window clients that land on the new process fall back to a full handshake until they hold one of its tickets. Both degrade to a full handshake, never to an error. `__test__/session-resumption.spec.ts` covers all of this end-to-end (TLS 1.2 and 1.3, reload, rotation, and suspended-route resolutions).
 
 ### Send + Sync in napi classes
 napi `Buffer` contains raw pointers (`*mut napi_env__`, `*mut napi_ref__`) that are not `Sync`. The `SymphonyProxyWrap` struct must be `Send + Sync`. Solution: napi types (`Buffer`, `JsCertConfig`, etc.) are only used as constructor/method *parameters*, immediately converted to plain Rust (`Vec<u8>`, `ListenerTlsSpec`, etc.), and never stored in the struct. The struct only holds types that are provably `Send + Sync`.

__test__/session-resumption.spec.ts

@@ -42,8 +42,8 @@ describe('TLS session resumption', () => {
 	});
 
 	after(async () => {
-		await proxy.stop();
-		await echo.close();
+		await proxy?.stop();
+		await echo?.close();
 	});
 
 	for (const maxVersion of ['TLSv1.3', 'TLSv1.2'] as const) {
@@ -122,3 +122,58 @@ describe('TLS session resumption', () => {
 		await sleep(50);
 	});
 });
+
+// A resolved route's TLS config comes from resolveConnection(), not the route table. The suspended
+// route below carries no cert of its own, so a resumed second connection can only mean that both
+// resolutions were handed the same ServerConfig — and with it the same ticketer and session store.
+describe('TLS session resumption – suspended routes resolved with a cert', () => {
+	const cert = generateSelfSignedCert('localhost');
+	let proxyPort: number;
+	let echo: Awaited<ReturnType<typeof startEchoServer>>;
+	let proxy: SymphonyProxy;
+	let resolutions = 0;
+
+	before(async () => {
+		echo = await startEchoServer();
+		proxyPort = await getFreePort();
+		proxy = new SymphonyProxy({
+			listeners: [{ host: '127.0.0.1', port: proxyPort }],
+			routes: [{ sni: 'localhost', upstreams: [], terminateTls: false, suspended: true, suspendTimeoutMs: 5000 }],
+		});
+		proxy.on('suspended', (conn) => {
+			resolutions++;
+			proxy.resolveConnection(conn.id, {
+				upstream: { kind: 'tcp', host: '127.0.0.1', port: echo.port },
+				terminateTls: true,
+				cert: { certChain: cert.cert, privateKey: cert.key },
+			});
+		});
+		await proxy.start();
+		await sleep(50);
+	});
+
+	after(async () => {
+		await proxy?.stop();
+		await echo?.close();
+	});
+
+	for (const maxVersion of ['TLSv1.3', 'TLSv1.2'] as const) {
+		it(`resumes a ${maxVersion} session issued to an earlier resolution of the same cert`, async () => {
+			const resolutionsBefore = resolutions;
+			const first = await tlsHandshake({ port: proxyPort, servername: 'localhost', caCert: cert.cert, maxVersion });
+			assert.equal(first.protocol, maxVersion);
+			assert.equal(first.reused, false);
+			assert.ok(first.session, 'the server must issue a session to resume against');
+
+			const second = await tlsHandshake({
+				port: proxyPort,
+				servername: 'localhost',
+				caCert: cert.cert,
+				maxVersion,
+				session: first.session,
+			});
+			assert.equal(second.reused, true, 'each resolution must not mint its own ServerConfig for an unchanged cert');
+			assert.equal(resolutions - resolutionsBefore, 2, 'both connections must have gone through resolveConnection');
+		});
+	}
+});

__test__/suspended.spec.ts

@@ -106,9 +106,9 @@ describe('Suspended routes – hold then resolve', () => {
 
 		// 3. Resolve the connection — this triggers the TLS handshake + proxying.
 		//    terminateTls: true = proxy terminates TLS from the client, then forwards
-		//    plaintext to the upstream. The cert must be provided here because the
-		//    resolved route builds its own TLS config (the original route's config
-		//    is not reused for resolved connections).
+		//    plaintext to the upstream. The cert must be provided here: a resolved
+		//    route takes its TLS config from the resolution, not from the suspended
+		//    route's own cert.
 		proxy.resolveConnection(capturedConn!.id, {
 			upstream: { kind: 'tcp', host: '127.0.0.1', port: echo.port },
 			terminateTls: true,

src/balancer.rs

@@ -178,16 +178,12 @@ impl UdsBalancer {
 		}
 	}
 
-	/// Current active connection count per socket path, for metrics.
-	pub fn connection_counts(&self) -> Vec<(String, u32)> {
-		self.sockets
-			.iter()
-			.map(|s| (s.path.to_string(), s.active.load(Ordering::Relaxed)))
-			.collect()
+	pub fn has_affinity(&self) -> bool {
+		self.affinity.is_some()
 	}
 
-	/// Evict affinity entries older than TTL with no active connections.
-	/// Called by the background eviction task.
+	/// Evict affinity entries older than TTL — the ones `pick()` would no longer honour.
+	/// Called by `router::evict_affinity_periodically`.
 	pub fn evict_affinity(&self) {
 		let Some(aff) = &self.affinity else { return };
 		let now = now_ns();
@@ -197,6 +193,11 @@ impl UdsBalancer {
 		});
 	}
 
+	#[cfg(test)]
+	pub(crate) fn affinity_len(&self) -> usize {
+		self.affinity.as_ref().map_or(0, |aff| aff.entries.len())
+	}
+
 	/// Returns true if any slot has a pid/tid configured for CPU monitoring.
 	pub fn has_monitored_slots(&self) -> bool {
 		self.sockets.iter().any(|s| s.pid.is_some())
@@ -266,3 +267,36 @@ impl Drop for BalancerGuard {
 		self.balancer.decrement(&self.path);
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn evict_affinity_drops_only_entries_past_their_ttl() {
+		let balancer = UdsBalancer::new(
+			vec![UdsSlotSpec {
+				path: "/tmp/symphony-test.sock".to_string(),
+				pid: None,
+				tid: None,
+			}],
+			true,
+			60_000,
+		);
+		let stale: IpAddr = "192.0.2.1".parse().unwrap();
+		let fresh: IpAddr = "192.0.2.2".parse().unwrap();
+		balancer.pick(Some(stale));
+		balancer.pick(Some(fresh));
+		let aff = balancer.affinity.as_ref().unwrap();
+		aff.entries
+			.get(&stale)
+			.unwrap()
+			.last_seen_ns
+			.store(0, Ordering::Relaxed);
+
+		balancer.evict_affinity();
+
+		assert!(!aff.entries.contains_key(&stale));
+		assert!(aff.entries.contains_key(&fresh));
+	}
+}

src/http_proxy.rs

@@ -159,11 +159,6 @@ pub fn is_transfer_encoding_chunked(headers: &[u8]) -> bool {
 		.unwrap_or(false)
 }
 
-/// Return `true` if the `Connection: close` header is present.
-pub fn is_connection_close(headers: &[u8]) -> bool {
-	connection_has_token(headers, "close")
-}
-
 /// Return `true` if a protocol upgrade is requested (e.g. WebSocket).
 /// `Connection` may be split across several fields (`Connection: keep-alive` +
 /// `Connection: Upgrade`) — every field's token list counts.
@@ -172,11 +167,6 @@ pub fn is_upgrade(headers: &[u8]) -> bool {
 		&& connection_has_token(headers, "upgrade")
 }
 
-/// Parse the HTTP response status code (first line: `HTTP/1.x NNN ...`).
-pub fn status_code(headers: &[u8]) -> u16 {
-	parse_status(headers).unwrap_or(200)
-}
-
 /// Status code of a response head, or `None` when the status line is malformed.
 fn parse_status(headers: &[u8]) -> Option<u16> {
 	let line_end = headers.windows(2).position(|w| w == b"\r\n")?;
@@ -254,14 +244,6 @@ pub fn with_connection_close(headers: &[u8]) -> Vec<u8> {
 	rewrite_connection_header(headers, b"Connection: close\r\n")
 }
 
-/// Return a copy of `headers` with the `Connection` field replaced by (or
-/// inserted as) `Connection: keep-alive`.
-///
-/// `headers` must include the trailing `\r\n\r\n`.
-pub fn with_connection_keep_alive(headers: &[u8]) -> Vec<u8> {
-	rewrite_connection_header(headers, b"Connection: keep-alive\r\n")
-}
-
 /// Return a copy of `headers` with any `Content-Length` and `Transfer-Encoding`
 /// fields removed.  Intended for ACME proxy requests which we treat as
 /// body-less GETs — stripping these headers avoids deadlocking the upstream if
@@ -334,22 +316,6 @@ fn rewrite_connection_header(headers: &[u8], replacement: &[u8]) -> Vec<u8> {
 	out
 }
 
-/// Insert `X-Forwarded-For: <ip>` after the first request line.
-pub fn insert_x_forwarded_for(headers: &[u8], peer_ip: std::net::IpAddr) -> Vec<u8> {
-	let xff = format!("X-Forwarded-For: {}\r\n", peer_ip);
-	// End of first line (method SP path SP version \r\n)
-	let insert_at = headers
-		.windows(2)
-		.position(|w| w == b"\r\n")
-		.map(|p| p + 2)
-		.unwrap_or(0);
-	let mut out = Vec::with_capacity(headers.len() + xff.len());
-	out.extend_from_slice(&headers[..insert_at]);
-	out.extend_from_slice(xff.as_bytes());
-	out.extend_from_slice(&headers[insert_at..]);
-	out
-}
-
 // ── Body copying ───────────────────────────────────────────────────────────────
 
 /// Copy exactly `n` bytes from `reader` to `writer`.

src/proxy.rs

@@ -7,8 +7,8 @@ use crate::proxy_conn::{
 	ConnContext, JsEvent, DEFAULT_COPY_BUFFER_SIZE, MAX_COPY_BUFFER_SIZE, MIN_COPY_BUFFER_SIZE,
 };
 use crate::router::{
-	build_route_table, requires_http_protocol, ForwardFingerprint, ListenerTlsSpec, LiveRouteTable,
-	RouteProtocol, RouteSpec, SourceAddressMode, UpstreamSpec,
+	build_route_table, evict_affinity_periodically, requires_http_protocol, ForwardFingerprint,
+	ListenerTlsSpec, LiveRouteTable, RouteProtocol, RouteSpec, SourceAddressMode, UpstreamSpec,
 };
 use crate::suspended::{build_resolved_route, ResolveSpec, ResolveUpstream, SuspendedRegistry};
 use crate::tls::TlsConfigCache;
@@ -338,8 +338,8 @@ pub struct SymphonyProxyWrap {
 	// would serialise all connections onto one OS thread.  By creating our own
 	// multi-thread runtime and using its Handle to spawn, every accept loop and
 	// connection handler gets distributed across the full CPU count.
-	// Handle is Send+Sync; Runtime is Send-only, so it lives in a Mutex.
-	rt: Mutex<Option<Runtime>>,
+	// Never read: owning it is what keeps the runtime alive until the proxy is dropped.
+	_rt: Runtime,
 	rt_handle: RtHandle,
 }
 
@@ -508,14 +508,14 @@ impl SymphonyProxyWrap {
 			keepalive,
 			client_read_buffer_size,
 			upstream_read_buffer_size,
-			route_table: Arc::new(LiveRouteTable(arc_swap::ArcSwap::new(Arc::new(table)))),
+			route_table: LiveRouteTable::new(table),
 			suspended_registry: SuspendedRegistry::new(),
 			global_metrics: Arc::new(GlobalMetrics::default()),
 			listener_states,
 			shutdown_tx: Mutex::new(None),
 			tls_cache: Mutex::new(tls_cache),
 			js_emit: Arc::new(js_emit),
-			rt: Mutex::new(Some(rt)),
+			_rt: rt,
 			rt_handle,
 		})
 	}
@@ -590,6 +590,12 @@ impl SymphonyProxyWrap {
 			}
 		});
 
+		self.rt_handle.spawn(evict_affinity_periodically(
+			self.route_table.clone(),
+			Duration::from_secs(60),
+			tx.subscribe(),
+		));
+
 		// Spawn a periodic IP state eviction task per listener that has protection.
 		// Eviction bounds ip_table memory growth under diverse-IP traffic / attack.
 		// evict() is O(N) DashMap::retain with shard locks and per-entry float math —
@@ -888,7 +894,9 @@ impl SymphonyProxyWrap {
 		let route_result = route.map(|r| {
 			parse_resolve_spec(&r)
 				.map_err(|e| e.reason)
-				.and_then(|spec| build_resolved_route(&spec).map_err(|e| e.to_string()))
+				.and_then(|spec| {
+					build_resolved_route(&spec, &self.tls_cache).map_err(|e| e.to_string())
+				})
 		});
 
 		let resolved = match route_result {