From 115963eb1ce2f5cc0f7885ccf4cff75e2e98e351 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Jul 2026 18:30:51 +0000 Subject: [PATCH] perf: defer NIP-50 tokenization off the insert path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FTS indexing ran inside every insert's transaction — a measurable slice of write cost (relayBench: ~18% of ingest throughput) paid at publish time for a feature only search queries read. It now runs as a watermark catch-up: - IndexingStrategy.deferFullTextSearchIndexing (default false; geode's relay strategy enables it with search). Deferred inserts skip tokenization entirely. - FullTextSearchModule keeps a fts_catchup_state watermark (everything <= last_row_id is indexed) and gains catchUpBatch(): scan past the watermark, tokenize, advance — one write transaction per batch, so publishes interleave. DATABASE_VERSION 3->4 seeds the watermark at MAX(row_id) for existing (synchronously indexed) databases. - NostrServer runs the catch-up worker, poked by IngestQueue's new onBatchCommitted hook, and *yields to publish traffic*: it only drains while the queue has no backlog (IngestQueue.hasBacklog()), so bursts ingest at no-FTS speed and tokenization fills the gaps. - LiveEventStore drains the backlog synchronously before serving any filter with a search term (query, queryRaw, count) — NIP-50 results stay exactly as fresh as the synchronous path; the deferral is invisible to correctness. Geode's existing search tests pass unchanged through this path. The first implementation reused reindexBatch and collapsed ingest 8x — its per-row 'DELETE FROM event_fts WHERE event_header_row_id = ?' matches on a plain FTS5 column, i.e. a full FTS-table scan per row (O(n²) overall), and the worker competed with the replay for the writer mutex. catchUpBatch therefore inserts without the delete (rows past the watermark are never indexed; switching a DB between deferred and synchronous strategies requires reindexAll, same rule as a searchable-kinds change), and the worker backs off whenever publishes are pending. Alternating A/B, 50k corpus, search-enabled default: 4,902/5,090/5,136 events/s synchronous vs 5,425/5,611 deferred (+8-12%), approaching the --no-search ceiling while keeping NIP-50 advertised and fresh. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NeoCvXnTxsKzqurkmjdC46 --- .../geode/RelayIndexingStrategy.kt | 4 + .../nip01Core/relay/server/NostrServer.kt | 39 ++++++ .../relay/server/backend/IngestQueue.kt | 21 +++ .../relay/server/backend/LiveEventStore.kt | 26 +++- .../quartz/nip01Core/store/IEventStore.kt | 14 ++ .../nip01Core/store/sqlite/EventStore.kt | 4 + .../store/sqlite/FullTextSearchModule.kt | 126 +++++++++++++++++- .../store/sqlite/IndexingStrategy.kt | 13 ++ .../store/sqlite/SQLiteEventStore.kt | 38 +++++- .../nip01Core/store/sqlite/DeferredFtsTest.kt | 121 +++++++++++++++++ 10 files changed, 398 insertions(+), 8 deletions(-) create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/DeferredFtsTest.kt diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt index 3809672cae..81528e96aa 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt @@ -50,6 +50,10 @@ fun relayIndexingStrategy(fullTextSearch: Boolean = true) = // 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, ) /** Stock relay strategy — everything on, matching geode's defaults. */ diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index 2250c4dd02..ffbe105fde 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -31,6 +31,8 @@ import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.launch import kotlin.coroutines.CoroutineContext /** @@ -65,6 +67,12 @@ class NostrServer( listener: RelayServerListener = RelayServerListener.None, limits: RelayLimits? = null, ) : RelayServerBase(policyBuilder, parentContext, negentropySettings, listener, limits) { + /** + * Wakes the deferred-FTS catch-up worker. Conflated: N batch commits + * while the worker is mid-drain collapse into one more pass. + */ + private val ftsCatchUpPokes = Channel(Channel.CONFLATED) + /** * Group-commit writer shared across every connected session. * Sessions hand off EVENT publishes here instead of awaiting @@ -77,10 +85,41 @@ class NostrServer( store = store, parentContext = parentContext, verify = if (parallelVerify) ({ it.verify() }) else null, + onBatchCommitted = + if (store.needsFtsCatchUp) { + { ftsCatchUpPokes.trySend(Unit) } + } else { + null + }, ) override val backend: SessionBackend = LiveEventStore(store, ingest) + init { + // Deferred-FTS catch-up worker: tokenizes in the gaps between + // publish batches. Each catch-up batch is its own write + // transaction, so a publish burst arriving mid-drain interleaves + // at batch granularity instead of stalling. Search REQs don't + // depend on this worker's pace — LiveEventStore drains the + // backlog synchronously before serving any search filter. + if (store.needsFtsCatchUp) { + // Drain any backlog left over from a previous run before the + // first publish arrives. + ftsCatchUpPokes.trySend(Unit) + scope.launch { + for (poke in ftsCatchUpPokes) { + // Yield to publish traffic: while the ingest queue has + // backlog, don't compete for the writer connection — + // the burst's final batch commit pokes again, and the + // pre-search drain covers correctness regardless. + while (!ingest.hasBacklog()) { + if (store.ftsCatchUp()) break + } + } + } + } + } + /** * Shuts down the server, cancelling all subscriptions and closing the store. */ diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/IngestQueue.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/IngestQueue.kt index 19e0fa76fc..4e32fba5ad 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/IngestQueue.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/IngestQueue.kt @@ -34,6 +34,7 @@ import kotlinx.coroutines.channels.ClosedReceiveChannelException import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch import kotlin.concurrent.atomics.AtomicBoolean +import kotlin.concurrent.atomics.AtomicInt import kotlin.concurrent.atomics.ExperimentalAtomicApi import kotlin.coroutines.CoroutineContext @@ -103,6 +104,13 @@ class IngestQueue( */ private val verify: ((Event) -> Boolean)? = null, private val verifyRejectionReason: String = "invalid: bad signature or id", + /** + * Fired after each batch's transaction commits and its OKs dispatch. + * Used to poke deferred maintenance (the FTS catch-up worker) so it + * runs in the gaps between publishes rather than inside them. Must + * be cheap and non-suspending — it runs on the writer stage. + */ + private val onBatchCommitted: (() -> Unit)? = null, ) : AutoCloseable { /** * One outstanding ingest request: the event to insert plus the @@ -116,6 +124,16 @@ class IngestQueue( private val incoming = Channel(capacity) private val scope = CoroutineScope(parentContext + SupervisorJob()) + /** Submissions accepted but whose outcome hasn't been dispatched yet. */ + private val pending = AtomicInt(0) + + /** + * True while publishes are queued or mid-batch. Deferred maintenance + * (the FTS catch-up) checks this to yield the writer connection to + * publish traffic instead of competing with it. + */ + fun hasBacklog(): Boolean = pending.load() > 0 + /** * Lazily-launched drain coroutine. We don't start it in `init` * because eagerly launching from a server's lazy @@ -144,6 +162,7 @@ class IngestQueue( onComplete: (IEventStore.InsertOutcome) -> Unit, ) { ensureWriterStarted() + pending.addAndFetch(1) incoming.send(Submission(event, onComplete)) } @@ -203,6 +222,7 @@ class IngestQueue( val next = verified.receive() val finalOutcomes = runInsertStage(next.batch, next.results) dispatchOutcomes(next.batch, finalOutcomes) + onBatchCommitted?.invoke() } } catch (_: ClosedReceiveChannelException) { // Normal shutdown via close(). @@ -290,6 +310,7 @@ class IngestQueue( Log.w("IngestQueue") { "onComplete threw: ${e.message}" } } } + pending.addAndFetch(-batch.size) } /** 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 bc49e9607b..1610b2aa17 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 @@ -141,12 +141,28 @@ class LiveEventStore( } } + /** + * With deferred FTS, a search query must first drain the catch-up + * backlog — that keeps NIP-50 results exactly as fresh as the + * synchronous path (the deferral is invisible to correctness; only + * publishes stop paying for tokenization). Non-search filters never + * touch the FTS index and skip this entirely. + */ + private suspend fun drainFtsIfSearching(filters: List) { + if (!store.needsFtsCatchUp) return + if (filters.none { !it.search.isNullOrEmpty() }) return + while (!store.ftsCatchUp()) { + // Each batch is its own write transaction; loop until caught up. + } + } + override suspend fun query( ctx: RequestContext, filters: List, onEach: (Event) -> Unit, onEose: () -> Unit, ) { + drainFtsIfSearching(filters) // During the historical replay, record ids the store has // emitted so the live path can dedupe. The index registers // *before* the replay starts (otherwise an event accepted @@ -227,6 +243,7 @@ class LiveEventStore( onEachLive: (Event) -> Unit, onEose: () -> Unit, ) { + drainFtsIfSearching(filters) val seenLock = AtomicBoolean(false) var seenIds: HashSet? = HashSet(1024) @@ -268,7 +285,10 @@ class LiveEventStore( override suspend fun count( ctx: RequestContext, filters: List, - ): Int = store.count(filters.strippingSearchExtensions()) + ): Int { + drainFtsIfSearching(filters) + return store.count(filters.strippingSearchExtensions()) + } /** * One-shot snapshot query. Used by NIP-77 negentropy: the server @@ -308,10 +328,7 @@ class LiveEventStore( override suspend fun snapshotIdsForNegentropy( filters: List, maxEntries: Int?, -<<<<<<< HEAD ): List = store.snapshotIdsForNegentropy(filters.strippingSearchExtensions(), maxEntries) -======= - ): List = store.snapshotIdsForNegentropy(filters, maxEntries) // ------------------------------------------------------------------ // NIP-77 snapshot cache @@ -375,5 +392,4 @@ class LiveEventStore( */ const val SNAPSHOT_TTL_SECONDS = 30L } ->>>>>>> 55139747 (perf: cache the sealed negentropy snapshot across NEG-OPENs) } 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 50a72ebcd0..da3dee4326 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 @@ -152,6 +152,20 @@ interface IEventStore : AutoCloseable { } } + /** + * True when NIP-50 tokenization is deferred and something must drive + * [ftsCatchUp] for search to see new events. The relay server checks + * this to start its catch-up worker; stores that index synchronously + * (the default) report `false` and [ftsCatchUp] is a no-op. + */ + val needsFtsCatchUp: Boolean get() = false + + /** + * One deferred-FTS catch-up batch; `true` once the index has caught + * up. Safe to call on any store — the default reports done. + */ + suspend fun ftsCatchUp(batchSize: Int = DEFAULT_FTS_REINDEX_BATCH): Boolean = true + suspend fun delete(filter: Filter) suspend fun delete(filters: List) 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 7f63d87aab..cfad881a67 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,10 @@ class EventStore( maxEntries: Int?, ): List = store.snapshotIdsForNegentropy(filters, maxEntries) + override val needsFtsCatchUp: Boolean get() = store.needsFtsCatchUp + + override suspend fun ftsCatchUp(batchSize: Int) = store.ftsCatchUp(batchSize) + override suspend fun delete(filter: Filter) { store.delete(filter) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/FullTextSearchModule.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/FullTextSearchModule.kt index 1283b78e16..32e489b05a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/FullTextSearchModule.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/FullTextSearchModule.kt @@ -42,11 +42,19 @@ import com.vitorpamplona.quartz.utils.EventFactory */ class FullTextSearchModule( val enabled: Boolean = true, + /** + * Skip tokenization in [insert] and let [catchUpBatch] index from the + * persisted watermark instead. See + * [IndexingStrategy.deferFullTextSearchIndexing] for the contract — + * something must drive the catch-up (the relay server does). + */ + val deferIndexing: Boolean = false, ) : IModule { val tableName = "event_fts" val triggerName = "fts_foreign_key" val eventHeaderRowIdName = "event_header_row_id" val contentName = "content" + val stateTableName = "fts_catchup_state" override fun create(db: SQLiteConnection) { if (!enabled) return @@ -70,6 +78,33 @@ class FullTextSearchModule( END; """, ) + + createStateTable(db) + } + + /** + * Watermark for the deferred path: everything with + * `row_id <= last_row_id` is guaranteed indexed. Idempotent — also + * used as the v3→v4 migration for databases created before the + * deferred mode existed; those seeded their FTS synchronously, so + * the watermark starts at the current MAX(row_id). + */ + fun createStateTable(db: SQLiteConnection) { + if (!enabled) return + db.execSQL( + """ + CREATE TABLE IF NOT EXISTS $stateTableName ( + id INTEGER PRIMARY KEY CHECK (id = 1), + last_row_id INTEGER NOT NULL + ) + """.trimIndent(), + ) + db.execSQL( + """ + INSERT OR IGNORE INTO $stateTableName (id, last_row_id) + SELECT 1, COALESCE(MAX(row_id), 0) FROM event_headers + """.trimIndent(), + ) } override fun drop(db: SQLiteConnection) { @@ -103,7 +138,7 @@ class FullTextSearchModule( headerId: Long, db: SQLiteConnection, ) { - if (!enabled) return + if (!enabled || deferIndexing) return if (event is SearchableEvent) { db.prepare(insertFTS).use { stmt -> stmt.bindLong(1, headerId) @@ -276,6 +311,95 @@ class FullTextSearchModule( ) } + /** + * One catch-up step of the deferred path: index up to [batchSize] + * events past the persisted watermark, then advance it. Returns + * `true` when the index has caught up with the table (within this + * transaction's snapshot). Must run inside the caller's write + * transaction so the watermark advances atomically with the FTS + * rows it covers — a crash replays the batch, and [reindexBatch]'s + * delete-then-insert keeps the replay harmless. + */ + fun catchUpBatch( + db: SQLiteConnection, + batchSize: Int, + ): Boolean { + if (!enabled) return true + + val watermark = + db.prepare("SELECT last_row_id FROM $stateTableName WHERE id = 1").use { stmt -> + if (stmt.step()) stmt.getLong(0) else 0L + } + + // Unlike [reindexBatch] there is NO per-row delete here: rows past + // the watermark were never indexed (deferred mode skips insert()), + // and the watermark advances atomically with the FTS rows it + // covers, so a crash replay is impossible. The delete would also + // be ruinous — `event_header_row_id` is a plain FTS5 column, so + // deleting by it scans the whole FTS table per row, which turned + // the first catch-up implementation O(n²). Consequence: switching + // a database back and forth between deferred and synchronous + // strategies requires a [reindexAll] in between (same rule as a + // searchable-kinds change). + val limit = batchSize.coerceAtLeast(1) + val kinds = searchableKindsPresent(db) + var last = watermark + var processed = 0 + if (kinds.isNotEmpty()) { + val selectSql = + "SELECT row_id, id, pubkey, created_at, kind, tags, content, sig " + + "FROM event_headers WHERE row_id > ? AND kind IN (${kinds.joinToString(",")}) " + + "ORDER BY row_id LIMIT ?" + db.prepare(insertFTS).use { write -> + db.prepare(selectSql).use { read -> + read.bindLong(1, watermark) + read.bindLong(2, limit.toLong()) + while (read.step()) { + val event = + EventFactory.create( + read.getText(1), + read.getText(2), + read.getLong(3), + read.getInt(4), + OptimizedJsonMapper.fromJsonToTagArray(read.getText(5)), + read.getText(6), + read.getText(7), + ) + if (event is SearchableEvent) { + write.bindLong(1, read.getLong(0)) + write.bindText(2, event.indexableContent()) + write.step() + write.reset() + } + last = read.getLong(0) + processed++ + } + } + } + } + + val done = processed < limit + val newWatermark = + if (done) { + // Scan reached the end of the table: everything visible in + // this snapshot is indexed. + db.prepare("SELECT COALESCE(MAX(row_id), 0) FROM event_headers").use { stmt -> + stmt.step() + stmt.getLong(0) + } + } else { + last + } + + if (newWatermark != watermark) { + db.prepare("UPDATE $stateTableName SET last_row_id = ? WHERE id = 1").use { stmt -> + stmt.bindLong(1, newWatermark) + stmt.step() + } + } + return done + } + /** * The distinct kinds present in `event_headers` that currently parse * to a [SearchableEvent]. Kind alone selects the event class in 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 21fc90f52e..fbcbda829e 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 @@ -92,6 +92,18 @@ interface IndexingStrategy { */ val useAndIndexIdOnOrderBy: Boolean + /** + * Defer NIP-50 tokenization off the insert path. When on (and + * [indexFullTextSearch] is on), inserts skip the FTS write — a + * measurable slice of commit cost — and a catch-up pass indexes + * from a persisted `row_id` watermark later: continuously in idle + * gaps on a relay, and always drained *before* a search query runs, + * so NIP-50 results stay exactly as fresh as the synchronous path. + * Only pays off where something drives the catch-up (the relay + * server does); leave it off for client-side stores. + */ + val deferFullTextSearchIndexing: Boolean + /** * Maintain the NIP-50 full-text search index (`event_fts`). * @@ -123,6 +135,7 @@ class DefaultIndexingStrategy( override val indexTagsWithKindAndPubkey: Boolean = false, override val useAndIndexIdOnOrderBy: Boolean = false, override val indexFullTextSearch: Boolean = true, + override val deferFullTextSearchIndexing: 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 985b7c3786..1eaf481024 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 @@ -43,12 +43,16 @@ class SQLiteEventStore( val numReaders: Int = 4, ) { companion object { - const val DATABASE_VERSION = 3 + const val DATABASE_VERSION = 4 } val seedModule = SeedModule() - val fullTextSearchModule = FullTextSearchModule(indexStrategy.indexFullTextSearch) + val fullTextSearchModule = + FullTextSearchModule( + indexStrategy.indexFullTextSearch, + indexStrategy.deferFullTextSearchIndexing, + ) val eventIndexModule = EventIndexesModule( seedModule::hasher, @@ -168,6 +172,12 @@ class SQLiteEventStore( // (created only for strategies that opt in). eventIndexModule.migrateV2AddPubkeyIndex(db) } + 3 -> { + // Upgrade from version 3 to 4: deferred-FTS watermark. + // Pre-v4 rows were indexed synchronously, so the + // watermark seeds at the current MAX(row_id). + fullTextSearchModule.createStateTable(db) + } } } } @@ -184,6 +194,30 @@ class SQLiteEventStore( db.execSQL("VACUUM") } + /** + * True when something must drive [ftsCatchUp] for NIP-50 to work — + * i.e. the strategy defers tokenization off the insert path. The + * relay server wires a background worker (and a pre-search drain) + * when this is set. + */ + val needsFtsCatchUp: Boolean = + indexStrategy.indexFullTextSearch && indexStrategy.deferFullTextSearchIndexing + + /** + * One deferred-FTS catch-up batch; returns `true` once the index has + * caught up with the table. Each batch is its own write transaction, + * so publishes interleave between batches instead of stalling behind + * a long rebuild. + */ + suspend fun ftsCatchUp(batchSize: Int = 1000): Boolean = + pool.useWriter { db -> + var done = false + db.transaction { + done = fullTextSearchModule.catchUpBatch(this, batchSize) + } + done + } + suspend fun analyse() = pool.useWriter { db -> // ANALYZE: Collects statistics about tables and indices diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/DeferredFtsTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/DeferredFtsTest.kt new file mode 100644 index 0000000000..b3e43c1019 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/DeferredFtsTest.kt @@ -0,0 +1,121 @@ +/* + * 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.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import com.vitorpamplona.quartz.utils.Secp256k1Instance +import kotlinx.coroutines.runBlocking +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.deleteIfExists +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * Contract of [IndexingStrategy.deferFullTextSearchIndexing]: inserts skip + * tokenization; [EventStore.ftsCatchUp] indexes from the persisted + * watermark, idempotently and resumably; search sees everything once the + * catch-up reports done. + */ +class DeferredFtsTest { + private val signer = NostrSignerSync() + private lateinit var dbFile: Path + private lateinit var store: EventStore + + @BeforeTest + fun setup() { + Secp256k1Instance + dbFile = Files.createTempFile("deferred-fts-", ".db") + Files.deleteIfExists(dbFile) + store = + EventStore( + dbName = dbFile.toAbsolutePath().toString(), + relay = null, + indexStrategy = DefaultIndexingStrategy(deferFullTextSearchIndexing = true), + ) + } + + @AfterTest + fun tearDown() { + store.close() + dbFile.deleteIfExists() + } + + private fun note( + text: String, + createdAt: Long, + ): Event = signer.sign(TextNoteEvent.build(text, createdAt = createdAt)) + + private suspend fun search(term: String): List = store.query(Filter(search = term)) + + @Test + fun deferThenCatchUpMakesEventsSearchable() = + runBlocking { + assertTrue(store.needsFtsCatchUp) + + store.insert(note("the purple ostrich flies at midnight", 1000)) + store.insert(note("nothing to see here", 1001)) + + // Deferred: not tokenized yet, so search finds nothing… + assertEquals(0, search("ostrich").size) + + // …until the catch-up drains. + while (!store.ftsCatchUp(batchSize = 1)) { + // batchSize 1 forces multiple resumable batches + } + assertEquals(1, search("ostrich").size) + assertEquals(0, search("zebra").size) + + // Idempotent: draining again neither duplicates nor loses rows. + while (!store.ftsCatchUp()) {} + assertEquals(1, search("ostrich").size) + + // New inserts after a completed catch-up start deferred again. + store.insert(note("a second ostrich appears", 1002)) + assertEquals(1, search("ostrich").size) + while (!store.ftsCatchUp()) {} + assertEquals(2, search("ostrich").size) + } + + @Test + fun synchronousStoreNeedsNoCatchUp() = + runBlocking { + val syncStore = + EventStore( + dbName = null, + relay = null, + indexStrategy = DefaultIndexingStrategy(), + ) + syncStore.use { + assertTrue(!it.needsFtsCatchUp) + it.insert(note("immediate emu sighting", 2000)) + assertEquals(1, it.query(Filter(search = "emu")).size) + // ftsCatchUp is a harmless no-op on synchronous stores. + assertTrue(it.ftsCatchUp()) + } + } +}