perf(relay): persistent-map FilterIndex snapshots, benchmark the sub population

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

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<S : Any> {
      * 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<S>(
-        val buckets: Map<BucketKey, Set<S>> = emptyMap(),
-        val assignments: Map<S, Set<BucketKey>> = emptyMap(),
+        val buckets: PersistentMap<BucketKey, PersistentSet<S>> = persistentHashMapOf(),
+        val assignments: PersistentMap<S, PersistentSet<BucketKey>> = persistentHashMapOf(),
     )
 
     private val state: AtomicReference<State<S>> = AtomicReference(State())
@@ -181,17 +193,18 @@ class FilterIndex<S : Any> {
         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<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 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
         }
     }

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<Long>()
                 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<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>()
@@ -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()