perf(relay): remove live-path allocations in LiveEventStore + FilterIndex

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RzkoN3SJHCZWRAiadXXG4w

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

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)

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","<subId>"]` — 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"
     }

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","<eventId>",<true|false>,"<message>"]` — 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"
 

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
+}

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
  *
@@ -109,14 +115,26 @@ class FilterIndex<S : Any> {
     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 full maps — registration happens on every REQ open/close.
      */
     private data class State<S>(
-        val buckets: Map<BucketKey, Set<S>> = emptyMap(),
-        val assignments: Map<S, Set<BucketKey>> = emptyMap(),
+        val ids: PersistentMap<HexKey, PersistentSet<S>> = persistentHashMapOf(),
+        val authors: PersistentMap<HexKey, PersistentSet<S>> = persistentHashMapOf(),
+        val tags: PersistentMap<String, PersistentMap<String, PersistentSet<S>>> = persistentHashMapOf(),
+        val kinds: PersistentMap<Int, PersistentSet<S>> = persistentHashMapOf(),
+        val unindexed: PersistentSet<S> = persistentHashSetOf(),
+        val assignments: PersistentMap<S, PersistentSet<BucketKey>> = persistentHashMapOf(),
     )
 
     private val state: AtomicReference<State<S>> = AtomicReference(State())
@@ -181,18 +199,23 @@ class FilterIndex<S : Any> {
         while (true) {
             val current = state.load()
             val keys = current.assignments[subscriber] ?: return
-            val newBuckets = current.buckets.toMutableMap()
+            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 - subscriber
-                if (next.isEmpty()) {
-                    newBuckets.remove(key)
-                } else {
-                    newBuckets[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 - 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
         }
     }
 
@@ -202,19 +225,22 @@ class FilterIndex<S : Any> {
      * 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<S> {
         val s = state.load()
-        if (s.buckets.isEmpty()) return emptySet()
+        if (s.assignments.isEmpty()) return emptySet()
         val result = LinkedHashSet<S>()
-        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
@@ -235,19 +261,75 @@ class FilterIndex<S : Any> {
         keys: List<BucketKey>,
     ) {
         if (keys.isEmpty()) return
-        val keySet = keys.toSet()
+        val keySet = keys.toPersistentHashSet()
         while (true) {
             val current = state.load()
-            val newBuckets = current.buckets.toMutableMap()
+            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] ?: emptySet()
-                if (subscriber in cur) continue
-                newBuckets[key] = cur + subscriber
+                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 = if (existing == null) keySet else existing + keySet
-            val newAssignments = current.assignments + (subscriber to merged)
-            if (state.compareAndSet(current, State(newBuckets, newAssignments))) return
+            val merged = existing?.addAll(keySet) ?: keySet
+            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 <K> PersistentMap<K, PersistentSet<S>>.addSub(
+        key: K,
+        sub: S,
+    ): PersistentMap<K, PersistentSet<S>> {
+        val cur = this[key] ?: persistentHashSetOf()
+        val next = cur.add(sub)
+        return if (next === cur) this else put(key, next)
+    }
+
+    private fun <K> PersistentMap<K, PersistentSet<S>>.removeSub(
+        key: K,
+        sub: S,
+    ): PersistentMap<K, PersistentSet<S>> {
+        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<String, PersistentMap<String, PersistentSet<S>>>.addTagSub(
+        letter: String,
+        value: String,
+        sub: S,
+    ): PersistentMap<String, PersistentMap<String, PersistentSet<S>>> {
+        val inner = this[letter] ?: persistentHashMapOf()
+        val newInner = inner.addSub(value, sub)
+        return if (newInner === inner) this else put(letter, newInner)
+    }
+
+    private fun PersistentMap<String, PersistentMap<String, PersistentSet<S>>>.removeTagSub(
+        letter: String,
+        value: String,
+        sub: S,
+    ): PersistentMap<String, PersistentMap<String, PersistentSet<S>>> {
+        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)
         }
     }
 

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
@@ -331,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)) },
                         )
                     }

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<Filter>,
-        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<String>? = HashSet()
+
+        private inline fun <R> 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<String>? = HashSet(1024)
-
-        fun <R> 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<Event>(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<Filter>,
         onEachStored: (RawEvent) -> Unit,
-        onEachLive: (Event) -> Unit,
+        onEachLive: (Event, String) -> Unit,
         onEose: () -> Unit,
     ) {
         drainFtsIfSearching(filters)
-        val seenLock = AtomicBoolean(false)
-        var seenIds: HashSet<String>? = HashSet(1024)
-
-        fun <R> 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)

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt

@@ -78,9 +78,9 @@ interface SessionBackend {
         ctx: RequestContext,
         filters: List<Filter>,
         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(

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

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"
 
     /**

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<Filter>.strippingSearchExtensions(): List<Filter> {
+    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 {

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.
  */

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt

@@ -32,11 +32,14 @@ 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
 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
 
@@ -65,6 +68,8 @@ 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
+        const val FANOUT_SUBS = 200
     }
 
     private fun hexId(seed: Int): String = seed.toString(16).padStart(64, '0')
@@ -122,16 +127,19 @@ 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<Long>()
                 val t0 = System.nanoTime()
                 val job =
-                    scope.launch {
+                    scope.launch(start = CoroutineStart.UNDISPATCHED) {
                         live.queryRaw(
                             ctx = ctx,
                             filters = listOf(filterFor(round)),
                             onEachStored = {},
-                            onEachLive = {},
+                            onEachLive = { _, _ -> },
                             onEose = { eose.complete(System.nanoTime() - t0) },
                         )
                     }
@@ -143,6 +151,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<Unit>()
+                    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<Long>()
@@ -160,15 +195,67 @@ 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<Long>()
+            var fanStart = 0L
+            val fanJobs =
+                (0 until FANOUT_SUBS).map {
+                    val ready = CompletableDeferred<Unit>()
+                    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<Event>(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)
             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)})")
+            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()

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/<pk>/` 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<Event>(kindDriven).size }
+                time(store, "planner (cost-based pick)") { store.query<Event>(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<Event>(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<Event>(authorOnly).size }
+                time(store, "author-only floor") { store.query<Event>(authorOnly).size }
             } finally {
                 store.close()
                 if (root.exists()) {