Scale copy-buffer memory with concurrent transfers, not connection count (#37)

CLAUDE.md

@@ -37,8 +37,9 @@ TCP accept (SO_REUSEPORT per worker thread)
   └─ [suspended.rs  register, emit 'suspended', await oneshot]
   └─ tls.rs       TlsAcceptor::accept() with handshake timeout (if terminate_tls)
   └─ upstream.rs  connect(Destination, peer_ip) → UpstreamStream
-  └─ tokio::io::copy_bidirectional_with_sizes wrapped in idle_timeout
-       (per-direction buffers from readBufferSize / client|upstreamReadBufferSize)
+  └─ copy::copy_bidirectional_lazy wrapped in idle_timeout
+       (readBufferSize / client|upstreamReadBufferSize is a per-direction MAXIMUM, not a
+       permanent allocation — see src/copy.rs)
   └─ RAII drop: BalancerGuard, ActiveGuard — all counter decrements happen here
 ```
 
@@ -127,6 +128,67 @@ Each tokio worker thread gets its own listening socket on the same address via `
 ### Suspended connections via oneshot channels
 Each suspended connection gets a `tokio::sync::oneshot::channel`. The sender is stored in a `DashMap<u64, Sender>`. `resolveConnection()` removes the sender and fires it — synchronous from the JS side (no async needed). `oneshot` is used rather than `mpsc` because exactly one resolution is possible per connection. If no resolution arrives within `suspendTimeoutMs`, the `timeout(rx.await)` in `proxy_conn.rs` returns an error and the TCP stream is dropped.
 
+### Escalating copy buffer instead of a permanent per-connection allocation (`src/copy.rs`)
+`tokio::io::copy_bidirectional_with_sizes` allocates its two per-direction `CopyBuffer`s once and
+holds them for the connection's whole life, whether or not it is transferring — at a million
+mostly-idle MQTT subscribers, `readBufferSize × 2` held forever per connection is dead weight.
+`copy_bidirectional_lazy` is a direct port of tokio's own `copy_bidirectional_impl`/`CopyBuffer`
+state machine (`transfer_one_direction` + `TransferState::Running/ShuttingDown/Done`, `poll_fn`
+over both directions), with one addition: `LazyCopyBuffer` starts each direction at a small fixed
+size (`PROBE_BUFFER_SIZE`, 512 B) and escalates straight to the full configured `max_buf_size`
+only once **two consecutive** reads exactly saturate the current buffer — real evidence of a
+sustained burst — dropping straight back to the floor once the direction actually **parks** with
+nothing left to write, not on the first under-capacity read. One full read isn't enough escalation
+evidence: a message that happens to exactly match the current (small) buffer size is a coincidence,
+not a burst, and escalating on it would leave the (now oversized) buffer resident for however long
+the connection then sits idle, however large `max_buf_size` is configured. Requiring a second
+confirming read bounds that coincidence to the floor size at the cost of one extra small-buffer
+round trip on every genuine burst — negligible. Symmetrically, shrinking is deferred to the actual
+park rather than the first under-capacity read: a connection with continuously active but
+variably-sized traffic (never actually idle) would otherwise reallocate on every undersized read
+only to grow right back on the next burst.
+
+**None of it is unconditional.** The saving scales with connection count and the per-burst cost of
+growing/releasing a buffer does not, so the behaviour is gated on the proxy's live
+`GlobalMetrics::active_connections` (`copy::LazyBufferGate`, config `lazyCopyBufferThreshold`,
+default 1000). Below the threshold each direction allocates its full configured buffer once and
+never resizes — byte-for-byte the old `copy_bidirectional_with_sizes` behaviour — so the six-stream
+replication port-set pays nothing for a memory problem it does not have, while the 100k-subscriber
+MQTT port-set is far above the threshold and gets the full mechanism. `0` engages always; a value
+above peak concurrency disables it. The gauge is re-read at **every** resize decision rather than
+latched per connection: a connection established while the proxy was quiet would otherwise hold a
+full-size buffer for its whole life however busy the proxy later became, and long-lived connections
+accumulating while idle is exactly the shape this exists for.
+
+**A recreate does not drop established connections.** `stop()` sends the shutdown broadcast (ending
+the accept loops) and sleeps 100 ms; it never aborts connection tasks, and the runtime lives in the
+napi wrap until GC. So the construction-frozen proxy fields (`readBufferSize`, its two overrides,
+`lazyCopyBufferThreshold`, `workerThreads`) apply to *new* connections only — existing sessions keep
+running on the old values for as long as they stay open. `__test__/server.spec.ts` pins both halves
+of this: that editing one of those fields forces a recreate (the signature does its job, with a
+route-only control proving it is the signature and not "recreates on every write"), and that a held
+connection survives that recreate and still proxies.
+
+Reusing tokio's own poll-based structure (rather than `tokio::io::split` plus independent
+per-direction `async fn`s, tried first) matters for two reasons found by an independent review of
+that version: `split` wraps each side in an `Arc<Mutex<_>>` — two heap allocations and a
+lock/unlock on every read/write/flush/shutdown, *per connection*, which itself scales with
+connection count, exactly what this change exists to avoid — and a hand-rolled `async fn` pump
+that just calls `write_all` does not flush before parking on the next read the way tokio's
+`poll_copy` does. That gap is real: `write_all` only guarantees the data reached the writer's
+internal buffer, not the wire (`tokio-rustls` in particular defers sending encrypted records
+until `poll_flush`) — a TLS client that sends one request and waits for the reply would never see
+it, both sides sitting idle until `idle_timeout` cleaned up the session. A non-blocking
+`poll_read`-with-`Waker::noop()` peek was also tried, to opportunistically drain "whatever's
+already queued" without the extra confirming iteration; under load it silently stranded a
+connection's wakeup (reproduced empirically — an increasing fraction of connections stopped
+responding as concurrency grew) and was dropped in favor of the two-consecutive-full-reads scheme,
+which uses only ordinary polling. `readBufferSize`/`client|upstreamReadBufferSize` are therefore a
+per-transfer *maximum*, not a permanent allocation. Propagating a failed `poll_shutdown()` (not
+swallowing it) is what makes `try_join!`-equivalent short-circuiting work: an error on either
+side — including a shutdown failure — ends the whole copy immediately rather than waiting on the
+other direction.
+
 ### 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))`. Built at config-parse time in `tls.rs::TlsConfigCache`. Important for deployments where many routes share a wildcard cert.
 

README.md

@@ -71,6 +71,7 @@ console.log('proxy listening on :443');
 | `readBufferSize` | `number` | `8192` | Per-direction copy buffer size in bytes, clamped to `[512, 1048576]`. See [Copy buffers and per-connection memory](#copy-buffers-and-per-connection-memory) |
 | `clientReadBufferSize` | `number` | `readBufferSize` | Overrides `readBufferSize` for the client→upstream direction only |
 | `upstreamReadBufferSize` | `number` | `readBufferSize` | Overrides `readBufferSize` for the upstream→client direction only |
+| `lazyCopyBufferThreshold` | `number` | `1000` | Active connections at or above which copy buffers escalate and release instead of being held at full size. `0` always; above peak concurrency disables it. See [Copy buffers and per-connection memory](#copy-buffers-and-per-connection-memory) |
 
 ### `ListenerConfig`
 
@@ -757,35 +758,69 @@ docker run --rm -v $(pwd):/build -w /build \
 
 ## Copy buffers and per-connection memory
 
-Each proxied connection holds two copy buffers — one per direction — for its entire life, whether or
-not it is transferring anything. So buffer memory scales with *connection count*, not with traffic:
-
-```
-buffer bytes = (clientReadBufferSize + upstreamReadBufferSize) × connections
-```
-
-Each unset override falls back to `readBufferSize`, so with symmetric sizing that is just
-`2 × readBufferSize × connections`. At the 8192-byte default it is 16 KiB per connection: 4.0 GiB at
-262k connections, 5.1 GiB at 333k. The asymmetric MQTT setting recommended below is
-`1024 + 4096` = 5 KiB per connection, which is where the general form matters.
-The knob is per proxy, and the right value is the opposite for the two shapes of traffic symphony
-carries:
+`readBufferSize` (and its per-direction overrides) is a *maximum*, not a permanent allocation. Each
+direction starts at a small fixed floor (1 KiB total across both directions) and escalates to the
+configured maximum only once it observes a sustained burst — two consecutive reads that fill the
+current buffer — dropping straight back to the floor once the direction actually parks with nothing
+left to write, not on every single under-capacity read (that would reallocate a connection that is
+still continuously active but simply has variably-sized traffic). So only the memory *above* that
+floor scales with concurrently bursting transfers; the 1 KiB/connection floor itself still scales
+with connection count, same as before. A million idle MQTT subscribers cost about 0.95 GiB in floors
+(1 KiB × 1,000,000 connections), not `readBufferSize × 2 × 1,000,000`.
+
+### When escalation is active: `lazyCopyBufferThreshold`
+
+The escalate/release behaviour above is not unconditional, because its economics are one-sided. The
+memory it saves scales with connection count, while its cost — two allocations plus the zeroing of
+the new buffer, per burst — does not. A replication port-set carrying six bulk streams would pay
+that on every burst to save a few hundred KiB it was never short of; a hundred thousand parked MQTT
+subscribers are the entire reason the mechanism exists.
+
+So it engages only once the proxy is actually carrying enough connections for the saving to be
+worth having. `lazyCopyBufferThreshold` (default `1000`) is the proxy-wide active connection count
+at or above which buffers escalate and release. Below it, each direction gets its full configured
+buffer once and never resizes — identical to symphony's behaviour before this mechanism existed,
+with no resize churn at all.
+
+The default puts the two shapes symphony carries on opposite sides without anyone configuring it: a
+replication port-set (~6 connections) stays static, an MQTT fan-out port-set (100k+) escalates. Set
+it to `0` to engage always, or above the port-set's peak concurrency to disable it entirely. It is
+per proxy, so those two port-sets can differ.
+
+The count is re-read at each resize decision rather than fixed when a connection is established.
+That matters for the case this exists for: connections that arrive while the proxy is quiet would
+otherwise hold a full-size buffer for their entire lives no matter how busy it later got, and
+long-lived connections accumulating while idle is precisely the MQTT shape. Instead they begin
+releasing at their next park once the proxy crosses the threshold.
+
+```
+worst-case buffer bytes = (clientReadBufferSize + upstreamReadBufferSize) × connections bursting right now
+```
+
+That's the ceiling if every connection happened to be mid-burst simultaneously — a useful number to
+size against, but not the steady-state cost, which sits near the 1 KiB/connection floor regardless of
+`readBufferSize`. Each unset override falls back to `readBufferSize`, so with symmetric sizing the
+ceiling is `2 × readBufferSize × connections bursting right now`. At the 8192-byte default that is 16
+KiB per bursting connection. The asymmetric MQTT setting recommended below is `1024 + 4096` = 5 KiB
+per bursting connection, which is where the general form matters.
+The knob still bounds how large a *single* transfer's buffer may grow, and the right bound is the
+opposite for the two shapes of traffic symphony carries:
 
 | Traffic | Connections | Payloads | Suggested |
 |---|---|---|---|
 | Native MQTT (`8883`) | 100k–1M | hundreds of bytes | `clientReadBufferSize: 1024`, `upstreamReadBufferSize: 4096` |
 | HTTPS (`443`) | thousands | mixed | leave at the default |
 | Operations API (`9925`) | tens | can be large | leave at the default |
-| Replication (`9933`) | ~6 | bulk streams | leave, or raise — 64 KiB across 6 connections is 768 KB total |
+| Replication (`9933`) | ~6 | bulk streams | leave, or raise — 64 KiB across 6 connections bursting at once is 768 KB total |
 
 MQTT is worth splitting by direction: after `SUBSCRIBE` a client sends almost nothing but `PINGREQ`,
-while the broker carries the whole fan-out. `1024`/`4096` is 5 KiB per connection against the
-default's 16 KiB — 3.5 GiB saved at 333k connections — and buys more downstream headroom than a
-symmetric 2048 would.
+while the broker carries the whole fan-out. `1024`/`4096` bounds a bursting connection to 5 KiB
+against the default's 16 KiB ceiling, and buys more downstream headroom than a symmetric 2048 would.
 
 Going small costs CPU, not correctness: a payload larger than the buffer is simply copied in more
-iterations. On a TLS-terminating listener those extra iterations are not even syscalls, since the
-reads come out of rustls's already-decrypted buffer.
+iterations, and a connection that keeps bursting re-escalates after two full reads. On a
+TLS-terminating listener those extra iterations are not even syscalls, since the reads come out of
+rustls's already-decrypted buffer.
 
 Two limits on where these settings apply:
 
@@ -796,16 +831,22 @@ Two limits on where these settings apply:
   PROXY-protocol routes, including every UDS route, take the plain path and are governed normally.
 - **A config reload cannot change these.** They are frozen when the proxy is constructed, so
   changing one makes `symphony-server` recreate the proxy rather than hot-swap it. `SO_REUSEPORT`
-  means there is no *bind* gap, but established connections on the old proxy are **not** drained —
-  `stop()` waits 100 ms and connection tasks are detached — so they are all dropped. On a
-  high-connection-count listener, treat a buffer-size edit as a reconnect event, not a live tune.
+  means there is no *bind* gap. Established connections on the old proxy are **not** drained and
+  **not** dropped either: `stop()` sends the shutdown broadcast, which ends the accept loops, then
+  sleeps 100 ms — it never aborts connection tasks, and the tokio runtime lives inside the addon
+  until the old proxy is garbage-collected. Those sessions keep running on the *old* buffer sizes
+  for as long as they stay open. So a buffer-size edit applies to new connections only, and on a
+  listener whose connections are long-lived by design (MQTT subscribers) that can mean the change
+  reaches almost nothing until those clients reconnect. Plan the edit around a reconnect rather
+  than expecting it to take effect fleet-wide on reload.
 
 > **Upgrading:** before this setting was applied to the copy loop, `readBufferSize` had no effect —
-> every connection got 8 KiB per direction regardless of what the config said, and the default
-> documented here was `65536`. A config that leaves it unset is unaffected. A config that *sets* it
-> explicitly now gets what it asked for, so a value copied from the old documented default becomes
-> 128 KiB per connection instead of 16 KiB. Drop or remove such a value before upgrading a
-> high-connection-count deployment.
+> every connection got a fixed 8 KiB per direction regardless of what the config said (held for the
+> connection's whole life, not escalating/shrinking), and the default documented here was `65536`. A
+> config that leaves it unset is unaffected. A config that *sets* it explicitly now gets what it
+> asked for as a per-transfer ceiling, so a value copied from the old documented default raises that
+> ceiling to 128 KiB per bursting connection instead of 16 KiB. Drop or remove such a value before
+> upgrading a high-connection-count deployment.
 
 Two caveats when sizing a node from this: