perf(quartz/geode): serve full-set NEG-OPENs from the live negentropy index

Milestones 2-3 of quartz/plans/2026-07-03-incremental-negentropy-storage.md. SQLiteEventStore now maintains the LiveNegentropyIndex when the strategy opts in (geode's does by default; [negentropy].live_index = false turns it off; app-side stores are untouched): - Write paths collect a LiveIndexDelta and apply it after COMMIT while still holding the writer mutex — rolled-back savepoint rows never reach the index and updates land in exact commit order (also vs the rebuild, which runs under the same mutex). - Replaceable/addressable overwrites report the row their BEFORE-INSERT trigger displaces via one indexed pre-SELECT that mirrors the trigger predicate (including the NIP-01 lowest-id tie-break and the d_tag-NULL case). - Paths that can't itemize (kind-5, vanish, delete-by-filter/id, expiration sweeps, clearDB) invalidate; the next NEG-OPEN rebuilds from one scan on the writer connection. - Until that first NEG-OPEN populates the index, ingest pays zero bookkeeping — the populated check happens under the writer mutex so it can't race the rebuild. LiveEventStore serves a single unconstrained filter (the relay-relay sync default; relayBench sends exactly this) from the index; everything else keeps the scan+seal path and its single-slot cache. Micro-benchmark at 50k events (LiveNegentropyBenchmark, in-container): scan+seal cold path 80-100 ms; index post-write open 9-16 ms (~5-10x). The relayBench A/B is the acceptance gate and comes next. Correctness: LiveNegentropyIndexStoreTest asserts index content == snapshotIdsForNegentropy scan after every mutation pattern (overwrites, losers, kind-5 rebuilds, filter deletes, mixed-outcome batches, transactions); the full geode suite (NIP-77 + interop sync tests) runs with the index on. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A
2026-08-12 01:07:46 +00:00 · 2026-07-03 23:05:23 +00:00
parent 7ad7dee5fa
commit 73ee5cce99
11 changed files with 620 additions and 30 deletions
@@ -126,7 +126,12 @@ fun main(args: Array<String>) {
        cliInfoFile?.let { RelayInfo.fromFile(it) }
            ?: config.resolveInfo(fullTextSearch)

-    val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl, indexStrategy = relayIndexingStrategy(fullTextSearch))
+    val store: IEventStore =
+        EventStore(
+            dbName = dbFile,
+            relay = advertisedUrl,
+            indexStrategy = relayIndexingStrategy(fullTextSearch, config.negentropy.live_index),
+        )

    val policyBuilder: () -> IRelayPolicy = {
        composePolicy(config, advertisedUrl, requireAuth, optionalAuth, verifySigs, parallelVerify)
@@ -40,21 +40,30 @@ import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy
 *   per-event tokenization on ingest (relayBench measured it at roughly a
 *   quarter of write cost) at the price of `search` filters matching
 *   nothing — pair it with a NIP-11 doc that doesn't advertise 50.
+ * @param liveNegentropyIndex keep the always-current `(created_at, id)`
+ *   set that serves full-corpus NIP-77 NEG-OPENs without a scan + seal
+ *   (strfry answers those off its live tree; the scan+seal path measured
+ *   ~340 ms per cold open at 50k events). Costs ~40 B/event of heap and
+ *   one indexed pre-SELECT per replaceable insert.
+ *   `[negentropy].live_index = false` turns it off.
 */
-fun relayIndexingStrategy(fullTextSearch: Boolean = true) =
-    DefaultIndexingStrategy(
-        indexEventsByCreatedAtAlone = true,
-        // Authors-only filters (no kinds) are relay-common — archives,
-        // migration tools. strfry maintains the same (pubkey, created_at)
-        // index unconditionally; without it the filter walks the whole
-        // time index.
-        indexEventsByPubkeyAlone = true,
-        indexFullTextSearch = fullTextSearch,
-        // Tokenize off the commit path; NostrServer drives the catch-up
-        // worker and search queries drain it first, so NIP-50 stays
-        // exactly as fresh while publishes stop paying for it.
-        deferFullTextSearchIndexing = fullTextSearch,
-    )
+fun relayIndexingStrategy(
+    fullTextSearch: Boolean = true,
+    liveNegentropyIndex: Boolean = true,
+) = DefaultIndexingStrategy(
+    indexEventsByCreatedAtAlone = true,
+    // Authors-only filters (no kinds) are relay-common — archives,
+    // migration tools. strfry maintains the same (pubkey, created_at)
+    // index unconditionally; without it the filter walks the whole
+    // time index.
+    indexEventsByPubkeyAlone = true,
+    indexFullTextSearch = fullTextSearch,
+    // Tokenize off the commit path; NostrServer drives the catch-up
+    // worker and search queries drain it first, so NIP-50 stays
+    // exactly as fresh while publishes stop paying for it.
+    deferFullTextSearchIndexing = fullTextSearch,
+    maintainLiveNegentropyIndex = liveNegentropyIndex,
+)

 /** Stock relay strategy — everything on, matching geode's defaults. */
 val RelayIndexingStrategy = relayIndexingStrategy()
@@ -153,6 +153,12 @@ data class StaticConfig(
        val frame_size_limit: Long = 500_000L,
        val max_sync_events: Int = 1_000_000,
        val max_sessions_per_connection: Int = 200,
+        /**
+         * Keep an always-current in-memory `(created_at, id)` set so
+         * full-corpus NEG-OPENs skip the table scan + seal (strfry
+         * parity). ~40 B per stored event of heap; on by default.
+         */
+        val live_index: Boolean = true,
    )

    data class AuthorizationSection(
@@ -381,6 +381,15 @@ class LiveEventStore(
        filters: List<Filter>,
        maxEntries: Int,
    ): IStorage? {
+        // Full-set NEG-OPENs — a single unconstrained filter, the shape
+        // relay-relay sync sends by default — are served from the store's
+        // always-current index when it maintains one: no scan, no
+        // O(n log n) seal, cold or not. `null` falls through to the scan
+        // path, which also owns the over-cap NEG-ERR detection.
+        if (filters.size == 1 && filters[0].isEmpty()) {
+            store.liveNegentropySnapshot(maxEntries)?.let { return it }
+        }
+
        val generation = writeGeneration.load()
        val key = filters.joinToString("") { it.toJson() } + "cap=$maxEntries"
        val now = TimeUtils.now()
@@ -20,6 +20,7 @@
 */
 package com.vitorpamplona.quartz.nip01Core.store

+import com.vitorpamplona.negentropy.storage.IStorage
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -152,6 +153,18 @@ interface IEventStore : AutoCloseable {
        }
    }

+    /**
+     * Sealed negentropy storage of the store's FULL set from an
+     * always-current in-memory index — the fast path for NEG-OPENs whose
+     * filter matches everything, skipping [snapshotIdsForNegentropy]'s
+     * scan and the O(n log n) seal. Returns `null` when the store keeps
+     * no such index (the default; see
+     * [com.vitorpamplona.quartz.nip01Core.store.sqlite.IndexingStrategy.maintainLiveNegentropyIndex])
+     * or the set exceeds [maxEntries] — callers fall back to the scan
+     * path either way.
+     */
+    suspend fun liveNegentropySnapshot(maxEntries: Int): IStorage? = null
+
    /**
     * True when NIP-50 tokenization is deferred and something must drive
     * [ftsCatchUp] for search to see new events. The relay server checks
@@ -194,6 +194,8 @@ class ObservableEventStore(
        maxEntries: Int?,
    ): List<IdAndTime> = inner.snapshotIdsForNegentropy(filters, maxEntries)

+    override suspend fun liveNegentropySnapshot(maxEntries: Int) = inner.liveNegentropySnapshot(maxEntries)
+
    override suspend fun delete(filter: Filter) {
        inner.delete(filter)
        _changes.emit(StoreChange.DeleteByFilter(listOf(filter)))
@@ -76,6 +76,8 @@ class EventStore(
        maxEntries: Int?,
    ): List<IdAndTime> = store.snapshotIdsForNegentropy(filters, maxEntries)

+    override suspend fun liveNegentropySnapshot(maxEntries: Int) = store.liveNegentropySnapshot(maxEntries)
+
    override val needsFtsCatchUp: Boolean get() = store.needsFtsCatchUp

    override suspend fun ftsCatchUp(batchSize: Int) = store.ftsCatchUp(batchSize)
@@ -119,6 +119,17 @@ interface IndexingStrategy {
     */
    val indexFullTextSearch: Boolean

+    /**
+     * Maintain an always-current in-memory `(created_at, id)` set (a
+     * [com.vitorpamplona.quartz.nip77Negentropy.LiveNegentropyIndex]) so
+     * NIP-77 NEG-OPENs over the full corpus skip the scan + O(n log n)
+     * seal — strfry answers those off its live tree. Costs ~40 B/event
+     * of heap plus one indexed pre-SELECT per replaceable insert, which
+     * only makes sense on a *relay*; client-side stores don't serve
+     * NEG-OPENs, so the default is **off**.
+     */
+    val maintainLiveNegentropyIndex: Boolean get() = false
+
    fun shouldIndex(
        kind: Int,
        tag: Tag,
@@ -136,6 +147,7 @@ class DefaultIndexingStrategy(
    override val useAndIndexIdOnOrderBy: Boolean = false,
    override val indexFullTextSearch: Boolean = true,
    override val deferFullTextSearchIndexing: Boolean = false,
+    override val maintainLiveNegentropyIndex: Boolean = false,
 ) : IndexingStrategy {
    override fun shouldIndex(
        kind: Int,
@@ -24,16 +24,23 @@ import androidx.sqlite.SQLiteConnection
 import androidx.sqlite.SQLiteDriver
 import androidx.sqlite.SQLiteException
 import androidx.sqlite.driver.bundled.BundledSQLiteDriver
+import com.vitorpamplona.negentropy.storage.IStorage
+import com.vitorpamplona.quartz.nip01Core.core.AddressableEvent
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.core.isAddressable
 import com.vitorpamplona.quartz.nip01Core.core.isEphemeral
+import com.vitorpamplona.quartz.nip01Core.core.isReplaceable
 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.nip09Deletions.DeletionEvent
 import com.vitorpamplona.quartz.nip40Expiration.isExpired
+import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent
+import com.vitorpamplona.quartz.nip77Negentropy.LiveNegentropyIndex

 class SQLiteEventStore(
    val driver: SQLiteDriver = BundledSQLiteDriver(),
@@ -74,6 +81,16 @@ class SQLiteEventStore(
            indexStrategy,
        )

+    /**
+     * Always-current `(created_at, id)` set for NIP-77 (see
+     * [IndexingStrategy.maintainLiveNegentropyIndex]). Populated lazily
+     * by the first [liveNegentropySnapshot]; kept current by the write
+     * paths through a [LiveIndexDelta] applied only after each
+     * transaction commits, so a rolled-back row never leaks into it.
+     */
+    val liveNegentropyIndex: LiveNegentropyIndex? =
+        if (indexStrategy.maintainLiveNegentropyIndex) LiveNegentropyIndex() else null
+
    val modules =
        listOf(
            seedModule,
@@ -182,10 +199,12 @@ class SQLiteEventStore(
        }
    }

-    suspend fun clearDB() =
+    suspend fun clearDB() {
        pool.useWriter { db ->
            modules.reversed().forEach { it.deleteAll(db) }
        }
+        liveNegentropyIndex?.invalidate()
+    }

    suspend fun vacuum() =
        pool.useWriter { db ->
@@ -225,15 +244,127 @@ class SQLiteEventStore(
            db.execSQL("ANALYZE")
        }

+    /**
+     * Live-index mutations gathered during one write transaction and
+     * applied only after its COMMIT — a rolled-back row (savepoint or
+     * whole-transaction failure) must never reach the index, and the
+     * index must never advertise an id before it is durable. `null`
+     * when [liveNegentropyIndex] is off: zero cost on the write path.
+     */
+    internal class LiveIndexDelta {
+        val added = ArrayList<IdAndTime>()
+        val removed = ArrayList<IdAndTime>()
+
+        /**
+         * Set when a row's side effects delete OTHER rows in ways this
+         * delta can't itemize (kind-5 targets, vanish-by-pubkey). The
+         * whole index is dropped and lazily rebuilt from one scan.
+         */
+        var invalidateAll = false
+    }
+
+    /**
+     * Starts delta tracking for a write transaction, or `null` when there
+     * is nothing to track — index off, or not (yet) populated. An
+     * unpopulated index is covered by its next [LiveNegentropyIndex.rebuild]
+     * scan instead, so ingest pays zero bookkeeping until the first
+     * NEG-OPEN actually builds the index.
+     *
+     * MUST be called while already holding the writer connection: the
+     * populated check races [liveNegentropySnapshot]'s rebuild otherwise
+     * (rebuild flips `populated` under the writer mutex — deciding out
+     * here could skip tracking for a transaction that commits after the
+     * rebuild's scan, silently losing its rows).
+     */
+    private fun newDeltaOrNull(): LiveIndexDelta? = if (liveNegentropyIndex?.isPopulated() == true) LiveIndexDelta() else null
+
+    /** Records one successfully inserted row into the pending delta. */
+    private fun LiveIndexDelta.recordAccepted(
+        event: Event,
+        displaced: IdAndTime?,
+    ) {
+        if (event is DeletionEvent || (event is RequestToVanishEvent && event.shouldVanishFrom(relay))) {
+            // Their own row lands too, but the rebuild scan picks it up.
+            invalidateAll = true
+            return
+        }
+        displaced?.let { removed += it }
+        added += IdAndTime(event.createdAt, event.id)
+    }
+
+    private fun applyAfterCommit(delta: LiveIndexDelta?) {
+        val index = liveNegentropyIndex ?: return
+        if (delta == null) return
+        if (delta.invalidateAll) {
+            index.invalidate()
+            return
+        }
+        delta.removed.forEach(index::remove)
+        delta.added.forEach(index::insert)
+    }
+
+    /**
+     * The row the replaceable/addressable BEFORE-INSERT trigger is about
+     * to delete for [event], or `null` when nothing will be displaced.
+     * Must run *before* the insert (the trigger fires during it) and
+     * mirrors the trigger predicates exactly — including the NIP-01
+     * lowest-id-wins tie-break. At most one row can match thanks to the
+     * partial unique indexes, and the same indexes make this lookup
+     * O(log n). Only called when the live index is on.
+     */
+    private fun displacedBy(
+        event: Event,
+        db: SQLiteConnection,
+    ): IdAndTime? {
+        // The addressable branch requires the parsed class to carry a
+        // d-tag, mirroring the header insert: events without one store
+        // d_tag NULL, and the trigger's `d_tag = NEW.d_tag` never
+        // matches NULL — so nothing gets displaced.
+        val addressable = event.kind.isAddressable() && event is AddressableEvent
+        val sql =
+            when {
+                event.kind.isReplaceable() ->
+                    """
+                    SELECT created_at, id FROM event_headers
+                    WHERE kind = ? AND pubkey = ?
+                      AND (created_at < ? OR (created_at = ? AND id > ?))
+                    """.trimIndent()
+
+                addressable ->
+                    """
+                    SELECT created_at, id FROM event_headers
+                    WHERE kind = ? AND pubkey = ? AND d_tag = ?
+                      AND kind >= 30000 AND kind < 40000
+                      AND (created_at < ? OR (created_at = ? AND id > ?))
+                    """.trimIndent()
+
+                else -> return null
+            }
+        db.prepare(sql).use { stmt ->
+            var i = 1
+            stmt.bindLong(i++, event.kind.toLong())
+            stmt.bindText(i++, event.pubKey)
+            if (addressable) stmt.bindText(i++, (event as AddressableEvent).dTag())
+            stmt.bindLong(i++, event.createdAt)
+            stmt.bindLong(i++, event.createdAt)
+            stmt.bindText(i, event.id)
+            if (!stmt.step()) return null
+            return IdAndTime(stmt.getLong(0), stmt.getText(1))
+        }
+    }
+
    private fun innerInsertEvent(
        event: Event,
        db: SQLiteConnection,
+        delta: LiveIndexDelta? = null,
    ) {
+        val displaced = if (delta != null) displacedBy(event, db) else null
        val headerId = eventIndexModule.insert(event, db)
        deletionModule.insert(event, db)
        expirationModule.insert(event, headerId, db)
        fullTextSearchModule.insert(event, headerId, db)
        rightToVanishModule.insert(event, relay, headerId, db)
+        delta?.recordAccepted(event, displaced)
    }

    suspend fun insertEvent(event: Event) {
@@ -241,9 +372,15 @@ class SQLiteEventStore(
        if (event.kind.isEphemeral()) return

        pool.useWriter { db ->
+            val delta = newDeltaOrNull()
            db.transaction {
-                innerInsertEvent(event, this)
+                innerInsertEvent(event, this, delta)
            }
+            // Still holding the writer mutex: the transaction above has
+            // committed, and applying here keeps index updates in exact
+            // commit order across every writer (and the rebuild, which
+            // also runs under this mutex).
+            applyAfterCommit(delta)
        }
    }

@@ -268,11 +405,14 @@ class SQLiteEventStore(
        if (events.isEmpty()) return emptyList()
        val outcomes = ArrayList<IEventStore.InsertOutcome>(events.size)
        pool.useWriter { db ->
+            val delta = newDeltaOrNull()
            db.transaction {
                events.forEachIndexed { i, event ->
-                    outcomes.add(insertWithSavepoint(event, i, this))
+                    outcomes.add(insertWithSavepoint(event, i, this, delta))
                }
            }
+            // Post-commit, pre-mutex-release: see insertEvent.
+            applyAfterCommit(delta)
        }
        return outcomes
    }
@@ -281,6 +421,7 @@ class SQLiteEventStore(
        event: Event,
        index: Int,
        db: SQLiteConnection,
+        delta: LiveIndexDelta?,
    ): IEventStore.InsertOutcome {
        if (event.isExpired()) {
            return IEventStore.InsertOutcome.Rejected("blocked: Cannot insert an expired event")
@@ -290,7 +431,10 @@ class SQLiteEventStore(
        val sp = "ev$index"
        db.execSQL("SAVEPOINT $sp")
        return try {
-            innerInsertEvent(event, db)
+            // The delta records this row only after every module insert
+            // succeeded (last step of innerInsertEvent), so a savepoint
+            // rollback below leaves the pending delta untouched.
+            innerInsertEvent(event, db, delta)
            db.execSQL("RELEASE SAVEPOINT $sp")
            IEventStore.InsertOutcome.Accepted
        } catch (e: Throwable) {
@@ -304,24 +448,28 @@ class SQLiteEventStore(
        }
    }

-    inner class Transaction(
+    inner class Transaction internal constructor(
        val db: SQLiteConnection,
+        private val delta: LiveIndexDelta?,
    ) : IEventStore.ITransaction {
        override fun insert(event: Event) {
            if (event.isExpired()) throw SQLiteException("blocked: Cannot insert an expired event")
            if (event.kind.isEphemeral()) return

-            innerInsertEvent(event, db)
+            innerInsertEvent(event, db, delta)
        }
    }

    suspend fun transaction(body: Transaction.() -> Unit) {
        pool.useWriter { db ->
+            val delta = newDeltaOrNull()
            db.transaction {
-                with(Transaction(this)) {
+                with(Transaction(this, delta)) {
                    body()
                }
            }
+            // Post-commit, pre-mutex-release: see insertEvent.
+            applyAfterCommit(delta)
        }
    }

@@ -366,17 +514,57 @@ class SQLiteEventStore(
        maxEntries: Int? = null,
    ): List<IdAndTime> = pool.useReader { queryBuilder.snapshotIdsForNegentropy(filters, it, maxEntries) }

-    suspend fun delete(filter: Filter) = pool.useWriter { queryBuilder.delete(filter, it) }
+    suspend fun delete(filter: Filter) {
+        pool.useWriter { queryBuilder.delete(filter, it) }
+        // Delete-by-filter can't itemize what it removed; drop the live
+        // index and let the next NEG-OPEN rebuild from one scan.
+        liveNegentropyIndex?.invalidate()
+    }

-    suspend fun delete(filters: List<Filter>) = pool.useWriter { queryBuilder.delete(filters, it) }
+    suspend fun delete(filters: List<Filter>) {
+        pool.useWriter { queryBuilder.delete(filters, it) }
+        liveNegentropyIndex?.invalidate()
+    }

-    suspend fun delete(id: HexKey): Int =
-        pool.useWriter { db ->
-            db.execSQL("DELETE FROM event_headers WHERE id = ?", arrayOf(id))
-            db.changes()
+    suspend fun delete(id: HexKey): Int {
+        val changes =
+            pool.useWriter { db ->
+                db.execSQL("DELETE FROM event_headers WHERE id = ?", arrayOf(id))
+                db.changes()
+            }
+        if (changes > 0) liveNegentropyIndex?.invalidate()
+        return changes
+    }
+
+    suspend fun deleteExpiredEvents() {
+        val swept =
+            pool.useWriter { db ->
+                expirationModule.deleteExpiredEvents(db)
+                db.changes()
+            }
+        if (swept > 0) liveNegentropyIndex?.invalidate()
+    }
+
+    /**
+     * Sealed live-index snapshot of the FULL stored set for NIP-77, or
+     * `null` when the live index is off (strategy default) or the set
+     * exceeds [maxEntries] — callers fall back to the scan path either
+     * way. The first call after boot or an invalidation rebuilds the
+     * index from one scan **on the writer connection**, so no insert can
+     * commit between the scan and the rebuild — after that, the write
+     * paths keep it current and this is a cached-snapshot lookup.
+     */
+    suspend fun liveNegentropySnapshot(maxEntries: Int): IStorage? {
+        val index = liveNegentropyIndex ?: return null
+        if (!index.isPopulated()) {
+            pool.useWriter { db ->
+                if (!index.isPopulated()) {
+                    index.rebuild(queryBuilder.snapshotIdsForNegentropy(listOf(Filter()), db, null))
+                }
+            }
        }
-
-    suspend fun deleteExpiredEvents() = pool.useWriter { expirationModule.deleteExpiredEvents(it) }
+        return index.sealedSnapshot(maxEntries)
+    }

    /**
     * Wipe and rebuild the NIP-50 full-text search index for every
@@ -0,0 +1,219 @@
+/*
+ * 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 com.vitorpamplona.quartz.nip01Core.core.Event
+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.utils.EventFactory
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+/**
+ * The live negentropy index's one correctness rule: at every point, a
+ * snapshot must advertise **exactly** the `(created_at, id)` set a fresh
+ * scan of the store would — never a deleted/displaced id (peers would
+ * fetch dead events), never missing a committed one. Every mutation
+ * pattern the write path knows is run against a store with the index on
+ * and compared to `snapshotIdsForNegentropy`.
+ */
+class LiveNegentropyIndexStoreTest {
+    private val strategy = DefaultIndexingStrategy(maintainLiveNegentropyIndex = true)
+
+    private fun hexId(seed: Int): String = seed.toString().padStart(64, '0')
+
+    private fun pubkey(seed: Int): String = seed.toString().padStart(64, 'a')
+
+    private val sig = "0".repeat(128)
+
+    private fun event(
+        idSeed: Int,
+        kind: Int = 1,
+        createdAt: Long = idSeed.toLong(),
+        pubKey: String = pubkey(1),
+        tags: Array<Array<String>> = emptyArray(),
+        content: String = "",
+    ): Event = EventFactory.create(hexId(idSeed), pubKey, createdAt, kind, tags, content, sig)
+
+    private fun newStore() = EventStore(dbName = null, indexStrategy = strategy)
+
+    /** Index content and scan content must be identical, entry for entry. */
+    private suspend fun assertIndexMatchesScan(store: EventStore) {
+        val snapshot = assertNotNull(store.liveNegentropySnapshot(Int.MAX_VALUE), "index should serve a snapshot")
+        val fromIndex = snapshot.map { IdAndTime(it.timestamp, it.id.toHexString()) }
+        val fromScan =
+            store
+                .snapshotIdsForNegentropy(listOf(Filter()), null)
+                .sortedWith(compareBy({ it.createdAt }, { it.id }))
+        assertEquals(fromScan, fromIndex)
+    }
+
+    @Test
+    fun populatesLazilyAndTracksPlainInserts() =
+        runTest {
+            val store = newStore()
+            store.insert(event(1))
+            store.insert(event(2))
+
+            // First snapshot rebuilds from a scan…
+            assertIndexMatchesScan(store)
+
+            // …and later inserts are tracked incrementally.
+            store.insert(event(3))
+            store.batchInsert(listOf(event(4), event(5)))
+            assertIndexMatchesScan(store)
+
+            store.close()
+        }
+
+    @Test
+    fun replaceableOverwriteDropsTheDisplacedId() =
+        runTest {
+            val store = newStore()
+            store.insert(event(1, kind = 0, createdAt = 100, pubKey = pubkey(7)))
+            assertIndexMatchesScan(store)
+
+            // Newer kind-0 from the same author displaces the old row.
+            store.insert(event(2, kind = 0, createdAt = 200, pubKey = pubkey(7)))
+            assertIndexMatchesScan(store)
+            assertEquals(1, store.count(Filter(kinds = listOf(0))))
+
+            // An older one loses to the stored row and must not disturb
+            // the index (the store rejects it).
+            runCatching { store.insert(event(3, kind = 0, createdAt = 50, pubKey = pubkey(7))) }
+            assertIndexMatchesScan(store)
+
+            store.close()
+        }
+
+    @Test
+    fun addressableOverwriteDropsOnlyTheSameDTag() =
+        runTest {
+            val store = newStore()
+            val author = pubkey(9)
+            store.insert(event(1, kind = 30023, createdAt = 100, pubKey = author, tags = arrayOf(arrayOf("d", "post-a"))))
+            store.insert(event(2, kind = 30023, createdAt = 100, pubKey = author, tags = arrayOf(arrayOf("d", "post-b"))))
+            assertIndexMatchesScan(store)
+
+            // Overwrites post-a; post-b must survive.
+            store.insert(event(3, kind = 30023, createdAt = 200, pubKey = author, tags = arrayOf(arrayOf("d", "post-a"))))
+            assertIndexMatchesScan(store)
+            assertEquals(2, store.count(Filter(kinds = listOf(30023))))
+
+            store.close()
+        }
+
+    @Test
+    fun deletionEventInvalidatesAndRebuildMatches() =
+        runTest {
+            val store = newStore()
+            val author = pubkey(3)
+            val target = event(1, createdAt = 100, pubKey = author)
+            store.insert(target)
+            store.insert(event(2, createdAt = 110, pubKey = author))
+            assertIndexMatchesScan(store)
+
+            // Kind-5 removes the target AND stores itself; the index
+            // can't itemize that, so it must rebuild — and match.
+            store.insert(
+                event(4, kind = 5, createdAt = 120, pubKey = author, tags = arrayOf(arrayOf("e", target.id))),
+            )
+            assertIndexMatchesScan(store)
+            assertTrue(store.query<Event>(Filter(ids = listOf(target.id))).isEmpty())
+
+            store.close()
+        }
+
+    @Test
+    fun deleteByFilterInvalidatesAndRebuildMatches() =
+        runTest {
+            val store = newStore()
+            store.insert(event(1, kind = 1))
+            store.insert(event(2, kind = 7))
+            assertIndexMatchesScan(store)
+
+            store.delete(Filter(kinds = listOf(7)))
+            assertIndexMatchesScan(store)
+
+            store.close()
+        }
+
+    @Test
+    fun rejectedRowsInABatchNeverEnterTheIndex() =
+        runTest {
+            val store = newStore()
+            store.insert(event(1))
+            assertIndexMatchesScan(store)
+
+            // Duplicate id (rejected) mixed with a fresh row (accepted).
+            val outcomes = store.batchInsert(listOf(event(1), event(2)))
+            assertTrue(outcomes[0] is IEventStore.InsertOutcome.Rejected)
+            assertEquals(IEventStore.InsertOutcome.Accepted, outcomes[1])
+            assertIndexMatchesScan(store)
+
+            store.close()
+        }
+
+    @Test
+    fun transactionInsertsLandAfterCommit() =
+        runTest {
+            val store = newStore()
+            store.insert(event(1))
+            assertIndexMatchesScan(store)
+
+            store.transaction {
+                insert(event(2))
+                insert(event(3))
+            }
+            assertIndexMatchesScan(store)
+
+            store.close()
+        }
+
+    @Test
+    fun overCapFallsBackToNull() =
+        runTest {
+            val store = newStore()
+            store.insert(event(1))
+            store.insert(event(2))
+            store.insert(event(3))
+            assertIndexMatchesScan(store)
+
+            assertNull(store.liveNegentropySnapshot(2))
+            assertNotNull(store.liveNegentropySnapshot(3))
+
+            store.close()
+        }
+
+    @Test
+    fun defaultStrategyKeepsNoIndex() =
+        runTest {
+            val store = EventStore(dbName = null)
+            store.insert(event(1))
+            assertNull(store.liveNegentropySnapshot(Int.MAX_VALUE))
+            store.close()
+        }
+}
@@ -0,0 +1,125 @@
+/*
+ * 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.prodbench
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession
+import com.vitorpamplona.quartz.utils.EventFactory
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+/**
+ * Measures what the live negentropy index buys on a NEG-OPEN against the
+ * scan + O(n log n) seal it replaces, at the relayBench corpus size where
+ * the cold path measured ~340 ms (50k events; strfry serves ~21 ms).
+ *
+ * Three numbers, printed for the record:
+ *  1. scan+seal — the pre-index cold path (`snapshotIdsForNegentropy` +
+ *     `sealVector`), what every open pays when the single-slot cache
+ *     misses.
+ *  2. index first-open — includes the one-time lazy rebuild scan.
+ *  3. index post-write open — the steady state on a busy relay: a write
+ *     lands (invalidating any memoized snapshot), then a NEG-OPEN
+ *     arrives. This is the number that was previously ~scan+seal and is
+ *     the point of the feature.
+ *
+ * No speed assertion — container noise makes ratios flaky; correctness
+ * (identical content between both paths) is asserted instead. Timings
+ * are informational and the real A/B happens in relayBench.
+ */
+class LiveNegentropyBenchmark {
+    companion object {
+        const val EVENTS = 50_000
+        const val POST_WRITE_OPENS = 50
+    }
+
+    private fun hexId(seed: Int): String = seed.toString(16).padStart(64, '0')
+
+    private fun pubkey(seed: Int): String = (seed % 512).toString(16).padStart(64, 'a')
+
+    private val sig = "0".repeat(128)
+
+    private fun event(seed: Int): Event =
+        EventFactory.create(
+            id = hexId(seed),
+            pubKey = pubkey(seed),
+            createdAt = 1_600_000_000L + (seed * 7919) % 1_000_000, // scattered, not pre-sorted
+            kind = 1,
+            tags = emptyArray(),
+            content = "live negentropy benchmark $seed",
+            sig = sig,
+        )
+
+    @Test
+    fun coldOpenVsLiveIndexAt50k() =
+        runBlocking {
+            val store = EventStore(dbName = null, indexStrategy = DefaultIndexingStrategy(maintainLiveNegentropyIndex = true))
+
+            (1..EVENTS).chunked(2000).forEach { chunk ->
+                store.batchInsert(chunk.map { event(it) })
+            }
+
+            // 1. The pre-index cold path.
+            var scanSealNanos = 0L
+            var scanned = 0
+            run {
+                val t0 = System.nanoTime()
+                val entries = store.snapshotIdsForNegentropy(listOf(Filter()), null)
+                val sealed = NegentropyServerSession.sealVector(entries)
+                scanSealNanos = System.nanoTime() - t0
+                scanned = sealed.size()
+            }
+
+            // 2. First index open: pays the lazy rebuild once.
+            val t1 = System.nanoTime()
+            val first = assertNotNull(store.liveNegentropySnapshot(Int.MAX_VALUE))
+            val firstOpenNanos = System.nanoTime() - t1
+            assertEquals(scanned, first.size())
+
+            // 3. Steady state: one write per open, so every open pays the
+            // index's real per-open cost (copy + seal of sorted data),
+            // never the memoized-snapshot freebie.
+            var postWriteNanos = 0L
+            for (i in 0 until POST_WRITE_OPENS) {
+                store.insert(event(EVENTS + 1 + i))
+                val t2 = System.nanoTime()
+                assertNotNull(store.liveNegentropySnapshot(Int.MAX_VALUE))
+                postWriteNanos += System.nanoTime() - t2
+            }
+            val postWriteAvgMs = postWriteNanos / POST_WRITE_OPENS / 1e6
+
+            // Correctness cross-check after all the churn.
+            val finalSnapshot = assertNotNull(store.liveNegentropySnapshot(Int.MAX_VALUE))
+            assertEquals(EVENTS + POST_WRITE_OPENS, finalSnapshot.size())
+
+            println("LiveNegentropyBenchmark @ ${EVENTS / 1000}k events")
+            println("  scan+seal cold path:   ${"%8.2f".format(scanSealNanos / 1e6)} ms")
+            println("  index first open:      ${"%8.2f".format(firstOpenNanos / 1e6)} ms (includes one-time rebuild)")
+            println("  index post-write open: ${"%8.2f".format(postWriteAvgMs)} ms (avg of $POST_WRITE_OPENS)")
+
+            store.close()
+        }
+}