diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/relay/LocalRelayStore.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/relay/LocalRelayStore.kt index 566df9e006..1300067464 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/relay/LocalRelayStore.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/relay/LocalRelayStore.kt @@ -26,6 +26,7 @@ import com.vitorpamplona.amethyst.desktop.cache.DesktopLocalCache import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.CoroutineScope @@ -42,6 +43,18 @@ class LocalRelayStore( ) : AutoCloseable { companion object { val LOCAL_RELAY_URL: NormalizedRelayUrl = NormalizedRelayUrl("ws://localhost/amethyst-local/") + + /** + * Client defaults plus the authors-without-kinds index: shared + * ViewModels (`Nip65RelayListViewModel`, `PrivateOutboxRelayListViewModel`, + * `VanishRequestsState`) replay `authors`-only filters against this + * store, which full-scan without `(pubkey, created_at)` — the + * `(kind, pubkey, …)` index can't serve them, pubkey is its second + * column. A personal store is small, so the extra insert cost is + * negligible; existing DBs get the index built on next open via + * `ensureOptionalIndexes`. + */ + val INDEX_STRATEGY = DefaultIndexingStrategy(indexEventsByPubkeyAlone = true) } private fun dbDir(pubKeyHex: String): File = File(homeDir, ".amethyst/accounts/${pubKeyHex.take(8)}") @@ -87,14 +100,14 @@ class LocalRelayStore( dir.mkdirs() val path = File(dir, "events.db").absolutePath try { - store = EventStore(dbName = path, relay = LOCAL_RELAY_URL) + store = EventStore(dbName = path, relay = LOCAL_RELAY_URL, indexStrategy = INDEX_STRATEGY) _lastError.value = null refreshStats() } catch (e: Exception) { Log.w("LocalRelayStore") { "DB open failed, recreating: ${e.message}" } try { deleteDbFiles(path) - store = EventStore(dbName = path, relay = LOCAL_RELAY_URL) + store = EventStore(dbName = path, relay = LOCAL_RELAY_URL, indexStrategy = INDEX_STRATEGY) _lastError.value = "Database was recreated: ${e.message}" } catch (e2: Exception) { _lastError.value = "Cannot open local store: ${e2.message}" diff --git a/quartz/build.gradle.kts b/quartz/build.gradle.kts index 10c0817ee8..513bdd3cd5 100644 --- a/quartz/build.gradle.kts +++ b/quartz/build.gradle.kts @@ -94,6 +94,8 @@ kotlin { // Forward the negentropy-benchmark corpus size to the test JVM. System.getProperty("negBenchN")?.let { systemProperty("negBenchN", it) } System.getProperty("followBenchScale")?.let { systemProperty("followBenchScale", it) } + System.getProperty("tagBenchScale")?.let { systemProperty("tagBenchScale", it) } + System.getProperty("fsBenchScale")?.let { systemProperty("fsBenchScale", it) } // Opt-in JFR profiling of a benchmark run (-PnegProfile=/tmp/neg.jfr). (project.findProperty("negProfile") as? String)?.let { jvmArgs("-XX:+FlightRecorder", "-XX:StartFlightRecording=filename=$it,settings=profile,dumponexit=true") diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventIndexesModule.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventIndexesModule.kt index f0228217b0..dba6e2f9f3 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventIndexesModule.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventIndexesModule.kt @@ -147,13 +147,38 @@ class EventIndexesModule( */ fun migrateV2AddPubkeyIndex(db: SQLiteConnection) { if (!indexStrategy.indexEventsByPubkeyAlone) return - val orderBy = - if (indexStrategy.useAndIndexIdOnOrderBy) { - "created_at DESC, id ASC" - } else { - "created_at DESC" - } - db.execSQL("CREATE INDEX IF NOT EXISTS query_by_pubkey_created ON event_headers (pubkey, $orderBy)") + db.execSQL("CREATE INDEX IF NOT EXISTS query_by_pubkey_created ON event_headers (pubkey, ${orderByColumns()})") + } + + private fun orderByColumns() = + if (indexStrategy.useAndIndexIdOnOrderBy) { + "created_at DESC, id ASC" + } else { + "created_at DESC" + } + + /** + * Materializes any flag-gated index the current [indexStrategy] wants + * but the on-disk schema predates. Flags are runtime configuration, not + * schema — a deployment can flip one without a `user_version` bump — so + * this runs idempotently on every open. The first open after enabling a + * flag pays a one-time index build over the existing rows; subsequent + * opens are no-ops. A disabled flag never drops an existing index (that + * stays an operator decision). + */ + fun ensureOptionalIndexes(db: SQLiteConnection) { + if (indexStrategy.indexEventsByCreatedAtAlone) { + db.execSQL("CREATE INDEX IF NOT EXISTS query_by_created_at_id ON event_headers (${orderByColumns()})") + } + if (indexStrategy.indexEventsByPubkeyAlone) { + db.execSQL("CREATE INDEX IF NOT EXISTS query_by_pubkey_created ON event_headers (pubkey, ${orderByColumns()})") + } + if (indexStrategy.indexTagsByCreatedAtAlone) { + db.execSQL("CREATE INDEX IF NOT EXISTS query_by_tags_hash ON event_tags (tag_hash, created_at DESC)") + } + if (indexStrategy.indexTagsWithKindAndPubkey) { + db.execSQL("CREATE INDEX IF NOT EXISTS query_by_tags_hash_kind_pubkey ON event_tags (tag_hash, kind, pubkey_hash, created_at DESC)") + } } val sqlInsertHeader = 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 eea4950f40..3150b7625b 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 @@ -160,6 +160,11 @@ class SQLiteEventStore( setUserVersion(this, DATABASE_VERSION) } } + // Flag-gated indexes are runtime config, not schema: a + // deployment that flips an IndexingStrategy flag on an + // existing DB gets the index built here (idempotent, + // one-time cost), with no user_version bump involved. + eventIndexModule.ensureOptionalIndexes(db) }, ) } diff --git a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsQueryPlanner.kt b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsQueryPlanner.kt index f6958d8d80..431ccafe85 100644 --- a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsQueryPlanner.kt +++ b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsQueryPlanner.kt @@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.core.isAddressable import com.vitorpamplona.quartz.nip01Core.core.isReplaceable import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.store.sqlite.TagNameValueHasher +import java.nio.file.DirectoryStream import java.nio.file.Files import java.nio.file.Path import kotlin.io.path.exists @@ -36,27 +37,22 @@ import kotlin.io.path.exists * * Step-2 coverage: * - `ids` → direct canonical opens - * - `tagsAll`/`tags` → tag index union (first key) - * - `kinds` → kind index union - * - `authors` → author index union + * - `tagsAll`/`tags`/`kinds`/`authors` → cheapest index tree drives + * (capped entry-count comparison), the rest post-filter * - otherwise → full scan via every `idx/kind//` subtree * - * The planner is intentionally dumb about selectivity — "first available - * driver wins". A cost-based picker (smallest listing) can slot in - * later without changing callers. All FilterMatcher semantics (tag - * AND/OR, since/until, id, author, kind cross-checks) are enforced in - * the orchestrator, so picking a loose driver is correctness-safe. - * - * TODO: add the cost-based pick. The fixed tags → kinds → authors order - * makes `authors + kinds + limit` — the most common CLI shape (27 - * assembler call sites, every `amy feed`-style author timeline) — drive - * from the kind tree and post-filter the author. Measured by - * `FsDriverSelectionBenchmark` at 30k events: 149 ms via `idx/kind/1/` - * vs 3.4 ms via `idx/author//` with kind post-filtered (~44×), and - * the gap grows with the kind tree, not the result. Comparing candidate - * directory entry counts before walking (kind dirs vs author dirs vs tag - * dirs) is enough; the slot shortcut already rescues the - * replaceable/addressable subset. + * Driver choice is cost-based: every legal driver (each `tagsAll` value + * alone — AND semantics make any single value a complete driver — each + * `tags` key's value union, the kind set, the author set) opens a lazy + * directory iterator, all are drained in lockstep, and the first to + * exhaust — the smallest listing — drives. A giant tree (`idx/kind/1/` + * with a million entries) is therefore never read past ~the smallest + * candidate's size. Before the pick, the fixed tags → kinds → authors + * order sent `authors + kinds + limit` — the most common CLI shape — + * through the kind tree: 149 ms vs 3.4 ms (~44×) at 30k events per + * `FsDriverSelectionBenchmark`. All FilterMatcher semantics (tag AND/OR, + * since/until, id, author, kind cross-checks) are enforced in the + * orchestrator, so any driver pick is correctness-safe. */ internal class FsQueryPlanner( private val layout: FsLayout, @@ -85,19 +81,100 @@ internal class FsQueryPlanner( return ftsDriver(search) } - firstTagKey(filter)?.let { (name, values) -> - return mergeDesc(values.map { v -> walkDir(layout.tagValueDir(name, v, hasher.hash(name, v))) }) + val candidates = driverCandidates(filter) + if (candidates.isEmpty()) return allKindsDriver() + + return mergeDesc(cheapestDriver(candidates).map { walkDir(it) }) + } + + /** + * Every set of index directories that, walked and post-filtered, yields + * a superset of the filter's matches: + * - each `tagsAll` value alone (AND semantics — every match carries it), + * - each `tags` key's full value union (OR within a key, AND across), + * - the kind set, and the author set. + * Listed in the old fixed-priority order so [cheapestDriver] keeps that + * order on cost ties. + */ + private fun driverCandidates(filter: Filter): List> { + val out = ArrayList>() + filter.tagsAll?.forEach { (name, values) -> + values.forEach { v -> out.add(listOf(layout.tagValueDir(name, v, hasher.hash(name, v)))) } + } + filter.tags?.forEach { (name, values) -> + if (values.isNotEmpty()) { + out.add(values.map { v -> layout.tagValueDir(name, v, hasher.hash(name, v)) }) + } + } + filter.kinds?.takeIf { it.isNotEmpty() }?.let { kinds -> + out.add(kinds.map { layout.kindDir(it) }) + } + filter.authors?.takeIf { it.isNotEmpty() }?.let { authors -> + out.add(authors.map { layout.authorDir(it) }) + } + return out + } + + /** + * Smallest candidate by lockstep listing drain: one lazy directory + * iterator per candidate, all advanced [COST_BATCH] entries per round — + * the first to exhaust its listing is the smallest, so a giant tree is + * never read past ~the smallest candidate's size (a candidate that + * exhausts on round one costs the others one batch each). A candidate + * whose dirs are all missing exhausts immediately: driving from an empty + * mandatory predicate correctly yields an empty result. If every + * candidate survives [COST_CAP] entries, all are huge and relative + * driver choice stops mattering — the first (old fixed-priority order) + * wins. + */ + private fun cheapestDriver(candidates: List>): List { + if (candidates.size == 1) return candidates[0] + val cursors = candidates.map { EntryCursor(it) } + try { + var advanced = 0L + while (advanced < COST_CAP) { + for (i in cursors.indices) { + if (!cursors[i].skip(COST_BATCH)) return candidates[i] + } + advanced += COST_BATCH + } + return candidates[0] + } finally { + cursors.forEach { it.close() } + } + } + + /** Lazy entry iterator over a candidate's directories, in order. */ + private class EntryCursor( + dirs: List, + ) : AutoCloseable { + private val remaining = ArrayDeque(dirs) + private var stream: DirectoryStream? = null + private var iter: Iterator = emptyList().iterator() + + /** Advances up to [n] entries; false when the listing ends first. */ + fun skip(n: Int): Boolean { + var left = n + while (left > 0) { + if (iter.hasNext()) { + iter.next() + left-- + continue + } + close() + val dir = remaining.removeFirstOrNull() ?: return false + if (!Files.isDirectory(dir)) continue + stream = Files.newDirectoryStream(dir) + iter = stream!!.iterator() + } + return true } - filter.kinds?.let { kinds -> - return mergeDesc(kinds.map { walkDir(layout.kindDir(it)) }) + override fun close() { + stream?.close() + stream = null + iter = emptyList().iterator() } - - filter.authors?.let { authors -> - return mergeDesc(authors.map { walkDir(layout.authorDir(it)) }) - } - - return allKindsDriver() } /** @@ -312,14 +389,15 @@ internal class FsQueryPlanner( var top: Candidate, ) - // ---- helpers ------------------------------------------------------ + private companion object { + /** Entries each candidate's cursor advances per lockstep round. */ + const val COST_BATCH = 64 - /** First tag filter with at least one value, preferring `tagsAll`. */ - private fun firstTagKey(filter: Filter): Pair>? { - filter.tagsAll?.firstNonEmpty()?.let { return it } - filter.tags?.firstNonEmpty()?.let { return it } - return null + /** + * Stop draining once every candidate has survived this many + * entries: past it they are all huge, relative choice stops + * mattering, and the first candidate in priority order wins. + */ + const val COST_CAP = 65_536L } - - private fun Map>.firstNonEmpty(): Pair>? = entries.firstOrNull { it.value.isNotEmpty() }?.let { it.key to it.value } } diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Scenarios.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Scenarios.kt index 5c73ac5ed5..2c29711e45 100644 --- a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Scenarios.kt +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Scenarios.kt @@ -70,11 +70,11 @@ object Scenarios { } val topAuthors = notesByAuthor.entries.sortedWith(compareByDescending> { it.value }.thenBy { it.key }).map { it.key } - val hottestThread = + val hotNotes = eTagRefs.entries .sortedWith(compareByDescending> { it.value }.thenBy { it.key }) - .firstOrNull() - ?.key + .map { it.key } + val hottestThread = hotNotes.firstOrNull() val mostMentioned = pTagRefs.entries .sortedWith(compareByDescending> { it.value }.thenBy { it.key }) @@ -86,6 +86,23 @@ object Scenarios { .firstOrNull() ?.key + // The author that most often tags the most-mentioned pubkey — a + // conversation pair for the tag ∩ author (DM-room) query shape. + val conversationPeer = + mostMentioned?.let { me -> + val byAuthor = HashMap() + for (e in events) { + if (e.kind != 1 || e.pubKey == me) continue + if (e.tags.any { it.size >= 2 && it[0] == "p" && it[1] == me }) { + byAuthor.merge(e.pubKey, 1, Int::plus) + } + } + byAuthor.entries + .sortedWith(compareByDescending> { it.value }.thenBy { it.key }) + .firstOrNull() + ?.key + } + // Evenly spread sample of note ids — a "fetch these 100 events" batch. val idSample = if (noteIds.size <= 100) { @@ -159,6 +176,33 @@ object Scenarios { ), ) } + if (mostMentioned != null && conversationPeer != null) { + // The tag ∩ author ∩ kind shape (65 client assembler call + // sites: NIP-04 DM rooms, reports-by-follows, follows-scoped + // community feeds). Modeled on kind 1 because public corpora + // carry no DMs; the index path exercised is identical. + add( + Scenario( + "conversation", + "notes by one author tagging the most-mentioned pubkey (DM-room shape)", + Filter(kinds = listOf(1), authors = listOf(conversationPeer), tags = mapOf("p" to listOf(mostMentioned)), limit = 500), + ), + ) + } + if (hotNotes.size > 1) { + // Large-IN tag watcher: per-value streams come sorted off the + // tag index but their union does not, exposing whether the + // store collects+sorts or merges. 150 values stays inside + // strfry's default 200-element filter cap. + val watched = hotNotes.take(150) + add( + Scenario( + "reactions-watch", + "reactions on the ${watched.size} hottest notes (visible-feed reaction watcher)", + Filter(kinds = listOf(7), tags = mapOf("e" to watched), limit = 500), + ), + ) + } topHashtag?.let { add( Scenario(