mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-09 08:04:45 +00:00
perf(graperank): mass pre-connect, cached DNS, and an amy relay census probe
Most of a from-scratch crawl's wall clock was connection setup, re-paid serially: the relay pool tears down a relay's socket ~300ms after its last drain unsubscribes, OkHttp allowed only 256 concurrent WS handshakes, and every dial re-ran an uncached blocking getaddrinfo (10-30s per dead domain, once per per-user path URL of the same host). Open the connections once, in parallel, and keep them: - GrapeRankCrawler: the warm-pool trick (never-matching REQ that only holds the socket) now covers the whole candidate universe instead of the top 20 - seeded at crawl start from the reachability cache's live set (one parallel connection storm, Config.knownLiveRelays) and refreshed each round with newly learned outbox relays, capped by Config.preconnectCap (FD-budget aware, --preconnect-cap / --no-preconnect). - CachingDns (quartz jvmAndroid): 10-min positive + negative DNS cache with in-flight per-host dedup; dead domains fail in microseconds instead of re-burning resolver timeouts, path URLs of one host resolve once. - cli Context: dispatcher and pre-connect caps derived from the process's open-files limit (UnixOperatingSystemMXBean), warning when ulimit is low. - amy graperank probe: relay census - mass-connects every relay the store knows (kind:10002 universe deduped per authority + cached verdicts) in waves, records live/dead with real measured rtt-open into the NIP-66 reachability cache (RelayProber + RelayReachabilityStore.recordProbed), so the next crawl skips dead relays and waits once for the slow-but-alive. Single-server limits (FDs, ephemeral ports, DNS, threads, conntrack) and the design are documented in quartz/plans/2026-07-10-graperank-connect-storm.md. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013zEYRGKF943RgLaHTViJaB
This commit is contained in:
Claude
@@ -20,6 +20,7 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.cli
|
||||
|
||||
import com.sun.management.UnixOperatingSystemMXBean
|
||||
import com.vitorpamplona.amethyst.cli.stores.FileCashuKeysetCounterStore
|
||||
import com.vitorpamplona.amethyst.cli.stores.FileKeyPackageBundleStore
|
||||
import com.vitorpamplona.amethyst.cli.stores.FileMarmotMessageStore
|
||||
@@ -53,6 +54,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.CachingDns
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.TcpNoDelaySocketFactory
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
|
||||
@@ -83,6 +85,7 @@ import kotlinx.coroutines.selects.select
|
||||
import kotlinx.coroutines.withTimeoutOrNull
|
||||
import okhttp3.Dispatcher
|
||||
import okhttp3.OkHttpClient
|
||||
import java.lang.management.ManagementFactory
|
||||
import java.util.concurrent.TimeUnit
|
||||
|
||||
/**
|
||||
@@ -144,14 +147,24 @@ class Context(
|
||||
// already-open sockets, bounded by AdaptiveRelayLimiter), so it can't
|
||||
// trip a relay's REQ rate-limit — it only speeds connection setup. The
|
||||
// executor thread pool is unbounded on demand, so raising maxRequests
|
||||
// just lets more of those short-lived handshakes proceed at once. 7s
|
||||
// just lets more of those short-lived handshakes proceed at once
|
||||
// (1 platform thread per in-flight handshake — ~1k is fine on a JVM,
|
||||
// 10k is not). Sized against the process FD budget: every pending
|
||||
// handshake and every open socket is one file descriptor. 7s
|
||||
// (not 5s): a 5s cap struck too many merely-busy relays as connect
|
||||
// failures — the crawl treats a connect *timeout* as retryable anyway,
|
||||
// but the extra headroom lets slow-but-alive relays finish the handshake.
|
||||
.connectTimeout(7, TimeUnit.SECONDS)
|
||||
// DNS dominates a mass connection ramp without help: Dns.SYSTEM blocks a
|
||||
// dispatcher thread per lookup, the JVM's own cache lasts ~30s (useless
|
||||
// over a 30-minute crawl), a dead domain burns 10-30s of resolver
|
||||
// timeouts on EVERY re-dial, and the outbox model mints hundreds of
|
||||
// per-user URLs on one host — each a fresh lookup. The cache (10-min
|
||||
// positive + negative TTL, in-flight dedup per host) collapses all of it.
|
||||
.dns(CachingDns())
|
||||
.dispatcher(
|
||||
Dispatcher().apply {
|
||||
maxRequests = 256
|
||||
maxRequests = maxParallelHandshakes
|
||||
maxRequestsPerHost = 16
|
||||
},
|
||||
).build()
|
||||
@@ -1033,6 +1046,39 @@ class Context(
|
||||
*/
|
||||
private const val GIFT_WRAP_LOOKBACK_SECS: Long = 2L * 24 * 60 * 60
|
||||
|
||||
/** FDs reserved for everything that isn't a relay socket (store, jars, pipes, DNS). */
|
||||
private const val FD_RESERVE = 256L
|
||||
|
||||
/**
|
||||
* The process's max-open-files limit (`ulimit -n`), the hard ceiling on
|
||||
* concurrent sockets: every open WebSocket AND every in-flight handshake is
|
||||
* one file descriptor, and the JVM cannot raise its own rlimit. Falls back
|
||||
* to the conservative 1024 (the common soft default) when the platform bean
|
||||
* doesn't expose it.
|
||||
*/
|
||||
val maxFileDescriptors: Long =
|
||||
(ManagementFactory.getOperatingSystemMXBean() as? UnixOperatingSystemMXBean)
|
||||
?.maxFileDescriptorCount ?: 1024L
|
||||
|
||||
/**
|
||||
* Concurrent WS-upgrade handshakes (OkHttp Dispatcher.maxRequests): a quarter
|
||||
* of the FD budget, so pending dials can never crowd out the sockets already
|
||||
* held open (warm pool, drains, parked subs). Also bounds the transient
|
||||
* platform threads OkHttp spawns — one per in-flight handshake.
|
||||
*/
|
||||
val maxParallelHandshakes: Int =
|
||||
((maxFileDescriptors - FD_RESERVE) / 4).coerceIn(64, 1024).toInt()
|
||||
|
||||
/**
|
||||
* Default cap for the crawl's mass pre-connect (warm pool): half the FD
|
||||
* budget goes to held-open relay sockets, leaving the other half for
|
||||
* in-flight handshakes, drain/parked subscriptions and headroom. At the
|
||||
* common 1024-FD soft limit this is ~384; `ulimit -n 16384` unlocks the
|
||||
* full 4000. Overridable per-run with --preconnect-cap.
|
||||
*/
|
||||
val defaultPreconnectCap: Int =
|
||||
((maxFileDescriptors - FD_RESERVE) / 2).coerceIn(100, 4000).toInt()
|
||||
|
||||
/**
|
||||
* Build a Context but require an account with a usable identity —
|
||||
* signing verbs can't run without one. Throws [IllegalArgumentException]
|
||||
|
||||
@@ -609,6 +609,14 @@ private fun printUsage() {
|
||||
| new/changed ranks >= --min-rank (default 2), skips
|
||||
| unchanged, and retracts (kind:5) any card whose
|
||||
| target left the graph or fell below the cutoff.
|
||||
| graperank crawl [OBSERVER] network only: crawl the WoT graph (kind 3/10000/
|
||||
| [--max-hops N] [--preconnect-cap N] 1984/10002) into the local store without scoring.
|
||||
| [--no-preconnect] Pre-connects every known-live relay in one parallel
|
||||
| storm (seeded from the reachability cache).
|
||||
| graperank probe [--timeout SECS] relay census: mass-connect every relay the store
|
||||
| [--concurrency N] knows and record live/dead + measured rtt-open into
|
||||
| the reachability cache (NIP-66 kind:30166), so the
|
||||
| next crawl skips dead relays and waits once.
|
||||
| graperank update [--down] [--up] refresh every locally-known author's WoT record kinds
|
||||
| [--no-sync-deletions] [--timeout SECS] (0/3/10002/1984) from their own outbox: reads all
|
||||
| [--relay-concurrency N] [--author-chunk N] kind:10002 in the store, groups authors by write
|
||||
|
||||
@@ -46,6 +46,7 @@ import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent
|
||||
import com.vitorpamplona.quartz.nip09Deletions.DeletionIndex
|
||||
import com.vitorpamplona.quartz.nip51Lists.muteList.MuteListEvent
|
||||
import com.vitorpamplona.quartz.nip56Reports.ReportEvent
|
||||
import com.vitorpamplona.quartz.nip66RelayMonitor.reachability.RelayProber
|
||||
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.TrustProviderListEvent
|
||||
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.serviceProviders
|
||||
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.tags.ProviderTypes
|
||||
@@ -92,6 +93,9 @@ import kotlin.math.roundToInt
|
||||
* - `amy graperank score [OBSERVER]` — local only: build the graph from the store
|
||||
* and score (same as bare `--offline`). Instant and param-tunable; repeat with
|
||||
* different `--rigor`/`--attenuation`/cutoffs without re-crawling.
|
||||
* - `amy graperank probe` — the relay census: mass-connect every relay the store
|
||||
* knows and record live/dead + measured RTT into the reachability cache, so the
|
||||
* next crawl skips the dead and pre-connects the living in one parallel storm.
|
||||
* - bare `amy graperank [OBSERVER]` — the convenience combo: crawl then score.
|
||||
*
|
||||
* Sub-verbs complete the NIP-85 provider experience — the discovery layer that
|
||||
@@ -193,6 +197,7 @@ object GrapeRankCommand {
|
||||
// `sync` is the pre-rename name kept as a back-compat alias; `crawl` is
|
||||
// canonical (disambiguates from negentropy `amy sync` / `graperank update`).
|
||||
"crawl", "sync" -> crawl(dataDir, tail.drop(1).toTypedArray())
|
||||
"probe" -> probe(dataDir, tail.drop(1).toTypedArray())
|
||||
"update" -> update(dataDir, tail.drop(1).toTypedArray())
|
||||
"score" -> run(dataDir, tail.drop(1).toTypedArray(), forceOffline = true)
|
||||
else -> run(dataDir, tail)
|
||||
@@ -441,11 +446,25 @@ object GrapeRankCommand {
|
||||
// Aggregator kind:3 recovery for stragglers is on by default; --no-aggregators
|
||||
// disables it for A/B comparison.
|
||||
val aggregators = if (args.bool("no-aggregators")) emptySet() else CONTENT_AGGREGATOR_RELAYS
|
||||
// Seed the crawl with relays a prior run/monitor proved dead within the cache's
|
||||
// TTL, so the WS path never re-pays their connect timeouts (--no-reachability-cache
|
||||
// to skip). The crawl's own final live/dead set is flushed back by the caller.
|
||||
val knownDead =
|
||||
if (args.bool("no-reachability-cache")) emptySet() else ctx.reachability.snapshot().dead
|
||||
// Seed the crawl from the reachability cache (--no-reachability-cache to skip):
|
||||
// proven-dead relays within the TTL are never dialed (their connect timeouts
|
||||
// aren't re-paid), and the proven-live universe is pre-connected in one
|
||||
// parallel storm at crawl start so the connect wait is paid once, up front.
|
||||
// The crawl's own final live/dead set is flushed back by the caller.
|
||||
val reachability =
|
||||
if (args.bool("no-reachability-cache")) null else ctx.reachability.snapshot()
|
||||
val knownDead = reachability?.dead ?: emptySet()
|
||||
// Mass pre-connect sizing: every warm socket is one FD, so the default is
|
||||
// derived from the process's ulimit (--preconnect-cap to override,
|
||||
// --no-preconnect to fall back to the top-20 warm pool).
|
||||
val preconnectCap =
|
||||
if (args.bool("no-preconnect")) 0 else args.intFlag("preconnect-cap", Context.defaultPreconnectCap)
|
||||
if (preconnectCap > 0 && Context.maxFileDescriptors < 4096) {
|
||||
System.err.println(
|
||||
"[graperank] open-files limit is ${Context.maxFileDescriptors} → pre-connect capped at " +
|
||||
"$preconnectCap sockets; `ulimit -n 16384` before running unlocks a faster crawl",
|
||||
)
|
||||
}
|
||||
return GrapeRankCrawler(
|
||||
client = ctx.client,
|
||||
store = ctx.store,
|
||||
@@ -454,6 +473,8 @@ object GrapeRankCommand {
|
||||
GrapeRankCrawler.Config(
|
||||
relayListDiscoveryRelays = discoveryRelays,
|
||||
knownDeadRelays = knownDead,
|
||||
knownLiveRelays = if (preconnectCap > 0) reachability?.live.orEmpty() else emptySet(),
|
||||
preconnectCap = preconnectCap,
|
||||
contentFallbackRelays = contentFallback,
|
||||
contentAggregatorRelays = aggregators,
|
||||
maxRounds = args.intFlag("max-rounds", Int.MAX_VALUE),
|
||||
@@ -548,6 +569,86 @@ object GrapeRankCommand {
|
||||
return 0
|
||||
}
|
||||
|
||||
/**
|
||||
* `amy graperank probe [--timeout SECS] [--concurrency N]` —
|
||||
* the relay census. Mass-connects the ENTIRE relay universe the local store knows
|
||||
* (every relay advertised in any stored kind:10002, deduped per host, plus
|
||||
* everything already in the reachability cache) in parallel waves with a no-op
|
||||
* REQ, so the "is this relay alive, and how slow?" wait is paid once, up front,
|
||||
* concurrently — then records per-relay verdicts with real measured `rtt-open`
|
||||
* into the NIP-66 reachability cache (kind:30166).
|
||||
*
|
||||
* The next `graperank crawl` reads that cache to (a) skip the dead set without
|
||||
* dialing it and (b) pre-connect the live set in one storm — separating "working
|
||||
* but slow" (kept; the crawler's patient park path waits for them) from "not
|
||||
* working" (skipped entirely). Typical flow the first time:
|
||||
* `graperank crawl --max-hops 2` (cheap, saves the relay lists) → `graperank
|
||||
* probe` → full `graperank crawl`.
|
||||
*/
|
||||
private suspend fun probe(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
): Int {
|
||||
val args = Args(rest)
|
||||
val timeoutMs = args.longFlag("timeout", 15L) * 1000
|
||||
val waveSize = args.intFlag("concurrency", Context.defaultPreconnectCap)
|
||||
|
||||
Context.openOrAnonymous(dataDir).use { ctx ->
|
||||
ctx.prepare()
|
||||
val cached = ctx.reachability.snapshot()
|
||||
val universe = RelayProber.knownRelayUniverse(ctx.store) + cached.live + cached.dead
|
||||
if (universe.isEmpty()) {
|
||||
Output.emit(
|
||||
linkedMapOf<String, Any?>(
|
||||
"probed" to 0,
|
||||
"note" to "no relays known locally — run `amy graperank crawl` first to gather kind:10002 relay lists",
|
||||
),
|
||||
)
|
||||
return 0
|
||||
}
|
||||
|
||||
System.err.println(
|
||||
"[relay-probe] probing ${universe.size} relays in waves of $waveSize " +
|
||||
"(${timeoutMs / 1000}s per wave; open-files limit ${Context.maxFileDescriptors})",
|
||||
)
|
||||
val result =
|
||||
RelayProber(ctx.client) { System.err.println(it) }
|
||||
.probe(universe, timeoutMs, waveSize)
|
||||
|
||||
ctx.reachability.recordProbed(result.reachableRttMs(), result.deadRelays())
|
||||
|
||||
val rtts =
|
||||
result.reachable
|
||||
.map { it.rttOpenMs }
|
||||
.filter { it >= 0 }
|
||||
.sorted()
|
||||
|
||||
fun pct(p: Int): Long? = if (rtts.isEmpty()) null else rtts[(rtts.size - 1) * p / 100]
|
||||
val slowest =
|
||||
result.reachable
|
||||
.filter { it.rttOpenMs >= 0 }
|
||||
.sortedByDescending { it.rttOpenMs }
|
||||
.take(10)
|
||||
.map { mapOf("relay" to it.relay.url, "rtt_open_ms" to it.rttOpenMs) }
|
||||
val authWalled = result.reachable.count { it.error?.startsWith("closed:") == true }
|
||||
|
||||
Output.emit(
|
||||
linkedMapOf<String, Any?>(
|
||||
"probed" to result.verdicts.size,
|
||||
"reachable" to result.reachable.size,
|
||||
"dead" to result.dead.size,
|
||||
"closed_by_policy" to authWalled,
|
||||
"elapsed_ms" to result.elapsedMs,
|
||||
"rtt_open_p50_ms" to pct(50),
|
||||
"rtt_open_p90_ms" to pct(90),
|
||||
"rtt_open_p99_ms" to pct(99),
|
||||
"slowest" to slowest,
|
||||
),
|
||||
)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
/**
|
||||
* `amy graperank update [flags]` — refresh every locally-known author's WoT
|
||||
* record kinds (0 / 3 / 10002 / 1984) straight from their own outbox, so the
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
# GrapeRank crawl: connect once, wait once (connection storm)
|
||||
|
||||
## Problem
|
||||
|
||||
A from-scratch `amy graperank crawl` spends most of its wall clock **waiting to
|
||||
connect** to relays, not downloading. Three compounding causes, all on our side:
|
||||
|
||||
1. **Sockets are torn down between drains.** `NostrClient` reconciles the relay
|
||||
pool against the union of *active* subscriptions (`RelayPool.updatePool`,
|
||||
sampled every 300ms). A Phase-B drain unit unsubscribes when it finishes, so
|
||||
an outbox relay with no other in-flight drain is **disconnected ~300ms
|
||||
later** — and the next round re-pays DNS + TCP + TLS + WS-upgrade for the
|
||||
same relay. Only the top-20 "warm pool" relays kept a do-nothing
|
||||
subscription open. A crawl touches ~10–15k distinct relay URLs; most get
|
||||
dialed many times.
|
||||
2. **Handshake concurrency was capped at 256** (OkHttp `Dispatcher.maxRequests`;
|
||||
the OkHttp default is 64). Every WS upgrade is an async call holding a
|
||||
dispatcher slot for its whole DNS + TCP + TLS + upgrade; a dead relay holds
|
||||
a slot for the full 7s `connectTimeout`. At 256 slots, dialing 5k relays
|
||||
where half are dead is ~20 serialized waves — minutes of pure waiting that
|
||||
could be one parallel wait.
|
||||
3. **DNS is blocking, uncached and duplicated.** OkHttp's `Dns.SYSTEM` calls
|
||||
`InetAddress.getAllByName` (glibc `getaddrinfo`) on the dispatcher thread.
|
||||
The JVM's own cache is ~30s positive / ~10s negative — useless across a
|
||||
30-minute crawl that re-dials the same hosts every round. Worse, the outbox
|
||||
model mints hundreds of per-user *URLs* on one *host*
|
||||
(`filter.nostr.wine/npubA`, `/npubB`, …) and each URL is a fresh lookup. A
|
||||
dead domain is the worst case: resolver timeouts (5s × 2 attempts per
|
||||
nameserver) can hold a thread 10–30s, and the crawl re-resolves it from
|
||||
every straggler's outbox, every round.
|
||||
|
||||
## Single-server limits (what bounds "1000s of connections at once")
|
||||
|
||||
Measured/derived for a JVM CLI on one Linux box:
|
||||
|
||||
| Resource | Limit | Consequence |
|
||||
|---|---|---|
|
||||
| File descriptors | `ulimit -n` soft (often 1024–4096; systemd default 1024 soft / 512k+ hard) | 1 FD per socket. The JVM cannot raise its own rlimit — detect via `UnixOperatingSystemMXBean` and size the storm to fit; advise `ulimit -n 16384` for big crawls. |
|
||||
| Ephemeral ports | ~28k (`ip_local_port_range`) **per destination tuple** | Non-issue: connections go to thousands of *distinct* relays, each gets its own tuple space. |
|
||||
| Concurrent TCP connects | Kernel: effectively unbounded (a pending SYN is just a socket in `SYN_SENT`) | Dead-IP dials cost nothing in CPU but hold their FD + dispatcher slot for the connect timeout. Parallelism is the fix; refused/reset fails in 1 RTT. |
|
||||
| DNS | glibc `getaddrinfo` is blocking, ~1 thread per in-flight lookup; dead domains 10–30s | Cache positives AND negatives in-process; dedupe concurrent lookups for the same host (path-URL explosion). |
|
||||
| Threads | 1 platform thread per in-flight handshake (OkHttp) | 1–2k concurrent handshakes ≈ 1–2k transient threads — fine on a 64-bit JVM (~stack is virtual), but don't go to 10k. |
|
||||
| TLS | CPU per handshake (~1ms) | Negligible next to network RTTs; OkHttp shares an SSL session cache per client. |
|
||||
| Middleboxes | Home-router NAT/conntrack tables (~4–16k entries) | Operational caveat for residential operators; server/VPS operators unaffected (`nf_conntrack_max` default ~256k). |
|
||||
|
||||
Conclusion: with FDs raised and DNS cached, a single server comfortably opens
|
||||
**a few thousand concurrent connections**; the binding constraints are the FD
|
||||
soft limit and the dispatcher cap, both of which we now size/raise explicitly.
|
||||
|
||||
## Design
|
||||
|
||||
Three changes, all keeping completeness identical (same relays asked the same
|
||||
filters — only *when* sockets open changes):
|
||||
|
||||
1. **Crawl-wide warm pool = mass pre-connect.** The existing warm-pool trick
|
||||
(a never-matching `ids` filter that EOSEs instantly and just holds the
|
||||
socket) is extended from the top-20 relays to **every candidate relay we
|
||||
know** — seeded at crawl start from the NIP-66 reachability cache's live
|
||||
set, and refreshed every round with the relays learned from kind:10002s —
|
||||
capped by `preconnectCap` (FD-budget-aware, busiest relays first, dead
|
||||
relays excluded). Connections open in one parallel storm at start (and in
|
||||
the background as new relays are learned), stay up for the whole crawl, and
|
||||
every drain hits an already-open socket. This also implements the "crawl
|
||||
once with --max-hops to save the relay list" flow: the first (even shallow)
|
||||
crawl populates the store with 10002s and the reachability cache with
|
||||
live/dead verdicts; the next crawl pre-connects that whole universe and
|
||||
waits once.
|
||||
2. **Transport ceilings raised (CLI).** `Dispatcher.maxRequests` 256 → 1024
|
||||
(per-host stays 16 — politeness to path-multiplexed hosts is a *server*
|
||||
property), and a `CachingDns` (10-min positive + negative TTL, in-flight
|
||||
dedup per host) replaces `Dns.SYSTEM`. FD budget is detected at startup and
|
||||
sizes both the dispatcher and the default `preconnectCap`.
|
||||
3. **`amy graperank probe` — the relay census.** Reads the full known relay
|
||||
universe (every relay in stored kind:10002s + everything in the
|
||||
reachability cache), mass-connects it in waves with a never-matching REQ,
|
||||
and records per-relay verdicts with **real** `rtt-open` into the NIP-66
|
||||
store: connected → live (with measured RTT, however slow), cannot-connect /
|
||||
never-completed → dead (TTL'd, re-probed after expiry). Separates
|
||||
"working but slow" (kept, given the crawler's patient park path) from
|
||||
"not working" (skipped entirely) without burning crawl time on the
|
||||
distinction.
|
||||
|
||||
## Non-goals
|
||||
|
||||
- No change to REQ concurrency per relay (`AdaptiveRelayLimiter`) or the
|
||||
drain worker count — the 64-worker A/B showed the *REQ* fan-out re-floods
|
||||
busy hubs; this work only parallelizes and amortizes *connection setup*.
|
||||
- No async-DNS library dependency; cached blocking lookups on OkHttp's
|
||||
existing threads are sufficient once deduped and negative-cached.
|
||||
+80
-10
@@ -208,6 +208,27 @@ class GrapeRankCrawler(
|
||||
* permanently ignores an author's advertised home. See RelayReachabilityStore.
|
||||
*/
|
||||
val knownDeadRelays: Set<NormalizedRelayUrl> = emptySet(),
|
||||
/**
|
||||
* Relays a prior run (or another monitor) proved REACHABLE within the
|
||||
* reachability cache's TTL. Seeded into the crawl-wide warm pool at start, so
|
||||
* their DNS + TCP + TLS + WS-upgrade cost is paid ONCE, in one parallel
|
||||
* connection storm, before the first round — instead of serially inside each
|
||||
* drain that first routes to them. See [preconnectCap].
|
||||
*/
|
||||
val knownLiveRelays: Set<NormalizedRelayUrl> = emptySet(),
|
||||
/**
|
||||
* Cap on the crawl-wide warm pool — the mass pre-connect. The client tears a
|
||||
* relay's socket down ~300ms after its last subscription closes, so without a
|
||||
* warm hold every round re-pays the full connect latency per outbox relay.
|
||||
* The warm pool keeps a do-nothing subscription (never-matching filter, EOSEs
|
||||
* instantly, streams nothing) open to up to this many relays for the whole
|
||||
* crawl: busiest proven relays first, then next-round outbox candidates, then
|
||||
* [knownLiveRelays]. Each warm relay holds one socket (one file descriptor)
|
||||
* and one dormant sub — size against the process FD budget, and keep it above
|
||||
* the transport's concurrent-handshake cap only if FDs allow. `<= 0` falls
|
||||
* back to warming just the busiest [WARM_POOL_SIZE] relays (the old behavior).
|
||||
*/
|
||||
val preconnectCap: Int = 2500,
|
||||
)
|
||||
|
||||
/** What the crawl fetched — the counters the caller reports and the graph is built from. */
|
||||
@@ -484,6 +505,41 @@ class GrapeRankCrawler(
|
||||
.map { it.key }
|
||||
.toList()
|
||||
|
||||
/**
|
||||
* Refresh the crawl-wide warm pool — the mass pre-connect. Re-subscribing the
|
||||
* same [WARM_SUB_ID] with a new relay set just updates the desired-relay set:
|
||||
* relays already warm stay connected, new ones start their handshake NOW (in
|
||||
* parallel, in the background), so by the time a drain routes to them the
|
||||
* socket is already open. Priority under [Config.preconnectCap]: busiest
|
||||
* proven-live relays, then advertised-but-not-yet-contacted outboxes (exactly
|
||||
* the relays the next rounds will dial), then the prior run's known-live
|
||||
* universe. Dead relays are always excluded so the pool never redials them.
|
||||
*/
|
||||
fun refreshWarmPool() {
|
||||
val massCap = config.preconnectCap
|
||||
val cap = if (massCap > 0) massCap else WARM_POOL_SIZE
|
||||
val warm = LinkedHashSet<NormalizedRelayUrl>(cap * 2)
|
||||
warm.addAll(topLiveRelays(cap))
|
||||
if (massCap > 0) {
|
||||
if (warm.size < cap) {
|
||||
// Advertised write relays not yet proven live — next drains' targets.
|
||||
val advertised = writeRelayFreq.snapshot().entries.sortedByDescending { it.value }
|
||||
for ((relay, _) in advertised) {
|
||||
if (warm.size >= cap) break
|
||||
if (!isDead(relay)) warm.add(relay)
|
||||
}
|
||||
}
|
||||
if (warm.size < cap) {
|
||||
for (relay in config.knownLiveRelays) {
|
||||
if (warm.size >= cap) break
|
||||
if (!isDead(relay)) warm.add(relay)
|
||||
}
|
||||
}
|
||||
}
|
||||
if (warm.isEmpty()) return
|
||||
client.subscribe(WARM_SUB_ID, warm.associateWith { WARM_FILTERS }, null)
|
||||
}
|
||||
|
||||
/**
|
||||
* Background reachability culler. Cheaply TCP-probes the relays we've learned —
|
||||
* COLD TAIL FIRST — and drops the unreachable ones into [deadHosts] so the WS
|
||||
@@ -1425,6 +1481,19 @@ class GrapeRankCrawler(
|
||||
// by scope.cancel() at crawl end.
|
||||
config.reachabilityProbe?.let { probe -> scope.launch { cullUnreachable(probe) } }
|
||||
|
||||
// Mass pre-connect: start the handshake to every relay a prior run proved
|
||||
// live NOW, in one parallel storm, so the connect wait is paid once —
|
||||
// concurrently, before the first drain — instead of serially inside each
|
||||
// drain that first routes to a relay. The subscribe returns immediately;
|
||||
// sockets ramp in the background while round 1's discovery queries run.
|
||||
if (config.preconnectCap > 0 && config.knownLiveRelays.isNotEmpty()) {
|
||||
log(
|
||||
"[graperank] pre-connecting up to ${config.preconnectCap} of " +
|
||||
"${config.knownLiveRelays.size} known-live relays",
|
||||
)
|
||||
refreshWarmPool()
|
||||
}
|
||||
|
||||
while (rounds < config.maxRounds) {
|
||||
// Fold in whatever the parked (slow-but-alive) relays have delivered
|
||||
// since the last round — their late contact lists expand the frontier
|
||||
@@ -1453,12 +1522,11 @@ class GrapeRankCrawler(
|
||||
progBaseDone = done.size
|
||||
progConverging = false
|
||||
|
||||
// Refresh the warm pool to this round's busiest relays and keep that
|
||||
// subscription open — reusing the same subId just updates the
|
||||
// desired-relay set, so these sockets stay up across the round.
|
||||
topLiveRelays(WARM_POOL_SIZE).takeIf { it.isNotEmpty() }?.let { warm ->
|
||||
client.subscribe(WARM_SUB_ID, warm.associateWith { WARM_FILTERS }, null)
|
||||
}
|
||||
// Refresh the warm pool for this round and keep that subscription open —
|
||||
// reusing the same subId just updates the desired-relay set, so warm
|
||||
// sockets survive across rounds and newly-learned outbox relays start
|
||||
// connecting in the background before Phase B routes to them.
|
||||
refreshWarmPool()
|
||||
|
||||
val discoveredBefore = hopOf.size
|
||||
val fedBefore = contactListsFed
|
||||
@@ -1961,10 +2029,12 @@ class GrapeRankCrawler(
|
||||
// Most-used write relays kept as the known-good backbone for retrying users.
|
||||
private const val BACKBONE_SIZE = 30
|
||||
|
||||
// Warm pool: hold a do-nothing subscription open to the busiest relays for
|
||||
// the whole crawl, so the connections we reuse every round survive the
|
||||
// between-round routing gaps. The filter matches an impossible event id, so
|
||||
// the relay EOSEs immediately and streams nothing — it only keeps sockets warm.
|
||||
// Warm pool: hold a do-nothing subscription open to relays for the whole
|
||||
// crawl, so their connections survive the between-round routing gaps. The
|
||||
// filter matches an impossible event id, so the relay EOSEs immediately and
|
||||
// streams nothing — it only keeps sockets warm. With Config.preconnectCap > 0
|
||||
// the pool covers the whole candidate universe (mass pre-connect); this
|
||||
// constant is the busiest-relays fallback size when that is disabled.
|
||||
private const val WARM_POOL_SIZE = 20
|
||||
private const val WARM_SUB_ID = "graperank-warm"
|
||||
private val WARM_FILTERS = listOf(Filter(ids = listOf("0".repeat(64))))
|
||||
|
||||
+265
@@ -0,0 +1,265 @@
|
||||
/*
|
||||
* 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.quartz.nip66RelayMonitor.reachability
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
|
||||
import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.withTimeoutOrNull
|
||||
import kotlin.time.TimeSource
|
||||
|
||||
/**
|
||||
* Mass relay census: dials a whole relay universe in parallel waves and returns a
|
||||
* per-relay [Verdict] — reachable (with the measured WebSocket-open RTT and, when the
|
||||
* relay answered a no-op REQ, its time-to-EOSE) or dead (with the failure reason).
|
||||
*
|
||||
* The point is to pay the "is this relay alive, and how slow is it?" wait ONCE, up
|
||||
* front and concurrently, instead of rediscovering it serially inside a crawl: feed
|
||||
* the verdicts to [RelayReachabilityStore.recordProbed] and the next crawl skips the
|
||||
* dead set entirely and pre-connects the live set in one storm.
|
||||
*
|
||||
* Mechanics per wave:
|
||||
* - subscribe a never-matching filter (impossible event id) to every relay in the
|
||||
* wave at once — the client dials them all in parallel (bounded by the transport's
|
||||
* handshake concurrency), a healthy relay EOSEs immediately, and no event payload
|
||||
* is ever streamed;
|
||||
* - a connection listener captures the real WS-upgrade RTT per relay;
|
||||
* - any app-level answer (EOSE, or even a CLOSED — an auth-walled relay is still a
|
||||
* WORKING relay) marks it reachable; a connect failure marks it dead;
|
||||
* - a relay with no terminal by the wave deadline is dead if it never opened the
|
||||
* socket, reachable-but-slow if it did.
|
||||
*
|
||||
* Wave size should stay at or below the transport's concurrent-handshake cap (and
|
||||
* well below the process's file-descriptor budget) — beyond that, extra relays in a
|
||||
* wave just queue and eat the wave deadline without dialing.
|
||||
*/
|
||||
class RelayProber(
|
||||
private val client: INostrClient,
|
||||
private val log: (String) -> Unit = {},
|
||||
) {
|
||||
/** One relay's probe outcome. [rttOpenMs]/[rttEoseMs] are -1 when not observed. */
|
||||
class Verdict(
|
||||
val relay: NormalizedRelayUrl,
|
||||
val reachable: Boolean,
|
||||
val rttOpenMs: Long,
|
||||
val rttEoseMs: Long,
|
||||
val error: String?,
|
||||
)
|
||||
|
||||
class Result(
|
||||
val verdicts: List<Verdict>,
|
||||
val elapsedMs: Long,
|
||||
) {
|
||||
val reachable: List<Verdict> get() = verdicts.filter { it.reachable }
|
||||
|
||||
val dead: List<Verdict> get() = verdicts.filter { !it.reachable }
|
||||
|
||||
/** Per-relay open RTT for the reachable set, 0 (= "live, latency unknown") when unobserved. */
|
||||
fun reachableRttMs(): Map<NormalizedRelayUrl, Long> = reachable.associate { it.relay to it.rttOpenMs.coerceAtLeast(0) }
|
||||
|
||||
fun deadRelays(): Set<NormalizedRelayUrl> = dead.mapTo(HashSet()) { it.relay }
|
||||
}
|
||||
|
||||
/**
|
||||
* Probe every relay in [relays], [waveSize] at a time, giving each wave up to
|
||||
* [timeoutMs] to reach terminals. Returns one [Verdict] per input relay.
|
||||
*/
|
||||
suspend fun probe(
|
||||
relays: Collection<NormalizedRelayUrl>,
|
||||
timeoutMs: Long = 15_000,
|
||||
waveSize: Int = 1000,
|
||||
): Result {
|
||||
val mark = TimeSource.Monotonic.markNow()
|
||||
val all = ArrayList<Verdict>(relays.size)
|
||||
val distinct = relays.toSet()
|
||||
var done = 0
|
||||
for (wave in distinct.chunked(waveSize.coerceAtLeast(1))) {
|
||||
all += probeWave(wave, timeoutMs)
|
||||
done += wave.size
|
||||
if (distinct.size > wave.size) {
|
||||
val liveSoFar = all.count { it.reachable }
|
||||
log("[relay-probe] $done/${distinct.size} probed · $liveSoFar reachable")
|
||||
}
|
||||
}
|
||||
return Result(all, mark.elapsedNow().inWholeMilliseconds)
|
||||
}
|
||||
|
||||
private suspend fun probeWave(
|
||||
wave: List<NormalizedRelayUrl>,
|
||||
timeoutMs: Long,
|
||||
): List<Verdict> {
|
||||
val mark = TimeSource.Monotonic.markNow()
|
||||
val waveSet = wave.toHashSet()
|
||||
val openRtt = ConcurrentMap<NormalizedRelayUrl, Long>()
|
||||
val eoseMs = ConcurrentMap<NormalizedRelayUrl, Long>()
|
||||
val errors = ConcurrentMap<NormalizedRelayUrl, String>()
|
||||
// Every terminal (EOSE / CLOSED / cannot-connect) pings this with its relay so
|
||||
// the wait loop can stop as soon as the whole wave has resolved.
|
||||
val terminals = Channel<NormalizedRelayUrl>(Channel.UNLIMITED)
|
||||
|
||||
val connListener =
|
||||
object : RelayConnectionListener {
|
||||
override fun onConnected(
|
||||
relay: IRelayClient,
|
||||
pingMillis: Int,
|
||||
compressed: Boolean,
|
||||
) {
|
||||
if (relay.url in waveSet) openRtt.getOrPut(relay.url) { pingMillis.toLong() }
|
||||
}
|
||||
}
|
||||
|
||||
val subId = newSubId()
|
||||
val subListener =
|
||||
object : SubscriptionListener {
|
||||
override fun onEvent(
|
||||
event: Event,
|
||||
isLive: Boolean,
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
// The filter matches nothing; any stray event still proves liveness.
|
||||
eoseMs.getOrPut(relay) { mark.elapsedNow().inWholeMilliseconds }
|
||||
}
|
||||
|
||||
override fun onEose(
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
eoseMs.getOrPut(relay) { mark.elapsedNow().inWholeMilliseconds }
|
||||
terminals.trySend(relay)
|
||||
}
|
||||
|
||||
override fun onClosed(
|
||||
message: String,
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
// A CLOSED is an app-level ANSWER (auth wall, policy, …): the relay
|
||||
// is working. Record why so the caller can see the wall.
|
||||
errors.getOrPut(relay) { "closed:$message" }
|
||||
terminals.trySend(relay)
|
||||
}
|
||||
|
||||
override fun onCannotConnect(
|
||||
relay: NormalizedRelayUrl,
|
||||
message: String,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
errors.getOrPut(relay) { "cannot:$message" }
|
||||
terminals.trySend(relay)
|
||||
}
|
||||
}
|
||||
|
||||
client.addConnectionListener(connListener)
|
||||
try {
|
||||
client.subscribe(subId, wave.associateWith { PROBE_FILTERS }, subListener)
|
||||
val remaining = wave.toMutableSet()
|
||||
withTimeoutOrNull(timeoutMs) {
|
||||
while (remaining.isNotEmpty()) {
|
||||
remaining.remove(terminals.receive())
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
client.unsubscribe(subId)
|
||||
client.removeConnectionListener(connListener)
|
||||
terminals.close()
|
||||
}
|
||||
|
||||
return wave.map { relay ->
|
||||
val opened = openRtt[relay]
|
||||
val answered = eoseMs[relay]
|
||||
val error = errors[relay]
|
||||
// Reachable = the socket opened OR the relay answered at the app level
|
||||
// (EOSE, or a CLOSED — an auth/policy wall is still a working relay).
|
||||
// Only a connect failure, or silence with no socket, is dead.
|
||||
val cannot = error?.startsWith("cannot:") == true
|
||||
val reachable = !cannot && (opened != null || answered != null || error != null)
|
||||
Verdict(
|
||||
relay = relay,
|
||||
reachable = reachable,
|
||||
rttOpenMs = opened ?: -1,
|
||||
rttEoseMs = answered ?: -1,
|
||||
error = error,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
companion object {
|
||||
// A filter no event can match (ids are 64-hex of a hash): the relay answers
|
||||
// with an immediate EOSE and never streams a payload. Same trick as the
|
||||
// crawler's warm pool.
|
||||
private val PROBE_FILTERS = listOf(Filter(ids = listOf("0".repeat(64))))
|
||||
|
||||
/**
|
||||
* The relay universe the local store knows: every read/write relay advertised
|
||||
* in any stored kind:10002. Callers typically union this with the reachability
|
||||
* cache's live+dead sets so previously-probed relays are re-checked too.
|
||||
* `.onion` relays are excluded unless [includeOnion] — without a Tor transport
|
||||
* they'd only burn a wave slot. [maxPerAuthority] bounds how many distinct
|
||||
* URLs are kept per host[:port]: paid/filter relays mint one path URL per user
|
||||
* (`wss://filter.example/npubA`, `/npubB`, …), and probing hundreds of paths
|
||||
* of ONE server is redundant (liveness is a server property) and rude.
|
||||
*/
|
||||
suspend fun knownRelayUniverse(
|
||||
store: IEventStore,
|
||||
includeOnion: Boolean = false,
|
||||
maxPerAuthority: Int = 3,
|
||||
): Set<NormalizedRelayUrl> {
|
||||
val out = HashSet<NormalizedRelayUrl>()
|
||||
val perAuthority = HashMap<String, Int>()
|
||||
for (ev in store.query<Event>(Filter(kinds = listOf(AdvertisedRelayListEvent.KIND)))) {
|
||||
if (ev !is AdvertisedRelayListEvent) continue
|
||||
for (relay in ev.relaysNorm()) {
|
||||
if (!includeOnion && RelayUrlNormalizer.isOnion(relay.url)) continue
|
||||
if (relay in out) continue
|
||||
val authority = authorityOf(relay.url)
|
||||
val count = perAuthority[authority] ?: 0
|
||||
if (count >= maxPerAuthority) continue
|
||||
perAuthority[authority] = count + 1
|
||||
out.add(relay)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/** host[:port] between the ws/wss scheme and the first path slash. */
|
||||
private fun authorityOf(url: String): String {
|
||||
val afterScheme =
|
||||
when {
|
||||
url.startsWith("wss://") -> url.substring(6)
|
||||
url.startsWith("ws://") -> url.substring(5)
|
||||
else -> url
|
||||
}
|
||||
val slash = afterScheme.indexOf('/')
|
||||
return if (slash >= 0) afterScheme.substring(0, slash) else afterScheme
|
||||
}
|
||||
}
|
||||
}
|
||||
+15
@@ -142,6 +142,21 @@ class RelayReachabilityStore(
|
||||
for (relay in dead) if (relay !in reachable) writeOne(relay, up = false, now, rttOpenMs)
|
||||
}
|
||||
|
||||
/**
|
||||
* Like [record], but with a real, per-relay measured open round-trip — the shape a
|
||||
* dedicated probe (see RelayProber) produces. Each reachable relay's record carries
|
||||
* ITS OWN `rtt-open`, so the cache doubles as a latency census: a later reader can
|
||||
* separate fast relays from working-but-slow ones instead of only live from dead.
|
||||
*/
|
||||
suspend fun recordProbed(
|
||||
reachableRttMs: Map<NormalizedRelayUrl, Long>,
|
||||
dead: Set<NormalizedRelayUrl>,
|
||||
now: Long = TimeUtils.now(),
|
||||
) {
|
||||
for ((relay, rtt) in reachableRttMs) writeOne(relay, up = true, now, rtt.coerceAtLeast(0))
|
||||
for (relay in dead) if (relay !in reachableRttMs) writeOne(relay, up = false, now, 0)
|
||||
}
|
||||
|
||||
private suspend fun writeOne(
|
||||
relay: NormalizedRelayUrl,
|
||||
up: Boolean,
|
||||
|
||||
+103
@@ -0,0 +1,103 @@
|
||||
/*
|
||||
* 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.quartz.nip66RelayMonitor.reachability
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
|
||||
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
|
||||
import com.vitorpamplona.quartz.nip65RelayList.tags.AdvertisedRelayInfo
|
||||
import com.vitorpamplona.quartz.nip65RelayList.tags.AdvertisedRelayType
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class RelayProberUniverseTest {
|
||||
private fun norm(url: String) = RelayUrlNormalizer.normalizeOrNull(url)!!
|
||||
|
||||
private fun relayInfo(url: String) = AdvertisedRelayInfo(norm(url), AdvertisedRelayType.BOTH)
|
||||
|
||||
@Test
|
||||
fun collectsAdvertisedRelaysAcrossAuthors() =
|
||||
runBlocking {
|
||||
val store = EventStore(null)
|
||||
store.insert(AdvertisedRelayListEvent.create(listOf(relayInfo("wss://a.example"), relayInfo("wss://b.example")), NostrSignerSync()))
|
||||
store.insert(AdvertisedRelayListEvent.create(listOf(relayInfo("wss://b.example"), relayInfo("wss://c.example")), NostrSignerSync()))
|
||||
|
||||
val universe = RelayProber.knownRelayUniverse(store)
|
||||
assertEquals(setOf(norm("wss://a.example"), norm("wss://b.example"), norm("wss://c.example")), universe)
|
||||
store.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun capsPathUrlsPerAuthority() =
|
||||
runBlocking {
|
||||
val store = EventStore(null)
|
||||
// One paid/filter host advertised as one path URL per user.
|
||||
val paths = (1..10).map { relayInfo("wss://filter.example/user$it") }
|
||||
store.insert(AdvertisedRelayListEvent.create(paths + relayInfo("wss://solo.example"), NostrSignerSync()))
|
||||
|
||||
val universe = RelayProber.knownRelayUniverse(store, maxPerAuthority = 3)
|
||||
assertEquals(3, universe.count { it.url.contains("filter.example") }, "per-authority URLs must be capped")
|
||||
assertTrue(norm("wss://solo.example") in universe)
|
||||
store.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun onionRelaysAreExcludedByDefault() =
|
||||
runBlocking {
|
||||
val store = EventStore(null)
|
||||
store.insert(
|
||||
AdvertisedRelayListEvent.create(
|
||||
listOf(relayInfo("wss://clear.example"), relayInfo("ws://someonionaddressabcdefghijklmnop.onion")),
|
||||
NostrSignerSync(),
|
||||
),
|
||||
)
|
||||
|
||||
val universe = RelayProber.knownRelayUniverse(store)
|
||||
assertEquals(setOf(norm("wss://clear.example")), universe)
|
||||
|
||||
val withOnion = RelayProber.knownRelayUniverse(store, includeOnion = true)
|
||||
assertEquals(2, withOnion.size)
|
||||
store.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun recordProbedKeepsPerRelayRttAndLiveWins() =
|
||||
runBlocking {
|
||||
val store = EventStore(null)
|
||||
val reach = RelayReachabilityStore(store, NostrSignerInternal(KeyPair()))
|
||||
|
||||
val fast = norm("wss://fast.example")
|
||||
val slow = norm("wss://slow.example")
|
||||
val dead = norm("wss://dead.example")
|
||||
reach.recordProbed(mapOf(fast to 42L, slow to 9000L), setOf(dead, slow))
|
||||
|
||||
val snapshot = reach.snapshot()
|
||||
// A relay both probed-live and reported dead stays live (live wins).
|
||||
assertEquals(setOf(fast, slow), snapshot.live)
|
||||
assertEquals(setOf(dead), snapshot.dead)
|
||||
store.close()
|
||||
}
|
||||
}
|
||||
+107
@@ -0,0 +1,107 @@
|
||||
/*
|
||||
* 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.quartz.nip01Core.relay.sockets.okhttp
|
||||
|
||||
import okhttp3.Dns
|
||||
import java.net.InetAddress
|
||||
import java.net.UnknownHostException
|
||||
import java.util.concurrent.CompletableFuture
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* An in-process DNS cache for OkHttp that makes mass parallel connections viable:
|
||||
*
|
||||
* - **Positive cache** ([positiveTtlMs]): the JVM's own `InetAddress` cache holds
|
||||
* entries for ~30s — useless across a 30-minute crawl that re-dials the same
|
||||
* hosts every round. Resolved addresses are kept for the TTL so reconnects and
|
||||
* the outbox model's many per-user path URLs on one host
|
||||
* (`wss://filter.example/npubA`, `/npubB`, …) resolve instantly.
|
||||
* - **Negative cache** ([negativeTtlMs]): a dead domain is the single most
|
||||
* expensive lookup — glibc `getaddrinfo` retries nameservers for 10–30s while
|
||||
* holding an OkHttp dispatcher thread — and a crawl re-dials dead relays from
|
||||
* every straggler's outbox. The first [UnknownHostException] is remembered and
|
||||
* re-thrown immediately for the TTL, so later dials fail in microseconds.
|
||||
* - **In-flight dedup**: concurrent lookups of the same host (a connect storm
|
||||
* dialing hundreds of URLs on one authority at once) collapse onto a single
|
||||
* delegate call; the rest wait on its [CompletableFuture] instead of stacking
|
||||
* N identical blocking `getaddrinfo` calls.
|
||||
*
|
||||
* Only [UnknownHostException] is negative-cached — a resolver that *errors*
|
||||
* (interrupted, SecurityException, …) is not proof the name is bad, so those
|
||||
* propagate uncached. Entries are evicted lazily on the next lookup after
|
||||
* expiry; the map is bounded by the distinct-host universe (a few thousand for
|
||||
* a full crawl), so no active eviction is needed.
|
||||
*/
|
||||
class CachingDns(
|
||||
private val delegate: Dns = Dns.SYSTEM,
|
||||
private val positiveTtlMs: Long = 10 * 60_000L,
|
||||
private val negativeTtlMs: Long = 10 * 60_000L,
|
||||
private val nowMs: () -> Long = System::currentTimeMillis,
|
||||
) : Dns {
|
||||
private class Entry(
|
||||
/** Resolved addresses, or null for a cached resolution failure. */
|
||||
val addresses: List<InetAddress>?,
|
||||
val expiresAtMs: Long,
|
||||
)
|
||||
|
||||
private val cache = ConcurrentHashMap<String, Entry>()
|
||||
private val inFlight = ConcurrentHashMap<String, CompletableFuture<List<InetAddress>>>()
|
||||
|
||||
override fun lookup(hostname: String): List<InetAddress> {
|
||||
val hit = cache[hostname]
|
||||
if (hit != null) {
|
||||
if (nowMs() < hit.expiresAtMs) {
|
||||
return hit.addresses
|
||||
?: throw UnknownHostException("$hostname (cached DNS failure)")
|
||||
}
|
||||
cache.remove(hostname, hit)
|
||||
}
|
||||
|
||||
// One resolver call per host at a time: the creator runs the delegate,
|
||||
// everyone else who raced in blocks on the same future.
|
||||
val future = CompletableFuture<List<InetAddress>>()
|
||||
val existing = inFlight.putIfAbsent(hostname, future)
|
||||
if (existing != null) {
|
||||
return try {
|
||||
existing.join()
|
||||
} catch (e: Exception) {
|
||||
throw (e.cause as? UnknownHostException) ?: UnknownHostException("$hostname (concurrent lookup failed)")
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
val addresses = delegate.lookup(hostname)
|
||||
cache[hostname] = Entry(addresses, nowMs() + positiveTtlMs)
|
||||
future.complete(addresses)
|
||||
return addresses
|
||||
} catch (e: UnknownHostException) {
|
||||
cache[hostname] = Entry(null, nowMs() + negativeTtlMs)
|
||||
future.completeExceptionally(e)
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
// Not proof the name is bad (interrupt, security, …) — don't cache.
|
||||
future.completeExceptionally(e)
|
||||
throw e
|
||||
} finally {
|
||||
inFlight.remove(hostname, future)
|
||||
}
|
||||
}
|
||||
}
|
||||
+125
@@ -0,0 +1,125 @@
|
||||
/*
|
||||
* 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.quartz.nip01Core.relay.sockets.okhttp
|
||||
|
||||
import okhttp3.Dns
|
||||
import java.net.InetAddress
|
||||
import java.net.UnknownHostException
|
||||
import java.util.concurrent.CountDownLatch
|
||||
import java.util.concurrent.Executors
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class CachingDnsTest {
|
||||
private val addr = listOf(InetAddress.getByAddress("good.example", byteArrayOf(10, 0, 0, 1)))
|
||||
|
||||
private class CountingDns(
|
||||
val onLookup: (String) -> List<InetAddress>,
|
||||
) : Dns {
|
||||
val calls = AtomicInteger(0)
|
||||
|
||||
override fun lookup(hostname: String): List<InetAddress> {
|
||||
calls.incrementAndGet()
|
||||
return onLookup(hostname)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun cachesPositiveLookupsUntilTtl() {
|
||||
var now = 0L
|
||||
val delegate = CountingDns { addr }
|
||||
val dns = CachingDns(delegate, positiveTtlMs = 1000, negativeTtlMs = 1000, nowMs = { now })
|
||||
|
||||
assertEquals(addr, dns.lookup("good.example"))
|
||||
assertEquals(addr, dns.lookup("good.example"))
|
||||
assertEquals(1, delegate.calls.get(), "second lookup within TTL must be served from cache")
|
||||
|
||||
now = 1001
|
||||
assertEquals(addr, dns.lookup("good.example"))
|
||||
assertEquals(2, delegate.calls.get(), "expired entry must re-resolve")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun cachesUnknownHostFailuresUntilTtl() {
|
||||
var now = 0L
|
||||
val delegate = CountingDns { throw UnknownHostException(it) }
|
||||
val dns = CachingDns(delegate, positiveTtlMs = 1000, negativeTtlMs = 1000, nowMs = { now })
|
||||
|
||||
assertFailsWith<UnknownHostException> { dns.lookup("dead.example") }
|
||||
assertFailsWith<UnknownHostException> { dns.lookup("dead.example") }
|
||||
assertEquals(1, delegate.calls.get(), "second failing lookup within TTL must be the cached failure")
|
||||
|
||||
now = 1001
|
||||
assertFailsWith<UnknownHostException> { dns.lookup("dead.example") }
|
||||
assertEquals(2, delegate.calls.get(), "expired negative entry must re-resolve")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun nonUnknownHostErrorsAreNotCached() {
|
||||
val delegate = CountingDns { throw RuntimeException("resolver interrupted") }
|
||||
val dns = CachingDns(delegate, nowMs = { 0 })
|
||||
|
||||
assertFailsWith<RuntimeException> { dns.lookup("flaky.example") }
|
||||
assertFailsWith<RuntimeException> { dns.lookup("flaky.example") }
|
||||
assertEquals(2, delegate.calls.get(), "a transient resolver error must not be negative-cached")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun concurrentLookupsOfSameHostCollapseToOneDelegateCall() {
|
||||
val started = CountDownLatch(1)
|
||||
val release = CountDownLatch(1)
|
||||
val delegate =
|
||||
CountingDns {
|
||||
started.countDown()
|
||||
release.await(5, TimeUnit.SECONDS)
|
||||
addr
|
||||
}
|
||||
val dns = CachingDns(delegate, nowMs = { 0 })
|
||||
val pool = Executors.newFixedThreadPool(8)
|
||||
try {
|
||||
val results =
|
||||
(1..8).map {
|
||||
pool.submit<List<InetAddress>> {
|
||||
if (it == 1) {
|
||||
dns.lookup("busy.example")
|
||||
} else {
|
||||
// Wait until the first lookup is inside the delegate so the
|
||||
// rest genuinely race against an in-flight resolution.
|
||||
started.await(5, TimeUnit.SECONDS)
|
||||
dns.lookup("busy.example")
|
||||
}
|
||||
}
|
||||
}
|
||||
// Give the racers a moment to pile onto the in-flight future, then release.
|
||||
started.await(5, TimeUnit.SECONDS)
|
||||
Thread.sleep(50)
|
||||
release.countDown()
|
||||
for (f in results) assertEquals(addr, f.get(5, TimeUnit.SECONDS))
|
||||
assertTrue(delegate.calls.get() <= 2, "concurrent lookups should collapse (got ${delegate.calls.get()} delegate calls)")
|
||||
} finally {
|
||||
pool.shutdownNow()
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user