From 0f7d8696deb352a0bf4d2b5321fb7e3b21759483 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 20:12:50 +0000 Subject: [PATCH 1/4] perf(geode): adaptive outQueue + CIO pool sizing for 10k+ connections Implements geode/plans/2026-05-07-connection-scaling.md to push the single-relay connection ceiling past the ~2k floor measured by LoadBenchmark.connectionsHeldOpen. - WebSocketSessionPump: switch outQueue to Channel.UNLIMITED bounded by an AtomicInteger backlog cap. Idle connections no longer reserve an 8 192-slot fixed buffer; lazy-allocated head segments cost only a few hundred bytes per idle session. Slow-client policy preserved: once outstanding > MAX_OUTGOING_BUFFER, the queue is closed and the connection drops, so NIP-01 ordering is never silently violated. - LocalRelayServer / RelayConfig.NetworkSection: expose CIO connectionGroupSize / workerGroupSize / callGroupSize so big-VM operators can tune Ktor's event-loop pools without forking. Switch to the serverConfig {} + configure {} embeddedServer overload so CIO tunables can be set. - LoadBenchmark: add connectionsHeldOpen10k (asserts 10 000 idle WebSockets settle under a 1 GiB heap ceiling) and connectionsHeldOpenWithFanout (5 000 subscribers, 10 EPS fan-out for 10 s, reports p50/p99 last-fanout latency). - config.example.toml: document the three CIO tunables and the ulimit -n requirement for operators targeting >1k connections. --- geode/config.example.toml | 14 ++ .../vitorpamplona/geode/LocalRelayServer.kt | 123 ++++++++---- .../kotlin/com/vitorpamplona/geode/Main.kt | 3 + .../vitorpamplona/geode/config/RelayConfig.kt | 23 +++ .../geode/server/WebSocketSessionPump.kt | 67 +++++-- .../vitorpamplona/geode/perf/LoadBenchmark.kt | 186 ++++++++++++++++++ 6 files changed, 359 insertions(+), 57 deletions(-) diff --git a/geode/config.example.toml b/geode/config.example.toml index 5835a5527..3d706999b 100644 --- a/geode/config.example.toml +++ b/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 diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt index 3840bf482..56ef2f6da 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt +++ b/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) { - install(WebSockets) { - maxFrameBytes?.let { maxFrameSize = it } - } - routing { - // NIP-11: GET on the relay URL with Accept: - // application/nostr+json returns the relay info doc. - // We mount this *before* the webSocket route so Ktor - // serves NIP-11 for plain HTTP GETs and only upgrades - // to a WebSocket when the request is a WS upgrade. - get(path) { - val accept = call.request.header(HttpHeaders.Accept).orEmpty() - if (accept.contains("application/nostr+json")) { - call.response.headers.append("Access-Control-Allow-Origin", "*") - call.respondText( - relay.info.json, - ContentType.parse("application/nostr+json"), - ) - } else { - call.respondText( - "Use a Nostr client (NIP-01 WebSocket) or send Accept: application/nostr+json (NIP-11).", - ContentType.Text.Plain, - HttpStatusCode.UpgradeRequired, - ) + embeddedServer( + factory = CIO, + rootConfig = + serverConfig { + module { + install(WebSockets) { + maxFrameBytes?.let { maxFrameSize = it } + } + routing { + // NIP-11: GET on the relay URL with Accept: + // application/nostr+json returns the relay info doc. + // We mount this *before* the webSocket route so Ktor + // serves NIP-11 for plain HTTP GETs and only upgrades + // to a WebSocket when the request is a WS upgrade. + get(path) { + val accept = call.request.header(HttpHeaders.Accept).orEmpty() + if (accept.contains("application/nostr+json")) { + call.response.headers.append("Access-Control-Allow-Origin", "*") + call.respondText( + relay.info.json, + ContentType.parse("application/nostr+json"), + ) + } 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 - // 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, - ) - } - } - } + // Keep Ktor defaults unless the operator overrode + // them — Ktor's per-CPU sizing is sensible for + // most deployments, and over-threading hurts L1/L2 + // locality at low connection counts. + connGrp?.let { connectionGroupSize = it } + workGrp?.let { workerGroupSize = it } + callGrp?.let { callGroupSize = it } + }, + ) server.start(wait = false) engine = server.engine // Ktor 3.x made resolvedConnectors() suspend. We block here so diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 228c40709..e7b7a97b0 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -117,6 +117,9 @@ fun main(args: Array) { 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( diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt index b1b99634d..1417ec61f 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt +++ b/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( diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt index 7cd700a33..f5319779d 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt +++ b/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] - * frames behind, the connection is dropped rather than silently - * losing EVENT/EOSE — silent drop would corrupt NIP-01. + * Slow-client policy: once the outbound backlog reaches + * [MAX_OUTGOING_BUFFER] frames, the connection is dropped rather + * 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(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(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) - if (!res.isSuccess && !res.isClosed) { - // Buffer is full → slow client. Mark + close the - // queue; the writer drains, then the outer handler - // closes the WS session. + // The channel itself is UNLIMITED, so trySend can't + // report "full". Enforce the cap explicitly: increment + // first, refuse if we'd cross the bound, otherwise + // enqueue. + 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 - * per frame this caps per-session memory at ~2 MiB before - * we drop the connection. + * — the realistic upper bound for a relay client. At ~250 B + * per frame this caps per-session worst-case memory at + * ~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 } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt index a083dd812..839d22f58 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -139,6 +139,192 @@ 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() + 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) + } + } + val rt = Runtime.getRuntime() + val rssMb = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024) + println( + "target=$target opened=${opened.get()} eosed=${gotEose.get()} " + + "active=${server.activeSessionCount} elapsedMs=${opens.inWholeMilliseconds} " + + "heapMb=$rssMb", + ) + sockets.forEach { runCatching { it.cancel() } } + check(gotEose.get() == target.toLong()) { + "expected $target EOSE but got ${gotEose.get()} — connection scaling regression" + } + check(rssMb < 1024) { + "heap usage $rssMb 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() + // Per-event fan-out latency, capped at the # of + // events we expect (durationSeconds * targetEps). + val fanoutLatenciesNs = + java.util.concurrent.ConcurrentHashMap() + val firstSeenNs = + java.util.concurrent.ConcurrentHashMap() + + 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?, + ) { + val now = System.nanoTime() + firstSeenNs + .computeIfAbsent(event.id) { AtomicLong(now) } + fanoutLatenciesNs + .computeIfAbsent(event.id) { AtomicLong(now) } + .set(now) + received.incrementAndGet() + } + + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + 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() + 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 = + fanoutLatenciesNs.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 From 64624bdfdb548642b06747b595a121b959505f94 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 20:41:39 +0000 Subject: [PATCH 2/4] perf(quartz): streaming filter deserializer to remove tree allocation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The relay-inbound path (REQ / COUNT / NEG-OPEN) was the only place ManualFilterDeserializer still went through `jp.codec.readTree(jp)`, materializing a full ObjectNode per filter before walking it. The Command/Message/Event deserializers next door already stream straight off the JsonParser; this brings filter parsing onto the same shape. - Add ManualFilterDeserializer.fromJson(JsonParser): mirrors EventDeserializer's hand-rolled token loop. Dispatches on field name, reads the seven fixed keys + dynamic #x / &x tag arrays via nextToken / nextTextValue / longValue / intValue, and drops invalid entries silently (same tolerance as the tree-based mapNotNull path). - Wire the four internal call sites — three in CommandDeserializer (REQ, COUNT, NEG-OPEN) and the standalone FilterDeserializer — to the streaming overload. - Keep the existing fromJson(ObjectNode) overload as-is for any external/cross-format adapter that already has a tree in hand. For a single REQ with N filters this drops N ObjectNode trees per inbound frame, which at 10k connections × ~5 filters × 1 msg/s is ~50k tree allocations/sec we don't have to do. No behavioral change: all quartz JVM tests pass (including the cross-mapper round-trip suite that compares Jackson and KotlinSerialization output) and geode's relay tests pass. --- .../commands/toRelay/CommandDeserializer.kt | 16 +- .../relay/filters/FilterDeserializer.kt | 141 +++++++++++++++++- 2 files changed, 146 insertions(+), 11 deletions(-) diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toRelay/CommandDeserializer.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toRelay/CommandDeserializer.kt index 3b5a8d622..258e52c8d 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toRelay/CommandDeserializer.kt +++ b/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::class.java) { val filters = mutableListOf() while (jp.nextToken() != JsonToken.END_ARRAY) { - val filterObj: ObjectNode = jp.codec.readTree(jp) - val filter = ManualFilterDeserializer.fromJson(filterObj) - filters.add(filter) + // currentToken is now START_OBJECT for each filter; + // the streaming parser consumes through END_OBJECT. + filters.add(ManualFilterDeserializer.fromJson(jp)) } ReqCmd( @@ -68,9 +67,7 @@ class CommandDeserializer : StdDeserializer(Command::class.java) { val filters = mutableListOf() while (jp.nextToken() != JsonToken.END_ARRAY) { - val filterObj: ObjectNode = jp.codec.readTree(jp) - val filter = ManualFilterDeserializer.fromJson(filterObj) - filters.add(filter) + filters.add(ManualFilterDeserializer.fromJson(jp)) } CountCmd( @@ -102,9 +99,8 @@ class CommandDeserializer : StdDeserializer(Command::class.java) { NegOpenCmd.LABEL -> { val subId = jp.nextTextValue() - jp.nextToken() - val filterObj: ObjectNode = jp.codec.readTree(jp) - val filter = ManualFilterDeserializer.fromJson(filterObj) + jp.nextToken() // advance to filter's START_OBJECT + val filter = ManualFilterDeserializer.fromJson(jp) val initialMessage = jp.nextTextValue() NegOpenCmd( diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterDeserializer.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterDeserializer.kt index 05081abdd..cdab3a9cd 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterDeserializer.kt +++ b/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::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? = null + var authors: MutableList? = null + var kinds: MutableList? = null + var tags: MutableMap>? = null + var tagsAll: MutableMap>? = 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>().also { tags = it } + map[name.substring(1)] = readStringArray(jp) + } + + name.length > 1 && name[0] == '&' -> { + val map = tagsAll ?: mutableMapOf>().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 { + val out = mutableListOf() + 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 { + val out = mutableListOf() + 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() jsonObject.fieldNames().forEach { From 6c792fdf449aa637b4f33af2448ef5b08a812aa8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 20:50:10 +0000 Subject: [PATCH 3/4] audit(geode): drop dead firstSeenNs + clarify heap-vs-RSS in LoadBenchmark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-merge audit findings on the connection-scaling perf changes: - connectionsHeldOpenWithFanout populated firstSeenNs but never read it — only lastReceiveNs feeds the p50/p99 latency metric. Drop the unused map and rename the latency map to lastReceiveNs to match what it actually holds. - connectionsHeldOpen10k printed/asserted on a variable named rssMb that was actually JVM heap (`Runtime.totalMemory - freeMemory`). Rename to heapMb, force a GC + 200 ms settle before reading so the number reflects retained bytes rather than connect-ramp churn, and update the assertion message to say "JVM heap" not "heap usage". --- .../vitorpamplona/geode/perf/LoadBenchmark.kt | 35 ++++++++++--------- 1 file changed, 19 insertions(+), 16 deletions(-) diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt index 839d22f58..04a5867fc 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -199,19 +199,25 @@ class LoadBenchmark { 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() - val rssMb = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024) + 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=$rssMb", + "heapMb=$heapMb", ) sockets.forEach { runCatching { it.cancel() } } check(gotEose.get() == target.toLong()) { "expected $target EOSE but got ${gotEose.get()} — connection scaling regression" } - check(rssMb < 1024) { - "heap usage $rssMb MiB exceeded 1 GiB ceiling for $target idle connections" + check(heapMb < 1024) { + "JVM heap $heapMb MiB exceeded 1 GiB ceiling for $target idle connections" } } } @@ -238,11 +244,11 @@ class LoadBenchmark { val relayUrl = server.url.normalizeRelayUrl() val received = AtomicLong() val eosed = AtomicLong() - // Per-event fan-out latency, capped at the # of - // events we expect (durationSeconds * targetEps). - val fanoutLatenciesNs = - java.util.concurrent.ConcurrentHashMap() - val firstSeenNs = + // 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() repeat(subs) { i -> @@ -256,12 +262,9 @@ class LoadBenchmark { relay: NormalizedRelayUrl, forFilters: List?, ) { - val now = System.nanoTime() - firstSeenNs - .computeIfAbsent(event.id) { AtomicLong(now) } - fanoutLatenciesNs - .computeIfAbsent(event.id) { AtomicLong(now) } - .set(now) + lastReceiveNs + .computeIfAbsent(event.id) { AtomicLong() } + .set(System.nanoTime()) received.incrementAndGet() } @@ -306,7 +309,7 @@ class LoadBenchmark { } val perEventLastMs = - fanoutLatenciesNs.entries + lastReceiveNs.entries .mapNotNull { (id, last) -> publishedAt[id]?.let { (last.get() - it) / 1_000_000.0 } }.sorted() From 6f8e7bb5206440e03ec8323299fb63ade0fd62f3 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 21:07:03 +0000 Subject: [PATCH 4/4] docs(geode): update connection-scaling plan to reflect what shipped MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Marks Sketches A and B done, with a note that A took the simpler Channel.UNLIMITED + AtomicInteger cap path the original Risks section called out, sidestepping the channel-swap that the plan first sketched. Records that the streaming-filter slice of Sketch C landed in Quartz, and that the larger envelope-streaming work the plan called out is unnecessary because MessageDeserializer / CommandDeserializer / EventDeserializer were already streaming — only the filter sub-object went through readTree. Adds the verification benchmarks (connectionsHeldOpen10k, connectionsHeldOpenWithFanout) to the verification section, with a correction that what's measured is JVM heap not OS RSS. Carries forward the not-done items (fan-out de-duplication, filter-matching index, Netty engine) into the open-work section, pointing at live-broadcast-fanout-index.md for the highest-leverage remaining work. --- geode/plans/2026-05-07-connection-scaling.md | 209 ++++++++++++++----- 1 file changed, 159 insertions(+), 50 deletions(-) diff --git a/geode/plans/2026-05-07-connection-scaling.md b/geode/plans/2026-05-07-connection-scaling.md index fb1f8b353..e16ab3f11 100644 --- a/geode/plans/2026-05-07-connection-scaling.md +++ b/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 -producer side trySends and we observe queue depth crossing a high-water -mark (e.g. 75% full), grow the channel up to `MAX_OUTGOING_BUFFER = -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. +> Original plan: start at `INITIAL_OUTGOING_BUFFER = 64` and swap to a +> wider channel under a per-session lock when a high-water mark +> trips. **Not how it shipped.** -Expected: 90% of connections never fan out, so they stay at 64 slots -× ~512 B per ref ≈ 32 KB. At 5 000 conns that's ~160 MB → ~5 MB. -Hot-fanout connections still get the 2 MB cap. - -### B — per-relay event-loop pool sizing - -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: +What actually shipped is the simpler alternative the original Risks +section called out: `Channel.UNLIMITED` plus an `AtomicInteger` +backlog cap. kotlinx.coroutines' `BufferedChannel` allocates segments +lazily, so an unlimited channel pays only the small head-segment cost +on idle connections — there is no preallocated buffer to scale. ```kotlin -embeddedServer(CIO, ...) { - connectionGroupSize = max(2, Runtime.getRuntime().availableProcessors() / 2) - workerGroupSize = max(4, Runtime.getRuntime().availableProcessors()) - callGroupSize = max(8, Runtime.getRuntime().availableProcessors() * 4) +private val outQueue = Channel(capacity = Channel.UNLIMITED) +private val outstanding = AtomicInteger(0) + +// 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 -big VM can lift them. +Memory characteristic the plan asked for is intact: idle connections +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 -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. +### B — per-relay event-loop pool sizing ✅ shipped -This is more of a quartz-level change than geode-specific, but -geode's load benchmark is the right place to measure it. +Three optional knobs added to `[network]` in `RelayConfig`: -## 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; - asserts no FD exhaustion + RSS stays under 1 GB. -- `connectionsHeldOpenWithFanout` — 5 000 idle subscribers, - 10 EPS published; measures p99 fanout latency at scale. +`config.example.toml` documents the knobs and includes the operator +note that targeting >1k connections needs `ulimit -n 65536` (or +matching `LimitNOFILE=` in a systemd unit). -The current `connectionsHeldOpen` benchmark stays as the baseline -floor (~2 000 conns). +### C — reduce per-message JSON allocations ✅ partially shipped (in Quartz) -## 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 - must preserve OK ordering. A simpler alternative: keep capacity fixed, - but lazily allocate a small `ArrayDeque` only when the first - message is sent. Channels in kotlinx.coroutines do allocate up-front. -- **Bumping CIO group sizes can hurt**: more threads can mean worse - L1/L2 locality. Always benchmark before/after, don't trust - intuitive sizing. -- **OS-level FD limit**: per-process FD limit on Linux defaults to - 1024 in many environments. Document the `ulimit -n` requirement - for operators targeting >1k connections. +Audit of `quartz/.../jackson` showed the Command/Message envelope is +already streaming: + +| Path | Already streaming? | Tree alloc? | +| ----------------------- | ------------------ | -------------------------------------------------- | +| `MessageDeserializer` | yes | only for `COUNT` result (rare) | +| `CommandDeserializer` | yes | only for **filter sub-objects** in REQ/COUNT/NEG-OPEN | +| `EventDeserializer` | yes | none — `currentName().hashCode()` dispatch | +| `ManualFilterDeserializer` | **no** | `jp.codec.readTree(jp)` per filter | + +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.