From 2bad6779ce52fa3327f461c157c4465a138efec6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 01:13:23 +0000 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_013zEYRGKF943RgLaHTViJaB --- .../com/vitorpamplona/amethyst/cli/Context.kt | 50 +++- .../com/vitorpamplona/amethyst/cli/Main.kt | 8 + .../amethyst/cli/commands/GrapeRankCommand.kt | 111 +++++++- .../2026-07-10-graperank-connect-storm.md | 89 ++++++ .../graperank/GrapeRankCrawler.kt | 90 +++++- .../reachability/RelayProber.kt | 265 ++++++++++++++++++ .../reachability/RelayReachabilityStore.kt | 15 + .../reachability/RelayProberUniverseTest.kt | 103 +++++++ .../relay/sockets/okhttp/CachingDns.kt | 107 +++++++ .../relay/sockets/okhttp/CachingDnsTest.kt | 125 +++++++++ 10 files changed, 946 insertions(+), 17 deletions(-) create mode 100644 quartz/plans/2026-07-10-graperank-connect-storm.md create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberUniverseTest.kt create mode 100644 quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/CachingDns.kt create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/CachingDnsTest.kt diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt index e74998b91d..1142529607 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt @@ -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] diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt index c9aa8c2461..01cbec81dc 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt @@ -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 diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt index e9a7177e36..18c01309f6 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt @@ -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, + ): 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( + "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( + "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 diff --git a/quartz/plans/2026-07-10-graperank-connect-storm.md b/quartz/plans/2026-07-10-graperank-connect-storm.md new file mode 100644 index 0000000000..96313bc636 --- /dev/null +++ b/quartz/plans/2026-07-10-graperank-connect-storm.md @@ -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. diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankCrawler.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankCrawler.kt index 63ebda32bf..5aaa544c52 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankCrawler.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankCrawler.kt @@ -208,6 +208,27 @@ class GrapeRankCrawler( * permanently ignores an author's advertised home. See RelayReachabilityStore. */ val knownDeadRelays: Set = 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 = 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(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)))) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt new file mode 100644 index 0000000000..a95da3f70a --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt @@ -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, + val elapsedMs: Long, + ) { + val reachable: List get() = verdicts.filter { it.reachable } + + val dead: List get() = verdicts.filter { !it.reachable } + + /** Per-relay open RTT for the reachable set, 0 (= "live, latency unknown") when unobserved. */ + fun reachableRttMs(): Map = reachable.associate { it.relay to it.rttOpenMs.coerceAtLeast(0) } + + fun deadRelays(): Set = 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, + timeoutMs: Long = 15_000, + waveSize: Int = 1000, + ): Result { + val mark = TimeSource.Monotonic.markNow() + val all = ArrayList(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, + timeoutMs: Long, + ): List { + val mark = TimeSource.Monotonic.markNow() + val waveSet = wave.toHashSet() + val openRtt = ConcurrentMap() + val eoseMs = ConcurrentMap() + val errors = ConcurrentMap() + // 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(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?, + ) { + // The filter matches nothing; any stray event still proves liveness. + eoseMs.getOrPut(relay) { mark.elapsedNow().inWholeMilliseconds } + } + + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + eoseMs.getOrPut(relay) { mark.elapsedNow().inWholeMilliseconds } + terminals.trySend(relay) + } + + override fun onClosed( + message: String, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + // 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?, + ) { + 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 { + val out = HashSet() + val perAuthority = HashMap() + for (ev in store.query(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 + } + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayReachabilityStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayReachabilityStore.kt index 93f6648639..3d3631a5f0 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayReachabilityStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayReachabilityStore.kt @@ -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, + dead: Set, + 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, diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberUniverseTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberUniverseTest.kt new file mode 100644 index 0000000000..846f4ac3e9 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberUniverseTest.kt @@ -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() + } +} diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/CachingDns.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/CachingDns.kt new file mode 100644 index 0000000000..d6867dc06d --- /dev/null +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/CachingDns.kt @@ -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?, + val expiresAtMs: Long, + ) + + private val cache = ConcurrentHashMap() + private val inFlight = ConcurrentHashMap>>() + + override fun lookup(hostname: String): List { + 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>() + 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) + } + } +} diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/CachingDnsTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/CachingDnsTest.kt new file mode 100644 index 0000000000..1b6d0d441d --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/CachingDnsTest.kt @@ -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, + ) : Dns { + val calls = AtomicInteger(0) + + override fun lookup(hostname: String): List { + 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 { dns.lookup("dead.example") } + assertFailsWith { dns.lookup("dead.example") } + assertEquals(1, delegate.calls.get(), "second failing lookup within TTL must be the cached failure") + + now = 1001 + assertFailsWith { 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 { dns.lookup("flaky.example") } + assertFailsWith { 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> { + 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() + } + } +}