From 156176f5fe8c72df074beb6438f85d6722d0797f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 13:22:55 +0000 Subject: [PATCH 1/8] fix(quartz): stop RelayUrlNormalizer from accepting urls that can never be relays MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Validated against a 45k-entry corpus of relay-url hints exported from real events (317k tag occurrences). The normalizer was converting ~30k distinct https:// urls with paths (Mastodon/bridge actor urls from proxy tags, web pages, images) into wss:// addresses that can never answer, wasting connection attempts and relay-pool slots. - http(s) → ws(s) scheme swap now only applies to bare hosts (host[:port] plus optional trailing slash); an http url with a path, query or fragment is a web resource, not a mistyped relay. - Authority validation for all schemes: rejects empty hosts, userinfo (@), percent-encoding and commas in the host, and paths that start with // (the signature of a second pasted url, e.g. wss://https//host). - Interior whitespace and backslashes reject the whole string (multiple urls or prose in one field). - Zero-width characters (U+200B..D, U+2060, BOM) are stripped instead of corrupting the parse (wss://\u200Bnos.lol previously normalized to the scheme-less //nos.lol/). - Schemeless candidates must look like host[:port] (single colon, numeric port), rejecting addressable pointers (31990:pubkey:dtag) and bare scheme leftovers (wss:) before the expensive RFC 3986 parse. - Protocol-relative //host/ inputs normalize as wss:// instead of resolving to https://. - normalizeOrNull now double-checks the parser output still starts with ws(s):// and rejects otherwise. Corpus impact: 30,014 garbage urls (30,333 events) now rejected, 0 real relays lost (all 15,162 kept urls normalize byte-identically), 6 broken outputs fixed. fix() itself stays allocation-free on the happy path (~357ns vs ~318ns per call on the garbage-heavy corpus). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A9sSyh1QLJD3PZ18tVPPVK --- .../relay/normalizer/RelayUrlNormalizer.kt | 147 ++++++++++++++++-- .../nip01Core/relay/RelayUrlFormatterTest.kt | 81 ++++++++++ 2 files changed, 216 insertions(+), 12 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt index 04f8c9cc06..8949a0b82c 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt @@ -84,6 +84,97 @@ class RelayUrlNormalizer { private fun norm(url: String) = NormalizedRelayUrl(Rfc3986.normalize(url)) + private fun isInvisible(c: Char) = c == '\u200B' || c == '\u200C' || c == '\u200D' || c == '\u2060' || c == '\uFEFF' + + /** + * Scans the authority (host[:port]) that starts at [start] and ends at the first + * `/`, `?` or `#`. Returns the end index, or -1 when the authority is empty or + * contains characters that never appear in a real relay host (`@` userinfo, + * percent-encoding, commas). + */ + private fun authorityEnd( + url: String, + start: Int, + ): Int { + if (start >= url.length) return -1 + var i = start + if (url[i] == '[') { + // IPv6 literal: defer validation to the RFC 3986 parser + while (i < url.length && url[i] != '/' && url[i] != '?' && url[i] != '#') i++ + return i + } + while (i < url.length) { + val c = url[i] + if (c == '/' || c == '?' || c == '#') break + if (c == '@' || c == '%' || c == ',') return -1 + i++ + } + return if (i == start) -1 else i + } + + /** + * Accepts a ws/wss url whose host starts at [hostStart] if the authority is sane + * and the path does not start with `//` (the signature of a second URL or a broken + * `https//` pasted after the scheme, e.g. `wss://https//nostr.watch/relay/x`). + */ + private fun fixWs( + url: String, + hostStart: Int, + ): String? { + val end = authorityEnd(url, hostStart) + if (end < 0) return null + if (end + 1 < url.length && url[end] == '/' && url[end + 1] == '/') return null + return url + } + + /** + * Converts an http(s) url to ws(s) only when it is a bare host — nothing after + * `host[:port]` but an optional trailing `/`. An http url with a path, query or + * fragment (Mastodon actor urls from bridge `proxy` tags, web pages, images) is + * a web resource, not a relay: converting it creates a wss:// url that can never + * answer and only wastes connection attempts. + */ + private fun fixHttp( + url: String, + hostStart: Int, + newScheme: String, + ): String? { + val end = authorityEnd(url, hostStart) + if (end < 0) return null + val bareHost = end == url.length || (end == url.length - 1 && url[end] == '/') + if (!bareHost) return null + return "$newScheme${url.substring(hostStart)}" + } + + /** + * Validates a schemeless candidate: the part before the first `/` must look like + * `host` or `host:port` — letters, digits, `.`, `-`, `_`, plus at most one `:` + * followed by digits only. Rejects addressable-event pointers (`31990:hex:dtag`), + * bare scheme leftovers (`wss:`) and anything else that would otherwise be blindly + * prefixed with `wss://`. + */ + private fun isBareHostAndPath(url: String): Boolean { + if (url[0] == '[') return true // IPv6 literal: defer to the RFC 3986 parser + var i = 0 + var portStart = -1 + while (i < url.length) { + val c = url[i] + if (c == '/') break + if (c == ':') { + if (portStart >= 0) return false + portStart = i + 1 + } else if (portStart >= 0) { + if (c < '0' || c > '9') return false + } else if (!c.isLetterOrDigit() && c != '.' && c != '-' && c != '_') { + return false + } + i++ + } + if (i == 0) return false + if (portStart >= 0 && portStart == i) return false + return true + } + @OptIn(ExperimentalContracts::class) fun fix(rawUrl: String): String? { if (rawUrl.length < 4) return null @@ -109,17 +200,33 @@ class RelayUrlNormalizer { } } - val trimmed = + var trimmed = if (url[0].isWhitespace() || url[url.length - 1].isWhitespace()) { url.trim() } else { url } + // Single pass: interior whitespace means multiple urls or prose in one field, + // backslashes never appear in a real relay url; both are garbage. Invisible + // characters (zero-width spaces, BOM) are copy-paste artifacts — strip them. + var hasInvisible = false + for (c in trimmed) { + if (c == '\\') return null + if (c.isWhitespace()) return null + if (isInvisible(c)) hasInvisible = true + } + if (hasInvisible) { + trimmed = buildString(trimmed.length) { for (c in trimmed) if (!isInvisible(c)) append(c) } + if (trimmed.length < 4) return null + } + // fast for good wss:// urls if (isRelaySchemePrefix(trimmed)) { - if (isRelaySchemePrefixSecure(trimmed) || isRelaySchemePrefixInsecure(trimmed)) { - return trimmed + if (isRelaySchemePrefixSecure(trimmed)) { + return fixWs(trimmed, 6) + } else if (isRelaySchemePrefixInsecure(trimmed)) { + return fixWs(trimmed, 5) } } @@ -127,31 +234,31 @@ class RelayUrlNormalizer { if (isHttpPrefix(trimmed)) { if (isHttpSSuffix(trimmed)) { // https:// - return "wss://${trimmed.drop(8)}" + return fixHttp(trimmed, 8, "wss://") } else if (isHttpSuffix(trimmed)) { // http:// - return "ws://${trimmed.drop(7)}" + return fixHttp(trimmed, 7, "ws://") } } // fast for good ww:// urls if (trimmed.startsWith("ww://")) { - return "wss://${trimmed.drop(5)}" + return fixWs("wss://${trimmed.drop(5)}", 6) } // fast for good ww:// urls if (trimmed.startsWith("was://")) { - return "wss://${trimmed.drop(6)}" + return fixWs("wss://${trimmed.drop(6)}", 6) } // fast for good ww:// urls if (trimmed.startsWith("Wws://")) { - return "wss://${trimmed.drop(6)}" + return fixWs("wss://${trimmed.drop(6)}", 6) } // fast for good ww:// urls if (trimmed.startsWith("Wss://")) { - return "wss://${trimmed.drop(6)}" + return fixWs("wss://${trimmed.drop(6)}", 6) } if (trimmed.contains("://")) { @@ -160,10 +267,19 @@ class RelayUrlNormalizer { return null } - return if (isOnion(trimmed) || isLocalHost(trimmed)) { - "ws://$trimmed" + // protocol-relative urls (`//host/`) are just missing the scheme + val bare = if (trimmed.startsWith("//")) trimmed.drop(2) else trimmed + if (bare.length < 4) return null + + if (!isBareHostAndPath(bare)) { + Log.d("RelayUrlNormalizer") { "Rejected $url" } + return null + } + + return if (isOnion(bare) || isLocalHost(bare)) { + "ws://$bare" } else { - "wss://$trimmed" + "wss://$bare" } } @@ -186,6 +302,13 @@ class RelayUrlNormalizer { val fixed = fix(url) if (fixed != null) { val normalized = norm(fixed) + // the RFC 3986 parser can drop or replace the scheme on odd inputs; + // anything that is not ws(s):// at this point cannot be connected to. + if (!isRelayUrl(normalized.url)) { + Log.d("NormalizedRelayUrl") { "Rejected $url" } + normalizedUrls.put(url, NormalizationResult.Error) + return null + } normalizedUrls.put(url, NormalizationResult.Success(normalized)) normalized } else { diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt index 82287dcbad..fc708eddba 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt @@ -52,4 +52,85 @@ class RelayUrlFormatterTest { fun weirdRelay() { assertNull(RelayUrlNormalizer.normalizeOrNull("wss://relay%20list%20to%20discover%20the%20user's%20content")) } + + @Test + fun httpWithPathIsNotARelay() { + // Mastodon/bridge actor urls from `proxy` tags: web resources, not relays + assertNull(RelayUrlNormalizer.normalizeOrNull("https://mastodon.social/users/amanita_muscaria")) + assertNull(RelayUrlNormalizer.normalizeOrNull("https://fosstodon.org/ap/users/115532410310000993")) + assertNull(RelayUrlNormalizer.normalizeOrNull("http://example.com/relay")) + assertNull(RelayUrlNormalizer.normalizeOrNull("https://nostr.mom/?author=0")) + assertNull(RelayUrlNormalizer.normalizeOrNull("https://nostr.mom/#section")) + + // but bare hosts still convert, with or without port and trailing slash + assertEquals("wss://nostr.mom/", RelayUrlNormalizer.normalizeOrNull("https://nostr.mom")?.url) + assertEquals("wss://nostr.mom/", RelayUrlNormalizer.normalizeOrNull("https://nostr.mom/")?.url) + assertEquals("wss://nostr.mom:4443/", RelayUrlNormalizer.normalizeOrNull("https://nostr.mom:4443/")?.url) + assertEquals("ws://nostr.mom/", RelayUrlNormalizer.normalizeOrNull("http://nostr.mom")?.url) + } + + @Test + fun wsWithPathIsStillARelay() { + assertEquals("wss://relay.nostr.band/all", RelayUrlNormalizer.normalizeOrNull("wss://relay.nostr.band/all")?.url) + assertEquals( + "wss://bostr.lecturify.net/?accept=0,1", + RelayUrlNormalizer.normalizeOrNull("wss://bostr.lecturify.net/?accept=0,1")?.url, + ) + } + + @Test + fun brokenSchemeGarbage() { + assertNull(RelayUrlNormalizer.normalizeOrNull("wss:")) + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://https//nostr.watch/relay/nostr.21crypto.ch")) + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://https://lockbox.fiatjaf.com")) + assertNull(RelayUrlNormalizer.normalizeOrNull("ws://http//nos.lol")) + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://://plebstr.com")) + } + + @Test + fun nostrUriIsNotARelay() { + assertNull(RelayUrlNormalizer.normalizeOrNull("nostr://nrelay1qqxhwumn8ghj77tpvf6jumt9e2ckgn/")) + assertNull(RelayUrlNormalizer.normalizeOrNull("nostr://npub1dwy079xmpz7mk02kvz6wan49h02635umk32aa4ufek8t8mjxv58qy2nr22/")) + assertNull(RelayUrlNormalizer.normalizeOrNull("nostr:nrelay1qq8k2cnfwejhyum99eek7cmfv9kqsm7sdm")) + } + + @Test + fun addressablePointerIsNotARelay() { + assertNull(RelayUrlNormalizer.normalizeOrNull("31990:6be38f8c63df7dbf84db7ec4a6e6fbbd8d19dca3b980efad18585c46f04b26f9:mostr")) + } + + @Test + fun authorityGarbage() { + // userinfo, percent-encoding and commas never appear in a real relay host + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://catuaba@plebs.place/")) + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://africa.nostr.joburg%0A/")) + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://bitcoiner,social/")) + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://#web3/")) + assertNull(RelayUrlNormalizer.normalizeOrNull("name@domain.com")) + } + + @Test + fun interiorWhitespaceAndBackslashes() { + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://nos lol")) + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://nos.lol/ wss:/nostr.land/ avatar wss:/nostr.wine/")) + assertNull(RelayUrlNormalizer.normalizeOrNull("wss://\\\\relay.damus.io/")) + } + + @Test + fun invisibleCharactersAreStripped() { + assertEquals("wss://nos.lol/", RelayUrlNormalizer.normalizeOrNull("wss://\u200Bnos.lol")?.url) + assertEquals("wss://nos.lol/", RelayUrlNormalizer.normalizeOrNull("\uFEFFwss://nos.lol")?.url) + } + + @Test + fun protocolRelativeUrls() { + assertEquals("wss://relay.most.pub/", RelayUrlNormalizer.normalizeOrNull("//relay.most.pub/")?.url) + assertEquals("wss://nos.lol/", RelayUrlNormalizer.normalizeOrNull("//nos.lol/")?.url) + } + + @Test + fun ipv6AndLanHostsStillWork() { + assertEquals("ws://[31b:6f20:c7f2:3ddf::3221]/", RelayUrlNormalizer.normalizeOrNull("ws://[31b:6f20:c7f2:3ddf::3221]/")?.url) + assertEquals("ws://geyser-relay:7777/", RelayUrlNormalizer.normalizeOrNull("ws://geyser-relay:7777/")?.url) + } } From ba7cc72dd28bdffee33fba3cc459e1e6be8b0381 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 13:22:55 +0000 Subject: [PATCH 2/8] feat(cli): amy relay probe --file to census external relay-url candidates Feeds a file of raw candidate urls (one per line) through the same RelayUrlNormalizer the app uses, then probes the surviving clearnet set alongside the store's known universe. Rejected and onion counts are reported (file_urls/file_normalized/file_rejected in the JSON output), and results land in the NIP-66 kind:30166 reachability cache keyed by the normalized url as d-tag, as usual. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A9sSyh1QLJD3PZ18tVPPVK --- .../amethyst/cli/commands/RelayCommands.kt | 37 +++++++++++++++++-- 1 file changed, 34 insertions(+), 3 deletions(-) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt index 64084a9356..8734ba1b02 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageRelayListEvent import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrlOrNull import com.vitorpamplona.quartz.nip01Core.relay.normalizer.toHttp import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation @@ -50,6 +51,7 @@ import com.vitorpamplona.quartz.nip65RelayList.tags.AdvertisedRelayType import com.vitorpamplona.quartz.nip66RelayMonitor.reachability.RelayProber import okhttp3.OkHttpClient import okhttp3.Request +import java.io.File /** * `amy relay …` — manage every relay list this account maintains, mirroring @@ -112,10 +114,12 @@ object RelayCommands { | relay info URL fetch + print a relay's NIP-11 info document (stateless) | relay 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), + | [--file PATH] into the reachability cache (NIP-66 kind:30166), | so reachability-aware commands (graperank crawl/ | refresh) skip dead relays and wait once - | (--timeout: per wave, default 15s) + | (--timeout: per wave, default 15s; --file: also + | probe candidate urls, one per line, each run + | through the relay url normalizer first) """.trimMargin() // ------------------------------------------------------------------ @@ -337,12 +341,36 @@ object RelayCommands { // Relays dialed at once; --relay-concurrency accepted as the alias the // graperank verbs spell it with. val waveSize = args.intFlag("concurrency", args.intFlag("relay-concurrency", Context.defaultPreconnectCap)) + // Optional external candidate list: one raw url per line, run through the + // same RelayUrlNormalizer the app uses, so a probe doubles as a census of + // how a corpus of relay hints normalizes (rejects are counted, not dialed). + val fromFile = args.flag("file") args.rejectUnknown() + var fileRaw = 0 + var fileRejected = 0 + val fileRelays = HashSet() + if (fromFile != null) { + File(fromFile).forEachLine { line -> + if (line.isBlank()) return@forEachLine + fileRaw++ + val normalized = line.normalizeRelayUrlOrNull() + if (normalized == null) { + fileRejected++ + } else if (!RelayUrlNormalizer.isOnion(normalized.url)) { + fileRelays.add(normalized) + } + } + System.err.println( + "[relay-probe] $fromFile: $fileRaw urls → ${fileRelays.size} unique clearnet relays " + + "($fileRejected rejected by the normalizer)", + ) + } + Context.openOrAnonymous(dataDir).use { ctx -> ctx.prepare() val cached = ctx.reachability.snapshot() - val universe = RelayProber.knownRelayUniverse(ctx.store) + cached.live + cached.dead + val universe = RelayProber.knownRelayUniverse(ctx.store) + cached.live + cached.dead + fileRelays if (universe.isEmpty()) { Output.emit( linkedMapOf( @@ -381,6 +409,9 @@ object RelayCommands { Output.emit( linkedMapOf( "probed" to result.verdicts.size, + "file_urls" to (if (fromFile != null) fileRaw else null), + "file_normalized" to (if (fromFile != null) fileRelays.size else null), + "file_rejected" to (if (fromFile != null) fileRejected else null), "reachable" to result.reachable.size, "dead" to result.dead.size, "closed_by_policy" to authWalled, From d9a58950ecc24a566d3626948e106de6675c1be8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 13:57:00 +0000 Subject: [PATCH 3/8] feat(quartz): stream NIP-66 probe verdicts and expose them as signable 30166 templates RelayProber.probeFlow(urls) is a cold Flow that emits each relay's Verdict the moment the relay resolves (EOSE, CLOSED or connect failure) instead of at the end of the whole census; only silent relays wait for their wave's deadline. probeWave now resolves verdicts per-terminal, so the batch probe() shares the same path. Verdict.toDiscoveryEventTemplate() renders a verdict as an UNSIGNED kind:30166 template (d = normalized url, n network type, rtt-open when reachable, R auth when the probe hit a NIP-42 auth-required CLOSED) so an external consumer signs with its own monitor key: prober.probeFlow(urls).map { it.toDiscoveryEventTemplate() } .collect { publish(signer.sign(it)) } rtt-eose is deliberately never published as rtt-read: it is measured from the wave start (dial + TLS + queueing + read), and aggregators rank on rtt values. The RelayObserver/RelayMonitor path supplies honest rtt-read/rtt-write from real traffic. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A9sSyh1QLJD3PZ18tVPPVK --- .../reachability/RelayProber.kt | 101 ++++++-- .../reachability/RelayProberFlowTest.kt | 219 ++++++++++++++++++ 2 files changed, 301 insertions(+), 19 deletions(-) create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt 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 index a95da3f70a..364f4e1bd0 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt @@ -29,10 +29,19 @@ 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.signers.EventTemplate import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent +import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.RelayDiscoveryEvent +import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.networkType +import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.requirement +import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.rtt +import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.tags.RttType +import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.withTimeoutOrNull import kotlin.time.TimeSource @@ -102,7 +111,7 @@ class RelayProber( val distinct = relays.toSet() var done = 0 for (wave in distinct.chunked(waveSize.coerceAtLeast(1))) { - all += probeWave(wave, timeoutMs) + probeWave(wave, timeoutMs) { all += it } done += wave.size if (distinct.size > wave.size) { val liveSoFar = all.count { it.reachable } @@ -112,10 +121,33 @@ class RelayProber( return Result(all, mark.elapsedNow().inWholeMilliseconds) } + /** + * Streaming variant of [probe]: a cold [Flow] that emits each relay's [Verdict] + * the moment that relay resolves — an answering relay's verdict arrives as soon + * as its EOSE/CLOSED/connect-failure lands, not when the whole census ends. Only + * relays that stay silent wait for their wave's [timeoutMs] deadline. + * + * Probing starts when the flow is collected and pauses between waves while the + * collector is busy (emission is sequential). Pair each verdict with + * [toDiscoveryEventTemplate] to turn the stream into signable NIP-66 kind:30166 + * records for another process to sign and publish. + */ + fun probeFlow( + relays: Collection, + timeoutMs: Long = 15_000, + waveSize: Int = 1000, + ): Flow = + flow { + for (wave in relays.toSet().chunked(waveSize.coerceAtLeast(1))) { + probeWave(wave, timeoutMs) { emit(it) } + } + } + private suspend fun probeWave( wave: List, timeoutMs: Long, - ): List { + onVerdict: suspend (Verdict) -> Unit, + ) { val mark = TimeSource.Monotonic.markNow() val waveSet = wave.toHashSet() val openRtt = ConcurrentMap() @@ -178,22 +210,7 @@ class RelayProber( } } - 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 -> + fun verdictOf(relay: NormalizedRelayUrl): Verdict { val opened = openRtt[relay] val answered = eoseMs[relay] val error = errors[relay] @@ -202,7 +219,7 @@ class RelayProber( // 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( + return Verdict( relay = relay, reachable = reachable, rttOpenMs = opened ?: -1, @@ -210,6 +227,27 @@ class RelayProber( error = error, ) } + + client.addConnectionListener(connListener) + try { + client.subscribe(subId, wave.associateWith { PROBE_FILTERS }, subListener) + val remaining = wave.toMutableSet() + while (remaining.isNotEmpty()) { + val left = timeoutMs - mark.elapsedNow().inWholeMilliseconds + if (left <= 0) break + val relay = withTimeoutOrNull(left) { terminals.receive() } ?: break + // The emission happens OUTSIDE the timeout window so a collector that + // suspends on a verdict can never be cancelled mid-emission and lose it. + if (remaining.remove(relay)) onVerdict(verdictOf(relay)) + } + // Whatever is left resolved nothing by the deadline: dead if the socket + // never opened, reachable-but-slow if it did. + for (relay in remaining) onVerdict(verdictOf(relay)) + } finally { + client.unsubscribe(subId) + client.removeConnectionListener(connListener) + terminals.close() + } } companion object { @@ -263,3 +301,28 @@ class RelayProber( } } } + +/** + * This verdict as an UNSIGNED NIP-66 kind:30166 Relay Discovery template — the d-tag + * is the normalized relay url; sign it with the consumer's own monitor key (per + * NIP-66 a monitor is its own identity, so the prober never signs on its own). + * + * Only facts this probe actually observed are tagged: + * - `n` network type inferred from the url (clearnet/tor/i2p); + * - `rtt-open` when the relay was reachable — the measured WS-upgrade round trip, + * or 0 for "reachable, latency not observed" (liveness is the tag's PRESENCE); + * - `R auth` when the relay answered the probe REQ with a NIP-42 `auth-required` + * CLOSED — an observed auth wall, not a NIP-11 claim. + * + * [Verdict.rttEoseMs] is deliberately NOT written as `rtt-read`: it is measured from + * the wave start, so it bundles dial, TLS and any handshake queueing with the read — + * publishing it as a read round trip would hand aggregators an inflated latency. + * The [RelayObserver]/[RelayMonitor] path supplies honest `rtt-read`/`rtt-write` + * from real traffic instead. + */ +fun RelayProber.Verdict.toDiscoveryEventTemplate(createdAt: Long = TimeUtils.now()): EventTemplate = + RelayDiscoveryEvent.build(relay, createdAt = createdAt) { + networkType(RelayReachabilityStore.networkTypeOf(relay)) + if (reachable) rtt(RttType.OPEN, rttOpenMs.coerceAtLeast(0)) + if (error?.startsWith("closed:auth-required") == true) requirement("auth") + } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt new file mode 100644 index 0000000000..34b1ce15fc --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt @@ -0,0 +1,219 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip66RelayMonitor.reachability + +import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener +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.signers.EventTemplate +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.currentTime +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Pins [RelayProber.probeFlow]'s streaming contract: an answering relay's verdict + * is emitted the moment its terminal arrives, while silent relays only resolve at + * the wave deadline — and pins the observed-facts-only tag set of + * [toDiscoveryEventTemplate]. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class RelayProberFlowTest { + /** Captures the probe subscription so the test can play the relays. */ + private class ScriptedClient : INostrClient by EmptyNostrClient() { + var listener: SubscriptionListener? = null + + override fun subscribe( + subId: String, + filters: Map>, + listener: SubscriptionListener?, + ) { + this.listener = listener + } + } + + private val fast = RelayUrlNormalizer.normalize("wss://fast.example.com") + private val silent = RelayUrlNormalizer.normalize("wss://silent.example.com") + private val walled = RelayUrlNormalizer.normalize("wss://walled.example.com") + + @Test + fun answeringRelayStreamsBeforeTheWaveDeadline() = + runTest { + val client = ScriptedClient() + val arrivals = mutableListOf>() + + val collector = + launch { + RelayProber(client) + .probeFlow(listOf(fast, silent), timeoutMs = 10_000) + .collect { arrivals += it to currentTime } + } + launch { + delay(200) + client.listener!!.onEose(fast, null) + } + collector.join() + + assertEquals(listOf(fast, silent), arrivals.map { it.first.relay }) + // The EOSE'd relay resolved when it answered, not at the deadline … + val (fastVerdict, fastAt) = arrivals[0] + assertTrue(fastVerdict.reachable) + assertTrue(fastAt < 1_000, "verdict should stream at answer time, arrived at ${fastAt}ms") + // … while the silent one waited out the wave and is dead (socket never opened). + val (silentVerdict, silentAt) = arrivals[1] + assertFalse(silentVerdict.reachable) + assertTrue(silentAt >= 10_000, "silent relay must wait for the deadline, arrived at ${silentAt}ms") + } + + @Test + fun authWalledRelayIsReachableWithTheWallRecorded() = + runTest { + val client = ScriptedClient() + val arrivals = mutableListOf() + + val collector = + launch { + RelayProber(client) + .probeFlow(listOf(walled), timeoutMs = 10_000) + .collect { arrivals += it } + } + launch { + delay(100) + client.listener!!.onClosed("auth-required: sign in first", walled, null) + } + collector.join() + + assertEquals(1, arrivals.size) + assertTrue(arrivals[0].reachable, "an auth wall is an app-level answer: the relay works") + assertEquals("closed:auth-required: sign in first", arrivals[0].error) + } + + @Test + fun connectFailureStreamsImmediatelyAsDead() = + runTest { + val client = ScriptedClient() + val arrivals = mutableListOf>() + + val collector = + launch { + RelayProber(client) + .probeFlow(listOf(fast), timeoutMs = 10_000) + .collect { arrivals += it to currentTime } + } + launch { + delay(50) + client.listener!!.onCannotConnect(fast, "dns failure", null) + } + collector.join() + + val (verdict, at) = arrivals.single() + assertFalse(verdict.reachable) + assertEquals("cannot:dns failure", verdict.error) + assertTrue(at < 1_000, "a failed dial must not wait for the deadline, arrived at ${at}ms") + } + + // ------------------------------------------------------------------ + // toDiscoveryEventTemplate — only observed facts become tags + // ------------------------------------------------------------------ + + private fun tagsOf(template: EventTemplate<*>) = template.tags.map { it.toList() } + + @Test + fun reachableVerdictTemplateCarriesLivenessAndNetwork() { + val template = + RelayProber + .Verdict(fast, reachable = true, rttOpenMs = 150, rttEoseMs = 480, error = null) + .toDiscoveryEventTemplate(createdAt = 1000) + + val tags = tagsOf(template) + assertEquals(30166, template.kind) + assertEquals(1000, template.createdAt) + assertTrue(listOf("d", fast.url) in tags) + assertTrue(listOf("n", "clearnet") in tags) + assertTrue(listOf("rtt-open", "150") in tags) + // rtt-eose is wave-relative (dial + queue + read) — never published as rtt-read. + assertNull(tags.firstOrNull { it[0] == "rtt-read" }) + } + + @Test + fun deadVerdictTemplateHasNoRttOpen() { + val template = + RelayProber + .Verdict(silent, reachable = false, rttOpenMs = -1, rttEoseMs = -1, error = "cannot:timeout") + .toDiscoveryEventTemplate() + + val tags = tagsOf(template) + assertTrue(listOf("d", silent.url) in tags) + // Liveness is the PRESENCE of rtt-open; a dead record must not carry one. + assertNull(tags.firstOrNull { it[0] == "rtt-open" }) + } + + @Test + fun reachableWithoutMeasuredLatencyWritesZeroFlag() { + val template = + RelayProber + .Verdict(fast, reachable = true, rttOpenMs = -1, rttEoseMs = 300, error = null) + .toDiscoveryEventTemplate() + + // 0 = "reachable, latency not observed": the flag form, never an invented number. + assertTrue(listOf("rtt-open", "0") in tagsOf(template)) + } + + @Test + fun observedAuthWallBecomesARequirementTag() { + val template = + RelayProber + .Verdict(walled, reachable = true, rttOpenMs = 90, rttEoseMs = -1, error = "closed:auth-required: sign in") + .toDiscoveryEventTemplate() + + assertTrue(listOf("R", "auth") in tagsOf(template)) + } + + @Test + fun policyClosedIsNotAnAuthRequirement() { + val template = + RelayProber + .Verdict(walled, reachable = true, rttOpenMs = 90, rttEoseMs = -1, error = "closed:blocked: not welcome") + .toDiscoveryEventTemplate() + + assertNull(tagsOf(template).firstOrNull { it[0] == "R" }) + } + + @Test + fun onionRelayIsTaggedTor() { + val onion = RelayUrlNormalizer.normalize("ws://someonionaddressabcdefghijklmnop.onion") + val template = + RelayProber + .Verdict(onion, reachable = true, rttOpenMs = 900, rttEoseMs = -1, error = null) + .toDiscoveryEventTemplate() + + assertTrue(listOf("n", "tor") in tagsOf(template)) + } +} From a7eec1d605728ca931b6e596e6f9832091d4c358 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 14:18:05 +0000 Subject: [PATCH 4/8] =?UTF-8?q?feat(quartz):=20NIP-66=20check=20options=20?= =?UTF-8?q?=E2=80=94=20read-test=20filters,=20write-test=20event,=20cached?= =?UTF-8?q?=20NIP-11=20fetcher?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RelayProber.probe()/probeFlow() take a filters option choosing the check: LIVENESS_FILTERS (default, impossible-id REQ — EOSE proves liveness with no payload) or readTestFilter(limit = 1) — a REQ the relay must actually work for, querying and streaming real events, making Verdict.rttEoseMs a genuine read test. RelayProbeWriteTest.build() creates the write-check event: ephemeral kind 20166 (never stored by compliant relays) carrying a NIP-40 expiration tag 60s out as belt-and-braces for relays that store unknown ephemeral kinds. Publish it under the monitor key, time the OK for rtt-write, map rejection prefixes to R requirement tags — an OK false still proves the write path. Nip11Fetcher is the missing fetch seam for relay information documents, mirroring Nip05Fetcher: the interface lives in commonMain, OkHttpNip11Fetcher (jvmAndroid) does the Accept: application/nostr+json GET, and CachedNip11Fetcher wraps any implementation with a TTL cache — successes trusted for a day, failures remembered for five minutes so a census doesn't hammer hosts that just refused. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A9sSyh1QLJD3PZ18tVPPVK --- .../nip11RelayInfo/CachedNip11Fetcher.kt | 97 ++++++++++++++ .../quartz/nip11RelayInfo/Nip11Fetcher.kt | 44 ++++++ .../reachability/RelayProbeWriteTest.kt | 59 +++++++++ .../reachability/RelayProber.kt | 36 ++++- .../nip11RelayInfo/CachedNip11FetcherTest.kt | 125 ++++++++++++++++++ .../reachability/RelayProberFlowTest.kt | 51 +++++++ .../nip11RelayInfo/OkHttpNip11Fetcher.kt | 60 +++++++++ 7 files changed, 465 insertions(+), 7 deletions(-) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11Fetcher.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11Fetcher.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProbeWriteTest.kt create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11FetcherTest.kt create mode 100644 quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/OkHttpNip11Fetcher.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11Fetcher.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11Fetcher.kt new file mode 100644 index 0000000000..0805118b6b --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11Fetcher.kt @@ -0,0 +1,97 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip11RelayInfo + +import androidx.collection.LruCache +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.CancellationException + +/** + * TTL cache around any [Nip11Fetcher]. NIP-11 documents change rarely, so a + * successful fetch is served from memory for [ttlSeconds]; a FAILED fetch is + * also remembered — for the shorter [errorTtlSeconds] — so a mass census does + * not hammer a host that just refused, while still retrying it soon. + * + * Concurrent first fetches of the same relay are not deduplicated: both hit the + * network and the second result wins the cache slot. That is harmless (the + * document is idempotent) and keeps this class lock-free. + */ +class CachedNip11Fetcher( + private val delegate: Nip11Fetcher, + private val ttlSeconds: Long = DEFAULT_TTL_SECONDS, + private val errorTtlSeconds: Long = DEFAULT_ERROR_TTL_SECONDS, + maxEntries: Int = 1000, + private val now: () -> Long = { TimeUtils.now() }, +) : Nip11Fetcher { + private sealed interface Cached { + val at: Long + } + + private class Hit( + val info: Nip11RelayInformation, + override val at: Long, + ) : Cached + + private class Miss( + val message: String?, + override val at: Long, + ) : Cached + + private val cache = LruCache(maxEntries) + + /** The cached document if present and fresh; null otherwise. Never touches the network. */ + fun cachedOrNull(relay: NormalizedRelayUrl): Nip11RelayInformation? { + val hit = cache[relay] as? Hit ?: return null + return if (now() - hit.at < ttlSeconds) hit.info else null + } + + /** Drops the cache entry (success or failure) so the next [fetch] is fresh. */ + fun invalidate(relay: NormalizedRelayUrl) { + cache.remove(relay) + } + + override suspend fun fetch(relay: NormalizedRelayUrl): Nip11RelayInformation { + when (val cached = cache[relay]) { + is Hit -> if (now() - cached.at < ttlSeconds) return cached.info + is Miss -> + if (now() - cached.at < errorTtlSeconds) { + throw Nip11FetchException(cached.message ?: "cached NIP-11 failure for ${relay.url}") + } + null -> {} + } + return try { + delegate.fetch(relay).also { cache.put(relay, Hit(it, now())) } + } catch (e: Exception) { + if (e is CancellationException) throw e + cache.put(relay, Miss(e.message, now())) + throw e + } + } + + companion object { + /** Documents are near-static: trust a success for a day. */ + const val DEFAULT_TTL_SECONDS = 24L * 60 * 60 + + /** Failures are often transient: retry after five minutes. */ + const val DEFAULT_ERROR_TTL_SECONDS = 5L * 60 + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11Fetcher.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11Fetcher.kt new file mode 100644 index 0000000000..cc97860576 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11Fetcher.kt @@ -0,0 +1,44 @@ +/* + * 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.nip11RelayInfo + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl + +/** + * Fetches and parses a relay's NIP-11 information document (the + * `application/nostr+json` answer on the relay's https url). Mirrors the + * [com.vitorpamplona.quartz.nip05DnsIdentifiers.Nip05Fetcher] seam: the HTTP + * transport lives in a platform implementation (OkHttpNip11Fetcher on + * JVM/Android), so common code — probes, monitors, the CLI — depends only on + * this interface. Wrap any implementation in [CachedNip11Fetcher] to add a TTL + * cache. + * + * Throws [Nip11FetchException] (or a transport exception) when the document is + * unavailable or unparseable. + */ +interface Nip11Fetcher { + suspend fun fetch(relay: NormalizedRelayUrl): Nip11RelayInformation +} + +/** The relay answered, but not with a usable NIP-11 document (bad status, not JSON). */ +class Nip11FetchException( + message: String, +) : Exception(message) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProbeWriteTest.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProbeWriteTest.kt new file mode 100644 index 0000000000..12bc086047 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProbeWriteTest.kt @@ -0,0 +1,59 @@ +/* + * 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.signers.EventTemplate +import com.vitorpamplona.quartz.nip01Core.signers.eventTemplate +import com.vitorpamplona.quartz.nip40Expiration.ExpirationTag +import com.vitorpamplona.quartz.utils.TimeUtils + +/** + * The event a NIP-66 monitor publishes to measure a relay's WRITE path: publish + * one of these (signed with the monitor key), time the `OK`, and read the + * rejection prefix when refused (`auth-required:`/`restricted:`/`pow:` map to + * the discovery record's `R` requirement tags; a timed acceptance is `rtt-write`). + * + * The kind is EPHEMERAL (20000–29999 per NIP-01), so a compliant relay serves it + * to current subscribers and never stores it — the probe leaves nothing behind. + * [KIND] 20166 is this library's convention (30166 discovery minus the + * addressable range), not something NIP-66 standardizes; any ephemeral kind + * works. Belt-and-braces, the template also carries a NIP-40 `expiration` tag + * [EXPIRATION_SECONDS] out, so a relay that stores unknown ephemeral kinds + * anyway purges it promptly. + * + * A rejection is still a MEASUREMENT: an `OK false` proves the write path works + * and documents the relay's policy. Only silence is a failed write test. + */ +object RelayProbeWriteTest { + const val KIND = 20166 + + /** Storage-window ceiling for non-compliant relays that store ephemeral events. */ + const val EXPIRATION_SECONDS = 60L + + fun build( + content: String = "NIP-66 write probe", + createdAt: Long = TimeUtils.now(), + ): EventTemplate = + eventTemplate(KIND, content, createdAt) { + add(ExpirationTag.assemble(createdAt + EXPIRATION_SECONDS)) + } +} 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 index 364f4e1bd0..58663c2728 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt @@ -100,18 +100,24 @@ class RelayProber( /** * Probe every relay in [relays], [waveSize] at a time, giving each wave up to * [timeoutMs] to reach terminals. Returns one [Verdict] per input relay. + * + * [filters] is the REQ each relay is asked to answer. The default + * [LIVENESS_FILTERS] matches nothing, so an EOSE proves liveness without + * streaming a payload; pass [readTestFilter] to make [Verdict.rttEoseMs] a + * real read test instead (the relay must query and stream an actual event). */ suspend fun probe( relays: Collection, timeoutMs: Long = 15_000, waveSize: Int = 1000, + filters: List = LIVENESS_FILTERS, ): 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))) { - probeWave(wave, timeoutMs) { all += it } + probeWave(wave, timeoutMs, filters) { all += it } done += wave.size if (distinct.size > wave.size) { val liveSoFar = all.count { it.reachable } @@ -127,6 +133,9 @@ class RelayProber( * as its EOSE/CLOSED/connect-failure lands, not when the whole census ends. Only * relays that stay silent wait for their wave's [timeoutMs] deadline. * + * [filters] picks the check, as in [probe]: [LIVENESS_FILTERS] (default) or + * [readTestFilter]. + * * Probing starts when the flow is collected and pauses between waves while the * collector is busy (emission is sequential). Pair each verdict with * [toDiscoveryEventTemplate] to turn the stream into signable NIP-66 kind:30166 @@ -136,16 +145,18 @@ class RelayProber( relays: Collection, timeoutMs: Long = 15_000, waveSize: Int = 1000, + filters: List = LIVENESS_FILTERS, ): Flow = flow { for (wave in relays.toSet().chunked(waveSize.coerceAtLeast(1))) { - probeWave(wave, timeoutMs) { emit(it) } + probeWave(wave, timeoutMs, filters) { emit(it) } } } private suspend fun probeWave( wave: List, timeoutMs: Long, + filters: List, onVerdict: suspend (Verdict) -> Unit, ) { val mark = TimeSource.Monotonic.markNow() @@ -230,7 +241,7 @@ class RelayProber( client.addConnectionListener(connListener) try { - client.subscribe(subId, wave.associateWith { PROBE_FILTERS }, subListener) + client.subscribe(subId, wave.associateWith { filters }, subListener) val remaining = wave.toMutableSet() while (remaining.isNotEmpty()) { val left = timeoutMs - mark.elapsedNow().inWholeMilliseconds @@ -251,10 +262,21 @@ class RelayProber( } 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)))) + /** + * 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. This is the default check — pure liveness. + */ + val LIVENESS_FILTERS = listOf(Filter(ids = listOf("0".repeat(64)))) + + /** + * A REQ the relay must actually WORK for: query its store and stream up to + * [limit] real events before the EOSE. Pass to [probe]/[probeFlow] as + * [filters] to turn [Verdict.rttEoseMs] into a genuine read test rather + * than a liveness ping — the time still counts from the wave start (dial + * included), so compare it against [Verdict.rttOpenMs], not across waves. + */ + fun readTestFilter(limit: Int = 1) = listOf(Filter(limit = limit)) /** * The relay universe the local store knows: every read/write relay advertised diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11FetcherTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11FetcherTest.kt new file mode 100644 index 0000000000..95bd844de0 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11FetcherTest.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.nip11RelayInfo + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertNull + +class CachedNip11FetcherTest { + private val relay = RelayUrlNormalizer.normalize("wss://nostr.example.com") + + /** Counts network hits; serves [doc] or throws when [failing]. */ + private class FakeFetcher : Nip11Fetcher { + var calls = 0 + var failing = false + var doc = Nip11RelayInformation(name = "v1") + + override suspend fun fetch(relay: NormalizedRelayUrl): Nip11RelayInformation { + calls++ + if (failing) throw Nip11FetchException("boom") + return doc + } + } + + private fun cached( + delegate: FakeFetcher, + clock: () -> Long, + ) = CachedNip11Fetcher(delegate, ttlSeconds = 100, errorTtlSeconds = 10, now = clock) + + @Test + fun freshSuccessIsServedFromCache() = + runBlocking { + val net = FakeFetcher() + var now = 0L + val fetcher = cached(net) { now } + + assertEquals("v1", fetcher.fetch(relay).name) + now = 99 + assertEquals("v1", fetcher.fetch(relay).name) + assertEquals(1, net.calls, "second fetch inside the TTL must not touch the network") + } + + @Test + fun successExpiresAfterTtl() = + runBlocking { + val net = FakeFetcher() + var now = 0L + val fetcher = cached(net) { now } + + fetcher.fetch(relay) + net.doc = Nip11RelayInformation(name = "v2") + now = 100 + assertEquals("v2", fetcher.fetch(relay).name) + assertEquals(2, net.calls) + } + + @Test + fun failureIsCachedForItsOwnShorterTtl() = + runBlocking { + val net = FakeFetcher().apply { failing = true } + var now = 0L + val fetcher = cached(net) { now } + + assertFailsWith { fetcher.fetch(relay) } + now = 9 + assertFailsWith { fetcher.fetch(relay) } + assertEquals(1, net.calls, "a fresh failure must be served from cache, not re-fetched") + + now = 10 + net.failing = false + assertEquals("v1", fetcher.fetch(relay).name) + assertEquals(2, net.calls, "an expired failure must be retried") + } + + @Test + fun invalidateForcesAFreshFetch() = + runBlocking { + val net = FakeFetcher() + val fetcher = cached(net) { 0 } + + fetcher.fetch(relay) + fetcher.invalidate(relay) + fetcher.fetch(relay) + assertEquals(2, net.calls) + } + + @Test + fun cachedOrNullNeverTouchesTheNetwork() = + runBlocking { + val net = FakeFetcher() + var now = 0L + val fetcher = cached(net) { now } + + assertNull(fetcher.cachedOrNull(relay)) + assertEquals(0, net.calls) + + fetcher.fetch(relay) + assertEquals("v1", fetcher.cachedOrNull(relay)?.name) + now = 100 + assertNull(fetcher.cachedOrNull(relay), "a stale hit must not be served") + assertEquals(1, net.calls) + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt index 34b1ce15fc..2efc801324 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt @@ -49,6 +49,7 @@ class RelayProberFlowTest { /** Captures the probe subscription so the test can play the relays. */ private class ScriptedClient : INostrClient by EmptyNostrClient() { var listener: SubscriptionListener? = null + var sentFilters: Map>? = null override fun subscribe( subId: String, @@ -56,6 +57,7 @@ class RelayProberFlowTest { listener: SubscriptionListener?, ) { this.listener = listener + this.sentFilters = filters } } @@ -139,6 +141,55 @@ class RelayProberFlowTest { assertTrue(at < 1_000, "a failed dial must not wait for the deadline, arrived at ${at}ms") } + // ------------------------------------------------------------------ + // Check options — liveness default, read-test override, write-test event + // ------------------------------------------------------------------ + + @Test + fun livenessFilterIsTheDefaultCheck() = + runTest { + val client = ScriptedClient() + val collector = + launch { + RelayProber(client).probeFlow(listOf(fast), timeoutMs = 1_000).collect {} + } + launch { + delay(10) + assertEquals(RelayProber.LIVENESS_FILTERS, client.sentFilters!![fast]) + client.listener!!.onEose(fast, null) + } + collector.join() + } + + @Test + fun readTestFilterIsSentWhenChosen() = + runTest { + val client = ScriptedClient() + val collector = + launch { + RelayProber(client) + .probeFlow(listOf(fast), timeoutMs = 1_000, filters = RelayProber.readTestFilter()) + .collect {} + } + launch { + delay(10) + val sent = client.sentFilters!![fast]!!.single() + assertEquals(1, sent.limit, "read test defaults to limit 1") + assertNull(sent.ids, "read test must query real events, not the impossible id") + client.listener!!.onEose(fast, null) + } + collector.join() + } + + @Test + fun writeTestEventIsEphemeralAndSelfExpiring() { + val template = RelayProbeWriteTest.build(createdAt = 5000) + + assertEquals(20166, template.kind) + assertTrue(template.kind in 20000..29999, "the write probe must be an ephemeral kind") + assertTrue(listOf("expiration", "5060") in template.tags.map { it.toList() }) + } + // ------------------------------------------------------------------ // toDiscoveryEventTemplate — only observed facts become tags // ------------------------------------------------------------------ diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/OkHttpNip11Fetcher.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/OkHttpNip11Fetcher.kt new file mode 100644 index 0000000000..a28033c0d1 --- /dev/null +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/OkHttpNip11Fetcher.kt @@ -0,0 +1,60 @@ +/* + * 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.nip11RelayInfo + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.toHttp +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.coroutines.executeAsync + +/** + * OkHttp-backed [Nip11Fetcher]: GETs the relay's https url with + * `Accept: application/nostr+json` and parses the document. The client is + * resolved per relay so callers can route Tor/proxy relays through a different + * OkHttp instance (the same seam Amethyst's Nip11Retriever uses). + */ +class OkHttpNip11Fetcher( + private val okHttpClient: (NormalizedRelayUrl) -> OkHttpClient, +) : Nip11Fetcher { + override suspend fun fetch(relay: NormalizedRelayUrl): Nip11RelayInformation = + withContext(Dispatchers.IO) { + val request = + Request + .Builder() + .header("Accept", "application/nostr+json") + .url(relay.toHttp()) + .build() + + okHttpClient(relay).newCall(request).executeAsync().use { response -> + if (!response.isSuccessful) { + throw Nip11FetchException("HTTP ${response.code} fetching NIP-11 from ${relay.url}") + } + val body = response.body.string() + if (!body.startsWith("{")) { + throw Nip11FetchException("Not a NIP-11 document from ${relay.url}") + } + Nip11RelayInformation.fromJson(body) + } + } +} From fd5bd994a1df5e43408423c4979a24a12cc73d8f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 14:41:15 +0000 Subject: [PATCH 5/8] =?UTF-8?q?feat(quartz):=20read+write=20relay=20checks?= =?UTF-8?q?=20=E2=80=94=20observed=20facts=20only,=20no=20NIP-11=20claims?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RelayProber.readWriteCheck(relays, signer) is the deeper check pair for relays already proven live (warm sockets from a probe that just ran): - READ: a real limit-1 REQ the relay must query its store for, timed REQ→first answer (honest rtt-read on an open socket). - WRITE: one ephemeral RelayProbeWriteTest event signed by the monitor key, timed publish→OK (honest rtt-write). An OK false is a measured policy answer, kept with its NIP-01 machine-readable reason; only silence leaves the write side unobserved (writeAccepted = null). publishAndCollectResults now stamps each OK with its elapsedMs (a rejection is still a round trip; -1 when the relay never answered), so any caller gets write latency for free. toDiscoveryEventTemplate(readWrite = ...) folds the pair into the 30166 template: rtt-read/rtt-write when measured, R auth / R pow when the write was refused with auth-required:/pow:. NIP-11-derived tags (N supported NIPs, k kinds, T type) are deliberately NOT emitted — those are relay self-claims, and publishing them under a monitor signature without per-NIP compliance tests would launder claims into measurements. Per-NIP/per-kind compliance suites can come later as opt-in checks; open/read/write is the default surface. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A9sSyh1QLJD3PZ18tVPPVK --- .../accessories/NostrClientPublishExt.kt | 13 +- .../reachability/RelayProber.kt | 100 +++++++++-- .../reachability/RelayProberFlowTest.kt | 164 +++++++++++++++++- 3 files changed, 264 insertions(+), 13 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt index 9cbec380b4..112345d123 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt @@ -34,6 +34,7 @@ import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.withTimeoutOrNull +import kotlin.time.TimeSource /** * One relay's verdict on a published event: [accepted] plus the reason the @@ -45,6 +46,12 @@ import kotlinx.coroutines.withTimeoutOrNull class PublishResult( val accepted: Boolean, val message: String, + /** + * Milliseconds from the publish to this relay's OK (true or false — a rejection + * is still a measured round trip), or -1 when the relay never answered with an + * OK. On an already-open socket this is an honest NIP-66 `rtt-write`. + */ + val elapsedMs: Long = -1, ) { /** * True when this failure came from the transport (never connected, @@ -102,6 +109,7 @@ suspend fun INostrClient.publishAndCollectResults( timeoutInSeconds: Long = 15, ): Map { val resultChannel = Channel(UNLIMITED) + val mark = TimeSource.Monotonic.markNow() Log.d("publishAndConfirm") { "Waiting for ${relayList.size} responses" } @@ -134,7 +142,7 @@ suspend fun INostrClient.publishAndCollectResults( when (msg) { is OkMessage -> { if (msg.eventId == event.id) { - resultChannel.trySend(DetailedResult(relay.url, msg.success, msg.message)) + resultChannel.trySend(DetailedResult(relay.url, msg.success, msg.message, mark.elapsedNow().inWholeMilliseconds)) Log.d("publishAndConfirm") { "onSendResponse Received response for ${msg.eventId} from relay ${relay.url} message ${msg.message} success ${msg.success}" } } } @@ -160,7 +168,7 @@ suspend fun INostrClient.publishAndCollectResults( val currentResult = receivedResults[result.relay] // do not override a successful result. if (currentResult == null || !currentResult.accepted) { - receivedResults[result.relay] = PublishResult(result.success, result.message) + receivedResults[result.relay] = PublishResult(result.success, result.message, result.elapsedMs) } } } @@ -191,4 +199,5 @@ private class DetailedResult( val relay: NormalizedRelayUrl, val success: Boolean, val message: String, + val elapsedMs: Long = -1, ) 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 index 58663c2728..86e57c7e58 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt @@ -22,6 +22,8 @@ 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.accessories.PublishResult +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndCollectResults 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 @@ -30,6 +32,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.signers.EventTemplate +import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.RelayDiscoveryEvent @@ -83,6 +86,19 @@ class RelayProber( val error: String?, ) + /** + * One relay's read+write check outcome (see [readWriteCheck]). Latencies are + * -1 when unobserved; [writeAccepted] is null when the relay never answered + * the write with an OK (transport failure or silence). + */ + class ReadWriteVerdict( + val relay: NormalizedRelayUrl, + val rttReadMs: Long, + val rttWriteMs: Long, + val writeAccepted: Boolean?, + val writeMessage: String?, + ) + class Result( val verdicts: List, val elapsedMs: Long, @@ -153,6 +169,55 @@ class RelayProber( } } + /** + * The deeper, still-honest check pair: READ (a real limit-[readLimit] REQ the + * relay must query its store for) and WRITE (one ephemeral [RelayProbeWriteTest] + * event signed by [signer], the monitor key, timed to its OK). Everything is a + * direct observation — nothing is copied from the relay's NIP-11 self-claims. + * + * Run it against relays ALREADY PROVEN LIVE — typically [Result.reachable] of a + * [probe] that just ran, while the pool's sockets are still open. On a warm + * socket both numbers are honest NIP-66 rtts (`rtt-read`, `rtt-write`); against + * a cold relay they silently include the dial, so don't. + * + * A write REJECTION is still a measurement: `OK false` proves the write path + * works and documents policy ([ReadWriteVerdict.writeMessage] keeps the NIP-01 + * machine-readable reason; [toDiscoveryEventTemplate] maps `auth-required:` and + * `pow:` to `R` tags). Only silence leaves [ReadWriteVerdict.writeAccepted] null. + */ + suspend fun readWriteCheck( + relays: Collection, + signer: NostrSigner, + timeoutMs: Long = 15_000, + waveSize: Int = 1000, + readLimit: Int = 1, + ): Map { + val out = HashMap() + val distinct = relays.toSet() + for (wave in distinct.chunked(waveSize.coerceAtLeast(1))) { + val reads = HashMap() + probeWave(wave, timeoutMs, readTestFilter(readLimit)) { reads[it.relay] = it.rttEoseMs } + + val event = signer.sign(RelayProbeWriteTest.build()) + val writes = client.publishAndCollectResults(event, wave.toSet(), (timeoutMs / 1000).coerceAtLeast(1)) + + for (relay in wave) { + // Only a real OK (true or false) counts as an answer; transport + // failures and silence leave the write side unobserved. + val answered = writes[relay]?.takeUnless { it.isTransportFailure || it.message == PublishResult.NO_RESPONSE } + out[relay] = + ReadWriteVerdict( + relay = relay, + rttReadMs = reads[relay] ?: -1, + rttWriteMs = answered?.elapsedMs ?: -1, + writeAccepted = answered?.accepted, + writeMessage = answered?.message, + ) + } + } + return out + } + private suspend fun probeWave( wave: List, timeoutMs: Long, @@ -329,22 +394,37 @@ class RelayProber( * is the normalized relay url; sign it with the consumer's own monitor key (per * NIP-66 a monitor is its own identity, so the prober never signs on its own). * - * Only facts this probe actually observed are tagged: + * Only facts a probe actually observed are tagged: * - `n` network type inferred from the url (clearnet/tor/i2p); * - `rtt-open` when the relay was reachable — the measured WS-upgrade round trip, * or 0 for "reachable, latency not observed" (liveness is the tag's PRESENCE); - * - `R auth` when the relay answered the probe REQ with a NIP-42 `auth-required` - * CLOSED — an observed auth wall, not a NIP-11 claim. + * - `rtt-read`/`rtt-write` when a [RelayProber.readWriteCheck] result is passed + * as [readWrite] and actually measured that side; + * - `R auth` when the relay answered a probe with a NIP-42 `auth-required` + * (CLOSED on the REQ, or OK-false on the write) and `R pow` when the write + * was refused with a `pow:` reason — observed walls, not NIP-11 claims. * - * [Verdict.rttEoseMs] is deliberately NOT written as `rtt-read`: it is measured from - * the wave start, so it bundles dial, TLS and any handshake queueing with the read — - * publishing it as a read round trip would hand aggregators an inflated latency. - * The [RelayObserver]/[RelayMonitor] path supplies honest `rtt-read`/`rtt-write` - * from real traffic instead. + * NIP-11-derived tags (`N` supported NIPs, `k` kinds, `T` type) are deliberately + * absent: those are the relay's self-claims, and asserting them under a monitor + * signature without a per-NIP compliance test would launder claims into + * measurements. [Verdict.rttEoseMs] is likewise never written as `rtt-read` — it + * is wave-relative (dial + TLS + queueing), so the honest read number only comes + * from [readWrite] (or the [RelayObserver]/[RelayMonitor] real-traffic path). */ -fun RelayProber.Verdict.toDiscoveryEventTemplate(createdAt: Long = TimeUtils.now()): EventTemplate = +fun RelayProber.Verdict.toDiscoveryEventTemplate( + createdAt: Long = TimeUtils.now(), + readWrite: RelayProber.ReadWriteVerdict? = null, +): EventTemplate = RelayDiscoveryEvent.build(relay, createdAt = createdAt) { networkType(RelayReachabilityStore.networkTypeOf(relay)) if (reachable) rtt(RttType.OPEN, rttOpenMs.coerceAtLeast(0)) - if (error?.startsWith("closed:auth-required") == true) requirement("auth") + if (readWrite != null) { + if (readWrite.rttReadMs >= 0) rtt(RttType.READ, readWrite.rttReadMs) + if (readWrite.rttWriteMs >= 0) rtt(RttType.WRITE, readWrite.rttWriteMs) + } + val authWalled = + error?.startsWith("closed:auth-required") == true || + readWrite?.writeMessage?.startsWith("auth-required") == true + if (authWalled) requirement("auth") + if (readWrite?.writeMessage?.startsWith("pow:") == true) requirement("pow") } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt index 2efc801324..0195e52b53 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt @@ -20,13 +20,20 @@ */ package com.vitorpamplona.quartz.nip66RelayMonitor.reachability +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient 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.commands.toClient.OkMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command 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.signers.EventTemplate +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.delay import kotlinx.coroutines.launch @@ -46,10 +53,12 @@ import kotlin.test.assertTrue */ @OptIn(ExperimentalCoroutinesApi::class) class RelayProberFlowTest { - /** Captures the probe subscription so the test can play the relays. */ + /** Captures the probe subscription and publish so the test can play the relays. */ private class ScriptedClient : INostrClient by EmptyNostrClient() { var listener: SubscriptionListener? = null var sentFilters: Map>? = null + var published: Event? = null + val connListeners = mutableListOf() override fun subscribe( subId: String, @@ -59,6 +68,49 @@ class RelayProberFlowTest { this.listener = listener this.sentFilters = filters } + + override fun publish( + event: Event, + relayList: Set, + ) { + published = event + } + + override fun addConnectionListener(listener: RelayConnectionListener) { + connListeners += listener + } + + override fun removeConnectionListener(listener: RelayConnectionListener) { + connListeners -= listener + } + + /** Plays a relay's OK answer for the published event to every armed listener. */ + fun answerOk( + relay: NormalizedRelayUrl, + success: Boolean, + message: String, + ) { + val ok = OkMessage(published!!.id, success, message) + connListeners.toList().forEach { it.onIncomingMessage(FakeRelayClient(relay), "", ok) } + } + } + + private class FakeRelayClient( + override val url: NormalizedRelayUrl, + ) : IRelayClient { + override fun connect() = Unit + + override fun needsToReconnect() = false + + override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) = Unit + + override fun isConnected() = true + + override fun sendOrConnectAndSync(cmd: Command) = Unit + + override fun sendIfConnected(cmd: Command) = Unit + + override fun disconnect() = Unit } private val fast = RelayUrlNormalizer.normalize("wss://fast.example.com") @@ -190,6 +242,82 @@ class RelayProberFlowTest { assertTrue(listOf("expiration", "5060") in template.tags.map { it.toList() }) } + // ------------------------------------------------------------------ + // readWriteCheck — honest read + write measurements, nothing claimed + // ------------------------------------------------------------------ + + private suspend fun ScriptedClient.playReadThenWrite( + relay: NormalizedRelayUrl, + ok: Boolean?, + okMessage: String = "", + ) { + delay(50) + listener!!.onEose(relay, null) // read phase answers + while (published == null) delay(10) // write phase begins + if (ok != null) answerOk(relay, ok, okMessage) + } + + @Test + fun readWriteCheckMeasuresBothSides() = + runTest { + val client = ScriptedClient() + val signer = NostrSignerInternal(KeyPair()) + var result: Map? = null + + val check = + launch { + result = RelayProber(client).readWriteCheck(listOf(fast), signer, timeoutMs = 5_000) + } + launch { client.playReadThenWrite(fast, ok = true) } + check.join() + + val verdict = result!![fast]!! + assertTrue(verdict.rttReadMs >= 0, "an answered read must be measured") + assertTrue(verdict.rttWriteMs >= 0, "an answered write must be measured") + assertEquals(true, verdict.writeAccepted) + assertEquals(20166, client.published!!.kind, "the write test must use the ephemeral probe event") + } + + @Test + fun writeRejectionIsAnAnswerNotAFailure() = + runTest { + val client = ScriptedClient() + val signer = NostrSignerInternal(KeyPair()) + var result: Map? = null + + val check = + launch { + result = RelayProber(client).readWriteCheck(listOf(walled), signer, timeoutMs = 5_000) + } + launch { client.playReadThenWrite(walled, ok = false, okMessage = "pow: 28 bits needed") } + check.join() + + val verdict = result!![walled]!! + assertEquals(false, verdict.writeAccepted, "OK false is a measured policy answer") + assertEquals("pow: 28 bits needed", verdict.writeMessage) + assertTrue(verdict.rttWriteMs >= 0, "a rejection is still a round trip") + } + + @Test + fun silentWriteLeavesTheWriteSideUnobserved() = + runTest { + val client = ScriptedClient() + val signer = NostrSignerInternal(KeyPair()) + var result: Map? = null + + val check = + launch { + result = RelayProber(client).readWriteCheck(listOf(fast), signer, timeoutMs = 2_000) + } + launch { client.playReadThenWrite(fast, ok = null) } + check.join() + + val verdict = result!![fast]!! + assertTrue(verdict.rttReadMs >= 0) + assertNull(verdict.writeAccepted, "silence is not evidence about the write path") + assertEquals(-1, verdict.rttWriteMs) + } + // ------------------------------------------------------------------ // toDiscoveryEventTemplate — only observed facts become tags // ------------------------------------------------------------------ @@ -257,6 +385,40 @@ class RelayProberFlowTest { assertNull(tagsOf(template).firstOrNull { it[0] == "R" }) } + @Test + fun readWriteResultsBecomeRttTags() { + val verdict = RelayProber.Verdict(fast, reachable = true, rttOpenMs = 100, rttEoseMs = 300, error = null) + val readWrite = RelayProber.ReadWriteVerdict(fast, rttReadMs = 40, rttWriteMs = 55, writeAccepted = true, writeMessage = "") + + val tags = tagsOf(verdict.toDiscoveryEventTemplate(readWrite = readWrite)) + assertTrue(listOf("rtt-read", "40") in tags) + assertTrue(listOf("rtt-write", "55") in tags) + } + + @Test + fun unobservedReadWriteSidesStayUntagged() { + val verdict = RelayProber.Verdict(fast, reachable = true, rttOpenMs = 100, rttEoseMs = -1, error = null) + val readWrite = RelayProber.ReadWriteVerdict(fast, rttReadMs = -1, rttWriteMs = -1, writeAccepted = null, writeMessage = null) + + val tags = tagsOf(verdict.toDiscoveryEventTemplate(readWrite = readWrite)) + assertNull(tags.firstOrNull { it[0] == "rtt-read" }) + assertNull(tags.firstOrNull { it[0] == "rtt-write" }) + } + + @Test + fun writeRejectionReasonsBecomeRequirementTags() { + val verdict = RelayProber.Verdict(walled, reachable = true, rttOpenMs = 100, rttEoseMs = -1, error = null) + + val pow = RelayProber.ReadWriteVerdict(walled, -1, 30, writeAccepted = false, writeMessage = "pow: 28 bits needed") + assertTrue(listOf("R", "pow") in tagsOf(verdict.toDiscoveryEventTemplate(readWrite = pow))) + + val auth = RelayProber.ReadWriteVerdict(walled, -1, 30, writeAccepted = false, writeMessage = "auth-required: sign in") + assertTrue(listOf("R", "auth") in tagsOf(verdict.toDiscoveryEventTemplate(readWrite = auth))) + + val blocked = RelayProber.ReadWriteVerdict(walled, -1, 30, writeAccepted = false, writeMessage = "blocked: not welcome") + assertNull(tagsOf(verdict.toDiscoveryEventTemplate(readWrite = blocked)).firstOrNull { it[0] == "R" }) + } + @Test fun onionRelayIsTaggedTor() { val onion = RelayUrlNormalizer.normalize("ws://someonionaddressabcdefghijklmnop.onion") From c767298368a5b8495ca80a201218cbb9462eb39c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 15:14:17 +0000 Subject: [PATCH 6/8] =?UTF-8?q?fix(quartz):=20audit=20fixes=20=E2=80=94=20?= =?UTF-8?q?foreign-OK=20confirmation=20bug,=20normalizer=20hot-path=20allo?= =?UTF-8?q?cation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit findings across the branch, each verified with a failing test or measurement before the fix: - publishAndCollectResults counted an OK from a relay OUTSIDE relayList (same event id — a probe-wave straggler, or any republish of the same event to a different relay set) toward its confirmation window, ending the wait loop early and misreporting still-pending listed relays as NO_RESPONSE. The OK branch now carries the same relayList guard the onCannotConnect/onDisconnected branches always had. Regression test proves the failure without the guard. readWriteCheck additionally varies the probe event content per wave so wave N's confirmation window can never match wave N-1's event id at all. - RelayUrlNormalizer.fix() called trimEnd('%','2','0') unconditionally, allocating a full string copy for ANY url merely ending in '%', '2' or '0' — which includes every relay port ending in zero (wss://host:3030). Now gated on endsWith("%20"), keeping the hot path allocation-free; semantics unchanged (test pins both the trim and the untouched-port cases). - amy relay probe --file: unreadable file is now a clean bad_args error instead of a stack trace, and skipped onion urls are counted and reported (file_onion_skipped) instead of vanishing from the tally. - probeFlow KDoc now states that a slow collector eats into the current wave's absolute deadline (answers are still recorded; silent relays get less listening time), not just that it delays the next wave. Verified non-issues: androidx.collection LruCache is internally locked (safe for CachedNip11Fetcher/normalizer concurrency); probeWave's per-terminal emission cannot lose or double-emit verdicts (remaining-set guard, data maps read at emission time); existing publish callers all benefit from the OK guard rather than depending on the old behavior. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A9sSyh1QLJD3PZ18tVPPVK --- .../amethyst/cli/commands/RelayCommands.kt | 12 +++++-- .../accessories/NostrClientPublishExt.kt | 7 +++- .../relay/normalizer/RelayUrlNormalizer.kt | 11 +++---- .../reachability/RelayProber.kt | 14 +++++--- .../nip01Core/relay/RelayUrlFormatterTest.kt | 8 +++++ .../reachability/RelayProberFlowTest.kt | 32 +++++++++++++++++++ 6 files changed, 70 insertions(+), 14 deletions(-) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt index 8734ba1b02..292a19b6b0 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt @@ -349,21 +349,26 @@ object RelayCommands { var fileRaw = 0 var fileRejected = 0 + var fileOnion = 0 val fileRelays = HashSet() if (fromFile != null) { - File(fromFile).forEachLine { line -> + val candidates = File(fromFile) + if (!candidates.canRead()) return Output.error("bad_args", "cannot read --file $fromFile") + candidates.forEachLine { line -> if (line.isBlank()) return@forEachLine fileRaw++ val normalized = line.normalizeRelayUrlOrNull() if (normalized == null) { fileRejected++ - } else if (!RelayUrlNormalizer.isOnion(normalized.url)) { + } else if (RelayUrlNormalizer.isOnion(normalized.url)) { + fileOnion++ + } else { fileRelays.add(normalized) } } System.err.println( "[relay-probe] $fromFile: $fileRaw urls → ${fileRelays.size} unique clearnet relays " + - "($fileRejected rejected by the normalizer)", + "($fileRejected rejected by the normalizer, $fileOnion onion skipped)", ) } @@ -412,6 +417,7 @@ object RelayCommands { "file_urls" to (if (fromFile != null) fileRaw else null), "file_normalized" to (if (fromFile != null) fileRelays.size else null), "file_rejected" to (if (fromFile != null) fileRejected else null), + "file_onion_skipped" to (if (fromFile != null) fileOnion else null), "reachable" to result.reachable.size, "dead" to result.dead.size, "closed_by_policy" to authWalled, diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt index 112345d123..c1aa8c14a2 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt @@ -141,7 +141,12 @@ suspend fun INostrClient.publishAndCollectResults( when (msg) { is OkMessage -> { - if (msg.eventId == event.id) { + // The relayList guard matters, not just the id: the same event may + // have been published to OTHER relays by an earlier call (probe + // waves, republish), and counting their late OKs here would inflate + // receivedResults and end the wait loop before every listed relay + // answered — misreporting the missing ones as NO_RESPONSE. + if (msg.eventId == event.id && relay.url in relayList) { resultChannel.trySend(DetailedResult(relay.url, msg.success, msg.message, mark.elapsedNow().inWholeMilliseconds)) Log.d("publishAndConfirm") { "onSendResponse Received response for ${msg.eventId} from relay ${relay.url} message ${msg.message} success ${msg.success}" } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt index 8949a0b82c..bc5c524482 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt @@ -180,12 +180,11 @@ class RelayUrlNormalizer { if (rawUrl.length < 4) return null if (rawUrl.contains("%00")) return null - // Trim trailing %20 (percent-encoded spaces from malformed event data) - val url = - rawUrl.trimEnd('%', '2', '0').let { trimmed -> - // Only accept if we actually removed a trailing %20 pattern - if (trimmed.length < rawUrl.length && rawUrl.endsWith("%20")) trimmed else rawUrl - } + // Trim trailing %20 (percent-encoded spaces from malformed event data). + // The endsWith gate keeps the hot path allocation-free: trimEnd would + // copy the string for ANY url merely ending in '%', '2' or '0' — which + // includes every port ending in zero ("wss://host:3030"). + val url = if (rawUrl.endsWith("%20")) rawUrl.trimEnd('%', '2', '0') else rawUrl if (url.length < 4) return null // Reject URLs with %20 in the middle — these are garbage 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 index 86e57c7e58..5dcf878aad 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt @@ -152,8 +152,11 @@ class RelayProber( * [filters] picks the check, as in [probe]: [LIVENESS_FILTERS] (default) or * [readTestFilter]. * - * Probing starts when the flow is collected and pauses between waves while the - * collector is busy (emission is sequential). Pair each verdict with + * Probing starts when the flow is collected, and emission is sequential — a slow + * collector delays the next wave AND eats into the current wave's [timeoutMs] + * window (the deadline is absolute; answers keep being recorded while the + * collector runs, but silent relays get less listening time). Keep per-verdict + * work light, or buffer, when precise deadlines matter. Pair each verdict with * [toDiscoveryEventTemplate] to turn the stream into signable NIP-66 kind:30166 * records for another process to sign and publish. */ @@ -194,11 +197,14 @@ class RelayProber( ): Map { val out = HashMap() val distinct = relays.toSet() - for (wave in distinct.chunked(waveSize.coerceAtLeast(1))) { + for ((waveIndex, wave) in distinct.chunked(waveSize.coerceAtLeast(1)).withIndex()) { val reads = HashMap() probeWave(wave, timeoutMs, readTestFilter(readLimit)) { reads[it.relay] = it.rttEoseMs } - val event = signer.sign(RelayProbeWriteTest.build()) + // A distinct event id per wave (createdAt has second granularity, so the + // content must vary) keeps a straggler OK from an earlier wave's relays + // from ever matching this wave's confirmation window. + val event = signer.sign(RelayProbeWriteTest.build(content = "NIP-66 write probe $waveIndex")) val writes = client.publishAndCollectResults(event, wave.toSet(), (timeoutMs / 1000).coerceAtLeast(1)) for (relay in wave) { diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt index fc708eddba..6357c09591 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt @@ -53,6 +53,14 @@ class RelayUrlFormatterTest { assertNull(RelayUrlNormalizer.normalizeOrNull("wss://relay%20list%20to%20discover%20the%20user's%20content")) } + @Test + fun trailingPercentTwentyIsTrimmedButBareTrailingZeroIsNot() { + assertEquals("wss://nostr.mom/", RelayUrlNormalizer.normalizeOrNull("wss://nostr.mom%20")?.url) + // urls merely ending in '%', '2' or '0' (every port ending in zero) must pass untouched + assertEquals("wss://nostr.mom:3030/", RelayUrlNormalizer.normalizeOrNull("wss://nostr.mom:3030")?.url) + assertEquals("wss://nostr.mom:8020/", RelayUrlNormalizer.normalizeOrNull("wss://nostr.mom:8020")?.url) + } + @Test fun httpWithPathIsNotARelay() { // Mastodon/bridge actor urls from `proxy` tags: web resources, not relays diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt index 0195e52b53..153c6e5d3e 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt @@ -298,6 +298,38 @@ class RelayProberFlowTest { assertTrue(verdict.rttWriteMs >= 0, "a rejection is still a round trip") } + @Test + fun foreignRelayOkDoesNotEndTheWriteConfirmationEarly() = + runTest { + // A relay OUTSIDE the checked set answering with the same event id (a + // straggler from an earlier wave that got the same probe event) must not + // count toward the confirmation window — before the relayList guard in + // publishAndCollectResults, it ended the wait early and misreported the + // real relay as silent. + val client = ScriptedClient() + val signer = NostrSignerInternal(KeyPair()) + val foreign = RelayUrlNormalizer.normalize("wss://foreign.example.com") + var result: Map? = null + + val check = + launch { + result = RelayProber(client).readWriteCheck(listOf(fast), signer, timeoutMs = 5_000) + } + launch { + delay(50) + client.listener!!.onEose(fast, null) + while (client.published == null) delay(10) + client.answerOk(foreign, true, "") + delay(100) + client.answerOk(fast, true, "") + } + check.join() + + val verdict = result!![fast]!! + assertEquals(true, verdict.writeAccepted, "the listed relay's OK must still be awaited and recorded") + assertNull(result!![foreign], "the foreign relay must not appear in the result") + } + @Test fun silentWriteLeavesTheWriteSideUnobserved() = runTest { From 9a4f6c6cdd96500be8b35fe2e2490fe4e2562968 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 15:26:54 +0000 Subject: [PATCH 7/8] feat(quartz): optional kinds on the read-test filter (production finding) Verified the new probe surface end-to-end against production relays (probeFlow streaming, readWriteCheck, signed 30166 templates). One compatibility finding: purpose relays like purplepag.es reject any REQ that names no kind ('blocked: filters must specify at least one kind'), leaving their read side unobserved. readTestFilter/readWriteCheck now take an optional kinds list for those; the default stays kind-less because naming kinds also narrows the query on every other relay. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A9sSyh1QLJD3PZ18tVPPVK --- .../nip66RelayMonitor/reachability/RelayProber.kt | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) 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 index 5dcf878aad..d03fb01208 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt @@ -194,12 +194,13 @@ class RelayProber( timeoutMs: Long = 15_000, waveSize: Int = 1000, readLimit: Int = 1, + readKinds: List? = null, ): Map { val out = HashMap() val distinct = relays.toSet() for ((waveIndex, wave) in distinct.chunked(waveSize.coerceAtLeast(1)).withIndex()) { val reads = HashMap() - probeWave(wave, timeoutMs, readTestFilter(readLimit)) { reads[it.relay] = it.rttEoseMs } + probeWave(wave, timeoutMs, readTestFilter(readLimit, readKinds)) { reads[it.relay] = it.rttEoseMs } // A distinct event id per wave (createdAt has second granularity, so the // content must vary) keeps a straggler OK from an earlier wave's relays @@ -346,8 +347,17 @@ class RelayProber( * [filters] to turn [Verdict.rttEoseMs] into a genuine read test rather * than a liveness ping — the time still counts from the wave start (dial * included), so compare it against [Verdict.rttOpenMs], not across waves. + * + * [kinds] widens compatibility with purpose relays: some (purplepag.es) + * reject any REQ that names no kind with `blocked: filters must specify at + * least one kind`, which leaves their read side unobserved. Passing e.g. + * `listOf(0, 1)` satisfies them; the default stays kind-less because a + * kind list also narrows the query on every OTHER relay. */ - fun readTestFilter(limit: Int = 1) = listOf(Filter(limit = limit)) + fun readTestFilter( + limit: Int = 1, + kinds: List? = null, + ) = listOf(Filter(kinds = kinds, limit = limit)) /** * The relay universe the local store knows: every read/write relay advertised From 5ec35c77726c36cf9ee8581e7bc6e0f684ce4557 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 15:35:22 +0000 Subject: [PATCH 8/8] feat(quartz): default the read test to kind 0, limit 1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A kind-0, limit-1 REQ works everywhere: purpose relays (purplepag.es) reject kind-less filters outright, and practically every relay stores some profile. Verified against production — purplepag.es's read side now measures instead of going unobserved. Pass a different kinds list to probe a specific shelf, or null for a kind-less query on relays known to allow one. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A9sSyh1QLJD3PZ18tVPPVK --- .../nip66RelayMonitor/reachability/RelayProber.kt | 14 +++++++------- .../reachability/RelayProberFlowTest.kt | 1 + 2 files changed, 8 insertions(+), 7 deletions(-) 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 index d03fb01208..48d33d9185 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt @@ -194,7 +194,7 @@ class RelayProber( timeoutMs: Long = 15_000, waveSize: Int = 1000, readLimit: Int = 1, - readKinds: List? = null, + readKinds: List? = listOf(0), ): Map { val out = HashMap() val distinct = relays.toSet() @@ -348,15 +348,15 @@ class RelayProber( * than a liveness ping — the time still counts from the wave start (dial * included), so compare it against [Verdict.rttOpenMs], not across waves. * - * [kinds] widens compatibility with purpose relays: some (purplepag.es) - * reject any REQ that names no kind with `blocked: filters must specify at - * least one kind`, which leaves their read side unobserved. Passing e.g. - * `listOf(0, 1)` satisfies them; the default stays kind-less because a - * kind list also narrows the query on every OTHER relay. + * [kinds] defaults to kind 0: purpose relays (purplepag.es) reject any REQ + * that names no kind with `blocked: filters must specify at least one kind`, + * and practically every relay stores SOME profile — so a kind-0, limit-1 + * query works everywhere. Pass a different list to probe a specific shelf, + * or null for a kind-less query on relays known to allow one. */ fun readTestFilter( limit: Int = 1, - kinds: List? = null, + kinds: List? = listOf(0), ) = listOf(Filter(kinds = kinds, limit = limit)) /** diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt index 153c6e5d3e..8593489ab2 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt @@ -227,6 +227,7 @@ class RelayProberFlowTest { delay(10) val sent = client.sentFilters!![fast]!!.single() assertEquals(1, sent.limit, "read test defaults to limit 1") + assertEquals(listOf(0), sent.kinds, "read test defaults to kind 0 — accepted by purpose relays too") assertNull(sent.ids, "read test must query real events, not the impossible id") client.listener!!.onEose(fast, null) }