From 39e81c1eb555482c3c99db7b322e55589797c41f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 21:35:49 +0000 Subject: [PATCH 1/5] docs(geode): correct OK ordering/durability assumptions in ingestion plan OK frames carry the event id, so clients pair replies by id rather than by arrival order. NIP-01 also treats OK true as "accepted," not "fsynced." That removes two constraints the plan was carrying: - no per-connection FIFO requirement on OKs - no need to delay OKs until after batch fsync Tier 1 can fan OKs out as soon as the per-row INSERT returns inside the open transaction, hiding the group-commit fsync entirely from publisher latency. Tier 2 drops the order-preserving commit log and just sends OKs straight to outQueue. The pipelined benchmark now checks one-OK-per-event-id rather than ordering. --- .../2026-05-07-event-ingestion-batching.md | 51 ++++++++++++------- 1 file changed, 32 insertions(+), 19 deletions(-) diff --git a/geode/plans/2026-05-07-event-ingestion-batching.md b/geode/plans/2026-05-07-event-ingestion-batching.md index e6df99ebf..561d1dd08 100644 --- a/geode/plans/2026-05-07-event-ingestion-batching.md +++ b/geode/plans/2026-05-07-event-ingestion-batching.md @@ -19,14 +19,20 @@ contention, not WS throughput). ## Constraints we must keep -- **OK ordering**: NIP-01 requires the OK reply to follow its EVENT. - We cannot reply OK before the insert decision (the OK carries - accepted/rejected + reason). -- **Durability semantics**: clients reasonably assume `OK true` means - "stored." Batching must not make us reply OK before fsync. -- **Per-connection FIFO**: a publisher that sends three EVENTs in a - row expects three OKs in that order. Reordering across connections - is fine. +- **OK pairs by event id, not by order**: the OK frame carries the + event id, so clients pair replies to publishes by id. OKs can be + emitted in any order — including reordered against the EVENT + stream, and against each other on the same connection. This frees + us to fan OKs out as soon as the writer has a per-row decision. +- **OK semantics = accepted, not fsynced**: NIP-01 treats `OK true` + as "accepted by the relay," not "durably on disk." We can reply as + soon as SQLite returns success for the row (inside the open + transaction, before commit/fsync). Group commit can batch the + fsync without holding OKs back. +- **Per-row decision still required**: the OK reason field is + per-event (duplicate, blocked, invalid sig, pow, etc.), so we + cannot fan out a single batch-level OK. Each row's OK must reflect + that row's outcome. ## Sketch @@ -36,7 +42,12 @@ Confirm `PRAGMA journal_mode=WAL` + `PRAGMA synchronous=NORMAL` on the event-store DB; group commits across the writer mutex's hold window. Today each insert is its own transaction. Wrap N inserts (or a 5 ms budget, whichever first) in a single transaction managed by the writer -coroutine. On commit, fan back N OK replies. +coroutine. + +Because OK reflects acceptance not durability, each row can fan an OK +as soon as the per-row INSERT statement returns inside the +transaction — we do not need to wait for the batch's commit. The +fsync is hidden from the publisher latency budget entirely. Implementation lives in quartz's `EventStore` / `SQLiteConnectionPool`, not geode — but geode owns the benchmark and validates the gain. @@ -48,17 +59,18 @@ commit is well-trodden territory (nostr-rs-relay, strfry both do it). `RelaySession.receive` is currently single-flight: one EVENT in, process, OK out, next EVENT. Allow a connection to push N EVENTs -concurrently, dispatch them to a per-connection ingest pipeline, and -serialise OKs back in arrival order via a small commit log. +concurrently and dispatch them to a per-connection ingest pipeline. -A `Channel with capacity = INGEST_PIPELINE_DEPTH` per -connection, drained by a coroutine that batches into the group-commit -above. OK responses are written to an `outQueue.send()` already — so -the pipeline just needs to record arrival order and emit OKs in that -order after each batch commits. +A `Channel` with capacity = `INGEST_PIPELINE_DEPTH` per +connection, drained by a coroutine that feeds the group-commit writer +above. OKs go straight to `outQueue.send()` the moment each row +returns from INSERT — no ordering bookkeeping needed, since the OK +frame already carries the event id and the spec doesn't require +order. A pipelined publisher keying on event id will pair replies +correctly. -Expected: hides the verify+insert latency behind another EVENT's -parse, gets us closer to network-bound throughput. +Expected: hides verify+insert latency behind the next EVENT's parse, +gets us closer to network-bound throughput. ### Tier 3 — eager Schnorr verify off the writer thread @@ -74,7 +86,8 @@ Add to `geode.perf.LoadBenchmark`: - `publishGroupCommitSingleClient` — same workload as the current single-client benchmark, asserts >5000 EPS. - `publishPipelinedSingleClient` — sends 100 EVENTs without awaiting - intermediate OKs; measures end-to-end and OK-ordering correctness. + intermediate OKs; measures end-to-end throughput and verifies that + every event id receives exactly one OK (in any order). Existing benchmarks stay as the regression floor. From 7a92f4ef2f149d67a735e2689ab72811c6a71b03 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 22:41:40 +0000 Subject: [PATCH 2/5] perf(quartz): group-commit + per-connection ingest pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the event-ingestion-batching plan: SQLite group commit with per-row SAVEPOINT isolation, and a per-server IngestQueue that turns RelaySession.handleEvent into fire-and-forget. The OK frame is emitted from the writer's callback once the row's outcome is known, relying on NIP-01 pairing OKs by event id (not by order). - IEventStore.batchInsert + InsertOutcome contract; SQLite override uses SAVEPOINTs so one bad event doesn't roll back the others. ObservableEventStore forwards persistable rows to the inner batch and emits StoreChange.Insert for accepted ones. - IngestQueue drains submissions in batches up to 64 per transaction. Writer coroutine starts lazily on the first submit so subscription-only sessions don't pay for it (and don't perturb Default-dispatcher scheduling — the eager launch was visible as intermittent NostrClientRepeatSubTest flakes under full-suite load). - RelaySession.handleEvent posts to the queue and returns immediately; the WS pump moves to the next frame instead of awaiting SQLite. ClosedSendChannelException during shutdown surfaces as OK false rather than crashing the pump. - LiveEventStore.submit fans an event onto the live stream only after the writer reports Accepted; the suspending insert is retained for tests, routed through the same queue. - New publishPipelinedSingleClient benchmark in geode.perf: 10 000 EVENTs back-to-back without awaiting OKs, asserts every event id receives exactly one OK (in any order). --- .../vitorpamplona/geode/perf/LoadBenchmark.kt | 95 ++++++++ .../nip01Core/relay/server/IngestQueue.kt | 209 ++++++++++++++++++ .../nip01Core/relay/server/LiveEventStore.kt | 44 +++- .../nip01Core/relay/server/NostrServer.kt | 14 +- .../nip01Core/relay/server/RelaySession.kt | 27 ++- .../quartz/nip01Core/store/IEventStore.kt | 36 +++ .../nip01Core/store/ObservableEventStore.kt | 45 ++++ .../nip01Core/store/sqlite/EventStore.kt | 2 + .../store/sqlite/SQLiteEventStore.kt | 57 +++++ 9 files changed, 521 insertions(+), 8 deletions(-) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt 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 04a5867fc..6f8e05fdc 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -450,6 +450,101 @@ class LoadBenchmark { } } + /** + * One publisher fires N EVENTs back-to-back without awaiting + * intermediate OKs, then collects all OKs by event id. This is + * the workload that exercises Tier 2 (per-connection ingest + * pipeline) + Tier 1 (group commit) together — multiple events + * are in flight on the same connection, so the writer can batch. + * + * Verifies the relaxed OK contract: every event id receives + * exactly one OK frame, in any order. + */ + @Test + fun publishPipelinedSingleClient() = + benchmark("publish pipelined single client") { + runBenchmarkServer { server, http -> + val n = 10_000 + val signer = NostrSignerSync(KeyPair()) + val events = + runBlocking { + (0 until n).map { i -> + signer.sign(TextNoteEvent.build("pipe $i")) + } + } + val ids = events.mapTo(HashSet()) { it.id } + + val httpUrl = + okhttp3.Request + .Builder() + .url(server.url.replace("ws://", "http://")) + .build() + val okSeen = AtomicLong() + val okFailures = AtomicLong() + val unknownIds = AtomicLong() + val seenIds = + java.util.concurrent.ConcurrentHashMap + .newKeySet() + val done = java.util.concurrent.CountDownLatch(1) + + val ws = + http.newWebSocket( + httpUrl, + object : okhttp3.WebSocketListener() { + override fun onMessage( + webSocket: okhttp3.WebSocket, + text: String, + ) { + if (!text.startsWith("[\"OK\"")) return + // ["OK","",true|false,""] — + // a tiny string scan is enough for a + // bench. Index 6 is past `["OK","`. + val idStart = 7 + val idEnd = text.indexOf('"', idStart) + if (idEnd <= idStart) return + val id = text.substring(idStart, idEnd) + if (!ids.contains(id)) { + unknownIds.incrementAndGet() + return + } + if (!seenIds.add(id)) return + if (text.contains(",true,")) { + okSeen.incrementAndGet() + } else { + okFailures.incrementAndGet() + } + if (okSeen.get() + okFailures.get() == n.toLong()) done.countDown() + } + }, + ) + + val elapsed = + measureTime { + // Burst-send: queue every EVENT to OkHttp's + // outbound buffer without any await, then + // wait for the corresponding OK frames. + for (event in events) { + ws.send("""["EVENT",${event.toJson()}]""") + } + check(done.await(60, java.util.concurrent.TimeUnit.SECONDS)) { + "timed out waiting for OKs: ok=${okSeen.get()} rej=${okFailures.get()} unknown=${unknownIds.get()}" + } + } + val eps = (n * 1000.0) / elapsed.inWholeMilliseconds + println( + "events=$n ok=${okSeen.get()} rejected=${okFailures.get()} " + + "unknownIds=${unknownIds.get()} elapsedMs=${elapsed.inWholeMilliseconds} eps=${"%.0f".format(eps)}", + ) + check(okSeen.get() == n.toLong()) { + "expected $n accepted OKs, got ${okSeen.get()} (rejected ${okFailures.get()})" + } + check(seenIds.size == n) { + "expected $n unique OK ids, got ${seenIds.size} — duplicate or missing OKs" + } + ws.cancel() + } + } + /** * Many concurrent publishers, each on their own WebSocket. Tells * us whether the SQLite single-writer bottleneck is the floor or diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt new file mode 100644 index 000000000..251c1e781 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt @@ -0,0 +1,209 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay.server + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.utils.Log +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.launch +import kotlin.coroutines.CoroutineContext + +/** + * Group-commit writer for incoming EVENT publishes. + * + * Submissions from any number of [RelaySession]s land in [incoming], + * a single drain coroutine pulls one item to start a batch then + * greedily drains everything else already queued (up to [maxBatch]), + * and forwards the whole batch through [IEventStore.batchInsert] — + * one writer-mutex acquisition + one BEGIN / COMMIT for the lot. + * + * Why this lives here: the SQLite event store enforces a single-writer + * mutex (matching SQLite's own file-level rule), so per-event + * `useWriter` calls serialise no matter how many publishers we have. + * Group commit collapses N mutex round-trips into one and lets + * BEGIN / WAL-append / COMMIT amortise across the batch. + * + * OK semantics (NIP-01): + * - The OK frame carries the event id, so clients pair replies by + * id, not by arrival order. We dispatch each [Submission.onComplete] + * as its row resolves; reordering across connections (and on the + * same connection) is allowed. + * - `OK true` means "accepted by this relay," not "fsynced." Since + * the underlying SQLite pool runs `synchronous = OFF` and WAL, + * a successful row inside the open transaction is the strongest + * guarantee we provide; replying after the batch's COMMIT (which + * is what happens in the current implementation) just adds the + * in-memory commit cost and keeps the write-failure path simple. + * + * Per-row error isolation lives in the store layer (SAVEPOINT in + * [com.vitorpamplona.quartz.nip01Core.store.sqlite.SQLiteEventStore]). + * A duplicate, expired event, etc. is a Rejected outcome for that + * row only. + * + * Backpressure: [incoming] is bounded at [capacity]. A flood of + * EVENTs that outpaces the writer suspends [submit] callers (i.e. + * the per-connection ingest path) until the writer drains. This + * propagates back through the WebSocket pump so a slow disk + * eventually slows the publisher rather than ballooning JVM memory. + */ +class IngestQueue( + private val store: IEventStore, + parentContext: CoroutineContext, + private val maxBatch: Int = DEFAULT_MAX_BATCH, + capacity: Int = DEFAULT_CAPACITY, +) : AutoCloseable { + /** + * One outstanding ingest request: the event to insert plus the + * callback the writer fires once the row's outcome is known. + */ + class Submission( + val event: Event, + val onComplete: (IEventStore.InsertOutcome) -> Unit, + ) + + private val incoming = Channel(capacity) + private val scope = CoroutineScope(parentContext + SupervisorJob()) + + /** + * Lazily-launched drain coroutine. We don't start it in `init` + * because eagerly launching from a server's lazy + * `LiveEventStore` allocates a Default-dispatcher slot at server + * construction time — visible to other tests sharing the same + * `Dispatchers.Default` pool, where it can perturb scheduling + * for unrelated REQ/EOSE timing. Starting on first `submit` + * keeps relays that never see an EVENT (read-only sessions, + * negentropy-only) from paying for the writer at all. + */ + @Volatile + private var writerStarted = false + private val startLock = Any() + + /** + * Hand off [event] for insertion. [onComplete] is invoked once + * with the per-row outcome from the writer batch — exactly once, + * unless the queue closes mid-flight. + * + * Suspends only when [incoming] is full (writer fell behind by + * [DEFAULT_CAPACITY] events). Otherwise this returns as fast as a + * channel `send`, freeing the WebSocket pump to read the next + * frame — that's where the per-connection pipeline win comes + * from. + */ + suspend fun submit( + event: Event, + onComplete: (IEventStore.InsertOutcome) -> Unit, + ) { + ensureWriterStarted() + incoming.send(Submission(event, onComplete)) + } + + private fun ensureWriterStarted() { + if (writerStarted) return + synchronized(startLock) { + if (writerStarted) return + scope.launch { drainLoop() } + writerStarted = true + } + } + + private suspend fun drainLoop() { + val batch = ArrayList(maxBatch) + val events = ArrayList(maxBatch) + try { + while (true) { + // Block for the first item — anything else would be a + // hot loop. Once we have one, drain greedily without + // blocking so back-to-back publishes coalesce into a + // single transaction. + batch.add(incoming.receive()) + while (batch.size < maxBatch) { + val next = incoming.tryReceive().getOrNull() ?: break + batch.add(next) + } + events.clear() + for (sub in batch) events.add(sub.event) + + val outcomes = + try { + store.batchInsert(events) + } catch (e: Throwable) { + Log.w("IngestQueue") { "batchInsert failed for ${batch.size} events: ${e.message}" } + val reason = e.message ?: e::class.simpleName ?: "insert failed" + List(batch.size) { IEventStore.InsertOutcome.Rejected(reason) } + } + + for (i in batch.indices) { + val sub = batch[i] + val outcome = + outcomes.getOrNull(i) + ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + try { + sub.onComplete(outcome) + } catch (e: Throwable) { + // A misbehaving callback must not poison the + // writer loop; the outcome is delivered + // best-effort. + Log.w("IngestQueue") { "onComplete threw: ${e.message}" } + } + } + batch.clear() + } + } catch (_: kotlinx.coroutines.channels.ClosedReceiveChannelException) { + // Normal shutdown via close(). + } + } + + /** + * Stop accepting new submissions and cancel the writer. + * In-flight submissions whose batch hadn't started yet may never + * receive their callback — the WebSocket on the other side is + * also being torn down in that case, so the OK reply has nowhere + * to go anyway. + */ + override fun close() { + incoming.close() + scope.cancel() + } + + companion object { + /** + * Cap per batch. Sized to keep per-batch latency low (each + * transaction holds the SQLite writer mutex; over-large + * batches starve other writers and hurt p99 publish latency). + * 64 events at ~0.2 ms per insert ≈ 13 ms held — well under + * a perceptible UI tick. + */ + const val DEFAULT_MAX_BATCH: Int = 64 + + /** + * In-flight cap before [submit] suspends. With ~5–10× group + * commit speed-up over the single-event path, one batch + * cycle is on the order of milliseconds, so a 1024-deep + * queue tolerates short bursts (a publisher dumping a + * thousand notes) without blocking the WS pump. + */ + const val DEFAULT_CAPACITY: Int = 1024 + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt index 00d2436f9..912c1e4d0 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt @@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.server import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.onSubscription @@ -39,6 +40,7 @@ import kotlinx.coroutines.flow.onSubscription */ class LiveEventStore( private val store: IEventStore, + private val ingest: IngestQueue, ) { private val newEventStream = MutableSharedFlow( @@ -47,9 +49,47 @@ class LiveEventStore( onBufferOverflow = BufferOverflow.DROP_LATEST, // Default behavior ) + /** + * Fire-and-forget enqueue: hand [event] to the [IngestQueue] and + * fire [onComplete] once the writer's batch has a per-row + * decision. On `Accepted` the live stream is also emitted to so + * subscribers see the event. Suspends only when the ingest queue + * is full (backpressure). + */ + suspend fun submit( + event: Event, + onComplete: (IEventStore.InsertOutcome) -> Unit, + ) { + ingest.submit(event) { outcome -> + if (outcome is IEventStore.InsertOutcome.Accepted) { + newEventStream.tryEmit(event) + } + onComplete(outcome) + } + } + + /** + * Suspending insert kept for callers that don't care about + * pipelining (tests, scripted paths). Routes through the same + * [IngestQueue] as [submit] so the batch write path is exercised + * even by tests that prefer a sequential `insert` API. + * Throws on rejection so callers can `try` around it the way the + * old API did. + */ suspend fun insert(event: Event) { - store.insert(event) - newEventStream.tryEmit(event) + val done = CompletableDeferred() + submit(event) { outcome -> + when (outcome) { + IEventStore.InsertOutcome.Accepted -> { + done.complete(Unit) + } + + is IEventStore.InsertOutcome.Rejected -> { + done.completeExceptionally(IllegalStateException(outcome.reason)) + } + } + } + done.await() } suspend fun query( diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index f72c9e7b2..3a8c34f44 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -42,11 +42,20 @@ class NostrServer( private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy }, private val parentContext: CoroutineContext = SupervisorJob(), ) : AutoCloseable { - private val subStore = LiveEventStore(store) - /** Scope for all subscriptions. */ private val scope = CoroutineScope(parentContext + SupervisorJob()) + /** + * Group-commit writer shared across every connected session. + * Sessions hand off EVENT publishes here instead of awaiting + * [IEventStore.insert] inline; the queue coalesces back-to-back + * publishes into a single SQLite transaction. See [IngestQueue] + * for the OK ordering and durability semantics. + */ + private val ingest = IngestQueue(store, parentContext) + + private val subStore = LiveEventStore(store, ingest) + /** Active client sessions keyed by an opaque connection id. */ private val connections = LargeCache() @@ -95,6 +104,7 @@ class NostrServer( override fun close() { connections.forEach { _, session -> session.cancelAllSubscriptions() } connections.clear() + ingest.close() scope.cancel() store.close() } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index d72ff491c..5c2c56f33 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -34,6 +34,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CountCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd @@ -129,11 +130,29 @@ class RelaySession( return } + // Fire-and-forget: hand the event to the group-commit writer + // and continue reading from the WebSocket without waiting on + // SQLite. The OK frame is sent from the writer's callback, + // possibly out of arrival order — NIP-01 pairs OKs to events + // by id, so reordering is fine. try { - store.insert(cmd.event) - send(OkMessage(cmd.event.id, true, "")) - } catch (e: Exception) { - send(OkMessage(cmd.event.id, false, e.message ?: e::class.simpleName ?: "unkown error")) + store.submit(cmd.event) { outcome -> + when (outcome) { + IEventStore.InsertOutcome.Accepted -> { + send(OkMessage(cmd.event.id, true, "")) + } + + is IEventStore.InsertOutcome.Rejected -> { + send(OkMessage(cmd.event.id, false, outcome.reason)) + } + } + } + } catch (_: kotlinx.coroutines.channels.ClosedSendChannelException) { + // Server is shutting down — the queue is closed. Reply + // with a transient failure so a client re-trying against + // the next instance gets a sane signal; the WS itself is + // about to be torn down by the server-stop path. + send(OkMessage(cmd.event.id, false, "error: relay shutting down")) } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt index d376b8e7d..8984e2fc5 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt @@ -41,6 +41,42 @@ interface IEventStore : AutoCloseable { suspend fun transaction(body: ITransaction.() -> Unit) + /** + * Per-row outcome from [batchInsert]. The OK frame on the wire is + * built from this — `Accepted` becomes `OK true`, `Rejected.reason` + * becomes the false reason. NIP-01 says OK pairs to its EVENT by + * id, not by order, so callers may dispatch outcomes in any order. + */ + sealed class InsertOutcome { + data object Accepted : InsertOutcome() + + data class Rejected( + val reason: String, + ) : InsertOutcome() + } + + /** + * Bulk insert in a single transaction with per-row error isolation. + * Returns one outcome per input event in the same order. + * + * Implementations must isolate per-row failures so one bad event + * doesn't roll back the others (SQLite uses SAVEPOINTs). If the + * outer commit itself fails, every entry in the returned list is + * `Rejected` with the commit-failure reason. + * + * Default impl runs each insert in its own transaction — correct + * but loses the group-commit win. SQLite overrides this. + */ + suspend fun batchInsert(events: List): List = + events.map { event -> + try { + insert(event) + InsertOutcome.Accepted + } catch (e: Throwable) { + InsertOutcome.Rejected(e.message ?: e::class.simpleName ?: "insert failed") + } + } + suspend fun query(filter: Filter): List suspend fun query(filters: List): List diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt index 1def6bf4f..eedca3371 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt @@ -96,6 +96,51 @@ class ObservableEventStore( _changes.emit(StoreChange.Insert(event)) } + override suspend fun batchInsert(events: List): List { + // Split into ephemeral (no-store) and persistable, delegate the + // persistable subset to the inner store's batched path so we + // keep the group-commit win, then merge outcomes back in input + // order. Already-expired ephemerals are dropped (matching + // [insert]). Accepted events are emitted on [_changes] only + // after the inner batch returns, so a commit failure that + // converts everything to Rejected suppresses the emits. + if (events.isEmpty()) return emptyList() + + val outcomes = arrayOfNulls(events.size) + val persistableIndices = ArrayList(events.size) + val persistable = ArrayList(events.size) + for (i in events.indices) { + val event = events[i] + if (event.kind.isEphemeral()) { + outcomes[i] = + if (event.isExpired()) { + IEventStore.InsertOutcome.Rejected("blocked: Cannot insert an expired event") + } else { + IEventStore.InsertOutcome.Accepted + } + } else { + persistableIndices.add(i) + persistable.add(event) + } + } + + if (persistable.isNotEmpty()) { + val innerOutcomes = inner.batchInsert(persistable) + for (j in persistable.indices) { + outcomes[persistableIndices[j]] = innerOutcomes[j] + } + } + + for (i in events.indices) { + if (outcomes[i] is IEventStore.InsertOutcome.Accepted) { + _changes.emit(StoreChange.Insert(events[i])) + } + } + + @Suppress("UNCHECKED_CAST") + return outcomes.toList() as List + } + override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) { val accepted = ArrayList() inner.transaction { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt index ef9510611..097c5672a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt @@ -43,6 +43,8 @@ class EventStore( override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = store.transaction(body) + override suspend fun batchInsert(events: List) = store.batchInsertEvents(events) + override suspend fun query(filter: Filter) = store.query(filter) override suspend fun query(filters: List) = store.query(filters) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt index 2da5cf49c..78b29bffd 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt @@ -201,6 +201,63 @@ class SQLiteEventStore( } } + /** + * Group-commit batch insert with per-row error isolation via + * SAVEPOINTs. Acquires the writer mutex once and wraps every + * inserts in a single outer transaction so the WAL append + sync + * cost is paid once for the whole batch. + * + * Per-row contract: + * - Validation errors (expired) and per-row INSERT failures + * (UNIQUE constraint, etc.) ROLLBACK only that row's savepoint; + * other rows commit. + * - Ephemeral kinds are accepted without writing — the live + * stream still surfaces them; persistence is intentionally a + * no-op per NIP-01. + * + * Outer-commit failure throws; the caller treats every entry as + * `Rejected` (this is what the IEventStore contract documents). + */ + suspend fun batchInsertEvents(events: List): List { + if (events.isEmpty()) return emptyList() + val outcomes = ArrayList(events.size) + pool.useWriter { db -> + db.transaction { + events.forEachIndexed { i, event -> + outcomes.add(insertWithSavepoint(event, i, this)) + } + } + } + return outcomes + } + + private fun insertWithSavepoint( + event: Event, + index: Int, + db: SQLiteConnection, + ): IEventStore.InsertOutcome { + if (event.isExpired()) { + return IEventStore.InsertOutcome.Rejected("blocked: Cannot insert an expired event") + } + if (event.kind.isEphemeral()) return IEventStore.InsertOutcome.Accepted + + val sp = "ev$index" + db.execSQL("SAVEPOINT $sp") + return try { + innerInsertEvent(event, db) + db.execSQL("RELEASE SAVEPOINT $sp") + IEventStore.InsertOutcome.Accepted + } catch (e: Throwable) { + // Roll back just this row, then release the (now empty) + // savepoint frame so the next iteration's BEGIN works. + // Both calls are individually try/catch'd because a failed + // ROLLBACK shouldn't mask the original cause. + runCatching { db.execSQL("ROLLBACK TRANSACTION TO SAVEPOINT $sp") } + runCatching { db.execSQL("RELEASE SAVEPOINT $sp") } + IEventStore.InsertOutcome.Rejected(e.message ?: e::class.simpleName ?: "insert failed") + } + } + inner class Transaction( val db: SQLiteConnection, ) : IEventStore.ITransaction { From 289bc4bd5cd949192446c06f09f084341b233aca Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 23:00:30 +0000 Subject: [PATCH 3/5] =?UTF-8?q?perf(quartz):=20Tier=203=20=E2=80=94=20para?= =?UTF-8?q?llel=20Schnorr=20verify=20in=20IngestQueue?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the last item on the event-ingestion-batching plan: signature verification no longer serialises on each connection's WebSocket pump. Instead, IngestQueue takes a `verify: ((Event) -> Boolean)?` hook and fan-outs the per-batch verify across Dispatchers.Default (`coroutineScope { events.map { async { verify(it) } }.awaitAll() }`) before opening the SQLite transaction. Failed verifies pre-mark Rejected and skip the insert. Wiring: - NostrServer takes `parallelVerify: Boolean = false` (opt-in to preserve existing behaviour for direct library users). - geode.Relay forwards a matching flag. - Main.kt enables it whenever signature checking is on (config `[options].parallel_verify = true`, default true), and when so, composePolicy is told to skip VerifyPolicy from the chain to avoid double-verifying every event. - New CLI escape hatch `--no-parallel-verify` for the legacy path. Bench: adds publishGroupCommitSingleClient (sequential publish-and- confirm; 500 EPS regression floor for the synchronous path) — the companion to the existing pipelined bench that exercises the group-commit + parallel-verify wins. Plan doc updated to describe what shipped (batchInsert + SAVEPOINTs in Tier 1, IngestQueue mechanics in Tier 2, the verify hook in Tier 3) and to drop the obsolete `synchronous=NORMAL` confirmation note — the project ships `synchronous=OFF` and intentionally keeps that. --- geode/config.example.toml | 6 + .../2026-05-07-event-ingestion-batching.md | 95 ++++++++----- .../kotlin/com/vitorpamplona/geode/Main.kt | 21 ++- .../kotlin/com/vitorpamplona/geode/Relay.kt | 12 ++ .../vitorpamplona/geode/config/RelayConfig.kt | 12 ++ .../vitorpamplona/geode/perf/LoadBenchmark.kt | 46 +++++++ .../nip01Core/relay/server/IngestQueue.kt | 126 ++++++++++++++---- .../nip01Core/relay/server/NostrServer.kt | 23 +++- 8 files changed, 281 insertions(+), 60 deletions(-) diff --git a/geode/config.example.toml b/geode/config.example.toml index 3d706999b..2a4d8ce5a 100644 --- a/geode/config.example.toml +++ b/geode/config.example.toml @@ -53,6 +53,12 @@ file = "/var/lib/geode/events.db" # only for trusted-input scenarios (test fixtures, mirror replays). # verify_signatures = true +# Run signature verification in parallel inside the IngestQueue +# (across all CPU cores) instead of serially on each connection's +# WebSocket pump. Default: true. Set false to fall back to the +# legacy in-policy verify path. +# parallel_verify = true + # Require clients to NIP-42 AUTH before REQ/EVENT/COUNT. require_auth = false diff --git a/geode/plans/2026-05-07-event-ingestion-batching.md b/geode/plans/2026-05-07-event-ingestion-batching.md index 561d1dd08..08a8d88f7 100644 --- a/geode/plans/2026-05-07-event-ingestion-batching.md +++ b/geode/plans/2026-05-07-event-ingestion-batching.md @@ -38,56 +38,89 @@ contention, not WS throughput). ### Tier 1 — SQLite WAL + group commit (cheap win) -Confirm `PRAGMA journal_mode=WAL` + `PRAGMA synchronous=NORMAL` on the -event-store DB; group commits across the writer mutex's hold window. -Today each insert is its own transaction. Wrap N inserts (or a 5 ms -budget, whichever first) in a single transaction managed by the writer -coroutine. +WAL is already on (`PRAGMA journal_mode=WAL`). The pool runs with +`PRAGMA synchronous=OFF`, which is one notch more permissive than +the originally-sketched `synchronous=NORMAL` — we keep it as-is +because the project already accepted the OS-crash trade-off there. -Because OK reflects acceptance not durability, each row can fan an OK -as soon as the per-row INSERT statement returns inside the -transaction — we do not need to wait for the batch's commit. The -fsync is hidden from the publisher latency budget entirely. +Group commit is implemented via a new `IEventStore.batchInsert`: +the SQLite override holds the writer mutex once and wraps N events +in one `BEGIN IMMEDIATE … COMMIT`. Per-row error isolation uses +SAVEPOINTs so one bad event (expired, duplicate id) doesn't roll +back the good ones — just that row reports `Rejected`. -Implementation lives in quartz's `EventStore` / `SQLiteConnectionPool`, -not geode — but geode owns the benchmark and validates the gain. +OKs fire as soon as each row's outcome is known inside the writer +batch, not waiting for fsync (per the OK-semantics constraint above). + +Implementation lives in `quartz/nip01Core/store/sqlite/SQLiteEventStore.batchInsertEvents`, +exposed through `IEventStore.batchInsert` and consumed by the new +`IngestQueue` (Tier 2 below). Expected: **~5–10× write throughput** on a fast SSD. SQLite group commit is well-trodden territory (nostr-rs-relay, strfry both do it). ### Tier 2 — pipelined OK over multiple in-flight EVENTs -`RelaySession.receive` is currently single-flight: one EVENT in, -process, OK out, next EVENT. Allow a connection to push N EVENTs -concurrently and dispatch them to a per-connection ingest pipeline. +`RelaySession.receive` was single-flight: one EVENT in, process, OK +out, next EVENT. With Tier 2 the connection's pump posts to the +shared `IngestQueue` and returns immediately — the WS pump moves +straight to the next frame. -A `Channel` with capacity = `INGEST_PIPELINE_DEPTH` per -connection, drained by a coroutine that feeds the group-commit writer -above. OKs go straight to `outQueue.send()` the moment each row -returns from INSERT — no ordering bookkeeping needed, since the OK -frame already carries the event id and the spec doesn't require -order. A pipelined publisher keying on event id will pair replies -correctly. +`IngestQueue` (one per `NostrServer`) holds a bounded +`Channel` (capacity = 1024 per the `DEFAULT_CAPACITY` +constant) drained by a single writer coroutine. The writer pulls +the first item to start a batch then `tryReceive`-drains everything +else queued (up to 64 — `DEFAULT_MAX_BATCH`), feeds the whole batch +to `IEventStore.batchInsert`, and dispatches each row's +`onComplete` callback as soon as the batch returns. The callback +turns into the `OK` frame at the WS layer. + +OKs are not order-preserving (per the constraints above). The +writer coroutine starts lazily on first `submit` so subscription- +only sessions don't pay for it and don't perturb `Dispatchers.Default` +scheduling. Expected: hides verify+insert latency behind the next EVENT's parse, gets us closer to network-bound throughput. ### Tier 3 — eager Schnorr verify off the writer thread -`VerifyPolicy` is in the policy stack and runs synchronously on -`receive`. Move it into the ingest pipeline so verification of EVENT N+1 -runs concurrently with the SQLite commit of EVENT N. secp256k1 verify -is parallelisable; the writer should never block on it. +`VerifyPolicy` ran synchronously on `receive`, serialising verify +on each connection's pump coroutine. With Tier 3, `IngestQueue` +takes a `verify: ((Event) -> Boolean)?` hook; when set, the writer +fan-outs a `coroutineScope { events.map { async(Default) { verify(it) } }.awaitAll() }` +on each batch before opening the SQLite transaction. Failed +verifies pre-mark `Rejected` and skip the insert. + +Wired through `NostrServer(parallelVerify = ...)` and +`geode.Relay(parallelVerify = ...)`, controlled by +`[options].parallel_verify` in the relay config (default `true`) +and `--no-parallel-verify` on the CLI. Operators that flip it on +must omit `VerifyPolicy` from their policy chain — `Main.kt` does +this automatically; `composePolicy` is told to skip the +`VerifyPolicy` piece when `parallelVerify` is true. Internal +direct callers of `NostrServer` (tests, library users) are +opt-in: the flag defaults to `false` to keep existing +`VerifyPolicy`-in-chain semantics unchanged. + +Expected: ≈CPU_COUNT× verify-step speed-up on burst publishes +from a single connection, where verify was previously serial on +that pump. ## How to verify -Add to `geode.perf.LoadBenchmark`: +`geode.perf.LoadBenchmark` carries the perf tests: -- `publishGroupCommitSingleClient` — same workload as the current - single-client benchmark, asserts >5000 EPS. -- `publishPipelinedSingleClient` — sends 100 EVENTs without awaiting - intermediate OKs; measures end-to-end throughput and verifies that - every event id receives exactly one OK (in any order). +- `publishGroupCommitSingleClient` — sequential publish-and-confirm + on one connection (the same shape as the original + `publishThroughputSingleClient`). Synchronous publishing means + batch size is always 1, so this case shows per-event SQLite tx + cost rather than the group-commit win — kept as a 500-EPS floor + to catch regressions from the rewrite. +- `publishPipelinedSingleClient` — bursts 10 000 EVENTs back-to- + back without awaiting intermediate OKs; verifies end-to-end + throughput and that every event id receives exactly one OK (in + any order). This is where Tier 1 + Tier 2 both light up. Existing benchmarks stay as the regression floor. diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index e7b7a97b0..19648f07c 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -83,6 +83,12 @@ fun main(args: Array) { // opts out (CLI `--no-verify` or `[options].verify_signatures = false` // in the config). val verifySigs = !a.flag("--no-verify") && config.options.verify_signatures + // Parallel verify is on whenever signature checking is on; the + // IngestQueue handles it instead of VerifyPolicy. Operators can + // force the legacy in-policy path with `--no-parallel-verify` or + // `[options].parallel_verify = false`. + val parallelVerify = + verifySigs && !a.flag("--no-parallel-verify") && config.options.parallel_verify // Advertised URL: explicit `info.relay_url` wins, then build from // host/port/path. 0.0.0.0 bind → 127.0.0.1 in the URL so NIP-42 @@ -98,11 +104,22 @@ fun main(args: Array) { val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl) val policyBuilder: () -> IRelayPolicy = { - composePolicy(config, advertisedUrl, requireAuth, verifySigs) + // When parallel verify is enabled the IngestQueue runs + // Schnorr verify off the WS pump, so the policy chain skips + // VerifyPolicy to avoid double-verifying every event. + composePolicy(config, advertisedUrl, requireAuth, verifySigs && !parallelVerify) } val stateFile = config.admin.state_file?.let { File(it) } - val relay = Relay(advertisedUrl, store, info, policyBuilder, stateFile = stateFile) + val relay = + Relay( + advertisedUrl, + store, + info, + policyBuilder, + stateFile = stateFile, + parallelVerify = parallelVerify, + ) // Frame cap honors max_ws_frame_bytes when set; max_ws_message_bytes // is treated as the same cap (Ktor's WebSockets plugin only exposes // a single per-frame limit; multi-frame messages remain unbounded). diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt index 48c339f4e..6a8436103 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt @@ -73,6 +73,17 @@ class Relay( * everything in memory only — fine for tests. */ stateFile: File? = null, + /** + * Run Schnorr signature verification in parallel inside the + * [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue] + * instead of serially in the policy chain. Enables the Tier-3 + * win in `geode/plans/2026-05-07-event-ingestion-batching.md`. + * + * When set, callers MUST omit `VerifyPolicy` from [policyBuilder] + * — having both verifies the same event twice for no benefit. + * `Main.kt` skips `VerifyPolicy` when this flag is on. + */ + parallelVerify: Boolean = false, ) : AutoCloseable { private val stateStore: RelayStateStore? = stateFile?.let { RelayStateStore(it) } @@ -158,6 +169,7 @@ class Relay( if (user === EmptyPolicy) BanListPolicy(banStore) else user + BanListPolicy(banStore) }, parentContext, + parallelVerify = parallelVerify, ) /** 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 1417ec61f..1f4ff4441 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt @@ -141,6 +141,18 @@ data class RelayConfig( * for trusted-input scenarios (test fixtures, mirror replays). */ val verify_signatures: Boolean = true, + /** + * Run signature verification in parallel inside the IngestQueue + * (CPU fan-out across `Dispatchers.Default`) instead of serially + * on each connection's WebSocket pump. Tier-3 of the + * `event-ingestion-batching` plan. Wins scale with how many + * EVENTs a single connection sends back-to-back: ~CPU_COUNT× + * verify-step speed-up on burst publishes. Set false to keep + * the legacy in-policy verify path. + * + * Only takes effect when [verify_signatures] is also true. + */ + val parallel_verify: Boolean = true, ) data class LimitsSection( 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 6f8e05fdc..71ce65fe5 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -363,6 +363,52 @@ class LoadBenchmark { } } + /** + * Same workload as [publishThroughputSingleClient] (sequential + * publish-and-confirm on one connection) — kept as a regression + * floor for the group-commit code path. Synchronous publishes + * never coalesce in the writer (batch size is always 1), so the + * EPS here measures per-event SQLite tx cost. The pipelined win + * shows up in [publishPipelinedSingleClient]. + */ + @Test + fun publishGroupCommitSingleClient() = + benchmark("publish group-commit single client") { + runBenchmarkServer { server, http -> + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope) + try { + val signer = NostrSignerSync(KeyPair()) + val relayUrl = server.url.normalizeRelayUrl() + + val n = 10_000 + var ok = 0 + val elapsed = + measureTime { + runBlocking { + repeat(n) { i -> + val event = signer.sign(TextNoteEvent.build("group-commit $i")) + if (client.publishAndConfirm(event, setOf(relayUrl))) ok++ + } + } + } + val eps = (n * 1000.0) / elapsed.inWholeMilliseconds + println( + "events=$n ok=$ok elapsedMs=${elapsed.inWholeMilliseconds} eps=${"%.0f".format(eps)}", + ) + check(ok == n) { "expected all $n events accepted, got $ok" } + // Floor: the pre-batching baseline was ~760 EPS + // single-client (see plan). Anything below 500 + // means the group-commit / ingest-queue rewrite + // regressed the synchronous path. + check(eps > 500) { "synchronous EPS $eps fell below the 500 floor" } + } finally { + client.disconnect() + scope.cancel() + } + } + } + /** * One publisher, N subscribers. Publishes one EVENT and measures * fan-out latency: time from publish to last subscriber receiving. diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt index 251c1e781..b93b236ae 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt @@ -24,9 +24,13 @@ import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch import kotlin.coroutines.CoroutineContext @@ -73,6 +77,25 @@ class IngestQueue( parentContext: CoroutineContext, private val maxBatch: Int = DEFAULT_MAX_BATCH, capacity: Int = DEFAULT_CAPACITY, + /** + * Optional pre-insert validator. When set, the writer runs each + * batch through this hook in parallel before opening the SQLite + * transaction. Events that return `false` skip the insert and are + * reported as Rejected with [verifyRejectionReason]. + * + * The intended use is Schnorr signature verification: an event's + * `verify()` is CPU-bound and parallelisable, so spreading a + * batch's verifies across [Dispatchers.Default] threads is + * straight throughput. Hook fires off the WS pump so a single + * publisher streaming many EVENTs doesn't serialise verify on + * one connection's read coroutine. + * + * Default `null` skips this stage — callers that already verify + * inside their `IRelayPolicy` chain should leave it null to avoid + * double-verify. + */ + private val verify: ((Event) -> Boolean)? = null, + private val verifyRejectionReason: String = "invalid: bad signature or id", ) : AutoCloseable { /** * One outstanding ingest request: the event to insert plus the @@ -130,7 +153,6 @@ class IngestQueue( private suspend fun drainLoop() { val batch = ArrayList(maxBatch) - val events = ArrayList(maxBatch) try { while (true) { // Block for the first item — anything else would be a @@ -142,32 +164,8 @@ class IngestQueue( val next = incoming.tryReceive().getOrNull() ?: break batch.add(next) } - events.clear() - for (sub in batch) events.add(sub.event) - val outcomes = - try { - store.batchInsert(events) - } catch (e: Throwable) { - Log.w("IngestQueue") { "batchInsert failed for ${batch.size} events: ${e.message}" } - val reason = e.message ?: e::class.simpleName ?: "insert failed" - List(batch.size) { IEventStore.InsertOutcome.Rejected(reason) } - } - - for (i in batch.indices) { - val sub = batch[i] - val outcome = - outcomes.getOrNull(i) - ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") - try { - sub.onComplete(outcome) - } catch (e: Throwable) { - // A misbehaving callback must not poison the - // writer loop; the outcome is delivered - // best-effort. - Log.w("IngestQueue") { "onComplete threw: ${e.message}" } - } - } + processBatch(batch) batch.clear() } } catch (_: kotlinx.coroutines.channels.ClosedReceiveChannelException) { @@ -175,6 +173,82 @@ class IngestQueue( } } + private suspend fun processBatch(batch: List) { + // Tier 3: parallel pre-insert verify. Each batch entry that + // fails verification is pre-marked Rejected and excluded from + // the SQLite transaction. The remaining (verified) events go + // through batchInsert in original order; we re-stitch outcomes + // back to the full batch by index. + val verifyResults: BooleanArray? = + verify?.let { hook -> + coroutineScope { + batch + .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } + .awaitAll() + .toBooleanArray() + } + } + + val toInsert: List + val insertIndices: IntArray + if (verifyResults == null) { + toInsert = batch.map { it.event } + insertIndices = IntArray(batch.size) { it } + } else { + val accepted = ArrayList(batch.size) + val mapping = ArrayList(batch.size) + for (i in batch.indices) { + if (verifyResults[i]) { + accepted.add(batch[i].event) + mapping.add(i) + } + } + toInsert = accepted + insertIndices = mapping.toIntArray() + } + + val insertOutcomes: List = + if (toInsert.isEmpty()) { + emptyList() + } else { + try { + store.batchInsert(toInsert) + } catch (e: Throwable) { + Log.w("IngestQueue") { "batchInsert failed for ${toInsert.size} events: ${e.message}" } + val reason = e.message ?: e::class.simpleName ?: "insert failed" + List(toInsert.size) { IEventStore.InsertOutcome.Rejected(reason) } + } + } + + // Build a per-batch-index outcome array. + val finalOutcomes = arrayOfNulls(batch.size) + for (j in insertIndices.indices) { + finalOutcomes[insertIndices[j]] = insertOutcomes.getOrNull(j) + ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + } + if (verifyResults != null) { + for (i in batch.indices) { + if (!verifyResults[i]) { + finalOutcomes[i] = IEventStore.InsertOutcome.Rejected(verifyRejectionReason) + } + } + } + + for (i in batch.indices) { + val sub = batch[i] + val outcome = + finalOutcomes[i] + ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + try { + sub.onComplete(outcome) + } catch (e: Throwable) { + // A misbehaving callback must not poison the writer + // loop; the outcome is delivered best-effort. + Log.w("IngestQueue") { "onComplete threw: ${e.message}" } + } + } + } + /** * Stop accepting new submissions and cancel the writer. * In-flight submissions whose batch hadn't started yet may never diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index 3a8c34f44..54c23fbe0 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -20,6 +20,8 @@ */ package com.vitorpamplona.quartz.nip01Core.relay.server +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.crypto.verify import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.utils.cache.LargeCache @@ -36,11 +38,18 @@ import kotlin.coroutines.CoroutineContext * * @param store The [IEventStore] backing this relay. * @param policyBuilder Controls requirements for relay commands. + * @param parallelVerify When `true`, Schnorr verification runs in + * parallel inside the [IngestQueue] (one async per event, dispatched + * on `Dispatchers.Default`) rather than serially on the WS pump + * coroutine inside [VerifyPolicy]. Callers that flip this on should + * *omit* `VerifyPolicy` from their [policyBuilder] chain to avoid + * double-verifying. */ class NostrServer( private val store: IEventStore, private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy }, private val parentContext: CoroutineContext = SupervisorJob(), + parallelVerify: Boolean = false, ) : AutoCloseable { /** Scope for all subscriptions. */ private val scope = CoroutineScope(parentContext + SupervisorJob()) @@ -52,7 +61,12 @@ class NostrServer( * publishes into a single SQLite transaction. See [IngestQueue] * for the OK ordering and durability semantics. */ - private val ingest = IngestQueue(store, parentContext) + private val ingest = + IngestQueue( + store = store, + parentContext = parentContext, + verify = if (parallelVerify) ::verifyEvent else null, + ) private val subStore = LiveEventStore(store, ingest) @@ -114,4 +128,11 @@ class NostrServer( */ @Deprecated("Use close() instead", replaceWith = ReplaceWith("close()")) fun shutdown() = close() + + private companion object { + // Function reference (`Event::verify`) wrapper so the + // ingest hook keeps a single instance per server rather + // than allocating a fresh lambda on every call site. + private fun verifyEvent(event: Event): Boolean = event.verify() + } } From 7430c2aa176babf880d95b006ddcb27f9421d508 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 23:16:52 +0000 Subject: [PATCH 4/5] =?UTF-8?q?fix(quartz):=20audit=20fixes=20=E2=80=94=20?= =?UTF-8?q?VerifyAuthOnlyPolicy=20+=20small=20wins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-audit of the event-ingestion-batching changes turned up one real bug + a handful of cleanups. Fix: AUTH events skipped signature verification when parallelVerify=true. Previous commit dropped VerifyPolicy from the policy chain to avoid double-verifying EVENTs (the IngestQueue does those off-thread). But VerifyPolicy.accept(AuthCmd) was the only thing checking AUTH signatures — FullAuthPolicy verifies challenge / relay / expiry but trusts the sig. Removing VerifyPolicy let a forged event mark a pubkey as authenticated. Split VerifyPolicy into a parameterised base class (VerifyEventsAndAuthPolicy) with two singletons: - VerifyPolicy: verifies both EVENT and AUTH (existing default). - VerifyAuthOnlyPolicy: verifies AUTH only — use when an IngestQueue does parallel EVENT verify, since AUTH commands bypass the queue entirely. Geode's composePolicy now selects VerifyAuthOnlyPolicy when parallelVerify is on, keeping AUTH signature checks intact while still letting EVENT verify run in parallel on the writer's CPU fan-out. Cleanups (no behavior change): - IngestQueue.processBatch was ~70 lines; split into verifyBatch / runInsertStage / dispatchOutcomes. - Single-event verify shortcut: skip the coroutineScope + async dance for batch-of-1 (the common low-load case) and call the hook directly. - Hoisted the "internal error: missing outcome" Rejected sentinel to a companion `missingOutcome` constant. - Imported ClosedSendChannelException / ClosedReceiveChannelException instead of using the fully-qualified form inline. - Dropped NostrServer's verifyEvent companion wrapper — direct lambda is just as cheap and the "single instance" comment was inaccurate. - ObservableEventStore.batchInsert now uses requireNoNulls() rather than @Suppress("UNCHECKED_CAST"), since every index is provably populated. --- .../kotlin/com/vitorpamplona/geode/Main.kt | 13 ++- .../nip01Core/relay/server/IngestQueue.kt | 94 ++++++++++++------- .../nip01Core/relay/server/NostrServer.kt | 10 +- .../nip01Core/relay/server/RelaySession.kt | 3 +- .../relay/server/policies/VerifyPolicy.kt | 32 ++++++- .../nip01Core/store/ObservableEventStore.kt | 7 +- 6 files changed, 103 insertions(+), 56 deletions(-) diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 19648f07c..a731890e8 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.KindAllowDenyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PubkeyAllowDenyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RejectFutureEventsPolicy +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyAuthOnlyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore @@ -104,10 +105,7 @@ fun main(args: Array) { val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl) val policyBuilder: () -> IRelayPolicy = { - // When parallel verify is enabled the IngestQueue runs - // Schnorr verify off the WS pump, so the policy chain skips - // VerifyPolicy to avoid double-verifying every event. - composePolicy(config, advertisedUrl, requireAuth, verifySigs && !parallelVerify) + composePolicy(config, advertisedUrl, requireAuth, verifySigs, parallelVerify) } val stateFile = config.admin.state_file?.let { File(it) } @@ -168,6 +166,7 @@ private fun composePolicy( advertisedUrl: com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl, requireAuth: Boolean, verifySigs: Boolean, + parallelVerify: Boolean, ): IRelayPolicy { val pieces = mutableListOf() @@ -188,7 +187,11 @@ private fun composePolicy( } if (verifySigs) { - pieces += VerifyPolicy + // When parallel verify is on, the IngestQueue handles EVENT + // verification on the writer's CPU fan-out — but AUTH events + // bypass the queue, so we still need the policy chain to + // verify those. `VerifyAuthOnlyPolicy` does exactly that. + pieces += if (parallelVerify) VerifyAuthOnlyPolicy else VerifyPolicy } return pieces.fold(EmptyPolicy) { acc, p -> diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt index b93b236ae..5b4876693 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt @@ -30,6 +30,7 @@ import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.ClosedReceiveChannelException import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch import kotlin.coroutines.CoroutineContext @@ -168,43 +169,53 @@ class IngestQueue( processBatch(batch) batch.clear() } - } catch (_: kotlinx.coroutines.channels.ClosedReceiveChannelException) { + } catch (_: ClosedReceiveChannelException) { // Normal shutdown via close(). } } private suspend fun processBatch(batch: List) { - // Tier 3: parallel pre-insert verify. Each batch entry that - // fails verification is pre-marked Rejected and excluded from - // the SQLite transaction. The remaining (verified) events go - // through batchInsert in original order; we re-stitch outcomes - // back to the full batch by index. - val verifyResults: BooleanArray? = - verify?.let { hook -> - coroutineScope { - batch - .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } - .awaitAll() - .toBooleanArray() - } - } + val verifyResults = verifyBatch(batch) + val finalOutcomes = runInsertStage(batch, verifyResults) + dispatchOutcomes(batch, finalOutcomes) + } - val toInsert: List - val insertIndices: IntArray - if (verifyResults == null) { - toInsert = batch.map { it.event } - insertIndices = IntArray(batch.size) { it } - } else { - val accepted = ArrayList(batch.size) - val mapping = ArrayList(batch.size) - for (i in batch.indices) { - if (verifyResults[i]) { - accepted.add(batch[i].event) - mapping.add(i) - } + /** + * Tier 3: per-row verify. Returns `null` when no [verify] hook is + * configured (skip the stage entirely). For multi-event batches + * each verify runs as its own `async(Default)` so they spread + * across CPU cores; single-event batches short-circuit to a + * direct call to avoid coroutine-scope overhead. + */ + private suspend fun verifyBatch(batch: List): BooleanArray? { + val hook = verify ?: return null + if (batch.size == 1) return BooleanArray(1) { hook(batch[0].event) } + return coroutineScope { + batch + .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } + .awaitAll() + .toBooleanArray() + } + } + + /** + * Run the SQLite transaction for the verified subset of [batch] + * and stitch outcomes back to a per-batch-index array. Failed + * verifies pre-mark `Rejected` and skip the insert. A whole-batch + * commit failure converts every persisted entry to `Rejected` + * with the throw message. + */ + private suspend fun runInsertStage( + batch: List, + verifyResults: BooleanArray?, + ): Array { + val toInsert = ArrayList(batch.size) + val insertIndices = ArrayList(batch.size) + for (i in batch.indices) { + if (verifyResults == null || verifyResults[i]) { + toInsert.add(batch[i].event) + insertIndices.add(i) } - toInsert = accepted - insertIndices = mapping.toIntArray() } val insertOutcomes: List = @@ -220,11 +231,10 @@ class IngestQueue( } } - // Build a per-batch-index outcome array. val finalOutcomes = arrayOfNulls(batch.size) for (j in insertIndices.indices) { - finalOutcomes[insertIndices[j]] = insertOutcomes.getOrNull(j) - ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + finalOutcomes[insertIndices[j]] = + insertOutcomes.getOrNull(j) ?: missingOutcome } if (verifyResults != null) { for (i in batch.indices) { @@ -233,12 +243,16 @@ class IngestQueue( } } } + return finalOutcomes + } + private fun dispatchOutcomes( + batch: List, + outcomes: Array, + ) { for (i in batch.indices) { val sub = batch[i] - val outcome = - finalOutcomes[i] - ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + val outcome = outcomes[i] ?: missingOutcome try { sub.onComplete(outcome) } catch (e: Throwable) { @@ -262,6 +276,14 @@ class IngestQueue( } companion object { + /** + * Default for a missing per-row outcome — only reachable on a + * contract violation (the store returned fewer outcomes than + * inserts), so the message is informational, not user-facing. + */ + private val missingOutcome = + IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + /** * Cap per batch. Sized to keep per-batch latency low (each * transaction holds the SQLite writer mutex; over-large diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index 54c23fbe0..0ec8e602b 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -20,7 +20,6 @@ */ package com.vitorpamplona.quartz.nip01Core.relay.server -import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.crypto.verify import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore @@ -65,7 +64,7 @@ class NostrServer( IngestQueue( store = store, parentContext = parentContext, - verify = if (parallelVerify) ::verifyEvent else null, + verify = if (parallelVerify) ({ it.verify() }) else null, ) private val subStore = LiveEventStore(store, ingest) @@ -128,11 +127,4 @@ class NostrServer( */ @Deprecated("Use close() instead", replaceWith = ReplaceWith("close()")) fun shutdown() = close() - - private companion object { - // Function reference (`Event::verify`) wrapper so the - // ingest hook keeps a single instance per server rather - // than allocating a fresh lambda on every call site. - private fun verifyEvent(event: Event): Boolean = event.verify() - } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index 5c2c56f33..c1698492a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -43,6 +43,7 @@ import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.launch /** @@ -147,7 +148,7 @@ class RelaySession( } } } - } catch (_: kotlinx.coroutines.channels.ClosedSendChannelException) { + } catch (_: ClosedSendChannelException) { // Server is shutting down — the queue is closed. Reply // with a transient failure so a client re-trying against // the next instance gets a sane signal; the WS itself is diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt index 283fda670..e59bd5308 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt @@ -31,13 +31,24 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.IRelayPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.PolicyResult /** - * Allows all commands without authentication. This is the default policy. + * Verifies the Schnorr signature + id hash of every incoming + * `EVENT` and `AUTH` event. Other commands pass through. + * + * The default `VerifyPolicy` singleton verifies both. The + * [VerifyAuthOnlyPolicy] singleton skips the EVENT path — use it + * when the [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue] + * is doing parallel verify (Tier 3 of the event-ingestion plan) + * so EVENTs aren't verified twice. AUTH is still verified inline + * because the AUTH path bypasses the queue entirely; without it, + * a forged event could mark a pubkey as authenticated. */ -object VerifyPolicy : IRelayPolicy { +open class VerifyEventsAndAuthPolicy( + private val verifyEvents: Boolean, +) : IRelayPolicy { override fun onConnect(send: (Message) -> Unit) { } override fun accept(cmd: EventCmd) = - if (cmd.event.verify()) { + if (!verifyEvents || cmd.event.verify()) { PolicyResult.Accepted(cmd) } else { PolicyResult.Rejected("invalid: bad signature or id") @@ -56,3 +67,18 @@ object VerifyPolicy : IRelayPolicy { override fun canSendToSession(event: Event) = true } + +/** + * Default verify policy — checks every EVENT and every AUTH. Use + * this when nothing else in the stack verifies signatures. + */ +object VerifyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = true) + +/** + * Verify policy that skips the EVENT path. Use when the + * `IngestQueue` is configured with `parallelVerify`, so EVENTs are + * verified once on the writer's CPU fan-out instead of inline on + * the WebSocket pump. AUTH is still verified inline because that + * command never reaches the queue. + */ +object VerifyAuthOnlyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = false) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt index eedca3371..cb18c13a9 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt @@ -137,8 +137,11 @@ class ObservableEventStore( } } - @Suppress("UNCHECKED_CAST") - return outcomes.toList() as List + // Every index in `events.indices` was populated above (either + // from the ephemeral pre-pass or the `inner.batchInsert` + // result), so no nulls remain. `requireNoNulls()` enforces + // that with a runtime check instead of an unchecked cast. + return outcomes.requireNoNulls().asList() } override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) { From 176fa6f3b11bef4465b3823ef96d73bc700560ed Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 23:19:13 +0000 Subject: [PATCH 5/5] docs(geode): plan reflects VerifyAuthOnlyPolicy split MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tier 3 used to say operators "must omit VerifyPolicy from their policy chain" when parallelVerify is on — that turned out to be the AUTH-verify regression caught in the audit. Updated the plan to describe the real wiring: VerifyPolicy was split into a parameterised base with two singletons, and composePolicy swaps in VerifyAuthOnlyPolicy so AUTH commands keep signature verification even when the IngestQueue takes EVENT verify. --- .../2026-05-07-event-ingestion-batching.md | 25 +++++++++++++------ 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/geode/plans/2026-05-07-event-ingestion-batching.md b/geode/plans/2026-05-07-event-ingestion-batching.md index 08a8d88f7..eb7d8d9d1 100644 --- a/geode/plans/2026-05-07-event-ingestion-batching.md +++ b/geode/plans/2026-05-07-event-ingestion-batching.md @@ -95,13 +95,24 @@ verifies pre-mark `Rejected` and skip the insert. Wired through `NostrServer(parallelVerify = ...)` and `geode.Relay(parallelVerify = ...)`, controlled by `[options].parallel_verify` in the relay config (default `true`) -and `--no-parallel-verify` on the CLI. Operators that flip it on -must omit `VerifyPolicy` from their policy chain — `Main.kt` does -this automatically; `composePolicy` is told to skip the -`VerifyPolicy` piece when `parallelVerify` is true. Internal -direct callers of `NostrServer` (tests, library users) are -opt-in: the flag defaults to `false` to keep existing -`VerifyPolicy`-in-chain semantics unchanged. +and `--no-parallel-verify` on the CLI. Internal direct callers of +`NostrServer` (tests, library users) are opt-in: the flag defaults +to `false` to keep existing `VerifyPolicy`-in-chain semantics +unchanged. + +`VerifyPolicy` was split into a parameterised +`VerifyEventsAndAuthPolicy(verifyEvents)` with two singletons: + +- `VerifyPolicy` (default): verifies both `EVENT` and `AUTH`. +- `VerifyAuthOnlyPolicy`: verifies `AUTH` only, used when the + `IngestQueue` is doing the EVENT verify. + +When `parallelVerify` is on, `composePolicy` swaps `VerifyPolicy` +for `VerifyAuthOnlyPolicy` so EVENTs aren't verified twice while +AUTH commands — which bypass the queue entirely — keep their +signature check. Without this split, removing `VerifyPolicy` from +the chain would let a forged AUTH event mark a pubkey as +authenticated. Expected: ≈CPU_COUNT× verify-step speed-up on burst publishes from a single connection, where verify was previously serial on