From 36571cfb94b5851532d519792aaa9d74b19d2709 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 5 Jul 2026 02:29:58 +0000 Subject: [PATCH] test(geode): sync-benchmark knobs + corpus-server tool for the negentropy comparison MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MirrorSyncThroughputTest: add -DsyncVerify (default true — `strfry sync` always verifies received events, so the negentropy sink verifies too for an apples-to-apples comparison) and -DsyncFts (default true; pass false to match strfry, which has no NIP-50). Both forwarded through the geode test task. - CorpusServerMain: a benchmark-only tool that boots a real geode relay (geode's default indexing) preloaded with an NDJSON corpus over a file-backed store and serves forever, so `strfry sync` / another geode / the negentropy sink can reconcile against a geode source holding the same 1M corpus a strfry source does. Used to run the 4-pair 1M negentropy sync comparison (geode↔geode, strfry→geode, strfry↔strfry, geode→strfry). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_012EZeWww5TJnzBZKPoc6mvU --- geode/build.gradle.kts | 2 + .../geode/tools/CorpusServerMain.kt | 101 ++++++++++++++++++ .../geode/mirror/MirrorSyncThroughputTest.kt | 15 ++- 3 files changed, 115 insertions(+), 3 deletions(-) create mode 100644 geode/src/main/kotlin/com/vitorpamplona/geode/tools/CorpusServerMain.kt diff --git a/geode/build.gradle.kts b/geode/build.gradle.kts index b2945b44c8..594458773f 100644 --- a/geode/build.gradle.kts +++ b/geode/build.gradle.kts @@ -78,6 +78,8 @@ tasks.withType().configureEach { System.getProperty("syncSourceUrl")?.let { systemProperty("syncSourceUrl", it) } System.getProperty("syncLiveIndex")?.let { systemProperty("syncLiveIndex", it) } System.getProperty("syncBackfillSeconds")?.let { systemProperty("syncBackfillSeconds", it) } + System.getProperty("syncFts")?.let { systemProperty("syncFts", it) } + System.getProperty("syncVerify")?.let { systemProperty("syncVerify", it) } maxHeapSize = System.getProperty("testHeap") ?: maxHeapSize // Opt-in JFR profiling (-PnegProfile=/tmp/neg.jfr). (project.findProperty("negProfile") as? String)?.let { diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/tools/CorpusServerMain.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/tools/CorpusServerMain.kt new file mode 100644 index 0000000000..935e81c334 --- /dev/null +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/tools/CorpusServerMain.kt @@ -0,0 +1,101 @@ +/* + * 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.tools + +import com.vitorpamplona.geode.KtorRelay +import com.vitorpamplona.geode.RelayEngine +import com.vitorpamplona.geode.RelayIndexingStrategy +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +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 kotlinx.coroutines.runBlocking +import java.io.File +import java.util.concurrent.CountDownLatch + +/** + * Boots a real geode relay (geode's default indexing: FTS + live negentropy + * index) preloaded with a corpus of nostr events read from an NDJSON file, then + * serves forever so external clients — `strfry sync`, another geode, the + * relayBench negentropy sink — can reconcile against it. + * + * A benchmark-only source: it exists so the negentropy sync comparison has a + * geode relay holding the same corpus a strfry source does, reachable over the + * production WebSocket transport. + * + * Usage: `CorpusServerMain [maxCount]` + */ +fun main(args: Array) { + if (args.size < 2) { + System.err.println("usage: CorpusServerMain [maxCount]") + return + } + val port = args[0].toInt() + val corpus = File(args[1]) + val maxCount = args.getOrNull(2)?.toInt() ?: Int.MAX_VALUE + + // File-backed so a 1M in-memory corpus here doesn't compete for RAM with an + // in-memory sink in the same box during the comparison. Fresh each boot. + val dbFile = "/tmp/geode-source-$port.sqlite" + listOf("", "-wal", "-shm").forEach { File(dbFile + it).delete() } + val store = EventStore(dbName = dbFile, indexStrategy = RelayIndexingStrategy) + val engine = RelayEngine(url = "ws://127.0.0.1:$port/".normalizeRelayUrl(), store = store) + + println("CorpusServerMain: loading up to $maxCount events from ${corpus.name}…") + val loaded = + runBlocking { + var total = 0 + val batch = ArrayList(10_000) + corpus.bufferedReader().useLines { lines -> + for (line in lines) { + if (total >= maxCount) break + if (line.isBlank()) continue + val event = runCatching { OptimizedJsonMapper.fromJson(line) }.getOrNull() ?: continue + batch.add(event) + if (batch.size == 10_000) { + store.batchInsert(batch) + total += batch.size + batch.clear() + if (total % 200_000 == 0) println(" …loaded $total") + } + } + } + if (batch.isNotEmpty()) { + store.batchInsert(batch) + total += batch.size + } + total + } + val count = runBlocking { store.count(Filter()) } + + val server = KtorRelay(engine, host = "127.0.0.1", port = port).start() + println("CorpusServerMain: READY port=$port loaded=$loaded distinct=$count") + + // Serve until the JVM is killed. + Runtime.getRuntime().addShutdownHook( + Thread { + server.stop(gracePeriodMillis = 0, timeoutMillis = 500) + engine.close() + }, + ) + CountDownLatch(1).await() +} diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorSyncThroughputTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorSyncThroughputTest.kt index f115d58667..4df0baf4b5 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorSyncThroughputTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorSyncThroughputTest.kt @@ -89,11 +89,20 @@ class MirrorSyncThroughputTest { // `-DsyncLiveIndex=false` disables the live negentropy index to isolate its // O(n)-insert cost during out-of-order backfill. private val liveIndex = System.getProperty("syncLiveIndex")?.toBoolean() ?: true + + // NIP-50 full-text search (geode's default; strfry has none) — pass + // `-DsyncFts=false` to match strfry for an apples-to-apples sync. + private val fts = System.getProperty("syncFts")?.toBoolean() ?: true + + // Schnorr verification on the sink's ingest. `strfry sync` always verifies + // received events, so the negentropy sink defaults to verifying too; pass + // `-DsyncVerify=false` for the trusted-mirror (skip-verify) rate. + private val verifySink = System.getProperty("syncVerify")?.toBoolean() ?: true private val strategy = DefaultIndexingStrategy( indexEventsByCreatedAtAlone = true, indexEventsByPubkeyAlone = true, - indexFullTextSearch = true, + indexFullTextSearch = fts, deferFullTextSearchIndexing = true, maintainLiveNegentropyIndex = liveIndex, ) @@ -252,7 +261,7 @@ class MirrorSyncThroughputTest { val consumer = launch { for (ev in handoff) { - downstream.server.ingest(ev, skipVerify = true) { } + downstream.server.ingest(ev, skipVerify = !verifySink) { } } } @@ -407,7 +416,7 @@ class MirrorSyncThroughputTest { val raw = withTimeout(180_000) { incoming.receive() } if (raw.startsWith("[\"EVENT\",\"$subId\"")) { val ev = (OptimizedJsonMapper.fromJsonToMessage(raw) as? EventMessage)?.event ?: continue - downstream.server.ingest(ev, skipVerify = true) { } + downstream.server.ingest(ev, skipVerify = !verifySink) { } } else if (raw.startsWith("[\"EOSE\",\"$subId\"") || raw.startsWith("[\"CLOSED\",\"$subId\"")) { ws.send("""["CLOSE","$subId"]""") break