diff --git a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md b/quartz/plans/2026-07-02-nostrclient-receiver-perf.md index 3002da4fdb..389d5f61f4 100644 --- a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md +++ b/quartz/plans/2026-07-02-nostrclient-receiver-perf.md @@ -169,6 +169,83 @@ Findings: `onNewEvent`/stats counters, so paging/EOSE bookkeeping would need the cheap counters kept ahead of the skip. +## Bulk download: N-million events from one relay (download + parse only) + +**Harness:** `quartz/src/jvmTest/.../relay/prodbench/BulkDownloadBenchmark.kt` +(gated: `./gradlew :quartz:jvmTest --tests "*.BulkDownloadBenchmark" -PprodRelayBench=1`) + +### Local ceilings (geode on localhost TCP, 100k seeded events, 4 cores) + +| strategy | events/s | wall for 100k | +|---|---|---| +| quartz, 1 conn, one giant REQ | 476 | 210s (and 2 events silently missing) | +| quartz, 4 conns time-sharded, giant REQs | 2,001 | 50s | +| quartz, 1 conn, paged 1000/page | 12,480 | 8.0s | +| **quartz, 4 conns time-sharded + paged** | **24,210** | **4.1s** | +| raw socket (no parse), one giant REQ | ~676 | timed out at 120s (81k/100k) | + +- **Giant single REQs are a trap.** The raw (no-parse) variant proves it's + server-side: geode streams a 100k-event response at ~700 events/s + (per-frame cost in the session pump / query streaming path — needs its own + investigation, `WebSocketSessionPump`), and the quartz run came back 2 + events short. Public relays are worse: they clamp limits and may drop + frames or the connection under output backpressure. Paged cursors through + the very same server ran 26× faster. +- **Sharding the `created_at` range across connections stacks with paging:** + 4 sharded + paged connections ≈ 2× one paged connection locally (24k/s), + and each connection gets its own receiver coroutine, so parse parallelizes + for free. + +### Per-frame strategies, offline (50k frames, ~580B each) + +| strategy | events/s | µs/frame | +|---|---|---| +| full `fromJsonToMessage`, 1 thread | 276k | 3.6 | +| full parse, 4 threads | 657k | 1.5 (aggregate) | +| id-scan only (archive raw, parse lazily) | 2.96M | 0.34 | + +Parse of small events is ~3.6µs; the earlier production capture (mixed sizes, +damus 8.7KB events) measured 32µs. Either way **parse is not the bottleneck +for bulk download**: one core parses 10M small events in ~36s, and a 4-core +fan-out does it in ~15s. The wire and the relay's page cadence dominate. + +### Production test case: kinds=[30382] on nip85.nosfabrica.com (cap 20k) + +| strategy | wall | events/s | notes | +|---|---|---|---| +| paged download (until-cursor) | 5.4s | 3,711 | 41 pages ≈ 500/page (relay clamp) | +| NIP-77 negentropy sync | 12.7s | 1,575 | full set = 31,241 ids, 9 reconcile windows | + +Negentropy correctly enumerated the whole 31,241-id set (splitting 9 windows +around the relay's `max_sync_events` cap) and streamed id-batch REQs — but +for a **cold** download it was 2.4× slower than plain paging: the reconcile +rounds and by-id lookups cost more than sequential pages when you need +*everything anyway*. Negentropy's win is **incremental re-sync**: once the +10M events are local, the next sync transfers only fingerprints + the diff +instead of re-paging the world. (kind 30382 events carry empty `content` — +all data in tags — so the MB/s column reads 0.) + +### Answer for "10M events from one relay, fastest" + +At nosfabrica's measured page cadence, one connection ≈ 3.7k events/s → 10M +in ~45 min. To improve, in order: + +1. **Page with until-cursors — never one giant REQ** (silent drops, server + slow paths, relay clamps). +2. **Shard the `created_at` range across K connections** to the same relay + (each shard pages independently; no cursor dependency between shards). + Local: 2× at K=4; WAN, where RTT dominates page turnaround, closer to + linear until the relay rate-limits per-IP. +3. **Don't optimize parse first** — it's 2–5% of the budget. If the goal is + archival, id-scan + store raw frames (0.34µs/frame) and parse lazily + in parallel later. +4. **Use negentropy for the second sync onward**, not the first. +5. **Memory:** the per-connection channel is UNLIMITED — at 10M events a + sink slower than the socket accumulates heap without bound. A bulk + downloader should bound the channel (blocking the OkHttp reader thread is + fine — that's TCP backpressure doing its job) and stream events to disk, + never hold the set. + ## Recommendations (in order of value/risk) 1. **Move Schnorr verification off the receiver coroutine** in the app's diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/BulkDownloadBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/BulkDownloadBenchmark.kt new file mode 100644 index 0000000000..23fbd9f6b2 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/BulkDownloadBenchmark.kt @@ -0,0 +1,419 @@ +/* + * 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.KtorRelay +import com.vitorpamplona.geode.RelayEngine +import com.vitorpamplona.geode.fixtures.SyntheticEvents +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySync +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.LimitsPolicy +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RelayLimits +import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.runBlocking +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicLong +import kotlin.test.Test + +/** + * Answers: "we need to download N-million events from a single relay as fast + * as possible — considering just download and parsing, what helps?" + * + * Three parts: + * + * 1. LOCAL CEILINGS — a real geode relay on a localhost TCP port, seeded with + * [LOCAL_EVENTS] events, so network bandwidth and server latency are ~free + * and the client stack is what's measured. Variants: 1 connection vs + * [SHARDS] connections with the `created_at` range sharded across them; + * single giant REQ vs realistic 1000/page cursor pagination (a second + * server instance clamps limits to force paging); and the full quartz + * stack (parse + dispatch) vs a raw OkHttp socket that only counts frames + * (no parse), which isolates the parse share of the pipeline. + * + * 2. OFFLINE PARSE STRATEGIES — what to do with each frame: full + * `fromJsonToMessage`, full parse fanned across cores, or an id-only scan + * (the "archive the raw frame now, parse lazily later" strategy). + * + * 3. PRODUCTION TEST CASE — NIP-77 negentropy sync of `kinds:[30382]` from + * nip85.nosfabrica.com (capped at [PROD_MAX_EVENTS]) against plain + * until-cursor paging of the same filter. Negentropy enumerates the id set + * server-side and downloads by id-batch with 8 concurrent REQs on one + * connection — the protocol-level answer to serial page round-trips. + * + * Gated behind PROD_RELAY_BENCH (same as ProductionReceiverBenchmark): + * ./gradlew :quartz:jvmTest --tests "*.BulkDownloadBenchmark" -PprodRelayBench=1 + */ +class BulkDownloadBenchmark { + companion object { + const val LOCAL_EVENTS = 100_000 + const val SHARDS = 4 + const val PAGE_LIMIT = 1_000 + const val LOCAL_PAGE_TIMEOUT_MS = 120_000L + + const val PROD_RELAY = "wss://nip85.nosfabrica.com" + const val PROD_KIND = 30382 + const val PROD_MAX_EVENTS = 20_000 + } + + private fun report( + name: String, + events: Long, + bytes: Long, + wallNanos: Long, + ) { + println( + " %-34s %,8d events %6.1f MB wall=%6.0fms -> %,8.0f events/s %5.1f MB/s" + .format( + name, + events, + bytes / 1e6, + wallNanos / 1e6, + events * 1e9 / wallNanos, + bytes * 1e3 / wallNanos, + ), + ) + } + + // ------------------------------------------------------------------ + // Part 1: local ceilings against a seeded geode relay on localhost TCP + // ------------------------------------------------------------------ + + /** Splits [1, LOCAL_EVENTS] (the seeded createdAt space) into [shards] windows. */ + private fun windows(shards: Int): List> { + val step = LOCAL_EVENTS / shards + return (0 until shards).map { i -> + val lo = i * step + 1L + val hi = if (i == shards - 1) LOCAL_EVENTS.toLong() else (i + 1) * step.toLong() + lo to hi + } + } + + /** Full quartz stack: NostrClient + fetchAllPages, one client per shard. */ + private suspend fun quartzDownload( + name: String, + relayUrl: NormalizedRelayUrl, + httpClient: OkHttpClient, + shards: Int, + ) { + val count = AtomicLong(0) + val bytes = AtomicLong(0) + val start = System.nanoTime() + coroutineScope { + windows(shards) + .map { (lo, hi) -> + async(Dispatchers.IO) { + val client = NostrClient(BasicOkHttpWebSocket.Builder { httpClient }) + try { + client.fetchAllPages( + relay = relayUrl, + filters = listOf(Filter(kinds = listOf(1), since = lo, until = hi)), + timeoutMs = LOCAL_PAGE_TIMEOUT_MS, + ) { event -> + count.incrementAndGet() + bytes.addAndGet(event.content.length.toLong()) + } + } finally { + client.close() + } + } + }.awaitAll() + } + val wall = System.nanoTime() - start + report(name, count.get(), bytes.get(), wall) + if (count.get() != LOCAL_EVENTS.toLong()) { + println(" !! expected $LOCAL_EVENTS events, got ${count.get()}") + } + } + + /** + * Raw OkHttp socket per shard: counts EVENT frames, no JSON parse, no + * quartz client. The download-only floor the parse stage sits on top of. + */ + private fun rawDownload( + name: String, + serverWsUrl: String, + httpClient: OkHttpClient, + shards: Int, + ) { + val count = AtomicLong(0) + val bytes = AtomicLong(0) + val latch = CountDownLatch(shards) + val start = System.nanoTime() + + val sockets = + windows(shards).mapIndexed { i, (lo, hi) -> + httpClient.newWebSocket( + Request.Builder().url(serverWsUrl).build(), + object : okhttp3.WebSocketListener() { + override fun onOpen( + webSocket: okhttp3.WebSocket, + response: Response, + ) { + webSocket.send("""["REQ","raw$i",{"kinds":[1],"since":$lo,"until":$hi}]""") + } + + override fun onMessage( + webSocket: okhttp3.WebSocket, + text: String, + ) { + if (text.startsWith("[\"EVENT\"")) { + count.incrementAndGet() + bytes.addAndGet(text.length.toLong()) + } else if (text.startsWith("[\"EOSE\"")) { + latch.countDown() + } + } + + override fun onFailure( + webSocket: okhttp3.WebSocket, + t: Throwable, + response: Response?, + ) { + latch.countDown() + } + }, + ) + } + + latch.await(120, TimeUnit.SECONDS) + val wall = System.nanoTime() - start + sockets.forEach { it.cancel() } + report(name, count.get(), bytes.get(), wall) + } + + private fun localCeilings(httpClient: OkHttpClient) { + println("\n=== LOCAL CEILINGS: geode on localhost, $LOCAL_EVENTS seeded events ===") + + val placeholderA = "ws://127.0.0.1:7771/".normalizeRelayUrl() + val placeholderB = "ws://127.0.0.1:7772/".normalizeRelayUrl() + + val engine = RelayEngine(url = placeholderA) + val seedNanos = + runBlocking { + val events = + (1..LOCAL_EVENTS).map { + SyntheticEvents.fakeEvent( + idSeed = it, + kind = 1, + pubKey = SyntheticEvents.hexId(it % 1000 + 1), + createdAt = it.toLong(), + content = "bulk download benchmark payload $it ".repeat(6), + ) + } + val t = System.nanoTime() + events.chunked(2_000).forEach { engine.store.batchInsert(it) } + System.nanoTime() - t + } + println(" (seeded in %.1fs)".format(seedNanos / 1e9)) + + // Same store, second engine that clamps every REQ to PAGE_LIMIT — forces + // fetchAllPages into realistic until-cursor pagination. + val pagedEngine = + RelayEngine( + url = placeholderB, + store = engine.store, + policyBuilder = { LimitsPolicy(RelayLimits(defaultLimit = PAGE_LIMIT, maxLimit = PAGE_LIMIT)) }, + ) + + val openServer = KtorRelay(engine, port = 0).start() + val pagedServer = KtorRelay(pagedEngine, port = 0).start() + try { + val openUrl = openServer.url.normalizeRelayUrl() + val pagedUrl = pagedServer.url.normalizeRelayUrl() + + runBlocking { + rawDownload("raw 1-conn single-REQ (no parse)", openServer.url, httpClient, shards = 1) + rawDownload("raw $SHARDS-conn sharded (no parse)", openServer.url, httpClient, shards = SHARDS) + quartzDownload("quartz 1-conn single-REQ", openUrl, httpClient, shards = 1) + quartzDownload("quartz $SHARDS-conn sharded", openUrl, httpClient, shards = SHARDS) + quartzDownload("quartz 1-conn paged $PAGE_LIMIT", pagedUrl, httpClient, shards = 1) + quartzDownload("quartz $SHARDS-conn sharded+paged", pagedUrl, httpClient, shards = SHARDS) + } + } finally { + openServer.stop(gracePeriodMillis = 200, timeoutMillis = 1_000) + pagedServer.stop(gracePeriodMillis = 200, timeoutMillis = 1_000) + pagedEngine.close() + engine.close() + } + } + + // ------------------------------------------------------------------ + // Part 2: offline per-frame parse strategies + // ------------------------------------------------------------------ + + private fun offlineParseStrategies() { + println("\n=== OFFLINE: per-frame strategies (50k synthetic frames) ===") + val frames = + (1..50_000).map { + val event = + SyntheticEvents.fakeEvent( + idSeed = it, + kind = 1, + pubKey = SyntheticEvents.hexId(it % 1000 + 1), + createdAt = it.toLong(), + content = "bulk download benchmark payload $it ".repeat(6), + ) + """["EVENT","sub",${event.toJson()}]""" + } + val totalBytes = frames.sumOf { it.length.toLong() } + + // warmup + frames.take(5_000).forEach { OptimizedJsonMapper.fromJsonToMessage(it) } + + var t = System.nanoTime() + frames.forEach { OptimizedJsonMapper.fromJsonToMessage(it) } + report("full parse, 1 thread", frames.size.toLong(), totalBytes, System.nanoTime() - t) + + val cores = Runtime.getRuntime().availableProcessors() + t = System.nanoTime() + runBlocking(Dispatchers.Default) { + frames + .chunked((frames.size + cores - 1) / cores) + .map { chunk -> async { chunk.forEach { OptimizedJsonMapper.fromJsonToMessage(it) } } } + .awaitAll() + } + report("full parse, $cores threads", frames.size.toLong(), totalBytes, System.nanoTime() - t) + + // id-only scan: what "write the raw frame to disk now, parse lazily + // later" pays per frame to dedup/route. + var sink = 0 + t = System.nanoTime() + frames.forEach { frame -> + val i = frame.indexOf("\"id\":\"") + if (i >= 0) sink += frame[i + 6].code + } + report("id-scan only (archive raw)", frames.size.toLong(), totalBytes, System.nanoTime() - t) + check(sink != 0) + } + + // ------------------------------------------------------------------ + // Part 3: production — negentropy vs paging for kind 30382 on nosfabrica + // ------------------------------------------------------------------ + + private suspend fun productionNegentropy(httpClient: OkHttpClient) { + println("\n=== PRODUCTION: $PROD_RELAY kinds=[$PROD_KIND], capped at $PROD_MAX_EVENTS events ===") + val relay = PROD_RELAY.normalizeRelayUrl() + + // A) NIP-77 negentropy: server enumerates the id set, we download by + // id-batches with 8 concurrent REQs on the same connection. + run { + val client = NostrClient(BasicOkHttpWebSocket.Builder { httpClient }) + try { + val count = AtomicLong(0) + val bytes = AtomicLong(0) + val start = System.nanoTime() + val result = + client.negentropySync( + relay = relay, + filter = Filter(kinds = listOf(PROD_KIND)), + maxEvents = PROD_MAX_EVENTS, + onProgress = { need, downloaded -> + if (downloaded % 5_000 == 0 && downloaded > 0) { + println(" … negentropy progress: need=$need downloaded=$downloaded") + } + }, + ) { event -> + count.incrementAndGet() + bytes.addAndGet(event.content.length.toLong()) + } + val wall = System.nanoTime() - start + report("negentropy sync", count.get(), bytes.get(), wall) + println( + " relay reported need=${result.needCount} ids for the full set, " + + "reconciled in ${result.windows} window(s), downloaded=${result.downloaded}", + ) + } catch (e: NegentropySyncException) { + println(" !! negentropy failed: ${e.reason} — ${e.message}") + } catch (e: Exception) { + println(" !! negentropy errored: ${e::class.simpleName} ${e.message}") + } finally { + client.close() + } + } + + // B) Plain until-cursor paging of the same filter, same cap. + run { + val client = NostrClient(BasicOkHttpWebSocket.Builder { httpClient }) + try { + val count = AtomicLong(0) + val bytes = AtomicLong(0) + var pages = 0 + val start = System.nanoTime() + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(PROD_KIND), limit = PROD_MAX_EVENTS)), + timeoutMs = 30_000L, + onNewPage = { pages++ }, + ) { event -> + count.incrementAndGet() + bytes.addAndGet(event.content.length.toLong()) + } + val wall = System.nanoTime() - start + report("paged download", count.get(), bytes.get(), wall) + println(" pages=${pages + 1}") + } catch (e: Exception) { + println(" !! paged download errored: ${e::class.simpleName} ${e.message}") + } finally { + client.close() + } + } + } + + // ------------------------------------------------------------------ + + @Test + fun bulkDownloadBenchmark() { + if (System.getenv("PROD_RELAY_BENCH") == null && System.getProperty("prodRelayBench") == null) { + println("BulkDownloadBenchmark skipped. Run with -PprodRelayBench=1 to enable.") + return + } + + val httpClient = + OkHttpClient + .Builder() + .connectTimeout(15, TimeUnit.SECONDS) + .readTimeout(120, TimeUnit.SECONDS) + .pingInterval(30, TimeUnit.SECONDS) + .build() + + println("=== BULK DOWNLOAD BENCHMARK === cores=${Runtime.getRuntime().availableProcessors()}") + + localCeilings(httpClient) + offlineParseStrategies() + runBlocking { productionNegentropy(httpClient) } + + httpClient.dispatcher.executorService.shutdown() + } +}