From 5988db010d18bd9f94374346dce9eb7a00d29ea6 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 14:34:57 +0000 Subject: [PATCH 1/8] =?UTF-8?q?test(store):=20add=20tag=E2=88=A9author=20a?= =?UTF-8?q?nd=20FS=20driver-selection=20benchmarks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 2026-07 client filter-assembler survey mapped 551 Filter constructions to ~12 query archetypes. Two hot shapes had no benchmark coverage in prodbench or relayBench, and the FS store had none at all: - TagAuthorIndexBenchmark: the DM-room shape (kinds + authors + #p, 65 assembler call sites) with indexTagsWithKindAndPubkey off vs on, including the insert-cost delta of the extra index; plus the reactions watcher (kinds=[7], #e IN 300, limit) cold and since-bounded, which has no tag-side k-way merge today. - FsDriverSelectionBenchmark: FsQueryPlanner's fixed driver order (tags → kinds → authors) on authors+kinds+limit — the most common CLI shape — comparing the current kind-tree driver against an author-tree driver with kind post-filter. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w --- .../prodbench/TagAuthorIndexBenchmark.kt | 184 ++++++++++++++++++ .../store/fs/FsDriverSelectionBenchmark.kt | 158 +++++++++++++++ 2 files changed, 342 insertions(+) create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/TagAuthorIndexBenchmark.kt create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/TagAuthorIndexBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/TagAuthorIndexBenchmark.kt new file mode 100644 index 0000000000..7ef0fc7ac8 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/TagAuthorIndexBenchmark.kt @@ -0,0 +1,184 @@ +/* + * 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.utils.EventFactory +import kotlinx.coroutines.runBlocking +import kotlin.test.Test + +/** + * Measures the two tag-path query shapes the client filter-assembler survey + * (2026-07) found hot but that no existing benchmark covers: + * + * 1. **tag ∩ author (DM-room shape)** — `kinds=[4] AND authors=[peer] AND + * #p=[me] LIMIT n`. 65 assembler call sites build this shape (every + * NIP-04 chat room, reports-by-follows, follows-scoped community feeds). + * [com.vitorpamplona.quartz.nip01Core.store.sqlite.IndexingStrategy.indexTagsWithKindAndPubkey] + * gates a covering `(tag_hash, kind, pubkey_hash, created_at)` index for + * it, but the flag is off everywhere (including geode). Without it the + * plan seeks `(tag_hash, kind)` and reads EVERY DM the user has ever + * received before filtering to the one peer. This compares query latency + * with the flag off vs on, and the batch-insert cost the extra index adds. + * + * 2. **large-IN tag watcher (reactions shape)** — `kinds=[7] AND + * #e=[hundreds of note ids] LIMIT n`. The per-value streams come sorted + * off `(tag_hash, kind, created_at)`, but their union does not, so SQLite + * collects every matching row and TEMP-B-TREE sorts to the limit — the + * tag-index analogue of the follow-feed regression + * [MergeQueryExecutor] fixed for author streams. Reported with and + * without a `since` bound to show what EOSE-warm steady state hides. + * + * Size the seed with `-DtagBenchScale=N` (default 1 ≈ ~200k events). + */ +class TagAuthorIndexBenchmark { + companion object { + val SCALE = System.getProperty("tagBenchScale")?.toInt() ?: 1 + } + + private val hex = "0123456789abcdef" + + private fun mix(seed: Long): Long { + var z = seed + -0x61c8864680b583ebL + z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L + z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L + return z xor (z ushr 31) + } + + private fun hex64( + salt: Long, + index: Int, + ): String { + val out = CharArray(64) + for (w in 0 until 4) { + val v = mix(salt * 1_000_003 + index.toLong() * 4 + w) + for (b in 0 until 8) { + val byte = ((v ushr (b * 8)) and 0xFF).toInt() + out[(w * 8 + b) * 2] = hex[byte ushr 4] + out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF] + } + } + return String(out) + } + + private val sig = "0".repeat(128) + private var idSeq = 0 + + private fun ev( + pubkey: String, + createdAt: Long, + kind: Int, + tags: Array>, + ): Event = EventFactory.create(hex64(7, idSeq++), pubkey, createdAt, kind, tags, "", sig) + + private fun seedEvents(): List { + idSeq = 0 + val base = 1_700_000_000L + val span = 3_000_000L // ~35 days + val me = hex64(9, 0) + val events = ArrayList(220_000 * SCALE) + + // DM inbox: 200 peers, 300 DMs each → 60k kind-4 rows sharing the + // same (p:me) tag hash. The room query wants one peer's 300. + val peers = (0 until 200).map { hex64(2, it) } + for ((i, peer) in peers.withIndex()) { + repeat(300 * SCALE) { + val ts = base + (mix(i * 131L + it) and 0x7fffffff) % span + events.add(ev(peer, ts, 4, arrayOf(arrayOf("p", me)))) + } + } + + // Notification noise: 2000 authors mention me in kind-1 notes, so + // (p:me) spans multiple kinds like a real inbox does. + repeat(40_000 * SCALE) { + val author = hex64(3, it % 2_000) + val ts = base + (mix(it * 17L) and 0x7fffffff) % span + events.add(ev(author, ts, 1, arrayOf(arrayOf("p", me)))) + } + + // Reactions: 100k kind-7 events spread over 5000 target notes, for + // the large-IN watcher shape. + val noteIds = (0 until 5_000).map { hex64(5, it) } + repeat(100_000 * SCALE) { + val author = hex64(4, it % 3_000) + val ts = base + (mix(it * 29L) and 0x7fffffff) % span + events.add(ev(author, ts, 7, arrayOf(arrayOf("e", noteIds[it % noteIds.size])))) + } + return events + } + + @Test + fun compareTagAuthorIndex() = + runBlocking { + val events = seedEvents() + val me = hex64(9, 0) + val peers = (0 until 200).map { hex64(2, it) } + val noteIds = (0 until 5_000).map { hex64(5, it) } + + println("─ TagAuthorIndexBenchmark: ${events.size} events (scale=$SCALE) ─") + + val strategies = + listOf( + "flag-off" to DefaultIndexingStrategy(indexFullTextSearch = false), + "flag-on " to DefaultIndexingStrategy(indexFullTextSearch = false, indexTagsWithKindAndPubkey = true), + ) + + for ((label, strategy) in strategies) { + val store = EventStore(dbName = null, indexStrategy = strategy) + + val t0 = System.nanoTime() + events.chunked(10_000).forEach { store.batchInsert(it) } + val insertMs = (System.nanoTime() - t0) / 1e6 + println(" ═ $label ═ insert: %.0f ms (%.1f µs/event)".format(insertMs, insertMs * 1000 / events.size)) + + // 1. DM room: one peer's DMs out of the whole (p:me) inbox. + val room = Filter(kinds = listOf(4), authors = listOf(peers[42]), tags = mapOf("p" to listOf(me)), limit = 100) + time(store, "dm-room (#p ∩ author ∩ kind, limit 100)", room) + + // 2. Reactions watcher: 300 note ids, cold (no since). + val watcher = Filter(kinds = listOf(7), tags = mapOf("e" to noteIds.take(300)), limit = 500) + time(store, "reactions (#e IN 300, limit 500, cold)", watcher) + + // 3. Same watcher, EOSE-warm (since bounds the window). + val warm = watcher.copy(since = 1_700_000_000L + 2_900_000L) + time(store, "reactions (#e IN 300, limit 500, since)", warm) + + store.close() + } + } + + private suspend fun time( + store: EventStore, + label: String, + filter: Filter, + ) { + repeat(3) { store.query(filter) } + val runs = 10 + var rows = 0 + val start = System.nanoTime() + repeat(runs) { rows = store.query(filter).size } + val ms = (System.nanoTime() - start) / 1e6 / runs + println(" %-42s %8.2f ms (%d rows)".format(label, ms, rows)) + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt new file mode 100644 index 0000000000..54058191ba --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt @@ -0,0 +1,158 @@ +/* + * 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.fs + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.utils.EventFactory +import kotlinx.coroutines.runBlocking +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.exists +import kotlin.test.Test + +/** + * Quantifies [FsQueryPlanner]'s "first available driver wins" ordering + * (tags → kinds → authors) on the `authors + kinds + limit` shape — the + * most common CLI query (27 assembler call sites; every `amy feed`-style + * author timeline over non-replaceable kinds). + * + * `Filter(authors=[pk], kinds=[1], limit=n)` drives from `idx/kind/1/` + * (the biggest tree in any real store) and post-filters the author, even + * though `idx/author//` holds exactly that author's events. The + * benchmark times: + * + * - **kind-driver (current)**: the filter as the planner runs it today. + * - **author-driver (proposed)**: same result set, but driven from the + * author tree with the kind check as a post-filter — what a cost-based + * picker (compare candidate directory sizes) would choose. + * + * Also reports the author-only shape (`authors + limit`) as the floor: the + * planner already picks the author tree there, so its time is the target. + * + * Size the seed with `-DfsBenchScale=N` (default 1 ≈ ~30k events; each + * event is a file + ~3 hardlinks, so seeding dominates wall time). + */ +class FsDriverSelectionBenchmark { + companion object { + val SCALE = System.getProperty("fsBenchScale")?.toInt() ?: 1 + } + + private val hex = "0123456789abcdef" + + private fun mix(seed: Long): Long { + var z = seed + -0x61c8864680b583ebL + z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L + z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L + return z xor (z ushr 31) + } + + private fun hex64( + salt: Long, + index: Int, + ): String { + val out = CharArray(64) + for (w in 0 until 4) { + val v = mix(salt * 1_000_003 + index.toLong() * 4 + w) + for (b in 0 until 8) { + val byte = ((v ushr (b * 8)) and 0xFF).toInt() + out[(w * 8 + b) * 2] = hex[byte ushr 4] + out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF] + } + } + return String(out) + } + + private val sig = "0".repeat(128) + private var idSeq = 0 + + private fun ev( + pubkey: String, + createdAt: Long, + kind: Int, + ): Event = EventFactory.create(hex64(7, idSeq++), pubkey, createdAt, kind, emptyArray(), "", sig) + + @Test + fun compareDrivers() = + runBlocking { + val root: Path = Files.createTempDirectory("fs-driver-bench-") + val store = FsEventStore(root) + try { + val base = 1_700_000_000L + val span = 3_000_000L + val target = hex64(9, 0) + + // Background: 300 authors × 100 kind-1 notes. + val bg = ArrayList(30_000 * SCALE + 300) + repeat(30_000 * SCALE) { + val author = hex64(1, it % 300) + bg.add(ev(author, base + (mix(it * 31L) and 0x7fffffff) % span, 1)) + } + // Target author: 200 kind-1 notes + 50 kind-7 reactions. + repeat(200) { bg.add(ev(target, base + (mix(it * 131L) and 0x7fffffff) % span, 1)) } + repeat(50) { bg.add(ev(target, base + (mix(it * 61L) and 0x7fffffff) % span, 7)) } + + val t0 = System.nanoTime() + store.transaction { bg.forEach { insert(it) } } + val insertMs = (System.nanoTime() - t0) / 1e6 + println("─ FsDriverSelectionBenchmark: ${bg.size} events (scale=$SCALE), seed %.0f ms ─".format(insertMs)) + + // Current planner: kinds present → kind tree drives, author + // is a post-filter over the whole kind-1 listing. + val kindDriven = Filter(authors = listOf(target), kinds = listOf(1), limit = 50) + time(store, "kind-driver (current planner)") { store.query(kindDriven).size } + + // Proposed: drive from the author tree, post-filter kind — + // same semantics, what a cost-based picker would run. + time(store, "author-driver (proposed)") { + store + .query(Filter(authors = listOf(target), limit = 250)) + .asSequence() + .filter { it.kind == 1 } + .take(50) + .count() + } + + // Floor: author-only shape, planner already optimal here. + val authorOnly = Filter(authors = listOf(target), limit = 50) + time(store, "author-only (planner floor)") { store.query(authorOnly).size } + } finally { + store.close() + if (root.exists()) { + Files.walk(root).use { it.sorted(Comparator.reverseOrder()).forEach { p -> Files.deleteIfExists(p) } } + } + } + } + + private inline fun time( + store: FsEventStore, + label: String, + run: () -> Int, + ) { + repeat(3) { run() } + val runs = 10 + var rows = 0 + val start = System.nanoTime() + repeat(runs) { rows = run() } + val ms = (System.nanoTime() - start) / 1e6 / runs + println(" %-32s %8.2f ms (%d rows)".format(label, ms, rows)) + } +} From 41240eeab9e9d32470379c0d15d89ba1f816ba21 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 14:39:21 +0000 Subject: [PATCH 2/8] docs(store): record measured index/planner gaps as TODOs in both stores Real numbers from the two new benchmarks, so the trade-offs are on the decision points instead of in a chat log: - IndexingStrategy.indexTagsWithKindAndPubkey: the KDoc called the kinds+authors+tags shape "rarely used", but the client assembler survey found 65 call sites. TagAuthorIndexBenchmark @ 200k events: DM-room query 9.4 ms -> 0.6 ms (~15x) with the flag on, insert cost +14% (41.5 -> 47.3 us/event). TODO: re-evaluate defaults (geode). - MergeQueryExecutor: tag-path analogue of the follow-feed collect-all sort (kinds + #e IN [hundreds] + limit never merges). Measured 12.8 ms cold / 6.0 ms since-bounded at 200k events; revisit if relayBench shows it at relay scale. - FsQueryPlanner: fixed driver order sends authors+kinds+limit (the most common CLI shape) through the kind tree. FsDriverSelection- Benchmark @ 30k events: 149 ms -> 3.4 ms (~44x) driving from the author tree; TODO: cost-based pick by directory entry counts. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w --- .../nip01Core/store/sqlite/IndexingStrategy.kt | 15 ++++++++++++--- .../nip01Core/store/sqlite/MergeQueryExecutor.kt | 10 ++++++++++ .../quartz/nip01Core/store/fs/FsQueryPlanner.kt | 11 +++++++++++ 3 files changed, 33 insertions(+), 3 deletions(-) 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 816f50f0f5..4f2e848c12 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 @@ -74,9 +74,18 @@ interface IndexingStrategy { * Activate this if you see too many Tag-centric Filters without * kind AND pubkey at the same time. * - * This is a rarely used index (reports by your follows or - * NIP-04 DMs for instance) that becomes quite large without - * major gains. + * This shape (reports by your follows, NIP-04 DM rooms, follows-scoped + * community feeds) is not rare on the client side: the 2026-07 filter + * assembler survey counted 65 call sites building + * `kinds + authors + tags`. Without this index the plan seeks + * `(tag_hash, kind)` and reads every row for that tag/kind before + * filtering the author. + * + * TODO: re-evaluate the off-by-default choice (especially for geode) + * with `TagAuthorIndexBenchmark` (jvmTest prodbench). At 200k events: + * DM-room query 9.4 ms → 0.6 ms (~15×) with the flag on, for a batch + * insert cost of 41.5 → 47.3 µs/event (+14%) — measure at target + * corpus size before flipping, since the index competes for page cache. * * Keep in mind that activating too many indexes increases the size of the * DB so much that the indexes themselves won't fit in memory, requiring diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt index 62660454db..24b62c56d5 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt @@ -52,6 +52,16 @@ import androidx.sqlite.SQLiteStatement * exactly at a same-second boundary. */ internal object MergeQueryExecutor { + // TODO: the same collect-all + TEMP-B-TREE-sort pattern exists one index + // over, on the tag path: `kinds + tags(#e IN [hundreds]) + limit` (the + // reactions/replies watcher archetype) unions per-value streams that are + // each sorted off `(tag_hash, kind, created_at)` and sorts the union. + // `streamCount` currently rejects any filter with tags, so those queries + // never merge. Measured by `TagAuthorIndexBenchmark` at 200k events: + // `#e IN 300, limit 500` costs 12.8 ms cold / 6.0 ms since-bounded — + // tolerable client-side, but it scales with matching history like the + // follow-feed shape did; extend the merge to per-tag-value streams if + // relay-scale runs (relayBench) show it in the profile. const val COLS = "id, pubkey, created_at, kind, tags, content, sig" /** 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 1c64f50360..f6958d8d80 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 @@ -46,6 +46,17 @@ import kotlin.io.path.exists * 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. */ internal class FsQueryPlanner( private val layout: FsLayout, From a3e239d33fc9e4a3cfa9c6ed1c1ca11745267f6a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 14:54:48 +0000 Subject: [PATCH 3/8] feat(store): cost-based FS driver pick, runtime index materialization, new relayBench shapes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Acts on the measured gaps from TagAuthorIndexBenchmark and FsDriverSelectionBenchmark: - FsQueryPlanner: replace the fixed tags -> kinds -> authors driver order with a cost-based pick. Every legal driver (each tagsAll value, 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, so a giant idx/kind tree is never read past ~the smallest candidate's size. Fixes the 149 ms vs 3.4 ms (~44x at 30k events) authors+kinds+limit regression. - EventIndexesModule.ensureOptionalIndexes + SQLiteEventStore: flag- gated indexes are runtime config, not schema. An idempotent CREATE INDEX IF NOT EXISTS pass now runs on every open, so flipping an IndexingStrategy flag on an existing DB builds the index without a user_version bump. - Desktop LocalRelayStore: enable indexEventsByPubkeyAlone. Shared ViewModels (Nip65RelayList, PrivateOutboxRelayList, VanishRequests) replay authors-only filters that full-scanned without (pubkey, created_at); existing DBs pick the index up on next open. - relayBench Scenarios: add "conversation" (tag ∩ author ∩ kind, the DM-room shape, 65 client assembler call sites) and "reactions-watch" (kind 7 + #e IN 150 hottest notes) so the uncovered archetypes get head-to-head numbers vs strfry. - quartz build: forward tagBenchScale/fsBenchScale to the test JVM. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w --- .../amethyst/desktop/relay/LocalRelayStore.kt | 17 +- quartz/build.gradle.kts | 2 + .../store/sqlite/EventIndexesModule.kt | 39 ++++- .../store/sqlite/SQLiteEventStore.kt | 5 + .../nip01Core/store/fs/FsQueryPlanner.kt | 152 +++++++++++++----- .../com/vitorpamplona/relaybench/Scenarios.kt | 50 +++++- 6 files changed, 216 insertions(+), 49 deletions(-) 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( From 57ffb3386d59c6693897c36c41f74a185da99353 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 15:07:12 +0000 Subject: [PATCH 4/8] feat(geode): enable the tag+kind+pubkey index, refresh measured docs TagAuthorIndexBenchmark at 1M events settles the flag: the DM-room shape (kinds + authors + #p, 65 client assembler call sites) drops 14.2 ms -> 0.66 ms (~21x, growing with corpus size) while batch-insert cost stays inside run noise (49.0 vs 47.4 us/event). Existing relay DBs build the index on next open via ensureOptionalIndexes. Also refreshes the docs the numbers made stale: IndexingStrategy KDoc now records the 200k and 1M measurements instead of a TODO, MergeQueryExecutor's tag-merge note points at the new relayBench reactions-watch scenario, FsQueryPlanner/FsDriverSelectionBenchmark reflect the landed cost-based pick (149 ms -> 4.0 ms at 30k events), and RELAY.md documents that strategy flag flips materialize indexes on the next open. Verified: quartz jvmTest store suites, geode test (126), desktopApp LocalRelayStore tests (5, incl. reopening a default-strategy DB with the new pubkey-alone flag), relayBench compiles. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w --- .../geode/RelayIndexingStrategy.kt | 8 ++++ quartz/RELAY.md | 2 + .../store/sqlite/IndexingStrategy.kt | 13 +++-- .../store/sqlite/MergeQueryExecutor.kt | 11 +++-- .../nip01Core/store/fs/FsQueryPlanner.kt | 5 +- .../store/fs/FsDriverSelectionBenchmark.kt | 47 ++++++++++--------- 6 files changed, 51 insertions(+), 35 deletions(-) diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt index 33fa9f4441..97dde44807 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt @@ -57,6 +57,14 @@ fun relayIndexingStrategy( // index unconditionally; without it the filter walks the whole // time index. indexEventsByPubkeyAlone = true, + // The tag ∩ author ∩ kind shape (DM rooms, reports-by-follows, + // follows-scoped community feeds — 65 client assembler call sites) + // otherwise reads every row for the tag/kind before filtering the + // author. TagAuthorIndexBenchmark @ 1M events: 14.2 ms -> 0.66 ms + // (~21x, growing with corpus size) with insert cost inside run noise + // (49.0 vs 47.4 µs/event). Existing DBs build the index on next open + // via ensureOptionalIndexes. + indexTagsWithKindAndPubkey = true, indexFullTextSearch = fullTextSearch, // Tokenize off the commit path; NostrServer drives the catch-up // worker and search queries drain it first, so NIP-50 stays diff --git a/quartz/RELAY.md b/quartz/RELAY.md index 807b65caee..01bc5fc736 100644 --- a/quartz/RELAY.md +++ b/quartz/RELAY.md @@ -83,6 +83,8 @@ val store = EventStore( By default, all single-letter tags with values are indexed. Override `shouldIndex(kind, tag)` for custom behavior. More indexes = faster queries but larger database. +Flag flips are safe on existing databases: any flag-gated index the strategy wants but the on-disk schema lacks is created on the next open (idempotent `CREATE INDEX IF NOT EXISTS`, one-time build cost) — no schema version bump involved. Disabling a flag never drops an existing index. + `indexFullTextSearch` defaults to `true` and controls the NIP-50 full-text index (`event_fts`). Set it to `false` when search is served elsewhere (e.g. a Vespa backend, or a `SearchEventSource` as shown below): inserts skip the FTS tokenization cost, no `event_fts` table/trigger is created, and any filter carrying a non-empty `search` term returns no matches. ## Non-Storage Relays (search, redirector, computed) 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 4f2e848c12..01362e03fb 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 @@ -81,11 +81,14 @@ interface IndexingStrategy { * `(tag_hash, kind)` and reads every row for that tag/kind before * filtering the author. * - * TODO: re-evaluate the off-by-default choice (especially for geode) - * with `TagAuthorIndexBenchmark` (jvmTest prodbench). At 200k events: - * DM-room query 9.4 ms → 0.6 ms (~15×) with the flag on, for a batch - * insert cost of 41.5 → 47.3 µs/event (+14%) — measure at target - * corpus size before flipping, since the index competes for page cache. + * Measured by `TagAuthorIndexBenchmark` (jvmTest prodbench): the + * DM-room query drops 9.4 ms → 0.6 ms (~15×) at 200k events and + * 14.2 ms → 0.66 ms (~21×) at 1M — the gap grows with corpus size — + * while batch-insert cost stays inside run noise (49.0 vs 47.4 + * µs/event at 1M). geode enables it; the client default stays off + * because a client store's per-tag row counts are bounded by one + * user's data. Flipping it on an existing DB is safe: the index is + * built on next open by `EventIndexesModule.ensureOptionalIndexes`. * * Keep in mind that activating too many indexes increases the size of the * DB so much that the indexes themselves won't fit in memory, requiring diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt index 24b62c56d5..34ada1e64c 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt @@ -57,11 +57,12 @@ internal object MergeQueryExecutor { // reactions/replies watcher archetype) unions per-value streams that are // each sorted off `(tag_hash, kind, created_at)` and sorts the union. // `streamCount` currently rejects any filter with tags, so those queries - // never merge. Measured by `TagAuthorIndexBenchmark` at 200k events: - // `#e IN 300, limit 500` costs 12.8 ms cold / 6.0 ms since-bounded — - // tolerable client-side, but it scales with matching history like the - // follow-feed shape did; extend the merge to per-tag-value streams if - // relay-scale runs (relayBench) show it in the profile. + // never merge. Measured by `TagAuthorIndexBenchmark`: `#e IN 300, + // limit 500` costs 12.8 ms cold at 200k events and 14.2 ms at 1M + // (6.7 ms with indexTagsWithKindAndPubkey on) — tolerable, but it + // scales with matching history like the follow-feed shape did; extend + // the merge to per-tag-value streams if the relayBench + // `reactions-watch` scenario shows it in the profile vs strfry. const val COLS = "id, pubkey, created_at, kind, tags, content, sig" /** 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 431ccafe85..f650bfd613 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 @@ -49,8 +49,9 @@ import kotlin.io.path.exists * 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, + * through the kind tree: 149 ms fixed-order vs 4.0 ms cost-based at 30k + * events per `FsDriverSelectionBenchmark` (floor: author-only at + * 1.4 ms). 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. */ diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt index 54058191ba..320e3fcb9f 100644 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt @@ -30,23 +30,24 @@ import kotlin.io.path.exists import kotlin.test.Test /** - * Quantifies [FsQueryPlanner]'s "first available driver wins" ordering - * (tags → kinds → authors) on the `authors + kinds + limit` shape — the - * most common CLI query (27 assembler call sites; every `amy feed`-style - * author timeline over non-replaceable kinds). + * Guards [FsQueryPlanner]'s cost-based driver pick on the + * `authors + kinds + limit` shape — the most common CLI query (27 + * assembler call sites; every `amy feed`-style author timeline over + * non-replaceable kinds). * - * `Filter(authors=[pk], kinds=[1], limit=n)` drives from `idx/kind/1/` - * (the biggest tree in any real store) and post-filters the author, even - * though `idx/author//` holds exactly that author's events. The - * benchmark times: + * Under the pre-pick fixed order (tags → kinds → authors), + * `Filter(authors=[pk], kinds=[1], limit=n)` drove from `idx/kind/1/` + * (the biggest tree in any real store) and post-filtered the author: + * 149 ms at 30k events. The lockstep pick drives from the author tree + * and runs at ~4 ms. The benchmark times: * - * - **kind-driver (current)**: the filter as the planner runs it today. - * - **author-driver (proposed)**: same result set, but driven from the - * author tree with the kind check as a post-filter — what a cost-based - * picker (compare candidate directory sizes) would choose. - * - * Also reports the author-only shape (`authors + limit`) as the floor: the - * planner already picks the author tree there, so its time is the target. + * - **planner (cost-based pick)**: the filter as the planner runs it — + * should sit near the floor, far below a kind-tree walk. + * - **author-driver emulation**: the author tree walked via an + * authors-only query with the kind check applied by the caller — the + * reference the pick is expected to match or beat. + * - **author-only floor**: `authors + limit` with no kind, the cheapest + * possible walk of the same tree. * * Size the seed with `-DfsBenchScale=N` (default 1 ≈ ~30k events; each * event is a file + ~3 hardlinks, so seeding dominates wall time). @@ -115,14 +116,14 @@ class FsDriverSelectionBenchmark { val insertMs = (System.nanoTime() - t0) / 1e6 println("─ FsDriverSelectionBenchmark: ${bg.size} events (scale=$SCALE), seed %.0f ms ─".format(insertMs)) - // Current planner: kinds present → kind tree drives, author - // is a post-filter over the whole kind-1 listing. + // The planner's own pick — expected to choose the author + // tree over the ~30k-entry kind-1 tree. val kindDriven = Filter(authors = listOf(target), kinds = listOf(1), limit = 50) - time(store, "kind-driver (current planner)") { store.query(kindDriven).size } + time(store, "planner (cost-based pick)") { store.query(kindDriven).size } - // Proposed: drive from the author tree, post-filter kind — - // same semantics, what a cost-based picker would run. - time(store, "author-driver (proposed)") { + // Reference: author tree walked explicitly, kind checked + // by the caller — the pick should match or beat this. + time(store, "author-driver emulation") { store .query(Filter(authors = listOf(target), limit = 250)) .asSequence() @@ -131,9 +132,9 @@ class FsDriverSelectionBenchmark { .count() } - // Floor: author-only shape, planner already optimal here. + // Floor: author-only shape, the cheapest walk of the tree. val authorOnly = Filter(authors = listOf(target), limit = 50) - time(store, "author-only (planner floor)") { store.query(authorOnly).size } + time(store, "author-only floor") { store.query(authorOnly).size } } finally { store.close() if (root.exists()) { From 37923d91019840cd98cb7bada21cc4f821f1bf7a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 16:14:12 +0000 Subject: [PATCH 5/8] perf(relay): run the stored REQ replay undispatched SmallReqFloorBenchmark showed the per-REQ floor on small results is dominated by pipeline, not the store (raw query 0.125 ms vs 0.785 ms session REQ->EOSE in-process). Half of the dispatch slice was the scheduler hop between handleReq's launch and the query coroutine: starting the job with CoroutineStart.UNDISPATCHED runs the stored replay and EOSE inline on the receiving coroutine (the reader-pool acquire doesn't suspend when a connection is free), parking only at the live tail. Measured: dispatch+frames slice 0.397 -> 0.207 ms. Commands on a connection are processed sequentially, so nothing can target the subscription before the job lands in the registry at the first suspension point. Verified: quartz relay.server suite, SmallReqFloorBenchmark, geode full test suite. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w --- .../quartz/nip01Core/relay/server/RelaySession.kt | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index 6acacc1705..f73493f19c 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -49,6 +49,7 @@ import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.Job import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.launch @@ -291,8 +292,16 @@ class RelaySession( // Policy may rewrite filters to match the user's access level. val filters = (result as PolicyResult.Accepted).cmd.filters + // UNDISPATCHED: the stored replay runs inline on this coroutine — + // the reader-pool acquire doesn't suspend when a connection is + // free, so EVENT frames and EOSE go out without a scheduler hop + // (SmallReqFloorBenchmark: the hop was most of the dispatch + // slice on small REQs). The coroutine first parks at the live + // tail (awaitCancellation), which is when launch returns and the + // job lands in [subscriptions]; commands on this connection are + // processed sequentially, so nothing can target the sub earlier. val job = - scope.launch { + scope.launch(start = CoroutineStart.UNDISPATCHED) { try { if (policy.filtersOutgoingEvents) { // Screened path: every event is materialized so the From 6033957b1ca8d7fec23c50118bb5f1654afec686 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 16:22:13 +0000 Subject: [PATCH 6/8] perf(relay): persistent-map FilterIndex snapshots, benchmark the sub population MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FilterIndex registration runs on every REQ open/close but built each new snapshot by copying both full maps — O(S) work and allocation per REQ with S live subscriptions. Persistent (HAMT) maps keep the wait-free single-load reads and CAS write loop while making a write O(keys x log S) with structural sharing. SmallReqFloorBenchmark grows a B@1k stage (1000 idle parked subscriptions) to make the population cost visible, and its B stage now enters queryRaw undispatched like production does: @1000 subs the per-REQ cost drops 0.225 -> 0.151 ms and the measured population penalty falls below run noise (was +0.011 ms per REQ). With this and the undispatched replay, the in-process floor above the raw store query is ~0.11 ms (was ~0.66 ms as first measured): A 0.120, B 0.203, C 0.239 ms on a quiet machine. Verified: FilterIndex tests, quartz relay.server suite. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w --- .../nip01Core/relay/filters/FilterIndex.kt | 53 ++++++++++++------- .../relay/prodbench/SmallReqFloorBenchmark.kt | 36 ++++++++++++- 2 files changed, 68 insertions(+), 21 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt index 1cae5b471d..a5c058ec6d 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt @@ -22,6 +22,11 @@ package com.vitorpamplona.quartz.nip01Core.relay.filters import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey +import kotlinx.collections.immutable.PersistentMap +import kotlinx.collections.immutable.PersistentSet +import kotlinx.collections.immutable.persistentHashMapOf +import kotlinx.collections.immutable.persistentHashSetOf +import kotlinx.collections.immutable.toPersistentHashSet import kotlin.concurrent.atomics.AtomicReference import kotlin.concurrent.atomics.ExperimentalAtomicApi @@ -62,9 +67,10 @@ import kotlin.concurrent.atomics.ExperimentalAtomicApi * copy-on-write CAS loops, mirroring the * `nip86RelayManagement.server.BanStore` pattern. Reads in * [candidatesFor] and [forEach] are wait-free single-load atomic. - * Writes (subscription register / unregister) copy the inner maps - * — fine for this workload because writes are subscription-rate - * (rare) while reads are event-rate (frequent). + * Writes (subscription register / unregister) build the next snapshot + * from persistent (HAMT) maps — O(keys × log S) with structural + * sharing rather than a full O(S) copy of both maps, since on a relay + * a write happens on every REQ open and close. * * ## What the index does NOT cover * @@ -113,10 +119,16 @@ class FilterIndex { * subscribers registered under it; [assignments] is the reverse * map used by [unregister] to find a subscriber's keys without * scanning every bucket. + * + * Persistent (HAMT) maps/sets: a register/unregister produces the + * next snapshot in O(keys × log S) with structural sharing, instead + * of copying both full maps — registration happens on every REQ + * open/close, so with S live subscriptions the full copy was + * O(S) work and O(S) allocation per REQ. */ private data class State( - val buckets: Map> = emptyMap(), - val assignments: Map> = emptyMap(), + val buckets: PersistentMap> = persistentHashMapOf(), + val assignments: PersistentMap> = persistentHashMapOf(), ) private val state: AtomicReference> = AtomicReference(State()) @@ -181,17 +193,18 @@ class FilterIndex { while (true) { val current = state.load() val keys = current.assignments[subscriber] ?: return - val newBuckets = current.buckets.toMutableMap() + var newBuckets = current.buckets for (key in keys) { val cur = newBuckets[key] ?: continue - val next = cur - subscriber - if (next.isEmpty()) { - newBuckets.remove(key) - } else { - newBuckets[key] = next - } + val next = cur.remove(subscriber) + newBuckets = + if (next.isEmpty()) { + newBuckets.remove(key) + } else { + newBuckets.put(key, next) + } } - val newAssignments = current.assignments - subscriber + val newAssignments = current.assignments.remove(subscriber) if (state.compareAndSet(current, State(newBuckets, newAssignments))) return } } @@ -235,18 +248,18 @@ class FilterIndex { keys: List, ) { if (keys.isEmpty()) return - val keySet = keys.toSet() + val keySet = keys.toPersistentHashSet() while (true) { val current = state.load() - val newBuckets = current.buckets.toMutableMap() + var newBuckets = current.buckets for (key in keySet) { - val cur = newBuckets[key] ?: emptySet() - if (subscriber in cur) continue - newBuckets[key] = cur + subscriber + val cur = newBuckets[key] ?: persistentHashSetOf() + val next = cur.add(subscriber) + if (next !== cur) newBuckets = newBuckets.put(key, next) } val existing = current.assignments[subscriber] - val merged = if (existing == null) keySet else existing + keySet - val newAssignments = current.assignments + (subscriber to merged) + val merged = existing?.addAll(keySet) ?: keySet + val newAssignments = current.assignments.put(subscriber, merged) if (state.compareAndSet(current, State(newBuckets, newAssignments))) return } } diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt index 9fe3a6ec33..cd6eb4525b 100644 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt @@ -32,6 +32,7 @@ import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore import com.vitorpamplona.quartz.utils.EventFactory import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel @@ -65,6 +66,7 @@ class SmallReqFloorBenchmark { const val AUTHORS = 2_500 // ~20 events per author, matching author-archive const val ROUNDS = 400 const val WARMUP = 100 + const val IDLE_SUBS = 1_000 } private fun hexId(seed: Int): String = seed.toString(16).padStart(64, '0') @@ -122,11 +124,14 @@ class SmallReqFloorBenchmark { } // --- B: backend queryRaw to EOSE (live machinery included) --- + // UNDISPATCHED mirrors the production path (RelaySession.handleReq + // starts the query coroutine undispatched), so B−A is the live + // machinery itself, not a benchmark-only scheduler hop. suspend fun timeBackend(round: Int): Long { val eose = CompletableDeferred() val t0 = System.nanoTime() val job = - scope.launch { + scope.launch(start = CoroutineStart.UNDISPATCHED) { live.queryRaw( ctx = ctx, filters = listOf(filterFor(round)), @@ -143,6 +148,33 @@ class SmallReqFloorBenchmark { val b = LongArray(ROUNDS) repeat(ROUNDS) { b[it] = timeBackend(it) } + // --- B@1k: same, with 1000 idle live subscriptions parked --- + // Register/unregister cost scales with the live population + // (FilterIndex mutates a shared snapshot per REQ open/close), + // which the single-sub stage can't see. Each idle sub filters + // on an author absent from the corpus: 0-row replay, then parks + // at the live tail and stays registered. + val idleJobs = + (0 until IDLE_SUBS).map { i -> + val ready = CompletableDeferred() + val job = + scope.launch(start = CoroutineStart.UNDISPATCHED) { + live.queryRaw( + ctx = ctx, + filters = listOf(Filter(authors = listOf(hexId(1_000_000 + i)), kinds = listOf(1), limit = 1)), + onEachStored = {}, + onEachLive = {}, + onEose = { ready.complete(Unit) }, + ) + } + ready.await() + job + } + repeat(WARMUP) { timeBackend(it) } + val b1k = LongArray(ROUNDS) + repeat(ROUNDS) { b1k[it] = timeBackend(it) } + idleJobs.forEach { it.cancel() } + // --- C: full session dispatch, REQ json in → EOSE frame out --- suspend fun timeSession(round: Int): Long { val eose = CompletableDeferred() @@ -164,10 +196,12 @@ class SmallReqFloorBenchmark { val mA = median(a) val mB = median(b) + val mB1k = median(b1k) val mC = median(c) println("SmallReqFloorBenchmark @ ${EVENTS / 1000}k events, ~${rowsA / ROUNDS} rows/req, medians of $ROUNDS") println(" A raw store query: ${"%6.3f".format(mA)} ms") println(" B backend queryRaw→EOSE: ${"%6.3f".format(mB)} ms (live machinery +${"%6.3f".format(mB - mA)})") + println(" B@${IDLE_SUBS} idle subs: ${"%6.3f".format(mB1k)} ms (population cost +${"%6.3f".format(mB1k - mB)})") println(" C session REQ→EOSE: ${"%6.3f".format(mC)} ms (dispatch+frames +${"%6.3f".format(mC - mB)})") server.close() From 387bfe99ee109bb920032b274c9211344868535b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 19:23:21 +0000 Subject: [PATCH 7/8] perf(relay): drop per-op allocations on the frame + search paths Three hot-path allocation cuts the SmallReqFloorBenchmark stages flagged: - strippingSearchExtensions: index-loop guard returns the same list with zero allocation when no filter carries a search term (every non-search REQ/COUNT/snapshot, the overwhelming majority). - EoseMessage/OkMessage: direct-buildString wire form on the escape-free fast path (EOSE per REQ, OK per publish), skipping the generic serializer's node tree; exotic subIds/reasons fall back. Shared isEscapeFreeAscii helper in WireJson.kt, mirroring NegMsgMessage. Verified: quartz relay.server + message-frame suites (110 tests). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w --- .../relay/commands/toClient/EoseMessage.kt | 16 +++++++++ .../relay/commands/toClient/OkMessage.kt | 19 ++++++++++ .../relay/commands/toClient/WireJson.kt | 36 +++++++++++++++++++ .../quartz/nip50Search/SearchQuery.kt | 12 +++++++ 4 files changed, 83 insertions(+) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/WireJson.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/EoseMessage.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/EoseMessage.kt index d6912571b0..f376e4bc82 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/EoseMessage.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/EoseMessage.kt @@ -25,6 +25,22 @@ class EoseMessage( ) : Message { override fun label() = LABEL + /** + * Wire form is `["EOSE",""]` — sent once per REQ, so it is on + * the per-subscription floor. Splice it directly when [subId] needs no + * escaping (the common case: client-chosen sub ids are short ASCII), + * skipping the generic serializer's node tree. Byte-identical output; + * any exotic subId falls back. + */ + override fun toJson(): String { + if (!isEscapeFreeAscii(subId)) return super.toJson() + return buildString(subId.length + 12) { + append("[\"EOSE\",\"") + append(subId) + append("\"]") + } + } + companion object { const val LABEL = "EOSE" } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/OkMessage.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/OkMessage.kt index 8a29794a44..47df9bedb0 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/OkMessage.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/OkMessage.kt @@ -29,6 +29,25 @@ class OkMessage( ) : Message { override fun label() = LABEL + /** + * Wire form is `["OK","",,""]` — sent + * once per published EVENT. [eventId] is validated hex (always + * escape-free); splice directly when [message] also needs no escaping, + * which covers the empty-string success ack and the plain-ASCII + * rejection reasons. Byte-identical output; a reason with quotes or + * non-ASCII falls back to the generic serializer. + */ + override fun toJson(): String { + if (!isEscapeFreeAscii(message)) return super.toJson() + return buildString(eventId.length + message.length + 20) { + append("[\"OK\",\"") + append(eventId) + append(if (success) "\",true,\"" else "\",false,\"") + append(message) + append("\"]") + } + } + companion object { const val LABEL = "OK" diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/WireJson.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/WireJson.kt new file mode 100644 index 0000000000..63e1a47e5e --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/WireJson.kt @@ -0,0 +1,36 @@ +/* + * 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.commands.toClient + +/** + * True when every char of [s] is printable ASCII (0x20–0x7e) and not a JSON + * metacharacter (`"` / `\`) — i.e. exactly the bytes a JSON string encoder + * would emit verbatim between the quotes. Frame builders use this to gate a + * direct-`buildString` fast path against the generic serializer: when it holds + * the spliced output is byte-identical, and any exotic value (control chars, + * quotes, non-ASCII) falls back to the escaping serializer. + */ +internal fun isEscapeFreeAscii(s: String): Boolean { + for (c in s) { + if (c < ' ' || c > '~' || c == '"' || c == '\\') return false + } + return true +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip50Search/SearchQuery.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip50Search/SearchQuery.kt index 12b3af995d..f96b620d5c 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip50Search/SearchQuery.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip50Search/SearchQuery.kt @@ -202,8 +202,20 @@ fun Filter.strippingSearchExtensions(): Filter { /** * Applies [strippingSearchExtensions] to every filter, returning this * same list when no filter carried extension tokens. + * + * This runs on every REQ/COUNT/snapshot, and the overwhelming majority + * carry no `search` term at all, so the no-search case must not allocate: + * bail before building any list when nothing could be stripped. */ fun List.strippingSearchExtensions(): List { + var hasSearch = false + for (i in indices) { + if (!this[i].search.isNullOrEmpty()) { + hasSearch = true + break + } + } + if (!hasSearch) return this var changed = false val out = map { From 078758888a5829c09981a102e1b372114a5f3a8e Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 19:23:38 +0000 Subject: [PATCH 8/8] perf(relay): remove live-path allocations in LiveEventStore + FilterIndex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The remaining SmallReqFloorBenchmark waste, on the per-row replay, the per-event live fanout, and the per-accepted-event index probe: - LiveEventStore replay dedupe: a SeenIds holder with an inline lock replaces the local-fn-plus-lambda that allocated one closure per streamed row (and again per live delivery). Its HashSet is created empty so the JVM defers the backing table to the first add — a 0-row replay no longer allocates a 1024-slot table (was ~4 MB across the benchmark's 1000 idle subs). - Live fanout serializes the event body once and passes it through onEachLive(event, body); RelaySession splices it into the per-sub frame prefix. An event matching N live subscriptions paid N identical Jackson passes before; now one. queryRaw's onEachLive signature gains the body arg (EventSourceBackend default serializes inline, no cross-sub memo, no regression). Measured: fanout 1->200 live subs 0.50 ms (2.5 us/sub). - FilterIndex holds subscribers in one persistent map per dimension, so candidatesFor (once per accepted ingest event) probes with the event's own fields and allocates no IdKey/AuthorKey/KindKey/TagKey wrappers; BucketKey now lives only in the rare register/unregister bookkeeping. SmallReqFloorBenchmark grows a fanout stage (200 live subs, one submit) to anchor the fanout number; it drives `live` directly and guards the await with withTimeout so a future fanout regression fails fast. Verified: quartz relay.server + FilterIndex suites (110 tests), SmallReqFloorBenchmark, geode suite (126 tests). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w --- .../nip01Core/relay/filters/FilterIndex.kt | 137 +++++++++++++----- .../nip01Core/relay/server/RelaySession.kt | 15 +- .../relay/server/backend/LiveEventStore.kt | 136 +++++++++-------- .../relay/server/backend/SessionBackend.kt | 4 +- .../relay/prodbench/SmallReqFloorBenchmark.kt | 57 +++++++- 5 files changed, 247 insertions(+), 102 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt index a5c058ec6d..bee08fc159 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt @@ -115,19 +115,25 @@ class FilterIndex { private object Unindexed : BucketKey /** - * Single immutable snapshot. [buckets] maps a key to the set of - * subscribers registered under it; [assignments] is the reverse - * map used by [unregister] to find a subscriber's keys without - * scanning every bucket. + * Single immutable snapshot. Subscribers are held in one map per + * indexable dimension so [candidatesFor] — called once per accepted + * ingest event, the hot read — can probe each dimension with the + * event's own field (`event.id`, `event.pubKey`, `tag[0]`/`tag[1]`, + * `event.kind`) and allocate no key-wrapper objects. [assignments] is + * the reverse map ([S] → the [BucketKey]s it occupies) used by + * [unregister]; the wrappers live only here, built on the rare + * register path. * * Persistent (HAMT) maps/sets: a register/unregister produces the * next snapshot in O(keys × log S) with structural sharing, instead - * of copying both full maps — registration happens on every REQ - * open/close, so with S live subscriptions the full copy was - * O(S) work and O(S) allocation per REQ. + * of copying full maps — registration happens on every REQ open/close. */ private data class State( - val buckets: PersistentMap> = persistentHashMapOf(), + val ids: PersistentMap> = persistentHashMapOf(), + val authors: PersistentMap> = persistentHashMapOf(), + val tags: PersistentMap>> = persistentHashMapOf(), + val kinds: PersistentMap> = persistentHashMapOf(), + val unindexed: PersistentSet = persistentHashSetOf(), val assignments: PersistentMap> = persistentHashMapOf(), ) @@ -193,19 +199,23 @@ class FilterIndex { while (true) { val current = state.load() val keys = current.assignments[subscriber] ?: return - var newBuckets = current.buckets + var ids = current.ids + var authors = current.authors + var tags = current.tags + var kinds = current.kinds + var unindexed = current.unindexed for (key in keys) { - val cur = newBuckets[key] ?: continue - val next = cur.remove(subscriber) - newBuckets = - if (next.isEmpty()) { - newBuckets.remove(key) - } else { - newBuckets.put(key, next) - } + when (key) { + is IdKey -> ids = ids.removeSub(key.id, subscriber) + is AuthorKey -> authors = authors.removeSub(key.author, subscriber) + is KindKey -> kinds = kinds.removeSub(key.kind, subscriber) + is TagKey -> tags = tags.removeTagSub(key.letter, key.value, subscriber) + Unindexed -> unindexed = unindexed.remove(subscriber) + } } - val newAssignments = current.assignments.remove(subscriber) - if (state.compareAndSet(current, State(newBuckets, newAssignments))) return + val next = + State(ids, authors, tags, kinds, unindexed, current.assignments.remove(subscriber)) + if (state.compareAndSet(current, next)) return } } @@ -215,19 +225,22 @@ class FilterIndex { * candidate to handle negative constraints. * * Iteration order is insertion-stable per call but otherwise - * unspecified. + * unspecified. Allocates only the result set — dimensions are + * probed with the event's own fields, no key wrappers. */ fun candidatesFor(event: Event): Set { val s = state.load() - if (s.buckets.isEmpty()) return emptySet() + if (s.assignments.isEmpty()) return emptySet() val result = LinkedHashSet() - s.buckets[Unindexed]?.let { result.addAll(it) } - s.buckets[IdKey(event.id)]?.let { result.addAll(it) } - s.buckets[AuthorKey(event.pubKey)]?.let { result.addAll(it) } - s.buckets[KindKey(event.kind)]?.let { result.addAll(it) } - for (tag in event.tags) { - if (tag.size >= 2 && tag[0].length == 1) { - s.buckets[TagKey(tag[0], tag[1])]?.let { result.addAll(it) } + if (s.unindexed.isNotEmpty()) result.addAll(s.unindexed) + s.ids[event.id]?.let { result.addAll(it) } + s.authors[event.pubKey]?.let { result.addAll(it) } + s.kinds[event.kind]?.let { result.addAll(it) } + if (s.tags.isNotEmpty()) { + for (tag in event.tags) { + if (tag.size >= 2 && tag[0].length == 1) { + s.tags[tag[0]]?.get(tag[1])?.let { result.addAll(it) } + } } } return result @@ -251,16 +264,72 @@ class FilterIndex { val keySet = keys.toPersistentHashSet() while (true) { val current = state.load() - var newBuckets = current.buckets + var ids = current.ids + var authors = current.authors + var tags = current.tags + var kinds = current.kinds + var unindexed = current.unindexed for (key in keySet) { - val cur = newBuckets[key] ?: persistentHashSetOf() - val next = cur.add(subscriber) - if (next !== cur) newBuckets = newBuckets.put(key, next) + when (key) { + is IdKey -> ids = ids.addSub(key.id, subscriber) + is AuthorKey -> authors = authors.addSub(key.author, subscriber) + is KindKey -> kinds = kinds.addSub(key.kind, subscriber) + is TagKey -> tags = tags.addTagSub(key.letter, key.value, subscriber) + Unindexed -> unindexed = unindexed.add(subscriber) + } } val existing = current.assignments[subscriber] val merged = existing?.addAll(keySet) ?: keySet - val newAssignments = current.assignments.put(subscriber, merged) - if (state.compareAndSet(current, State(newBuckets, newAssignments))) return + val next = State(ids, authors, tags, kinds, unindexed, current.assignments.put(subscriber, merged)) + if (state.compareAndSet(current, next)) return + } + } + + // Per-dimension add/remove of one subscriber, returning the same map + // instance when nothing changed so the CAS builds minimal new nodes. + private fun PersistentMap>.addSub( + key: K, + sub: S, + ): PersistentMap> { + val cur = this[key] ?: persistentHashSetOf() + val next = cur.add(sub) + return if (next === cur) this else put(key, next) + } + + private fun PersistentMap>.removeSub( + key: K, + sub: S, + ): PersistentMap> { + val cur = this[key] ?: return this + val next = cur.remove(sub) + return when { + next === cur -> this + next.isEmpty() -> remove(key) + else -> put(key, next) + } + } + + private fun PersistentMap>>.addTagSub( + letter: String, + value: String, + sub: S, + ): PersistentMap>> { + val inner = this[letter] ?: persistentHashMapOf() + val newInner = inner.addSub(value, sub) + return if (newInner === inner) this else put(letter, newInner) + } + + private fun PersistentMap>>.removeTagSub( + letter: String, + value: String, + sub: S, + ): PersistentMap>> { + val inner = this[letter] ?: return this + val newInner = inner.removeSub(value, sub) + return when { + newInner === inner -> this + newInner.isEmpty() -> remove(letter) + else -> put(letter, newInner) } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index f73493f19c..7aa1e208d6 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -340,7 +340,20 @@ class RelaySession( }, ) }, - onEachLive = { event -> send(EventMessage(cmd.subId, event)) }, + // Live events arrive with their wire body already + // serialized (once per event, shared across every + // matching subscription): splice it into the same + // per-sub frame prefix as the stored replay, no + // per-event EventMessage or re-serialize. + onEachLive = { _, body -> + sendRaw( + buildString(framePrefix.length + body.length + 1) { + append(framePrefix) + append(body) + append(']') + }, + ) + }, onEose = { send(EoseMessage(cmd.subId)) }, ) } 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 1d35362134..9f5f8c4ecc 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 @@ -74,13 +74,57 @@ class LiveEventStore( * One live REQ subscription. Carries the filters (for the * post-index `match` re-check needed for negative constraints * like `since` / `until` / `tagsAll`) and the delivery callback - * the index dispatches into. Identity-keyed inside [FilterIndex]. + * the index dispatches into. [deliver] receives the event and its + * pre-serialized wire body (memoized once per fanout across all + * matching subscribers). Identity-keyed inside [FilterIndex]. */ private class LiveSubscription( val filters: List, - val deliver: (Event) -> Unit, + val deliver: (Event, String) -> Unit, ) + /** + * Replay-dedupe set for one REQ. During the historical replay the + * store's ids are [record]ed here so the concurrent live path can + * drop an event the replay also emitted; after EOSE the set is + * [release]d and the live path forwards everything. + * + * Written from the replay coroutine and read from the [IngestQueue] + * drain coroutine (via `fanout`), so every access takes a tiny spin + * lock — [locked] is `inline`, so the per-row `record` / `isDuplicate` + * calls allocate no closure. The backing `HashSet` is created empty + * up front (so the register-before-replay race guarantee holds) but + * the JVM defers its table allocation to the first `add`, so a + * zero-row replay costs only the empty set object, not a sized table. + * It MUST stay a mutable set under a lock, never a copy-on-add + * immutable set — `set + id` per row made large replays O(n²). + */ + private class SeenIds { + private val lock = AtomicBoolean(false) + private var ids: HashSet? = HashSet() + + private inline fun locked(block: () -> R): R { + while (lock.exchange(true)) { + while (lock.load()) { } + } + try { + return block() + } finally { + lock.store(false) + } + } + + fun record(id: String) { + locked { ids?.add(id) } + } + + fun isDuplicate(id: String): Boolean = locked { ids?.contains(id) ?: false } + + fun release() { + locked { ids = null } + } + } + /** * Fire-and-forget enqueue: hand [event] to the [IngestQueue] and * fire [onComplete] once the writer's batch has a per-row @@ -149,9 +193,17 @@ class LiveEventStore( * batch writer. */ private fun fanout(event: Event) { - for (sub in index.candidatesFor(event)) { + val candidates = index.candidatesFor(event) + if (candidates.isEmpty()) return + // Serialize the wire body at most once for this event, no matter + // how many subscriptions match it — the old path re-serialized the + // whole event per matching subscriber, so a note landing in N live + // feeds paid N identical Jackson passes. Lazy so a fanout that + // matches nothing (index over-approximates) serializes nothing. + var body: String? = null + for (sub in candidates) { if (sub.filters.any { it.match(event) }) { - sub.deliver(event) + sub.deliver(event, body ?: event.toJson().also { body = it }) } } } @@ -178,61 +230,33 @@ class LiveEventStore( 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 - // mid-replay would slip past the live path entirely — same - // race the previous SharedFlow-based implementation closed - // with `onSubscription`). - // - // The set is read from the [IngestQueue] drain coroutine (in - // `deliver`, called synchronously from `fanout`) and written - // from this coroutine (the historical-replay closure below), - // so access is guarded by a tiny spin lock (contains/add, - // never I/O). It MUST be a mutable set under a lock, not an - // immutable Set under an AtomicReference with copy-on-add: - // `set + id` copies the whole set per streamed event, which - // made large replays accidentally O(n²) — a 100k-event REQ - // crawled at ~700 events/s and the rate degraded as the - // response grew (see the plan doc's giant-REQ finding). - // - // Once cleared to null after EOSE, `deliver` short-circuits - // and every live event is forwarded. - val seenLock = AtomicBoolean(false) - var seenIds: HashSet? = HashSet(1024) - - fun seenLocked(block: () -> R): R { - while (seenLock.exchange(true)) { - while (seenLock.load()) { } - } - try { - return block() - } finally { - seenLock.store(false) - } - } + // The index registers *before* the replay starts (otherwise an + // event accepted mid-replay would slip past the live path entirely + // — same race the previous SharedFlow-based implementation closed + // with `onSubscription`), and [SeenIds] bridges the two coroutines: + // the replay records ids here, the live `deliver` drops duplicates, + // and after EOSE the set is released so every live event forwards. + val seen = SeenIds() val sub = LiveSubscription( filters = filters, - deliver = { event -> - val duplicate = seenLocked { seenIds?.contains(event.id) ?: false } - if (duplicate) return@LiveSubscription - onEach(event) + deliver = { event, _ -> + if (!seen.isDuplicate(event.id)) onEach(event) }, ) index.register(filters, sub) try { store.query(filters.strippingSearchExtensions()) { event -> - seenLocked { seenIds?.add(event.id) } + seen.record(event.id) onEach(event) } onEose() // Drop the dedupe set so the live path stops paying for // it. From this point the index drives delivery and // duplicates are no longer possible. - seenLocked { seenIds = null } + seen.release() // Suspend until the caller's coroutine is cancelled // (e.g. NIP-01 CLOSE or connection drop). The `finally` // unregisters from the index. @@ -255,42 +279,28 @@ class LiveEventStore( ctx: RequestContext, filters: List, onEachStored: (RawEvent) -> Unit, - onEachLive: (Event) -> Unit, + onEachLive: (Event, String) -> Unit, onEose: () -> Unit, ) { drainFtsIfSearching(filters) - val seenLock = AtomicBoolean(false) - var seenIds: HashSet? = HashSet(1024) - - fun seenLocked(block: () -> R): R { - while (seenLock.exchange(true)) { - while (seenLock.load()) { } - } - try { - return block() - } finally { - seenLock.store(false) - } - } + val seen = SeenIds() val sub = LiveSubscription( filters = filters, - deliver = { event -> - val duplicate = seenLocked { seenIds?.contains(event.id) ?: false } - if (duplicate) return@LiveSubscription - onEachLive(event) + deliver = { event, body -> + if (!seen.isDuplicate(event.id)) onEachLive(event, body) }, ) index.register(filters, sub) try { store.rawQuery(filters.strippingSearchExtensions()) { raw -> - seenLocked { seenIds?.add(raw.id) } + seen.record(raw.id) onEachStored(raw) } onEose() - seenLocked { seenIds = null } + seen.release() awaitCancellation() } finally { index.unregister(sub) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt index 1119d9e9be..f379364cc4 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt @@ -78,9 +78,9 @@ interface SessionBackend { ctx: RequestContext, filters: List, onEachStored: (RawEvent) -> Unit, - onEachLive: (Event) -> Unit, + onEachLive: (Event, String) -> Unit, onEose: () -> Unit, - ): Unit = query(ctx, filters, onEachLive, onEose) + ): Unit = query(ctx, filters, { onEachLive(it, it.toJson()) }, onEose) /** Answers a NIP-45 COUNT with an exact cardinality for the caller in [ctx]. */ suspend fun count( diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt index cd6eb4525b..4c2c839965 100644 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt @@ -38,6 +38,8 @@ import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import java.util.concurrent.atomic.AtomicInteger import kotlin.test.Test import kotlin.test.assertEquals @@ -67,6 +69,7 @@ class SmallReqFloorBenchmark { const val ROUNDS = 400 const val WARMUP = 100 const val IDLE_SUBS = 1_000 + const val FANOUT_SUBS = 200 } private fun hexId(seed: Int): String = seed.toString(16).padStart(64, '0') @@ -136,7 +139,7 @@ class SmallReqFloorBenchmark { ctx = ctx, filters = listOf(filterFor(round)), onEachStored = {}, - onEachLive = {}, + onEachLive = { _, _ -> }, onEose = { eose.complete(System.nanoTime() - t0) }, ) } @@ -163,7 +166,7 @@ class SmallReqFloorBenchmark { ctx = ctx, filters = listOf(Filter(authors = listOf(hexId(1_000_000 + i)), kinds = listOf(1), limit = 1)), onEachStored = {}, - onEachLive = {}, + onEachLive = { _, _ -> }, onEose = { ready.complete(Unit) }, ) } @@ -192,6 +195,54 @@ class SmallReqFloorBenchmark { val c = LongArray(ROUNDS) repeat(ROUNDS) { c[it] = timeSession(it) } + // --- fanout: one live event → FANOUT_SUBS live subscriptions --- + // All subs register on `live` directly (via queryRaw, same backend + // we submit into) and filter an author with no stored events (0-row + // replay, then park live). Submitting one matching event fans out to + // every sub; the body is serialized once and spliced per sub, so + // this measures the shared-serialization path (#2). skipVerify so + // the synthetic sig is accepted. Fewer rounds than A–C: each round + // is FANOUT_SUBS deliveries and a real group-commit insert. + val fanAuthor = hexId(9_000_001) + val delivered = AtomicInteger(0) + var fanDone = CompletableDeferred() + var fanStart = 0L + val fanJobs = + (0 until FANOUT_SUBS).map { + val ready = CompletableDeferred() + val job = + scope.launch(start = CoroutineStart.UNDISPATCHED) { + live.queryRaw( + ctx = ctx, + filters = listOf(Filter(authors = listOf(fanAuthor), kinds = listOf(1))), + onEachStored = {}, + onEachLive = { _, _ -> + if (delivered.incrementAndGet() == FANOUT_SUBS) { + fanDone.complete(System.nanoTime() - fanStart) + } + }, + onEose = { ready.complete(Unit) }, + ) + } + ready.await() + job + } + val fanRounds = 60 + val fanWarmup = 15 + val fan = LongArray(fanRounds) + var fanSeq = 0 + repeat(fanWarmup + fanRounds) { r -> + delivered.set(0) + fanDone = CompletableDeferred() + val ev = EventFactory.create(hexId(9_500_000 + fanSeq), fanAuthor, 1_700_000_000L + fanSeq, 1, emptyArray(), "fanout $fanSeq", sig) + fanSeq++ + fanStart = System.nanoTime() + live.submit(ev, skipVerify = true) {} + val nanos = withTimeout(30_000) { fanDone.await() } + if (r >= fanWarmup) fan[r - fanWarmup] = nanos + } + fanJobs.forEach { it.cancel() } + assertEquals(true, rowsA > 0, "author filters must return rows") val mA = median(a) @@ -203,6 +254,8 @@ class SmallReqFloorBenchmark { println(" B backend queryRaw→EOSE: ${"%6.3f".format(mB)} ms (live machinery +${"%6.3f".format(mB - mA)})") println(" B@${IDLE_SUBS} idle subs: ${"%6.3f".format(mB1k)} ms (population cost +${"%6.3f".format(mB1k - mB)})") println(" C session REQ→EOSE: ${"%6.3f".format(mC)} ms (dispatch+frames +${"%6.3f".format(mC - mB)})") + val mFan = median(fan) + println(" fanout 1→$FANOUT_SUBS live subs: ${"%6.3f".format(mFan)} ms (${"%.2f".format(mFan * 1000 / FANOUT_SUBS)} µs/sub; body serialized once)") server.close() scope.cancel()