From 6095e5768e245f6d9e77a4a087bcbf4c6acf9aea Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 21:57:49 +0000 Subject: [PATCH] =?UTF-8?q?test(sync):=201M=20sync-throughput=20harness=20?= =?UTF-8?q?for=20strfry=E2=86=94strfry,=20geode=E2=86=94geode,=20strfry?= =?UTF-8?q?=E2=86=92geode?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_012EZeWww5TJnzBZKPoc6mvU --- .../geode/mirror/MirrorSyncThroughputTest.kt | 173 ++++++++++++++++++ relayBench/sync-throughput-strfry.sh | 43 +++++ 2 files changed, 216 insertions(+) create mode 100644 geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorSyncThroughputTest.kt create mode 100755 relayBench/sync-throughput-strfry.sh diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorSyncThroughputTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorSyncThroughputTest.kt new file mode 100644 index 0000000000..92b8496500 --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorSyncThroughputTest.kt @@ -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(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("════════════════════════════════════════════") + } +} diff --git a/relayBench/sync-throughput-strfry.sh b/relayBench/sync-throughput-strfry.sh new file mode 100755 index 0000000000..6854a0074d --- /dev/null +++ b/relayBench/sync-throughput-strfry.sh @@ -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 +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" < "$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