perf(quartz): inline fast path for bounded small REQs

Backlog item 2 (small-REQ dispatch floor), first cut. relayBench at 50k shows geode ~2.5x slower than strfry when a REQ returns ~20 events (author-archive 1.18 vs 0.48 ms EOSE p50) while WINNING the 500-event scenarios — with tiny results the fixed per-REQ cost dominates. A new stage benchmark (SmallReqFloorBenchmark) decomposes the floor: at ~21 rows, raw SQL is 0.18 ms and the remaining ~0.42 ms is live-subscription machinery plus per-REQ coroutine dispatch. A REQ whose filters all carry a limit summing to <= 256 now runs its stored replay inline on the receive coroutine and only retains a live-tail handle (SessionBackend.queryRawInline; LiveEventStore's queryRaw is re-expressed on the same core) — no per-REQ launch, no Job, no dispatcher handoffs, no awaitCancellation scaffolding. Unbounded REQs keep the launched path so CLOSE can always interrupt a giant replay. In-process time-to-EOSE for ~21-row REQs drops 0.60 -> 0.50 ms; the bigger effect expected under concurrency (no per-REQ Job+dispatch churn) is relayBench's to judge. InlineReqFastPathTest pins wire-behavior parity: stored-then-EOSE ordering, live tail delivery and CLOSE detachment, same-subId replacement, launched fallback for unbounded and over-cap REQs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A
2026-07-30 19:46:17 +00:00 · 2026-07-03 23:54:19 +00:00
parent 0d6a9b1e4d
commit fb29d655fb
5 changed files with 508 additions and 26 deletions
@@ -35,6 +35,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CountCmd
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.server.backend.RequestContext
 import com.vitorpamplona.quartz.nip01Core.relay.server.backend.SessionBackend
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.IRelayPolicy
@@ -49,7 +50,6 @@ import com.vitorpamplona.quartz.utils.Log
 import com.vitorpamplona.quartz.utils.cache.LargeCache
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.ClosedSendChannelException
 import kotlinx.coroutines.launch
 import kotlin.concurrent.atomics.AtomicLong
@@ -75,7 +75,16 @@ class RelaySession(
     */
    val id: Long = nextConnectionId(),
 ) : AutoCloseable {
-    private val subscriptions = LargeCache<String, Job>()
+    /**
+     * One active REQ. Launched subscriptions cancel their coroutine;
+     * inline subscriptions (the bounded small-REQ fast path, which never
+     * had a coroutine) close their live-tail registration directly.
+     */
+    private fun interface ActiveSubscription {
+        fun cancel()
+    }
+
+    private val subscriptions = LargeCache<String, ActiveSubscription>()

    /**
     * The authenticated-identity store for this connection. The engine is the
@@ -103,8 +112,8 @@ class RelaySession(

    private fun addSubscription(
        subId: String,
-        job: Job,
-    ) = subscriptions.put(subId, job)
+        sub: ActiveSubscription,
+    ) = subscriptions.put(subId, sub)

    private fun cancelSubscription(subId: String): Boolean =
        subscriptions.remove(subId)?.let {
@@ -113,7 +122,7 @@ class RelaySession(
        } ?: false

    fun cancelAllSubscriptions() {
-        subscriptions.forEach { _, job -> job.cancel() }
+        subscriptions.forEach { _, sub -> sub.cancel() }
        subscriptions.clear()
        negentropy.clear()
    }
@@ -263,7 +272,28 @@ class RelaySession(
    }

    // -- NIP-01: REQ ----------------------------------------------------------
-    private fun handleReq(cmd: ReqCmd) {
+
+    /**
+     * A REQ qualifies for the inline fast path when its replay is
+     * provably small: every filter carries a `limit` and the limits sum
+     * to at most [INLINE_REPLAY_MAX_ROWS]. Inline replays run on this
+     * connection's receive coroutine — no per-REQ launch, no dispatcher
+     * handoffs (profiled at ~2/3 of a ~20-row REQ's time-to-EOSE) — at
+     * the price of delaying the connection's NEXT command until EOSE,
+     * which a bounded replay keeps in the tens-of-rows range. Unbounded
+     * REQs keep the launched path so a CLOSE can always interrupt them.
+     */
+    private fun isInlineEligible(filters: List<Filter>): Boolean {
+        var total = 0
+        for (f in filters) {
+            val limit = f.limit ?: return false
+            total += limit
+            if (total > INLINE_REPLAY_MAX_ROWS) return false
+        }
+        return true
+    }
+
+    private suspend fun handleReq(cmd: ReqCmd) {
        // Ask the policy whether a *new* subscription may open (e.g. a
        // max_subscriptions cap). A re-REQ on an existing id replaces it
        // 1-for-1 and doesn't grow the count, so it's exempt. Checked before
@@ -288,6 +318,54 @@ class RelaySession(
        // Policy may rewrite filters to match the user's access level.
        val filters = (result as PolicyResult.Accepted).cmd.filters

+        // The `["EVENT","<subId>",` prefix is built once per
+        // subscription, not per row (zero-decode paths only).
+        fun framePrefix() =
+            buildString {
+                append("[\"EVENT\",")
+                RawEvent.appendJsonQuoted(this, cmd.subId)
+                append(',')
+            }
+
+        fun sendStored(
+            framePrefix: String,
+            raw: RawEvent,
+        ) = sendRaw(
+            buildString(framePrefix.length + raw.jsonTags.length + raw.content.length + 256) {
+                append(framePrefix)
+                raw.appendJsonObjectTo(this)
+                append(']')
+            },
+        )
+
+        // Small-REQ fast path: a provably bounded replay runs inline on
+        // this receive coroutine — no launch, no dispatcher handoffs —
+        // and only the live-tail handle is retained. See
+        // [SessionBackend.queryRawInline] for the contract.
+        if (!policy.filtersOutgoingEvents && isInlineEligible(filters)) {
+            val handle =
+                try {
+                    val prefix = framePrefix()
+                    store.queryRawInline(
+                        ctx = requestContext,
+                        filters = filters,
+                        onEachStored = { raw -> sendStored(prefix, raw) },
+                        onEachLive = { event -> send(EventMessage(cmd.subId, event)) },
+                        onEose = { send(EoseMessage(cmd.subId)) },
+                    )
+                } catch (e: CancellationException) {
+                    throw e
+                } catch (e: Exception) {
+                    send(ClosedMessage.of(cmd.subId, MachineReadablePrefix.ERROR, e.message ?: "query failed"))
+                    return
+                }
+            if (handle != null) {
+                addSubscription(cmd.subId) { handle.close() }
+                return
+            }
+            // Backend without an inline path: fall through to launch.
+        }
+
        val job =
            scope.launch {
                try {
@@ -308,26 +386,11 @@ class RelaySession(
                        // Zero-decode path: the stored replay splices raw
                        // storage strings straight into wire frames — no tags
                        // parse, no Event materialization, no re-serialize.
-                        // The `["EVENT","<subId>",` prefix is built once per
-                        // subscription, not per row.
-                        val framePrefix =
-                            buildString {
-                                append("[\"EVENT\",")
-                                RawEvent.appendJsonQuoted(this, cmd.subId)
-                                append(',')
-                            }
+                        val prefix = framePrefix()
                        store.queryRaw(
                            ctx = requestContext,
                            filters = filters,
-                            onEachStored = { raw ->
-                                sendRaw(
-                                    buildString(framePrefix.length + raw.jsonTags.length + raw.content.length + 256) {
-                                        append(framePrefix)
-                                        raw.appendJsonObjectTo(this)
-                                        append(']')
-                                    },
-                                )
-                            },
+                            onEachStored = { raw -> sendStored(prefix, raw) },
                            onEachLive = { event -> send(EventMessage(cmd.subId, event)) },
                            onEose = { send(EoseMessage(cmd.subId)) },
                        )
@@ -343,7 +406,7 @@ class RelaySession(
                }
            }

-        addSubscription(cmd.subId, job)
+        addSubscription(cmd.subId) { job.cancel() }
    }

    // -- NIP-01: CLOSE --------------------------------------------------------
@@ -363,5 +426,15 @@ class RelaySession(

        /** Allocates the next process-unique connection id. */
        fun nextConnectionId(): Long = connectionIdSeq.fetchAndAdd(1L)
+
+        /**
+         * Ceiling on the summed filter limits for the inline REQ fast
+         * path. Sized so the worst-case inline replay stays around a
+         * millisecond (~20-row REQs profile at ~0.2 ms of storage work)
+         * — small enough that delaying the connection's next command is
+         * unnoticeable, large enough to cover the profile/thread/
+         * notifications-style lookups clients hammer relays with.
+         */
+        const val INLINE_REPLAY_MAX_ROWS = 256
    }
 }
@@ -258,6 +258,31 @@ class LiveEventStore(
        onEachLive: (Event) -> Unit,
        onEose: () -> Unit,
    ) {
+        val handle = queryRawInline(ctx, filters, onEachStored, onEachLive, onEose)
+        try {
+            // Suspend until the caller's coroutine is cancelled (NIP-01
+            // CLOSE or connection drop); the live tail runs meanwhile.
+            awaitCancellation()
+        } finally {
+            handle.close()
+        }
+    }
+
+    /**
+     * The registration + replay + EOSE core of [queryRaw], run entirely
+     * on the calling coroutine. Small bounded REQs take this directly
+     * (see [SessionBackend.queryRawInline]) and skip the per-REQ
+     * coroutine; [queryRaw] wraps it with `awaitCancellation` for the
+     * launched path. Never returns `null` — this backend always
+     * supports the inline path.
+     */
+    override suspend fun queryRawInline(
+        ctx: RequestContext,
+        filters: List<Filter>,
+        onEachStored: (RawEvent) -> Unit,
+        onEachLive: (Event) -> Unit,
+        onEose: () -> Unit,
+    ): SessionBackend.LiveSubscriptionHandle {
        drainFtsIfSearching(filters)
        val seenLock = AtomicBoolean(false)
        var seenIds: HashSet<String>? = HashSet(1024)
@@ -290,11 +315,14 @@ class LiveEventStore(
                onEachStored(raw)
            }
            onEose()
+            // Drop the dedupe set so the live path stops paying for it.
            seenLocked { seenIds = null }
-            awaitCancellation()
-        } finally {
+        } catch (e: Throwable) {
+            // A failed replay must not leak the registration.
            index.unregister(sub)
+            throw e
        }
+        return SessionBackend.LiveSubscriptionHandle { index.unregister(sub) }
    }

    override suspend fun count(
@@ -82,6 +82,42 @@ interface SessionBackend {
        onEose: () -> Unit,
    ): Unit = query(ctx, filters, onEachLive, onEose)

+    /**
+     * Keeps a live registration alive after [queryRawInline] returned.
+     * [close] detaches it — idempotent, callable from any thread.
+     */
+    fun interface LiveSubscriptionHandle {
+        fun close()
+    }
+
+    /**
+     * Inline variant of [queryRaw] for BOUNDED replays: runs the stored
+     * replay and [onEose] on the *calling* coroutine — no per-REQ
+     * `launch`, no dispatcher handoffs, no job to keep alive — and
+     * returns a [LiveSubscriptionHandle] whose `close()` ends the live
+     * tail (live events keep arriving via [onEachLive] from the ingest
+     * path until then). This is the small-REQ fast path: profiling put
+     * the per-REQ coroutine machinery at ~2/3 of a ~20-row REQ's
+     * time-to-EOSE.
+     *
+     * Callers MUST only use it when the replay is bounded (e.g. every
+     * filter carries a small `limit`): the replay occupies the session's
+     * receive coroutine, so a giant replay here would delay that
+     * connection's subsequent commands — including the CLOSE that could
+     * have cancelled it. Unbounded REQs belong on [queryRaw] in a
+     * separate coroutine.
+     *
+     * Returns `null` when the backend has no inline path (the default) —
+     * callers fall back to [queryRaw].
+     */
+    suspend fun queryRawInline(
+        ctx: RequestContext,
+        filters: List<Filter>,
+        onEachStored: (RawEvent) -> Unit,
+        onEachLive: (Event) -> Unit,
+        onEose: () -> Unit,
+    ): LiveSubscriptionHandle? = null
+
    /** Answers a NIP-45 COUNT with an exact cardinality for the caller in [ctx]. */
    suspend fun count(
        ctx: RequestContext,
@@ -0,0 +1,169 @@
+/*
+ * 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.server
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * NIP-01 semantics of the inline small-REQ fast path (a REQ whose
+ * filters all carry a small `limit` runs its replay on the receive
+ * coroutine — see [RelaySession.INLINE_REPLAY_MAX_ROWS]). The wire
+ * behavior must be indistinguishable from the launched path: stored
+ * replay, EOSE, live tail, CLOSE handling, and same-subId replacement.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class InlineReqFastPathTest {
+    private fun hexId(n: Int): String = n.toString().padStart(64, '0')
+
+    private val pubkey = "46fcbe3065eaf1ae7811465924e48923363ff3f526bd6f73d7c184b16bd8ce4d"
+    private val sig = "0".repeat(128)
+
+    private fun testEvent(
+        idSeed: Int,
+        kind: Int = 1,
+        createdAt: Long = idSeed.toLong(),
+    ) = Event(hexId(idSeed), pubkey, createdAt, kind, emptyArray(), "note $idSeed", sig)
+
+    private suspend fun serverWith(
+        dispatcher: kotlinx.coroutines.CoroutineDispatcher,
+        vararg events: Event,
+    ): NostrServer {
+        val store = EventStore(null)
+        events.forEach { store.insert(it) }
+        return NostrServer(store = store, policyBuilder = { EmptyPolicy }, parentContext = dispatcher)
+    }
+
+    @Test
+    fun boundedReqRepliesStoredThenEoseThenLiveTail() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val server = serverWith(dispatcher, testEvent(1), testEvent(2, kind = 7))
+
+            val frames = mutableListOf<String>()
+            val session = server.connect { frames.add(it) }
+
+            // limit=10 → inline path.
+            session.receive("""["REQ","small",{"kinds":[1],"limit":10}]""")
+
+            assertEquals(1, frames.count { it.startsWith("[\"EVENT\",\"small\"") })
+            assertTrue(frames.first { it.startsWith("[\"EVENT\"") }.contains(hexId(1)))
+            assertTrue(frames.last().startsWith("[\"EOSE\",\"small\""))
+
+            // Live tail: a matching publish after EOSE must reach the sub.
+            session.receive("""["EVENT",${testEvent(3).toJson()}]""")
+            assertTrue(frames.any { it.startsWith("[\"EVENT\",\"small\"") && it.contains(hexId(3)) })
+            // Non-matching kind stays out.
+            session.receive("""["EVENT",${testEvent(4, kind = 7).toJson()}]""")
+            assertTrue(frames.none { it.startsWith("[\"EVENT\",\"small\"") && it.contains(hexId(4)) })
+
+            server.close()
+        }
+
+    @Test
+    fun closeDetachesTheInlineLiveTail() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val server = serverWith(dispatcher, testEvent(1))
+
+            val frames = mutableListOf<String>()
+            val session = server.connect { frames.add(it) }
+
+            session.receive("""["REQ","small",{"kinds":[1],"limit":10}]""")
+            session.receive("""["CLOSE","small"]""")
+
+            session.receive("""["EVENT",${testEvent(2).toJson()}]""")
+            assertTrue(frames.none { it.startsWith("[\"EVENT\",\"small\"") && it.contains(hexId(2)) })
+
+            server.close()
+        }
+
+    @Test
+    fun sameSubIdReplacesInlineSubscription() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val server = serverWith(dispatcher, testEvent(1), testEvent(2, kind = 7))
+
+            val frames = mutableListOf<String>()
+            val session = server.connect { frames.add(it) }
+
+            session.receive("""["REQ","sub",{"kinds":[1],"limit":10}]""")
+            // Replace with a kind-7 filter (still inline). The old live
+            // tail must be gone: a new kind-1 publish stays silent, a
+            // kind-7 one is delivered.
+            session.receive("""["REQ","sub",{"kinds":[7],"limit":10}]""")
+
+            session.receive("""["EVENT",${testEvent(3, kind = 1).toJson()}]""")
+            assertTrue(frames.none { it.startsWith("[\"EVENT\",\"sub\"") && it.contains(hexId(3)) })
+            session.receive("""["EVENT",${testEvent(4, kind = 7).toJson()}]""")
+            assertTrue(frames.any { it.startsWith("[\"EVENT\",\"sub\"") && it.contains(hexId(4)) })
+
+            server.close()
+        }
+
+    @Test
+    fun unboundedReqStillWorksViaLaunchedPath() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val server = serverWith(dispatcher, testEvent(1))
+
+            val frames = mutableListOf<String>()
+            val session = server.connect { frames.add(it) }
+
+            // No limit → not inline-eligible; must behave identically.
+            session.receive("""["REQ","big",{"kinds":[1]}]""")
+
+            assertTrue(frames.any { it.startsWith("[\"EVENT\",\"big\"") && it.contains(hexId(1)) })
+            assertTrue(frames.any { it.startsWith("[\"EOSE\",\"big\"") })
+
+            session.receive("""["EVENT",${testEvent(2).toJson()}]""")
+            assertTrue(frames.any { it.startsWith("[\"EVENT\",\"big\"") && it.contains(hexId(2)) })
+
+            server.close()
+        }
+
+    @Test
+    fun oversizedLimitSumIsNotInlineEligible() =
+        runTest {
+            // Behavior parity either way — this documents the boundary:
+            // summed limits over the cap route to the launched path and
+            // still answer correctly.
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val server = serverWith(dispatcher, testEvent(1))
+
+            val frames = mutableListOf<String>()
+            val session = server.connect { frames.add(it) }
+
+            session.receive("""["REQ","big",{"kinds":[1],"limit":${RelaySession.INLINE_REPLAY_MAX_ROWS + 1}}]""")
+
+            assertTrue(frames.any { it.startsWith("[\"EVENT\",\"big\"") && it.contains(hexId(1)) })
+            assertTrue(frames.any { it.startsWith("[\"EOSE\",\"big\"") })
+
+            server.close()
+        }
+}
@@ -0,0 +1,176 @@
+/*
+ * 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.relay.server.NostrServer
+import com.vitorpamplona.quartz.nip01Core.relay.server.backend.IngestQueue
+import com.vitorpamplona.quartz.nip01Core.relay.server.backend.LiveEventStore
+import com.vitorpamplona.quartz.nip01Core.relay.server.backend.RequestContext
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
+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.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+/**
+ * Decomposes the fixed per-REQ cost on SMALL results — relayBench shows
+ * geode ~2.5× slower than strfry when a REQ returns ~20 events
+ * (author-archive 1.18 vs 0.48 ms EOSE p50 at 50k), while geode WINS the
+ * 500-event scenarios. With tiny result sets the per-REQ floor dominates,
+ * so this isolates its stages:
+ *
+ *  A. raw store query (SQL + row decode) — the unavoidable part
+ *  B. LiveEventStore.queryRaw to EOSE — adds live-subscription
+ *     registration (FilterIndex), the per-REQ dedupe HashSet, and the
+ *     search-drain check
+ *  C. full session dispatch (RelaySession.receive of the REQ json) —
+ *     adds command parse, policy, coroutine launch, frame assembly,
+ *     and the send callback
+ *
+ * B−A is the live machinery; C−B is dispatch+serialization. Printed
+ * per-stage medians tell which one owns the floor. No speed assertions —
+ * numbers are informational; changes get A/B'd in relayBench.
+ */
+class SmallReqFloorBenchmark {
+    companion object {
+        const val EVENTS = 50_000
+        const val AUTHORS = 2_500 // ~20 events per author, matching author-archive
+        const val ROUNDS = 400
+        const val WARMUP = 100
+    }
+
+    private fun hexId(seed: Int): String = seed.toString(16).padStart(64, '0')
+
+    private fun pubkey(seed: Int): String = (seed % AUTHORS).toString(16).padStart(64, 'a')
+
+    private val sig = "0".repeat(128)
+
+    private fun event(seed: Int): Event =
+        EventFactory.create(
+            id = hexId(seed),
+            pubKey = pubkey(seed),
+            createdAt = 1_600_000_000L + (seed * 7919) % 1_000_000,
+            kind = 1,
+            tags = emptyArray(),
+            content = "small req floor benchmark $seed",
+            sig = sig,
+        )
+
+    private fun median(samples: LongArray): Double {
+        samples.sort()
+        return samples[samples.size / 2] / 1e6
+    }
+
+    @Test
+    fun perReqFloorAt50k() =
+        runBlocking {
+            val store = EventStore(dbName = null, indexStrategy = DefaultIndexingStrategy(indexEventsByPubkeyAlone = true))
+            (1..EVENTS).chunked(2000).forEach { chunk -> store.batchInsert(chunk.map { event(it) }) }
+
+            val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+            val live = LiveEventStore(store, IngestQueue(store, scope.coroutineContext))
+            val ctx =
+                object : RequestContext {
+                    override val connectionId = 0L
+                    override val policy = EmptyPolicy
+                    override val authenticatedUsers = emptySet<String>()
+                }
+            val server = NostrServer(store = store, policyBuilder = { EmptyPolicy }, parentContext = scope.coroutineContext)
+
+            fun filterFor(round: Int) = Filter(authors = listOf(pubkey(round)), kinds = listOf(1), limit = 50)
+
+            fun filterJson(round: Int) = """["REQ","floor$round",{"authors":["${pubkey(round)}"],"kinds":[1],"limit":50}]"""
+
+            // --- A: raw store query ---
+            val a = LongArray(ROUNDS)
+            var rowsA = 0
+            repeat(WARMUP) { store.rawQuery(listOf(filterFor(it))) {} }
+            repeat(ROUNDS) { round ->
+                val t0 = System.nanoTime()
+                var n = 0
+                store.rawQuery(listOf(filterFor(round))) { n++ }
+                a[round] = System.nanoTime() - t0
+                rowsA += n
+            }
+
+            // --- B: backend queryRaw to EOSE (live machinery included) ---
+            suspend fun timeBackend(round: Int): Long {
+                val eose = CompletableDeferred<Long>()
+                val t0 = System.nanoTime()
+                val job =
+                    scope.launch {
+                        live.queryRaw(
+                            ctx = ctx,
+                            filters = listOf(filterFor(round)),
+                            onEachStored = {},
+                            onEachLive = {},
+                            onEose = { eose.complete(System.nanoTime() - t0) },
+                        )
+                    }
+                val nanos = eose.await()
+                job.cancel()
+                return nanos
+            }
+            repeat(WARMUP) { timeBackend(it) }
+            val b = LongArray(ROUNDS)
+            repeat(ROUNDS) { b[it] = timeBackend(it) }
+
+            // --- C: full session dispatch, REQ json in → EOSE frame out ---
+            suspend fun timeSession(round: Int): Long {
+                val eose = CompletableDeferred<Long>()
+                val t0 = System.nanoTime()
+                val session =
+                    server.connect { frame ->
+                        if (frame.startsWith("[\"EOSE\"")) eose.complete(System.nanoTime() - t0)
+                    }
+                session.receive(filterJson(round))
+                val nanos = eose.await()
+                session.close()
+                return nanos
+            }
+            repeat(WARMUP) { timeSession(it) }
+            val c = LongArray(ROUNDS)
+            repeat(ROUNDS) { c[it] = timeSession(it) }
+
+            assertEquals(true, rowsA > 0, "author filters must return rows")
+
+            val mA = median(a)
+            val mB = median(b)
+            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("  C session REQ→EOSE:       ${"%6.3f".format(mC)} ms  (dispatch+frames +${"%6.3f".format(mC - mB)})")
+
+            server.close()
+            scope.cancel()
+        }
+}