mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-07-22 07:48:27 +00:00
feat(store): cost-based FS driver pick, runtime index materialization, new relayBench shapes
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w
This commit is contained in:
Claude
@@ -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}"
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -147,13 +147,38 @@ class EventIndexesModule(
|
||||
*/
|
||||
fun migrateV2AddPubkeyIndex(db: SQLiteConnection) {
|
||||
if (!indexStrategy.indexEventsByPubkeyAlone) return
|
||||
val 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"
|
||||
}
|
||||
db.execSQL("CREATE INDEX IF NOT EXISTS query_by_pubkey_created ON event_headers (pubkey, $orderBy)")
|
||||
|
||||
/**
|
||||
* 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 =
|
||||
|
||||
@@ -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)
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
@@ -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/<k>/` 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/<pk>/` 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) })
|
||||
}
|
||||
|
||||
filter.kinds?.let { kinds ->
|
||||
return mergeDesc(kinds.map { walkDir(layout.kindDir(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<List<Path>> {
|
||||
val out = ArrayList<List<Path>>()
|
||||
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
|
||||
}
|
||||
|
||||
filter.authors?.let { authors ->
|
||||
return mergeDesc(authors.map { walkDir(layout.authorDir(it)) })
|
||||
/**
|
||||
* 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<Path>>): List<Path> {
|
||||
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() }
|
||||
}
|
||||
}
|
||||
|
||||
return allKindsDriver()
|
||||
/** Lazy entry iterator over a candidate's directories, in order. */
|
||||
private class EntryCursor(
|
||||
dirs: List<Path>,
|
||||
) : AutoCloseable {
|
||||
private val remaining = ArrayDeque(dirs)
|
||||
private var stream: DirectoryStream<Path>? = null
|
||||
private var iter: Iterator<Path> = emptyList<Path>().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
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
stream?.close()
|
||||
stream = null
|
||||
iter = emptyList<Path>().iterator()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -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<String, List<String>>? {
|
||||
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<String, List<String>>.firstNonEmpty(): Pair<String, List<String>>? = entries.firstOrNull { it.value.isNotEmpty() }?.let { it.key to it.value }
|
||||
}
|
||||
|
||||
@@ -70,11 +70,11 @@ object Scenarios {
|
||||
}
|
||||
|
||||
val topAuthors = notesByAuthor.entries.sortedWith(compareByDescending<Map.Entry<String, Int>> { it.value }.thenBy { it.key }).map { it.key }
|
||||
val hottestThread =
|
||||
val hotNotes =
|
||||
eTagRefs.entries
|
||||
.sortedWith(compareByDescending<Map.Entry<String, Int>> { it.value }.thenBy { it.key })
|
||||
.firstOrNull()
|
||||
?.key
|
||||
.map { it.key }
|
||||
val hottestThread = hotNotes.firstOrNull()
|
||||
val mostMentioned =
|
||||
pTagRefs.entries
|
||||
.sortedWith(compareByDescending<Map.Entry<String, Int>> { 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<String, Int>()
|
||||
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<Map.Entry<String, Int>> { 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(
|
||||
|
||||
Reference in New Issue
Block a user