diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt index f37c1b5509..c0887ed607 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt @@ -65,6 +65,7 @@ import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation import com.vitorpamplona.quartz.nip17Dm.settings.ChatMessageRelayListEvent import com.vitorpamplona.quartz.nip46RemoteSigner.signer.NostrSignerRemote import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent +import com.vitorpamplona.quartz.nip66RelayMonitor.reachability.RelayObserver import com.vitorpamplona.quartz.nip66RelayMonitor.reachability.RelayReachabilityStore import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.Dispatchers @@ -210,8 +211,12 @@ class Context( * (auth-required / rate-limited / restricted / …), and NIP-42 AUTH * challenges — so a failed REQ can be explained instead of guessed at. * Registered on [client] for the life of this run. + * + * Quartz's [RelayObserver], which also measures the connect/read/write + * round trips behind that feedback and is what a [RelayMonitor] publishes + * as NIP-66. One listener now answers both questions. */ - val relayDiagnostics: RelayDiagnostics = RelayDiagnostics().also { client.addConnectionListener(it) } + val relayDiagnostics: RelayObserver = RelayObserver().also { client.addConnectionListener(it) } /** * Adaptive per-relay concurrent-subscription cap. Starts every relay diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/RelayDiagnostics.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/RelayDiagnostics.kt deleted file mode 100644 index be275a71dd..0000000000 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/RelayDiagnostics.kt +++ /dev/null @@ -1,96 +0,0 @@ -/* - * 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.amethyst.cli - -import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener -import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient -import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage -import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage -import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message -import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage -import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.atomic.AtomicLong - -/** - * Client-wide tally of the relay feedback the crawl would otherwise never see: - * `NOTICE` frames, `CLOSED` reasons (`auth-required` / `rate-limited` / - * `restricted` / …), and NIP-42 `AUTH` challenges. Registered as a - * [RelayConnectionListener] on the shared client, so every incoming message - * during a run is counted and a REQ failure can be explained instead of - * guessed at. - * - * Callbacks fire on the per-relay socket threads, so all state is concurrent. - */ -class RelayDiagnostics : RelayConnectionListener { - private val closedByReason = ConcurrentHashMap() - private val noticeSamples = ConcurrentHashMap() - private val authChallenges = AtomicLong() - - override fun onIncomingMessage( - relay: IRelayClient, - msgStr: String, - msg: Message, - ) { - when (msg) { - // CLOSED reasons follow the NIP-01 machine-readable "word: text" - // convention, so the prefix categorises the failure. - is ClosedMessage -> bump(closedByReason, prefix(msg.message)) - // NOTICE is free-form; keep the (truncated) text so recurring - // relay complaints ("too many concurrent REQs", …) are visible. - is NoticeMessage -> if (noticeSamples.size < MAX_DISTINCT_NOTICES) bump(noticeSamples, msg.message.trim().take(80)) - is AuthMessage -> authChallenges.incrementAndGet() - else -> Unit - } - } - - private fun bump( - map: ConcurrentHashMap, - key: String, - ) { - map.getOrPut(key) { AtomicLong() }.incrementAndGet() - } - - /** The NIP-01 machine-readable prefix (`word` before `:`), or `other`. */ - private fun prefix(message: String): String { - val head = message.substringBefore(':').trim().lowercase() - return head.ifEmpty { "other" }.take(24) - } - - fun hadFeedback(): Boolean = authChallenges.get() > 0 || closedByReason.isNotEmpty() || noticeSamples.isNotEmpty() - - /** JSON-friendly summary for the command output. */ - fun snapshot(): Map = - mapOf( - "auth_challenges" to authChallenges.get(), - "closed_by_reason" to closedByReason.entries.associate { it.key to it.value.get() }.toSortedMap(), - "notices" to noticeSamples.values.sumOf { it.get() }, - "notice_top" to - noticeSamples.entries - .sortedByDescending { it.value.get() } - .take(TOP_NOTICES) - .map { "${it.key} (${it.value.get()})" }, - ) - - companion object { - private const val MAX_DISTINCT_NOTICES = 500 - private const val TOP_NOTICES = 8 - } -} diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankCrawl.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankCrawl.kt index d3316d619d..d34a31f1ee 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankCrawl.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankCrawl.kt @@ -129,7 +129,7 @@ object GrapeRankCrawl { /** Echo any relay NOTICE/CLOSED feedback + adaptive throttling the crawl saw. */ internal fun reportRelayFeedback(ctx: Context) { if (ctx.relayDiagnostics.hadFeedback()) { - System.err.println("[graperank] relay feedback: ${ctx.relayDiagnostics.snapshot()}") + System.err.println("[graperank] relay feedback: ${ctx.relayDiagnostics.summary()}") } if (ctx.relayLimiter.hadThrottling()) { System.err.println("[graperank] relay throttling: ${ctx.relayLimiter.snapshot()}") @@ -204,7 +204,7 @@ object GrapeRankCrawl { "observer" to observer, "crawl_rounds" to stats.rounds, "relays_contacted" to stats.relaysContacted, - "relay_feedback" to if (ctx.relayDiagnostics.hadFeedback()) ctx.relayDiagnostics.snapshot() else null, + "relay_feedback" to if (ctx.relayDiagnostics.hadFeedback()) ctx.relayDiagnostics.summary() else null, "relay_throttling" to if (ctx.relayLimiter.hadThrottling()) ctx.relayLimiter.snapshot() else null, "max_hop_reached" to (stats.hopHistogram.keys.maxOrNull() ?: 0), "users_by_hop" to stats.hopHistogram.mapKeys { it.key.toString() }, diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankScore.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankScore.kt index bcd6a68a1f..4a147dc115 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankScore.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankScore.kt @@ -191,7 +191,7 @@ object GrapeRankScore { "observer" to observer, "crawl_rounds" to (crawlStats?.rounds ?: 0), "relays_contacted" to (crawlStats?.relaysContacted ?: 0), - "relay_feedback" to if (ctx.relayDiagnostics.hadFeedback()) ctx.relayDiagnostics.snapshot() else null, + "relay_feedback" to if (ctx.relayDiagnostics.hadFeedback()) ctx.relayDiagnostics.summary() else null, "relay_throttling" to if (ctx.relayLimiter.hadThrottling()) ctx.relayLimiter.snapshot() else null, "max_hop_reached" to (hopHistogram.keys.maxOrNull() ?: 0), "users_by_hop" to hopHistogram.mapKeys { it.key.toString() }, diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayMonitor.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayMonitor.kt new file mode 100644 index 0000000000..a1e8a7bfcb --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayMonitor.kt @@ -0,0 +1,161 @@ +/* + * 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.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner +import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlin.concurrent.Volatile + +/** + * NIP-66 relay monitoring for a client that was going to talk to relays anyway. + * + * Construct one, and from then on every connection the client makes is measured + * ([RelayObserver]), signed and stored as a kind:30166 ([RelayReachabilityStore]) + * on an interval, and folded back into a cheap [isKnownDead] the caller consults + * when it picks relays. There is nothing else to wire. + * + * ## Reading is the cheap side, and it has to be + * + * A relay picker runs per event — thousands of times a second in an outbox + * fan-out — so [isKnownDead] answers from an in-memory snapshot refreshed on the + * same interval as the writes, never from a store query. The store round trip + * happens [refreshIntervalMs] apart, not per routing decision. + * + * ## The signer is required + * + * Measuring relay quality and letting others check it IS NIP-66; a monitor that + * cannot sign is not a monitor. Making the signer optional would also create the + * failure this library keeps trying to design out — a component configured, + * silent, and doing nothing. A client that should not publish simply does not + * construct one of these, which is a decision visible where it is made. + * + * Note that a monitor is its own identity: per NIP-66 it has its own pubkey, + * profile and relay list, distinct from any user account the client also holds. + * + * ## What ends up in the record + * + * Only what was observed — connect, read and write round trips, whether the + * relay actually demanded AUTH, and the network type implied by the url. Nothing + * is copied out of a relay's NIP-11 document: that is the relay's own claim + * about itself, available to anyone who asks, and re-publishing it under a + * monitor's signature would add nothing but an opportunity to go stale. + */ +class RelayMonitor( + private val client: INostrClient, + store: IEventStore, + private val scope: CoroutineScope, + signer: NostrSigner, + ttlSeconds: Long = RelayReachabilityStore.DEFAULT_TTL_SECONDS, + private val flushIntervalMs: Long = DEFAULT_FLUSH_INTERVAL_MS, + private val refreshIntervalMs: Long = DEFAULT_REFRESH_INTERVAL_MS, + private val onError: (String) -> Unit = {}, +) : AutoCloseable { + val observer = RelayObserver() + + private val reachability = RelayReachabilityStore(store, signer, ttlSeconds) + + @Volatile private var snapshot: RelayReachabilityStore.Snapshot? = null + + init { + client.addConnectionListener(observer) + scope.launch { flushLoop() } + scope.launch { refreshLoop() } + } + + /** + * Skip this relay? Answers from memory, so it is safe to call per routing + * decision. False until the first [refresh] completes — an unknown relay is + * one to try, never one to shun. + */ + fun isKnownDead(relay: NormalizedRelayUrl): Boolean = snapshot?.isKnownDead(relay) == true + + /** Relays proven unreachable within the TTL and not seen live since. */ + fun deadSet(): Set = snapshot?.dead ?: emptySet() + + /** Relays with a recent successful open, from any monitor whose records we hold. */ + fun liveSet(): Set = snapshot?.live ?: emptySet() + + /** Re-read the reachability records, including any other monitor's that arrived. */ + suspend fun refresh() { + runCatching { snapshot = reachability.snapshot() } + .onFailure { onError("could not read relay reachability: ${it.message}") } + } + + /** + * Sign and store what has been observed since the last flush. Returns how + * many records were written. + * + * A relay whose state has not changed is skipped: re-writing its record + * would refresh a freshness window that nothing re-measured. + */ + suspend fun flush(): Int { + val fresh = observer.collectUnreported() + if (fresh.isEmpty()) return 0 + return runCatching { reachability.record(fresh, TimeUtils.now()) } + .onFailure { onError("could not write relay reachability: ${it.message}") } + .getOrDefault(0) + } + + private suspend fun flushLoop() { + while (scope.isActive) { + delay(flushIntervalMs) + flush() + } + } + + private suspend fun refreshLoop() { + // Immediately, then on the interval: the first thing a run should know is + // what the last one learned, before it dials anything. + refresh() + while (scope.isActive) { + delay(refreshIntervalMs) + refresh() + } + } + + /** + * Detach and stop measuring. Does NOT flush — the last write needs a + * coroutine and a bound on how long a shutdown may block, both of which + * belong to the caller. Call [flush] inside your own timeout first. + */ + override fun close() { + runCatching { client.removeConnectionListener(observer) } + } + + companion object { + /** + * Five minutes: long enough that a flapping relay does not mint a record + * per flap, short enough that a crash loses little. The records are + * replaceable, so writing again costs one document, not one more. + */ + const val DEFAULT_FLUSH_INTERVAL_MS = 5 * 60 * 1000L + + /** How often the in-memory dead/live view is re-read from the store. */ + const val DEFAULT_REFRESH_INTERVAL_MS = 5 * 60 * 1000L + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayObserver.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayObserver.kt new file mode 100644 index 0000000000..05f869ce32 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayObserver.kt @@ -0,0 +1,313 @@ +/* + * 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.listeners.RelayConnectionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap +import kotlin.concurrent.Volatile +import kotlin.time.TimeSource + +/** + * What a client learns about relays just by talking to them. + * + * A NIP-66 monitor normally probes: it opens connections for the sole purpose of + * measuring, and then throws them away. A client that is already subscribing, + * fetching and publishing has better data available for free — measured under + * real load, against the relays it actually uses, at the concurrency it actually + * runs. Attached to a client as a [RelayConnectionListener], this collects it. + * + * Everything here is **observed**. Nothing is copied from a relay's NIP-11 + * document, and that is deliberate: a relay's self-description is available to + * anyone who asks for it, so republishing it under a monitor's signature adds + * nothing but a chance to be stale. Where the two disagree — a relay that + * advertises open reads and then sends AUTH — the observation is the half worth + * having, and copying the claim would erase it. + * + * ## Threading + * + * Callbacks for one relay arrive on that relay's own socket thread, so a single + * [Observation] is only ever written by one thread. The fields are `@Volatile` + * for visibility to the reader that publishes them, not for mutual exclusion, + * and the counters need no atomics. The client-wide tallies ARE shared and use + * [ConcurrentMap.merge]. + * + * ## Why nothing is removed + * + * Observations are marked reported rather than deleted. A long-lived connection + * only fires `onConnected` once, so a measurement that vanished when it was + * published would leave the relays we know best — an upstream whose socket has + * been open for hours — with nothing to say about them ever again. The map is + * bounded by the number of distinct relays the client has ever dialled. + */ +class RelayObserver : RelayConnectionListener { + class Observation( + val url: NormalizedRelayUrl, + ) { + // Monotonic marks, not wall clock: these measure durations, and a clock + // step mid-connection must not produce a negative or wild latency. + @Volatile var connectingAt: TimeSource.Monotonic.ValueTimeMark? = null + + @Volatile var rttOpenMs: Long? = null + + @Volatile var firstReqAt: TimeSource.Monotonic.ValueTimeMark? = null + + @Volatile var rttReadMs: Long? = null + + @Volatile var firstEventAt: TimeSource.Monotonic.ValueTimeMark? = null + + @Volatile var rttWriteMs: Long? = null + + /** It opened, or served something. Nothing more is claimed by this. */ + @Volatile var reachable: Boolean = false + + /** Why it did not open, verbatim from the transport. */ + @Volatile var error: String? = null + + /** It sent AUTH, or CLOSED a subscription demanding it. Measured, not read off NIP-11. */ + @Volatile var authRequired: Boolean = false + + /** The NIP-01 machine-readable prefix of the last CLOSED. */ + @Volatile var closedReason: String? = null + + /** The last NOTICE text, truncated — often the only explanation a relay gives. */ + @Volatile var notice: String? = null + + /** Set by every observation, cleared when published. See the class doc. */ + @Volatile var unreported: Boolean = false + + internal fun touch() { + unreported = true + } + } + + private val seen = ConcurrentMap() + + // Client-wide tallies, across every relay. Separate from the per-relay state + // because they answer a different question — "how did this run go" rather + // than "what shall I record about this relay" — and because a summary must + // survive publishing, which clears the per-relay flags. + private val closedByReason = ConcurrentMap() + private val noticeSamples = ConcurrentMap() + private val authChallenges = ConcurrentMap() + + private fun of(relay: IRelayClient) = seen.getOrPut(relay.url) { Observation(relay.url) } + + override fun onConnecting(relay: IRelayClient) { + val o = of(relay) + o.connectingAt = TimeSource.Monotonic.markNow() + // Cleared, not kept: a reconnect is a fresh attempt, and carrying an old + // error forward would report a working relay as broken for as long as the + // process lives after one bad minute. + o.error = null + o.touch() + } + + override fun onConnected( + relay: IRelayClient, + pingMillis: Int, + compressed: Boolean, + ) { + val o = of(relay) + o.reachable = true + o.error = null + o.connectingAt?.let { o.rttOpenMs = it.elapsedNow().inWholeMilliseconds.coerceAtLeast(0) } + o.touch() + } + + override fun onCannotConnect( + relay: IRelayClient, + errorMessage: String, + ) { + val o = of(relay) + // NOT `reachable = false`. A relay that answered an hour ago and is down + // now is a different thing from one that never answered at all, and only + // the writer decides which record that becomes. + o.error = errorMessage.take(MAX_TEXT) + o.touch() + } + + /** + * Outgoing commands start the read and write clocks — the FIRST of each per + * relay, since a later REQ on a warm socket measures nothing about the relay. + */ + override fun onSent( + relay: IRelayClient, + cmdStr: String, + cmd: Command, + success: Boolean, + ) { + if (!success) return + val o = of(relay) + when (cmd) { + is ReqCmd -> if (o.firstReqAt == null) o.firstReqAt = TimeSource.Monotonic.markNow() + is EventCmd -> if (o.firstEventAt == null) o.firstEventAt = TimeSource.Monotonic.markNow() + else -> Unit + } + } + + override fun onIncomingMessage( + relay: IRelayClient, + msgStr: String, + msg: Message, + ) { + val o = of(relay) + when (msg) { + is EoseMessage -> { + if (o.rttReadMs == null) { + o.firstReqAt?.let { + o.rttReadMs = it.elapsedNow().inWholeMilliseconds.coerceAtLeast(0) + o.touch() + } + } + } + + is OkMessage -> { + if (o.rttWriteMs == null) { + o.firstEventAt?.let { + o.rttWriteMs = it.elapsedNow().inWholeMilliseconds.coerceAtLeast(0) + o.touch() + } + } + } + + // Serving an event is proof of life even from a relay that never + // sends EOSE — some do not, and treating those as unresponsive would + // shed relays that work perfectly well. Guarded because this fires + // for EVERY event on every socket: an unconditional write here would + // bounce a cache line between threads to say nothing new. + is EventMessage -> { + if (!o.reachable) { + o.reachable = true + o.touch() + } + } + + is AuthMessage -> { + o.authRequired = true + o.touch() + authChallenges.merge(relay.url.url, 1L) { a, b -> a + b } + } + + is NoticeMessage -> { + val text = msg.message.trim().take(NOTICE_KEY) + o.notice = text + o.touch() + if (noticeSamples.size() < MAX_DISTINCT_NOTICES) noticeSamples.merge(text, 1L) { a, b -> a + b } + } + + is ClosedMessage -> { + val reason = prefixOf(msg.message) + o.closedReason = reason + // NIP-42 refusal, in the shape relays use when the subscription + // is what got rejected rather than the connection. + if (reason == AUTH_REQUIRED) o.authRequired = true + o.touch() + closedByReason.merge(reason, 1L) { a, b -> a + b } + } + + else -> Unit + } + } + + /** + * Everything observed since the last call, marked reported as it is read. + * + * A relay whose state has not changed is left out: writing its record again + * would refresh a freshness window that nothing re-measured. + */ + fun collectUnreported(): List = + seen + .snapshot() + .values + .filter { it.unreported } + .onEach { it.unreported = false } + + /** Every relay ever observed, whether or not it has changed. */ + fun all(): Collection = seen.snapshot().values + + fun observationOf(relay: NormalizedRelayUrl): Observation? = seen[relay] + + fun hadFeedback(): Boolean = authChallenges.size() > 0 || closedByReason.size() > 0 || noticeSamples.size() > 0 + + /** + * A run-level summary of the feedback relays gave — the frames a client + * otherwise never surfaces, so a failed REQ can be explained instead of + * guessed at. + */ + fun summary(): Map { + val notices = noticeSamples.snapshot() + return mapOf( + "auth_challenges" to authChallenges.snapshot().values.sum(), + "auth_required_relays" to authChallenges.size(), + // Sorted into a LinkedHashMap rather than toSortedMap(): that one is + // java.util and this file is commonMain, so it built on JVM and broke + // the native targets. + "closed_by_reason" to + closedByReason + .snapshot() + .entries + .sortedBy { it.key } + .associate { it.key to it.value }, + "notices" to notices.values.sum(), + "notice_top" to + notices.entries + .sortedByDescending { it.value } + .take(TOP_NOTICES) + .map { "${it.key} (${it.value})" }, + ) + } + + companion object { + private const val MAX_TEXT = 200 + private const val NOTICE_KEY = 80 + private const val MAX_DISTINCT_NOTICES = 500 + private const val TOP_NOTICES = 8 + private const val AUTH_REQUIRED = "auth-required" + private const val OTHER = "other" + private const val MAX_PREFIX = 24 + + /** + * The NIP-01 machine-readable prefix — the word before `:` — or `other`. + * + * The colon is required. `substringBefore` returns the WHOLE string when + * the separator is absent, so without this check a relay's free-form + * CLOSED prose became its own tally key and the map's cardinality grew + * with the number of distinct sentences relays happened to write. + */ + fun prefixOf(message: String): String { + if (!message.contains(':')) return OTHER + val head = message.substringBefore(':').trim().lowercase() + return head.ifEmpty { OTHER }.take(MAX_PREFIX) + } + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayReachabilityStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayReachabilityStore.kt index 3d3631a5f0..fe28f94954 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayReachabilityStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayReachabilityStore.kt @@ -27,6 +27,7 @@ import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner import com.vitorpamplona.quartz.nip01Core.store.IEventStore 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.NetworkType import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.tags.RttType @@ -157,6 +158,53 @@ class RelayReachabilityStore( for (relay in dead) if (relay !in reachableRttMs) writeOne(relay, up = false, now, 0) } + /** + * Write everything a run observed, one replaceable record per relay. + * + * Skips a relay it learned nothing about — one that was never dialled, or + * only started connecting. Silence is not evidence, and a record written on + * no observation would refresh a freshness window nothing re-measured. + * + * Returns the number of records written. + */ + suspend fun record( + observations: Collection, + now: Long = TimeUtils.now(), + ): Int { + var written = 0 + for (o in observations) { + if (!o.reachable && o.error == null) continue + writeObserved(o, now) + written++ + } + return written + } + + private suspend fun writeObserved( + o: RelayObserver.Observation, + now: Long, + ) { + val template = + RelayDiscoveryEvent.build(o.url, createdAt = now) { + networkType(networkTypeOf(o.url)) + if (o.reachable) { + // Liveness is the presence of rtt-open, per NIP-66. A relay we + // reached without timing the open — served us an event on a + // socket that was already up — still gets the tag so it reads + // as live, but never an invented latency: 0 would be a lie + // aggregators rank on. + o.rttOpenMs?.let { rtt(RttType.OPEN, it) } ?: rtt(RttType.OPEN, 0) + o.rttReadMs?.let { rtt(RttType.READ, it) } + o.rttWriteMs?.let { rtt(RttType.WRITE, it) } + } + // Observed, not read off NIP-11: this relay actually challenged + // us. A relay advertising open reads and then demanding AUTH is + // exactly what a monitor exists to catch. + if (o.authRequired) requirement("auth") + } + store.insert(signer.sign(template)) + } + private suspend fun writeOne( relay: NormalizedRelayUrl, up: Boolean, diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayObserverTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayObserverTest.kt new file mode 100644 index 0000000000..f1192871f7 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayObserverTest.kt @@ -0,0 +1,255 @@ +/* + * 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.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * These records get published under a monitor's own key, so what matters is what + * the observer is willing to CLAIM: an unmeasured latency must never be reported + * as a measurement, a relay nobody dialled must never be reported at all, and one + * bad minute must not bury a relay that works. + */ +class RelayObserverTest { + private val url = RelayUrlNormalizer.normalize("wss://relay.example") + private val other = RelayUrlNormalizer.normalize("wss://other.example") + + 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 fun client(u: NormalizedRelayUrl) = FakeRelayClient(u) + + private fun RelayObserver.only() = collectUnreported().single() + + // ---- what we measured --------------------------------------------------- + + @Test + fun `an opened connection is timed rather than assumed`() { + val o = RelayObserver() + o.onConnecting(client(url)) + o.onConnected(client(url), 1, true) + + val obs = o.only() + assertTrue(obs.reachable) + assertNotNull(obs.rttOpenMs, "rtt-open must be measured — aggregators rank on it") + assertNull(obs.error) + } + + @Test + fun `the read clock runs from the first REQ to the first EOSE`() { + val o = RelayObserver() + o.onConnecting(client(url)) + o.onConnected(client(url), 1, true) + o.onSent(client(url), "", ReqCmd("sub", emptyList()), true) + o.onIncomingMessage(client(url), "", EoseMessage("sub")) + + assertNotNull(o.only().rttReadMs) + } + + @Test + fun `the write clock runs from the first EVENT to its OK`() { + val o = RelayObserver() + o.onConnecting(client(url)) + o.onConnected(client(url), 1, true) + o.onIncomingMessage(client(url), "", OkMessage("id", true, "")) + + assertNull(o.collectUnreported().single().rttWriteMs, "an OK with nothing sent behind it times nothing") + } + + @Test + fun `a non-REQ command does not start the read clock`() { + val o = RelayObserver() + o.onConnecting(client(url)) + o.onConnected(client(url), 1, true) + o.onSent(client(url), "", CloseCmd("sub"), true) + o.onIncomingMessage(client(url), "", EoseMessage("sub")) + + assertNull(o.only().rttReadMs) + } + + // ---- what we refuse to claim -------------------------------------------- + + @Test + fun `a connection that never opened records the reason and no latency`() { + val o = RelayObserver() + o.onConnecting(client(url)) + o.onCannotConnect(client(url), "Expected HTTP 101 response but was '503 Service Unavailable'") + + val obs = o.only() + assertFalse(obs.reachable) + assertNull(obs.rttOpenMs, "nothing opened, so there is nothing to time") + assertTrue(obs.error!!.contains("503")) + } + + @Test + fun `a relay that answered stays answered through a later failure`() { + // A relay that worked a minute ago and blipped now is not the same thing + // as one that never answered, and only the writer decides which record + // that becomes. A single failure must not erase the success under it. + val o = RelayObserver() + o.onConnecting(client(url)) + o.onConnected(client(url), 1, true) + o.onCannotConnect(client(url), "connection reset") + + assertTrue(o.only().reachable, "one bad minute must not bury a relay that answered") + } + + @Test + fun `a reconnect clears the previous attempt's error`() { + val o = RelayObserver() + o.onConnecting(client(url)) + o.onCannotConnect(client(url), "timeout") + o.onConnecting(client(url)) + + assertNull(o.only().error, "a stale error would report a live relay as broken forever") + } + + // ---- AUTH, which is why an anonymous crawl finds a relay empty ------------ + + @Test + fun `a demand for AUTH is recorded from either shape`() { + val challenged = RelayObserver() + challenged.onIncomingMessage(client(url), "", AuthMessage("challenge")) + assertTrue(challenged.only().authRequired) + + val closed = RelayObserver() + closed.onIncomingMessage(client(url), "", ClosedMessage("sub", "auth-required: subscribers only")) + val obs = closed.only() + assertTrue(obs.authRequired) + assertEquals("auth-required", obs.closedReason) + } + + @Test + fun `a CLOSED that is not about auth is categorised rather than misread`() { + val o = RelayObserver() + o.onIncomingMessage(client(url), "", ClosedMessage("sub", "rate-limited: slow down")) + + val obs = o.only() + assertEquals("rate-limited", obs.closedReason) + assertFalse(obs.authRequired, "only an auth refusal means auth is required") + } + + // ---- publishing bookkeeping --------------------------------------------- + + @Test + fun `an unchanged relay is not re-reported but its measurement survives`() { + // Re-writing a record refreshes its freshness window, so a relay nobody + // re-measured must be left out. But the measurement itself has to stay: + // a long-lived socket fires onConnected once, and if publishing erased + // it, the relays we know best would be the ones we could never describe + // again. + val o = RelayObserver() + o.onConnecting(client(url)) + o.onConnected(client(url), 1, true) + + val first = o.collectUnreported().single() + assertNotNull(first.rttOpenMs) + assertEquals(0, o.collectUnreported().size, "nothing new to say") + + o.onIncomingMessage(client(url), "", NoticeMessage("slow down")) + val second = o.collectUnreported().single() + assertEquals(first.rttOpenMs, second.rttOpenMs, "the last real measurement still stands") + } + + @Test + fun `each relay is observed on its own`() { + val o = RelayObserver() + o.onConnecting(client(url)) + o.onConnected(client(url), 1, true) + o.onConnecting(client(other)) + o.onCannotConnect(client(other), "nodename nor servname provided") + + val byUrl = o.collectUnreported().associateBy { it.url } + assertTrue(byUrl.getValue(url).reachable) + assertFalse(byUrl.getValue(other).reachable) + } + + // ---- the run-level summary (what RelayDiagnostics used to give) ----------- + + @Test + fun `the summary tallies feedback across every relay`() { + val o = RelayObserver() + assertFalse(o.hadFeedback()) + + o.onIncomingMessage(client(url), "", AuthMessage("c1")) + o.onIncomingMessage(client(other), "", AuthMessage("c2")) + o.onIncomingMessage(client(url), "", ClosedMessage("s", "rate-limited: slow")) + o.onIncomingMessage(client(other), "", ClosedMessage("s", "rate-limited: slow")) + o.onIncomingMessage(client(url), "", NoticeMessage("too many REQs")) + + assertTrue(o.hadFeedback()) + val s = o.summary() + assertEquals(2L, s["auth_challenges"]) + assertEquals(2, s["auth_required_relays"]) + assertEquals(mapOf("rate-limited" to 2L), s["closed_by_reason"]) + assertEquals(1L, s["notices"]) + } + + @Test + fun `the summary outlives publishing`() { + // It answers "how did this run go", which must not be reset by the + // unrelated act of writing records out. + val o = RelayObserver() + o.onIncomingMessage(client(url), "", AuthMessage("c")) + o.collectUnreported() + + assertTrue(o.hadFeedback(), "a flush must not erase the run's tally") + assertEquals(1L, o.summary()["auth_challenges"]) + } + + @Test + fun `a machine-readable prefix is extracted or falls back to other`() { + assertEquals("auth-required", RelayObserver.prefixOf("auth-required: come back signed")) + assertEquals("other", RelayObserver.prefixOf("just some prose")) + assertEquals("other", RelayObserver.prefixOf("")) + } +}