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..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 @@ -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,41 @@ 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 + var fileOnion = 0 + val fileRelays = HashSet() + if (fromFile != null) { + 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)) { + fileOnion++ + } else { + fileRelays.add(normalized) + } + } + System.err.println( + "[relay-probe] $fromFile: $fileRaw urls → ${fileRelays.size} unique clearnet relays " + + "($fileRejected rejected by the normalizer, $fileOnion onion skipped)", + ) + } + 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 +414,10 @@ 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), + "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 9cbec380b4..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 @@ -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" } @@ -133,8 +141,13 @@ suspend fun INostrClient.publishAndCollectResults( when (msg) { is OkMessage -> { - if (msg.eventId == event.id) { - resultChannel.trySend(DetailedResult(relay.url, msg.success, msg.message)) + // 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}" } } } @@ -160,7 +173,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 +204,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/nip01Core/relay/normalizer/RelayUrlNormalizer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt index 04f8c9cc06..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 @@ -84,17 +84,107 @@ 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 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 @@ -109,17 +199,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 +233,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 +266,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 +301,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/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 a95da3f70a..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 @@ -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 @@ -29,10 +31,20 @@ 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.signers.NostrSigner 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 @@ -74,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, @@ -91,18 +116,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))) { - all += probeWave(wave, timeoutMs) + probeWave(wave, timeoutMs, filters) { all += it } done += wave.size if (distinct.size > wave.size) { val liveSoFar = all.count { it.reachable } @@ -112,10 +143,94 @@ 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. + * + * [filters] picks the check, as in [probe]: [LIVENESS_FILTERS] (default) or + * [readTestFilter]. + * + * 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. + */ + fun probeFlow( + 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, filters) { emit(it) } + } + } + + /** + * 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, + readKinds: List? = listOf(0), + ): 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, 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 + // 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) { + // 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, - ): List { + filters: List, + onVerdict: suspend (Verdict) -> Unit, + ) { val mark = TimeSource.Monotonic.markNow() val waveSet = wave.toHashSet() val openRtt = ConcurrentMap() @@ -178,22 +293,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 +302,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,13 +310,54 @@ class RelayProber( error = error, ) } + + client.addConnectionListener(connListener) + try { + client.subscribe(subId, wave.associateWith { 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 { - // 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. + * + * [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? = listOf(0), + ) = listOf(Filter(kinds = kinds, limit = limit)) /** * The relay universe the local store knows: every read/write relay advertised @@ -263,3 +404,43 @@ 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 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); + * - `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. + * + * 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(), + readWrite: RelayProber.ReadWriteVerdict? = null, +): EventTemplate = + RelayDiscoveryEvent.build(relay, createdAt = createdAt) { + networkType(RelayReachabilityStore.networkTypeOf(relay)) + if (reachable) rtt(RttType.OPEN, rttOpenMs.coerceAtLeast(0)) + 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/nip01Core/relay/RelayUrlFormatterTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt index 82287dcbad..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 @@ -52,4 +52,93 @@ class RelayUrlFormatterTest { fun weirdRelay() { 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 + 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) + } } 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 new file mode 100644 index 0000000000..8593489ab2 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt @@ -0,0 +1,465 @@ +/* + * 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.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 +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 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, + filters: Map>, + listener: SubscriptionListener?, + ) { + 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") + 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") + } + + // ------------------------------------------------------------------ + // 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") + 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) + } + 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() }) + } + + // ------------------------------------------------------------------ + // 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 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 { + 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 + // ------------------------------------------------------------------ + + 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 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") + val template = + RelayProber + .Verdict(onion, reachable = true, rttOpenMs = 900, rttEoseMs = -1, error = null) + .toDiscoveryEventTemplate() + + assertTrue(listOf("n", "tor") in tagsOf(template)) + } +} 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) + } + } +}