Merge pull request #2767 from vitorpamplona/claude/connection-scaling-plan-YVjc8

Scale relay to 10k+ concurrent connections with streaming JSON parsing

geode/config.example.toml

@@ -25,6 +25,20 @@ contact = "admin@example.com"
 host = "0.0.0.0"
 port = 7447
 path = "/"
+# Ktor CIO event-loop pool sizing. Leave commented-out for sensible
+# per-CPU defaults (typical for <2k concurrent connections). Lift on
+# big-VM deployments targeting 10k+ connections — over-threading at
+# low connection counts hurts L1/L2 cache locality, so always
+# benchmark before/after when tuning these.
+#
+# Operators targeting >1k concurrent WebSockets should also raise the
+# OS file-descriptor limit: `ulimit -n 65536` (or higher) before
+# launching, plus a matching `LimitNOFILE=` in any systemd unit. The
+# default of 1024 on most distros caps the relay well below 1k FDs
+# (one per WS plus DB and listening sockets).
+# connection_group_size = 4
+# worker_group_size = 16
+# call_group_size = 64
 
 [database]
 # True keeps an in-memory SQLite db (events vanish on restart). Useful

geode/plans/2026-05-07-connection-scaling.md

@@ -1,5 +1,14 @@
 # Connection scaling: pushing past 2 000
 
+> **Status (2026-05-07):** Sketches A and B shipped on
+> `claude/connection-scaling-plan-YVjc8`. Sketch C landed as a smaller
+> slice in Quartz — the streaming-filter cut — once the audit showed
+> the rest of the plan's premise was overstated. Verification
+> benchmarks (`connectionsHeldOpen10k`, `connectionsHeldOpenWithFanout`)
+> are wired up but only run under `-DrunLoadBenchmark=true`. Remaining
+> open work, including fan-out de-duplication, is now tracked in
+> [`live-broadcast-fanout-index.md`](./2026-05-07-live-broadcast-fanout-index.md).
+
 ## Problem
 
 Current measurement (`LoadBenchmark.connectionsHeldOpen`): **~2 000
@@ -25,69 +34,169 @@ the channel array, even though most connections never fan out.
 
 ## Sketch
 
-### A — adaptive outQueue capacity
+### A — adaptive outQueue capacity ✅ shipped
 
-Start every connection with `INITIAL_OUTGOING_BUFFER = 64`. When the
+> Original plan: start at `INITIAL_OUTGOING_BUFFER = 64` and swap to a
-producer side trySends and we observe queue depth crossing a high-water
+> wider channel under a per-session lock when a high-water mark
-mark (e.g. 75% full), grow the channel up to `MAX_OUTGOING_BUFFER =
+> trips. **Not how it shipped.**
-8192`. This is not how `kotlinx.coroutines.channels.Channel` is
-structured (capacity is fixed at construction), so the implementation
-is "swap in a wider channel under a per-session lock when watermark
-trips" — drains the old, then routes new sends through the new.
 
-Expected: 90% of connections never fan out, so they stay at 64 slots
+What actually shipped is the simpler alternative the original Risks
-× ~512 B per ref ≈ 32 KB. At 5 000 conns that's ~160 MB → ~5 MB.
+section called out: `Channel.UNLIMITED` plus an `AtomicInteger`
-Hot-fanout connections still get the 2 MB cap.
+backlog cap. kotlinx.coroutines' `BufferedChannel` allocates segments
-
+lazily, so an unlimited channel pays only the small head-segment cost
-### B — per-relay event-loop pool sizing
+on idle connections — there is no preallocated buffer to scale.
-
-Ktor CIO defaults to one event-loop thread per available CPU.
-Beyond a few thousand connections, this becomes the bottleneck — and
-none of geode's per-connection work is CPU-bound (it's mostly waiting
-on incoming frames). Tune CIO via:
 
 ```kotlin
-embeddedServer(CIO, ...) {
+private val outQueue = Channel<String>(capacity = Channel.UNLIMITED)
-    connectionGroupSize = max(2, Runtime.getRuntime().availableProcessors() / 2)
+private val outstanding = AtomicInteger(0)
-    workerGroupSize    = max(4, Runtime.getRuntime().availableProcessors())
+
-    callGroupSize      = max(8, Runtime.getRuntime().availableProcessors() * 4)
+// producer side
+val depth = outstanding.incrementAndGet()
+if (depth > MAX_OUTGOING_BUFFER) {            // 8192
+    outstanding.decrementAndGet()
+    droppedForBackpressure = true
+    outQueue.close()                          // NIP-01: drop the conn
+    return@connect
+}
+val res = outQueue.trySend(json)
+if (!res.isSuccess) outstanding.decrementAndGet()  // closed concurrently
+
+// writer side
+for (json in outQueue) {
+    ws.outgoing.send(Frame.Text(json))
+    outstanding.decrementAndGet()
 }
 ```
 
-Expose these through `RelayConfig.NetworkSection` so an operator on a
+Memory characteristic the plan asked for is intact: idle connections
-big VM can lift them.
+no longer reserve an 8 192-slot fixed buffer; hot fan-out connections
+still get bounded at the same 2 MiB cap before the slow-client cutoff
+fires. NIP-01 ordering is preserved (no silent drop — connection is
+killed at the cap).
 
-### C — reduce per-message JSON allocations
+Implementation: `geode/.../server/WebSocketSessionPump.kt`. The
+channel-swap approach was rejected because
+`Channel.UNLIMITED` already gives the lazy-allocation behavior the
+swap was simulating, with none of the swap's race surface.
 
-`OptimizedJsonMapper.fromJsonToCommand` allocates a `JsonNode` tree per
+### B — per-relay event-loop pool sizing ✅ shipped
-incoming frame. At 10k connections with 1 msg/s each that's 10k tree
-allocations/sec. Investigate streaming Jackson + reusing `ObjectMapper`
-per session, or using kotlinx-serialization's lower-overhead path.
 
-This is more of a quartz-level change than geode-specific, but
+Three optional knobs added to `[network]` in `RelayConfig`:
-geode's load benchmark is the right place to measure it.
 
-## How to verify
+```toml
+[network]
+host = "0.0.0.0"
+port = 7447
+path = "/"
+# connection_group_size = 4
+# worker_group_size = 16
+# call_group_size = 64
+```
 
-Add to `geode.perf.LoadBenchmark`:
+Default is **`null` (Ktor default)** — no behavior change unless an
+operator explicitly tunes them. The values are wired through
+`LocalRelayServer` into the new `embeddedServer(factory = CIO,
+rootConfig = serverConfig {…}, configure = {…})` overload (the
+short-form `embeddedServer(factory, host, port) {…}` overload doesn't
+expose CIO config). The auto-connector that the short form created
+now has to be added explicitly via `connector { host = …; port = …}`.
 
-- `connectionsHeldOpen10k` — opens 10 000 idle WebSocket connections;
+`config.example.toml` documents the knobs and includes the operator
-  asserts no FD exhaustion + RSS stays under 1 GB.
+note that targeting >1k connections needs `ulimit -n 65536` (or
-- `connectionsHeldOpenWithFanout` — 5 000 idle subscribers,
+matching `LimitNOFILE=` in a systemd unit).
-  10 EPS published; measures p99 fanout latency at scale.
 
-The current `connectionsHeldOpen` benchmark stays as the baseline
+### C — reduce per-message JSON allocations ✅ partially shipped (in Quartz)
-floor (~2 000 conns).
 
-## Risks
+> Original plan claim: "`OptimizedJsonMapper.fromJsonToCommand`
+> allocates a `JsonNode` tree per incoming frame." **Overstated.**
 
-- **Adaptive channel swap is fiddly**: drains under the producer's nose
+Audit of `quartz/.../jackson` showed the Command/Message envelope is
-  must preserve OK ordering. A simpler alternative: keep capacity fixed,
+already streaming:
-  but lazily allocate a small `ArrayDeque<String>` only when the first
+
-  message is sent. Channels in kotlinx.coroutines do allocate up-front.
+| Path                    | Already streaming? | Tree alloc?                                        |
-- **Bumping CIO group sizes can hurt**: more threads can mean worse
+| ----------------------- | ------------------ | -------------------------------------------------- |
-  L1/L2 locality. Always benchmark before/after, don't trust
+| `MessageDeserializer`   | yes                | only for `COUNT` result (rare)                     |
-  intuitive sizing.
+| `CommandDeserializer`   | yes                | only for **filter sub-objects** in REQ/COUNT/NEG-OPEN |
-- **OS-level FD limit**: per-process FD limit on Linux defaults to
+| `EventDeserializer`     | yes                | none — `currentName().hashCode()` dispatch         |
-  1024 in many environments. Document the `ulimit -n` requirement
+| `ManualFilterDeserializer` | **no**          | `jp.codec.readTree(jp)` per filter                 |
-  for operators targeting >1k connections.
+
+So the only relay-inbound tree allocation worth chasing was filter
+parsing — the bulk of the per-frame allocations on a REQ-heavy
+relay.
+
+What shipped: a streaming `ManualFilterDeserializer.fromJson(jp:
+JsonParser)` modeled exactly on `EventDeserializer`. Token-loop with
+field-name dispatch (`ids` / `authors` / `kinds` / `since` / `until` /
+`limit` / `search`, plus dynamic `#x` / `&x` tag keys), and
+`readStringArray` / `readIntArray` helpers that drop invalid entries
+silently to match the tree path's `mapNotNull { asTextOrNull() }`
+tolerance. Wired into all four internal call sites:
+`FilterDeserializer.deserialize` and the three `CommandDeserializer`
+paths (REQ, COUNT, NEG-OPEN).
+
+The tree-based `fromJson(ObjectNode)` overload is retained for
+external/cross-format adapters (Quartz is a published library).
+
+What was NOT done — and why:
+- **Streaming Jackson for the Command envelope**: already streaming.
+  No allocation to remove.
+- **kotlinx-serialization for the inbound path**: not pursued. The
+  cross-mapper round-trip tests in `KotlinSerializationMapperTest`
+  show the two formats are interchangeable, but the engine swap is a
+  much larger lift than the filter cut and there's no evidence the
+  KS path is faster on this code shape.
+- **Per-session `ObjectMapper`**: Jackson's `ObjectMapper` is
+  thread-safe and stateless — sharing one is the recommended pattern.
+  Per-session would *increase* allocation, not decrease it.
+
+## How to verify ✅ shipped
+
+Two new benchmarks in `geode.perf.LoadBenchmark`, gated behind
+`-DrunLoadBenchmark=true`:
+
+- **`connectionsHeldOpen10k`** — opens 10 000 idle WebSocket
+  connections, asserts every one settles to EOSE inside 120 s, and
+  measures retained JVM heap (after `System.gc()` + 200 ms settle)
+  with a 1 GiB ceiling assertion. Requires `ulimit -n 32768` on
+  Linux.
+- **`connectionsHeldOpenWithFanout`** — 5 000 subscribers all
+  matching `kinds:[1]`, one publisher emitting `targetEps × duration`
+  events, prints p50 / p99 last-fanout latency. No assertion on
+  latency — just regression-detection via stdout logging.
+
+The original `connectionsHeldOpen` benchmark stays as the **baseline
+floor (~2 000 conns)** for before/after comparisons.
+
+Note on heap-vs-RSS: the original plan said "RSS stays under 1 GB"
+but the JVM can only measure heap from inside; `Runtime.totalMemory
+- freeMemory` is what the benchmark asserts on. RSS will be higher
+because of code, native buffers, off-heap (Ktor CIO), etc.
+
+## Risks (post-implementation)
+
+- ~~**Adaptive channel swap is fiddly**~~ — sidestepped by using
+  `Channel.UNLIMITED` instead of swapping bounded channels.
+- **Bumping CIO group sizes can hurt** — kept the defaults `null`.
+  Operators must opt in, and the docstrings explicitly say to
+  benchmark before/after.
+- **OS-level FD limit** — documented in `config.example.toml` next to
+  the CIO knobs. Test prereq is also documented in the benchmark
+  KDoc.
+
+## Open work
+
+- **Fan-out de-duplication** — when one EVENT matches N subscribers,
+  we currently re-serialize and copy the JSON N times into N
+  channels. Caching one pre-serialized payload per event and
+  broadcasting a shared reference is a much bigger win than anything
+  in this plan; tracked in
+  [`live-broadcast-fanout-index.md`](./2026-05-07-live-broadcast-fanout-index.md).
+- **Filter-matching index** — same plan. At 10k conns × ~5 filters
+  that's 50k evaluations per published EVENT, almost all of which
+  could be culled by indexing subscriptions on `kinds` / `authors` /
+  `#e` / `#p`.
+- **Netty engine evaluation** — Ktor's Netty engine handles many idle
+  connections with measurably lower per-connection overhead than
+  CIO. Not pursued here because it changes the transport layer
+  wholesale; revisit only if the CIO knobs in (B) prove insufficient
+  for an operator at 20k+ connections.

geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt

@@ -32,8 +32,10 @@ import io.ktor.http.ContentType
 import io.ktor.http.HttpHeaders
 import io.ktor.http.HttpStatusCode
 import io.ktor.server.application.install
+import io.ktor.server.application.serverConfig
 import io.ktor.server.cio.CIO
 import io.ktor.server.cio.CIOApplicationEngine
+import io.ktor.server.engine.connector
 import io.ktor.server.engine.embeddedServer
 import io.ktor.server.request.header
 import io.ktor.server.response.respondText
@@ -109,6 +111,16 @@ class LocalRelayServer(
      * RPC payload.
      */
     val maxAdminBodyBytes: Int = 1 shl 20,
+    /**
+     * Ktor CIO acceptor-thread count. `null` keeps Ktor's default.
+     * Lift on machines with many cores when targeting 10k+
+     * concurrent connections — see `[network]` config docs.
+     */
+    val connectionGroupSize: Int? = null,
+    /** Ktor CIO worker-thread count. `null` keeps Ktor's default. */
+    val workerGroupSize: Int? = null,
+    /** Ktor CIO call-handling thread count. `null` keeps Ktor's default. */
+    val callGroupSize: Int? = null,
 ) {
     private val infoHolder =
         object : Nip86Server.InfoHolder {
@@ -167,51 +179,78 @@ class LocalRelayServer(
      * [url] is safe to read on the very next line.
      */
     fun start(): LocalRelayServer {
+        // Snapshot the constructor-supplied overrides into locals so
+        // the `configure` lambda below can assign to its receiver
+        // without the names colliding with outer properties.
+        val connGrp = connectionGroupSize
+        val workGrp = workerGroupSize
+        val callGrp = callGroupSize
+        val bindHost = host
+        val bindPort = port
         val server =
-            embeddedServer(CIO, host = host, port = port) {
+            embeddedServer(
-                install(WebSockets) {
+                factory = CIO,
-                    maxFrameBytes?.let { maxFrameSize = it }
+                rootConfig =
-                }
+                    serverConfig {
-                routing {
+                        module {
-                    // NIP-11: GET on the relay URL with Accept:
+                            install(WebSockets) {
-                    // application/nostr+json returns the relay info doc.
+                                maxFrameBytes?.let { maxFrameSize = it }
-                    // We mount this *before* the webSocket route so Ktor
+                            }
-                    // serves NIP-11 for plain HTTP GETs and only upgrades
+                            routing {
-                    // to a WebSocket when the request is a WS upgrade.
+                                // NIP-11: GET on the relay URL with Accept:
-                    get(path) {
+                                // application/nostr+json returns the relay info doc.
-                        val accept = call.request.header(HttpHeaders.Accept).orEmpty()
+                                // We mount this *before* the webSocket route so Ktor
-                        if (accept.contains("application/nostr+json")) {
+                                // serves NIP-11 for plain HTTP GETs and only upgrades
-                            call.response.headers.append("Access-Control-Allow-Origin", "*")
+                                // to a WebSocket when the request is a WS upgrade.
-                            call.respondText(
+                                get(path) {
-                                relay.info.json,
+                                    val accept = call.request.header(HttpHeaders.Accept).orEmpty()
-                                ContentType.parse("application/nostr+json"),
+                                    if (accept.contains("application/nostr+json")) {
-                            )
+                                        call.response.headers.append("Access-Control-Allow-Origin", "*")
-                        } else {
+                                        call.respondText(
-                            call.respondText(
+                                            relay.info.json,
-                                "Use a Nostr client (NIP-01 WebSocket) or send Accept: application/nostr+json (NIP-11).",
+                                            ContentType.parse("application/nostr+json"),
-                                ContentType.Text.Plain,
+                                        )
-                                HttpStatusCode.UpgradeRequired,
+                                    } else {
-                            )
+                                        call.respondText(
+                                            "Use a Nostr client (NIP-01 WebSocket) or send Accept: application/nostr+json (NIP-11).",
+                                            ContentType.Text.Plain,
+                                            HttpStatusCode.UpgradeRequired,
+                                        )
+                                    }
+                                }
+                                // NIP-86: POST application/nostr+json+rpc with a NIP-98
+                                // signed Authorization header → JSON-RPC dispatch.
+                                post(path) {
+                                    nip86Route.handle(call)
+                                }
+                                webSocket(path) {
+                                    if (shuttingDown) {
+                                        // Just return — Ktor closes the WS for us.
+                                        return@webSocket
+                                    }
+                                    WebSocketSessionPump(this).pump(
+                                        server = relay.server,
+                                        registerSession = activeSessions::add,
+                                        unregisterSession = activeSessions::remove,
+                                    )
+                                }
+                            }
                         }
+                    },
+                configure = {
+                    connector {
+                        host = bindHost
+                        port = bindPort
                     }
-                    // NIP-86: POST application/nostr+json+rpc with a NIP-98
+                    // Keep Ktor defaults unless the operator overrode
-                    // signed Authorization header → JSON-RPC dispatch.
+                    // them — Ktor's per-CPU sizing is sensible for
-                    post(path) {
+                    // most deployments, and over-threading hurts L1/L2
-                        nip86Route.handle(call)
+                    // locality at low connection counts.
-                    }
+                    connGrp?.let { connectionGroupSize = it }
-                    webSocket(path) {
+                    workGrp?.let { workerGroupSize = it }
-                        if (shuttingDown) {
+                    callGrp?.let { callGroupSize = it }
-                            // Just return — Ktor closes the WS for us.
+                },
-                            return@webSocket
+            )
-                        }
-                        WebSocketSessionPump(this).pump(
-                            server = relay.server,
-                            registerSession = activeSessions::add,
-                            unregisterSession = activeSessions::remove,
-                        )
-                    }
-                }
-            }
         server.start(wait = false)
         engine = server.engine
         // Ktor 3.x made resolvedConnectors() suspend. We block here so

geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt

@@ -117,6 +117,9 @@ fun main(args: Array<String>) {
             maxFrameBytes = frameLimit,
             adminPubkeys = config.admin.pubkeys.toSet(),
             publicUrl = config.admin.public_url,
+            connectionGroupSize = config.network.connection_group_size,
+            workerGroupSize = config.network.worker_group_size,
+            callGroupSize = config.network.call_group_size,
         ).start()
 
     Runtime.getRuntime().addShutdownHook(

geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt

@@ -96,6 +96,29 @@ data class RelayConfig(
         val host: String = "0.0.0.0",
         val port: Int = 7447,
         val path: String = "/",
+        /**
+         * Ktor CIO acceptor-thread count. `null` (default) keeps Ktor's
+         * default sizing — fine up to a few thousand concurrent
+         * connections. On big-VM deployments targeting 10k+
+         * connections, lift this to roughly half the available cores
+         * so the acceptor doesn't starve workers.
+         */
+        val connection_group_size: Int? = null,
+        /**
+         * Ktor CIO worker-thread count (handles socket I/O). `null`
+         * keeps Ktor's default. Each connection's WebSocket read/write
+         * is dispatched onto this pool; for many idle long-lived
+         * connections the pool can stay small, but 10k+ connections
+         * benefit from sizing this to the full CPU count.
+         */
+        val worker_group_size: Int? = null,
+        /**
+         * Ktor CIO call-handling thread count. `null` keeps Ktor's
+         * default. Sized higher than [worker_group_size] because each
+         * call (incl. WebSocket upgrade) may suspend on I/O — at
+         * 10k+ connections, ~4× cores is a reasonable starting point.
+         */
+        val call_group_size: Int? = null,
     )
 
     data class DatabaseSection(

geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt

@@ -29,6 +29,7 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ClosedSendChannelException
 import kotlinx.coroutines.channels.consumeEach
 import kotlinx.coroutines.launch
+import java.util.concurrent.atomic.AtomicInteger
 
 /**
  * Per-WebSocket pump that owns the bounded outbound queue and the
@@ -44,15 +45,39 @@ import kotlinx.coroutines.launch
  *   3. `finally`-style teardown closes the queue, cancels the
  *      writer, unregisters the session, and closes it.
  *
- * Slow-client policy: when [outQueue] fills, [SESSION_OUTGOING_BUFFER]
+ * Slow-client policy: once the outbound backlog reaches
- * frames behind, the connection is dropped rather than silently
+ * [MAX_OUTGOING_BUFFER] frames, the connection is dropped rather
- * losing EVENT/EOSE — silent drop would corrupt NIP-01.
+ * than silently losing EVENT/EOSE — silent drop would corrupt
+ * NIP-01.
+ *
+ * Memory model: the outbound queue is `Channel.UNLIMITED`, which in
+ * kotlinx.coroutines allocates segments lazily — an idle connection
+ * pays only a small head-segment cost. The cap is enforced via
+ * [outstanding] rather than the channel's own capacity so we don't
+ * reserve a fixed-size buffer up-front for every connection. At
+ * 5 000+ idle connections this matters: an 8 192-slot fixed buffer
+ * per connection would otherwise dominate JVM heap usage even
+ * though the vast majority of connections never fan out.
  */
 internal class WebSocketSessionPump(
     private val ws: DefaultWebSocketServerSession,
 ) {
-    private val outQueue = Channel<String>(capacity = SESSION_OUTGOING_BUFFER)
+    /**
-    private var droppedForBackpressure = false
+     * Unbounded channel — bounded by [outstanding] above, not by the
+     * channel's own capacity. See class kdoc for memory rationale.
+     */
+    private val outQueue = Channel<String>(capacity = Channel.UNLIMITED)
+
+    /**
+     * Number of frames queued but not yet written to the socket.
+     * Producer increments before [Channel.trySend]; writer decrements
+     * after the frame is handed to Ktor. When this would cross
+     * [MAX_OUTGOING_BUFFER] we treat the client as slow and close
+     * the queue.
+     */
+    private val outstanding = AtomicInteger(0)
+
+    @Volatile private var droppedForBackpressure = false
 
     suspend fun pump(
         server: NostrServer,
@@ -64,6 +89,7 @@ internal class WebSocketSessionPump(
                 try {
                     for (json in outQueue) {
                         ws.outgoing.send(Frame.Text(json))
+                        outstanding.decrementAndGet()
                     }
                 } catch (_: ClosedSendChannelException) {
                     // socket closed — outer handler runs normal teardown.
@@ -71,13 +97,22 @@ internal class WebSocketSessionPump(
             }
         val session =
             server.connect { json ->
-                val res = outQueue.trySend(json)
+                // The channel itself is UNLIMITED, so trySend can't
-                if (!res.isSuccess && !res.isClosed) {
+                // report "full". Enforce the cap explicitly: increment
-                    // Buffer is full → slow client. Mark + close the
+                // first, refuse if we'd cross the bound, otherwise
-                    // queue; the writer drains, then the outer handler
+                // enqueue.
-                    // closes the WS session.
+                val depth = outstanding.incrementAndGet()
+                if (depth > MAX_OUTGOING_BUFFER) {
+                    outstanding.decrementAndGet()
                     droppedForBackpressure = true
                     outQueue.close()
+                    return@connect
+                }
+                val res = outQueue.trySend(json)
+                if (!res.isSuccess) {
+                    // Channel was closed concurrently (e.g. teardown).
+                    // Roll back the counter; nothing more to do.
+                    outstanding.decrementAndGet()
                 }
             }
         registerSession(session)
@@ -98,17 +133,19 @@ internal class WebSocketSessionPump(
 
     companion object {
         /**
-         * Per-session outbound buffer size. When a slow client falls
+         * Per-session outbound backlog cap. When a slow client falls
          * this many frames behind, we close their connection rather
          * than silently dropping further frames (which would corrupt
          * NIP-01 by missing EVENT/EOSE messages).
          *
          * Sized to hold fan-out for a connection holding several
          * thousand subscriptions when one event matches all of them
-         * — the realistic upper bound for a relay client. At ~250B
+         * — the realistic upper bound for a relay client. At ~250 B
-         * per frame this caps per-session memory at ~2 MiB before
+         * per frame this caps per-session worst-case memory at
-         * we drop the connection.
+         * ~2 MiB before we drop the connection. Idle connections
+         * pay only the small head-segment cost of an unlimited
+         * channel (≈ a few hundred bytes), not the full cap.
          */
-        const val SESSION_OUTGOING_BUFFER: Int = 8192
+        const val MAX_OUTGOING_BUFFER: Int = 8192
     }
 }

geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt

@@ -139,6 +139,195 @@ class LoadBenchmark {
             }
         }
 
+    /**
+     * Holds 10 000 idle WebSocket connections open against a single
+     * relay. Verifies that the adaptive outQueue (sketch A in
+     * [connection-scaling plan][1]) lets us cross the ~2 000-connection
+     * floor measured by [connectionsHeldOpen] without FD exhaustion or
+     * runaway RSS.
+     *
+     * RUN PREREQ: requires a process FD limit ≥ ~12 000 (each WS uses
+     * one FD on each side plus margin). On Linux: `ulimit -n 32768`
+     * before launching the test JVM.
+     *
+     * [1]: geode/plans/2026-05-07-connection-scaling.md
+     */
+    @Test
+    fun connectionsHeldOpen10k() =
+        benchmark("connections held open 10k") {
+            val target = 10_000
+            runBenchmarkServer { server, http ->
+                val httpUrl =
+                    okhttp3.Request
+                        .Builder()
+                        .url(server.url.replace("ws://", "http://"))
+                        .build()
+                val sockets = java.util.concurrent.CopyOnWriteArrayList<okhttp3.WebSocket>()
+                val opened = AtomicLong()
+                val gotEose = AtomicLong()
+                val opens =
+                    measureTime {
+                        repeat(target) {
+                            val ws =
+                                http.newWebSocket(
+                                    httpUrl,
+                                    object : okhttp3.WebSocketListener() {
+                                        override fun onOpen(
+                                            webSocket: okhttp3.WebSocket,
+                                            response: okhttp3.Response,
+                                        ) {
+                                            opened.incrementAndGet()
+                                            webSocket.send(
+                                                """["REQ","s",{"kinds":[1],"limit":1}]""",
+                                            )
+                                        }
+
+                                        override fun onMessage(
+                                            webSocket: okhttp3.WebSocket,
+                                            text: String,
+                                        ) {
+                                            if (text.startsWith("[\"EOSE\"")) {
+                                                gotEose.incrementAndGet()
+                                            }
+                                        }
+                                    },
+                                )
+                            sockets += ws
+                        }
+                        val deadline = System.currentTimeMillis() + 120_000
+                        while (gotEose.get() < target && System.currentTimeMillis() < deadline) {
+                            Thread.sleep(50)
+                        }
+                    }
+                // JVM heap usage, not OS RSS — we can only measure
+                // what the JVM itself has allocated. Force a GC first
+                // so the reading reflects retained bytes, not in-flight
+                // allocation churn from the connect ramp-up.
+                val rt = Runtime.getRuntime()
+                System.gc()
+                Thread.sleep(200)
+                val heapMb = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024)
+                println(
+                    "target=$target opened=${opened.get()} eosed=${gotEose.get()} " +
+                        "active=${server.activeSessionCount} elapsedMs=${opens.inWholeMilliseconds} " +
+                        "heapMb=$heapMb",
+                )
+                sockets.forEach { runCatching { it.cancel() } }
+                check(gotEose.get() == target.toLong()) {
+                    "expected $target EOSE but got ${gotEose.get()} — connection scaling regression"
+                }
+                check(heapMb < 1024) {
+                    "JVM heap $heapMb MiB exceeded 1 GiB ceiling for $target idle connections"
+                }
+            }
+        }
+
+    /**
+     * 5 000 idle subscribers, one publisher emitting 10 EPS for 10 s.
+     * Measures fan-out latency at scale — exercises the queue path
+     * for a connection that *does* fan out, not just an idle one.
+     *
+     * Each subscriber matches every published event (`kinds:[1]`),
+     * so a single EVENT generates 5 000 outbound frames per tick.
+     */
+    @Test
+    fun connectionsHeldOpenWithFanout() =
+        benchmark("connections held open with fanout") {
+            val subs = 5_000
+            val durationSeconds = 10
+            val targetEps = 10
+            runBenchmarkServer { server, http ->
+                val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+                val subClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope)
+                val pubClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope)
+                try {
+                    val relayUrl = server.url.normalizeRelayUrl()
+                    val received = AtomicLong()
+                    val eosed = AtomicLong()
+                    // Last-receive timestamp per event id. The N-th
+                    // subscriber to deliver wins; combined with the
+                    // publish timestamp this gives us the full fan-out
+                    // duration to the slowest subscriber.
+                    val lastReceiveNs =
+                        java.util.concurrent.ConcurrentHashMap<String, AtomicLong>()
+
+                    repeat(subs) { i ->
+                        subClient.subscribe(
+                            "fanout-$i",
+                            mapOf(relayUrl to listOf(Filter(kinds = listOf(1)))),
+                            object : SubscriptionListener {
+                                override fun onEvent(
+                                    event: com.vitorpamplona.quartz.nip01Core.core.Event,
+                                    isLive: Boolean,
+                                    relay: NormalizedRelayUrl,
+                                    forFilters: List<Filter>?,
+                                ) {
+                                    lastReceiveNs
+                                        .computeIfAbsent(event.id) { AtomicLong() }
+                                        .set(System.nanoTime())
+                                    received.incrementAndGet()
+                                }
+
+                                override fun onEose(
+                                    relay: NormalizedRelayUrl,
+                                    forFilters: List<Filter>?,
+                                ) {
+                                    eosed.incrementAndGet()
+                                }
+                            },
+                        )
+                    }
+
+                    runBlocking {
+                        withTimeout(120_000) {
+                            while (eosed.get() < subs) kotlinx.coroutines.delay(100)
+                        }
+                    }
+                    println("$subs subs ready; publishing ${targetEps * durationSeconds} events at $targetEps EPS...")
+
+                    val signer = NostrSignerSync(KeyPair())
+                    val publishedAt = java.util.concurrent.ConcurrentHashMap<String, Long>()
+                    val totalEvents = targetEps * durationSeconds
+                    val tickIntervalMs = 1000L / targetEps
+
+                    runBlocking {
+                        repeat(totalEvents) { i ->
+                            val event = signer.sign(TextNoteEvent.build("fanout-$i"))
+                            publishedAt[event.id] = System.nanoTime()
+                            pubClient.publishAndConfirm(event, setOf(relayUrl))
+                            kotlinx.coroutines.delay(tickIntervalMs)
+                        }
+                    }
+
+                    // Wait for fan-out completion (or 30s, whichever first).
+                    runBlocking {
+                        withTimeout(30_000) {
+                            while (received.get() < subs.toLong() * totalEvents) {
+                                kotlinx.coroutines.delay(100)
+                            }
+                        }
+                    }
+
+                    val perEventLastMs =
+                        lastReceiveNs.entries
+                            .mapNotNull { (id, last) ->
+                                publishedAt[id]?.let { (last.get() - it) / 1_000_000.0 }
+                            }.sorted()
+                    val p50 = perEventLastMs.getOrNull(perEventLastMs.size / 2) ?: -1.0
+                    val p99 = perEventLastMs.getOrNull((perEventLastMs.size * 99) / 100) ?: -1.0
+                    println(
+                        "subs=$subs events=$totalEvents received=${received.get()}/${subs.toLong() * totalEvents} " +
+                            "p50LastFanoutMs=${"%.1f".format(p50)} " +
+                            "p99LastFanoutMs=${"%.1f".format(p99)}",
+                    )
+                } finally {
+                    subClient.disconnect()
+                    pubClient.disconnect()
+                    scope.cancel()
+                }
+            }
+        }
+
     /**
      * One publisher sends 10k events serially. Measures the round-trip
      * `EVENT` → `OK true` time, which is dominated by SQLite write

quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toRelay/CommandDeserializer.kt

@@ -24,7 +24,6 @@ import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.core.JsonToken
 import com.fasterxml.jackson.databind.DeserializationContext
 import com.fasterxml.jackson.databind.deser.std.StdDeserializer
-import com.fasterxml.jackson.databind.node.ObjectNode
 import com.vitorpamplona.quartz.nip01Core.jackson.EventDeserializer
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.filters.ManualFilterDeserializer
@@ -52,9 +51,9 @@ class CommandDeserializer : StdDeserializer<Command>(Command::class.java) {
                     val filters = mutableListOf<Filter>()
 
                     while (jp.nextToken() != JsonToken.END_ARRAY) {
-                        val filterObj: ObjectNode = jp.codec.readTree(jp)
+                        // currentToken is now START_OBJECT for each filter;
-                        val filter = ManualFilterDeserializer.fromJson(filterObj)
+                        // the streaming parser consumes through END_OBJECT.
-                        filters.add(filter)
+                        filters.add(ManualFilterDeserializer.fromJson(jp))
                     }
 
                     ReqCmd(
@@ -68,9 +67,7 @@ class CommandDeserializer : StdDeserializer<Command>(Command::class.java) {
                     val filters = mutableListOf<Filter>()
 
                     while (jp.nextToken() != JsonToken.END_ARRAY) {
-                        val filterObj: ObjectNode = jp.codec.readTree(jp)
+                        filters.add(ManualFilterDeserializer.fromJson(jp))
-                        val filter = ManualFilterDeserializer.fromJson(filterObj)
-                        filters.add(filter)
                     }
 
                     CountCmd(
@@ -102,9 +99,8 @@ class CommandDeserializer : StdDeserializer<Command>(Command::class.java) {
 
                 NegOpenCmd.LABEL -> {
                     val subId = jp.nextTextValue()
-                    jp.nextToken()
+                    jp.nextToken() // advance to filter's START_OBJECT
-                    val filterObj: ObjectNode = jp.codec.readTree(jp)
+                    val filter = ManualFilterDeserializer.fromJson(jp)
-                    val filter = ManualFilterDeserializer.fromJson(filterObj)
                     val initialMessage = jp.nextTextValue()
 
                     NegOpenCmd(

quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterDeserializer.kt

@@ -21,6 +21,7 @@
 package com.vitorpamplona.quartz.nip01Core.relay.filters
 
 import com.fasterxml.jackson.core.JsonParser
+import com.fasterxml.jackson.core.JsonToken
 import com.fasterxml.jackson.databind.DeserializationContext
 import com.fasterxml.jackson.databind.deser.std.StdDeserializer
 import com.fasterxml.jackson.databind.node.ObjectNode
@@ -32,11 +33,149 @@ class FilterDeserializer : StdDeserializer<Filter>(Filter::class.java) {
     override fun deserialize(
         jp: JsonParser,
         ctxt: DeserializationContext,
-    ): Filter = ManualFilterDeserializer.fromJson(jp.codec.readTree(jp))
+    ): Filter = ManualFilterDeserializer.fromJson(jp)
 }
 
 class ManualFilterDeserializer {
     companion object {
+        /**
+         * Streaming filter parser. Reads field-by-field off [jp] without
+         * materializing an intermediate `JsonNode` tree — same shape as
+         * [com.vitorpamplona.quartz.nip01Core.jackson.EventDeserializer],
+         * which is what makes high-fan-out REQ traffic cheap on the
+         * relay-inbound path.
+         *
+         * Caller must position the parser so [jp.currentToken] is the
+         * `START_OBJECT` opening the filter. On return, [jp.currentToken]
+         * is the matching `END_OBJECT`.
+         *
+         * Tolerant by design — invalid array entries (wrong type, JSON
+         * null) are silently dropped, mirroring the
+         * `mapNotNull { it.asTextOrNull() }` behavior of the tree-based
+         * overload below. Unknown top-level fields are skipped via
+         * [JsonParser.skipChildren].
+         */
+        fun fromJson(jp: JsonParser): Filter {
+            var ids: MutableList<String>? = null
+            var authors: MutableList<String>? = null
+            var kinds: MutableList<Int>? = null
+            var tags: MutableMap<String, List<String>>? = null
+            var tagsAll: MutableMap<String, List<String>>? = null
+            var since: Long? = null
+            var until: Long? = null
+            var limit: Int? = null
+            var search: String? = null
+
+            while (jp.nextToken() != JsonToken.END_OBJECT) {
+                val name = jp.currentName()
+                jp.nextToken() // advance to value
+                when {
+                    name == "ids" -> {
+                        ids = readStringArray(jp)
+                    }
+
+                    name == "authors" -> {
+                        authors = readStringArray(jp)
+                    }
+
+                    name == "kinds" -> {
+                        kinds = readIntArray(jp)
+                    }
+
+                    name == "since" -> {
+                        if (jp.currentToken != JsonToken.VALUE_NULL) since = jp.longValue
+                    }
+
+                    name == "until" -> {
+                        if (jp.currentToken != JsonToken.VALUE_NULL) until = jp.longValue
+                    }
+
+                    name == "limit" -> {
+                        if (jp.currentToken != JsonToken.VALUE_NULL) limit = jp.intValue
+                    }
+
+                    name == "search" -> {
+                        if (jp.currentToken != JsonToken.VALUE_NULL) search = jp.text
+                    }
+
+                    name.length > 1 && name[0] == '#' -> {
+                        val map = tags ?: mutableMapOf<String, List<String>>().also { tags = it }
+                        map[name.substring(1)] = readStringArray(jp)
+                    }
+
+                    name.length > 1 && name[0] == '&' -> {
+                        val map = tagsAll ?: mutableMapOf<String, List<String>>().also { tagsAll = it }
+                        map[name.substring(1)] = readStringArray(jp)
+                    }
+
+                    else -> {
+                        jp.skipChildren()
+                    }
+                }
+            }
+
+            return Filter(
+                ids = ids,
+                authors = authors,
+                kinds = kinds,
+                tags = tags,
+                tagsAll = tagsAll,
+                since = since,
+                until = until,
+                limit = limit,
+                search = search,
+            )
+        }
+
+        /**
+         * Reads a string array off [jp]. Drops non-string entries and
+         * JSON nulls — matches the `mapNotNull { it.asTextOrNull() }`
+         * tolerance of the tree-based path. Returns an empty list when
+         * the value is anything other than a `START_ARRAY` (incl.
+         * `null`), so callers don't have to special-case that.
+         */
+        private fun readStringArray(jp: JsonParser): MutableList<String> {
+            val out = mutableListOf<String>()
+            if (jp.currentToken == JsonToken.START_ARRAY) {
+                while (jp.nextToken() != JsonToken.END_ARRAY) {
+                    if (jp.currentToken == JsonToken.VALUE_STRING) {
+                        out.add(jp.text)
+                    } else if (jp.currentToken == JsonToken.START_OBJECT || jp.currentToken == JsonToken.START_ARRAY) {
+                        jp.skipChildren()
+                    }
+                }
+            } else if (jp.currentToken == JsonToken.START_OBJECT) {
+                jp.skipChildren()
+            }
+            return out
+        }
+
+        /**
+         * Reads an int array off [jp]. Same tolerance rules as
+         * [readStringArray] — non-numeric entries are dropped.
+         */
+        private fun readIntArray(jp: JsonParser): MutableList<Int> {
+            val out = mutableListOf<Int>()
+            if (jp.currentToken == JsonToken.START_ARRAY) {
+                while (jp.nextToken() != JsonToken.END_ARRAY) {
+                    when (jp.currentToken) {
+                        JsonToken.VALUE_NUMBER_INT, JsonToken.VALUE_NUMBER_FLOAT -> out.add(jp.intValue)
+                        JsonToken.START_OBJECT, JsonToken.START_ARRAY -> jp.skipChildren()
+                        else -> Unit
+                    }
+                }
+            } else if (jp.currentToken == JsonToken.START_OBJECT) {
+                jp.skipChildren()
+            }
+            return out
+        }
+
+        /**
+         * Tree-based overload kept for callers that already have an
+         * `ObjectNode` in hand (e.g. cross-format adapters). New code on
+         * the relay-inbound path should use the streaming overload —
+         * this one materializes the full filter tree first.
+         */
         fun fromJson(jsonObject: ObjectNode): Filter {
             val tagsIn = mutableListOf<String>()
             jsonObject.fieldNames().forEach {