diff --git a/geode/config.example.toml b/geode/config.example.toml index 3d706999b..2a4d8ce5a 100644 --- a/geode/config.example.toml +++ b/geode/config.example.toml @@ -53,6 +53,12 @@ file = "/var/lib/geode/events.db" # only for trusted-input scenarios (test fixtures, mirror replays). # verify_signatures = true +# Run signature verification in parallel inside the IngestQueue +# (across all CPU cores) instead of serially on each connection's +# WebSocket pump. Default: true. Set false to fall back to the +# legacy in-policy verify path. +# parallel_verify = true + # Require clients to NIP-42 AUTH before REQ/EVENT/COUNT. require_auth = false diff --git a/geode/plans/2026-05-07-event-ingestion-batching.md b/geode/plans/2026-05-07-event-ingestion-batching.md index e6df99ebf..eb7d8d9d1 100644 --- a/geode/plans/2026-05-07-event-ingestion-batching.md +++ b/geode/plans/2026-05-07-event-ingestion-batching.md @@ -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 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` (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. diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index e7b7a97b0..a731890e8 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.KindAllowDenyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PubkeyAllowDenyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RejectFutureEventsPolicy +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyAuthOnlyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore @@ -83,6 +84,12 @@ fun main(args: Array) { // opts out (CLI `--no-verify` or `[options].verify_signatures = false` // in the config). val verifySigs = !a.flag("--no-verify") && config.options.verify_signatures + // Parallel verify is on whenever signature checking is on; the + // IngestQueue handles it instead of VerifyPolicy. Operators can + // force the legacy in-policy path with `--no-parallel-verify` or + // `[options].parallel_verify = false`. + val parallelVerify = + verifySigs && !a.flag("--no-parallel-verify") && config.options.parallel_verify // Advertised URL: explicit `info.relay_url` wins, then build from // host/port/path. 0.0.0.0 bind → 127.0.0.1 in the URL so NIP-42 @@ -98,11 +105,19 @@ fun main(args: Array) { 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() @@ -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(EmptyPolicy) { acc, p -> diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt index 48c339f4e..6a8436103 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt @@ -73,6 +73,17 @@ class Relay( * everything in memory only — fine for tests. */ stateFile: File? = null, + /** + * Run Schnorr signature verification in parallel inside the + * [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue] + * instead of serially in the policy chain. Enables the Tier-3 + * win in `geode/plans/2026-05-07-event-ingestion-batching.md`. + * + * When set, callers MUST omit `VerifyPolicy` from [policyBuilder] + * — having both verifies the same event twice for no benefit. + * `Main.kt` skips `VerifyPolicy` when this flag is on. + */ + parallelVerify: Boolean = false, ) : AutoCloseable { private val stateStore: RelayStateStore? = stateFile?.let { RelayStateStore(it) } @@ -158,6 +169,7 @@ class Relay( if (user === EmptyPolicy) BanListPolicy(banStore) else user + BanListPolicy(banStore) }, parentContext, + parallelVerify = parallelVerify, ) /** diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt index 1417ec61f..1f4ff4441 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt @@ -141,6 +141,18 @@ data class RelayConfig( * for trusted-input scenarios (test fixtures, mirror replays). */ val verify_signatures: Boolean = true, + /** + * Run signature verification in parallel inside the IngestQueue + * (CPU fan-out across `Dispatchers.Default`) instead of serially + * on each connection's WebSocket pump. Tier-3 of the + * `event-ingestion-batching` plan. Wins scale with how many + * EVENTs a single connection sends back-to-back: ~CPU_COUNT× + * verify-step speed-up on burst publishes. Set false to keep + * the legacy in-policy verify path. + * + * Only takes effect when [verify_signatures] is also true. + */ + val parallel_verify: Boolean = true, ) data class LimitsSection( diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt index 04a5867fc..71ce65fe5 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -363,6 +363,52 @@ class LoadBenchmark { } } + /** + * Same workload as [publishThroughputSingleClient] (sequential + * publish-and-confirm on one connection) — kept as a regression + * floor for the group-commit code path. Synchronous publishes + * never coalesce in the writer (batch size is always 1), so the + * EPS here measures per-event SQLite tx cost. The pipelined win + * shows up in [publishPipelinedSingleClient]. + */ + @Test + fun publishGroupCommitSingleClient() = + benchmark("publish group-commit single client") { + runBenchmarkServer { server, http -> + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope) + try { + val signer = NostrSignerSync(KeyPair()) + val relayUrl = server.url.normalizeRelayUrl() + + val n = 10_000 + var ok = 0 + val elapsed = + measureTime { + runBlocking { + repeat(n) { i -> + val event = signer.sign(TextNoteEvent.build("group-commit $i")) + if (client.publishAndConfirm(event, setOf(relayUrl))) ok++ + } + } + } + val eps = (n * 1000.0) / elapsed.inWholeMilliseconds + println( + "events=$n ok=$ok elapsedMs=${elapsed.inWholeMilliseconds} eps=${"%.0f".format(eps)}", + ) + check(ok == n) { "expected all $n events accepted, got $ok" } + // Floor: the pre-batching baseline was ~760 EPS + // single-client (see plan). Anything below 500 + // means the group-commit / ingest-queue rewrite + // regressed the synchronous path. + check(eps > 500) { "synchronous EPS $eps fell below the 500 floor" } + } finally { + client.disconnect() + scope.cancel() + } + } + } + /** * One publisher, N subscribers. Publishes one EVENT and measures * fan-out latency: time from publish to last subscriber receiving. @@ -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() + val done = java.util.concurrent.CountDownLatch(1) + + val ws = + http.newWebSocket( + httpUrl, + object : okhttp3.WebSocketListener() { + override fun onMessage( + webSocket: okhttp3.WebSocket, + text: String, + ) { + if (!text.startsWith("[\"OK\"")) return + // ["OK","",true|false,""] — + // a tiny string scan is enough for a + // bench. Index 6 is past `["OK","`. + val idStart = 7 + val idEnd = text.indexOf('"', idStart) + if (idEnd <= idStart) return + val id = text.substring(idStart, idEnd) + if (!ids.contains(id)) { + unknownIds.incrementAndGet() + return + } + if (!seenIds.add(id)) return + if (text.contains(",true,")) { + okSeen.incrementAndGet() + } else { + okFailures.incrementAndGet() + } + if (okSeen.get() + okFailures.get() == n.toLong()) done.countDown() + } + }, + ) + + val elapsed = + measureTime { + // Burst-send: queue every EVENT to OkHttp's + // outbound buffer without any await, then + // wait for the corresponding OK frames. + for (event in events) { + ws.send("""["EVENT",${event.toJson()}]""") + } + check(done.await(60, java.util.concurrent.TimeUnit.SECONDS)) { + "timed out waiting for OKs: ok=${okSeen.get()} rej=${okFailures.get()} unknown=${unknownIds.get()}" + } + } + val eps = (n * 1000.0) / elapsed.inWholeMilliseconds + println( + "events=$n ok=${okSeen.get()} rejected=${okFailures.get()} " + + "unknownIds=${unknownIds.get()} elapsedMs=${elapsed.inWholeMilliseconds} eps=${"%.0f".format(eps)}", + ) + check(okSeen.get() == n.toLong()) { + "expected $n accepted OKs, got ${okSeen.get()} (rejected ${okFailures.get()})" + } + check(seenIds.size == n) { + "expected $n unique OK ids, got ${seenIds.size} — duplicate or missing OKs" + } + ws.cancel() + } + } + /** * Many concurrent publishers, each on their own WebSocket. Tells * us whether the SQLite single-writer bottleneck is the floor or diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt new file mode 100644 index 000000000..5b4876693 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt @@ -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(capacity) + private val scope = CoroutineScope(parentContext + SupervisorJob()) + + /** + * Lazily-launched drain coroutine. We don't start it in `init` + * because eagerly launching from a server's lazy + * `LiveEventStore` allocates a Default-dispatcher slot at server + * construction time — visible to other tests sharing the same + * `Dispatchers.Default` pool, where it can perturb scheduling + * for unrelated REQ/EOSE timing. Starting on first `submit` + * keeps relays that never see an EVENT (read-only sessions, + * negentropy-only) from paying for the writer at all. + */ + @Volatile + private var writerStarted = false + private val startLock = Any() + + /** + * Hand off [event] for insertion. [onComplete] is invoked once + * with the per-row outcome from the writer batch — exactly once, + * unless the queue closes mid-flight. + * + * Suspends only when [incoming] is full (writer fell behind by + * [DEFAULT_CAPACITY] events). Otherwise this returns as fast as a + * channel `send`, freeing the WebSocket pump to read the next + * frame — that's where the per-connection pipeline win comes + * from. + */ + suspend fun submit( + event: Event, + onComplete: (IEventStore.InsertOutcome) -> Unit, + ) { + ensureWriterStarted() + incoming.send(Submission(event, onComplete)) + } + + private fun ensureWriterStarted() { + if (writerStarted) return + synchronized(startLock) { + if (writerStarted) return + scope.launch { drainLoop() } + writerStarted = true + } + } + + private suspend fun drainLoop() { + val batch = ArrayList(maxBatch) + 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) { + 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): BooleanArray? { + val hook = verify ?: return null + if (batch.size == 1) return BooleanArray(1) { hook(batch[0].event) } + return coroutineScope { + batch + .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } + .awaitAll() + .toBooleanArray() + } + } + + /** + * Run the SQLite transaction for the verified subset of [batch] + * and stitch outcomes back to a per-batch-index array. Failed + * verifies pre-mark `Rejected` and skip the insert. A whole-batch + * commit failure converts every persisted entry to `Rejected` + * with the throw message. + */ + private suspend fun runInsertStage( + batch: List, + verifyResults: BooleanArray?, + ): Array { + val toInsert = ArrayList(batch.size) + val insertIndices = ArrayList(batch.size) + for (i in batch.indices) { + if (verifyResults == null || verifyResults[i]) { + toInsert.add(batch[i].event) + insertIndices.add(i) + } + } + + val insertOutcomes: List = + if (toInsert.isEmpty()) { + emptyList() + } else { + try { + store.batchInsert(toInsert) + } catch (e: Throwable) { + Log.w("IngestQueue") { "batchInsert failed for ${toInsert.size} events: ${e.message}" } + val reason = e.message ?: e::class.simpleName ?: "insert failed" + List(toInsert.size) { IEventStore.InsertOutcome.Rejected(reason) } + } + } + + val finalOutcomes = arrayOfNulls(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, + outcomes: Array, + ) { + 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 + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt index 00d2436f9..912c1e4d0 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt @@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.server import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.onSubscription @@ -39,6 +40,7 @@ import kotlinx.coroutines.flow.onSubscription */ class LiveEventStore( private val store: IEventStore, + private val ingest: IngestQueue, ) { private val newEventStream = MutableSharedFlow( @@ -47,9 +49,47 @@ class LiveEventStore( onBufferOverflow = BufferOverflow.DROP_LATEST, // Default behavior ) + /** + * Fire-and-forget enqueue: hand [event] to the [IngestQueue] and + * fire [onComplete] once the writer's batch has a per-row + * decision. On `Accepted` the live stream is also emitted to so + * subscribers see the event. Suspends only when the ingest queue + * is full (backpressure). + */ + suspend fun submit( + event: Event, + onComplete: (IEventStore.InsertOutcome) -> Unit, + ) { + ingest.submit(event) { outcome -> + if (outcome is IEventStore.InsertOutcome.Accepted) { + newEventStream.tryEmit(event) + } + onComplete(outcome) + } + } + + /** + * Suspending insert kept for callers that don't care about + * pipelining (tests, scripted paths). Routes through the same + * [IngestQueue] as [submit] so the batch write path is exercised + * even by tests that prefer a sequential `insert` API. + * Throws on rejection so callers can `try` around it the way the + * old API did. + */ suspend fun insert(event: Event) { - store.insert(event) - newEventStream.tryEmit(event) + val done = CompletableDeferred() + submit(event) { outcome -> + when (outcome) { + IEventStore.InsertOutcome.Accepted -> { + done.complete(Unit) + } + + is IEventStore.InsertOutcome.Rejected -> { + done.completeExceptionally(IllegalStateException(outcome.reason)) + } + } + } + done.await() } suspend fun query( diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index f72c9e7b2..0ec8e602b 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -20,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() @@ -95,6 +117,7 @@ class NostrServer( override fun close() { connections.forEach { _, session -> session.cancelAllSubscriptions() } connections.clear() + ingest.close() scope.cancel() store.close() } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index d72ff491c..c1698492a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -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")) } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt index 283fda670..e59bd5308 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt @@ -31,13 +31,24 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.IRelayPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.PolicyResult /** - * Allows all commands without authentication. This is the default policy. + * Verifies the Schnorr signature + id hash of every incoming + * `EVENT` and `AUTH` event. Other commands pass through. + * + * The default `VerifyPolicy` singleton verifies both. The + * [VerifyAuthOnlyPolicy] singleton skips the EVENT path — use it + * when the [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue] + * is doing parallel verify (Tier 3 of the event-ingestion plan) + * so EVENTs aren't verified twice. AUTH is still verified inline + * because the AUTH path bypasses the queue entirely; without it, + * a forged event could mark a pubkey as authenticated. */ -object VerifyPolicy : IRelayPolicy { +open class VerifyEventsAndAuthPolicy( + private val verifyEvents: Boolean, +) : IRelayPolicy { override fun onConnect(send: (Message) -> Unit) { } override fun accept(cmd: EventCmd) = - if (cmd.event.verify()) { + if (!verifyEvents || cmd.event.verify()) { PolicyResult.Accepted(cmd) } else { PolicyResult.Rejected("invalid: bad signature or id") @@ -56,3 +67,18 @@ object VerifyPolicy : IRelayPolicy { override fun canSendToSession(event: Event) = true } + +/** + * Default verify policy — checks every EVENT and every AUTH. Use + * this when nothing else in the stack verifies signatures. + */ +object VerifyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = true) + +/** + * Verify policy that skips the EVENT path. Use when the + * `IngestQueue` is configured with `parallelVerify`, so EVENTs are + * verified once on the writer's CPU fan-out instead of inline on + * the WebSocket pump. AUTH is still verified inline because that + * command never reaches the queue. + */ +object VerifyAuthOnlyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = false) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt index d376b8e7d..8984e2fc5 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt @@ -41,6 +41,42 @@ interface IEventStore : AutoCloseable { suspend fun transaction(body: ITransaction.() -> Unit) + /** + * Per-row outcome from [batchInsert]. The OK frame on the wire is + * built from this — `Accepted` becomes `OK true`, `Rejected.reason` + * becomes the false reason. NIP-01 says OK pairs to its EVENT by + * id, not by order, so callers may dispatch outcomes in any order. + */ + sealed class InsertOutcome { + data object Accepted : InsertOutcome() + + data class Rejected( + val reason: String, + ) : InsertOutcome() + } + + /** + * Bulk insert in a single transaction with per-row error isolation. + * Returns one outcome per input event in the same order. + * + * Implementations must isolate per-row failures so one bad event + * doesn't roll back the others (SQLite uses SAVEPOINTs). If the + * outer commit itself fails, every entry in the returned list is + * `Rejected` with the commit-failure reason. + * + * Default impl runs each insert in its own transaction — correct + * but loses the group-commit win. SQLite overrides this. + */ + suspend fun batchInsert(events: List): List = + events.map { event -> + try { + insert(event) + InsertOutcome.Accepted + } catch (e: Throwable) { + InsertOutcome.Rejected(e.message ?: e::class.simpleName ?: "insert failed") + } + } + suspend fun query(filter: Filter): List suspend fun query(filters: List): List diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt index 1def6bf4f..cb18c13a9 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt @@ -96,6 +96,54 @@ class ObservableEventStore( _changes.emit(StoreChange.Insert(event)) } + override suspend fun batchInsert(events: List): List { + // Split into ephemeral (no-store) and persistable, delegate the + // persistable subset to the inner store's batched path so we + // keep the group-commit win, then merge outcomes back in input + // order. Already-expired ephemerals are dropped (matching + // [insert]). Accepted events are emitted on [_changes] only + // after the inner batch returns, so a commit failure that + // converts everything to Rejected suppresses the emits. + if (events.isEmpty()) return emptyList() + + val outcomes = arrayOfNulls(events.size) + val persistableIndices = ArrayList(events.size) + val persistable = ArrayList(events.size) + for (i in events.indices) { + val event = events[i] + if (event.kind.isEphemeral()) { + outcomes[i] = + if (event.isExpired()) { + IEventStore.InsertOutcome.Rejected("blocked: Cannot insert an expired event") + } else { + IEventStore.InsertOutcome.Accepted + } + } else { + persistableIndices.add(i) + persistable.add(event) + } + } + + if (persistable.isNotEmpty()) { + val innerOutcomes = inner.batchInsert(persistable) + for (j in persistable.indices) { + outcomes[persistableIndices[j]] = innerOutcomes[j] + } + } + + for (i in events.indices) { + if (outcomes[i] is IEventStore.InsertOutcome.Accepted) { + _changes.emit(StoreChange.Insert(events[i])) + } + } + + // 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() inner.transaction { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt index ef9510611..097c5672a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt @@ -43,6 +43,8 @@ class EventStore( override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = store.transaction(body) + override suspend fun batchInsert(events: List) = store.batchInsertEvents(events) + override suspend fun query(filter: Filter) = store.query(filter) override suspend fun query(filters: List) = store.query(filters) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt index 2da5cf49c..78b29bffd 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt @@ -201,6 +201,63 @@ class SQLiteEventStore( } } + /** + * Group-commit batch insert with per-row error isolation via + * SAVEPOINTs. Acquires the writer mutex once and wraps every + * inserts in a single outer transaction so the WAL append + sync + * cost is paid once for the whole batch. + * + * Per-row contract: + * - Validation errors (expired) and per-row INSERT failures + * (UNIQUE constraint, etc.) ROLLBACK only that row's savepoint; + * other rows commit. + * - Ephemeral kinds are accepted without writing — the live + * stream still surfaces them; persistence is intentionally a + * no-op per NIP-01. + * + * Outer-commit failure throws; the caller treats every entry as + * `Rejected` (this is what the IEventStore contract documents). + */ + suspend fun batchInsertEvents(events: List): List { + if (events.isEmpty()) return emptyList() + val outcomes = ArrayList(events.size) + pool.useWriter { db -> + db.transaction { + events.forEachIndexed { i, event -> + outcomes.add(insertWithSavepoint(event, i, this)) + } + } + } + return outcomes + } + + private fun insertWithSavepoint( + event: Event, + index: Int, + db: SQLiteConnection, + ): IEventStore.InsertOutcome { + if (event.isExpired()) { + return IEventStore.InsertOutcome.Rejected("blocked: Cannot insert an expired event") + } + if (event.kind.isEphemeral()) return IEventStore.InsertOutcome.Accepted + + val sp = "ev$index" + db.execSQL("SAVEPOINT $sp") + return try { + innerInsertEvent(event, db) + db.execSQL("RELEASE SAVEPOINT $sp") + IEventStore.InsertOutcome.Accepted + } catch (e: Throwable) { + // Roll back just this row, then release the (now empty) + // savepoint frame so the next iteration's BEGIN works. + // Both calls are individually try/catch'd because a failed + // ROLLBACK shouldn't mask the original cause. + runCatching { db.execSQL("ROLLBACK TRANSACTION TO SAVEPOINT $sp") } + runCatching { db.execSQL("RELEASE SAVEPOINT $sp") } + IEventStore.InsertOutcome.Rejected(e.message ?: e::class.simpleName ?: "insert failed") + } + } + inner class Transaction( val db: SQLiteConnection, ) : IEventStore.ITransaction {