perf: relay read/write path — ordering index, zero-decode REQ replay, prepared-statement cache

Three changes, each validated head-to-head against strfry with relayBench (same 10k-event corpus a1cd3517a8296911, stock configs, sig verify on): - geode: RelayIndexingStrategy turns on indexEventsByCreatedAtAlone for the relay's stores (quartz's DefaultIndexingStrategy stays off for client-side stores). A relay can't predict client filters, and the cheapest REQ of all — {"limit":N} — was a full-table scan + top-N sort: 40 ms and O(table) growth before; 11 ms and index-streamed (first event 30 ms -> 1.9 ms) after. - quartz: zero-decode REQ replay. Stored events now stream as RawEvent (tags kept in serialized form) and are spliced directly into wire frames — no tags parse, no EventFactory dispatch, no re-serialize per row. Gated on the new IRelayPolicy.filtersOutgoingEvents capability: policies that can veto per-event delivery (none today) keep the materialized path; everyone else skips it. Live post-EOSE delivery is unchanged (live matching needs Event objects). - quartz: StatementCachingConnection wraps the pool's writer and reader connections, replaying prepared statements instead of re-preparing per event/REQ (eager reset on return keeps cursors from holding table locks; a 256-statement cap bounds client-controlled filter-shape variety). Ingest went 3,000 -> 4,700 events/s (+55%) — prepare overhead was the single largest non-crypto write cost. Net effect on the benchmark: ingest gap vs strfry narrowed from 2.2x to 1.8x, the firehose latency gap from 4x to 1.2x, and geode now wins 4 of 9 query-latency scenarios (notifications, hashtag, by-ids, recent-window) plus most concurrent-throughput scenarios, while keeping the smaller on-disk footprint. Result sets stayed byte-identical across relays and NIP-77 sync still converges. Measured but deliberately NOT taken: per-row SAVEPOINT elision (+4%, within run noise — not worth weakening batch error isolation), FTS-off (+25% ingest but drops NIP-50), --no-verify (+45% but unfair/unsafe). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NeoCvXnTxsKzqurkmjdC46
2026-08-08 23:54:39 +00:00 · 2026-07-03 18:37:39 +00:00
parent 5f3a790d56
commit 5f0a629e18
15 changed files with 438 additions and 37 deletions
@@ -113,7 +113,7 @@ fun main(args: Array<String>) {
        cliInfoFile?.let { RelayInfo.fromFile(it) }
            ?: config.resolveInfo()

-    val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl)
+    val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl, indexStrategy = RelayIndexingStrategy)

    val policyBuilder: () -> IRelayPolicy = {
        composePolicy(config, advertisedUrl, requireAuth, optionalAuth, verifySigs, parallelVerify)
@@ -58,7 +58,7 @@ import kotlin.coroutines.CoroutineContext
 */
 class RelayEngine(
    val url: NormalizedRelayUrl,
-    val store: IEventStore = EventStore(dbName = null, relay = url),
+    val store: IEventStore = EventStore(dbName = null, relay = url, indexStrategy = RelayIndexingStrategy),
    /**
     * Runtime configuration handle — owns the persistence path (when
     * any), the NIP-11 doc seed, and the seed for the NIP-86 ban /
@@ -0,0 +1,42 @@
+/*
+ * 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.geode
+
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy
+
+/**
+ * Index strategy for a *public relay*, as opposed to quartz's
+ * [DefaultIndexingStrategy] which is tuned for client-side stores that
+ * only ever query their own kinds.
+ *
+ * A relay cannot predict its clients' filters, and the cheapest query a
+ * client can send — `{"limit": N}`, the firehose/landing REQ — carries no
+ * kind, author or tag for the planner to use. Without an index on
+ * `created_at` alone that REQ is a full-table scan + top-N sort on every
+ * poll; relayBench measured it 4× slower than strfry at 10k events and
+ * the gap grows linearly with the table. The extra index costs one B-tree
+ * insert per event, which the same benchmark shows is noise next to the
+ * Schnorr verify + tag indexing already paid on the write path.
+ */
+val RelayIndexingStrategy =
+    DefaultIndexingStrategy(
+        indexEventsByCreatedAtAlone = true,
+    )
@@ -40,6 +40,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.backend.SessionBackend
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.IRelayPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PolicyResult
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
+import com.vitorpamplona.quartz.nip01Core.store.RawEvent
 import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd
 import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd
 import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
@@ -125,6 +126,15 @@ class RelaySession(
        }
    }

+    /** [send] for frames that are already wire-format JSON (the raw REQ path). */
+    private fun sendRaw(json: String) {
+        try {
+            onSend(json)
+        } catch (e: Exception) {
+            Log.w("ClientSession") { "Failed to send to ${e.message}" }
+        }
+    }
+
    override fun close() {
        cancelAllSubscriptions()
        onClose(this)
@@ -281,16 +291,47 @@ class RelaySession(
        val job =
            scope.launch {
                try {
-                    store.query(
-                        ctx = requestContext,
-                        filters = filters,
-                        onEach = { event ->
-                            if (policy.canSendToSession(event)) {
-                                send(EventMessage(cmd.subId, event))
+                    if (policy.filtersOutgoingEvents) {
+                        // Screened path: every event is materialized so the
+                        // policy can veto it per session.
+                        store.query(
+                            ctx = requestContext,
+                            filters = filters,
+                            onEach = { event ->
+                                if (policy.canSendToSession(event)) {
+                                    send(EventMessage(cmd.subId, event))
+                                }
+                            },
+                            onEose = { send(EoseMessage(cmd.subId)) },
+                        )
+                    } else {
+                        // Zero-decode path: the stored replay splices raw
+                        // storage strings straight into wire frames — no tags
+                        // parse, no Event materialization, no re-serialize.
+                        // The `["EVENT","<subId>",` prefix is built once per
+                        // subscription, not per row.
+                        val framePrefix =
+                            buildString {
+                                append("[\"EVENT\",")
+                                RawEvent.appendJsonQuoted(this, cmd.subId)
+                                append(',')
                            }
-                        },
-                        onEose = { send(EoseMessage(cmd.subId)) },
-                    )
+                        store.queryRaw(
+                            ctx = requestContext,
+                            filters = filters,
+                            onEachStored = { raw ->
+                                sendRaw(
+                                    buildString(framePrefix.length + raw.jsonTags.length + raw.content.length + 256) {
+                                        append(framePrefix)
+                                        raw.appendJsonObjectTo(this)
+                                        append(']')
+                                    },
+                                )
+                            },
+                            onEachLive = { event -> send(EventMessage(cmd.subId, event)) },
+                            onEose = { send(EoseMessage(cmd.subId)) },
+                        )
+                    }
                } catch (e: CancellationException) {
                    // Subscription was closed – this is expected.
                    throw e
@@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterIndex
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
+import com.vitorpamplona.quartz.nip01Core.store.RawEvent
 import com.vitorpamplona.quartz.nip50Search.strippingSearchExtensions
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.awaitCancellation
@@ -205,6 +206,51 @@ class LiveEventStore(
        }
    }

+    /**
+     * Zero-decode variant of [query]: the historical replay streams
+     * [com.vitorpamplona.quartz.nip01Core.store.RawEvent] rows straight
+     * from storage (no tags parse, no Event materialization, no
+     * re-serialization), while the live path after EOSE is identical to
+     * [query]'s. Same registration-before-replay ordering and the same
+     * immutable-set dedupe against events accepted mid-replay.
+     */
+    override suspend fun queryRaw(
+        ctx: RequestContext,
+        filters: List<Filter>,
+        onEachStored: (RawEvent) -> Unit,
+        onEachLive: (Event) -> Unit,
+        onEose: () -> Unit,
+    ) {
+        val seenIds = AtomicReference<Set<String>?>(emptySet())
+
+        val sub =
+            LiveSubscription(
+                filters = filters,
+                deliver = { event ->
+                    val seen = seenIds.load()
+                    if (seen != null && seen.contains(event.id)) return@LiveSubscription
+                    onEachLive(event)
+                },
+            )
+
+        index.register(filters, sub)
+        try {
+            store.rawQuery(filters) { raw ->
+                while (true) {
+                    val current = seenIds.load() ?: break
+                    if (raw.id in current) break
+                    if (seenIds.compareAndSet(current, current + raw.id)) break
+                }
+                onEachStored(raw)
+            }
+            onEose()
+            seenIds.store(null)
+            awaitCancellation()
+        } finally {
+            index.unregister(sub)
+        }
+    }
+
    override suspend fun count(
        ctx: RequestContext,
        filters: List<Filter>,
@@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
+import com.vitorpamplona.quartz.nip01Core.store.RawEvent

 /**
 * The data plane a [RelaySession] talks to: how REQ/COUNT are answered, how
@@ -58,6 +59,27 @@ interface SessionBackend {
        onEose: () -> Unit,
    )

+    /**
+     * [query] variant for sessions whose policy does not filter outgoing
+     * events (see
+     * [com.vitorpamplona.quartz.nip01Core.relay.server.policies.IRelayPolicy.filtersOutgoingEvents]):
+     * the stored replay is delivered as [RawEvent]s so the transport can
+     * splice storage strings straight into wire frames without
+     * materializing an [Event] per row. Live events after EOSE still
+     * arrive as [Event]s via [onEachLive] — they exist in object form
+     * already, and live matching needs them.
+     *
+     * The default falls back to [query], treating every delivery as live —
+     * correct for any backend, just without the zero-decode win.
+     */
+    suspend fun queryRaw(
+        ctx: RequestContext,
+        filters: List<Filter>,
+        onEachStored: (RawEvent) -> Unit,
+        onEachLive: (Event) -> Unit,
+        onEose: () -> Unit,
+    ): Unit = query(ctx, filters, onEachLive, onEose)
+
    /** Answers a NIP-45 COUNT with an exact cardinality for the caller in [ctx]. */
    suspend fun count(
        ctx: RequestContext,
@@ -133,10 +133,24 @@ interface IRelayPolicy {
     * Called for each event that matches a subscription's filters. Return
     * true to deliver the event, false to suppress it for this session.
     *
+     * Any implementation that can return `false` here MUST also override
+     * [filtersOutgoingEvents] to return `true`, otherwise the session may
+     * take the zero-decode REQ fast path that never consults this method.
+     *
     * @param event The event about to be sent.
     */
    fun canSendToSession(event: Event): Boolean = true

+    /**
+     * Declares whether [canSendToSession] can ever suppress an event for
+     * this session. When `false` (the default, matching the default
+     * [canSendToSession] that always allows), the engine answers REQs
+     * through the zero-decode raw path — storage strings are spliced
+     * straight into wire frames and [canSendToSession] is never called.
+     * When `true`, every delivered event is materialized and screened.
+     */
+    val filtersOutgoingEvents: Boolean get() = false
+
    operator fun plus(other: IRelayPolicy): IRelayPolicy = PolicyStack(this, other)
 }

@@ -77,4 +77,6 @@ class PolicyStack(
    }

    override fun canSendToSession(event: Event): Boolean = policies.all { it.canSendToSession(event) }
+
+    override val filtersOutgoingEvents: Boolean get() = policies.any { it.filtersOutgoingEvents }
 }
@@ -101,6 +101,21 @@ interface IEventStore : AutoCloseable {
        onEach: (T) -> Unit,
    )

+    /**
+     * Streams matching events in storage form — tags still serialized,
+     * nothing materialized — for read paths that only put events back on
+     * the wire (see [RawEvent]). The default decodes and re-wraps so
+     * every store stays correct; SQLite overrides with a true zero-decode
+     * row read.
+     */
+    suspend fun rawQuery(
+        filters: List<Filter>,
+        onEach: (RawEvent) -> Unit,
+    ): Unit =
+        query<Event>(filters) { event ->
+            onEach(RawEvent.fromEvent(event))
+        }
+
    suspend fun count(filter: Filter): Int

    suspend fun count(filters: List<Filter>): Int
@@ -0,0 +1,129 @@
+/*
+ * 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.store
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.core.Kind
+import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
+import com.vitorpamplona.quartz.utils.EventFactory
+
+/**
+ * An event as it sits in storage: fields as plain strings, `tags` still in
+ * its serialized-JSON form. This is the zero-decode currency of the relay
+ * read path — a REQ that only needs to put the event back on the wire can
+ * splice these strings straight into the outgoing frame via
+ * [appendJsonObjectTo], skipping the tags parse, the [EventFactory]
+ * dispatch to a kind-specific class, and the full re-serialization that
+ * materializing an [Event] would cost per row.
+ */
+class RawEvent(
+    val id: HexKey,
+    val pubKey: HexKey,
+    val createdAt: Long,
+    val kind: Kind,
+    val jsonTags: String,
+    val content: String,
+    val sig: HexKey,
+) {
+    fun <T : Event> toEvent() =
+        EventFactory.create<T>(
+            id,
+            pubKey,
+            createdAt,
+            kind,
+            OptimizedJsonMapper.fromJsonToTagArray(jsonTags),
+            content,
+            sig,
+        )
+
+    /**
+     * Appends this event as a NIP-01 JSON object, reusing the stored
+     * strings verbatim: `id`/`pubkey`/`sig` are validated hex, [jsonTags]
+     * is already JSON; only [content] needs string escaping.
+     */
+    fun appendJsonObjectTo(builder: StringBuilder) {
+        builder
+            .append("{\"id\":\"")
+            .append(id)
+            .append("\",\"pubkey\":\"")
+            .append(pubKey)
+            .append("\",\"created_at\":")
+            .append(createdAt)
+            .append(",\"kind\":")
+            .append(kind)
+            .append(",\"tags\":")
+            .append(jsonTags)
+            .append(",\"content\":")
+        appendJsonQuoted(builder, content)
+        builder
+            .append(",\"sig\":\"")
+            .append(sig)
+            .append("\"}")
+    }
+
+    companion object {
+        fun fromEvent(event: Event) =
+            RawEvent(
+                id = event.id,
+                pubKey = event.pubKey,
+                createdAt = event.createdAt,
+                kind = event.kind,
+                jsonTags = OptimizedJsonMapper.toJson(event.tags),
+                content = event.content,
+                sig = event.sig,
+            )
+
+        /**
+         * Appends [value] as a JSON string literal, quotes included.
+         * Minimal spec-compliant escaping (RFC 8259 §7): `"`, `\`, the
+         * short control escapes, and `\u00XX` for the rest of C0.
+         * Everything else — including non-ASCII — passes through raw,
+         * which is valid JSON and preserves the exact code points the
+         * event id was hashed over.
+         */
+        fun appendJsonQuoted(
+            builder: StringBuilder,
+            value: String,
+        ) {
+            builder.append('"')
+            for (ch in value) {
+                when {
+                    ch == '"' -> builder.append("\\\"")
+                    ch == '\\' -> builder.append("\\\\")
+                    ch == '\n' -> builder.append("\\n")
+                    ch == '\r' -> builder.append("\\r")
+                    ch == '\t' -> builder.append("\\t")
+                    ch == '\b' -> builder.append("\\b")
+                    ch == '\u000C' -> builder.append("\\f")
+                    ch < ' ' -> {
+                        builder.append("\\u")
+                        val hex = ch.code.toString(16)
+                        repeat(4 - hex.length) { builder.append('0') }
+                        builder.append(hex)
+                    }
+                    else -> builder.append(ch)
+                }
+            }
+            builder.append('"')
+        }
+    }
+}
@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
 import com.vitorpamplona.quartz.nip01Core.store.FtsReindexProgress
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
+import com.vitorpamplona.quartz.nip01Core.store.RawEvent

 /**
 * SQLite-backed [IEventStore] with default DB-file name and relay
@@ -61,6 +62,11 @@ class EventStore(
        onEach: (T) -> Unit,
    ) = store.query(filters, onEach)

+    override suspend fun rawQuery(
+        filters: List<Filter>,
+        onEach: (RawEvent) -> Unit,
+    ) = store.rawQuery(filters, onEach)
+
    override suspend fun count(filter: Filter) = store.count(filter)

    override suspend fun count(filters: List<Filter>) = store.count(filters)
@@ -29,6 +29,7 @@ import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
 import com.vitorpamplona.quartz.nip01Core.core.isAddressable
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
+import com.vitorpamplona.quartz.nip01Core.store.RawEvent
 import com.vitorpamplona.quartz.nip01Core.store.sqlite.sql.where
 import com.vitorpamplona.quartz.utils.EventFactory

@@ -82,14 +82,21 @@ class SQLiteConnectionPool(
    private val readerChannel: Channel<SQLiteConnection>?

    init {
-        writer = openConnection()
+        // The writer replays the same INSERT statements for every event —
+        // cache its prepared statements so sqlite3_prepare is paid once per
+        // SQL string instead of once per row. (Migrations run through the
+        // same wrapper; DDL statements just cache and stay unused.)
+        writer = StatementCachingConnection(openConnection())
        onMigrate(writer)

        if (isInMemory) {
            readers = emptyList()
            readerChannel = null
        } else {
-            readers = List(numReaders) { openConnection() }
+            // Readers replay the same filter shapes all day (feeds, threads,
+            // profile hydrations) — caching their prepared statements pays
+            // the parse/plan cost once per shape per connection.
+            readers = List(numReaders) { StatementCachingConnection(openConnection()) }
            readerChannel = Channel(numReaders)
            readers.forEach { readerChannel.trySend(it) }
        }
@@ -26,16 +26,14 @@ import androidx.sqlite.SQLiteException
 import androidx.sqlite.driver.bundled.BundledSQLiteDriver
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
-import com.vitorpamplona.quartz.nip01Core.core.Kind
-import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
 import com.vitorpamplona.quartz.nip01Core.core.isEphemeral
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.store.FtsReindexProgress
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
+import com.vitorpamplona.quartz.nip01Core.store.RawEvent
 import com.vitorpamplona.quartz.nip40Expiration.isExpired
-import com.vitorpamplona.quartz.utils.EventFactory

 class SQLiteEventStore(
    val driver: SQLiteDriver = BundledSQLiteDriver(),
@@ -372,24 +370,3 @@ class SQLiteEventStore(

    fun close() = pool.close()
 }
-
-class RawEvent(
-    val id: HexKey,
-    val pubKey: HexKey,
-    val createdAt: Long,
-    val kind: Kind,
-    val jsonTags: String,
-    val content: String,
-    val sig: HexKey,
-) {
-    fun <T : Event> toEvent() =
-        EventFactory.create<T>(
-            id,
-            pubKey,
-            createdAt,
-            kind,
-            OptimizedJsonMapper.fromJsonToTagArray(jsonTags),
-            content,
-            sig,
-        )
-}
@@ -0,0 +1,99 @@
+/*
+ * 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.store.sqlite
+
+import androidx.sqlite.SQLiteConnection
+import androidx.sqlite.SQLiteStatement
+
+/**
+ * A [SQLiteConnection] decorator that keeps every prepared statement and
+ * hands the same handle back on the next [prepare] of the same SQL. The
+ * event-ingest hot path prepares the identical INSERT statements once per
+ * event; sqlite3_prepare is pure overhead the second time around.
+ *
+ * Callers keep their idiomatic `prepare(sql).use { … }` blocks untouched:
+ * the statement wrapper turns `close()` into a return-to-cache no-op (the
+ * real reset + clearBindings happens on the next checkout). Statements are
+ * only truly finalized when the connection itself closes.
+ *
+ * Constraints, by design of the call sites:
+ *  - **Not thread-safe** — same contract as the underlying connection,
+ *    which the pool already serializes (single writer under a mutex).
+ *  - **No overlapping use of the same SQL** — checking out one SQL string
+ *    twice without closing the first use would alias one native handle.
+ *    Insert/query paths never nest the same statement; a checkout while
+ *    the previous one is still open falls back to an uncached statement.
+ */
+class StatementCachingConnection(
+    private val delegate: SQLiteConnection,
+    /**
+     * Ceiling on retained statements. Query SQL embeds one `?` per filter
+     * element, so shape variety is client-controlled — without a cap a
+     * long-lived relay connection would accumulate native handles without
+     * bound. Once full, unseen SQL just prepares uncached. 256 comfortably
+     * covers the write path's fixed set plus the recurring filter shapes.
+     */
+    private val maxCachedStatements: Int = 256,
+) : SQLiteConnection by delegate {
+    private val cache = HashMap<String, CachedStatement>()
+
+    override fun prepare(sql: String): SQLiteStatement {
+        val cached =
+            cache[sql] ?: run {
+                if (cache.size >= maxCachedStatements) return delegate.prepare(sql)
+                CachedStatement(delegate.prepare(sql)).also { cache[sql] = it }
+            }
+        if (cached.checkedOut) {
+            // Same SQL prepared while the previous handle is still in use —
+            // stay correct with a plain uncached statement.
+            return delegate.prepare(sql)
+        }
+        cached.checkedOut = true
+        cached.clearBindings()
+        return cached
+    }
+
+    override fun close() {
+        cache.values.forEach { runCatching { it.finalize() } }
+        cache.clear()
+        delegate.close()
+    }
+
+    private class CachedStatement(
+        private val delegate: SQLiteStatement,
+    ) : SQLiteStatement by delegate {
+        var checkedOut = false
+
+        /**
+         * Return to cache. The real handle stays prepared, but must be
+         * reset *now*: an un-reset statement keeps its cursor (and its
+         * table read locks) open, which turns a later `DELETE`/`DROP` on
+         * the same connection into `SQLITE_LOCKED: database table is
+         * locked`.
+         */
+        override fun close() {
+            runCatching { delegate.reset() }
+            checkedOut = false
+        }
+
+        fun finalize() = delegate.close()
+    }
+}