diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 8a8408fd51..2a019ff766 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -13,7 +13,10 @@ a non-interactive JVM command-line client that drives the same `quartz` + `commo humans, agents, and interop tests. `quic` is a from-scratch pure-Kotlin QUIC v1 + HTTP/3 + WebTransport client (no JNI, no BouncyCastle), built because no Android-compatible Java QUIC library exists. `geode` is a standalone JVM Nostr relay (Ktor) built on quartz's -relay-server code; smaller modules are `benchmark` (Android macrobenchmarks) and +relay-server code; smaller modules are `benchmark` (Android macrobenchmarks), +`relayBench` (head-to-head relay benchmark — boots geode, strfry and other +relay binaries, replays a shared deterministic corpus, measures ingest/query/ +NIP-77 sync; `./relayBench/run.sh`, see `relayBench/README.md`) and `quic-interop` (QUIC interop runner, lives at `quic/interop`). `nestsClient` runs the audio-room protocol on top of `:quic` for the NIP-53 audio-rooms feature. It implements both IETF `draft-ietf-moq-transport-17` (under `moq/`) and **moq-lite Lite-03** (kixelated's variant, under `moq/lite/`); the diff --git a/relayBench/.gitignore b/relayBench/.gitignore new file mode 100644 index 0000000000..d80aa7e4d6 --- /dev/null +++ b/relayBench/.gitignore @@ -0,0 +1,3 @@ +.cache/ +.corpus-cache/ +results/ diff --git a/relayBench/README.md b/relayBench/README.md new file mode 100644 index 0000000000..e0a2a822e4 --- /dev/null +++ b/relayBench/README.md @@ -0,0 +1,111 @@ +# relayBench + +Head-to-head benchmark for Nostr relay implementations. Boots each relay as a +real external process on loopback under an equivalent setup — persistent +storage, signature verification on, no auth, stock limits — replays the same +event corpus into each, and renders a side-by-side report. + +```bash +./relayBench/run.sh # geode vs strfry, 10k-event synthetic corpus +./relayBench/run.sh --quick # 2k-event smoke run +./relayBench/run.sh --real # replay the checked-in real-event dump (2024, ~30k events) +``` + +`run.sh` builds `:geode:installDist` and the harness, and resolves strfry from +`$STRFRY_BIN`, the `PATH`, or by building it from source into +`relayBench/.cache/` (first run only). `SKIP_STRFRY=1` / `SKIP_GEODE=1` skip a +side. + +## What is measured + +**Ingest — receipt ➜ queryable.** One connection publishes an event; from the +same instant a second connection hammer-polls `REQ {"ids":[id]}` until the +event comes back. This measures exactly "how long after the relay receives an +event can a REQ return it", which is not the same thing as the OK ack — the +report also shows OK latency and what fraction of events were already +queryable when their OK arrived. + +**Ingest — throughput.** The corpus is replayed over N connections (default 4) +with a bounded number of unacked EVENTs in flight, wall-clocked from first +send to last OK. Accepted/rejected counts come from the OKs. + +**Queries.** Filters modeled on what real clients send, derived from the +corpus itself so they hit meaningful data: global feed, profile hydration, +home feed (150 follows), hottest thread, notifications for the most-mentioned +pubkey, hashtag feed, 100-id batch fetch, recent time window. Each runs +warmup + measured rounds on one connection (time-to-first-event / time-to-EOSE +percentiles) and then from 8 connections at once (aggregate events/second). +All filters stay inside strfry's default limits, and the number of events each +relay returns is cross-checked — a ⚠ in the report means the relays disagree +about the result set, which invalidates the speed comparison for that row. + +**NIP-77 negentropy sync — every pair of relays.** Both sides get an 80% slice +of the corpus (60% overlap); the harness plays the `strfry sync` role with one +side's dataset as its local set and measures: initial reconciliation against +each relay as server (time, NEG-MSG rounds, wire bytes), the delta transfer to +convergence, and the steady-state reconcile of identical sets. Convergence is +verified, so this doubles as an interop test. + +**Storage.** On-disk footprint after full ingest (LMDB vs SQLite vs whatever). + +## Corpora + +The corpus is the controlled variable: every relay sees the same events in +the same order, and reports carry a `fingerprint` (sha256 over event ids) so +two runs are comparable only when fingerprints match. + +| source | flag | notes | +|---|---|---| +| **synthetic** (default) | `--events N --seed S` | Deterministic to the byte: seeded keys, fixed timestamps, seed-derived BIP-340 nonces. Same spec ⇒ identical NDJSON on any machine. Zipf-popularity authors, threads, reactions, reposts, zap pairs, hashtags. | +| **real dump** | `--real` | The `quartz` test fixture `nostr_vitor_startup_data.json.gz` — ~31k unique real events from 2024 with a rich kind mix (notes, chats, DMs, zaps, reports, communities). | +| **contact lists** | `--corpus contact-lists.gz --limit 100000 --max-event-bytes 1048576 --max-tags 20000` | 2.1M real kind-3 contact lists (heavy events, ~1.3 kB avg, up to 100+ kB). Grab it with `pip install gdown && gdown 1yyC93xY9sDsEsa351ZAMhtAXwBUh3LYT`. Raising the size/tag caps reconfigures strfry to match, so both relays still accept the full stream. | +| **any dump** | `--corpus FILE` | NDJSON or a single JSON array, gzipped or plain (sniffed by magic bytes). | +| **fresh download** | `--download [urls]` | Pages recent events out of public relays (damus/nos.lol/primal by default). | + +Every source goes through the same preparation: dedup by id, drop unsigned +events (NIP-17 rumors), kind-5 deletions and ephemerals (order-dependent or +unqueryable — they would make relays disagree for reasons unrelated to +performance), drop events over the size/tag caps, verify every Schnorr +signature in parallel, sort chronologically. The prepared corpus is cached in +`relayBench/.corpus-cache/` as NDJSON next to a `manifest.json` with the +fingerprint and kind histogram — that file pair is a shareable, citable +benchmark artifact. + +> **Why this corpus matters:** there is no de-facto community benchmark corpus +> today. Existing relay benchmarks (rnostr's and privkeyio's nostr-bench, +> mattn's scripts) each synthesize their own events with unspecified +> distributions, so published numbers aren't reproducible corpus-controlled; +> the only shared dataset (Wellorder's early-1m) is frozen in January 2023. +> relayBench's synthetic spec ("seed 1, n=10000, v1" ⇒ byte-identical corpus) +> and manifest/fingerprint convention are designed so other relay authors can +> run the exact same workload and publish comparable numbers. + +## Adding another relay + +No code needed if the relay can be launched from a command line: + +```bash +./relayBench/run.sh --relay 'nostr-rs-relay=/usr/bin/nostr-rs-relay --db {dir} --port {port}' +``` + +`{port}` and `{dir}` are substituted at launch; the process must listen on +`127.0.0.1:{port}` with persistent storage under `{dir}`, verification on and +no auth. If the relay needs a config file, point the template at a small +wrapper script that writes one (see `StrfryRelay` in +`relays/RelayUnderTest.kt` for the pattern — adding a first-class subclass is +~20 lines). + +## Output + +The terminal report shows each metric as name / bar / value rows with the +winner starred, followed by a head-to-head summary. Every run also writes: + +- `relayBench/results//report.md` — shareable Markdown +- `relayBench/results//results.json` — raw numbers for tooling + +## Direct harness invocation + +`run.sh` is a thin wrapper; the harness itself is +`relayBench/build/install/relaybench/bin/relaybench` — see `--help` for all +options (`--samples`, `--publishers`, `--window`, `--query-rounds`, +`--query-conns`, `--no-sync`, `--out`, `--keep-data`, …). diff --git a/relayBench/build.gradle.kts b/relayBench/build.gradle.kts new file mode 100644 index 0000000000..1dccf54b54 --- /dev/null +++ b/relayBench/build.gradle.kts @@ -0,0 +1,38 @@ +import org.jetbrains.kotlin.gradle.dsl.JvmTarget + +plugins { + alias(libs.plugins.jetbrainsKotlinJvm) + application +} + +application { + mainClass.set("com.vitorpamplona.relaybench.MainKt") + applicationName = "relaybench" + // Percentile latencies get skewed by GC pauses in the *client* — give the + // harness enough heap that it never becomes the bottleneck being measured. + applicationDefaultJvmArgs = listOf("-Xmx2g") +} + +kotlin { + jvmToolchain(21) + compilerOptions { + jvmTarget.set(JvmTarget.JVM_21) + } +} + +dependencies { + // Event model, Schnorr signing, Filter + wire-format JSON. The harness + // reuses quartz so the corpus is byte-identical to what Amethyst emits. + implementation(project(":quartz")) + + implementation(libs.kotlinx.coroutines.core) + implementation(libs.jackson.module.kotlin) + implementation(libs.okhttp) + + // JNI secp256k1 backend quartz needs at runtime on plain JVM. + runtimeOnly(libs.secp256k1.kmp.jni.jvm) + + testImplementation(libs.kotlin.test) + testImplementation(libs.kotlinx.coroutines.test) + testImplementation(libs.secp256k1.kmp.jni.jvm) +} diff --git a/relayBench/run.sh b/relayBench/run.sh new file mode 100755 index 0000000000..a01fb758c9 --- /dev/null +++ b/relayBench/run.sh @@ -0,0 +1,98 @@ +#!/usr/bin/env bash +# +# relaybench — one-command Nostr relay benchmark (geode vs strfry vs anything). +# +# ./relayBench/run.sh # deterministic 10k synthetic corpus +# ./relayBench/run.sh --quick # 2k-event smoke run +# ./relayBench/run.sh --real # replay the checked-in real-event dump +# ./relayBench/run.sh --corpus f.gz --limit 50000 +# ./relayBench/run.sh --relay 'myrelay=/path/bin --port {port} --db {dir}' +# +# strfry resolution order: $STRFRY_BIN → `strfry` on PATH → build from +# source into relayBench/.cache (needs the deps listed in the error text). +# Skip strfry entirely with SKIP_STRFRY=1; skip geode with SKIP_GEODE=1. +set -euo pipefail + +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +ROOT="$(dirname "$HERE")" +CACHE="$HERE/.cache" + +ARGS=() +for arg in "$@"; do + case "$arg" in + --real) + ARGS+=("--corpus" "$ROOT/quartz/src/commonTest/resources/nostr_vitor_startup_data.json.gz") + ;; + *) + ARGS+=("$arg") + ;; + esac +done + +# ---------- strfry ---------- +resolve_strfry() { + if [[ -n "${STRFRY_BIN:-}" ]]; then + echo "$STRFRY_BIN" + return + fi + if command -v strfry > /dev/null 2>&1; then + command -v strfry + return + fi + if [[ -x "$CACHE/strfry/strfry" ]]; then + echo "$CACHE/strfry/strfry" + return + fi + echo "building strfry from source (first run only — a few minutes)…" >&2 + mkdir -p "$CACHE" + if [[ ! -d "$CACHE/strfry" ]]; then + git clone --depth 1 https://github.com/hoytech/strfry.git "$CACHE/strfry" >&2 + fi + ( + cd "$CACHE/strfry" + git submodule update --init --depth 1 >&2 + make setup-golpe >&2 + make -j"$(nproc)" >&2 + ) || { + cat >&2 << 'EOF' + +strfry build failed. On Debian/Ubuntu install its deps with: + + sudo apt install build-essential libyaml-perl libtemplate-perl \ + libregexp-grammars-perl libssl-dev zlib1g-dev liblmdb-dev \ + libflatbuffers-dev libsecp256k1-dev libzstd-dev + +or point STRFRY_BIN at an existing binary, or SKIP_STRFRY=1 to bench without it. +EOF + return 1 + } + echo "$CACHE/strfry/strfry" +} + +STRFRY="" +if [[ "${SKIP_STRFRY:-0}" != "1" ]]; then + STRFRY="$(resolve_strfry)" || exit 1 + echo "strfry: $STRFRY" +fi + +# ---------- build geode + harness ---------- +echo "building geode + relaybench…" +GRADLE_TASKS=(":relayBench:installDist") +if [[ "${SKIP_GEODE:-0}" != "1" ]]; then + GRADLE_TASKS+=(":geode:installDist") +fi +(cd "$ROOT" && ./gradlew -q "${GRADLE_TASKS[@]}") + +RELAY_FLAGS=() +if [[ "${SKIP_GEODE:-0}" != "1" ]]; then + RELAY_FLAGS+=("--geode-bin" "$ROOT/geode/build/install/geode/bin/geode") +fi +if [[ -n "$STRFRY" ]]; then + RELAY_FLAGS+=("--strfry-bin" "$STRFRY") +fi + +# ---------- run ---------- +cd "$ROOT" +exec "$ROOT/relayBench/build/install/relaybench/bin/relaybench" \ + "${RELAY_FLAGS[@]}" \ + "${ARGS[@]}" diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Main.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Main.kt new file mode 100644 index 0000000000..26b3fedb6e --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Main.kt @@ -0,0 +1,343 @@ +/* + * 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.relaybench + +import com.vitorpamplona.relaybench.bench.BenchmarkRunner +import com.vitorpamplona.relaybench.corpus.Corpus +import com.vitorpamplona.relaybench.corpus.CorpusDownloader +import com.vitorpamplona.relaybench.corpus.CorpusSource +import com.vitorpamplona.relaybench.corpus.CorpusSpec +import com.vitorpamplona.relaybench.relays.CustomRelay +import com.vitorpamplona.relaybench.relays.GeodeRelay +import com.vitorpamplona.relaybench.relays.RelayUnderTest +import com.vitorpamplona.relaybench.relays.StrfryRelay +import okhttp3.Dispatcher +import okhttp3.OkHttpClient +import okhttp3.Protocol +import java.io.File +import java.time.ZonedDateTime +import java.time.format.DateTimeFormatter +import java.util.concurrent.TimeUnit +import kotlin.system.exitProcess + +/** + * relaybench — head-to-head Nostr relay benchmark. + * + * Boots each relay implementation as a real external process on loopback + * with equivalent setups (persistent storage, signature verification on, + * no auth, stock limits), replays the same corpus into each, and measures: + * + * - insertion: receipt➜queryable-by-REQ latency, OK-ack latency, and + * corpus replay throughput; + * - queries: popular client filters (home feed, thread, notifications, + * profiles, hashtag, by-ids, …) — time-to-EOSE and events/second, + * single-client and concurrent; + * - NIP-77 negentropy sync between every pair of relays. + * + * Normally driven via `relayBench/run.sh`, which builds the binaries and + * passes `--geode-bin`/`--strfry-bin` here. + */ +class Options( + val geodeBin: String?, + val strfryBin: String?, + val customRelays: List>, + val events: Int, + val seed: Long, + val baseTime: Long, + val corpusFile: File?, + val downloadFrom: List?, + val limit: Int, + val maxEventBytes: Int, + val maxTags: Int, + val visibilitySamples: Int, + val publishers: Int, + val window: Int, + val warmupRounds: Int, + val queryRounds: Int, + val queryConnections: Int, + val concurrentRounds: Int, + val skipSync: Boolean, + val outDir: File, + val workDir: File, + val cacheDir: File, + val keepData: Boolean, +) + +private val USAGE = + """ + relaybench — benchmark Nostr relay implementations head to head. + + Relays (at least one): + --geode-bin PATH geode launcher (from :geode:installDist) + --strfry-bin PATH strfry binary + --relay NAME=CMD any other relay; CMD may use {port} and {dir} + (repeatable) + + Corpus (default: deterministic synthetic): + --events N synthetic corpus size [10000] + --seed N synthetic corpus seed [1] + --base-time EPOCH|now newest synthetic timestamp [fixed 2026-06-01] + --corpus FILE NDJSON or JSON-array dump, optionally gzipped + --limit N cap corpus events after preparation + --download [URLS] build corpus from public relays (comma-separated, + defaults to damus/nos.lol/primal) + --max-event-bytes N event size ceiling (corpus filter + relay config) + [65536] + --max-tags N tag count ceiling [2000] + + Benchmark shape: + --samples N visibility (receipt➜REQ) samples [200] + --publishers N ingest connections [4] + --window N unacked EVENTs in flight/conn [200] + --query-rounds N measured rounds per scenario [10] + --query-conns N concurrent query connections [8] + --no-sync skip the pairwise NIP-77 sync phase + --quick small corpus + fewer rounds (smoke test) + + Output: + --out DIR report directory [relayBench/results] + --work DIR scratch dir for relay data [temp] + --keep-data keep relay data dirs after the run + --no-color plain terminal output + """.trimIndent() + +fun main(args: Array) { + val options = + parseArgs(args) ?: run { + println(USAGE) + exitProcess(2) + } + + val http = + OkHttpClient + .Builder() + .protocols(listOf(Protocol.HTTP_1_1)) + .socketFactory(NostrSocket.NO_DELAY_SOCKETS) + .readTimeout(0, TimeUnit.MILLISECONDS) + .apply { + val dispatcher = Dispatcher() + dispatcher.maxRequests = 10_000 + dispatcher.maxRequestsPerHost = 10_000 + dispatcher(dispatcher) + }.build() + + val log: (String) -> Unit = { println(it) } + + val relays = buildRelays(options) + if (relays.isEmpty()) { + System.err.println("No relays configured. Pass --geode-bin, --strfry-bin or --relay.\n") + println(USAGE) + exitProcess(2) + } + + println() + println("── corpus ──────────────────────────────────────────────") + val corpus = loadCorpus(options, http, log) + if (corpus.events.isEmpty()) { + System.err.println("Corpus is empty — nothing to benchmark.") + exitProcess(1) + } + val scenarios = Scenarios.derive(corpus) + println(" ${corpus.events.size} events ready (fingerprint ${corpus.fingerprint}); ${scenarios.size} query scenarios derived") + + val workDir = options.workDir.apply { mkdirs() } + val startedAt = ZonedDateTime.now() + + val results = + relays.map { relay -> + println() + println("── ${relay.name} ─────────────────────────────────────────") + BenchmarkRunner.runSingle(relay, corpus, scenarios, options, workDir, http, log) + } + + val syncs = + if (options.skipSync || relays.size < 2) { + emptyList() + } else { + println() + println("── NIP-77 sync pairs ──────────────────────────────────") + BenchmarkRunner.runSyncPairs(relays, corpus, options, workDir, http, log) + } + + val run = BenchRun(startedAt, corpus, options, results, syncs) + + // Reports. + val stamp = startedAt.format(DateTimeFormatter.ofPattern("yyyy-MM-dd-HHmmss")) + val outDir = File(options.outDir, stamp).apply { mkdirs() } + File(outDir, "report.md").writeText(Report.markdown(run)) + File(outDir, "results.json").writeText(Report.json(run)) + + println(Report.terminal(run)) + println(" full report: ${File(outDir, "report.md").path}") + println(" raw data: ${File(outDir, "results.json").path}") + println() + + if (!options.keepData) workDir.deleteRecursively() + http.dispatcher.executorService.shutdown() + http.connectionPool.evictAll() + + if (results.any { it.error != null }) exitProcess(1) +} + +private fun buildRelays(options: Options): List = + buildList { + options.geodeBin?.let { add(GeodeRelay(it)) } + options.strfryBin?.let { add(StrfryRelay(it, options.maxEventBytes, options.maxTags)) } + options.customRelays.forEach { (name, cmd) -> add(CustomRelay(name, cmd)) } + } + +private fun loadCorpus( + options: Options, + http: OkHttpClient, + log: (String) -> Unit, +): Corpus = + when { + options.corpusFile != null -> + CorpusSource.fromFile( + options.corpusFile, + options.limit, + options.cacheDir, + log, + options.maxEventBytes, + options.maxTags, + ) + options.downloadFrom != null -> + CorpusDownloader.download( + options.downloadFrom.ifEmpty { CorpusDownloader.DEFAULT_RELAYS }, + if (options.limit > 0) options.limit else options.events, + options.cacheDir, + http, + log, + ) + else -> + CorpusSource.synthetic( + CorpusSpec( + seed = options.seed, + events = options.events, + baseTime = options.baseTime, + spanSeconds = CorpusSpec.DEFAULT_SPAN_SECONDS, + ), + options.cacheDir, + log, + ) + } + +private fun parseArgs(args: Array): Options? { + val map = HashMap>() + val flags = HashSet() + var i = 0 + val valueOptions = + setOf( + "--geode-bin", + "--strfry-bin", + "--relay", + "--events", + "--seed", + "--base-time", + "--corpus", + "--limit", + "--download", + "--max-event-bytes", + "--max-tags", + "--samples", + "--publishers", + "--window", + "--query-rounds", + "--query-conns", + "--out", + "--work", + ) + while (i < args.size) { + val arg = args[i] + when { + arg == "--help" || arg == "-h" -> return null + arg.contains('=') && arg.substringBefore('=') in valueOptions -> { + map.getOrPut(arg.substringBefore('=')) { mutableListOf() }.add(arg.substringAfter('=')) + } + arg in valueOptions -> { + // --download may appear with no value (use default relays). + val next = args.getOrNull(i + 1) + if (next != null && !next.startsWith("--")) { + map.getOrPut(arg) { mutableListOf() }.add(next) + i++ + } else if (arg == "--download") { + map.getOrPut(arg) { mutableListOf() }.add("") + } else { + System.err.println("Missing value for $arg") + return null + } + } + arg.startsWith("--") -> flags.add(arg) + else -> { + System.err.println("Unknown argument: $arg") + return null + } + } + i++ + } + + fun one(key: String): String? = map[key]?.lastOrNull() + + fun int( + key: String, + default: Int, + ): Int = one(key)?.toIntOrNull() ?: default + + val quick = "--quick" in flags + if ("--no-color" in flags) Report.disableColor() + + val moduleDir = File("relayBench").takeIf { it.isDirectory } ?: File(".") + return Options( + geodeBin = one("--geode-bin"), + strfryBin = one("--strfry-bin"), + customRelays = + (map["--relay"] ?: emptyList()).map { + val name = it.substringBefore('=') + val cmd = it.substringAfter('=') + name to cmd + }, + events = int("--events", if (quick) 2000 else 10_000), + seed = one("--seed")?.toLongOrNull() ?: 1L, + baseTime = + when (val t = one("--base-time")) { + null -> CorpusSpec.DEFAULT_BASE_TIME + "now" -> System.currentTimeMillis() / 1000 + else -> t.toLongOrNull() ?: CorpusSpec.DEFAULT_BASE_TIME + }, + corpusFile = one("--corpus")?.let { File(it) }, + downloadFrom = map["--download"]?.lastOrNull()?.split(',')?.filter { it.isNotBlank() }, + limit = int("--limit", 0), + maxEventBytes = int("--max-event-bytes", CorpusSource.DEFAULT_MAX_EVENT_BYTES), + maxTags = int("--max-tags", CorpusSource.DEFAULT_MAX_TAGS), + visibilitySamples = int("--samples", if (quick) 50 else 200), + publishers = int("--publishers", 4), + window = int("--window", 200), + warmupRounds = if (quick) 1 else 2, + queryRounds = int("--query-rounds", if (quick) 5 else 10), + queryConnections = int("--query-conns", if (quick) 4 else 8), + concurrentRounds = if (quick) 2 else 5, + skipSync = "--no-sync" in flags, + outDir = one("--out")?.let { File(it) } ?: File(moduleDir, "results"), + workDir = one("--work")?.let { File(it) } ?: File(System.getProperty("java.io.tmpdir"), "relaybench-${System.nanoTime()}"), + cacheDir = File(moduleDir, ".corpus-cache"), + keepData = "--keep-data" in flags, + ) +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/NostrSocket.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/NostrSocket.kt new file mode 100644 index 0000000000..7f09807fdb --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/NostrSocket.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.relaybench + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.withTimeout +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import java.net.InetAddress +import java.net.InetSocketAddress +import java.net.Socket +import javax.net.SocketFactory + +/** + * The thinnest possible NIP-01 wire client: one WebSocket, every inbound + * text frame pushed into an unbounded [Channel]. Each benchmark phase owns + * its sockets outright and parses frames itself, so no subscription router + * sits between the relay and the measurement — what we time is the wire, + * not this harness. + */ +class NostrSocket private constructor( + private val ws: WebSocket, + val incoming: Channel, +) { + fun send(text: String): Boolean = ws.send(text) + + fun publish(eventJson: String): Boolean = send("""["EVENT",$eventJson]""") + + fun req( + subId: String, + filterJson: String, + ): Boolean = send("""["REQ","$subId",$filterJson]""") + + fun close(subId: String): Boolean = send("""["CLOSE","$subId"]""") + + fun disconnect() { + ws.close(1000, "bench-done") + } + + companion object { + /** + * Sockets with Nagle disabled. The JDK default (Nagle on) delays a + * small write that follows another unacked small write — e.g. a REQ + * sent right after the previous round's CLOSE — by a full delayed-ACK + * interval (~40 ms), which would drown every latency this harness + * measures. Verified: with the default factory every REQ→EOSE round + * floors at ~44 ms against both geode and strfry; with TCP_NODELAY + * the same rounds take ~0.3 ms. + */ + val NO_DELAY_SOCKETS: SocketFactory = + object : SocketFactory() { + private fun socket() = Socket().apply { tcpNoDelay = true } + + override fun createSocket(): Socket = socket() + + override fun createSocket( + host: String?, + port: Int, + ): Socket = socket().apply { connect(InetSocketAddress(host, port)) } + + override fun createSocket( + host: String?, + port: Int, + localHost: InetAddress?, + localPort: Int, + ): Socket = throw UnsupportedOperationException() + + override fun createSocket( + host: InetAddress?, + port: Int, + ): Socket = socket().apply { connect(InetSocketAddress(host, port)) } + + override fun createSocket( + address: InetAddress?, + port: Int, + localAddress: InetAddress?, + localPort: Int, + ): Socket = throw UnsupportedOperationException() + } + + suspend fun connect( + http: OkHttpClient, + wsUrl: String, + ): NostrSocket { + val incoming = Channel(Channel.UNLIMITED) + val opened = CompletableDeferred() + val httpUrl = wsUrl.replaceFirst("ws://", "http://").replaceFirst("wss://", "https://") + val ws = + http.newWebSocket( + Request.Builder().url(httpUrl).build(), + object : WebSocketListener() { + override fun onOpen( + webSocket: WebSocket, + response: Response, + ) { + opened.complete(Unit) + } + + override fun onMessage( + webSocket: WebSocket, + text: String, + ) { + incoming.trySend(text) + } + + override fun onClosed( + webSocket: WebSocket, + code: Int, + reason: String, + ) { + incoming.close() + } + + override fun onFailure( + webSocket: WebSocket, + t: Throwable, + response: Response?, + ) { + opened.completeExceptionally(t) + incoming.close(t) + } + }, + ) + withTimeout(10_000) { opened.await() } + return NostrSocket(ws, incoming) + } + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Percentiles.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Percentiles.kt new file mode 100644 index 0000000000..8b277bab93 --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Percentiles.kt @@ -0,0 +1,51 @@ +/* + * 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.relaybench + +/** Latency distribution summary, in milliseconds. */ +data class Percentiles( + val samples: Int, + val mean: Double, + val p50: Double, + val p90: Double, + val p99: Double, + val max: Double, +) { + companion object { + fun ofNanos(nanos: List): Percentiles { + if (nanos.isEmpty()) return Percentiles(0, 0.0, 0.0, 0.0, 0.0, 0.0) + val sorted = nanos.sorted() + + fun pct(p: Double): Double { + val idx = ((sorted.size - 1) * p).toInt() + return sorted[idx] / 1_000_000.0 + } + return Percentiles( + samples = sorted.size, + mean = sorted.average() / 1_000_000.0, + p50 = pct(0.50), + p90 = pct(0.90), + p99 = pct(0.99), + max = sorted.last() / 1_000_000.0, + ) + } + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Report.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Report.kt new file mode 100644 index 0000000000..636d5c9354 --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Report.kt @@ -0,0 +1,408 @@ +/* + * 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.relaybench + +import com.fasterxml.jackson.databind.SerializationFeature +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.vitorpamplona.relaybench.bench.RelayResult +import com.vitorpamplona.relaybench.bench.SyncBenchmark +import com.vitorpamplona.relaybench.corpus.Corpus +import java.time.ZonedDateTime +import java.time.format.DateTimeFormatter + +/** Everything one benchmark run produced, ready to render. */ +class BenchRun( + val startedAt: ZonedDateTime, + val corpus: Corpus, + val options: Options, + val relays: List, + val syncs: List, +) + +/** + * Renders the run three ways: a rich ANSI terminal report (bars, colors, + * winners), a Markdown report for sharing, and raw JSON for tooling. + */ +object Report { + private const val BAR_WIDTH = 30 + + private var color = System.getenv("NO_COLOR") == null + + fun disableColor() { + color = false + } + + private fun paint( + code: String, + text: String, + ) = if (color) "\u001B[${code}m$text\u001B[0m" else text + + private fun bold(t: String) = paint("1", t) + + private fun dim(t: String) = paint("2", t) + + private fun green(t: String) = paint("32;1", t) + + private fun cyan(t: String) = paint("36", t) + + private fun yellow(t: String) = paint("33", t) + + private fun red(t: String) = paint("31;1", t) + + private fun bar( + value: Double, + max: Double, + best: Boolean, + ): String { + if (max <= 0 || value.isNaN()) return "" + val cells = ((value / max) * BAR_WIDTH).toInt().coerceIn(if (value > 0) 1 else 0, BAR_WIDTH) + val b = "█".repeat(cells) + "░".repeat(BAR_WIDTH - cells) + return if (best) green(b) else cyan(b) + } + + private fun fmt(v: Double): String = + when { + v >= 1000 -> "%,.0f".format(v) + v >= 10 -> "%.1f".format(v) + else -> "%.2f".format(v) + } + + private fun bytesHuman(b: Long): String = + when { + b >= 1 shl 30 -> "%.2f GiB".format(b / 1073741824.0) + b >= 1 shl 20 -> "%.1f MiB".format(b / 1048576.0) + b >= 1 shl 10 -> "%.1f KiB".format(b / 1024.0) + else -> "$b B" + } + + /** + * One metric across all relays: name column, bar, value, star on the + * winner. [higherIsBetter] flips both the star and the bar scaling + * (for latencies the *shortest* bar wins, so bars stay proportional + * to the raw value and the star marks the minimum). + */ + private fun StringBuilder.metricRows( + rows: List>, + unit: String, + higherIsBetter: Boolean, + detail: Map = emptyMap(), + ) { + val present = rows.mapNotNull { (n, v) -> v?.let { n to it } } + if (present.isEmpty()) return + val max = present.maxOf { it.second } + val bestValue = if (higherIsBetter) present.maxOf { it.second } else present.minOf { it.second } + val nameWidth = rows.maxOf { it.first.length } + for ((name, value) in rows) { + if (value == null) { + appendLine(" ${name.padEnd(nameWidth)} ${dim("(failed)")}") + continue + } + val best = value == bestValue && present.size > 1 + val star = if (best) green(" ★") else "" + val valueText = (fmt(value) + " " + unit).let { if (best) bold(it) else it } + val extra = detail[name]?.let { " ${dim(it)}" } ?: "" + appendLine(" ${name.padEnd(nameWidth)} ${bar(value, max, best)} $valueText$star$extra") + } + } + + private fun StringBuilder.sectionTitle(title: String) { + appendLine() + appendLine(bold("▌ $title")) + } + + fun terminal(run: BenchRun): String = + buildString { + val line = "═".repeat(74) + appendLine(bold(line)) + appendLine(bold(" RELAY BENCH") + dim(" · ${run.startedAt.format(DateTimeFormatter.RFC_1123_DATE_TIME)}")) + appendLine(" corpus: ${run.corpus.source} — ${"%,d".format(run.corpus.events.size)} events, fingerprint ${run.corpus.fingerprint}") + val kinds = + run.corpus + .kindHistogram() + .entries + .sortedByDescending { it.value } + .take(8) + .joinToString(" ") { "k${it.key}:${"%,d".format(it.value)}" } + appendLine(" kinds: $kinds") + appendLine( + " relays: " + + run.relays.joinToString(" · ") { r -> + val v = r.info?.let { "${it.software ?: r.name} ${it.version ?: ""}".trim() } ?: r.name + if (r.error != null) red("$v (FAILED)") else bold(v) + }, + ) + appendLine(bold(line)) + + run.relays.filter { it.error != null }.forEach { + appendLine(red(" ✗ ${it.name} failed: ${it.error?.lineSequence()?.first()}")) + } + + val ok = run.relays.filter { it.error == null } + if (ok.isNotEmpty()) { + sectionTitle("INGEST — receipt ➜ queryable by REQ (idle relay, ${ok.firstNotNullOfOrNull { it.visibility?.samples } ?: 0} samples)") + metricRows(ok.map { it.name to it.visibility?.visibleLatency?.p50 }, "ms p50", higherIsBetter = false, detail = ok.associate { it.name to "p99 ${fmt(it.visibility?.visibleLatency?.p99 ?: 0.0)} ms" }) + appendLine(dim(" OK-ack latency:")) + metricRows(ok.map { it.name to it.visibility?.okLatency?.p50 }, "ms p50", higherIsBetter = false, detail = ok.associate { it.name to "p99 ${fmt(it.visibility?.okLatency?.p99 ?: 0.0)} ms" }) + ok.forEach { r -> + val pct = (r.visibility?.visibleByOkTime ?: 0.0) * 100 + appendLine(dim(" ${r.name}: ${"%.0f".format(pct)}% of events were already queryable when their OK arrived")) + } + + sectionTitle("INGEST — corpus replay throughput (${run.options.publishers} connections, window ${run.options.window})") + metricRows( + ok.map { it.name to it.throughput?.eventsPerSec }, + "events/s", + higherIsBetter = true, + detail = + ok.associate { + it.name to + "${"%,d".format(it.throughput?.accepted ?: 0)} accepted, ${"%,d".format(it.throughput?.rejected ?: 0)} rejected, ${fmt((it.throughput?.wallMs ?: 0.0) / 1000)} s" + }, + ) + + sectionTitle("STORAGE — on-disk footprint after full ingest") + metricRows(ok.map { it.name to it.storageBytes.toDouble() / 1048576.0 }, "MiB", higherIsBetter = false) + + // ---- queries ---- + val scenarioKeys = ok.flatMap { r -> r.queries.map { it.scenario.key } }.distinct() + if (scenarioKeys.isNotEmpty()) { + sectionTitle("QUERIES — time to EOSE, single client (p50, ${run.options.queryRounds} rounds)") + for (key in scenarioKeys) { + val results = ok.associateWith { r -> r.queries.find { it.scenario.key == key } } + val any = results.values.filterNotNull().firstOrNull() ?: continue + val counts = + results.values + .filterNotNull() + .map { it.eventsPerRound } + .distinct() + val mismatch = if (counts.size > 1) red(" ⚠ result sets differ: $counts") else dim(" ${counts.first()} events") + appendLine(" ${bold(key)} ${dim("— " + any.scenario.description)}$mismatch") + metricRows( + ok.map { it.name to results[it]?.timeToEose?.p50 }, + "ms", + higherIsBetter = false, + detail = ok.associate { it.name to "p99 ${fmt(results[it]?.timeToEose?.p99 ?: 0.0)} ms, first event ${fmt(results[it]?.timeToFirst?.p50 ?: 0.0)} ms" }, + ) + } + + sectionTitle("QUERIES — aggregate throughput (${run.options.queryConnections} concurrent connections)") + for (key in scenarioKeys) { + val results = ok.associateWith { r -> r.queries.find { it.scenario.key == key } } + appendLine(" ${bold(key)}") + metricRows(ok.map { it.name to results[it]?.concurrentEventsPerSec }, "events/s", higherIsBetter = true) + } + } + } + + // ---- sync ---- + if (run.syncs.isNotEmpty()) { + sectionTitle("NIP-77 NEGENTROPY SYNC — pairwise (80%/80% slices, 60% overlap)") + for (pair in run.syncs) { + val status = if (pair.converged) green("converged ✓") else red("DID NOT CONVERGE ✗") + appendLine(" ${bold("${pair.serverA} ⇄ ${pair.serverB}")} ${dim("${"%,d".format(pair.syncableEvents)} syncable events")} $status") + pair.error?.let { appendLine(red(" error: $it")) } + if (pair.initialReconcile.isNotEmpty()) { + appendLine(dim(" initial reconcile (server side doing the range work):")) + metricRows( + pair.initialReconcile.map { (name, s) -> name to s.ms }, + "ms", + higherIsBetter = false, + detail = + pair.initialReconcile.mapValues { (_, s) -> + "${s.rounds} rounds, ${bytesHuman(s.wireBytes)} on the wire, need ${s.needCount} / have ${s.haveCount}" + + (s.error?.let { " ! $it" } ?: "") + }, + ) + appendLine(dim(" delta transfer: ${pair.transferredToA} → ${pair.serverA}, ${pair.transferredToB} → ${pair.serverB} in ${fmt(pair.transferMs)} ms")) + appendLine(dim(" steady-state reconcile of identical sets:")) + metricRows( + pair.identicalReconcile.map { (name, s) -> name to s.ms }, + "ms", + higherIsBetter = false, + detail = pair.identicalReconcile.mapValues { (_, s) -> "${s.rounds} rounds, ${bytesHuman(s.wireBytes)}" }, + ) + } + } + } + + // ---- verdict ---- + val okRelays = run.relays.filter { it.error == null } + if (okRelays.size > 1) { + sectionTitle("HEAD-TO-HEAD") + appendLine(headToHead(okRelays)) + } + appendLine() + appendLine(bold(line)) + } + + private fun headToHead(relays: List): String = + buildString { + fun crown( + label: String, + winner: String?, + note: String, + ) { + if (winner != null) appendLine(" $label: ${green(bold(winner))} ${dim(note)}") + } + + val ingest = relays.mapNotNull { r -> r.throughput?.let { r.name to it.eventsPerSec } } + ingest.maxByOrNull { it.second }?.let { (name, eps) -> + val runnerUp = ingest.filter { it.first != name }.maxByOrNull { it.second } + val ratio = runnerUp?.let { "%.1f× faster than ${it.first}".format(eps / it.second) } ?: "" + crown("ingest throughput", name, ratio) + } + val visible = relays.mapNotNull { r -> r.visibility?.let { r.name to it.visibleLatency.p50 } } + visible.minByOrNull { it.second }?.let { (name, ms) -> + val runnerUp = visible.filter { it.first != name }.minByOrNull { it.second } + val ratio = runnerUp?.let { "%.1f× lower p50 than ${it.first}".format(it.second / ms) } ?: "" + crown("receipt➜queryable latency", name, ratio) + } + val queryWins = HashMap() + val keys = relays.flatMap { r -> r.queries.map { it.scenario.key } }.distinct() + for (key in keys) { + relays + .mapNotNull { r -> r.queries.find { it.scenario.key == key }?.let { r.name to it.timeToEose.p50 } } + .minByOrNull { it.second } + ?.let { queryWins.merge(it.first, 1, Int::plus) } + } + queryWins.maxByOrNull { it.value }?.let { (name, wins) -> + crown("query latency", name, "fastest EOSE in $wins of ${keys.size} scenarios") + } + } + + // ------------------------------------------------------------------ + fun markdown(run: BenchRun): String = + buildString { + appendLine("# Relay Bench — ${run.startedAt.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME)}") + appendLine() + appendLine("- **Corpus**: ${run.corpus.source} — ${"%,d".format(run.corpus.events.size)} events, fingerprint `${run.corpus.fingerprint}`") + appendLine("- **Relays**: " + run.relays.joinToString(", ") { r -> r.info?.let { "${it.software} ${it.version}" } ?: r.name }) + appendLine("- **Settings**: ${run.options.publishers} publisher conns, window ${run.options.window}, ${run.options.queryRounds} query rounds, ${run.options.queryConnections} concurrent conns") + appendLine() + val ok = run.relays.filter { it.error == null } + run.relays.filter { it.error != null }.forEach { appendLine("> ⚠ **${it.name} failed**: ${it.error?.lineSequence()?.first()}") } + + appendLine("## Ingest") + appendLine() + appendLine("| metric | " + ok.joinToString(" | ") { it.name } + " |") + appendLine("|---|" + ok.joinToString("|") { "---:" } + "|") + + fun row( + label: String, + value: (RelayResult) -> String, + ) = appendLine("| $label | " + ok.joinToString(" | ") { value(it) } + " |") + row("receipt➜queryable p50 (ms)") { fmt(it.visibility?.visibleLatency?.p50 ?: Double.NaN) } + row("receipt➜queryable p99 (ms)") { fmt(it.visibility?.visibleLatency?.p99 ?: Double.NaN) } + row("OK ack p50 (ms)") { fmt(it.visibility?.okLatency?.p50 ?: Double.NaN) } + row("queryable by OK time") { "%.0f%%".format((it.visibility?.visibleByOkTime ?: 0.0) * 100) } + row("replay throughput (events/s)") { fmt(it.throughput?.eventsPerSec ?: Double.NaN) } + row("accepted / rejected") { "${it.throughput?.accepted} / ${it.throughput?.rejected}" } + row("storage after ingest") { bytesHuman(it.storageBytes) } + appendLine() + + appendLine("## Queries") + appendLine() + appendLine("| scenario | events | " + ok.joinToString(" | ") { "${it.name} EOSE p50 (ms)" } + " | " + ok.joinToString(" | ") { "${it.name} @${run.options.queryConnections}conn (ev/s)" } + " |") + appendLine("|---|---:|" + ok.joinToString("|") { "---:" } + "|" + ok.joinToString("|") { "---:" } + "|") + val keys = ok.flatMap { r -> r.queries.map { it.scenario.key } }.distinct() + for (key in keys) { + val cells = ok.map { r -> r.queries.find { it.scenario.key == key } } + val counts = cells.filterNotNull().map { it.eventsPerRound }.distinct() + val eventsCell = if (counts.size > 1) "⚠ $counts" else "${counts.firstOrNull() ?: "-"}" + appendLine( + "| $key | $eventsCell | " + + cells.joinToString(" | ") { fmt(it?.timeToEose?.p50 ?: Double.NaN) } + " | " + + cells.joinToString(" | ") { fmt(it?.concurrentEventsPerSec ?: Double.NaN) } + " |", + ) + } + appendLine() + + if (run.syncs.isNotEmpty()) { + appendLine("## NIP-77 negentropy sync") + appendLine() + for (pair in run.syncs) { + appendLine("### ${pair.serverA} ⇄ ${pair.serverB} — ${if (pair.converged) "converged ✓" else "did not converge ✗"}") + appendLine() + appendLine("| phase | server | ms | rounds | wire | need/have |") + appendLine("|---|---|---:|---:|---:|---|") + pair.initialReconcile.forEach { (name, s) -> + appendLine("| initial reconcile | $name | ${fmt(s.ms)} | ${s.rounds} | ${bytesHuman(s.wireBytes)} | ${s.needCount}/${s.haveCount} |") + } + pair.identicalReconcile.forEach { (name, s) -> + appendLine("| identical-set reconcile | $name | ${fmt(s.ms)} | ${s.rounds} | ${bytesHuman(s.wireBytes)} | 0/0 |") + } + appendLine() + appendLine("Delta transfer: ${pair.transferredToA} events → ${pair.serverA}, ${pair.transferredToB} → ${pair.serverB} in ${fmt(pair.transferMs)} ms.") + appendLine() + } + } + } + + fun json(run: BenchRun): String { + val mapper = jacksonObjectMapper().enable(SerializationFeature.INDENT_OUTPUT) + val tree = + mapOf( + "startedAt" to run.startedAt.toString(), + "corpus" to + mapOf( + "source" to run.corpus.source, + "events" to run.corpus.events.size, + "fingerprint" to run.corpus.fingerprint, + "kinds" to run.corpus.kindHistogram(), + ), + "options" to + mapOf( + "publishers" to run.options.publishers, + "window" to run.options.window, + "visibilitySamples" to run.options.visibilitySamples, + "queryRounds" to run.options.queryRounds, + "queryConnections" to run.options.queryConnections, + ), + "relays" to + run.relays.map { r -> + mapOf( + "name" to r.name, + "software" to r.info?.software, + "version" to r.info?.version, + "error" to r.error, + "visibility" to r.visibility, + "throughput" to r.throughput, + "storageBytes" to r.storageBytes, + "queries" to + r.queries.map { q -> + mapOf( + "scenario" to q.scenario.key, + "description" to q.scenario.description, + "filter" to q.scenario.filterJson, + "eventsPerRound" to q.eventsPerRound, + "timeToFirst" to q.timeToFirst, + "timeToEose" to q.timeToEose, + "sequentialEventsPerSec" to q.sequentialEventsPerSec, + "concurrentEventsPerSec" to q.concurrentEventsPerSec, + ) + }, + ) + }, + "syncs" to run.syncs, + ) + return mapper.writeValueAsString(tree) + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Scenarios.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Scenarios.kt new file mode 100644 index 0000000000..43da3db874 --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Scenarios.kt @@ -0,0 +1,177 @@ +/* + * 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.relaybench + +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.relaybench.corpus.Corpus + +/** + * One REQ workload, named after the client behavior it models. The filter + * is derived from the corpus itself (top authors, hottest thread, most + * mentioned pubkey, …) so the same scenario keys stay meaningful whether + * the corpus is synthetic, the checked-in real dump, or freshly downloaded. + * + * Everything stays inside strfry's *default* limits (≤200 filter elements, + * limit ≤500, ≤3 tag kinds per filter) — the point is comparing relays on + * out-of-the-box configs, not tuning around them. + */ +data class Scenario( + val key: String, + val description: String, + val filter: Filter, +) { + val filterJson: String get() = filter.toJson() +} + +object Scenarios { + fun derive(corpus: Corpus): List { + val events = corpus.events + + // Frequency maps that mirror what relay indexes serve most. + val notesByAuthor = HashMap() + val eTagRefs = HashMap() + val pTagRefs = HashMap() + val tTagRefs = HashMap() + val noteIds = ArrayList() + + for (e in events) { + if (e.kind == 1) { + notesByAuthor.merge(e.pubKey, 1, Int::plus) + noteIds.add(e.id) + } + if (e.kind == 1 || e.kind == 6 || e.kind == 7 || e.kind == 9735) { + for (tag in e.tags) { + if (tag.size < 2) continue + when (tag[0]) { + "e" -> if (tag[1].length == 64) eTagRefs.merge(tag[1], 1, Int::plus) + "p" -> if (tag[1].length == 64) pTagRefs.merge(tag[1], 1, Int::plus) + "t" -> if (e.kind == 1) tTagRefs.merge(tag[1].lowercase(), 1, Int::plus) + } + } + } + } + + val topAuthors = notesByAuthor.entries.sortedWith(compareByDescending> { it.value }.thenBy { it.key }).map { it.key } + val hottestThread = + eTagRefs.entries + .sortedWith(compareByDescending> { it.value }.thenBy { it.key }) + .firstOrNull() + ?.key + val mostMentioned = + pTagRefs.entries + .sortedWith(compareByDescending> { it.value }.thenBy { it.key }) + .firstOrNull() + ?.key + val topHashtag = + tTagRefs.entries + .sortedWith(compareByDescending> { it.value }.thenBy { it.key }) + .firstOrNull() + ?.key + + // Evenly spread sample of note ids — a "fetch these 100 events" batch. + val idSample = + if (noteIds.size <= 100) { + noteIds.toList() + } else { + (0 until 100).map { noteIds[it * noteIds.size / 100] } + } + + val maxTime = events.maxOf { it.createdAt } + val minTime = events.minOf { it.createdAt } + val recentSince = maxTime - ((maxTime - minTime) / 5).coerceAtLeast(1) + + return buildList { + add( + Scenario( + "firehose", + "latest 500 events of any kind (relay landing query)", + Filter(limit = 500), + ), + ) + add( + Scenario( + "global-feed", + "latest 500 text notes (global tab)", + Filter(kinds = listOf(1), limit = 500), + ), + ) + if (topAuthors.isNotEmpty()) { + add( + Scenario( + "profiles", + "metadata for ${topAuthors.take(50).size} pubkeys (profile hydration)", + Filter(kinds = listOf(0), authors = topAuthors.take(50)), + ), + ) + add( + Scenario( + "follow-feed", + "notes+reposts from ${topAuthors.take(150).size} follows (home feed)", + Filter(kinds = listOf(1, 6), authors = topAuthors.take(150), limit = 500), + ), + ) + } + hottestThread?.let { + add( + Scenario( + "thread", + "replies/reposts/reactions/zaps on the hottest note (thread view)", + Filter(kinds = listOf(1, 6, 7, 9735), tags = mapOf("e" to listOf(it))), + ), + ) + } + mostMentioned?.let { + add( + Scenario( + "notifications", + "events tagging the most-mentioned pubkey (notifications tab)", + Filter(kinds = listOf(1, 6, 7, 9735), tags = mapOf("p" to listOf(it)), limit = 500), + ), + ) + } + topHashtag?.let { + add( + Scenario( + "hashtag", + "latest notes under #$it (hashtag feed)", + Filter(kinds = listOf(1), tags = mapOf("t" to listOf(it)), limit = 500), + ), + ) + } + if (idSample.isNotEmpty()) { + add( + Scenario( + "by-ids", + "batch fetch of ${idSample.size} known event ids", + Filter(ids = idSample), + ), + ) + } + add( + Scenario( + "recent-window", + "notes since the corpus' last 20% time window", + Filter(kinds = listOf(1), since = recentSince, limit = 500), + ), + ) + } + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/BenchmarkRunner.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/BenchmarkRunner.kt new file mode 100644 index 0000000000..c8d2b68d7e --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/BenchmarkRunner.kt @@ -0,0 +1,219 @@ +/* + * 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.relaybench.bench + +import com.vitorpamplona.relaybench.Options +import com.vitorpamplona.relaybench.Scenario +import com.vitorpamplona.relaybench.corpus.Corpus +import com.vitorpamplona.relaybench.relays.Nip11Info +import com.vitorpamplona.relaybench.relays.RelayUnderTest +import com.vitorpamplona.relaybench.relays.RunningRelay +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.runBlocking +import okhttp3.OkHttpClient +import java.io.File + +/** Everything measured for one relay implementation. */ +class RelayResult( + val name: String, + val info: Nip11Info?, + val visibility: IngestBenchmark.VisibilityResult?, + val throughput: IngestBenchmark.ThroughputResult?, + val queries: List, + val storageBytes: Long, + val error: String? = null, +) + +/** + * Runs the full single-relay pipeline — boot fresh, probe visibility on an + * idle relay, replay the corpus for throughput, run every query scenario, + * measure disk footprint, shut down — then the pairwise sync pipeline for + * every relay combination. + */ +object BenchmarkRunner { + fun runSingle( + relay: RelayUnderTest, + corpus: Corpus, + scenarios: List, + options: Options, + workDir: File, + http: OkHttpClient, + log: (String) -> Unit, + ): RelayResult { + val running = + try { + relay.start(workDir) + } catch (e: Exception) { + return RelayResult(relay.name, null, null, null, emptyList(), 0, error = e.message) + } + try { + val info = running.fetchInfo(http) + log(" ${relay.name} up at ${running.wsUrl} (${info?.software ?: "?"} ${info?.version ?: ""})") + + // Visibility samples: non-replaceable events only (a superseded + // replaceable would never come back by id and the probe would spin). + val sampleIds = HashSet() + val visibilitySamples = + corpus.events + .filter { it.kind == 1 } + .ifEmpty { corpus.events.filter { it.kind !in setOf(0, 3) && it.kind !in 10000..19999 && it.kind !in 30000..39999 } } + .let { pool -> + if (pool.size <= options.visibilitySamples) { + pool + } else { + (0 until options.visibilitySamples).map { pool[it * pool.size / options.visibilitySamples] } + } + }.onEach { sampleIds.add(it.id) } + + log(" visibility probe: ${visibilitySamples.size} sequential EVENT→REQ round trips…") + val visibility = + runBlocking { IngestBenchmark.visibility(running.wsUrl, visibilitySamples, http, log) } + running.ensureAlive() + + val rest = corpus.events.filter { it.id !in sampleIds } + log(" ingest throughput: ${rest.size} events over ${options.publishers} connections (window ${options.window})…") + val throughput = + runBlocking { IngestBenchmark.throughput(running.wsUrl, rest, options.publishers, options.window, http) } + log(" ${"%,.0f".format(throughput.eventsPerSec)} events/s (${throughput.accepted} accepted, ${throughput.rejected} rejected)") + running.ensureAlive() + + val queries = + scenarios.map { scenario -> + log(" query [${scenario.key}] ${options.queryRounds} rounds + ${options.queryConnections}×${options.concurrentRounds} concurrent…") + val result = + runBlocking { + QueryBenchmark.run( + running.wsUrl, + scenario, + warmupRounds = options.warmupRounds, + rounds = options.queryRounds, + concurrency = options.queryConnections, + concurrentRounds = options.concurrentRounds, + http = http, + ) + } + log( + " ${result.eventsPerRound} events, EOSE p50 ${"%.1f".format(result.timeToEose.p50)} ms, " + + "${"%,.0f".format(result.sequentialEventsPerSec)} ev/s seq, ${"%,.0f".format(result.concurrentEventsPerSec)} ev/s @${options.queryConnections}conn", + ) + running.ensureAlive() + result + } + + val storage = running.storageBytes() + return RelayResult(relay.name, info, visibility, throughput, queries, storage) + } catch (e: Exception) { + return RelayResult( + relay.name, + null, + null, + null, + emptyList(), + 0, + error = "${e.message}\nrelay log tail:\n${running.logFile + .takeIf { it.exists() } + ?.readLines() + ?.takeLast(10) + ?.joinToString("\n") ?: ""}", + ) + } finally { + running.stop() + } + } + + fun runSyncPairs( + relays: List, + corpus: Corpus, + options: Options, + workDir: File, + http: OkHttpClient, + log: (String) -> Unit, + ): List { + if (relays.size < 2) return emptyList() + val effective = SyncBenchmark.effectiveEvents(corpus.events) + val n = effective.size + val sliceA = effective.subList(0, n * 4 / 5) + val sliceB = effective.subList(n / 5, n) + + val results = ArrayList() + for (i in relays.indices) { + for (j in i + 1 until relays.size) { + val a = relays[i] + val b = relays[j] + log(" sync pair ${a.name} ⇄ ${b.name}: $n syncable events, 80%/80% slices, 60% overlap") + val pairDir = File(workDir, "sync-${a.name}-${b.name}").apply { mkdirs() } + var runningA: RunningRelay? = null + var runningB: RunningRelay? = null + try { + runningA = a.start(pairDir) + runningB = b.start(pairDir) + val urlA = runningA.wsUrl + val urlB = runningB.wsUrl + log(" seeding both sides (${sliceA.size} + ${sliceB.size} events)…") + runBlocking { + coroutineScope { + listOf( + async { IngestBenchmark.throughput(urlA, sliceA, options.publishers, options.window, http) }, + async { IngestBenchmark.throughput(urlB, sliceB, options.publishers, options.window, http) }, + ).awaitAll() + } + } + results += + runBlocking { + SyncBenchmark.runPair( + a.name, + urlA, + b.name, + urlB, + effective, + sliceA, + sliceB, + options.window, + http, + log, + ) + } + } catch (e: Exception) { + log(" ! sync pair failed: ${e.message}") + results += + SyncBenchmark.PairResult( + a.name, + b.name, + n, + emptyMap(), + 0.0, + 0, + 0, + emptyMap(), + converged = false, + error = e.message, + ) + } finally { + runningA?.stop() + runningB?.stop() + } + } + } + return results + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/IngestBenchmark.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/IngestBenchmark.kt new file mode 100644 index 0000000000..874649a9e2 --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/IngestBenchmark.kt @@ -0,0 +1,253 @@ +/* + * 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.relaybench.bench + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.relaybench.NostrSocket +import com.vitorpamplona.relaybench.Percentiles +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.TimeoutCancellationException +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout +import okhttp3.OkHttpClient +import java.util.concurrent.atomic.AtomicInteger + +/** + * Write-path metrics. + * + * **Visibility latency** answers the exact question "how long from the + * relay *receiving* an EVENT until a REQ can return it": one connection + * publishes, and from the same send-instant a second connection + * hammer-polls `REQ {ids:[id]}` until the event comes back. The polled + * REQ round-trip on loopback (~sub-ms) bounds the probe's resolution. + * The OK-ack latency and whether the event was readable no later than + * its OK (i.e. does the ack imply queryable — NIP-20 semantics) are + * recorded alongside. + * + * **Throughput** replays the corpus over N connections with a bounded + * number of unacked EVENTs in flight per connection, timing wall clock + * from first send to last OK. + */ +object IngestBenchmark { + private val mapper = jacksonObjectMapper() + + data class VisibilityResult( + val okLatency: Percentiles, + val visibleLatency: Percentiles, + val visibleByOkTime: Double, + val samples: Int, + val rejected: Int, + ) + + data class ThroughputResult( + val events: Int, + val accepted: Int, + val rejected: Int, + val wallMs: Double, + val eventsPerSec: Double, + val connections: Int, + ) + + /** (okNanos, visibleNanos, visibleBeforeOk) for one accepted event, or null if rejected. */ + private class Sample( + val okNanos: Long, + val visibleNanos: Long, + val visibleByOk: Boolean, + ) + + suspend fun visibility( + wsUrl: String, + samples: List, + http: OkHttpClient, + log: (String) -> Unit, + ): VisibilityResult { + val publisher = NostrSocket.connect(http, wsUrl) + val prober = NostrSocket.connect(http, wsUrl) + val okNanos = ArrayList(samples.size) + val visibleNanos = ArrayList(samples.size) + var byOkHits = 0 + var rejected = 0 + var timeouts = 0 + + try { + for ((i, event) in samples.withIndex()) { + if (timeouts >= 5) { + log(" ! aborting visibility probe after $timeouts timeouts") + break + } + val sample = + try { + probeOne(publisher, prober, event, i) + } catch (_: TimeoutCancellationException) { + timeouts++ + continue + } + if (sample == null) { + rejected++ + } else { + okNanos += sample.okNanos + visibleNanos += sample.visibleNanos + if (sample.visibleByOk) byOkHits++ + } + } + } finally { + publisher.disconnect() + prober.disconnect() + } + if (rejected > 0) log(" ! $rejected visibility samples were rejected by the relay") + return VisibilityResult( + okLatency = Percentiles.ofNanos(okNanos), + visibleLatency = Percentiles.ofNanos(visibleNanos), + visibleByOkTime = if (visibleNanos.isEmpty()) 0.0 else byOkHits.toDouble() / visibleNanos.size, + samples = visibleNanos.size, + rejected = rejected, + ) + } + + private suspend fun probeOne( + publisher: NostrSocket, + prober: NostrSocket, + event: Event, + index: Int, + ): Sample? = + withTimeout(30_000) { + coroutineScope { + val start = System.nanoTime() + + val okAt = + async { + for (raw in publisher.incoming) { + val node = runCatching { mapper.readTree(raw) }.getOrNull() ?: continue + if (node[0]?.asText() == "OK" && node[1]?.asText() == event.id) { + return@async Pair(System.nanoTime(), node[2]?.asBoolean() == true) + } + } + Pair(0L, false) + } + + val visibleAt = + async { + var polls = 0 + while (true) { + val subId = "v$index-${polls++}" + prober.req(subId, """{"ids":["${event.id}"]}""") + var foundAt = 0L + for (raw in prober.incoming) { + val node = runCatching { mapper.readTree(raw) }.getOrNull() ?: continue + if (node[1]?.asText() != subId) continue + when (node[0]?.asText()) { + "EVENT" -> foundAt = System.nanoTime() + "EOSE", "CLOSED" -> { + prober.close(subId) + if (foundAt != 0L) return@async foundAt + break + } + } + } + } + @Suppress("UNREACHABLE_CODE") + 0L + } + + check(publisher.publish(OptimizedJsonMapper.toJson(event))) { "publish failed" } + + val (okTime, accepted) = okAt.await() + if (!accepted) { + visibleAt.cancelAndJoin() + return@coroutineScope null + } + val visibleTime = visibleAt.await() + Sample(okTime - start, visibleTime - start, visibleTime <= okTime) + } + } + + suspend fun throughput( + wsUrl: String, + events: List, + connections: Int, + window: Int, + http: OkHttpClient, + ): ThroughputResult { + val accepted = AtomicInteger() + val rejected = AtomicInteger() + val slices = List(connections) { c -> events.filterIndexed { i, _ -> i % connections == c } } + + val start = System.nanoTime() + withContext(Dispatchers.IO) { + slices + .filter { it.isNotEmpty() } + .map { slice -> async { publishSlice(wsUrl, slice, window, http, accepted, rejected) } } + .awaitAll() + } + val wallMs = (System.nanoTime() - start) / 1_000_000.0 + return ThroughputResult( + events = events.size, + accepted = accepted.get(), + rejected = rejected.get(), + wallMs = wallMs, + eventsPerSec = events.size / (wallMs / 1000.0), + connections = connections, + ) + } + + private suspend fun publishSlice( + wsUrl: String, + slice: List, + window: Int, + http: OkHttpClient, + accepted: AtomicInteger, + rejected: AtomicInteger, + ) { + val socket = NostrSocket.connect(http, wsUrl) + try { + coroutineScope { + val inFlight = Semaphore(window) + val reader = + launch { + var acked = 0 + for (raw in socket.incoming) { + // Fast-path: only OK frames matter here. + if (!raw.startsWith("[\"OK\"")) continue + val node = runCatching { mapper.readTree(raw) }.getOrNull() ?: continue + if (node[2]?.asBoolean() == true) accepted.incrementAndGet() else rejected.incrementAndGet() + inFlight.release() + if (++acked == slice.size) break + } + } + for (event in slice) { + inFlight.acquire() + check(socket.publish(OptimizedJsonMapper.toJson(event))) { "publish failed (socket closed?)" } + } + reader.join() + } + } finally { + socket.disconnect() + } + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/QueryBenchmark.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/QueryBenchmark.kt new file mode 100644 index 0000000000..6fd2daedd9 --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/QueryBenchmark.kt @@ -0,0 +1,163 @@ +/* + * 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.relaybench.bench + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.vitorpamplona.relaybench.NostrSocket +import com.vitorpamplona.relaybench.Percentiles +import com.vitorpamplona.relaybench.Scenario +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout +import okhttp3.OkHttpClient + +/** + * Read-path metrics for one [Scenario]: + * + * - **sequential**: R fresh REQ→EOSE rounds on a single connection — + * time-to-first-event and time-to-EOSE percentiles, plus the derived + * events/second a single client experiences. + * - **concurrent**: the same REQ issued from C connections at once, each + * running R rounds — aggregate events/second, i.e. how the relay holds + * up when a popular filter is hot. + * + * The number of events each round returns is recorded so the harness can + * flag relays that disagree about the result set (a correctness smell that + * would invalidate the speed comparison). + */ +object QueryBenchmark { + private val mapper = jacksonObjectMapper() + + data class ScenarioResult( + val scenario: Scenario, + val eventsPerRound: Int, + val timeToFirst: Percentiles, + val timeToEose: Percentiles, + val sequentialEventsPerSec: Double, + val concurrentEventsPerSec: Double, + val concurrentConnections: Int, + ) + + suspend fun run( + wsUrl: String, + scenario: Scenario, + warmupRounds: Int, + rounds: Int, + concurrency: Int, + concurrentRounds: Int, + http: OkHttpClient, + ): ScenarioResult { + // --- sequential --- + val socket = NostrSocket.connect(http, wsUrl) + val firstNanos = ArrayList(rounds) + val eoseNanos = ArrayList(rounds) + var eventsPerRound = 0 + try { + repeat(warmupRounds + rounds) { round -> + val r = oneRound(socket, scenario, "q$round") + if (round >= warmupRounds) { + firstNanos += r.firstEventNanos + eoseNanos += r.eoseNanos + eventsPerRound = r.events + } + } + } finally { + socket.disconnect() + } + val totalSeqSec = eoseNanos.sum() / 1_000_000_000.0 + val seqEps = if (totalSeqSec > 0) eventsPerRound * rounds / totalSeqSec else 0.0 + + // --- concurrent --- + var concEps = 0.0 + if (concurrency > 1) { + val start = System.nanoTime() + val counts = + withContext(Dispatchers.IO) { + (0 until concurrency) + .map { c -> + async { + val s = NostrSocket.connect(http, wsUrl) + try { + var total = 0 + repeat(concurrentRounds) { round -> + total += oneRound(s, scenario, "c$c-$round").events + } + total + } finally { + s.disconnect() + } + } + }.awaitAll() + } + val wallSec = (System.nanoTime() - start) / 1_000_000_000.0 + concEps = if (wallSec > 0) counts.sum() / wallSec else 0.0 + } + + return ScenarioResult( + scenario = scenario, + eventsPerRound = eventsPerRound, + timeToFirst = Percentiles.ofNanos(firstNanos), + timeToEose = Percentiles.ofNanos(eoseNanos), + sequentialEventsPerSec = seqEps, + concurrentEventsPerSec = concEps, + concurrentConnections = concurrency, + ) + } + + private class Round( + val events: Int, + val firstEventNanos: Long, + val eoseNanos: Long, + ) + + private suspend fun oneRound( + socket: NostrSocket, + scenario: Scenario, + subId: String, + ): Round = + withTimeout(60_000) { + val start = System.nanoTime() + socket.req(subId, scenario.filterJson) + var events = 0 + var firstAt = 0L + for (raw in socket.incoming) { + // Cheap dispatch on the frame prefix; EVENT frames don't + // need full JSON parsing just to be counted. + if (raw.startsWith("[\"EVENT\"")) { + if (firstAt == 0L) firstAt = System.nanoTime() + events++ + continue + } + val node = runCatching { mapper.readTree(raw) }.getOrNull() ?: continue + val type = node[0]?.asText() + if ((type == "EOSE" || type == "CLOSED") && node[1]?.asText() == subId) break + } + val eoseAt = System.nanoTime() + socket.close(subId) + Round( + events = events, + firstEventNanos = (if (firstAt == 0L) eoseAt else firstAt) - start, + eoseNanos = eoseAt - start, + ) + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/SyncBenchmark.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/SyncBenchmark.kt new file mode 100644 index 0000000000..a002ddaf37 --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/SyncBenchmark.kt @@ -0,0 +1,281 @@ +/* + * 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.relaybench.bench + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage +import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession +import com.vitorpamplona.relaybench.NostrSocket +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.withTimeout +import okhttp3.OkHttpClient + +/** + * NIP-77 negentropy sync between two relays, driven the way `strfry sync` + * drives it: this harness plays the role of the syncing agent that holds + * one relay's dataset locally and reconciles it against the other relay. + * + * Per pair, both relays get an 80% slice of the corpus with 60% overlap; + * then we measure, in order: + * + * 1. **reconcile** against each relay as the NEG server (time, NEG-MSG + * round trips, wire bytes) — this is the interesting server-side + * data structure work; + * 2. **delta transfer** — REQ the missing events off one side and + * publish them to the other, both directions, until every OK lands + * (plain NIP-01 follow-up, dominated by ingest speed); + * 3. **re-reconcile with identical sets** — the steady-state "nothing to + * do" sync cost that a periodic mirror would pay. + * + * Reconciliation runs over the *effective* corpus — replaceable kinds + * collapsed to their newest version — because that's what a relay stores; + * superseded versions would otherwise show up as phantom diffs and the + * pair would never converge. + */ +object SyncBenchmark { + private val mapper = jacksonObjectMapper() + + data class ReconcileStats( + val ms: Double, + val rounds: Int, + val wireBytes: Long, + val needCount: Int, + val haveCount: Int, + val error: String?, + ) + + data class PairResult( + val serverA: String, + val serverB: String, + val syncableEvents: Int, + val initialReconcile: Map, + val transferMs: Double, + val transferredToA: Int, + val transferredToB: Int, + val identicalReconcile: Map, + val converged: Boolean, + val error: String? = null, + ) + + /** Newest version per replaceable/addressable key; everything else untouched. */ + fun effectiveEvents(events: List): List { + val byKey = LinkedHashMap(events.size) + for (e in events) { + val key = + when { + e.kind == 0 || e.kind == 3 || e.kind in 10000..19999 -> "${e.kind}:${e.pubKey}" + e.kind in 30000..39999 -> { + val d = e.tags.firstOrNull { it.size >= 2 && it[0] == "d" }?.get(1) ?: "" + "${e.kind}:${e.pubKey}:$d" + } + else -> e.id + } + val prev = byKey[key] + if (prev == null || e.createdAt > prev.createdAt || (e.createdAt == prev.createdAt && e.id < prev.id)) { + byKey[key] = e + } + } + return byKey.values.toList() + } + + class NegotiationResult( + val haveIds: Set, + val needIds: Set, + val rounds: Int, + val wireBytes: Long, + val ms: Double, + val error: String?, + ) + + /** One full NEG-OPEN → NEG-MSG* → NEG-CLOSE negotiation. */ + suspend fun reconcile( + wsUrl: String, + localEvents: List, + http: OkHttpClient, + maxRounds: Int = 128, + ): NegotiationResult { + val socket = NostrSocket.connect(http, wsUrl) + val haveIds = mutableSetOf() + val needIds = mutableSetOf() + var rounds = 0 + var bytes = 0L + val start = System.nanoTime() + + fun elapsedMs() = (System.nanoTime() - start) / 1_000_000.0 + + try { + val session = NegentropySession("bench-sync", Filter(), localEvents, frameSizeLimit = 0) + val open = OptimizedJsonMapper.toJson(session.open()) + bytes += open.length + check(socket.send(open)) { "send NEG-OPEN failed" } + + while (rounds < maxRounds) { + val raw = withTimeout(60_000) { socket.incoming.receive() } + if (raw.startsWith("[\"EVENT\"") || raw.startsWith("[\"OK\"")) continue + bytes += raw.length + rounds++ + when (val msg = OptimizedJsonMapper.fromJsonToMessage(raw)) { + is NegErrMessage -> return NegotiationResult(haveIds, needIds, rounds, bytes, elapsedMs(), "NEG-ERR: ${msg.reason}") + is NoticeMessage -> return NegotiationResult(haveIds, needIds, rounds, bytes, elapsedMs(), "NOTICE: ${msg.message}") + is NegMsgMessage -> { + val r = session.processMessage(msg.message) + haveIds += r.haveIds + needIds += r.needIds + if (r.isComplete()) { + socket.send("""["NEG-CLOSE","bench-sync"]""") + return NegotiationResult(haveIds, needIds, rounds, bytes, elapsedMs(), null) + } + val next = OptimizedJsonMapper.toJson(r.nextCmd!!) + bytes += next.length + check(socket.send(next)) { "send NEG-MSG failed" } + } + else -> {} // stray frame from another subsystem; ignore + } + } + return NegotiationResult(haveIds, needIds, rounds, bytes, elapsedMs(), "no convergence in $maxRounds rounds") + } finally { + socket.disconnect() + } + } + + /** REQ [ids] off [fromUrl] in ≤100-id batches; returns the events. */ + suspend fun fetchByIds( + fromUrl: String, + ids: Collection, + http: OkHttpClient, + ): List { + if (ids.isEmpty()) return emptyList() + val socket = NostrSocket.connect(http, fromUrl) + val fetched = ArrayList(ids.size) + try { + for ((batchIdx, batch) in ids.chunked(100).withIndex()) { + withTimeout(60_000) { + val subId = "fetch-$batchIdx" + socket.req(subId, Filter(ids = batch).toJson()) + for (raw in socket.incoming) { + val node = runCatching { mapper.readTree(raw) }.getOrNull() ?: continue + if (node[1]?.asText() != subId) continue + when (node[0]?.asText()) { + "EVENT" -> + runCatching { OptimizedJsonMapper.fromJson(node[2].toString()) } + .getOrNull() + ?.let { fetched.add(it) } + "EOSE", "CLOSED" -> { + socket.close(subId) + return@withTimeout + } + } + } + } + } + } finally { + socket.disconnect() + } + return fetched + } + + /** + * Full pair benchmark. [urlA]/[urlB] must already hold [sliceA]/[sliceB] + * (both slices of [effective], which must be replaceable-collapsed). + */ + suspend fun runPair( + nameA: String, + urlA: String, + nameB: String, + urlB: String, + effective: List, + sliceA: List, + sliceB: List, + window: Int, + http: OkHttpClient, + log: (String) -> Unit, + ): PairResult { + val idsA = sliceA.mapTo(HashSet()) { it.id } + val idsB = sliceB.mapTo(HashSet()) { it.id } + + // 1. Initial reconcile against each side as server. + log(" reconciling (local=$nameA's ${sliceA.size} events) against $nameB…") + val againstB = reconcile(urlB, sliceA, http) + log(" ${againstB.rounds} rounds, need=${againstB.needIds.size} have=${againstB.haveIds.size} in ${"%.0f".format(againstB.ms)} ms") + log(" reconciling (local=$nameB's ${sliceB.size} events) against $nameA…") + val againstA = reconcile(urlA, sliceB, http) + log(" ${againstA.rounds} rounds, need=${againstA.needIds.size} have=${againstA.haveIds.size} in ${"%.0f".format(againstA.ms)} ms") + + val expectedOnlyB = idsB.count { it !in idsA } + val expectedOnlyA = idsA.count { it !in idsB } + val diffCorrect = + againstB.error == null && + againstA.error == null && + againstB.needIds.size == expectedOnlyB && + againstB.haveIds.size == expectedOnlyA && + againstA.needIds.size == expectedOnlyA && + againstA.haveIds.size == expectedOnlyB + + // 2. Delta transfer: close the diff in both directions. + val transferStart = System.nanoTime() + val byId = effective.associateBy { it.id } + val toA = fetchByIds(urlB, againstB.needIds, http) + val toB = againstB.haveIds.mapNotNull { byId[it] } + coroutineScope { + listOf( + async { if (toA.isNotEmpty()) IngestBenchmark.throughput(urlA, toA, 1, window, http) }, + async { if (toB.isNotEmpty()) IngestBenchmark.throughput(urlB, toB, 1, window, http) }, + ).awaitAll() + } + val transferMs = (System.nanoTime() - transferStart) / 1_000_000.0 + log(" delta transfer: ${toA.size} → $nameA, ${toB.size} → $nameB in ${"%.0f".format(transferMs)} ms") + + // 3. Steady-state: reconcile identical sets. + log(" re-reconciling identical sets…") + val identicalA = reconcile(urlA, effective, http) + val identicalB = reconcile(urlB, effective, http) + val converged = + diffCorrect && + identicalA.error == null && + identicalB.error == null && + identicalA.needIds.isEmpty() && + identicalA.haveIds.isEmpty() && + identicalB.needIds.isEmpty() && + identicalB.haveIds.isEmpty() + + fun stats(r: NegotiationResult) = ReconcileStats(r.ms, r.rounds, r.wireBytes, r.needIds.size, r.haveIds.size, r.error) + + return PairResult( + serverA = nameA, + serverB = nameB, + syncableEvents = effective.size, + initialReconcile = mapOf(nameA to stats(againstA), nameB to stats(againstB)), + transferMs = transferMs, + transferredToA = toA.size, + transferredToB = toB.size, + identicalReconcile = mapOf(nameA to stats(identicalA), nameB to stats(identicalB)), + converged = converged, + ) + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/Corpus.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/Corpus.kt new file mode 100644 index 0000000000..ff973a792c --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/Corpus.kt @@ -0,0 +1,148 @@ +/* + * 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.relaybench.corpus + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import java.io.File +import java.security.MessageDigest + +/** + * The benchmark corpus: a flat, chronologically ordered list of *fully + * signed* Nostr events plus where it came from. + * + * Corpora are cached/shared as NDJSON (one event JSON per line) with a + * `manifest.json` sibling describing how they were built and a fingerprint + * (sha256 over all event ids) so two parties can confirm they benchmarked + * against the same data. Synthetic corpora are fully deterministic given + * (seed, size, baseTime) — same inputs, same event ids on any machine — + * which is what makes results comparable across relay implementations and + * across the community. + */ +class Corpus( + val events: List, + val source: String, + val spec: CorpusSpec?, +) { + val fingerprint: String by lazy { + val digest = MessageDigest.getInstance("SHA-256") + events.forEach { digest.update(it.id.toByteArray()) } + digest.digest().joinToString("") { "%02x".format(it) }.take(16) + } + + fun kindHistogram(): Map = events.groupingBy { it.kind }.eachCount().toSortedMap() +} + +/** Deterministic inputs of a synthetic corpus. */ +data class CorpusSpec( + val seed: Long, + val events: Int, + /** + * Newest timestamp in the corpus. Fixed by default (not wall clock) so + * the same spec yields byte-identical event ids forever. Must stay + * inside strfry's default accept window (now - 3 years, now + 15 min). + */ + val baseTime: Long, + /** Corpus spans this many seconds ending at [baseTime]. */ + val spanSeconds: Long, +) { + fun cacheName(): String = "corpus-v$VERSION-seed$seed-n$events-t$baseTime.ndjson" + + companion object { + /** Bump when the generator's output changes for the same spec. */ + const val VERSION = 1 + + /** 2026-06-01T00:00:00Z. */ + const val DEFAULT_BASE_TIME = 1_780_272_000L + + const val DEFAULT_SPAN_SECONDS = 30L * 24 * 3600 + } +} + +object CorpusIO { + fun write( + file: File, + corpus: Corpus, + ) { + file.parentFile?.mkdirs() + file.bufferedWriter().use { out -> + corpus.events.forEach { event -> + out.write(OptimizedJsonMapper.toJson(event)) + out.write("\n") + } + } + writeManifest(File(file.parentFile, file.nameWithoutExtension + ".manifest.json"), corpus) + } + + fun read( + file: File, + source: String = file.name, + ): Corpus { + var skipped = 0 + val events = + file.useLines { lines -> + lines + .filter { it.isNotBlank() } + .mapNotNull { line -> + runCatching { OptimizedJsonMapper.fromJson(line) } + .onFailure { skipped++ } + .getOrNull() + }.toList() + } + if (skipped > 0) println(" ! skipped $skipped unparseable lines in ${file.name}") + return Corpus(events, source, spec = null) + } + + private fun writeManifest( + file: File, + corpus: Corpus, + ) { + val kinds = + corpus + .kindHistogram() + .entries + .joinToString(",\n") { (k, v) -> " \"$k\": $v" } + val specJson = + corpus.spec?.let { + """ + | "generator": "amethyst-relaybench", + | "generatorVersion": ${CorpusSpec.VERSION}, + | "seed": ${it.seed}, + | "baseTime": ${it.baseTime}, + | "spanSeconds": ${it.spanSeconds}, + """.trimMargin() + } ?: " \"generator\": \"external\"," + file.writeText( + """ + |{ + | "source": "${corpus.source}", + |$specJson + | "events": ${corpus.events.size}, + | "fingerprint": "${corpus.fingerprint}", + | "kinds": { + |$kinds + | } + |} + | + """.trimMargin(), + ) + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusDownloader.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusDownloader.kt new file mode 100644 index 0000000000..de74cfd44e --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusDownloader.kt @@ -0,0 +1,140 @@ +/* + * 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.relaybench.corpus + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +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.relaybench.NostrSocket +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull +import okhttp3.OkHttpClient +import java.io.File + +/** + * Assembles a fresh real-world corpus by paginating recent events out of + * public relays. Complements the checked-in dump: use this when the corpus + * should reflect *today's* event mix. The result goes through the same + * [CorpusSource.prepare] pipeline (dedup, verify, filter) as every other + * source, then is cached as NDJSON. + */ +object CorpusDownloader { + val DEFAULT_RELAYS = listOf("wss://relay.damus.io", "wss://nos.lol", "wss://relay.primal.net") + + private val KIND_BUCKETS = listOf(listOf(0, 3), listOf(1), listOf(6, 7), listOf(9735), listOf(30023, 1111)) + private val mapper = jacksonObjectMapper() + + fun download( + relayUrls: List, + target: Int, + cacheDir: File, + http: OkHttpClient, + log: (String) -> Unit, + ): Corpus { + val key = "corpus-download-${relayUrls.hashCode()}-n$target.ndjson" + val cached = File(cacheDir, key) + if (cached.exists()) { + log(" reusing downloaded corpus ${cached.name}") + return CorpusIO.read(cached, source = "download:${relayUrls.joinToString(",")}") + } + + val perRelay = (target * 2 / relayUrls.size).coerceAtLeast(500) + val collected = LinkedHashMap(target * 2) + runBlocking { + relayUrls + .map { url -> + async { + runCatching { downloadFrom(url, perRelay, log) } + .onFailure { log(" ! $url failed: ${it.message}") } + .getOrDefault(emptyList()) + } + }.awaitAll() + }.flatten().forEach { collected.putIfAbsent(it.id, it) } + + log(" downloaded ${collected.size} unique events from ${relayUrls.size} relays") + val corpus = CorpusSource.prepare(collected.values.toList(), target, "download:${relayUrls.joinToString(",")}", log) + CorpusIO.write(cached, corpus) + return corpus + } + + private suspend fun downloadFrom( + url: String, + target: Int, + log: (String) -> Unit, + ): List { + val http = OkHttpClient.Builder().build() + val socket = NostrSocket.connect(http, url) + val events = ArrayList(target) + try { + for (kinds in KIND_BUCKETS) { + var until: Long? = null + var pages = 0 + val quota = target / KIND_BUCKETS.size + var got = 0 + while (got < quota && pages < 40) { + val filter = Filter(kinds = kinds, limit = 500, until = until) + val page = requestPage(socket, "dl-${kinds.first()}-$pages", filter) ?: break + if (page.isEmpty()) break + events += page + got += page.size + until = page.minOf { it.createdAt } - 1 + pages++ + } + } + } finally { + socket.disconnect() + http.dispatcher.executorService.shutdown() + } + log(" $url: ${events.size} events") + return events + } + + /** One REQ page; null on timeout or socket close. */ + private suspend fun requestPage( + socket: NostrSocket, + subId: String, + filter: Filter, + ): List? = + withTimeoutOrNull(30_000) { + socket.req(subId, filter.toJson()) + val page = ArrayList(filter.limit ?: 500) + for (raw in socket.incoming) { + val node = runCatching { mapper.readTree(raw) }.getOrNull() ?: continue + when (node[0]?.asText()) { + "EVENT" -> + if (node[1]?.asText() == subId) { + runCatching { OptimizedJsonMapper.fromJson(node[2].toString()) } + .getOrNull() + ?.let { page.add(it) } + } + "EOSE", "CLOSED" -> + if (node[1]?.asText() == subId) { + socket.close(subId) + return@withTimeoutOrNull page + } + } + } + page + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusGenerator.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusGenerator.kt new file mode 100644 index 0000000000..dbc2496731 --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusGenerator.kt @@ -0,0 +1,369 @@ +/* + * 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.relaybench.corpus + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.core.toHexKey +import com.vitorpamplona.quartz.nip01Core.crypto.EventAssembler +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.runBlocking +import java.security.MessageDigest +import kotlin.math.min +import kotlin.math.pow +import kotlin.random.Random + +/** + * Deterministic synthetic corpus with a realistic social shape: + * + * - Zipf-distributed author popularity (a few loud accounts, a long tail). + * - Profiles (kind 0) + contact lists (kind 3) for every author. + * - Notes (kind 1), ~1/3 of them replies clustered into threads. + * - Reposts (kind 6), reactions (kind 7) and zap request/receipt pairs + * (9734 inside 9735) targeting earlier notes with a recency bias, so + * "hot threads" and "popular people" exist for the query scenarios. + * - ~15% of notes carry `t` hashtags drawn from a small popular set. + * + * Every random draw comes from a single seeded [Random] consumed in a fixed + * order, timestamps derive from the spec's fixed baseTime, and even the + * BIP-340 aux nonces are seed-derived — so the same [CorpusSpec] reproduces + * a byte-identical corpus (ids *and* signatures) on any machine. That is + * what lets two parties compare relay results against "corpus seed 1, + * n=10000" without shipping the file. (Deterministic nonces are fine here: + * each signature covers a distinct message, and these keys sign nothing + * outside the benchmark.) + * + * Signing is the expensive part and is parallelized in three dependency + * waves (referenced ids must exist before the referencing template is + * finalized): 1) profiles/contacts/root notes, 2) replies, 3) reactions, + * reposts and zaps. + */ +object CorpusGenerator { + private val WORDS = + ( + "the of to and in that for on with as it is was at by from be this have or an are not you your " + + "we they he she will one all would there what so up out if about who get which go me when make " + + "can like time no just him know take people into year good some could them see other than then " + + "now look only come its over think also back after use two how our work first well way even new " + + "want because any these give day most us nostr relay zap note client protocol key sign event " + + "freedom bitcoin coffee build ship test run friend photo music idea question answer thanks great" + ).split(" ") + + private val HASHTAGS = + listOf( + "nostr", + "bitcoin", + "introductions", + "foodstr", + "plebchain", + "art", + "music", + "photography", + "zapathon", + "asknostr", + ) + + private val REACTIONS = listOf("+", "+", "+", "+", "🤙", "❤️", "🫂", "😂") + + private enum class SlotType { PROFILE, CONTACTS, ROOT_NOTE, REPLY, REPOST, REACTION, ZAP } + + private class Slot( + val type: SlotType, + val createdAt: Long, + val actor: Int, + var targetSlot: Int = -1, + var rootSlot: Int = -1, + var text: String = "", + var hashtags: List = emptyList(), + var follows: List = emptyList(), + ) + + fun generate( + spec: CorpusSpec, + log: (String) -> Unit = {}, + ): Corpus { + val rng = Random(spec.seed) + val n = spec.events + val authorCount = (n / 20).coerceIn(20, 2000) + + // Sequential draws → deterministic keys. + val authorKeys = List(authorCount) { KeyPair(privKey = rng.nextBytes(32)) } + val authorHex = authorKeys.map { it.pubKey.toHexKey() } + val zapperKeys = List(min(5, authorCount)) { KeyPair(privKey = rng.nextBytes(32)) } + val zapperHex = zapperKeys.map { it.pubKey.toHexKey() } + + /** Seed-derived BIP-340 aux nonce, unique per (slot, sub-signature). */ + fun nonce( + slot: Int, + sub: Int, + ): ByteArray = + MessageDigest + .getInstance("SHA-256") + .digest("relaybench-nonce:${spec.seed}:$slot:$sub".toByteArray()) + + fun sign( + key: KeyPair, + pubHex: String, + slot: Int, + sub: Int, + createdAt: Long, + kind: Int, + tags: Array>, + content: String, + ): Event = EventAssembler.hashAndSign(pubHex, createdAt, kind, tags, content, key.privKey!!, nonce(slot, sub)) + + // Zipf-ish popularity: weight of rank r ∝ 1/(r+1)^0.9. + val weights = DoubleArray(authorCount) { 1.0 / (it + 1.0).pow(0.9) } + val cumulative = DoubleArray(authorCount) + var acc = 0.0 + for (i in weights.indices) { + acc += weights[i] + cumulative[i] = acc + } + + fun sampleAuthor(): Int { + val x = rng.nextDouble() * acc + val idx = cumulative.toList().binarySearch { it.compareTo(x) } + return if (idx >= 0) idx else (-idx - 1).coerceAtMost(authorCount - 1) + } + + fun sentence(): String { + val count = 5 + rng.nextInt(35) + return (0 until count).joinToString(" ") { WORDS[rng.nextInt(WORDS.size)] } + } + + // ---- Pass 1: plan every slot sequentially (all randomness here). ---- + val step = spec.spanSeconds.toDouble() / n + + fun createdAt(slot: Int): Long = + spec.baseTime - spec.spanSeconds + (slot * step).toLong() + + rng.nextLong(0, (step * 0.9).toLong().coerceAtLeast(1)) + + val bulkCount = (n - 2 * authorCount).coerceAtLeast(0) + val bulkTypes = ArrayList(bulkCount) + repeat(bulkCount) { + val x = rng.nextDouble() + bulkTypes += + when { + x < 0.36 -> SlotType.ROOT_NOTE + x < 0.55 -> SlotType.REPLY + x < 0.63 -> SlotType.REPOST + x < 0.93 -> SlotType.REACTION + else -> SlotType.ZAP + } + } + + val slots = ArrayList(n) + // Profiles and contact lists are each author's oldest events. + for (a in 0 until authorCount) { + if (slots.size >= n) break + slots += Slot(SlotType.PROFILE, createdAt(slots.size), a) + if (slots.size >= n) break + val followCount = 10 + rng.nextInt(min(120, authorCount)) + val follows = (0 until followCount).map { sampleAuthor() }.distinct() + slots += + Slot(SlotType.CONTACTS, createdAt(slots.size), a).apply { this.follows = follows } + } + + val noteSlots = ArrayList() // indices of ROOT_NOTE and REPLY slots + + fun sampleNoteSlot(): Int { + // 70% recency-biased (hot content), 30% uniform (long tail). + val size = noteSlots.size + return if (rng.nextDouble() < 0.7) { + val back = (rng.nextDouble().pow(2) * min(size, 500)).toInt() + noteSlots[size - 1 - back.coerceAtMost(size - 1)] + } else { + noteSlots[rng.nextInt(size)] + } + } + + for (type in bulkTypes) { + if (slots.size >= n) break + val idx = slots.size + val effective = if (noteSlots.isEmpty() && type != SlotType.ROOT_NOTE) SlotType.ROOT_NOTE else type + val slot = Slot(effective, createdAt(idx), sampleAuthor()) + when (effective) { + SlotType.ROOT_NOTE -> { + slot.text = sentence() + if (rng.nextDouble() < 0.15) { + slot.hashtags = List(1 + rng.nextInt(2)) { HASHTAGS[(rng.nextDouble().pow(2) * HASHTAGS.size).toInt()] }.distinct() + } + noteSlots += idx + } + SlotType.REPLY -> { + slot.text = sentence() + slot.targetSlot = sampleNoteSlot() + slot.rootSlot = + slots[slot.targetSlot].let { t -> if (t.type == SlotType.REPLY) t.rootSlot else slot.targetSlot } + noteSlots += idx + } + SlotType.REPOST, SlotType.REACTION, SlotType.ZAP -> { + slot.targetSlot = sampleNoteSlot() + if (effective == SlotType.REACTION) slot.text = REACTIONS[rng.nextInt(REACTIONS.size)] + } + else -> {} + } + slots += slot + } + + // ---- Pass 2: sign in dependency waves, in parallel. ---- + val events = arrayOfNulls(slots.size) + + fun signSlot(i: Int): Event { + val s = slots[i] + val key = authorKeys[s.actor] + val pub = authorHex[s.actor] + return when (s.type) { + SlotType.PROFILE -> + sign( + key, + pub, + i, + 0, + s.createdAt, + 0, + emptyArray(), + """{"name":"user-${s.actor}","about":"synthetic relaybench profile ${s.actor}","picture":"https://example.com/${s.actor}.png"}""", + ) + SlotType.CONTACTS -> + sign( + key, + pub, + i, + 0, + s.createdAt, + 3, + s.follows.map { arrayOf("p", authorHex[it]) }.toTypedArray(), + "", + ) + SlotType.ROOT_NOTE -> + sign( + key, + pub, + i, + 0, + s.createdAt, + 1, + s.hashtags.map { arrayOf("t", it) }.toTypedArray(), + if (s.hashtags.isEmpty()) s.text else s.text + " " + s.hashtags.joinToString(" ") { "#$it" }, + ) + SlotType.REPLY -> { + val root = events[s.rootSlot]!! + sign( + key, + pub, + i, + 0, + s.createdAt, + 1, + arrayOf(arrayOf("e", root.id, "", "root"), arrayOf("p", root.pubKey)), + s.text, + ) + } + SlotType.REPOST -> { + val target = events[s.targetSlot]!! + sign( + key, + pub, + i, + 0, + s.createdAt, + 6, + arrayOf(arrayOf("e", target.id), arrayOf("p", target.pubKey)), + OptimizedJsonMapper.toJson(target), + ) + } + SlotType.REACTION -> { + val target = events[s.targetSlot]!! + sign( + key, + pub, + i, + 0, + s.createdAt, + 7, + arrayOf(arrayOf("e", target.id), arrayOf("p", target.pubKey), arrayOf("k", "1")), + s.text, + ) + } + SlotType.ZAP -> { + val target = events[s.targetSlot]!! + val request = + sign( + key, + pub, + i, + 0, + s.createdAt, + 9734, + arrayOf(arrayOf("e", target.id), arrayOf("p", target.pubKey), arrayOf("relays", "ws://localhost")), + "", + ) + val zapperIdx = s.actor % zapperKeys.size + sign( + zapperKeys[zapperIdx], + zapperHex[zapperIdx], + i, + 1, + s.createdAt, + 9735, + arrayOf( + arrayOf("p", target.pubKey), + arrayOf("e", target.id), + arrayOf("bolt11", "lnbc210n1relaybenchfake${s.actor}"), + arrayOf("description", OptimizedJsonMapper.toJson(request)), + ), + "", + ) + } + } + } + + val waves = + listOf( + slots.indices.filter { slots[it].type in setOf(SlotType.PROFILE, SlotType.CONTACTS, SlotType.ROOT_NOTE) }, + slots.indices.filter { slots[it].type == SlotType.REPLY }, + slots.indices.filter { slots[it].type in setOf(SlotType.REPOST, SlotType.REACTION, SlotType.ZAP) }, + ) + val start = System.nanoTime() + runBlocking { + waves.forEach { wave -> + coroutineScope { + wave + .chunked(512) + .map { chunk -> + async(Dispatchers.Default) { chunk.forEach { events[it] = signSlot(it) } } + }.awaitAll() + } + } + } + val signed = events.filterNotNull() + log(" signed ${signed.size} events (${zapCount(signed)} of them zap receipts embed a signed 9734) in ${(System.nanoTime() - start) / 1_000_000} ms") + return Corpus(signed, "synthetic seed=${spec.seed} n=${spec.events}", spec) + } + + private fun zapCount(events: List) = events.count { it.kind == 9735 } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusSource.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusSource.kt new file mode 100644 index 0000000000..79702a99ed --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusSource.kt @@ -0,0 +1,239 @@ +/* + * 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.relaybench.corpus + +import com.fasterxml.jackson.core.JsonFactory +import com.fasterxml.jackson.core.JsonToken +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.ObjectMapper +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.crypto.verify +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.runBlocking +import java.io.File +import java.io.PushbackInputStream +import java.util.zip.GZIPInputStream + +/** + * Turns any raw event source (deterministic generator, NDJSON file, a JSON + * array dump like quartz's `nostr_vitor_startup_data.json.gz`, or a live + * download) into a *benchmark-ready* corpus, and caches the result so + * subsequent runs skip the expensive preparation. + * + * Preparation makes the corpus fair and hang-proof for every relay: + * - dedup by id (multi-relay dumps repeat events heavily), + * - drop unsigned events (NIP-17 kind-14 rumors are unsigned by design), + * - drop kind-5 deletions (order-dependent side effects would make query + * counts diverge between relays for reasons unrelated to performance), + * - drop ephemeral kinds (accepted but never queryable — would deadlock + * the visibility probe), + * - drop events beyond strfry's default hard limits (64 KiB serialized, + * 2000 tags, 1 KiB tag values) so both relays are offered an identical, + * fully ingestable stream, + * - verify every Schnorr signature in parallel and drop invalid ones (a + * verifying relay would reject them), + * - sort chronologically (stable) so replaceable-event supersession is + * deterministic. + */ +object CorpusSource { + const val DEFAULT_MAX_EVENT_BYTES = 65536 + const val DEFAULT_MAX_TAGS = 2000 + private const val MAX_TAG_VALUE_BYTES = 1024 + + fun synthetic( + spec: CorpusSpec, + cacheDir: File, + log: (String) -> Unit, + ): Corpus { + val cached = File(cacheDir, spec.cacheName()) + if (cached.exists()) { + log(" reusing cached corpus ${cached.name}") + return Corpus(CorpusIO.read(cached).events, "synthetic seed=${spec.seed} n=${spec.events}", spec) + } + log(" generating deterministic synthetic corpus (seed=${spec.seed}, n=${spec.events})…") + val corpus = CorpusGenerator.generate(spec, log) + CorpusIO.write(cached, corpus) + log(" cached to ${cached.path}") + return corpus + } + + fun fromFile( + file: File, + limit: Int, + cacheDir: File, + log: (String) -> Unit, + maxEventBytes: Int = DEFAULT_MAX_EVENT_BYTES, + maxTags: Int = DEFAULT_MAX_TAGS, + ): Corpus { + val key = "corpus-file-${file.name}-${file.length()}-limit$limit-b$maxEventBytes-t$maxTags.ndjson" + val cached = File(cacheDir, key) + if (cached.exists()) { + log(" reusing prepared corpus ${cached.name}") + return CorpusIO.read(cached, source = "file:${file.name}") + } + log(" reading ${file.name}…") + // For multi-million-event dumps a --limit also caps how much we + // read (with headroom for events preparation will drop), keeping + // memory bounded instead of materializing gigabytes. + val readCap = if (limit > 0) (limit * 3 / 2) + 1000 else Int.MAX_VALUE + val raw = readAnyFormat(file, readCap) + log(" ${raw.size} raw events read; preparing (dedup, filter, verify signatures)…") + val corpus = prepare(raw, limit, "file:${file.name}", log, maxEventBytes, maxTags) + CorpusIO.write(cached, corpus) + log(" cached prepared corpus to ${cached.path}") + return corpus + } + + /** + * Reads NDJSON or a single JSON array, gzipped or plain (sniffed by + * magic bytes, not filename — Drive exports lie about extensions), + * streaming up to [readCap] events. + */ + private fun readAnyFormat( + file: File, + readCap: Int, + ): List { + val base = PushbackInputStream(file.inputStream().buffered(1 shl 16), 2) + val magic = ByteArray(2) + val read = base.read(magic) + if (read > 0) base.unread(magic, 0, read) + val isGzip = read == 2 && magic[0] == 0x1f.toByte() && magic[1] == 0x8b.toByte() + val stream = if (isGzip) GZIPInputStream(base, 1 shl 16) else base + stream.use { input -> + val pushback = PushbackInputStream(input, 1) + var first: Int + do { + first = pushback.read() + } while (first != -1 && Character.isWhitespace(first)) + if (first == -1) return emptyList() + pushback.unread(first) + return if (first.toChar() == '[') { + // One huge JSON array — stream element by element. + val mapper = ObjectMapper() + val parser = JsonFactory().createParser(pushback) + parser.codec = mapper + check(parser.nextToken() == JsonToken.START_ARRAY) + val events = ArrayList(1 shl 18) + while (events.size < readCap && parser.nextToken() == JsonToken.START_OBJECT) { + val node = parser.readValueAsTree() + toEvent(node)?.let { events.add(it) } + } + events + } else { + // NDJSON. + pushback.bufferedReader().useLines { lines -> + lines + .filter { it.isNotBlank() } + .mapNotNull { runCatching { OptimizedJsonMapper.fromJson(it) }.getOrNull() } + .take(readCap) + .toList() + } + } + } + } + + private fun toEvent(node: JsonNode): Event? = + runCatching { + Event( + id = node["id"].asText(), + pubKey = node["pubkey"].asText(), + createdAt = node["created_at"].asLong(), + kind = node["kind"].asInt(), + tags = node["tags"].map { tag -> tag.map { it.asText() }.toTypedArray() }.toTypedArray(), + content = node["content"].asText(), + sig = node["sig"]?.asText() ?: "", + ) + }.getOrNull() + + fun prepare( + raw: List, + limit: Int, + source: String, + log: (String) -> Unit, + maxEventBytes: Int = DEFAULT_MAX_EVENT_BYTES, + maxTags: Int = DEFAULT_MAX_TAGS, + ): Corpus { + val drops = LinkedHashMap() + + fun drop(reason: String) = drops.merge(reason, 1, Int::plus) + + val seen = HashSet(raw.size * 2) + val unique = + raw.filter { e -> + when { + !seen.add(e.id) -> { + drop("duplicate id") + false + } + e.sig.length != 128 -> { + drop("missing/short sig (e.g. NIP-17 rumors)") + false + } + e.kind == 5 -> { + drop("kind-5 deletion (order-dependent)") + false + } + e.kind in 20000..29999 -> { + drop("ephemeral kind") + false + } + e.tags.size > maxTags -> { + drop("more than $maxTags tags") + false + } + e.tags.any { t -> t.any { it.toByteArray().size > MAX_TAG_VALUE_BYTES } } -> { + drop("tag value over ${MAX_TAG_VALUE_BYTES}B") + false + } + OptimizedJsonMapper.toJson(e).toByteArray().size > maxEventBytes -> { + drop("serialized size over ${maxEventBytes}B") + false + } + else -> true + } + } + + val verified = + runBlocking { + coroutineScope { + unique + .chunked(2048) + .map { chunk -> async(Dispatchers.Default) { chunk.filter { it.verify() } } } + .awaitAll() + .flatten() + } + } + val invalidSigs = unique.size - verified.size + if (invalidSigs > 0) drops["invalid signature"] = invalidSigs + + val sorted = verified.sortedWith(compareBy({ it.createdAt }, { it.id })) + val capped = if (limit in 1 until sorted.size) sorted.takeLast(limit) else sorted + if (capped.size < sorted.size) drops["over --limit (kept newest)"] = sorted.size - capped.size + + drops.forEach { (reason, count) -> log(" dropped $count: $reason") } + log(" prepared corpus: ${capped.size} events") + return Corpus(capped, source, spec = null) + } +} diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/relays/RelayUnderTest.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/relays/RelayUnderTest.kt new file mode 100644 index 0000000000..631efe5c54 --- /dev/null +++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/relays/RelayUnderTest.kt @@ -0,0 +1,221 @@ +/* + * 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.relaybench.relays + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import okhttp3.OkHttpClient +import okhttp3.Request +import java.io.File +import java.net.ServerSocket +import java.net.Socket + +/** + * A relay implementation the harness can boot as an external process. + * + * Contract: [command] must start a relay that listens on `127.0.0.1:port` + * with persistent storage under `dataDir`, signature verification ON, no + * auth required, and otherwise stock defaults — that's the "similar setup" + * every implementation is compared under. Adding a relay to the benchmark + * means adding one subclass (or using [CustomRelay] with a command + * template, no code needed). + */ +abstract class RelayUnderTest( + val name: String, +) { + abstract fun command( + port: Int, + dataDir: File, + ): List + + /** Hook for writing config files before launch. */ + open fun prepare( + port: Int, + dataDir: File, + ) {} + + fun start(workDir: File): RunningRelay { + val port = ServerSocket(0).use { it.localPort } + val dataDir = File(workDir, "$name-$port").apply { mkdirs() } + prepare(port, dataDir) + val logFile = File(dataDir, "$name.log") + val process = + ProcessBuilder(command(port, dataDir)) + .redirectErrorStream(true) + .redirectOutput(logFile) + .start() + + val deadline = System.currentTimeMillis() + 30_000 + while (System.currentTimeMillis() < deadline) { + if (!process.isAlive) break + try { + Socket("127.0.0.1", port).close() + return RunningRelay(this, process, port, dataDir, logFile) + } catch (_: Exception) { + Thread.sleep(50) + } + } + process.destroyForcibly() + throw IllegalStateException( + "$name did not open port $port within 30s. Log tail:\n" + logFile.tail(20), + ) + } +} + +class RunningRelay( + val relay: RelayUnderTest, + private val process: Process, + val port: Int, + val dataDir: File, + val logFile: File, +) { + val wsUrl: String get() = "ws://127.0.0.1:$port" + + fun ensureAlive() { + check(process.isAlive) { + "${relay.name} process died mid-benchmark. Log tail:\n" + logFile.tail(20) + } + } + + /** NIP-11 info document — name/software/version for the report header. */ + fun fetchInfo(http: OkHttpClient): Nip11Info? = + runCatching { + val request = + Request + .Builder() + .url("http://127.0.0.1:$port/") + .header("Accept", "application/nostr+json") + .build() + http.newCall(request).execute().use { response -> + val node = jacksonObjectMapper().readTree(response.body.string()) + Nip11Info( + software = node["software"]?.asText()?.substringAfterLast('/')?.removeSuffix(".git"), + version = node["version"]?.asText(), + ) + } + }.getOrNull() + + /** On-disk footprint of everything the relay wrote (DB + logs excluded). */ + fun storageBytes(): Long = + dataDir + .walkTopDown() + .filter { it.isFile && it != logFile } + .sumOf { it.length() } + + fun stop() { + process.destroy() + if (!process.waitFor(10, java.util.concurrent.TimeUnit.SECONDS)) { + process.destroyForcibly() + process.waitFor() + } + } +} + +data class Nip11Info( + val software: String?, + val version: String?, +) + +/** Geode — the standalone JVM relay from this repo (`:geode:installDist`). */ +class GeodeRelay( + private val bin: String, +) : RelayUnderTest("geode") { + override fun command( + port: Int, + dataDir: File, + ): List = + listOf( + bin, + "--host", + "127.0.0.1", + "--port", + "$port", + "--db", + File(dataDir, "geode.sqlite").absolutePath, + ) +} + +/** + * strfry — writes a minimal config (proven against strfry v1 by geode's + * interop tests) and lets compiled-in defaults govern everything else. + * `rejectEventsOlderThanSeconds` is raised to 100 years (0 would reject + * *everything*, not disable the check) so historical corpora (like the + * checked-in 2024 dump) replay 1:1; geode has no old-event cutoff, so this + * keeps the two setups equivalent. `nofiles = 0` skips the setrlimit call, + * which fails inside containers with a low hard limit. Event size + * and tag-count ceilings come from the harness options because the corpus + * is pre-filtered to the same numbers — both relays are always offered a + * stream they are configured to fully accept. + */ +class StrfryRelay( + private val bin: String, + private val maxEventBytes: Int, + private val maxTags: Int, +) : RelayUnderTest("strfry") { + override fun prepare( + port: Int, + dataDir: File, + ) { + File(dataDir, "strfry-db").mkdirs() + File(dataDir, "strfry.conf").writeText( + """ + db = "${File(dataDir, "strfry-db").absolutePath}" + events { + maxEventSize = $maxEventBytes + rejectEventsOlderThanSeconds = 3155760000 + maxNumTags = $maxTags + } + relay { + bind = "127.0.0.1" + port = $port + nofiles = 0 + maxWebsocketPayloadSize = ${maxEventBytes + 65536} + } + """.trimIndent(), + ) + } + + override fun command( + port: Int, + dataDir: File, + ): List = listOf(bin, "--config", File(dataDir, "strfry.conf").absolutePath, "relay") +} + +/** + * Any other relay, from a command template: `{port}` and `{dir}` are + * substituted at launch. Example: + * + * --relay 'nostr-rs-relay=/usr/bin/nostr-rs-relay --db {dir} --port {port}' + */ +class CustomRelay( + name: String, + private val template: String, +) : RelayUnderTest(name) { + override fun command( + port: Int, + dataDir: File, + ): List = + template + .split(Regex("\\s+")) + .filter { it.isNotBlank() } + .map { it.replace("{port}", "$port").replace("{dir}", dataDir.absolutePath) } +} + +private fun File.tail(lines: Int): String = if (exists()) readLines().takeLast(lines).joinToString("\n") else "(no log)" diff --git a/relayBench/src/test/kotlin/com/vitorpamplona/relaybench/CorpusDeterminismTest.kt b/relayBench/src/test/kotlin/com/vitorpamplona/relaybench/CorpusDeterminismTest.kt new file mode 100644 index 0000000000..36c66588c5 --- /dev/null +++ b/relayBench/src/test/kotlin/com/vitorpamplona/relaybench/CorpusDeterminismTest.kt @@ -0,0 +1,97 @@ +/* + * 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.relaybench + +import com.vitorpamplona.quartz.nip01Core.crypto.verify +import com.vitorpamplona.relaybench.bench.SyncBenchmark +import com.vitorpamplona.relaybench.corpus.CorpusGenerator +import com.vitorpamplona.relaybench.corpus.CorpusSpec +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class CorpusDeterminismTest { + private val spec = + CorpusSpec( + seed = 42, + events = 400, + baseTime = CorpusSpec.DEFAULT_BASE_TIME, + spanSeconds = 3600, + ) + + @Test + fun sameSpecSameEventIds() { + val a = CorpusGenerator.generate(spec) + val b = CorpusGenerator.generate(spec) + assertEquals(a.events.size, b.events.size) + assertEquals(a.events.map { it.id }, b.events.map { it.id }, "same spec must reproduce identical event ids") + assertEquals(a.fingerprint, b.fingerprint) + } + + @Test + fun corpusIsFullySignedAndShaped() { + val corpus = CorpusGenerator.generate(spec) + assertEquals(spec.events, corpus.events.size) + assertTrue(corpus.events.all { it.verify() }, "every generated event must carry a valid signature") + + val kinds = corpus.kindHistogram() + // Social shape: profiles, contacts, notes, reactions must all exist. + for (kind in listOf(0, 1, 3, 7)) { + assertTrue((kinds[kind] ?: 0) > 0, "expected kind $kind in corpus, got $kinds") + } + // Timestamps stay inside the spec window (strfry accepts them). + assertTrue(corpus.events.all { it.createdAt in (spec.baseTime - spec.spanSeconds)..spec.baseTime }) + // References point backwards: replies/reactions tag already-published events. + val idsSoFar = HashSet() + for (e in corpus.events) { + for (tag in e.tags) { + if (tag.size >= 2 && tag[0] == "e") { + assertTrue(tag[1] in idsSoFar, "event ${e.id} (kind ${e.kind}) references a later event") + } + } + idsSoFar.add(e.id) + } + } + + @Test + fun scenariosDeriveFromCorpus() { + val corpus = CorpusGenerator.generate(spec) + val scenarios = Scenarios.derive(corpus) + val keys = scenarios.map { it.key } + assertTrue("global-feed" in keys && "follow-feed" in keys && "thread" in keys, "expected core scenarios, got $keys") + // Filters must be inside strfry's default request limits. + for (s in scenarios) { + assertTrue((s.filter.limit ?: 0) <= 500, "${s.key} limit over strfry maxFilterLimit") + assertTrue((s.filter.authors?.size ?: 0) + (s.filter.ids?.size ?: 0) <= 200, "${s.key} filter too large") + } + } + + @Test + fun effectiveEventsCollapsesReplaceables() { + val corpus = CorpusGenerator.generate(spec) + val effective = SyncBenchmark.effectiveEvents(corpus.events) + // The generator emits exactly one kind-0 and one kind-3 per author, + // so nothing should collapse — but the result must never grow. + assertTrue(effective.size <= corpus.events.size) + val replaceableKeys = effective.filter { it.kind == 0 }.map { it.pubKey } + assertEquals(replaceableKeys.size, replaceableKeys.distinct().size) + } +} diff --git a/settings.gradle.kts b/settings.gradle.kts index 023ac563fe..307259219a 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -40,5 +40,6 @@ include(":quic") include(":nestsClient") include(":desktopApp") include(":cli") +include(":relayBench") include(":quic-interop") project(":quic-interop").projectDir = file("quic/interop")