From 73ee5cce994bb28199d29d277194239eabf602f8 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Jul 2026 23:05:23 +0000 Subject: [PATCH] perf(quartz/geode): serve full-set NEG-OPENs from the live negentropy index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A --- .../kotlin/com/vitorpamplona/geode/Main.kt | 7 +- .../geode/RelayIndexingStrategy.kt | 37 +-- .../geode/config/StaticConfig.kt | 6 + .../relay/server/backend/LiveEventStore.kt | 9 + .../quartz/nip01Core/store/IEventStore.kt | 13 ++ .../nip01Core/store/ObservableEventStore.kt | 2 + .../nip01Core/store/sqlite/EventStore.kt | 2 + .../store/sqlite/IndexingStrategy.kt | 12 + .../store/sqlite/SQLiteEventStore.kt | 218 +++++++++++++++-- .../sqlite/LiveNegentropyIndexStoreTest.kt | 219 ++++++++++++++++++ .../prodbench/LiveNegentropyBenchmark.kt | 125 ++++++++++ 11 files changed, 620 insertions(+), 30 deletions(-) create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/LiveNegentropyIndexStoreTest.kt create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/LiveNegentropyBenchmark.kt diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index d096b34f8f..19e470853b 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -126,7 +126,12 @@ fun main(args: Array) { 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) diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt index 81528e96aa..9b3a116f3f 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt @@ -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() diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt index 7191da08f3..7079a3fa28 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt @@ -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( diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt index 6bed5eae8d..1d35362134 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt @@ -381,6 +381,15 @@ class LiveEventStore( filters: List, 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() 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 da3dee4326..28122a586d 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 @@ -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 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 35e0427f25..ca0d372d6d 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 @@ -194,6 +194,8 @@ class ObservableEventStore( maxEntries: Int?, ): List = 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))) 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 cfad881a67..82d1df1c0b 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 @@ -76,6 +76,8 @@ class EventStore( maxEntries: Int?, ): List = 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) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/IndexingStrategy.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/IndexingStrategy.kt index fbcbda829e..aa7d38ca25 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/IndexingStrategy.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/IndexingStrategy.kt @@ -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, 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 1eaf481024..2a802b4c23 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 @@ -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() + val removed = ArrayList() + + /** + * 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(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 = 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) = pool.useWriter { queryBuilder.delete(filters, it) } + suspend fun delete(filters: List) { + 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 diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/LiveNegentropyIndexStoreTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/LiveNegentropyIndexStoreTest.kt new file mode 100644 index 0000000000..3102f162de --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/LiveNegentropyIndexStoreTest.kt @@ -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> = 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(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() + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/LiveNegentropyBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/LiveNegentropyBenchmark.kt new file mode 100644 index 0000000000..5abfc1672b --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/LiveNegentropyBenchmark.kt @@ -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() + } +}