diff --git a/quartz/plans/2026-07-04-small-req-floor.md b/quartz/plans/2026-07-04-small-req-floor.md new file mode 100644 index 0000000000..7723eba0c5 --- /dev/null +++ b/quartz/plans/2026-07-04-small-req-floor.md @@ -0,0 +1,58 @@ +# Small-REQ dispatch floor — investigated, inline fast path reverted + +**Status: closed (negative result recorded).** Backlog item 2 of the +relay performance campaign. + +## The gap + +relayBench at 50k events: geode WINS most 500-event query scenarios +(hashtag 5.4 vs 8.4 ms, recent-window 4.1 vs 8.5) but loses ~2.5× on +small results — author-archive (19 events) 1.2–1.7 ms vs strfry's +~0.5–0.6, thread (23) likewise — and the @8conn throughput inverts +(strfry 2–3× geode). With ~20-row responses, throughput ≈ 1/latency: +there is a fixed per-REQ floor. + +## Decomposition (SmallReqFloorBenchmark, kept in jvmTest) + +In-process at 50k events, ~21 rows/REQ, medians of 400: + +| stage | ms | +|---|---:| +| A raw store query (SQL + row decode) | 0.18 | +| B + live machinery (FilterIndex reg/unreg, dedupe set) | 0.36 | +| C + session dispatch (parse, launch, frames) | 0.60 | + +Note: in-memory DBs have no reader pool (`useReader` falls back to the +writer mutex), so absolute numbers are conservative vs the file-DB +bench setup. + +## What was tried and why it was reverted + +An inline fast path (`SessionBackend.queryRawInline`): REQs with +provably bounded replays (limit or ids-count summing ≤ 512) ran their +stored replay on the receive coroutine and kept only a live-tail +handle — no per-REQ `launch`, no Job, no dispatcher handoffs. It cut +in-process time-to-EOSE ~17% (0.60 → 0.50 ms) with full wire-behavior +parity (stored→EOSE order, live tail, CLOSE, same-subId replacement). + +Three relayBench runs (baseline, cap-256 [path not engaged — bench +filters carry limit=500 or no limit], cap-512 [engaged for +author-archive/by-ids/500-limit feeds]) showed **no movement outside +the container drift band** — strfry's own numbers drifted ±30% run to +run, and inline-eligible scenarios moved the same as ineligible ones. +Reverted per the keep-only-winners rule. + +## Where the floor actually is + +In-process REQ→EOSE is ~0.5–0.6 ms, but the wire-level p50 is +1.2–1.7 ms: the missing ~1 ms per REQ is transport-side — the Ktor CIO +frame write path, per-frame sends with no batching, and the client +round trip — which is backlog item 6 (websocket send path, 13–25% of +ingest CPU in the JFR profile) territory. strfry completes the whole +round trip in under 0.6 ms on a single event loop with uWebSockets. + +**Do not retry** coroutine-dispatch shaving for this gap without first +measuring the transport side: instrument the time between +`RelaySession`'s `onSend` invocation and the frame actually leaving +the socket, and compare permessage-deflate/frame-batching settings +against strfry's uWS configuration. diff --git a/quartz/plans/README.md b/quartz/plans/README.md index 577d380717..d8ce0a8cab 100644 --- a/quartz/plans/README.md +++ b/quartz/plans/README.md @@ -1,6 +1,6 @@ # quartz plans -_Audited 2026-06-30. 10 plans: 7 shipped (archived), 0 in-progress, 3 queued, 0 abandoned._ +_Audited 2026-06-30. 11 plans: 7 shipped (archived), 0 in-progress, 3 queued, 1 closed (negative result)._ ## Queued | Plan | Summary | @@ -8,6 +8,7 @@ _Audited 2026-06-30. 10 plans: 7 shipped (archived), 0 in-progress, 3 queued, 0 | [2026-05-08-local-headers-explorer.md](2026-05-08-local-headers-explorer.md) | Headers-only Bitcoin P2P client to verify NIP-03 OTS attestations without a trusted block explorer. | | [2026-06-12-giftwrap-deletion-requests.md](2026-06-12-giftwrap-deletion-requests.md) | Let a recipient-authored kind-5 delete/block a gift wrap (kind 1059) addressed to them. | | [2026-07-03-incremental-negentropy-storage.md](2026-07-03-incremental-negentropy-storage.md) | Always-current (created_at, id) index so cold NEG-OPENs stop paying a full scan + seal (~340 ms at 50k vs strfry's ~21 ms). | +| [2026-07-04-small-req-floor.md](2026-07-04-small-req-floor.md) | Small-REQ dispatch floor: decomposed, inline fast path tried and reverted (no wire-level win); floor is transport-side. | ## Archived (shipped) | Plan | Summary | 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 5f5c290af8..05cec0fad3 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,7 +35,6 @@ 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 @@ -50,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.Job import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.launch import kotlin.concurrent.atomics.AtomicLong @@ -75,16 +75,7 @@ class RelaySession( */ val id: Long = nextConnectionId(), ) : AutoCloseable { - /** - * 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() + private val subscriptions = LargeCache() /** * The authenticated-identity store for this connection. The engine is the @@ -112,8 +103,8 @@ class RelaySession( private fun addSubscription( subId: String, - sub: ActiveSubscription, - ) = subscriptions.put(subId, sub) + job: Job, + ) = subscriptions.put(subId, job) private fun cancelSubscription(subId: String): Boolean = subscriptions.remove(subId)?.let { @@ -122,7 +113,7 @@ class RelaySession( } ?: false fun cancelAllSubscriptions() { - subscriptions.forEach { _, sub -> sub.cancel() } + subscriptions.forEach { _, job -> job.cancel() } subscriptions.clear() negentropy.clear() } @@ -272,30 +263,7 @@ class RelaySession( } // -- NIP-01: REQ ---------------------------------------------------------- - - /** - * A REQ qualifies for the inline fast path when its replay is - * provably bounded: every filter carries a `limit`, or an `ids` list - * (ids are unique keys, so the result can't exceed the list), and - * the bounds 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 the row cap keeps in the - * single-digit-milliseconds range. Unbounded REQs keep the launched - * path so a CLOSE can always interrupt a genuinely giant replay. - */ - private fun isInlineEligible(filters: List): Boolean { - var total = 0 - for (f in filters) { - val bound = f.limit ?: f.ids?.size ?: return false - total += bound - if (total > INLINE_REPLAY_MAX_ROWS) return false - } - return true - } - - private suspend fun handleReq(cmd: ReqCmd) { + private 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 @@ -320,54 +288,6 @@ 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 { @@ -388,11 +308,26 @@ 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. - val prefix = framePrefix() + // The `["EVENT","",` prefix is built once per + // subscription, not per row. + val framePrefix = + buildString { + append("[\"EVENT\",") + RawEvent.appendJsonQuoted(this, cmd.subId) + append(',') + } store.queryRaw( ctx = requestContext, filters = filters, - onEachStored = { raw -> sendStored(prefix, raw) }, + onEachStored = { raw -> + sendRaw( + buildString(framePrefix.length + raw.jsonTags.length + raw.content.length + 256) { + append(framePrefix) + raw.appendJsonObjectTo(this) + append(']') + }, + ) + }, onEachLive = { event -> send(EventMessage(cmd.subId, event)) }, onEose = { send(EoseMessage(cmd.subId)) }, ) @@ -408,7 +343,7 @@ class RelaySession( } } - addSubscription(cmd.subId) { job.cancel() } + addSubscription(cmd.subId, job) } // -- NIP-01: CLOSE -------------------------------------------------------- @@ -428,16 +363,5 @@ class RelaySession( /** Allocates the next process-unique connection id. */ fun nextConnectionId(): Long = connectionIdSeq.fetchAndAdd(1L) - - /** - * Ceiling on the summed per-filter bounds (limit or ids count) - * for the inline REQ fast path. Sized so the worst-case inline - * replay stays in single-digit milliseconds (500-row replays - * measure ~5–8 ms of storage+frame work) — small enough that - * delaying the connection's next command is unnoticeable, large - * enough to cover the limit≤500 feed/notifications/archive - * shapes real clients hammer relays with. - */ - const val INLINE_REPLAY_MAX_ROWS = 512 } } 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 a397156143..1d35362134 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,31 +258,6 @@ 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) @@ -315,14 +290,11 @@ class LiveEventStore( onEachStored(raw) } onEose() - // Drop the dedupe set so the live path stops paying for it. seenLocked { seenIds = null } - } catch (e: Throwable) { - // A failed replay must not leak the registration. + awaitCancellation() + } finally { 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 eb3ab83ce5..1119d9e9be 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,42 +82,6 @@ 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 deleted file mode 100644 index d6e33b8a5e..0000000000 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InlineReqFastPathTest.kt +++ /dev/null @@ -1,169 +0,0 @@ -/* - * 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() - } -}