diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index 7b9038f..fc3f323 100644 --- a/.github/workflows/CI.yml +++ b/.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 diff --git a/CLAUDE.md b/CLAUDE.md index 873f692..ed6df6e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -135,17 +135,18 @@ Each suspended connection gets a `tokio::sync::oneshot::channel`. The sender is ### Per-route Arc deduplication (TlsConfigCache) Routes that share the same cert+mTLS combination share a single `Arc` 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`, `ListenerTlsSpec`, etc.), and never stored in the struct. The struct only holds types that are provably `Send + Sync`. diff --git a/__test__/session-resumption.spec.ts b/__test__/session-resumption.spec.ts index a04ace8..4be831c 100644 --- a/__test__/session-resumption.spec.ts +++ b/__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>; + 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'); + }); + } +}); diff --git a/__test__/suspended.spec.ts b/__test__/suspended.spec.ts index 1723052..ac79dce 100644 --- a/__test__/suspended.spec.ts +++ b/__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, diff --git a/src/balancer.rs b/src/balancer.rs index d5f6924..5b2ad47 100644 --- a/src/balancer.rs +++ b/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)); + } +} diff --git a/src/http_proxy.rs b/src/http_proxy.rs index 0109b54..a85007f 100644 --- a/src/http_proxy.rs +++ b/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 { let line_end = headers.windows(2).position(|w| w == b"\r\n")?; @@ -254,14 +244,6 @@ pub fn with_connection_close(headers: &[u8]) -> Vec { 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 { - 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 { out } -/// Insert `X-Forwarded-For: ` after the first request line. -pub fn insert_x_forwarded_for(headers: &[u8], peer_ip: std::net::IpAddr) -> Vec { - 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`. diff --git a/src/proxy.rs b/src/proxy.rs index 2e75e87..c9b8d29 100644 --- a/src/proxy.rs +++ b/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>, + // 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 { diff --git a/src/router.rs b/src/router.rs index ec393b7..e10f0bf 100644 --- a/src/router.rs +++ b/src/router.rs @@ -8,6 +8,7 @@ use std::net::SocketAddr; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::sync::broadcast; fn now_ns() -> u64 { SystemTime::now() @@ -235,6 +236,8 @@ pub struct RouteTable { /// All UdsBalancers that have at least one pid/tid-configured slot. /// Used by the CPU monitor task spawned in `proxy.rs::start()`. pub monitored_balancers: Vec>, + /// All UdsBalancers with IP affinity, for `evict_affinity_periodically`. + affinity_balancers: Vec>, /// SNIs whose cert failed to build in this table. Carried across a hot-swap so a /// persistently-broken route is logged only on the good→bad transition, not every reconcile. failing_snis: HashSet>, @@ -419,6 +422,7 @@ pub fn build_route_table( let mut exact: HashMap, Route> = HashMap::new(); let mut wildcard: Vec<(Arc, Route)> = Vec::new(); let mut monitored_balancers: Vec> = Vec::new(); + let mut affinity_balancers: Vec> = Vec::new(); let mut failing_snis: HashSet> = HashSet::new(); let mut configured_wildcards = HashSet::with_capacity(specs.len()); let mut dropped_exact: HashSet> = HashSet::new(); @@ -527,12 +531,15 @@ pub fn build_route_table( configured_wildcards.insert(suffix); } - // Collect UdsBalancers that have pid/tid slots for the monitor task. + // Collected from the route actually installed, which may be a carried-forward one. for dest in std::iter::once(&route.destination).chain(route.destination_h2.iter()) { if let Destination::UdsSet(ref bal) = dest { if bal.has_monitored_slots() { monitored_balancers.push(bal.clone()); } + if bal.has_affinity() { + affinity_balancers.push(bal.clone()); + } } } @@ -561,6 +568,7 @@ pub fn build_route_table( default, metric_identities, monitored_balancers, + affinity_balancers, failing_snis, dropped_exact, dropped_wildcard, @@ -791,14 +799,44 @@ impl LiveRouteTable { } } +/// `pick()` ignores an expired affinity entry but only overwrites it when that same IP returns, so +/// without this the maps grow with every distinct client IP until a reload. +/// +/// `DashMap::retain` is O(N) and write-locks each shard while scanning it, so it runs on the +/// blocking pool, as IP-state eviction does; a `pick()` hashing to the shard under scan still +/// waits for that one shard. +pub async fn evict_affinity_periodically( + live: Arc, + interval: Duration, + mut shutdown: broadcast::Receiver<()>, +) { + loop { + tokio::select! { + _ = shutdown.recv() => break, + _ = tokio::time::sleep(interval) => { + let table = live.0.load_full(); + if table.affinity_balancers.is_empty() { + continue; + } + let _ = tokio::task::spawn_blocking(move || { + for balancer in &table.affinity_balancers { + balancer.evict_affinity(); + } + }) + .await; + } + } + } +} + #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; // A self-signed cert (CERT_A) and its matching key (KEY_A), plus an unrelated key // (KEY_B). Pairing CERT_A with KEY_B reproduces the production rustls KeyMismatch a // cert rotation causes (leaf pubkey ≠ private key). - const CERT_A: &[u8] = b"-----BEGIN CERTIFICATE----- + pub(crate) const CERT_A: &[u8] = b"-----BEGIN CERTIFICATE----- MIIDNDCCAhygAwIBAgIUM+1LAIojftQSkEIBoBR0AV87XfowDQYJKoZIhvcNAQEL BQAwGzEZMBcGA1UEAwwQZ29vZC5leGFtcGxlLmNvbTAeFw0yNjA3MDYxNTQ4MzBa Fw0zNjA3MDMxNTQ4MzBaMBsxGTAXBgNVBAMMEGdvb2QuZXhhbXBsZS5jb20wggEi @@ -820,7 +858,7 @@ ZlgCEToIkYUjQVGSygmQqFBbRC5EJAPb+Wpx8N5Y2+g/Q2qy2aPKXhIcOwFSrtbB -----END CERTIFICATE----- "; - const KEY_A: &[u8] = b"-----BEGIN PRIVATE KEY----- + pub(crate) const KEY_A: &[u8] = b"-----BEGIN PRIVATE KEY----- MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCua7KJUHPYvO/P aqLDMrHGlEddpxkFGMifO87nj9QRnIpHcz+nWrdvH57QpBRdojBC/j9L2/ybaRVG M52OO5fJm1DH4veD9axofkOGWBp1yPqDlxe0g/wlreWtAAMRVqGODw/OOvcDwnok @@ -850,7 +888,7 @@ UKlOCXtHXb1XskMBV7W29w== -----END PRIVATE KEY----- "; - const KEY_B: &[u8] = b"-----BEGIN PRIVATE KEY----- + pub(crate) const KEY_B: &[u8] = b"-----BEGIN PRIVATE KEY----- MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQCWxNE5Z315MdlX Za+MKSJrdcJlm5zHfsBZ642On9Cc+oMe+U5+91RMUhiNt524CNvQyDMTntiD9wn7 zUaeQGZjP+wtUDcIG1S1EIXhoAWLWd+Jww/3WAYNxnauIW3QawlqH/aiJVxTgWB2 @@ -880,7 +918,7 @@ UlqL1DcgX6Szi9w/p7B4BZO9iA== -----END PRIVATE KEY----- "; - fn tls_route(sni: &str, cert: &[u8], key: &[u8]) -> RouteSpec { + pub(crate) fn tls_route(sni: &str, cert: &[u8], key: &[u8]) -> RouteSpec { RouteSpec { sni: sni.to_string(), metrics_group: String::new(), @@ -1372,4 +1410,113 @@ UlqL1DcgX6Szi9w/p7B4BZO9iA== ForwardFingerprint::NONE )); } + + fn uds_upstream(path: &str, protocol: Option<&str>, ip_affinity: bool) -> UpstreamSpec { + UpstreamSpec::Uds { + paths: vec![path.to_string()], + pids: vec![None], + tids: vec![None], + ip_affinity, + affinity_ttl_ms: 0, + protocol: protocol.map(str::to_string), + } + } + + /// A terminating route whose UDS upstreams include an h2 mirror, so it gets two balancers. + fn affinity_route(key: &[u8], ip_affinity: bool) -> RouteSpec { + let mut spec = tls_route("tenant.example.com", CERT_A, key); + spec.http2 = true; + spec.upstreams = vec![ + uds_upstream("/tmp/symphony-test.sock", None, ip_affinity), + uds_upstream("/tmp/symphony-test-h2.sock", Some("h2"), ip_affinity), + ]; + spec + } + + #[test] + fn affinity_balancers_come_from_the_installed_route() { + let mut cache = TlsConfigCache::new(); + let no_affinity = build_route_table( + &[affinity_route(KEY_A, false)], + &ListenerTlsSpec::empty(), + None, + &mut cache, + ) + .expect("build"); + assert!(no_affinity.affinity_balancers.is_empty()); + + let live = build_route_table( + &[affinity_route(KEY_A, true)], + &ListenerTlsSpec::empty(), + Some(&no_affinity), + &mut cache, + ) + .expect("build"); + assert_eq!( + live.affinity_balancers.len(), + 2, + "the h2 mirror's balancer carries its own affinity map" + ); + + // A cert mismatch carries the live route forward — affinity and all — even though the + // rejected spec turned affinity off. Its maps keep growing, so they must stay collected. + let carried = build_route_table( + &[affinity_route(KEY_B, false)], + &ListenerTlsSpec::empty(), + Some(&live), + &mut cache, + ) + .expect("carry forward"); + assert_eq!(carried.affinity_balancers.len(), 2); + assert!(carried + .affinity_balancers + .iter() + .zip(&live.affinity_balancers) + .all(|(a, b)| Arc::ptr_eq(a, b))); + } + + #[tokio::test] + async fn affinity_eviction_follows_the_live_table_and_stops_on_shutdown() { + let mut cache = TlsConfigCache::new(); + let live = LiveRouteTable::new( + build_route_table(&[], &ListenerTlsSpec::empty(), None, &mut cache).expect("build"), + ); + let (shutdown_tx, shutdown_rx) = broadcast::channel(1); + let task = tokio::spawn(evict_affinity_periodically( + live.clone(), + Duration::from_millis(5), + shutdown_rx, + )); + + // Affinity arrives with a reload after the task started, so it has to re-read the table. + live.swap( + build_route_table( + &[affinity_route(KEY_A, true)], + &ListenerTlsSpec::empty(), + None, + &mut cache, + ) + .expect("build"), + ); + let balancers = live.0.load().affinity_balancers.clone(); + for balancer in &balancers { + balancer.pick(Some("192.0.2.7".parse().unwrap())); + assert_eq!(balancer.affinity_len(), 1); + } + + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while balancers.iter().any(|b| b.affinity_len() > 0) { + assert!( + tokio::time::Instant::now() < deadline, + "expired affinity entries were never reclaimed" + ); + tokio::time::sleep(Duration::from_millis(5)).await; + } + + shutdown_tx.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("the task must exit on shutdown") + .unwrap(); + } } diff --git a/src/suspended.rs b/src/suspended.rs index 63a56a1..c09c548 100644 --- a/src/suspended.rs +++ b/src/suspended.rs @@ -1,9 +1,10 @@ use crate::balancer::{UdsBalancer, UdsSlotSpec}; use crate::router::{Destination, ForwardFingerprint, RouteProtocol, SourceAddressMode}; +use crate::tls::{CertSpec, MtlsSpec, TlsConfigCache}; use dashmap::DashMap; use rustls::ServerConfig; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use tokio::sync::oneshot; /// The resolved route sent back from JS via `resolveConnection()`. @@ -60,11 +61,6 @@ impl SuspendedRegistry { pub fn contains(&self, id: u64) -> bool { self.pending.contains_key(&id) } - - /// Number of currently pending suspended connections. - pub fn pending_count(&self) -> u64 { - self.pending.len() as u64 - } } // ── JS-side resolver spec ───────────────────────────────────────────────────── @@ -94,10 +90,13 @@ pub enum ResolveUpstream { }, } -/// Build a `ResolvedRoute` from a `ResolveSpec`. -pub fn build_resolved_route(spec: &ResolveSpec) -> crate::error::Result { - use crate::tls::{CertSpec, MtlsSpec, TlsConfigCache}; - +/// Build a `ResolvedRoute` from a `ResolveSpec`. `tls_cache` is the proxy's own, so every +/// resolution of one cert gets the same `ServerConfig` — and with it the session store and ticket +/// keys a returning client needs to resume. +pub fn build_resolved_route( + spec: &ResolveSpec, + tls_cache: &Mutex, +) -> crate::error::Result { let tls_config = if spec.terminate_tls { let cert_pem = spec.cert_pem.as_deref().ok_or_else(|| { crate::error::SymphonyError::Config( @@ -119,8 +118,8 @@ pub fn build_resolved_route(spec: &ResolveSpec) -> crate::error::Result crate::error::Result ResolveSpec { + ResolveSpec { + upstream: ResolveUpstream::Tcp("127.0.0.1:9".parse().unwrap()), + terminate_tls: true, + cert_pem: Some(CERT_A.to_vec()), + key_pem: Some(key.to_vec()), + mtls_ca_pem: None, + require_client_cert: false, + source_address_mode: SourceAddressMode::None, + forward_fingerprint: ForwardFingerprint::NONE, + http2: false, + protocol: RouteProtocol::Opaque, + } + } + + fn tls_config(spec: &ResolveSpec, cache: &Mutex) -> Arc { + build_resolved_route(spec, cache) + .expect("resolve") + .tls_config + .expect("terminating route") + } + + /// A committed route-table reload, as `update_config` runs it. + fn reload(cache: &Mutex, specs: &[RouteSpec]) -> crate::router::RouteTable { + let mut cache = cache.lock().unwrap(); + let table = + build_route_table(specs, &ListenerTlsSpec::empty(), None, &mut cache).expect("build"); + cache.retain_used(); + table + } + + #[test] + fn resolutions_of_one_cert_share_a_server_config() { + let cache = Mutex::new(TlsConfigCache::new()); + let spec = resolve_spec(KEY_A); + assert!(Arc::ptr_eq( + &tls_config(&spec, &cache), + &tls_config(&spec, &cache) + )); + } + + // One config means one ticketer, so anything that changes what a resumed session may skip + // (client auth, ALPN) must key a different one. + #[test] + fn tls_policy_keys_a_separate_server_config() { + let cache = Mutex::new(TlsConfigCache::new()); + let plain = resolve_spec(KEY_A); + let mut mtls_required = resolve_spec(KEY_A); + mtls_required.mtls_ca_pem = Some(CERT_A.to_vec()); + mtls_required.require_client_cert = true; + let mut mtls_optional = resolve_spec(KEY_A); + mtls_optional.mtls_ca_pem = Some(CERT_A.to_vec()); + let mut h2 = resolve_spec(KEY_A); + h2.http2 = true; + + let configs: Vec<_> = [&plain, &mtls_required, &mtls_optional, &h2] + .into_iter() + .map(|spec| tls_config(spec, &cache)) + .collect(); + for (i, a) in configs.iter().enumerate() { + for b in &configs[i + 1..] { + assert!(!Arc::ptr_eq(a, b)); + } + } + } + + #[test] + fn a_resolution_shares_a_table_route_config_for_the_same_cert() { + let cache = Mutex::new(TlsConfigCache::new()); + let route = crate::router::tests::tls_route("tenant.example.com", CERT_A, KEY_A); + let table = reload(&cache, &[route]); + let table_config = table + .resolve(Some("tenant.example.com")) + .and_then(|r| r.tls_config.clone()) + .expect("route config"); + assert!(Arc::ptr_eq( + &table_config, + &tls_config(&resolve_spec(KEY_A), &cache) + )); + } + + #[test] + fn a_held_resolved_config_survives_a_reload_and_an_idle_one_does_not() { + let cache = Mutex::new(TlsConfigCache::new()); + let spec = resolve_spec(KEY_A); + + let held = tls_config(&spec, &cache); + reload(&cache, &[]); + assert!( + Arc::ptr_eq(&held, &tls_config(&spec, &cache)), + "a config a live connection holds must outlive the sweep" + ); + + drop(held); + reload(&cache, &[]); + assert!( + cache.lock().unwrap().is_empty(), + "no table or connection references it any more" + ); + } + + #[test] + fn a_failed_resolution_leaves_the_cache_usable() { + let cache = Mutex::new(TlsConfigCache::new()); + assert!(build_resolved_route(&resolve_spec(KEY_B), &cache).is_err()); + let first = tls_config(&resolve_spec(KEY_A), &cache); + assert!(Arc::ptr_eq( + &first, + &tls_config(&resolve_spec(KEY_A), &cache) + )); + assert_eq!(cache.lock().unwrap().len(), 1); + } +} diff --git a/src/tls.rs b/src/tls.rs index 7966068..dad5a07 100644 --- a/src/tls.rs +++ b/src/tls.rs @@ -33,17 +33,29 @@ type CacheKey = ([u8; 32], [u8; 32], bool); /// renewal rebuilds the whole table — so a per-build cache means clients almost never get to /// resume. Keying on the cert bytes gives exactly the right lifetime: session state survives as /// long as the cert it was issued under, and rotating a cert retires it. +/// +/// `resolveConnection` draws from the same cache. No route table holds those configs, so they +/// live by refcount alone: kept while a connection holds one, dropped by the next build's +/// `clear_used` or by an insert past `evict_at` — whichever comes first. pub struct TlsConfigCache { cache: HashMap>, /// Keys touched since the last `retain_used()` — the mark half of mark-and-sweep. used: HashSet, + evict_at: usize, } +/// Floor for `evict_at`: how many configs nothing references (idle resolved-route certs) a proxy +/// that never reloads its route table keeps for resumption. Each one carries a session store +/// preallocated for 1024 entries, so without a bound they accumulate with every distinct cert +/// ever resolved. +const MIN_EVICT_AT: usize = 256; + impl TlsConfigCache { pub fn new() -> Self { Self { cache: HashMap::new(), used: HashSet::new(), + evict_at: MIN_EVICT_AT, } } @@ -66,7 +78,18 @@ impl TlsConfigCache { /// without the retain, repeatedly-failing reloads accumulate a config per attempt. pub(crate) fn clear_used(&mut self) { self.used.clear(); + self.evict_unreferenced(); + } + + /// Drop every config only the cache holds. A route table or a live connection holding one + /// keeps it, so the cost is one full handshake per returning client of an idle resolved route. + fn evict_unreferenced(&mut self) { self.cache.retain(|_, config| Arc::strong_count(config) > 1); + // Doubling keeps the scan amortised O(1) per insert even when referenced configs alone + // exceed the floor; shrinking keeps it so after a large referenced set drains, since + // `retain` walks the allocation, not just the live entries. + self.evict_at = MIN_EVICT_AT.max(self.cache.len() * 2); + self.cache.shrink_to(self.evict_at); } #[cfg(test)] @@ -79,12 +102,36 @@ impl TlsConfigCache { self.cache.is_empty() } + /// For a route-table build: also marks the key as requested by the table being built. pub fn get_or_build( &mut self, cert: &CertSpec, mtls: Option<&MtlsSpec>, http2: bool, ) -> Result> { + let (cache_key, cfg) = self.lookup_or_build(cert, mtls, http2)?; + // A hit is exactly as much "still in use" as a miss. + self.used.insert(cache_key); + Ok(cfg) + } + + /// For `resolveConnection`: no table will hold the result, so marking it would only add a key + /// nothing reads before the next build clears it. + pub fn get_or_build_unmarked( + &mut self, + cert: &CertSpec, + mtls: Option<&MtlsSpec>, + http2: bool, + ) -> Result> { + self.lookup_or_build(cert, mtls, http2).map(|(_, cfg)| cfg) + } + + fn lookup_or_build( + &mut self, + cert: &CertSpec, + mtls: Option<&MtlsSpec>, + http2: bool, + ) -> Result<(CacheKey, Arc)> { // Hash both chain and private key so routes sharing a cert but using different // keys (e.g. mid-rotation) get distinct ServerConfig allocations. let cert_key = sha256( @@ -104,15 +151,15 @@ impl TlsConfigCache { let cache_key = (cert_key, mtls_key, http2); if let Some(cfg) = self.cache.get(&cache_key) { - // Mark before returning: a hit is exactly as much "still in use" as a miss. - self.used.insert(cache_key); - return Ok(cfg.clone()); + return Ok((cache_key, cfg.clone())); } let cfg = build_server_config(cert, mtls, http2)?; self.cache.insert(cache_key, cfg.clone()); - self.used.insert(cache_key); - Ok(cfg) + if self.cache.len() > self.evict_at { + self.evict_unreferenced(); + } + Ok((cache_key, cfg)) } } @@ -189,3 +236,57 @@ fn sha256(data: &[u8]) -> [u8; 32] { out.copy_from_slice(digest.as_ref()); out } + +#[cfg(test)] +mod tests { + use super::*; + use crate::router::tests::{CERT_A, KEY_A}; + + /// The same cert with a distinct comment line: PEM parsing ignores it, the cache key doesn't. + fn distinct_cert(n: usize) -> CertSpec { + let mut chain = CERT_A.to_vec(); + chain.extend_from_slice(format!("# {n}\n").as_bytes()); + CertSpec { + cert_chain_pem: chain, + private_key_pem: KEY_A.to_vec().into(), + } + } + + // Without a reload nothing else evicts, so every distinct cert ever resolved would stay. + #[test] + fn unreferenced_configs_are_bounded_without_a_reload() { + let mut cache = TlsConfigCache::new(); + let held: Vec<_> = (0..4) + .map(|n| { + cache + .get_or_build_unmarked(&distinct_cert(n), None, false) + .unwrap() + }) + .collect(); + + for n in 4..MIN_EVICT_AT * 3 { + drop( + cache + .get_or_build_unmarked(&distinct_cert(n), None, false) + .unwrap(), + ); + assert!(cache.len() <= MIN_EVICT_AT + 1); + } + assert!( + cache.used.is_empty(), + "resolutions must not leave marks behind" + ); + + for (n, config) in held.iter().enumerate() { + assert!( + Arc::ptr_eq( + config, + &cache + .get_or_build_unmarked(&distinct_cert(n), None, false) + .unwrap() + ), + "a referenced config must survive pressure eviction" + ); + } + } +}