diff --git a/geode/build.gradle.kts b/geode/build.gradle.kts index 981d2a0abc..67588b19e4 100644 --- a/geode/build.gradle.kts +++ b/geode/build.gradle.kts @@ -69,11 +69,20 @@ tasks.withType().configureEach { systemProperty("runLoadBenchmark", System.getProperty("runLoadBenchmark") ?: "false") System.getProperty("fanoutScalingEvents")?.let { systemProperty("fanoutScalingEvents", it) } System.getProperty("fanoutScalingSubs")?.let { systemProperty("fanoutScalingSubs", it) } + // NegentropyServerReconcileBenchmark opt-in + sizing. + System.getProperty("negServerBench")?.let { systemProperty("negServerBench", it) } + System.getProperty("negBenchN")?.let { systemProperty("negBenchN", it) } + maxHeapSize = System.getProperty("testHeap") ?: maxHeapSize + // Opt-in JFR profiling (-PnegProfile=/tmp/neg.jfr). + (project.findProperty("negProfile") as? String)?.let { + jvmArgs("-XX:+FlightRecorder", "-XX:StartFlightRecording=filename=$it,settings=profile,dumponexit=true") + } // Show println output from test JVM so the benchmark numbers are // actually visible without grepping the report XML. testLogging { showStandardStreams = - (System.getProperty("runLoadBenchmark") == "true") + (System.getProperty("runLoadBenchmark") == "true") || + (System.getProperty("negServerBench") == "1") events("standard_out") } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/NegentropyServerReconcileBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/NegentropyServerReconcileBenchmark.kt new file mode 100644 index 0000000000..72af40fc3d --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/NegentropyServerReconcileBenchmark.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.geode.perf + +import com.vitorpamplona.geode.KtorRelay +import com.vitorpamplona.geode.RelayEngine +import com.vitorpamplona.geode.interop.InteropSyncDriver +import com.vitorpamplona.geode.relayIndexingStrategy +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore +import com.vitorpamplona.quartz.utils.EventFactory +import kotlinx.coroutines.runBlocking +import okhttp3.OkHttpClient +import kotlin.test.Test +import kotlin.test.assertEquals + +/** + * Reproduces relayBench's NIP-77 initial-reconcile phase against a **real + * in-process geode server** (KtorRelay → RelaySession → NegSessionRegistry + * → live-index snapshot → NegentropyServerSession), driven by the real + * client over a loopback WebSocket. Because client and server share one + * JVM, a JFR recording of this run attributes cost across geode's full + * request path — not just the reconciliation library. + * + * The companion [com.vitorpamplona.quartz.nip01Core.relay.prodbench.NegentropyReconcileBenchmark] + * times the *library only* (no server, no wire): ~200 ms server compute at + * the same 1M/200k-diff shape. The head-to-head relayBench run measured + * geode's live server at ~6 s. This benchmark closes that gap in a + * profilable single process, so we can see where the 30× goes. + * + * Opt-in (heavy — seeds up to 800k events): `-DnegServerBench=1`, size via + * `-DnegBenchN=1000000`. FTS is off (irrelevant to reconcile) so seeding is + * fast; the live negentropy index stays on, exactly as production runs it. + */ +class NegentropyServerReconcileBenchmark { + companion object { + val enabled = System.getProperty("negServerBench") == "1" + val N = System.getProperty("negBenchN")?.toInt() ?: 200_000 + const val BASE_TIME = 1_704_067_200L + const val EVENTS_PER_SECOND = 3 + } + + private fun mix(seed: Long): Long { + var z = seed + -0x61c8864680b583ebL + z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L + z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L + return z xor (z ushr 31) + } + + private val hexChars = "0123456789abcdef".toCharArray() + + private fun idFor(index: Int): String { + val out = CharArray(64) + for (w in 0 until 4) { + val v = mix(index.toLong() * 4 + w) + for (b in 0 until 8) { + val byte = ((v ushr (b * 8)) and 0xFF).toInt() + val pos = (w * 8 + b) * 2 + out[pos] = hexChars[byte ushr 4] + out[pos + 1] = hexChars[byte and 0xF] + } + } + return String(out) + } + + private val pubkey = "00".repeat(32) + private val sig = "0".repeat(128) + + private fun event(index: Int): Event = + EventFactory.create( + id = idFor(index), + pubKey = pubkey, + createdAt = BASE_TIME + (index / EVENTS_PER_SECOND).toLong(), + kind = 1, + tags = emptyArray(), + content = "", + sig = sig, + ) + + @Test + fun serverReconcileAgainstRealGeode() = + runBlocking { + if (!enabled) { + println("[skip] NegentropyServerReconcileBenchmark — set -DnegServerBench=1 to enable") + return@runBlocking + } + + // FTS off (not on the reconcile path); live index on (it is). + val store = EventStore(dbName = null, indexStrategy = relayIndexingStrategy(fullTextSearch = false)) + val relay = RelayEngine(url = "ws://127.0.0.1:7000/".normalizeRelayUrl(), store = store) + val server = KtorRelay(relay, host = "127.0.0.1", port = 0).start() + val http = OkHttpClient.Builder().build() + + try { + // Server holds [0.2N, N); client holds [0, 0.8N). Contiguous + // diff blocks (oldest 20% + newest 20%) — relayBench's shape. + val serverEvents = ((N * 2 / 10) until N).map { event(it) } + val clientEvents = (0 until (N * 8 / 10)).map { event(it) } + + val seedStart = System.nanoTime() + serverEvents.chunked(5000).forEach { store.batchInsert(it) } + val seedMs = (System.nanoTime() - seedStart) / 1e6 + + val expectedNeed = ((N * 8 / 10) until N).count() // server-only (newest 20%) + val expectedHave = (0 until (N * 2 / 10)).count() // client-only (oldest 20%) + + val driver = InteropSyncDriver(http) + val filter = Filter() + + val recStart = System.nanoTime() + val res = driver.negotiate(server.url, filter, clientEvents, timeoutMs = 120_000, maxRounds = 256) + val recMs = (System.nanoTime() - recStart) / 1e6 + + println("─ NegentropyServerReconcileBenchmark @ ${N / 1000}k (real geode server) ─") + println(" seed (${serverEvents.size} events): ${"%.0f".format(seedMs)} ms") + println(" rounds: ${res.rounds}") + println(" negotiate wall: ${"%.1f".format(recMs)} ms ← geode server + client + wire") + println(" need=${res.needIds.size} (exp $expectedNeed) have=${res.haveIds.size} (exp $expectedHave)") + println(" error: ${res.error}") + + assertEquals(null, res.error, "reconcile error") + assertEquals(expectedNeed, res.needIds.size, "need set") + assertEquals(expectedHave, res.haveIds.size, "have set") + } finally { + server.stop() + relay.close() + http.dispatcher.executorService.shutdown() + } + } +} diff --git a/quartz/build.gradle.kts b/quartz/build.gradle.kts index 74216db34b..efc1db5eb7 100644 --- a/quartz/build.gradle.kts +++ b/quartz/build.gradle.kts @@ -91,6 +91,12 @@ kotlin { (project.findProperty("prodRelayBench") as? String)?.let { environment("PROD_RELAY_BENCH", it) } + // Forward the negentropy-benchmark corpus size to the test JVM. + System.getProperty("negBenchN")?.let { systemProperty("negBenchN", it) } + // Opt-in JFR profiling of a benchmark run (-PnegProfile=/tmp/neg.jfr). + (project.findProperty("negProfile") as? String)?.let { + jvmArgs("-XX:+FlightRecorder", "-XX:StartFlightRecording=filename=$it,settings=profile,dumponexit=true") + } } tasks.withType().configureEach { diff --git a/quartz/plans/2026-07-04-negentropy-reconcile-profiling.md b/quartz/plans/2026-07-04-negentropy-reconcile-profiling.md new file mode 100644 index 0000000000..08eb96cb74 --- /dev/null +++ b/quartz/plans/2026-07-04-negentropy-reconcile-profiling.md @@ -0,0 +1,100 @@ +# NIP-77 reconcile profiling: where geode's sync time actually goes + +**Status: investigation + verified fix core (not yet wired).** Follow-up to +the 1M-event relayBench run (`relay.damus.io`, geode 1.12.6 vs strfry +v1-b80cda3), where geode lost the negentropy phase: initial reconcile +6,066 ms / 27 rounds vs strfry 1,270 ms / 14; identical-set reconcile +1,947 ms vs 557 ms (~3.5×). + +## What the numbers said, layer by layer + +Three benchmarks isolate each layer (all reproduce the relayBench slice +shape: 1M effective corpus, 80%/80% index slices, 60% overlap → a +**contiguous** oldest-20% + newest-20% diff, ~200k each way): + +| layer | tool | 1M / 200k-diff result | +|---|---|---| +| library reconcile only (no wire) | `quartz …prodbench.NegentropyReconcileBenchmark` | server **207 ms**, client 302 ms, 14 rounds, seal 331 ms | +| real geode server, one JVM, loopback ws | `geode …perf.NegentropyServerReconcileBenchmark` (`-DnegServerBench=1`) | negotiate **3,214 ms**, 14 rounds | +| two processes + real net | relayBench production | **6,066 ms**, 27 rounds | + +Key facts this establishes: + +1. **The reconciliation algorithm is not the bottleneck.** The pure + kmp-negentropy server loop is ~200 ms for the whole 14-round exchange. + My first cut of the library benchmark showed 22 s / 139 rounds — that + was a **benchmark bug**: slicing by array index while ids sort randomly + scatters the diff through sorted order (negentropy's worst case). Real + relayBench slices are contiguous *time* ranges; fixing the benchmark to + monotonic `created_at` dropped it to 200 ms / 14 rounds, matching + strfry's round count exactly. + +2. **The gap is geode's live server path, not framing.** Same 14 rounds, + same 500 KB frame cap as strfry, but 3,214 ms in-process (15× the + library loop) and 6,066 ms across processes. Rounds only inflate to 27 + in production from *seeding drift* (rejected events during seeding make + the stored sets differ from the clean slices → some scatter); strfry saw + less drift that run. + +3. **JFR of the real-geode run (server call-trees) splits as:** + - **~40% hex + UTF-8 + JSON serialization** of the hundred-KB hex NEG + payload each round (`Hex.encode`/`decode`, `UTF_8.encode`, Jackson). + - **~26% actual reconcile** (`StorageVector.forEach` + + `FingerprintCalculator.run` — the per-range fingerprint). + - rest: allocation (`MessageBuilder.branch`, `Arrays.copyOf`), one-time + seal/sort. + The client path is even heavier and dominated by `HashMap.putVal`/ + `resize` accumulating 400k have/need ids as hex strings. + +So geode-vs-strfry on sync is mostly the **JVM constant-factor tax on +serializing hex-in-JSON payloads**, which strfry pays in zero-copy C++ — +not a single fixable hotspot. `quartz.utils.Hex` is already an optimized +table-based codec, so there's no cheap win there. + +## The one algorithmic lever: prefix-sum fingerprints + +The ~26% reconcile slice *is* addressable, and it dominates the +steady-state identical-set reconcile (where geode loses 3.5×): every round +the library recomputes each range fingerprint with an **O(range) walk**. +Negentropy's fingerprint is `sha256( Σ id (mod 2²⁵⁶, 8 LE u32 limbs) ‖ +varint(count) )[0:16]` — the inner sum is **additive**, so a prefix-sum +table answers any range's raw sum in **O(1)** (limb-wise subtract with +borrow) + one sha256. + +`quartz …prodbench.NegentropyPrefixFingerprintTest` proves this: +- reproduces the library's fingerprint **bit-for-bit** over 2,000 random + ranges + boundaries at 50k; +- **460× faster per call** on a reconcile-shaped range mix (465 ms → 1.0 ms). + +For the top-of-tree fingerprints (each round re-walks ~all 800k ids) and +the identical-set case (16 full-corpus-ish fingerprints per reconcile), +this is the difference between an O(n) walk and a table lookup. + +## Why it isn't wired yet + +`com.vitorpamplona.negentropy.Negentropy` instantiates its own +`FingerprintCalculator` internally — there's no seam to inject a +prefix-sum-backed one from geode/quartz. Shipping it needs one of: + +1. **kmp-negentropy change** (cleanest): let `Negentropy` take a storage + that can answer `fingerprint(lo, hi)` itself, and have `StorageVector` + (or a new sealed storage) carry the prefix-sum table built at seal time. + `LiveNegentropyIndex` already keeps the sorted `(created_at, id)` set, so + the table is one extra pass at seal. +2. **quartz-side fast server**: reimplement the server reconcile against the + prefix-sum index. Larger and interop-critical (must match the wire byte + for byte with strfry) — the benchmarks + bit-exact test above are the + safety net for it. + +The serialization tax (~40%) is separate and only closes by writing the +hex payload straight into the output buffer as ASCII instead of +`bytes → hexString → JSON string → UTF-8 bytes`. + +## Artifacts (all landed here, gated/CI-safe) + +- `quartz …prodbench.NegentropyReconcileBenchmark` — library reconcile, + default 20k (fast CI correctness guard), `-DnegBenchN=1000000` for scale. +- `quartz …prodbench.NegentropyPrefixFingerprintTest` — the verified, + bit-exact prefix-sum core + per-call speedup (runs at 50k in CI). +- `geode …perf.NegentropyServerReconcileBenchmark` — real in-process geode + server, opt-in `-DnegServerBench=1`, JFR via `-PnegProfile=/path.jfr`. diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyPrefixFingerprintTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyPrefixFingerprintTest.kt new file mode 100644 index 0000000000..c5918c87d1 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyPrefixFingerprintTest.kt @@ -0,0 +1,208 @@ +/* + * 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.negentropy.fingerprint.FingerprintCalculator +import com.vitorpamplona.negentropy.storage.StorageVector +import com.vitorpamplona.quartz.utils.Hex +import com.vitorpamplona.quartz.utils.sha256.sha256 +import kotlin.random.Random +import kotlin.test.Test +import kotlin.test.assertContentEquals + +/** + * Validates the algorithmic fix for the one part of the NIP-77 server that + * profiling ([NegentropyReconcileBenchmark] + geode's server JFR) pinned as + * genuinely CPU-bound rather than serialization tax: the per-range + * fingerprint, which the reconciliation library recomputes from scratch — + * an O(range) walk over the storage — on **every** round. + * + * Negentropy's fingerprint is `sha256( Σ id (mod 2²⁵⁶, as 8 little-endian + * u32 limbs) ‖ varint(count) )[0:16]`. The inner sum is **additive**, so a + * prefix-sum table answers any range's raw sum in O(1) (limb-wise subtract + * with borrow), turning the top-of-tree fingerprints — which today re-walk + * hundreds of thousands of ids every round, and dominate the steady-state + * "identical sets" reconcile geode loses 3.5× on — into a constant-time + * lookup plus one sha256. + * + * This test proves the prefix-sum reproduces the library's fingerprint + * **bit-for-bit** over random ranges, and measures the per-call speedup. + * It does not wire the index into the wire path (the library instantiates + * its own [FingerprintCalculator]); it's the verified core for a + * kmp-negentropy change or a quartz-side fast server. + */ +class NegentropyPrefixFingerprintTest { + /** + * Prefix-sum fingerprint index over sorted ids. `prefix[k]` holds the + * 256-bit little-endian sum (8 u32 limbs) of the first `k` ids; a range + * sum is `prefix[hi] − prefix[lo]` limb-wise with borrow. Same limb math + * as the library's `FingerprintCalculator.add`, just accumulated once. + */ + class PrefixFingerprintIndex( + idsHex: List, + ) { + private val n = idsHex.size + + // 8 limbs per prefix position, row-major: prefix[k*8 + limb]. + private val prefix = LongArray((n + 1) * 8) + + init { + for (k in 0 until n) { + val id = Hex.decode(idsHex[k]) + var carry = 0L + for (limb in 0 until 8) { + val off = limb * 4 + val v = + (id[off].toLong() and 0xFF) or + ((id[off + 1].toLong() and 0xFF) shl 8) or + ((id[off + 2].toLong() and 0xFF) shl 16) or + ((id[off + 3].toLong() and 0xFF) shl 24) + val sum = (prefix[k * 8 + limb] and 0xFFFFFFFFL) + v + carry + prefix[(k + 1) * 8 + limb] = sum and 0xFFFFFFFFL + carry = sum ushr 32 + } + // Final carry out of limb 7 is dropped: sum is mod 2²⁵⁶. + } + } + + /** Fingerprint of `[lo, hi)` — O(1) sum + one sha256. */ + fun fingerprint( + lo: Int, + hi: Int, + ): ByteArray { + val buf = ByteArray(32) + var borrow = 0L + for (limb in 0 until 8) { + val diff = prefix[hi * 8 + limb] - prefix[lo * 8 + limb] - borrow + val v = diff and 0xFFFFFFFFL + borrow = if (diff < 0) 1 else 0 + val off = limb * 4 + buf[off] = (v and 0xFF).toByte() + buf[off + 1] = ((v shr 8) and 0xFF).toByte() + buf[off + 2] = ((v shr 16) and 0xFF).toByte() + buf[off + 3] = ((v shr 24) and 0xFF).toByte() + } + return sha256(buf + encodeVarInt(hi - lo)).copyOfRange(0, 16) + } + + /** Matches the library's message VarInt encoding for the count tag. */ + private fun encodeVarInt(n: Int): ByteArray { + if (n == 0) return byteArrayOf(0) + val limbs = ArrayList() + var num = n + while (num != 0) { + limbs.add(num and 127) + num = num ushr 7 + } + return ByteArray(limbs.size) { i -> + if (i == limbs.size - 1) { + limbs[limbs.size - 1 - i].toByte() + } else { + (limbs[limbs.size - 1 - i] or 128).toByte() + } + } + } + } + + private fun idFor(index: Int): String { + fun mix(seed: Long): Long { + var z = seed + -0x61c8864680b583ebL + z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L + z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L + return z xor (z ushr 31) + } + val hex = "0123456789abcdef" + val out = CharArray(64) + for (w in 0 until 4) { + val v = mix(index.toLong() * 4 + w) + for (b in 0 until 8) { + val byte = ((v ushr (b * 8)) and 0xFF).toInt() + out[(w * 8 + b) * 2] = hex[byte ushr 4] + out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF] + } + } + return String(out) + } + + @Test + fun prefixSumReproducesLibraryFingerprintBitForBit() { + val n = 50_000 + // Sorted so index order == storage order (ids compared as bytes). + val ids = (0 until n).map { idFor(it) }.sorted() + + val storage = StorageVector() + ids.forEachIndexed { i, id -> storage.insert(1_700_000_000L + i, id) } + storage.seal() + val library = FingerprintCalculator() + + val index = PrefixFingerprintIndex(ids) + val rnd = Random(42) + + // Ranges across every scale: tiny leaves, mid buckets, full corpus. + var checked = 0 + repeat(2000) { + val a = rnd.nextInt(n + 1) + val b = rnd.nextInt(n + 1) + val lo = minOf(a, b) + val hi = maxOf(a, b) + if (lo == hi) return@repeat + assertContentEquals( + library.run(storage, lo, hi).bytes, + index.fingerprint(lo, hi), + "fingerprint mismatch for [$lo,$hi)", + ) + checked++ + } + // A few whole-corpus and boundary ranges too. + assertContentEquals(library.run(storage, 0, n).bytes, index.fingerprint(0, n), "full range") + assertContentEquals(library.run(storage, 0, 1).bytes, index.fingerprint(0, 1), "first") + assertContentEquals(library.run(storage, n - 1, n).bytes, index.fingerprint(n - 1, n), "last") + + println("─ NegentropyPrefixFingerprint: verified $checked random ranges + boundaries at ${n / 1000}k ─") + + // Per-call speedup on a reconcile-shaped mix: the top of the tree + // fingerprints huge ranges (where prefix-sum wins most), leaves + // fingerprint tiny ranges. Weight toward the expensive large ranges. + val ranges = + buildList { + repeat(200) { add(0 to n) } // full-corpus (every round's top level) + repeat(400) { + val w = n / 16 + val lo = rnd.nextInt(n - w) + add(lo to lo + w) // 16-bucket split level + } + } + + var sink = 0 + val libStart = System.nanoTime() + for ((lo, hi) in ranges) sink = sink xor library.run(storage, lo, hi).bytes[0].toInt() + val libMs = (System.nanoTime() - libStart) / 1e6 + + val fastStart = System.nanoTime() + for ((lo, hi) in ranges) sink = sink xor index.fingerprint(lo, hi)[0].toInt() + val fastMs = (System.nanoTime() - fastStart) / 1e6 + + println(" ${ranges.size} reconcile-shaped range fingerprints:") + println(" library (O(range) walk): ${"%.1f".format(libMs)} ms") + println(" prefix-sum (O(1) + sha): ${"%.1f".format(fastMs)} ms") + println(" speedup: ${"%.1f".format(libMs / fastMs)}× (sink=$sink)") + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyReconcileBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyReconcileBenchmark.kt new file mode 100644 index 0000000000..ee828d9d52 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyReconcileBenchmark.kt @@ -0,0 +1,164 @@ +/* + * 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.filters.Filter +import com.vitorpamplona.quartz.nip01Core.store.IdAndTime +import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession +import kotlin.test.Test +import kotlin.test.assertEquals + +/** + * In-process reproduction of relayBench's NIP-77 initial-reconcile phase, + * with **no** network or JSON framing in the loop — so what it times is + * exactly the server-side reconciliation work that the head-to-head + * benchmark measured geode losing to strfry (1M corpus: geode 6,066 ms / + * 27 rounds vs strfry 1,270 ms / 14 rounds). + * + * The harness plays the unbounded initiator (`frameSizeLimit = 0`, like + * `SyncBenchmark.reconcile`); the [NegentropyServerSession] under test is + * the exact object geode's [com.vitorpamplona.quartz.nip01Core.relay.server.NegSessionRegistry] + * builds, at the same 500 KB frame cap. Server and client `processMessage` + * time is summed separately so the server's share is isolated. + * + * Not a CI assertion on speed (container noise) — it asserts convergence + * correctness and prints the breakdown. Size via `-DnegBenchN=1000000`. + */ +class NegentropyReconcileBenchmark { + companion object { + // Small by default so the correctness assertions run as a fast CI + // regression guard; scale to the relayBench shape with -DnegBenchN. + val N = System.getProperty("negBenchN")?.toInt() ?: 20_000 + const val FRAME_SIZE_LIMIT = 500_000L + + /** 2024-01-01. */ + const val BASE_TIME = 1_704_067_200L + + /** + * Events per second. `created_at` is monotonic with index so that + * sorted order == index order == a real chronological feed. That + * makes SyncBenchmark's index slices ([0,0.8N) vs [0.2N,N)) fall on + * *contiguous* time ranges — the diff is the oldest 20% + newest 20%, + * exactly like relayBench's slices, not a scatter that forces the + * reconciliation tree to split everywhere. + */ + const val EVENTS_PER_SECOND = 3 + } + + /** splitmix64 — well-distributed 64 pseudo-random bits from a counter. */ + private fun mix(seed: Long): Long { + var z = seed + -0x61c8864680b583ebL + z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L + z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L + return z xor (z ushr 31) + } + + private val hexChars = "0123456789abcdef".toCharArray() + + /** 64-char lowercase hex id from 4 mixed longs — distinct, unsorted. */ + private fun idFor(index: Int): String { + val out = CharArray(64) + for (w in 0 until 4) { + val v = mix(index.toLong() * 4 + w) + for (b in 0 until 8) { + val byte = ((v ushr (b * 8)) and 0xFF).toInt() + val pos = (w * 8 + b) * 2 + out[pos] = hexChars[byte ushr 4] + out[pos + 1] = hexChars[byte and 0xF] + } + } + return String(out) + } + + private fun entry(index: Int): IdAndTime { + // Monotonic time with same-second ties (broken by the random id), + // so index order matches negentropy's sorted order. + val createdAt = BASE_TIME + index / EVENTS_PER_SECOND + return IdAndTime(createdAt, idFor(index)) + } + + @Test + fun serverReconcileAtCorpusScale() { + // 80% / 80% slices with 60% overlap — SyncBenchmark's split. + // Server holds [0.2N, N); client (initiator) holds [0, 0.8N). + val clientEntries = (0 until (N * 8 / 10)).map { entry(it) } + val serverEntries = ((N * 2 / 10) until N).map { entry(it) } + + val clientIds = clientEntries.mapTo(HashSet()) { it.id } + val serverIds = serverEntries.mapTo(HashSet()) { it.id } + val expectedNeed = serverIds.count { it !in clientIds } // server has, client lacks + val expectedHave = clientIds.count { it !in serverIds } // client has, server lacks + + // Build (seal) time is measured separately — it's the NEG-OPEN cost, + // not per-round reconcile. + val sealStart = System.nanoTime() + val server = NegentropyServerSession("bench", serverEntries, FRAME_SIZE_LIMIT) + val sealMs = (System.nanoTime() - sealStart) / 1e6 + + val client = NegentropySession("bench", Filter(), clientEntries, frameSizeLimit = 0) + + var rounds = 0 + var wireBytes = 0L + var serverNanos = 0L + var clientNanos = 0L + val haveIds = HashSet() + val needIds = HashSet() + + val open = client.open() + var serverMsg: String? = open.initialMessage + + val loopStart = System.nanoTime() + while (serverMsg != null) { + wireBytes += serverMsg.length.toLong() + + val s0 = System.nanoTime() + val response = server.processMessage(serverMsg) + serverNanos += System.nanoTime() - s0 + + if (response == null) break // server produced nothing → done + wireBytes += response.message.length.toLong() + rounds++ + + val c0 = System.nanoTime() + val result = client.processMessage(response.message) + clientNanos += System.nanoTime() - c0 + + haveIds += result.haveIds + needIds += result.needIds + serverMsg = result.nextCmd?.message + } + val loopMs = (System.nanoTime() - loopStart) / 1e6 + + println("─ NegentropyReconcileBenchmark @ ${N / 1000}k (server ${serverEntries.size}, client ${clientEntries.size}) ─") + println(" rounds: $rounds") + println(" wire: ${"%.1f".format(wireBytes / 1024.0 / 1024.0)} MiB (hex NEG payloads, both directions)") + println(" avg frame fill: ${"%.0f".format(wireBytes.toDouble() / rounds / 1024)} KiB/round") + println(" seal (NEG-OPEN): ${"%.1f".format(sealMs)} ms") + println(" server reconcile: ${"%.1f".format(serverNanos / 1e6)} ms ← the number to beat") + println(" client reconcile: ${"%.1f".format(clientNanos / 1e6)} ms") + println(" loop wall: ${"%.1f".format(loopMs)} ms") + println(" need=${needIds.size} (exp $expectedNeed) have=${haveIds.size} (exp $expectedHave)") + + assertEquals(expectedNeed, needIds.size, "need set") + assertEquals(expectedHave, haveIds.size, "have set") + } +}