test(geode): sync-benchmark knobs + corpus-server tool for the negentropy comparison

- 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012EZeWww5TJnzBZKPoc6mvU

geode/build.gradle.kts

@@ -78,6 +78,8 @@ tasks.withType<Test>().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 {

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 <port> <corpus.ndjson> [maxCount]`
+ */
+fun main(args: Array<String>) {
+    if (args.size < 2) {
+        System.err.println("usage: CorpusServerMain <port> <corpus.ndjson> [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<Event>(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()
+}

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