perf(quartz): group-commit + per-connection ingest pipeline

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).
2026-05-07 22:41:40 +00:00
parent 39e81c1eb5
commit 7a92f4ef2f
9 changed files with 521 additions and 8 deletions
@@ -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<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)
+        val events = ArrayList<Event>(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
+    }
+}
@@ -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(
@@ -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<Int, RelaySession>()

@@ -95,6 +104,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
@@ -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"))
        }
    }

@@ -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,51 @@ 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]))
+            }
+        }
+
+        @Suppress("UNCHECKED_CAST")
+        return outcomes.toList() as List<IEventStore.InsertOutcome>
+    }
+
    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 {