feat: add relayBench — head-to-head relay benchmark (geode vs strfry vs any)

New :relayBench module that boots relay implementations as real external
processes under equivalent setups (persistent storage, sig verification on,
no auth) and compares them on the same corpus:

- Ingest: receipt->queryable-by-REQ latency (publish on one connection,
  hammer-poll REQ{ids} on another), OK-ack latency, and pipelined corpus
  replay throughput over N connections.
- Queries: client-realistic filters derived from the corpus (home feed,
  thread, notifications, profiles, hashtag, by-ids, ...), time-to-EOSE
  percentiles plus aggregate events/sec under concurrency; result-set
  counts are cross-checked between relays and mismatches flagged.
- NIP-77 negentropy sync between every relay pair: 80%/80% slices with 60%
  overlap, reconcile timing/rounds/wire-bytes per server side, delta
  transfer, steady-state identical-set reconcile, convergence verified.
- Storage footprint after full ingest.

Corpora (all cached as NDJSON + manifest with a sha256 id fingerprint so
results are comparable across runs and machines):
- synthetic (default): deterministic to the byte — seeded keys, fixed
  timestamps, seed-derived BIP-340 aux nonces — with a realistic social
  shape (zipf authors, threads, reactions, reposts, zap request/receipt
  pairs, hashtags);
- the checked-in real dump (quartz test fixture, ~31k unique 2024 events);
- external dumps (NDJSON or JSON array, gzip sniffed by magic bytes),
  e.g. the 2.1M contact-list archive, with --max-event-bytes/--max-tags
  raising both the corpus filter and the strfry config together;
- live download from public relays.

Every source runs through the same preparation: dedup, drop unsigned/
ephemeral/kind-5, enforce relay ingest caps, parallel Schnorr verify,
chronological sort.

relayBench/run.sh is the one-command entry point: builds geode + harness,
resolves strfry (STRFRY_BIN, PATH, or source build into .cache), runs the
suite and renders an ANSI report with per-metric bars and winners, plus
report.md and results.json under relayBench/results/<timestamp>/.

The harness client disables Nagle (TCP_NODELAY): with the JDK default, a
REQ following the previous round's CLOSE stalls a full delayed-ACK
interval and every latency floors at ~44 ms against both geode and strfry
(verified: ~0.3 ms with it off).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NeoCvXnTxsKzqurkmjdC46
This commit is contained in:
Claude
2026-07-03 03:22:56 +00:00
parent acaf768309
commit 5f3a790d56
21 changed files with 3514 additions and 1 deletions

View File

@@ -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

3
relayBench/.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
.cache/
.corpus-cache/
results/

111
relayBench/README.md Normal file
View File

@@ -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/<timestamp>/report.md` — shareable Markdown
- `relayBench/results/<timestamp>/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`, …).

View File

@@ -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)
}

98
relayBench/run.sh Executable file
View File

@@ -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[@]}"

View File

@@ -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<Pair<String, String>>,
val events: Int,
val seed: Long,
val baseTime: Long,
val corpusFile: File?,
val downloadFrom: List<String>?,
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<String>) {
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<RelayUnderTest> =
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<String>): Options? {
val map = HashMap<String, MutableList<String>>()
val flags = HashSet<String>()
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,
)
}

View File

@@ -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<String>,
) {
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<String>(Channel.UNLIMITED)
val opened = CompletableDeferred<Unit>()
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)
}
}
}

View File

@@ -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<Long>): 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,
)
}
}
}

View File

@@ -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<RelayResult>,
val syncs: List<SyncBenchmark.PairResult>,
)
/**
* 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<Pair<String, Double?>>,
unit: String,
higherIsBetter: Boolean,
detail: Map<String, String> = 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<RelayResult>): 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<String, Int>()
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)
}
}

View File

@@ -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<Scenario> {
val events = corpus.events
// Frequency maps that mirror what relay indexes serve most.
val notesByAuthor = HashMap<String, Int>()
val eTagRefs = HashMap<String, Int>()
val pTagRefs = HashMap<String, Int>()
val tTagRefs = HashMap<String, Int>()
val noteIds = ArrayList<String>()
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<Map.Entry<String, Int>> { it.value }.thenBy { it.key }).map { it.key }
val hottestThread =
eTagRefs.entries
.sortedWith(compareByDescending<Map.Entry<String, Int>> { it.value }.thenBy { it.key })
.firstOrNull()
?.key
val mostMentioned =
pTagRefs.entries
.sortedWith(compareByDescending<Map.Entry<String, Int>> { it.value }.thenBy { it.key })
.firstOrNull()
?.key
val topHashtag =
tTagRefs.entries
.sortedWith(compareByDescending<Map.Entry<String, Int>> { 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),
),
)
}
}
}

View File

@@ -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<QueryBenchmark.ScenarioResult>,
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<Scenario>,
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<String>()
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<RelayUnderTest>,
corpus: Corpus,
options: Options,
workDir: File,
http: OkHttpClient,
log: (String) -> Unit,
): List<SyncBenchmark.PairResult> {
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<SyncBenchmark.PairResult>()
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
}
}

View File

@@ -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<Event>,
http: OkHttpClient,
log: (String) -> Unit,
): VisibilityResult {
val publisher = NostrSocket.connect(http, wsUrl)
val prober = NostrSocket.connect(http, wsUrl)
val okNanos = ArrayList<Long>(samples.size)
val visibleNanos = ArrayList<Long>(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<Event>,
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<Event>,
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()
}
}
}

View File

@@ -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<Long>(rounds)
val eoseNanos = ArrayList<Long>(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,
)
}
}

View File

@@ -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<String, ReconcileStats>,
val transferMs: Double,
val transferredToA: Int,
val transferredToB: Int,
val identicalReconcile: Map<String, ReconcileStats>,
val converged: Boolean,
val error: String? = null,
)
/** Newest version per replaceable/addressable key; everything else untouched. */
fun effectiveEvents(events: List<Event>): List<Event> {
val byKey = LinkedHashMap<String, Event>(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<HexKey>,
val needIds: Set<HexKey>,
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<Event>,
http: OkHttpClient,
maxRounds: Int = 128,
): NegotiationResult {
val socket = NostrSocket.connect(http, wsUrl)
val haveIds = mutableSetOf<HexKey>()
val needIds = mutableSetOf<HexKey>()
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<HexKey>,
http: OkHttpClient,
): List<Event> {
if (ids.isEmpty()) return emptyList()
val socket = NostrSocket.connect(http, fromUrl)
val fetched = ArrayList<Event>(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<Event>,
sliceA: List<Event>,
sliceB: List<Event>,
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,
)
}
}

View File

@@ -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<Event>,
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<Int, Int> = 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(),
)
}
}

View File

@@ -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<String>,
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<String, Event>(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<Event> {
val http = OkHttpClient.Builder().build()
val socket = NostrSocket.connect(http, url)
val events = ArrayList<Event>(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<Event>? =
withTimeoutOrNull(30_000) {
socket.req(subId, filter.toJson())
val page = ArrayList<Event>(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
}
}

View File

@@ -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<String> = emptyList(),
var follows: List<Int> = 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<Array<String>>,
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<SlotType>(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<Slot>(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<Int>() // 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<Event>(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<Event>) = events.count { it.kind == 9735 }
}

View File

@@ -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<Event> {
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<Event>(1 shl 18)
while (events.size < readCap && parser.nextToken() == JsonToken.START_OBJECT) {
val node = parser.readValueAsTree<JsonNode>()
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<Event>,
limit: Int,
source: String,
log: (String) -> Unit,
maxEventBytes: Int = DEFAULT_MAX_EVENT_BYTES,
maxTags: Int = DEFAULT_MAX_TAGS,
): Corpus {
val drops = LinkedHashMap<String, Int>()
fun drop(reason: String) = drops.merge(reason, 1, Int::plus)
val seen = HashSet<String>(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)
}
}

View File

@@ -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<String>
/** 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<String> =
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<String> = 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<String> =
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)"

View File

@@ -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<String>()
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)
}
}

View File

@@ -40,5 +40,6 @@ include(":quic")
include(":nestsClient")
include(":desktopApp")
include(":cli")
include(":relayBench")
include(":quic-interop")
project(":quic-interop").projectDir = file("quic/interop")