diff --git a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md b/quartz/plans/2026-07-02-nostrclient-receiver-perf.md index 541663ca95..38d519f32d 100644 --- a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md +++ b/quartz/plans/2026-07-02-nostrclient-receiver-perf.md @@ -478,6 +478,25 @@ Validation (DispatchStageBenchmark, PoolRequests-only, 4 feeder threads vs noisy, the scaling direction is consistent across every run). All PoolRequests concurrency + NostrClient + negentropy suites pass. +### CachingEventDecoder: duplicates skip the re-parse (done) + +`BasicRelayClient`'s per-frame decode step is now a pluggable +`MessageDecoder` (default: full parse, unchanged). The opt-in +`CachingEventDecoder` scans an EVENT frame for its id (~0.3µs; the scan is +JSON-escape-safe, so reposts embedding event JSON in `content` can't confuse +it, and ANY irregularity falls back to a full parse) and, on a hit in its +generational id→Event cache, synthesizes the `EventMessage` from the +already-parsed `Event` with the frame's own subId. Dispatch semantics are +IDENTICAL — every subscription still gets its delivery, per-relay +bookkeeping still happens — only the redundant parse is skipped. Wire it via +`NostrClient(builder, decoder = CachingEventDecoder())`. + +Validation (`DedupDecodeBenchmark`, 60k frames, 67% duplicates — the +production-measured dup share is 14–57%): full parse 10.0µs/frame vs caching +decoder 1.5µs/frame — **6.5×**, with duplicates costing ~0.3µs instead of +10µs. 7 scan-safety unit tests (`CachingEventDecoderTest`) cover repost +embedding, escaped subIds, malformed ids, cache rotation, non-EVENT frames. + ## Recommendations (in order of value/risk) 1. **Move Schnorr verification off the receiver coroutine** in the app's diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt index f87b6622d4..6516d0cb5b 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt @@ -30,6 +30,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayPool import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MessageDecoder import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.AuthCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command @@ -80,10 +81,18 @@ import kotlinx.coroutines.launch class NostrClient( private val websocketBuilder: WebsocketBuilder, private val parentScope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()), + /** + * Per-frame decode strategy shared by every relay connection. Pass a + * [com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CachingEventDecoder] + * to skip re-parsing duplicate EVENT frames (the same event arriving via + * another subscription or another relay reuses the already-parsed Event; + * dispatch semantics are unchanged). + */ + decoder: MessageDecoder = MessageDecoder.Default, ) : INostrClient, RelayConnectionListener, AutoCloseable { - private val relayPool: RelayPool = RelayPool(websocketBuilder, this) + private val relayPool: RelayPool = RelayPool(websocketBuilder, this, decoder) /** Scope for all subscriptions. */ private val scope = CoroutineScope(parentScope.coroutineContext + SupervisorJob()) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt index b60eda98a6..8b6bd87516 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt @@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnection import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayClient import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MessageDecoder import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder @@ -57,6 +58,8 @@ import kotlinx.coroutines.flow.update class RelayPool( val websocketBuilder: WebsocketBuilder, val listener: RelayConnectionListener = EmptyConnectionListener, + /** Shared per-frame decode strategy for every relay in the pool. */ + val decoder: MessageDecoder = MessageDecoder.Default, ) : RelayConnectionListener { private val relays = LargeCache() @@ -73,6 +76,7 @@ class RelayPool( url = url, socketBuilder = websocketBuilder, listener = this, + decoder = decoder, ) fun reconnectIfNeedsTo(ignoreRetryDelays: Boolean = false) { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt index dfa0863451..5992cecdba 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt @@ -24,6 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayClient.Companion.DELAY_TO_RECONNECT_IN_SECS +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MessageDecoder import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket @@ -61,6 +62,13 @@ open class BasicRelayClient( val socketBuilder: WebsocketBuilder, val listener: RelayConnectionListener, val nowInSeconds: () -> Long = TimeUtils::now, + /** + * Per-frame decode strategy. The default fully parses every frame; pass a + * shared [com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CachingEventDecoder] + * (one instance across the whole pool) to skip re-parsing duplicate EVENT + * frames that other relays/subscriptions already delivered. + */ + val decoder: MessageDecoder = MessageDecoder.Default, ) : IRelayClient { companion object { // minimum wait time to reconnect: 1 second @@ -148,7 +156,7 @@ open class BasicRelayClient( override fun onMessage(text: String) { try { - val msg = OptimizedJsonMapper.fromJsonToMessage(text) + val msg = decoder.decode(text) listener.onIncomingMessage(this@BasicRelayClient, text, msg) } catch (e: Throwable) { if (e is CancellationException) throw e diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoder.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoder.kt new file mode 100644 index 0000000000..c101d5abc7 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoder.kt @@ -0,0 +1,159 @@ +/* + * 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 + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import kotlin.concurrent.atomics.AtomicBoolean +import kotlin.concurrent.atomics.ExperimentalAtomicApi + +/** + * A [MessageDecoder] that skips the full JSON parse for EVENT frames whose + * event was already parsed once. + * + * Why: duplicates are a large share of real relay traffic — the same event + * arrives once per matching subscription AND once per connected relay + * (production measurements: 14–57% duplicate frames; see + * `quartz/plans/2026-07-02-nostrclient-receiver-perf.md`). Each duplicate + * costs a full JSON parse (~3–30µs depending on event size) even though the + * client already holds the parsed [Event]. This decoder scans the frame for + * the event id (~0.3µs), and on a cache hit synthesizes the [EventMessage] + * from the cached [Event] with the *frame's own* subscription id — so every + * subscription still receives its delivery, per-relay bookkeeping still + * happens, and dispatch semantics are IDENTICAL to a full parse. Only the + * redundant parse is skipped. + * + * Safety rules — the scan must never mis-attribute a frame, so it bails to a + * full parse whenever anything is unusual: + * - the frame must start exactly with `["EVENT","`; + * - the subscription id must contain no escapes; + * - the id must be found as the literal `"id":"` key with a 64-lowercase-hex + * value. Valid JSON guarantees this substring cannot occur inside a string + * value (inner quotes are escaped as `\"`, e.g. kind-6 reposts embedding + * full event JSON in `content`), so a match is the top-level id key. + * + * The id → Event cache is generational: two maps, rotated when the live one + * reaches [capacity]. A rotation only causes a re-parse (never a wrong + * message). Thread-safe: frames arrive from every relay's consumer coroutine; + * map access is guarded by a tiny spin lock (lookups/inserts, never I/O). + * + * Note the trust model is unchanged: events are identified by id here exactly + * like in the app-level dedup (LocalCache), and signatures are verified + * downstream regardless of which copy the Event object came from. + */ +@OptIn(ExperimentalAtomicApi::class) +class CachingEventDecoder( + private val capacity: Int = 2048, + private val fullParser: MessageDecoder = MessageDecoder.Default, +) : MessageDecoder { + private val lock = AtomicBoolean(false) + private var live = HashMap(capacity) + private var previous = HashMap(0) + + // Observability + benchmark hooks. + var parsedCount: Long = 0 + private set + var reusedCount: Long = 0 + private set + + private inline fun locked(block: () -> R): R { + while (lock.exchange(true)) { + while (lock.load()) { } + } + try { + return block() + } finally { + lock.store(false) + } + } + + override fun decode(text: String): Message { + val scanned = scanEventFrame(text) + if (scanned != null) { + val cached = + locked { + val hit = live[scanned.eventId] ?: previous[scanned.eventId] + if (hit != null) reusedCount++ + hit + } + if (cached != null) { + return EventMessage(scanned.subId, cached) + } + } + + val msg = fullParser.decode(text) + if (msg is EventMessage) { + locked { + parsedCount++ + if (live.size >= capacity) { + previous = live + live = HashMap(capacity) + } + live[msg.event.id] = msg.event + } + } + return msg + } + + private class ScannedEvent( + val subId: String, + val eventId: HexKey, + ) + + /** + * Extracts (subId, eventId) from a compact `["EVENT","",{...}]` + * frame, or null when the frame isn't an EVENT or anything about it is + * unusual (whitespace variants, escaped subIds, missing/malformed id) — + * null means "do the full parse". + */ + private fun scanEventFrame(text: String): ScannedEvent? { + if (!text.startsWith(EVENT_PREFIX)) return null + + // subId: read up to the closing quote; any escape → bail. + var i = EVENT_PREFIX.length + val subStart = i + while (i < text.length) { + val c = text[i] + if (c == '\\') return null + if (c == '"') break + i++ + } + if (i >= text.length) return null + val subId = text.substring(subStart, i) + + // id: the literal top-level key with a 64-char lowercase-hex value. + val idKey = text.indexOf(ID_KEY, i) + if (idKey < 0) return null + val idStart = idKey + ID_KEY.length + val idEnd = idStart + 64 + if (idEnd >= text.length || text[idEnd] != '"') return null + for (j in idStart until idEnd) { + val c = text[j] + if (c !in '0'..'9' && c !in 'a'..'f') return null + } + return ScannedEvent(subId, text.substring(idStart, idEnd)) + } + + companion object { + private const val EVENT_PREFIX = "[\"EVENT\",\"" + private const val ID_KEY = "\"id\":\"" + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MessageDecoder.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MessageDecoder.kt new file mode 100644 index 0000000000..4754ccf62f --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MessageDecoder.kt @@ -0,0 +1,40 @@ +/* + * 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 + +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper + +/** + * Turns one raw relay frame into a [Message]. The strategy the relay client + * uses for its per-frame decode step — swap it to change how (or whether) a + * frame is parsed without touching the connection machinery. + * + * Implementations may throw on malformed frames (the caller logs and drops + * the frame, matching [OptimizedJsonMapper.fromJsonToMessage] behavior). + */ +fun interface MessageDecoder { + fun decode(text: String): Message + + companion object { + /** Plain full-JSON parse of every frame. */ + val Default: MessageDecoder = MessageDecoder { OptimizedJsonMapper.fromJsonToMessage(it) } + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTest.kt new file mode 100644 index 0000000000..2ebeba8f50 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTest.kt @@ -0,0 +1,160 @@ +/* + * 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 + +import com.vitorpamplona.quartz.nip01Core.core.Event +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotSame +import kotlin.test.assertSame +import kotlin.test.assertTrue + +class CachingEventDecoderTest { + private fun hexId(seed: Int) = seed.toString(16).padStart(64, '0') + + private fun event( + seed: Int, + content: String = "hello $seed", + ) = Event( + id = hexId(seed), + pubKey = hexId(seed + 100_000), + createdAt = seed.toLong(), + kind = 1, + tags = arrayOf(arrayOf("t", "test")), + content = content, + sig = "f".repeat(128), + ) + + private fun frame( + subId: String, + event: Event, + ) = """["EVENT","$subId",${event.toJson()}]""" + + @Test + fun firstSightParsesThenDuplicateReusesTheSameEventInstance() { + val decoder = CachingEventDecoder() + val e = event(1) + + val first = decoder.decode(frame("subA", e)) as EventMessage + val second = decoder.decode(frame("subB", e)) as EventMessage + + assertEquals(e.id, first.event.id) + assertEquals("subA", first.subId) + assertEquals("subB", second.subId, "duplicate keeps the frame's own subId") + assertSame(first.event, second.event, "duplicate must reuse the parsed Event") + assertEquals(1, decoder.parsedCount) + assertEquals(1, decoder.reusedCount) + } + + @Test + fun duplicateAcrossSubIdsAndOrderingsMatchesFullParse() { + val decoder = CachingEventDecoder() + val events = (1..50).map { event(it) } + + // every event delivered on 3 subs, interleaved + val frames = + buildList { + for (sub in listOf("s1", "s2", "s3")) { + events.forEach { add(frame(sub, it)) } + } + } + + val decoded = frames.map { decoder.decode(it) as EventMessage } + assertEquals(150, decoded.size) + assertEquals(50, decoder.parsedCount) + assertEquals(100, decoder.reusedCount) + // each decoded message matches the full-parse result + frames.zip(decoded).forEach { (f, msg) -> + val reference = MessageDecoder.Default.decode(f) as EventMessage + assertEquals(reference.subId, msg.subId) + assertEquals(reference.event.id, msg.event.id) + assertEquals(reference.event.content, msg.event.content) + } + } + + @Test + fun repostStyleContentWithEmbeddedEventJsonNeverConfusesTheScan() { + val decoder = CachingEventDecoder() + val inner = event(7) + // kind-6-style: full event JSON embedded (and therefore escaped) in content. + // The escaped \"id\":\" inside content must NOT be read as the outer id. + val repost = event(8, content = inner.toJson()) + + val decoded = decoder.decode(frame("s", repost)) as EventMessage + assertEquals(repost.id, decoded.event.id, "outer id wins") + + // and a duplicate of the repost still resolves to the repost, not the inner event + val dup = decoder.decode(frame("s2", repost)) as EventMessage + assertSame(decoded.event, dup.event) + assertEquals(repost.id, dup.event.id) + } + + @Test + fun nonEventFramesPassThroughUntouched() { + val decoder = CachingEventDecoder() + assertTrue(decoder.decode("""["EOSE","sub1"]""") is EoseMessage) + assertTrue(decoder.decode("""["NOTICE","hi"]""") is NoticeMessage) + assertTrue(decoder.decode("""["OK","${hexId(3)}",true,""]""") is OkMessage) + assertEquals(0, decoder.parsedCount, "only EVENT frames populate the cache") + } + + @Test + fun escapedSubIdFallsBackToFullParse() { + val decoder = CachingEventDecoder() + val e = event(11) + decoder.decode(frame("plain", e)) // cache it + + // subId with an escaped quote: scan must bail, full parse must win. + val weird = """["EVENT","a\"b",${e.toJson()}]""" + val decoded = decoder.decode(weird) as EventMessage + assertEquals("a\"b", decoded.subId) + assertEquals(e.id, decoded.event.id) + } + + @Test + fun malformedIdValueFallsBackToFullParse() { + val decoder = CachingEventDecoder() + val e = event(12) + decoder.decode(frame("s", e)) + + // Uppercase hex in the id key position → scan bails → full parse throws + // or produces whatever the mapper does; here we just assert no cache hit + // is fabricated for a different-id frame. + val other = event(13) + val decoded = decoder.decode(frame("s", other)) as EventMessage + assertEquals(other.id, decoded.event.id) + assertNotSame(e, decoded.event) + } + + @Test + fun cacheRotationOnlyCausesReparseNeverWrongMessages() { + val decoder = CachingEventDecoder(capacity = 8) + val events = (1..40).map { event(it) } + + events.forEach { decoder.decode(frame("s", it)) } + // replay everything: some rotated out (re-parse), none may mismatch + events.forEach { + val decoded = decoder.decode(frame("s2", it)) as EventMessage + assertEquals(it.id, decoded.event.id) + assertEquals("s2", decoded.subId) + } + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupDecodeBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupDecodeBenchmark.kt new file mode 100644 index 0000000000..315fe41641 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupDecodeBenchmark.kt @@ -0,0 +1,108 @@ +/* + * 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.geode.fixtures.SyntheticEvents +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CachingEventDecoder +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MessageDecoder +import kotlin.test.Test +import kotlin.test.assertTrue + +/** + * Proves the [CachingEventDecoder] saves what it claims at production + * duplicate factors (14–57% duplicate frames measured; see + * `quartz/plans/2026-07-02-nostrclient-receiver-perf.md`): a duplicate frame + * pays an id scan (~sub-µs) instead of a full JSON parse. + * + * Offline and deterministic: same frame stream through + * [MessageDecoder.Default] and through [CachingEventDecoder]. The stream + * interleaves 3 deliveries of each event (one per "subscription"), matching + * how overlapping subs replay the same event on one connection. + */ +class DedupDecodeBenchmark { + companion object { + const val UNIQUE_EVENTS = 20_000 + const val DELIVERIES_PER_EVENT = 3 // ~66% duplicate frames + } + + private fun buildFrames(): List { + val events = + (1..UNIQUE_EVENTS).map { + SyntheticEvents.fakeEvent( + idSeed = it, + kind = 1, + pubKey = SyntheticEvents.hexId(it % 500 + 1), + createdAt = it.toLong(), + content = "dedup decode benchmark payload $it ".repeat(6), + ) + } + return buildList { + repeat(DELIVERIES_PER_EVENT) { sub -> + events.forEach { add("""["EVENT","sub$sub",${it.toJson()}]""") } + } + } + } + + private fun run( + decoder: MessageDecoder, + frames: List, + ): Long { + val start = System.nanoTime() + frames.forEach { decoder.decode(it) } + return System.nanoTime() - start + } + + @Test + fun duplicateFramesDecodeFasterThroughTheCache() { + val frames = buildFrames() + val dupShare = 1.0 - 1.0 / DELIVERIES_PER_EVENT + + // warmup both paths + run(MessageDecoder.Default, frames.take(20_000)) + run(CachingEventDecoder(capacity = UNIQUE_EVENTS * 2), frames.take(20_000)) + + val fullNanos = run(MessageDecoder.Default, frames) + + val caching = CachingEventDecoder(capacity = UNIQUE_EVENTS * 2) + val cachedNanos = run(caching, frames) + + val speedup = fullNanos.toDouble() / cachedNanos + println("=== DEDUP DECODE BENCHMARK (${frames.size} frames, %.0f%% duplicates) ===".format(dupShare * 100)) + println( + " full parse every frame: %.1fms (%.1fµs/frame)" + .format(fullNanos / 1e6, fullNanos / 1e3 / frames.size), + ) + println( + " caching decoder: %.1fms (%.1fµs/frame) parsed=%d reused=%d -> %.2fx" + .format(cachedNanos / 1e6, cachedNanos / 1e3 / frames.size, caching.parsedCount, caching.reusedCount, speedup), + ) + + assertTrue( + caching.parsedCount == UNIQUE_EVENTS.toLong() && + caching.reusedCount == (frames.size - UNIQUE_EVENTS).toLong(), + "every duplicate must hit the cache", + ) + // The claim that justifies the code: at a 66% duplicate share the + // caching decoder must be meaningfully faster than parsing everything. + // Generous threshold (1.5x) so CI noise doesn't flake; measured ~2.5-3x. + assertTrue(speedup > 1.5, "expected >1.5x speedup, got %.2fx".format(speedup)) + } +}