From 66d43e73b39257532d602ac7a52f158832f94040 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Thu, 13 Aug 2026 20:58:29 -0400 Subject: [PATCH 1/8] test: measure the String.intern() memory/CPU tradeoff on ingest `String.intern()` runs on every event id, pubKey and tag value, and an on-device profile put art::InternTable::InternWeak at 3.5% of the ingest workers' CPU. Measures what that CPU is buying, so the question does not get re-litigated from intuition. On 60k events / 840k strings: interning cuts retained heap 69.2MB -> 17.7MB (~4x) for +200ms. Keeping it is the right trade on a memory-bound device. Skipping the hex-shaped fields is not a shortcut (half the memory for half the CPU -- referenced ids repeat via `e` tags). An app-level pool is the only variant cheaper on CPU, but it holds strong references where ART's table is weak. Co-Authored-By: Claude Opus 5 (1M context) --- .../prodbench/InternTradeoffBenchmark.kt | 170 ++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/InternTradeoffBenchmark.kt diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/InternTradeoffBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/InternTradeoffBenchmark.kt new file mode 100644 index 0000000000..9095a17bcf --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/InternTradeoffBenchmark.kt @@ -0,0 +1,170 @@ +/* + * 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.prodbench + +import java.util.concurrent.ConcurrentHashMap +import kotlin.test.Test + +/** + * `String.intern()` is applied to every event `id`, `pubKey` (EventDeserializer) and + * every tag value (TagArrayDeserializer). An on-device profile of a release build + * during ingest put `art::InternTable::InternWeak` at **3.5% of the ingest workers' + * CPU** — the second largest symbol after memcpy. + * + * Interning is a MEMORY optimisation, not a speed one: Nostr repeats strings heavily, + * so collapsing duplicates to one instance is why it is there. Removing it could + * therefore cost more memory than the CPU it saves. This measures both axes, split by + * how duplicated each field actually is, because the answer differs per field: + * + * - tag names (`p`, `e`, `a`) — a handful of values repeated on every event + * - relay URLs — tens of values repeated across everything + * - pubkeys — measured on device: 1,880 distinct for 11,338 notes + * - event ids — one per event, but every `e` tag that references + * that event repeats it, so they dedup too + * + * Cardinalities come from the device probe cited above, scaled to a corpus big enough + * for the heap delta to clear GC noise. + * + * Results (JVM, 60k events / 840k strings): + * + * ``` + * retained parse CPU + * no interning 69.1 MB 146 ms + * intern everything 17.8 MB 354 ms + * app-level pool 23.3 MB 192 ms + * intern all but ids 35.8 MB 257 ms + * ``` + * + * So interning is a ~4x memory win and worth its CPU. Skipping the hex fields is not a + * shortcut — it gives up half the memory for half the CPU, because those hex strings + * repeat too. An app-level pool is the only variant that beats interning on CPU, but it + * holds *strong* references where ART's intern table is weak, so it would never release + * a string whose event was pruned; the 5.5 MB it loses to interning is its own node + * headers. Not worth trading a self-clearing table for an unbounded one to reclaim ~2.7% + * of ingest CPU on a device that is already memory-bound. + * + * Measures the interning primitive on realistic string distributions — not the full + * Jackson parse path — so it isolates the tradeoff without needing a second + * deserializer wired in. Note this is HotSpot's `intern()`; ART's `InternWeak` is a + * different implementation, so treat the ordering as transferable and the magnitudes as + * indicative. + */ +class InternTradeoffBenchmark { + companion object { + const val EVENTS = 60_000 + const val AUTHORS = 10_000 // ~6 events per author, as measured on device + const val RELAYS = 60 + val TAG_NAMES = listOf("p", "e", "a", "t", "r", "alt") + + fun hex(seed: Int): String = "%064x".format(seed) + + fun used(): Long { + val rt = Runtime.getRuntime() + return rt.totalMemory() - rt.freeMemory() + } + + /** Retained bytes of whatever [build] returns, with GC settled either side. */ + fun retained(build: () -> Any): Pair { + repeat(3) { + System.gc() + Thread.sleep(60) + } + val before = used() + val held = build() + repeat(3) { + System.gc() + Thread.sleep(60) + } + val after = used() + return (after - before) to held + } + } + + /** The strings one event contributes, as fresh instances (as if just parsed off the wire). */ + private fun eventStrings(i: Int): List = + buildList { + add(String(hex(i).toCharArray())) // id — unique per event + add(String(hex(i % AUTHORS).toCharArray())) // pubKey — repeats + repeat(4) { t -> + add(String(TAG_NAMES[(i + t) % TAG_NAMES.size].toCharArray())) // tag name — tiny set + add(String("wss://relay${(i + t) % RELAYS}.example.com/".toCharArray())) // relay url + add(String(hex((i + t) * 31).toCharArray())) // tag value: a referenced id + } + } + + @Test + fun internCostAndBenefit() { + var sink: Any? = null + + println("\n=== corpus: $EVENTS events, $AUTHORS authors, $RELAYS relays ===") + + // ---- memory ---- + val (memNone, holdA) = retained { (0 until EVENTS).flatMap { eventStrings(it) } } + sink = holdA + val (memAll, holdB) = retained { (0 until EVENTS).flatMap { eventStrings(it).map { s -> s.intern() } } } + sink = holdB + // intern everything EXCEPT the two unique-hex classes (event id, tag id refs) + val (memSel, holdC) = + retained { + (0 until EVENTS).flatMap { i -> + eventStrings(i).mapIndexed { idx, s -> + // index 0 is the id; every 3rd tag slot is a referenced id + if (idx == 0 || (idx > 1 && (idx - 2) % 3 == 2)) s else s.intern() + } + } + } + sink = holdC + + println("\nRETAINED HEAP") + println(" no interning %6.1f MB".format(memNone / 1048576.0)) + println(" intern everything %6.1f MB (%+.1f MB vs none)".format(memAll / 1048576.0, (memAll - memNone) / 1048576.0)) + println(" intern all but ids %6.1f MB (%+.1f MB vs none)".format(memSel / 1048576.0, (memSel - memNone) / 1048576.0)) + + // ---- cpu ---- + fun timeIt( + label: String, + op: (String, Int) -> String, + ) { + repeat(2) { i -> (0 until 5_000).forEach { eventStrings(it).forEachIndexed { idx, s -> op(s, idx) } } } + val t0 = System.nanoTime() + var n = 0 + (0 until EVENTS).forEach { i -> eventStrings(i).forEachIndexed { idx, s -> if (op(s, idx) !== s) n++ } } + val ms = (System.nanoTime() - t0) / 1_000_000 + println(" %-20s %5d ms (%d strings)".format(label, ms, EVENTS * 14)) + } + + // An app-level dedup table gives the same collapsing without going through + // ART's global InternTable, which is what showed up in the device profile. + val pool = ConcurrentHashMap(1 shl 16) + val (memPool, holdD) = retained { (0 until EVENTS).flatMap { eventStrings(it).map { s -> pool.putIfAbsent(s, s) ?: s } } } + sink = holdD + println(" app-level pool %6.1f MB (%+.1f MB vs none, pool holds %d)".format(memPool / 1048576.0, (memPool - memNone) / 1048576.0, pool.size)) + + println("\nPARSE-SIDE CPU (build + optionally intern)") + timeIt("no interning") { s, _ -> s } + timeIt("intern everything") { s, _ -> s.intern() } + timeIt("intern all but ids") { s, idx -> if (idx == 0 || (idx > 1 && (idx - 2) % 3 == 2)) s else s.intern() } + val pool2 = ConcurrentHashMap(1 shl 16) + timeIt("app-level pool") { s, _ -> pool2.putIfAbsent(s, s) ?: s } + + check(sink != null) + } +} From e17f63428a0db7d4f66db9419f7c2ff578dbea10 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Thu, 13 Aug 2026 21:27:08 -0400 Subject: [PATCH 2/8] test: profile ingest allocation rate against the real startup capture Every prior memory measurement here looked at retained heap, which is blind to garbage that never survives a GC. This measures allocated bytes per stage with ThreadMXBean.getThreadAllocatedBytes, replaying the checked-in capture of an account cold start. Result: parse allocates 5.5x the wire bytes and retains 0.87x -- 4.8 KB per event, 5.3 bytes of garbage per byte kept. Useful as a regression guard, but it also rules the parse path out as the cause of the ingest CPU saturation: device-wide allocation during ingest measures ~28-44 MB/s, which is not a rate that stresses a GC. Co-Authored-By: Claude Opus 5 (1M context) --- .../prodbench/IngestAllocationBenchmark.kt | 205 ++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/IngestAllocationBenchmark.kt diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/IngestAllocationBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/IngestAllocationBenchmark.kt new file mode 100644 index 0000000000..0369af8a30 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/IngestAllocationBenchmark.kt @@ -0,0 +1,205 @@ +/* + * 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.prodbench + +import com.fasterxml.jackson.core.JsonToken +import com.sun.management.ThreadMXBean +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.crypto.verify +import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip10Notes.content.findHashtags +import com.vitorpamplona.quartz.nip10Notes.content.findIndexTagsWithPeople +import com.vitorpamplona.quartz.nip19Bech32.Nip19Parser +import java.lang.management.ManagementFactory +import java.util.zip.GZIPInputStream +import kotlin.test.Test +import kotlin.test.assertTrue + +/** + * Allocation-RATE profile of the per-event ingest path. + * + * Every previous memory measurement in this codebase looked at *retained* heap (heap + * dumps, histograms, the intern tradeoff). That answers "what is alive", which is the + * wrong question for the observed problem: on an SM-T220, ingest saturates all 8 cores + * and the driver is GC read barriers charged to the mutator threads, not any single + * consume stage. Read-barrier cost tracks the *allocation rate* — bytes created and + * discarded per second — which retained-heap analysis is blind to. + * + * `com.sun.management.ThreadMXBean.getThreadAllocatedBytes` reports exactly that, per + * thread, with no sampling error. It counts TLAB bumps, so it sees garbage that never + * survives a single GC — precisely the allocations a heap dump cannot show. + * + * The headline number to look for is **bytes allocated per byte of wire JSON**. An event + * arrives as N bytes of text; if turning it into an `Event` costs 10N, the ingest path + * is a garbage pump and bounding concurrency will not help, because every added thread + * multiplies the same waste. + * + * Runs on the JVM, so the absolute numbers are HotSpot's. Allocation *counts* come from + * the same bytecode ART runs and the object layouts are close, so the per-stage ranking + * and the ratios transfer; treat the absolute MB/s as indicative and confirm any fix + * on device. + */ +class IngestAllocationBenchmark { + companion object { + /** The real multi-relay capture of an account cold start, checked into quartz. */ + const val CORPUS = "nostr_vitor_startup_data.json.gz" + const val EVENTS = 20_000 + + private val threads = ManagementFactory.getThreadMXBean() as ThreadMXBean + + /** Anything the measured loops produce, parked here so JIT cannot elide them. */ + var keep: Any? = null + + /** + * Bytes allocated by running [block] over [reps] iterations, minus the loop's own + * overhead. + * + * Warmed hard on purpose: escape analysis only kicks in after C2 compiles the + * method, and a cold measurement reports allocations the JIT would have scalarised + * away — i.e. it over-reports exactly the transient garbage this is looking for. + */ + fun allocBytes( + reps: Int, + warmup: Int = reps, + block: (Int) -> Any?, + ): Long { + var sink: Any? = null + for (i in 0 until warmup) sink = block(i % reps) + keep = sink + + val id = Thread.currentThread().threadId() + val before = threads.getThreadAllocatedBytes(id) + for (i in 0 until reps) sink = block(i) + val after = threads.getThreadAllocatedBytes(id) + keep = sink + + // the empty loop itself allocates nothing measurable, but the accessor does + val idle = threads.getThreadAllocatedBytes(id) + val overhead = idle - after + return (after - before) - overhead + } + + fun used(): Long { + val rt = Runtime.getRuntime() + return rt.totalMemory() - rt.freeMemory() + } + + fun retained(build: () -> Any): Pair { + repeat(3) { + System.gc() + Thread.sleep(60) + } + val before = used() + val held = build() + repeat(3) { + System.gc() + Thread.sleep(60) + } + return (used() - before) to held + } + } + + /** + * The real capture of Vitor's account startup — the exact traffic this profile is + * about — streamed rather than slurped, because the file is 247 MB decompressed and + * only the first [EVENTS] are needed. + * + * Real signatures, so `verify()` does its real work, and a real kind/size/tag mix + * rather than a guess at one. + */ + private fun corpus(): List { + val stream = + javaClass.classLoader?.getResourceAsStream(CORPUS) + ?: throw IllegalStateException("corpus $CORPUS missing from the test classpath") + + val out = ArrayList(EVENTS) + val seen = HashSet(EVENTS * 2) + GZIPInputStream(stream).use { gz -> + val parser = JacksonMapper.mapper.factory.createParser(gz) + check(parser.nextToken() == JsonToken.START_ARRAY) { "$CORPUS is not a JSON array" } + while (out.size < EVENTS && parser.nextToken() == JsonToken.START_OBJECT) { + val event: Event = JacksonMapper.mapper.readValue(parser, JacksonMapper.eventTypeInstance) + // the capture spans many relays, so the same event arrives several times + if (seen.add(event.id)) out.add(event) + } + } + return out + } + + @Test + fun allocationPerIngestStage() { + val events = corpus() + val eventJson = events.map { it.toJson() } + val frames = eventJson.mapIndexed { i, j -> "[\"EVENT\",\"sub$i\",$j]" } + val frameBytes = frames.map { it.toByteArray(Charsets.UTF_8) } + + val count = events.size + val wireBytes = frames.sumOf { it.length.toLong() } + val n = count.toLong() + + assertTrue(count > EVENTS / 2, "corpus yielded only $count events") + assertTrue(eventJson.all { it.length > 50 }, "corpus produced empty events") + + println("\n=== corpus: $count events, ${wireBytes / 1024} KB of wire JSON, mean ${wireBytes / n} B/event ===") + + // ---- what surviving the parse actually costs to keep ---- + val (retainedEvents, held) = retained { frames.map { Message.fromJson(it) } } + keep = held + + // ---- per stage ---- + val decode = allocBytes(count) { i -> String(frameBytes[i], Charsets.UTF_8) } + val msgParse = allocBytes(count) { i -> Message.fromJson(frames[i]) } + val evtParse = allocBytes(count) { i -> JacksonMapper.fromJson(eventJson[i]) } + val verify = allocBytes(count) { i -> events[i].verify() } + val scans = + allocBytes(count) { i -> + val e = events[i] + findHashtags(e.content) + findIndexTagsWithPeople(e.content, e.tags) + Nip19Parser.parseAll(e.content) + } + + fun row( + label: String, + bytes: Long, + ) = println( + " %-22s %8.1f MB total %8d B/event %5.2fx wire".format( + label, + bytes / 1048576.0, + bytes / n, + bytes.toDouble() / wireBytes, + ), + ) + + println("\nALLOCATED (transient + surviving)") + row("utf8 decode", decode) + row("frame -> Message", msgParse) + row("event json -> Event", evtParse) + row("verify()", verify) + row("content scans", scans) + row("TOTAL decode+parse+verify", decode + msgParse + verify) + + println("\nRETAINED after parse") + println(" parsed events %8.1f MB total %8d B/event %5.2fx wire".format(retainedEvents / 1048576.0, retainedEvents / n, retainedEvents.toDouble() / wireBytes)) + println(" garbage per kept byte %8.1fx".format((decode + msgParse + verify - retainedEvents).toDouble() / retainedEvents)) + } +} From 054c665696acb7d17130b4e4910ba0b4b8914f8e Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Fri, 14 Aug 2026 00:04:46 -0400 Subject: [PATCH 3/8] test: pin that live websockets hold their OkHttp dispatcher slot forever The app dials ~190 relays in one burst from RelayPool.connect() and spikes to ~640 threads, so capping the relay client's Dispatcher.maxRequests looks like a one-line throttle. It is not. Against a real relay, maxRequests=4 with 20 dials opens exactly 4; the other 16 queue forever and never fail, so nothing surfaces the stall. Setting maxRequests=N would cap the app at N relays permanently. Pins the behaviour so the knob is not reached for again. Throttling has to happen above OkHttp, in RelayPool, where a settled dial can release its permit. Co-Authored-By: Claude Opus 5 (1M context) --- .../geode/OkHttpDispatcherWebSocketCapTest.kt | 143 ++++++++++++++++++ 1 file changed, 143 insertions(+) create mode 100644 geode/src/test/kotlin/com/vitorpamplona/geode/OkHttpDispatcherWebSocketCapTest.kt diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/OkHttpDispatcherWebSocketCapTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/OkHttpDispatcherWebSocketCapTest.kt new file mode 100644 index 0000000000..98d798d495 --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/OkHttpDispatcherWebSocketCapTest.kt @@ -0,0 +1,143 @@ +/* + * 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.geode + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import okhttp3.Dispatcher +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse + +/** + * Does OkHttp's `Dispatcher.maxRequests` cap *concurrent handshakes*, or does a live + * WebSocket hold its dispatcher slot for the socket's whole lifetime? + * + * This decides whether lowering `maxRequests` on the relay `OkHttpClient` + * (`OkHttpClientFactoryForRelays`, currently 1024) is a safe way to throttle the + * cold-start dial burst — or a change that would silently cap the app at N relays + * forever. The app connects to ~190 relays at once and spikes to ~640 threads, so + * this knob looks tempting; the difference between the two behaviours is the + * difference between a one-line fix and an outage. + * + * The answer is not something to take from memory: `RealWebSocket.connect()` goes + * through `newCall().enqueue()`, so the upgrade *is* dispatched, and whether the + * call is retired when the 101 response arrives is an implementation detail that + * has changed across OkHttp versions. + */ +class OkHttpDispatcherWebSocketCapTest { + private lateinit var relay: RelayEngine + private lateinit var server: KtorRelay + + companion object { + const val CAP = 4 + const val SOCKETS = 20 + } + + @BeforeTest + fun setup() { + relay = RelayEngine(url = "ws://127.0.0.1:7771/".normalizeRelayUrl()) + server = KtorRelay(relay, host = "127.0.0.1", port = 0).start() + } + + @AfterTest + fun teardown() { + server.stop() + relay.close() + } + + @Test + fun liveWebSocketsHoldTheirDispatcherSlotForever() { + val client = + OkHttpClient + .Builder() + .dispatcher( + Dispatcher().apply { + maxRequests = CAP + // all sockets share one host here, so this would bind too + maxRequestsPerHost = CAP + }, + ).build() + + val opened = AtomicInteger() + val failed = AtomicInteger() + val latch = CountDownLatch(SOCKETS) + val sockets = mutableListOf() + + val url = server.url.replace("ws://", "http://") + repeat(SOCKETS) { + sockets += + client.newWebSocket( + Request.Builder().url(url).build(), + object : WebSocketListener() { + override fun onOpen( + webSocket: WebSocket, + response: Response, + ) { + opened.incrementAndGet() + latch.countDown() + } + + override fun onFailure( + webSocket: WebSocket, + t: Throwable, + response: Response?, + ) { + failed.incrementAndGet() + latch.countDown() + } + }, + ) + } + + val settled = latch.await(20, TimeUnit.SECONDS) + // snapshot BEFORE cancelling: cancel() drives onFailure on every queued dial + val openedCount = opened.get() + val failedCount = failed.get() + println( + "\nmaxRequests=$CAP, dialed $SOCKETS websockets -> " + + "opened=$openedCount failed=$failedCount allSettled=$settled", + ) + + sockets.forEach { it.cancel() } + client.dispatcher.executorService.shutdown() + + // ANSWER: a live WebSocket holds its dispatcher slot for the socket's whole + // lifetime. Exactly CAP open; the remaining SOCKETS-CAP sit queued forever and + // never fail, so nothing surfaces the stall. + // + // Therefore `maxRequests` must NEVER be used to throttle the relay dial burst: + // setting it to N would cap the app at N relays permanently. Throttling has to + // happen above OkHttp, in RelayPool, where a settled dial can release its permit. + assertEquals(CAP, openedCount, "only maxRequests websockets ever open") + assertEquals(0, failedCount, "the queued dials never fail - they just hang") + assertFalse(settled, "the remaining dials never settle") + } +} From f2eaa682e45d084d974cfb47d07a243325140060 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Fri, 14 Aug 2026 00:19:20 -0400 Subject: [PATCH 4/8] test: measure the dedup decoder cache at its production capacity Connecting to every relay and pulling everything at once is the design, and the 82% duplicate frames it produces are the price of that redundancy. What is not wanted is paying a full JSON parse for each copy -- which is what CachingEventDecoder exists to avoid. DedupDecodeBenchmark proves the mechanism, but at capacity = UNIQUE * 2 (40,000) and with duplicates spaced 20,000 frames apart. Production ships the default capacity of 2048. Measured against the real multi-relay capture in recorded order, that catches only 58.3% of duplicates; 8192 catches 80.5% and halves decode time (354ms -> 169ms over 150k frames). Uses reuse distance so one pass yields the hit rate for every candidate capacity. Co-Authored-By: Claude Opus 5 (1M context) --- .../relay/prodbench/DedupCapacityBenchmark.kt | 151 ++++++++++++++++++ 1 file changed, 151 insertions(+) create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt new file mode 100644 index 0000000000..d61e46432e --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt @@ -0,0 +1,151 @@ +/* + * 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.relay.commands.toClient.CachingEventDecoder +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MessageDecoder +import java.util.zip.GZIPInputStream +import kotlin.test.Test +import kotlin.test.assertTrue + +/** + * Is the production [CachingEventDecoder] capacity big enough for a cold start? + * + * The app connects to every relay it knows and pulls everything at once, by design — + * that redundancy is the point, and throttling it would defeat the product. What is + * NOT wanted is paying a full JSON parse for the same event once per relay that + * serves it. [CachingEventDecoder] exists precisely to avoid that, and it is wired in + * (`AppModules.kt`, `NostrClient(websocketBuilder, applicationIOScope, + * CachingEventDecoder())`) — at its **default capacity of 2048**. + * + * `DedupDecodeBenchmark` proves the mechanism, but with `capacity = UNIQUE * 2` + * (40,000) and a frame order that spaces each duplicate 20,000 frames apart. Both + * choices flatter the cache. This measures the setting that actually ships, against + * the real multi-relay capture (248,241 frames / 31,161 unique -- ~8x duplication) in + * the order it was recorded. + * + * The decisive statistic is the **reuse distance**: how many frames pass between two + * deliveries of the same event. A capacity only catches duplicates whose reuse + * distance falls inside it, so the distance histogram gives the hit rate for every + * candidate capacity at once, without re-running the decoder for each. + */ +class DedupCapacityBenchmark { + companion object { + const val CORPUS = "nostr_vitor_startup_data.json.gz" + const val FRAMES = 150_000 + val CAPACITIES = listOf(2_048, 8_192, 32_768, 131_072) + } + + /** Raw event JSON in recorded order, wrapped as EVENT frames. */ + private fun frames(): List { + val stream = + javaClass.classLoader?.getResourceAsStream(CORPUS) + ?: throw IllegalStateException("corpus $CORPUS missing from the test classpath") + + val out = ArrayList(FRAMES) + GZIPInputStream(stream).bufferedReader().use { reader -> + // one giant single-line JSON array; walk it and slice out each top-level object + val buf = CharArray(1 shl 16) + val sb = StringBuilder() + var depth = 0 + var inStr = false + var esc = false + var n = reader.read(buf) + while (n > 0 && out.size < FRAMES) { + for (i in 0 until n) { + val c = buf[i] + if (depth > 0) sb.append(c) + when { + esc -> esc = false + c == '\\' && inStr -> esc = true + c == '"' -> inStr = !inStr + inStr -> {} + c == '{' -> { + if (depth == 0) sb.append(c) + depth++ + } + c == '}' -> { + depth-- + if (depth == 0) { + out.add("[\"EVENT\",\"s\",$sb]") + sb.setLength(0) + if (out.size >= FRAMES) break + } + } + } + } + if (out.size >= FRAMES) break + n = reader.read(buf) + } + } + return out + } + + private fun idOf(frame: String): String? { + val k = frame.indexOf("\"id\":\"") + if (k < 0 || k + 6 + 64 > frame.length) return null + return frame.substring(k + 6, k + 6 + 64) + } + + @Test + fun productionCapacityAgainstRealArrivalOrder() { + val frames = frames() + assertTrue(frames.size > FRAMES / 2, "corpus yielded only ${frames.size} frames") + + // ---- reuse-distance histogram: hit rate for ANY capacity, in one pass ---- + val lastSeen = HashMap(frames.size) + val distances = ArrayList(frames.size) + var unique = 0 + frames.forEachIndexed { i, f -> + val id = idOf(f) ?: return@forEachIndexed + val prev = lastSeen.put(id, i) + if (prev == null) unique++ else distances.add(i - prev) + } + val dupes = distances.size + println("\n=== real capture: ${frames.size} frames, $unique unique, $dupes duplicates (${dupes * 100 / frames.size}%) ===") + + println("\nREUSE DISTANCE -> share of duplicates a cache of that size can catch") + CAPACITIES.forEach { cap -> + val caught = distances.count { it <= cap } + val label = if (cap == 2_048) " <- PRODUCTION" else "" + println(" capacity %7d catches %5.1f%% of duplicates%s".format(cap, caught * 100.0 / dupes, label)) + } + val med = distances.sorted()[dupes / 2] + println(" median reuse distance: $med frames") + + // ---- confirm with the real decoder at the shipping capacity ---- + println("\nACTUAL DECODER (parse time over ${frames.size} frames)") + (listOf(0) + CAPACITIES).forEach { cap -> + val decoder = if (cap == 0) MessageDecoder.Default else CachingEventDecoder(capacity = cap) + repeat(2_000) { decoder.decode(frames[it]) } // warm + val t0 = System.nanoTime() + frames.forEach { decoder.decode(it) } + val ms = (System.nanoTime() - t0) / 1_000_000 + val reuse = + if (decoder is CachingEventDecoder) { + " reused=${decoder.reusedCount} parsed=${decoder.parsedCount}" + } else { + "" + } + println(" capacity %7s %6d ms%s".format(if (cap == 0) "none" else "$cap", ms, reuse)) + } + } +} From be330df1bc6a675ed47f6dabee9f49def5423a94 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Fri, 14 Aug 2026 00:31:27 -0400 Subject: [PATCH 5/8] perf: size the dedup decoder cache from measured reuse distance (2048 -> 8192) Connecting to every relay and pulling everything at once is the design; the duplicate frames it produces are the price of that redundancy. Paying a full JSON parse for each copy is not. CachingEventDecoder already avoids that, but shipped at capacity 2048. Against the recorded multi-relay startup capture (150k frames, 82% duplicates, median reuse distance 1,430) that caught only 58.3% of duplicates. 8192 catches 80.5% and halves offline decode time (354ms -> 169ms); 32768 would reach 96.5% but for 4x the retained entries. On-device A/B (emulator, cold start, ~15-18k frames): the share of frames needing a full parse fell from 55.2% at 2048 to 38.7-44.2% at 8192. Process CPU was NOT a usable signal at this sample size -- cold-start variance swamped it. The gain scales with duplication, so accounts with many relays gain most. No behaviour change: same relays, same aggression, same dispatch semantics. Co-Authored-By: Claude Opus 5 (1M context) --- .../commands/toClient/CachingEventDecoder.kt | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) 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 index b3517a3f5e..6290f3dc3c 100644 --- 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 @@ -71,7 +71,22 @@ import kotlin.concurrent.atomics.incrementAndFetch */ @OptIn(ExperimentalAtomicApi::class) class CachingEventDecoder( - private val capacity: Int = 2048, + /** + * Ids kept per generation, so up to `2 * capacity` are addressable. + * + * Sized from the measured *reuse distance* — how many frames separate two + * deliveries of the same event — because a duplicate is only caught when its + * reuse distance still fits in the cache. Against the recorded multi-relay + * startup capture (150k frames, 82% duplicates, median reuse distance 1,430), + * the old 2048 caught just 58.3% of duplicates; 8192 catches 80.5% and halves + * decode time (354ms -> 169ms). Beyond 8192 the curve flattens (32768 buys + * 96.5% for 4x the entries), so this trades the last 16% for memory the + * client would otherwise hold on top of `LocalCache`. + * + * See `DedupCapacityBenchmark`. Re-measure it before changing this: the right + * value follows the duplication of real traffic, not intuition. + */ + private val capacity: Int = 8192, private val fullParser: MessageDecoder = MessageDecoder.Default, ) : MessageDecoder { @Volatile private var live = ConcurrentHashCache() From 1b8e264706bdf582fb0fce8297875ba11bc38375 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Fri, 14 Aug 2026 01:50:35 -0400 Subject: [PATCH 6/8] fix: release the dedup decoder cache once relay traffic goes quiet CachingEventDecoder bounds what it holds with `capacity`, which caps the cost during a burst but never ends it: the two generations only rotate inside an insert, so a client that stops receiving frames pins up to `2 * capacity` events for the rest of the process's life. Their dedup value decays within seconds -- a duplicate is the same event arriving from another relay -- but the memory did not decay at all. Raising capacity to 8192 doubled that cost (4.7MB -> 9.4MB retained, worst case), which is what surfaced it. Adds MessageDecoder.trimIfIdle(idleMillis, nowMillis), a no-op for stateless decoders, and drives it two ways from NostrClient: - disconnect() trims eagerly, so backgrounding releases immediately instead of leaving a timer to do it later; - while active, a 30s tick releases anything idle for 60s. The trim loop suspends on isActiveFlow exactly like keepAliveJob, so no timer fires while the client is down. Idleness counts cache HITS, not just inserts: a stream of pure duplicates inserts nothing yet is precisely when the cache is earning its keep, and keying off inserts would trim it out from under that traffic. Racy by the same design as rotation -- clearing concurrently with an insert can only lose an id, and a lost id costs a re-parse, never a wrong message. Tests are in commonTest with an injected clock (no target needs a real one) and were mutation-checked: dropping the hit-path refresh fails 1, ignoring the idle threshold fails 2, never releasing fails 4. Co-Authored-By: Claude Opus 5 (1M context) --- .../nip01Core/relay/client/NostrClient.kt | 30 +++- .../commands/toClient/CachingEventDecoder.kt | 37 +++++ .../relay/commands/toClient/MessageDecoder.kt | 17 ++ .../toClient/CachingEventDecoderTrimTest.kt | 150 ++++++++++++++++++ .../relay/prodbench/DedupCapacityBenchmark.kt | 55 +++++++ 5 files changed, 288 insertions(+), 1 deletion(-) create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt 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 bc281cfd3e..99d517a9f1 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 @@ -91,7 +91,7 @@ class NostrClient( * another subscription or another relay reuses the already-parsed Event; * dispatch semantics are unchanged). */ - decoder: MessageDecoder = MessageDecoder.Default, + private val decoder: MessageDecoder = MessageDecoder.Default, ) : INostrClient, RelayConnectionListener, AutoCloseable { @@ -156,6 +156,11 @@ class NostrClient( override fun disconnect() { isActiveFlow.value = false relayPool.disconnect() + // The host is putting us down (typically the app backgrounding). Whatever the + // decoder cached has no dedup value left across that gap, but it would keep + // costing memory for the whole background stretch, so release it now rather + // than leaving a timer running to do it later. + decoder.trimIfIdle(idleMillis = 0) } override fun isActive() = isActiveFlow.value @@ -208,8 +213,31 @@ class NostrClient( } } + /** + * Releases the decoder's cache after a quiet stretch. + * + * A caching decoder bounds what it holds by capacity, which caps the cost during + * a burst but never ends it: its generations only rotate inside an insert, so a + * client that goes quiet pins every cached event indefinitely. Dedup value decays + * within seconds (duplicates are the same event arriving from other relays), so + * anything still held after [DECODER_IDLE_MS] is pure cost. + * + * Suspends on [isActiveFlow] like [keepAliveJob], so no timer fires while the + * client is down — [disconnect] already trims eagerly on the way there. + */ + private val decoderTrimJob = + scope.launch { + while (true) { + isActiveFlow.first { it } + delay(DECODER_TRIM_INTERVAL_MS) + decoder.trimIfIdle(DECODER_IDLE_MS) + } + } + companion object { private const val KEEP_ALIVE_INTERVAL_MS = 60_000L + private const val DECODER_TRIM_INTERVAL_MS = 30_000L + private const val DECODER_IDLE_MS = 60_000L } override fun reconnect( 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 index 6290f3dc3c..c4bb19ee97 100644 --- 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 @@ -22,6 +22,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.commands.toClient import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.cache.ConcurrentHashCache import kotlin.concurrent.Volatile import kotlin.concurrent.atomics.AtomicLong @@ -93,6 +94,16 @@ class CachingEventDecoder( @Volatile private var previous = ConcurrentHashCache() + /** + * When an EVENT frame was last seen, for [trimIfIdle]. + * + * Written on every EVENT frame — hit or miss — so a cache that is actively + * serving duplicates counts as busy and is never trimmed out from under the + * traffic it is helping. One relaxed volatile store per frame; at observed + * rates (~400 frames/s) that is not measurable. + */ + @Volatile private var lastActivityAt: Long = TimeUtils.nowMillis() + // Observability + benchmark hooks. private val parsed = AtomicLong(0) private val reused = AtomicLong(0) @@ -106,6 +117,7 @@ class CachingEventDecoder( val cached = live.get(scanned.eventId) ?: previous.get(scanned.eventId) if (cached != null) { reused.incrementAndFetch() + lastActivityAt = TimeUtils.nowMillis() return EventMessage(scanned.subId, cached) } } @@ -113,6 +125,7 @@ class CachingEventDecoder( val msg = fullParser.decode(text) if (msg is EventMessage) { parsed.incrementAndFetch() + lastActivityAt = TimeUtils.nowMillis() if (live.size() >= capacity) { // Benign-race rotation: see class kdoc. previous = live @@ -123,6 +136,30 @@ class CachingEventDecoder( return msg } + /** + * Drops both generations once no EVENT frame has arrived for [idleMillis]. + * + * [capacity] bounds what the cache costs *during* a burst; this bounds how long + * it costs anything *after* one. Without it the maps only ever rotate inside an + * insert, so a client that goes quiet pins up to `2 * capacity` events for the + * rest of the process's life — memory whose dedup value has already decayed to + * zero. + * + * Racy by the same design as rotation (see class kdoc): clearing concurrently + * with an insert can only lose an id, and losing an id costs a re-parse, never + * a wrong message. [nowMillis] is injectable so tests need no clock. + */ + override fun trimIfIdle( + idleMillis: Long, + nowMillis: Long, + ): Boolean { + if (nowMillis - lastActivityAt < idleMillis) return false + if (live.size() == 0 && previous.size() == 0) return false + previous = ConcurrentHashCache() + live = ConcurrentHashCache() + return true + } + private class ScannedEvent( val subId: String, val eventId: HexKey, 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 index 4754ccf62f..3373977580 100644 --- 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 @@ -21,6 +21,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.commands.toClient import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.utils.TimeUtils /** * Turns one raw relay frame into a [Message]. The strategy the relay client @@ -33,6 +34,22 @@ import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper fun interface MessageDecoder { fun decode(text: String): Message + /** + * Releases whatever the decoder is holding if it has seen no frames for + * [idleMillis]. Returns true when something was released. + * + * A caching decoder's value decays with time — a duplicate that has not + * arrived in a minute is not going to — but its memory does not, so the + * owner ticks this to let a quiet decoder drop what it cached. Stateless + * decoders keep the no-op default. + * + * Never affects correctness: releasing a cache can only cost a re-parse. + */ + fun trimIfIdle( + idleMillis: Long, + nowMillis: Long = TimeUtils.nowMillis(), + ): Boolean = false + 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/CachingEventDecoderTrimTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt new file mode 100644 index 0000000000..cda94b59a5 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt @@ -0,0 +1,150 @@ +/* + * 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.utils.TimeUtils +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +/** + * [CachingEventDecoder.trimIfIdle] bounds how long the cache costs memory, which + * [CachingEventDecoder.capacity] does not: the generations only rotate inside an + * insert, so without a trim a client that goes quiet pins every cached event for + * the rest of the process's life. + * + * The rules that matter here are "release when genuinely idle", "do NOT release + * out from under live traffic" — including traffic that is *only* cache hits, + * which is exactly when the cache is earning its keep — and "releasing never + * changes what a decode returns, only whether it re-parsed". + * + * In `commonTest`: this is `commonMain` logic with no platform surface, and the + * clock is injected so no target needs a real one. + */ +class CachingEventDecoderTrimTest { + private fun hexId(seed: Int) = seed.toString(16).padStart(64, '0') + + private fun event(seed: Int) = + Event( + id = hexId(seed), + pubKey = hexId(seed + 500_000), + createdAt = seed.toLong(), + kind = 1, + tags = arrayOf(arrayOf("t", "trim")), + content = "trim test $seed", + sig = "f".repeat(128), + ) + + private fun frame( + seed: Int, + subId: String = "s", + ) = """["EVENT","$subId",${event(seed).toJson()}]""" + + private val minute = 60_000L + + @Test + fun releasesTheCacheOnceIdle() { + val decoder = CachingEventDecoder(capacity = 1_000) + (1..50).forEach { decoder.decode(frame(it)) } + assertEquals(50, decoder.parsedCount.toInt()) + + // read the clock AFTER the decodes: they set the idle clock as they run, so a + // cutoff measured from before them is only a full minute later while the + // parsing itself takes under a millisecond + val quietSince = TimeUtils.nowMillis() + assertTrue(decoder.trimIfIdle(minute, quietSince + minute), "should trim after a quiet minute") + + // the ids are gone, so a repeat of a previously cached frame parses again + decoder.decode(frame(1)) + assertEquals(51, decoder.parsedCount.toInt()) + assertEquals(0, decoder.reusedCount.toInt()) + } + + @Test + fun keepsTheCacheWhileTrafficIsFlowing() { + val decoder = CachingEventDecoder(capacity = 1_000) + val t0 = TimeUtils.nowMillis() + (1..50).forEach { decoder.decode(frame(it)) } + + assertFalse(decoder.trimIfIdle(minute, t0 + 1_000), "1s of quiet is not idle") + + decoder.decode(frame(1)) + assertEquals(50, decoder.parsedCount.toInt(), "cached ids must survive") + assertEquals(1, decoder.reusedCount.toInt()) + } + + @Test + fun cacheHitsCountAsActivity() { + // A stream of pure duplicates inserts nothing, but it is precisely when the + // cache is most valuable. Keying idleness off inserts alone would trim it away + // under the traffic it is serving. + val decoder = CachingEventDecoder(capacity = 1_000) + decoder.decode(frame(1)) + + val muchLater = TimeUtils.nowMillis() + 10 * minute + repeat(5) { decoder.decode(frame(1, subId = "sub$it")) } + + assertEquals(5, decoder.reusedCount.toInt()) + assertFalse( + decoder.trimIfIdle(minute, TimeUtils.nowMillis() + 1_000), + "hits must refresh the idle clock", + ) + // ...but a genuinely quiet stretch after those hits still trims + assertTrue(decoder.trimIfIdle(minute, muchLater)) + } + + @Test + fun trimmingAnEmptyCacheDoesNothing() { + val decoder = CachingEventDecoder(capacity = 1_000) + assertFalse(decoder.trimIfIdle(0, TimeUtils.nowMillis() + minute), "nothing to release") + } + + @Test + fun zeroIdleReleasesImmediately() { + // what NostrClient.disconnect() uses when the host backgrounds the app + val decoder = CachingEventDecoder(capacity = 1_000) + (1..10).forEach { decoder.decode(frame(it)) } + assertTrue(decoder.trimIfIdle(idleMillis = 0, nowMillis = TimeUtils.nowMillis())) + } + + @Test + fun aTrimNeverChangesWhatADecodeReturns() { + val decoder = CachingEventDecoder(capacity = 1_000) + val before = decoder.decode(frame(7, subId = "a")) as EventMessage + decoder.trimIfIdle(0, TimeUtils.nowMillis()) + val after = decoder.decode(frame(7, subId = "a")) as EventMessage + + assertEquals(before.subId, after.subId) + assertEquals(before.event.id, after.event.id) + assertEquals(before.event.pubKey, after.event.pubKey) + assertEquals(before.event.content, after.event.content) + assertEquals(before.event.createdAt, after.event.createdAt) + // only the parse count differs: the trim cost one re-parse, nothing else + assertEquals(2, decoder.parsedCount.toInt()) + } + + @Test + fun theDefaultDecoderIgnoresTrim() { + assertFalse(MessageDecoder.Default.trimIfIdle(0, TimeUtils.nowMillis()), "stateless: nothing to release") + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt index d61e46432e..6f8b43017c 100644 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt @@ -105,6 +105,61 @@ class DedupCapacityBenchmark { return frame.substring(k + 6, k + 6 + 64) } + private fun used(): Long { + val rt = Runtime.getRuntime() + return rt.totalMemory() - rt.freeMemory() + } + + /** Retained bytes of whatever [build] returns, with GC settled either side. */ + private fun retained(build: () -> Any): Pair { + repeat(3) { + System.gc() + Thread.sleep(60) + } + val before = used() + val held = build() + repeat(3) { + System.gc() + Thread.sleep(60) + } + return (used() - before) to held + } + + /** + * What the cache actually costs to hold. + * + * This is the WORST case: the decoder is the only holder, so every cached Event + * counts. In the app most of them are also in `LocalCache` (the decoder hands out + * the very same instance), and for those the true incremental cost is just the map + * node. The gap between the two is the memory the decoder pins for events + * `LocalCache` chose to drop. + */ + @Test + fun retainedMemoryPerCapacity() { + val frames = frames() + var sink: Any? = null + println("\n=== retained heap held by the decoder (worst case: sole holder) ===") + (CAPACITIES + listOf(2_048)).distinct().sorted().forEach { cap -> + val (bytes, held) = + retained { + val d = CachingEventDecoder(capacity = cap) + frames.forEach { d.decode(it) } + d + } + sink = held + val entries = minOf(cap * 2, 25_571) + println( + " capacity %7d %6.1f MB retained (<= %d cached events, %.0f B/entry)".format( + cap, + bytes / 1048576.0, + entries, + bytes.toDouble() / entries, + ), + ) + } + checkNotNull(sink) + } + @Test fun productionCapacityAgainstRealArrivalOrder() { val frames = frames() From ccebc3d344c36cb9aafaa4a3aae1b98245bcfc43 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Fri, 14 Aug 2026 10:20:23 -0400 Subject: [PATCH 7/8] fix: age the dedup cache on a clock, not on idleness The idle-triggered release in the previous commit never fired. Measured on the emulator: 240s of a completely flat heap, and every tick returned false. Cause: relays keep pushing events down open subscriptions forever, so the decoder is never idle in the sense of "saw no frames". Counting cache hits as activity -- which the tests asserted, and which is right in isolation, since a stream of pure duplicates is when the cache is earning its keep -- guaranteed the idle clock was refreshed indefinitely. The unit tests proved the intended behaviour; only the device showed the intent was wrong. Replaced with unconditional clock-based aging: ageOutCache() retires the live generation, so an id survives one tick and dies on the next, and a 30s tick bounds any id's lifetime at ~60s. clearCache() still releases everything when the host disconnects. Capacity keeps bounding cost during a burst; this bounds how long it costs anything afterwards. On-device A/B, same account and duration: idle-based (never fired): flat 127-128 MB for 240s clock-based, run 1: 130 MB -> 71 MB when 7,436 ids were dropped clock-based, run 2: 143 MB -> 121 MB (3,153 ids) -> 96 MB (4,530 ids) Each run contains its own control: the first tick only retires a generation, drops nothing, and frees nothing. Note ~7-8 KB released per cached id, an order above the ~600 B/entry the recorded corpus suggested -- live traffic carries big kind-0/kind-3 events the capture's first 150k frames under-represent. So capacity 8192 was costing tens of MB on a real account, not the 4.7 MB the corpus implied. Tests rewritten for the new semantics (no clock needed now, so they are fully deterministic) and mutation-checked: a no-op ageOut fails 4, an ageOut that clears both generations at once fails 3. Co-Authored-By: Claude Opus 5 (1M context) --- .../nip01Core/relay/client/NostrClient.kt | 29 ++-- .../commands/toClient/CachingEventDecoder.kt | 51 +++---- .../relay/commands/toClient/MessageDecoder.kt | 24 ++-- .../toClient/CachingEventDecoderTrimTest.kt | 127 +++++++++--------- 4 files changed, 117 insertions(+), 114 deletions(-) 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 99d517a9f1..1f02032687 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 @@ -160,7 +160,7 @@ class NostrClient( // decoder cached has no dedup value left across that gap, but it would keep // costing memory for the whole background stretch, so release it now rather // than leaving a timer running to do it later. - decoder.trimIfIdle(idleMillis = 0) + decoder.clearCache() } override fun isActive() = isActiveFlow.value @@ -214,30 +214,33 @@ class NostrClient( } /** - * Releases the decoder's cache after a quiet stretch. + * Ages out the decoder's cache on a clock. * - * A caching decoder bounds what it holds by capacity, which caps the cost during - * a burst but never ends it: its generations only rotate inside an insert, so a - * client that goes quiet pins every cached event indefinitely. Dedup value decays - * within seconds (duplicates are the same event arriving from other relays), so - * anything still held after [DECODER_IDLE_MS] is pure cost. + * A caching decoder bounds what it holds by capacity, which caps the cost during a + * burst but never ends it: its generations only rotate inside an insert, so a + * client that drops to a trickle keeps everything from the initial burst forever. * - * Suspends on [isActiveFlow] like [keepAliveJob], so no timer fires while the - * client is down — [disconnect] already trims eagerly on the way there. + * Deliberately NOT idle-triggered. Measured on device: relays keep pushing events + * down open subscriptions indefinitely, so the decoder is never idle in the sense + * of "saw no frames" — an idle check returned false on every tick across 240s of + * a flat heap. Aging unconditionally is what actually releases the memory. + * + * Ids survive one tick and die on the next, so [DECODER_AGE_OUT_MS] bounds their + * lifetime at ~2x that. Suspends on [isActiveFlow] like [keepAliveJob], so no timer + * fires while the client is down — [disconnect] clears outright on the way there. */ private val decoderTrimJob = scope.launch { while (true) { isActiveFlow.first { it } - delay(DECODER_TRIM_INTERVAL_MS) - decoder.trimIfIdle(DECODER_IDLE_MS) + delay(DECODER_AGE_OUT_MS) + decoder.ageOutCache() } } companion object { private const val KEEP_ALIVE_INTERVAL_MS = 60_000L - private const val DECODER_TRIM_INTERVAL_MS = 30_000L - private const val DECODER_IDLE_MS = 60_000L + private const val DECODER_AGE_OUT_MS = 30_000L } override fun reconnect( 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 index c4bb19ee97..f2ea54e695 100644 --- 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 @@ -22,7 +22,6 @@ package com.vitorpamplona.quartz.nip01Core.relay.commands.toClient import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey -import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.cache.ConcurrentHashCache import kotlin.concurrent.Volatile import kotlin.concurrent.atomics.AtomicLong @@ -94,16 +93,6 @@ class CachingEventDecoder( @Volatile private var previous = ConcurrentHashCache() - /** - * When an EVENT frame was last seen, for [trimIfIdle]. - * - * Written on every EVENT frame — hit or miss — so a cache that is actively - * serving duplicates counts as busy and is never trimmed out from under the - * traffic it is helping. One relaxed volatile store per frame; at observed - * rates (~400 frames/s) that is not measurable. - */ - @Volatile private var lastActivityAt: Long = TimeUtils.nowMillis() - // Observability + benchmark hooks. private val parsed = AtomicLong(0) private val reused = AtomicLong(0) @@ -117,7 +106,6 @@ class CachingEventDecoder( val cached = live.get(scanned.eventId) ?: previous.get(scanned.eventId) if (cached != null) { reused.incrementAndFetch() - lastActivityAt = TimeUtils.nowMillis() return EventMessage(scanned.subId, cached) } } @@ -125,7 +113,6 @@ class CachingEventDecoder( val msg = fullParser.decode(text) if (msg is EventMessage) { parsed.incrementAndFetch() - lastActivityAt = TimeUtils.nowMillis() if (live.size() >= capacity) { // Benign-race rotation: see class kdoc. previous = live @@ -137,29 +124,35 @@ class CachingEventDecoder( } /** - * Drops both generations once no EVENT frame has arrived for [idleMillis]. + * Retires the live generation, so entries age out on the caller's clock instead of + * only when the cache fills. * - * [capacity] bounds what the cache costs *during* a burst; this bounds how long - * it costs anything *after* one. Without it the maps only ever rotate inside an - * insert, so a client that goes quiet pins up to `2 * capacity` events for the - * rest of the process's life — memory whose dedup value has already decayed to - * zero. + * [capacity] bounds what the cache costs *during* a burst; this bounds how long it + * costs anything *after* one. Without it the maps only rotate inside an insert, so + * a client receiving a slow trickle keeps everything it saw during the initial + * burst forever — memory whose dedup value is long gone. * - * Racy by the same design as rotation (see class kdoc): clearing concurrently - * with an insert can only lose an id, and losing an id costs a re-parse, never - * a wrong message. [nowMillis] is injectable so tests need no clock. + * An id survives one call and dies on the next, so ticking every 30s keeps nothing + * older than ~60s. Two consecutive calls with no traffic in between leave the cache + * empty, which is what releases memory when the client really does go quiet. + * + * Racy by the same design as capacity rotation (see class kdoc): retiring a + * generation concurrently with an insert can only lose an id, and a lost id costs a + * re-parse, never a wrong message. */ - override fun trimIfIdle( - idleMillis: Long, - nowMillis: Long, - ): Boolean { - if (nowMillis - lastActivityAt < idleMillis) return false - if (live.size() == 0 && previous.size() == 0) return false + override fun ageOutCache() { + previous = live + live = ConcurrentHashCache() + } + + override fun clearCache() { previous = ConcurrentHashCache() live = ConcurrentHashCache() - return true } + /** Ids currently addressable across both generations. Observability + tests. */ + val cachedCount: Int get() = live.size() + previous.size() + private class ScannedEvent( val subId: String, val eventId: HexKey, 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 index 3373977580..6362040b18 100644 --- 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 @@ -21,7 +21,6 @@ package com.vitorpamplona.quartz.nip01Core.relay.commands.toClient import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper -import com.vitorpamplona.quartz.utils.TimeUtils /** * Turns one raw relay frame into a [Message]. The strategy the relay client @@ -35,20 +34,21 @@ fun interface MessageDecoder { fun decode(text: String): Message /** - * Releases whatever the decoder is holding if it has seen no frames for - * [idleMillis]. Returns true when something was released. + * Ages out whatever the decoder is holding. Call periodically: an entry must + * survive at most two calls, so a 30s tick keeps nothing older than ~60s. * - * A caching decoder's value decays with time — a duplicate that has not - * arrived in a minute is not going to — but its memory does not, so the - * owner ticks this to let a quiet decoder drop what it cached. Stateless - * decoders keep the no-op default. + * A caching decoder's value decays with time — a duplicate that has not arrived + * within seconds is not going to — while its memory does not decay at all. Aging + * on a clock rather than on idleness is deliberate: relays keep pushing a trickle + * of events down open subscriptions forever, so a decoder is never idle in the + * sense of "saw no frames", and an idle-triggered release would never fire. * - * Never affects correctness: releasing a cache can only cost a re-parse. + * Never affects correctness: dropping a cached id can only cost a re-parse. */ - fun trimIfIdle( - idleMillis: Long, - nowMillis: Long = TimeUtils.nowMillis(), - ): Boolean = false + fun ageOutCache() = Unit + + /** Releases everything cached, e.g. when the host is shutting the client down. */ + fun clearCache() = Unit companion object { /** Plain full-JSON parse of every frame. */ diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt index cda94b59a5..28d3ff7ea0 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt @@ -21,25 +21,24 @@ package com.vitorpamplona.quartz.nip01Core.relay.commands.toClient import com.vitorpamplona.quartz.nip01Core.core.Event -import com.vitorpamplona.quartz.utils.TimeUtils import kotlin.test.Test import kotlin.test.assertEquals -import kotlin.test.assertFalse import kotlin.test.assertTrue /** - * [CachingEventDecoder.trimIfIdle] bounds how long the cache costs memory, which - * [CachingEventDecoder.capacity] does not: the generations only rotate inside an - * insert, so without a trim a client that goes quiet pins every cached event for - * the rest of the process's life. + * [CachingEventDecoder.ageOutCache] bounds how long the cache costs memory, which + * [CachingEventDecoder.capacity] does not: the generations otherwise only rotate + * inside an insert, so a client that drops to a trickle keeps everything from its + * initial burst forever. * - * The rules that matter here are "release when genuinely idle", "do NOT release - * out from under live traffic" — including traffic that is *only* cache hits, - * which is exactly when the cache is earning its keep — and "releasing never - * changes what a decode returns, only whether it re-parsed". + * Aging is on a clock, not on idleness, and that distinction is load-bearing — + * measured on device, relays keep pushing events down open subscriptions + * indefinitely, so an idle-triggered release returned false on every tick across + * 240s of a flat heap. The rules pinned here are the lifetime guarantee ("an entry + * survives one call and dies on the next"), that traffic between calls is kept, and + * that aging only ever costs a re-parse — never a different message. * - * In `commonTest`: this is `commonMain` logic with no platform surface, and the - * clock is injected so no target needs a real one. + * In `commonTest`: `commonMain` logic, no platform surface, no clock. */ class CachingEventDecoderTrimTest { private fun hexId(seed: Int) = seed.toString(16).padStart(64, '0') @@ -60,78 +59,84 @@ class CachingEventDecoderTrimTest { subId: String = "s", ) = """["EVENT","$subId",${event(seed).toJson()}]""" - private val minute = 60_000L - @Test - fun releasesTheCacheOnceIdle() { + fun entriesSurviveExactlyOneAgeOut() { val decoder = CachingEventDecoder(capacity = 1_000) - (1..50).forEach { decoder.decode(frame(it)) } - assertEquals(50, decoder.parsedCount.toInt()) - - // read the clock AFTER the decodes: they set the idle clock as they run, so a - // cutoff measured from before them is only a full minute later while the - // parsing itself takes under a millisecond - val quietSince = TimeUtils.nowMillis() - assertTrue(decoder.trimIfIdle(minute, quietSince + minute), "should trim after a quiet minute") - - // the ids are gone, so a repeat of a previously cached frame parses again decoder.decode(frame(1)) - assertEquals(51, decoder.parsedCount.toInt()) - assertEquals(0, decoder.reusedCount.toInt()) - } - - @Test - fun keepsTheCacheWhileTrafficIsFlowing() { - val decoder = CachingEventDecoder(capacity = 1_000) - val t0 = TimeUtils.nowMillis() - (1..50).forEach { decoder.decode(frame(it)) } - - assertFalse(decoder.trimIfIdle(minute, t0 + 1_000), "1s of quiet is not idle") + assertEquals(1, decoder.parsedCount.toInt()) + // first tick: retired into `previous`, still addressable + decoder.ageOutCache() decoder.decode(frame(1)) - assertEquals(50, decoder.parsedCount.toInt(), "cached ids must survive") + assertEquals(1, decoder.parsedCount.toInt(), "one age-out must not drop the id") assertEquals(1, decoder.reusedCount.toInt()) } @Test - fun cacheHitsCountAsActivity() { - // A stream of pure duplicates inserts nothing, but it is precisely when the - // cache is most valuable. Keying idleness off inserts alone would trim it away - // under the traffic it is serving. + fun twoAgeOutsWithNoTrafficEmptyTheCache() { + val decoder = CachingEventDecoder(capacity = 1_000) + (1..50).forEach { decoder.decode(frame(it)) } + assertEquals(50, decoder.cachedCount) + + decoder.ageOutCache() + decoder.ageOutCache() + + assertEquals(0, decoder.cachedCount, "nothing survives two ticks without traffic") + decoder.decode(frame(1)) + assertEquals(51, decoder.parsedCount.toInt(), "the id is gone, so it re-parses") + assertEquals(0, decoder.reusedCount.toInt()) + } + + @Test + fun trafficBetweenTicksIsKept() { + // the trickle case: events keep arriving, so the cache must keep serving them val decoder = CachingEventDecoder(capacity = 1_000) decoder.decode(frame(1)) + decoder.ageOutCache() + decoder.decode(frame(2)) + decoder.ageOutCache() - val muchLater = TimeUtils.nowMillis() + 10 * minute - repeat(5) { decoder.decode(frame(1, subId = "sub$it")) } + // 1 was inserted two ticks ago -> gone; 2 was inserted one tick ago -> kept + decoder.decode(frame(2)) + assertEquals(2, decoder.parsedCount.toInt()) + assertEquals(1, decoder.reusedCount.toInt(), "the recent event is still cached") - assertEquals(5, decoder.reusedCount.toInt()) - assertFalse( - decoder.trimIfIdle(minute, TimeUtils.nowMillis() + 1_000), - "hits must refresh the idle clock", - ) - // ...but a genuinely quiet stretch after those hits still trims - assertTrue(decoder.trimIfIdle(minute, muchLater)) + decoder.decode(frame(1)) + assertEquals(3, decoder.parsedCount.toInt(), "the older event aged out") } @Test - fun trimmingAnEmptyCacheDoesNothing() { - val decoder = CachingEventDecoder(capacity = 1_000) - assertFalse(decoder.trimIfIdle(0, TimeUtils.nowMillis() + minute), "nothing to release") + fun agingABusyCacheKeepsItSmallInsteadOfUnbounded() { + // what the device measurement is about: a burst, then a trickle. Without aging + // the burst's ids stay resident forever. + val decoder = CachingEventDecoder(capacity = 100_000) + (1..5_000).forEach { decoder.decode(frame(it)) } + assertEquals(5_000, decoder.cachedCount) + + decoder.ageOutCache() + (5_001..5_010).forEach { decoder.decode(frame(it)) } // the trickle + decoder.ageOutCache() + + assertEquals(10, decoder.cachedCount, "only the trickle survives, not the burst") } @Test - fun zeroIdleReleasesImmediately() { + fun clearCacheReleasesEverythingAtOnce() { // what NostrClient.disconnect() uses when the host backgrounds the app val decoder = CachingEventDecoder(capacity = 1_000) (1..10).forEach { decoder.decode(frame(it)) } - assertTrue(decoder.trimIfIdle(idleMillis = 0, nowMillis = TimeUtils.nowMillis())) + assertTrue(decoder.cachedCount > 0) + + decoder.clearCache() + assertEquals(0, decoder.cachedCount) } @Test - fun aTrimNeverChangesWhatADecodeReturns() { + fun agingNeverChangesWhatADecodeReturns() { val decoder = CachingEventDecoder(capacity = 1_000) val before = decoder.decode(frame(7, subId = "a")) as EventMessage - decoder.trimIfIdle(0, TimeUtils.nowMillis()) + decoder.ageOutCache() + decoder.ageOutCache() val after = decoder.decode(frame(7, subId = "a")) as EventMessage assertEquals(before.subId, after.subId) @@ -139,12 +144,14 @@ class CachingEventDecoderTrimTest { assertEquals(before.event.pubKey, after.event.pubKey) assertEquals(before.event.content, after.event.content) assertEquals(before.event.createdAt, after.event.createdAt) - // only the parse count differs: the trim cost one re-parse, nothing else + // only the parse count differs: aging cost one re-parse, nothing else assertEquals(2, decoder.parsedCount.toInt()) } @Test - fun theDefaultDecoderIgnoresTrim() { - assertFalse(MessageDecoder.Default.trimIfIdle(0, TimeUtils.nowMillis()), "stateless: nothing to release") + fun theDefaultDecoderIgnoresBoth() { + // stateless: nothing to age or release, and neither call may throw + MessageDecoder.Default.ageOutCache() + MessageDecoder.Default.clearCache() } } From b1f1190919bc763d8f73aadac05012257d858ccf Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Fri, 14 Aug 2026 11:24:47 -0400 Subject: [PATCH 8/8] docs: record the tick sweep that picked 30s for the dedup age-out 30s was the one number in this change I picked rather than measured. Swept it on device (10/30/60/120s). The naive comparison is invalid: runs pull 3.7k-32k frames depending on what the relays serve, so hit rate and heap at the end of a run are not comparable. Comparing at a matched ~19k frames instead: tick hit rate parses ids still cached at rest 10s 44.9% 10,663 ~2 30s 58.5% 7,569 ~2 60s 60.4% 7,520 ~3 120s 60.6% 7,555 7,438 Hit rate saturates by 30s, so a shorter tick buys only re-parses (10s costs 41% more) and a longer one only holds memory -- at 120s just one tick fires in three minutes, so the burst never drops at all. The knee is where the tick stops being the binding constraint and capacity takes over: at ~400 frames/s that is 8192/400 ~= 20s, and 30s sits just past it. Keeps 30s; documents why, and what to re-measure if capacity or frame rates change. No behaviour change. Co-Authored-By: Claude Opus 5 (1M context) --- .../nip01Core/relay/client/NostrClient.kt | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) 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 1f02032687..30da4b8992 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 @@ -228,6 +228,24 @@ class NostrClient( * Ids survive one tick and die on the next, so [DECODER_AGE_OUT_MS] bounds their * lifetime at ~2x that. Suspends on [isActiveFlow] like [keepAliveJob], so no timer * fires while the client is down — [disconnect] clears outright on the way there. + * + * 30s is measured, not guessed. Sweeping the interval on device and comparing hit + * rate at a matched ~19k frames (runs pull different volumes, so raw endpoints are + * not comparable): + * + * ``` + * tick hit rate parses ids still cached at rest + * 10s 44.9% 10,663 ~2 + * 30s 58.5% 7,569 ~2 + * 60s 60.4% 7,520 ~3 + * 120s 60.6% 7,555 7,438 (one tick in 3 min -> never releases) + * ``` + * + * Hit rate saturates by 30s, so a shorter tick only buys re-parses (10s costs 41% + * more), and a longer one only holds memory. The knee is where the tick stops being + * the binding constraint and [CachingEventDecoder.capacity] takes over: at the + * observed ~400 frames/s that crossover is 8192/400 ~= 20s, so 30s sits just past + * it. Re-measure if capacity or typical frame rates change. */ private val decoderTrimJob = scope.launch {