perf(quartz): Tier 3 — parallel Schnorr verify in IngestQueue

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

@@ -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
 

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
+WAL is already on (`PRAGMA journal_mode=WAL`). The pool runs with
-event-store DB; group commits across the writer mutex's hold window.
+`PRAGMA synchronous=OFF`, which is one notch more permissive than
-Today each insert is its own transaction. Wrap N inserts (or a 5 ms
+the originally-sketched `synchronous=NORMAL` — we keep it as-is
-budget, whichever first) in a single transaction managed by the writer
+because the project already accepted the OS-crash trade-off there.
-coroutine.
 
-Because OK reflects acceptance not durability, each row can fan an OK
+Group commit is implemented via a new `IEventStore.batchInsert`:
-as soon as the per-row INSERT statement returns inside the
+the SQLite override holds the writer mutex once and wraps N events
-transaction — we do not need to wait for the batch's commit. The
+in one `BEGIN IMMEDIATE … COMMIT`. Per-row error isolation uses
-fsync is hidden from the publisher latency budget entirely.
+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`,
+OKs fire as soon as each row's outcome is known inside the writer
-not geode — but geode owns the benchmark and validates the gain.
+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,
+`RelaySession.receive` was single-flight: one EVENT in, process, OK
-process, OK out, next EVENT. Allow a connection to push N EVENTs
+out, next EVENT. With Tier 2 the connection's pump posts to the
-concurrently and dispatch them to a per-connection ingest pipeline.
+shared `IngestQueue` and returns immediately — the WS pump moves
+straight to the next frame.
 
-A `Channel<EventCmd>` with capacity = `INGEST_PIPELINE_DEPTH` per
+`IngestQueue` (one per `NostrServer`) holds a bounded
-connection, drained by a coroutine that feeds the group-commit writer
+`Channel<Submission>` (capacity = 1024 per the `DEFAULT_CAPACITY`
-above. OKs go straight to `outQueue.send()` the moment each row
+constant) drained by a single writer coroutine. The writer pulls
-returns from INSERT — no ordering bookkeeping needed, since the OK
+the first item to start a batch then `tryReceive`-drains everything
-frame already carries the event id and the spec doesn't require
+else queued (up to 64 — `DEFAULT_MAX_BATCH`), feeds the whole batch
-order. A pipelined publisher keying on event id will pair replies
+to `IEventStore.batchInsert`, and dispatches each row's
-correctly.
+`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
+`VerifyPolicy` ran synchronously on `receive`, serialising verify
-`receive`. Move it into the ingest pipeline so verification of EVENT N+1
+on each connection's pump coroutine. With Tier 3, `IngestQueue`
-runs concurrently with the SQLite commit of EVENT N. secp256k1 verify
+takes a `verify: ((Event) -> Boolean)?` hook; when set, the writer
-is parallelisable; the writer should never block on it.
+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
+- `publishGroupCommitSingleClient` — sequential publish-and-confirm
-  single-client benchmark, asserts >5000 EPS.
+  on one connection (the same shape as the original
-- `publishPipelinedSingleClient` — sends 100 EVENTs without awaiting
+  `publishThroughputSingleClient`). Synchronous publishing means
-  intermediate OKs; measures end-to-end throughput and verifies that
+  batch size is always 1, so this case shows per-event SQLite tx
-  every event id receives exactly one OK (in any order).
+  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.
 

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

@@ -83,6 +83,12 @@ fun main(args: Array<String>) {
     // 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<String>) {
     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).

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,
         )
 
     /**

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(

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.

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<Submission>(maxBatch)
-        val events = ArrayList<Event>(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 =
+                processBatch(batch)
-                    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) {
@@ -175,6 +173,82 @@ class IngestQueue(
         }
     }
 
+    private suspend fun processBatch(batch: List<Submission>) {
+        // 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<Event>
+        val insertIndices: IntArray
+        if (verifyResults == null) {
+            toInsert = batch.map { it.event }
+            insertIndices = IntArray(batch.size) { it }
+        } else {
+            val accepted = ArrayList<Event>(batch.size)
+            val mapping = ArrayList<Int>(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<IEventStore.InsertOutcome> =
+            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<IEventStore.InsertOutcome>(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

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()
+    }
 }