test(sync): 1M sync-throughput harness for strfry↔strfry, geode↔geode, strfry→geode

Two pieces measuring how fast each sink pulls a large corpus from a source using its native sync client: - relayBench/sync-throughput-strfry.sh: `strfry import` N events into a source strfry, boot it, `strfry sync --dir down` an empty sink, report events/s. - MirrorSyncThroughputTest: geode downstream pulls via the real MirrorWorker (WebSocket). Default in-process geode source (geode→geode); with -DsyncSourceUrl it mirrors an external relay (e.g. strfry) for strfry→geode. Sized by -DsyncN. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012EZeWww5TJnzBZKPoc6mvU
2026-07-30 19:46:17 +00:00 · 2026-07-04 21:57:49 +00:00
parent 998e5fc422
commit 6095e5768e
2 changed files with 216 additions and 0 deletions
@@ -0,0 +1,173 @@
+/*
+ * 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.mirror
+
+import com.vitorpamplona.geode.KtorRelay
+import com.vitorpamplona.geode.RelayEngine
+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.TimeUtils
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.test.AfterTest
+import kotlin.test.Test
+
+/**
+ * Measures geode's mirror sync throughput (events/second) pulling a large set
+ * into an empty downstream geode over the real WebSocket transport.
+ *
+ * Source selection:
+ *  - default (`-DsyncSourceUrl` unset): an **in-process geode** upstream
+ *    ([KtorRelay]) preloaded with N events — the **geode→geode** number.
+ *  - `-DsyncSourceUrl=ws://host:port`: mirror that external relay (e.g. a strfry
+ *    loaded with the corpus) — the **strfry→geode** number. `-DsyncExpect=N`
+ *    sets the convergence target.
+ *
+ * Size with `-DsyncN` (default 1,000,000). Timed from worker start to the
+ * downstream reaching the target (or plateauing).
+ */
+class MirrorSyncThroughputTest {
+    private val downstreamStore = EventStore(null)
+    private val downstream =
+        RelayEngine(url = "ws://127.0.0.1:7899/".normalizeRelayUrl(), store = downstreamStore, parallelVerify = true)
+
+    private var upstreamStore: EventStore? = null
+    private var upstream: RelayEngine? = null
+    private var server: KtorRelay? = null
+    private var worker: MirrorWorker? = null
+
+    @AfterTest
+    fun tearDown() {
+        worker?.close()
+        server?.stop(gracePeriodMillis = 0, timeoutMillis = 1_000)
+        upstream?.close()
+        downstream.close()
+    }
+
+    private val hex = "0123456789abcdef"
+
+    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 fun hex64(
+        salt: Long,
+        index: Int,
+    ): String {
+        val out = CharArray(64)
+        for (w in 0 until 4) {
+            val v = mix(salt * 1_000_003 + 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 mirrorSyncThroughput() =
+        runBlocking {
+            val n = System.getProperty("syncN")?.toInt() ?: 1_000_000
+            val externalUrl = System.getProperty("syncSourceUrl")
+            val expect = System.getProperty("syncExpect")?.toInt() ?: n
+
+            val sourceUrl: String
+            if (externalUrl != null) {
+                sourceUrl = externalUrl
+                println("─ MirrorSyncThroughput: mirroring EXTERNAL source $externalUrl, expect $expect ─")
+            } else {
+                val store = EventStore(null).also { upstreamStore = it }
+                val now = TimeUtils.now()
+                val sig = "f".repeat(128)
+                val batch = ArrayList<Event>(10_000)
+                var loaded = 0
+                for (i in 0 until n) {
+                    batch.add(
+                        Event(
+                            id = hex64(7, i),
+                            pubKey = hex64(3, i % 20_000),
+                            createdAt = now - 60 - (i % 20_000),
+                            kind = 1,
+                            tags = emptyArray(),
+                            content = "e$i",
+                            sig = sig,
+                        ),
+                    )
+                    if (batch.size == 10_000) {
+                        store.batchInsert(batch)
+                        loaded += batch.size
+                        batch.clear()
+                    }
+                }
+                if (batch.isNotEmpty()) store.batchInsert(batch)
+                val eng = RelayEngine(url = "ws://127.0.0.1:7898/".normalizeRelayUrl(), store = store).also { upstream = it }
+                server = KtorRelay(eng, host = "127.0.0.1", port = 7898).start()
+                sourceUrl = "ws://127.0.0.1:7898/"
+                println("─ MirrorSyncThroughput: geode→geode, in-process source preloaded $loaded ─")
+            }
+
+            val startNanos = System.nanoTime()
+            worker =
+                MirrorWorker(
+                    upstreams =
+                        listOf(
+                            MirrorUpstream(
+                                url = sourceUrl.normalizeRelayUrl(),
+                                trusted = true,
+                                backfillSeconds = 90_000,
+                            ),
+                        ),
+                    server = downstream.server,
+                ).also { it.start() }
+
+            var reached = 0
+            var last = -1
+            var stable = 0
+            withTimeoutOrNull(600_000) {
+                while (true) {
+                    val c = downstreamStore.count(Filter())
+                    reached = c
+                    if (c >= expect) break
+                    if (c == last) {
+                        if (++stable >= 30) break // ~15s no progress
+                    } else {
+                        stable = 0
+                        last = c
+                    }
+                    delay(500)
+                }
+            }
+            val secs = (System.nanoTime() - startNanos) / 1e9
+            val eps = (reached / secs).toLong()
+            val src = if (externalUrl != null) "external" else "geode"
+            println("════════════════════════════════════════════")
+            println("$src→geode: synced $reached/$expect in %.1fs  =>  %,d events/s".format(secs, eps))
+            println("════════════════════════════════════════════")
+        }
+}
@@ -0,0 +1,43 @@
+#!/usr/bin/env bash
+# Measures strfry's native sync throughput: import N events into a source
+# strfry, boot it as a relay, then `strfry sync --dir down` an empty sink from
+# it and report events/second. Usage: sync-throughput-strfry.sh <N>
+set -euo pipefail
+N="${1:-1000000}"
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+STRFRY="${STRFRY_BIN:-$HERE/.cache/strfry/strfry}"
+CORPUS="${CORPUS:-$HERE/.corpus-cache/corpus-download-relay.damus.io-n1000000.ndjson}"
+WORK="${WORK:-/tmp/strfry-sync-$N}"
+PORT="${PORT:-18821}"
+
+rm -rf "$WORK"; mkdir -p "$WORK/src-db" "$WORK/dst-db"
+mkconf() { cat > "$1" <<EOF
+db = "$2"
+events { maxEventSize = 1048576 rejectEventsOlderThanSeconds = 3155760000 maxNumTags = 10000 }
+relay { bind = "127.0.0.1" port = $3 nofiles = 0 maxWebsocketPayloadSize = 1114112 }
+EOF
+}
+mkconf "$WORK/src.conf" "$WORK/src-db" "$PORT"
+mkconf "$WORK/dst.conf" "$WORK/dst-db" "$((PORT+1))"
+
+echo "importing $N events into strfry source…"
+head -n "$N" "$CORPUS" | "$STRFRY" --config="$WORK/src.conf" import > "$WORK/import.log" 2>&1
+srcN=$("$STRFRY" --config="$WORK/src.conf" export 2>/dev/null | wc -l)
+echo "source has $srcN events"
+
+"$STRFRY" --config="$WORK/src.conf" relay > "$WORK/src.log" 2>&1 &
+SRCPID=$!
+trap 'kill $SRCPID 2>/dev/null || true' EXIT
+sleep 3
+
+echo "syncing sink from source (strfry sync --dir down)…"
+t0=$(date +%s.%N)
+"$STRFRY" --config="$WORK/dst.conf" sync "ws://127.0.0.1:$PORT" --dir down --timeout 120 > "$WORK/sync.log" 2>&1
+t1=$(date +%s.%N)
+dstN=$("$STRFRY" --config="$WORK/dst.conf" export 2>/dev/null | wc -l)
+secs=$(echo "$t1 - $t0" | bc)
+eps=$(echo "scale=0; $dstN / $secs" | bc)
+echo "════════════════════════════════════════════"
+echo "strfry→strfry: synced $dstN/$srcN in ${secs}s  =>  $eps events/s"
+echo "════════════════════════════════════════════"
+grep -iE "reconcile complete|Have [0-9]" "$WORK/sync.log" | grep -viE "arguments|Current dir" | tail -3