Merge pull request #2769 from vitorpamplona/claude/fix-ok-message-ordering-fNXmT

Implement event ingestion batching with group commit and parallel verify
2026-05-07 19:21:44 -04:00
parent a38a56ea78 176fa6f3b1
commit 6d3bd3a727
15 changed files with 854 additions and 49 deletions
@@ -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

@@ -19,62 +19,119 @@ 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

 ### 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. On commit, fan back N OK replies.
+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.

-Implementation lives in quartz's `EventStore` / `SQLiteConnectionPool`,
-not geode — but geode owns the benchmark and validates the gain.
+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`.
+
+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, dispatch them to a per-connection ingest pipeline, and
-serialise OKs back in arrival order via a small commit log.
+`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<EventCmd> 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.
+`IngestQueue` (one per `NostrServer`) holds a bounded
+`Channel<Submission>` (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.

-Expected: hides the verify+insert latency behind another EVENT's
-parse, gets us closer to network-bound throughput.
+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. 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
+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 and OK-ordering correctness.
+- `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.

@@ -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
@@ -83,6 +84,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 +105,19 @@ fun main(args: Array<String>) {
    val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl)

    val policyBuilder: () -> IRelayPolicy = {
-        composePolicy(config, advertisedUrl, requireAuth, verifySigs)
+        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).
@@ -151,6 +166,7 @@ private fun composePolicy(
    advertisedUrl: com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl,
    requireAuth: Boolean,
    verifySigs: Boolean,
+    parallelVerify: Boolean,
 ): IRelayPolicy {
    val pieces = mutableListOf<IRelayPolicy>()

@@ -171,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<IRelayPolicy, IRelayPolicy>(EmptyPolicy) { acc, p ->
@@ -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,
        )

    /**
@@ -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(
@@ -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.
@@ -450,6 +496,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<String>()
+                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","<id>",true|false,"<reason>"] —
+                                // 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
@@ -0,0 +1,305 @@
+/*
+ * 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.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.channels.ClosedReceiveChannelException
+import kotlinx.coroutines.coroutineScope
+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,
+    /**
+     * 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
+     * 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<Submission>(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<Submission>(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)
+                }
+
+                processBatch(batch)
+                batch.clear()
+            }
+        } catch (_: ClosedReceiveChannelException) {
+            // Normal shutdown via close().
+        }
+    }
+
+    private suspend fun processBatch(batch: List<Submission>) {
+        val verifyResults = verifyBatch(batch)
+        val finalOutcomes = runInsertStage(batch, verifyResults)
+        dispatchOutcomes(batch, finalOutcomes)
+    }
+
+    /**
+     * 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<Submission>): 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<Submission>,
+        verifyResults: BooleanArray?,
+    ): Array<IEventStore.InsertOutcome?> {
+        val toInsert = ArrayList<Event>(batch.size)
+        val insertIndices = ArrayList<Int>(batch.size)
+        for (i in batch.indices) {
+            if (verifyResults == null || verifyResults[i]) {
+                toInsert.add(batch[i].event)
+                insertIndices.add(i)
+            }
+        }
+
+        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) }
+                }
+            }
+
+        val finalOutcomes = arrayOfNulls<IEventStore.InsertOutcome>(batch.size)
+        for (j in insertIndices.indices) {
+            finalOutcomes[insertIndices[j]] =
+                insertOutcomes.getOrNull(j) ?: missingOutcome
+        }
+        if (verifyResults != null) {
+            for (i in batch.indices) {
+                if (!verifyResults[i]) {
+                    finalOutcomes[i] = IEventStore.InsertOutcome.Rejected(verifyRejectionReason)
+                }
+            }
+        }
+        return finalOutcomes
+    }
+
+    private fun dispatchOutcomes(
+        batch: List<Submission>,
+        outcomes: Array<IEventStore.InsertOutcome?>,
+    ) {
+        for (i in batch.indices) {
+            val sub = batch[i]
+            val outcome = outcomes[i] ?: missingOutcome
+            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
+     * 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 {
+        /**
+         * 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
+         * 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
+    }
+}
@@ -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<Event>(
@@ -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<Unit>()
+        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(
@@ -20,6 +20,7 @@
 */
 package com.vitorpamplona.quartz.nip01Core.relay.server

+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,17 +37,38 @@ 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 {
-    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 = store,
+            parentContext = parentContext,
+            verify = if (parallelVerify) ({ it.verify() }) else null,
+        )
+
+    private val subStore = LiveEventStore(store, ingest)
+
    /** Active client sessions keyed by an opaque connection id. */
    private val connections = LargeCache<Int, RelaySession>()

@@ -95,6 +117,7 @@ class NostrServer(
    override fun close() {
        connections.forEach { _, session -> session.cancelAllSubscriptions() }
        connections.clear()
+        ingest.close()
        scope.cancel()
        store.close()
    }
@@ -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
@@ -42,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

 /**
@@ -129,11 +131,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 (_: 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"))
        }
    }

@@ -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)
@@ -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<Event>): List<InsertOutcome> =
+        events.map { event ->
+            try {
+                insert(event)
+                InsertOutcome.Accepted
+            } catch (e: Throwable) {
+                InsertOutcome.Rejected(e.message ?: e::class.simpleName ?: "insert failed")
+            }
+        }
+
    suspend fun <T : Event> query(filter: Filter): List<T>

    suspend fun <T : Event> query(filters: List<Filter>): List<T>
@@ -96,6 +96,54 @@ class ObservableEventStore(
        _changes.emit(StoreChange.Insert(event))
    }

+    override suspend fun batchInsert(events: List<Event>): List<IEventStore.InsertOutcome> {
+        // 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<IEventStore.InsertOutcome>(events.size)
+        val persistableIndices = ArrayList<Int>(events.size)
+        val persistable = ArrayList<Event>(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]))
+            }
+        }
+
+        // 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) {
        val accepted = ArrayList<Event>()
        inner.transaction {
@@ -43,6 +43,8 @@ class EventStore(

    override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = store.transaction(body)

+    override suspend fun batchInsert(events: List<Event>) = store.batchInsertEvents(events)
+
    override suspend fun <T : Event> query(filter: Filter) = store.query<T>(filter)

    override suspend fun <T : Event> query(filters: List<Filter>) = store.query<T>(filters)
@@ -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<Event>): List<IEventStore.InsertOutcome> {
+        if (events.isEmpty()) return emptyList()
+        val outcomes = ArrayList<IEventStore.InsertOutcome>(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 {