mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-09 08:04:45 +00:00
perf: defer NIP-50 tokenization off the insert path
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NeoCvXnTxsKzqurkmjdC46
This commit is contained in:
Claude
@@ -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. */
|
||||
|
||||
+39
@@ -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<Unit>(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.
|
||||
*/
|
||||
|
||||
+21
@@ -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<Submission>(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)
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+21
-5
@@ -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<Filter>) {
|
||||
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<Filter>,
|
||||
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<String>? = HashSet(1024)
|
||||
|
||||
@@ -268,7 +285,10 @@ class LiveEventStore(
|
||||
override suspend fun count(
|
||||
ctx: RequestContext,
|
||||
filters: List<Filter>,
|
||||
): 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<Filter>,
|
||||
maxEntries: Int?,
|
||||
<<<<<<< HEAD
|
||||
): List<IdAndTime> = store.snapshotIdsForNegentropy(filters.strippingSearchExtensions(), maxEntries)
|
||||
=======
|
||||
): List<IdAndTime> = 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)
|
||||
}
|
||||
|
||||
@@ -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<Filter>)
|
||||
|
||||
+4
@@ -76,6 +76,10 @@ class EventStore(
|
||||
maxEntries: Int?,
|
||||
): List<IdAndTime> = 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)
|
||||
}
|
||||
|
||||
+125
-1
@@ -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<Event>(
|
||||
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
|
||||
|
||||
+13
@@ -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,
|
||||
|
||||
+36
-2
@@ -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
|
||||
|
||||
+121
@@ -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<Event> = 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<Event>(Filter(search = "emu")).size)
|
||||
// ftsCatchUp is a harmless no-op on synchronous stores.
|
||||
assertTrue(it.ftsCatchUp())
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user