diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index 05cec0fad3..0ac21a81b5 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -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() + /** + * 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() /** * 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): 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","",` 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","",` 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 } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt index 1d35362134..a397156143 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt @@ -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, + onEachStored: (RawEvent) -> Unit, + onEachLive: (Event) -> Unit, + onEose: () -> Unit, + ): SessionBackend.LiveSubscriptionHandle { drainFtsIfSearching(filters) val seenLock = AtomicBoolean(false) var seenIds: HashSet? = 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( diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt index 1119d9e9be..eb3ab83ce5 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt @@ -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, + 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, diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InlineReqFastPathTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InlineReqFastPathTest.kt new file mode 100644 index 0000000000..d6e33b8a5e --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InlineReqFastPathTest.kt @@ -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() + 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() + 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() + 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() + 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() + 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() + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt new file mode 100644 index 0000000000..9fe3a6ec33 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt @@ -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() + } + 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() + 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() + 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() + } +}