mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-08 23:54:39 +00:00
Merge pull request #3836 from vitorpamplona/feat/nip66-relay-monitor
NIP-66: measure relays from the traffic a client already makes
This commit is contained in:
Vitor Pamplona
@@ -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
|
||||
|
||||
@@ -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<String, AtomicLong>()
|
||||
private val noticeSamples = ConcurrentHashMap<String, AtomicLong>()
|
||||
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<String, AtomicLong>,
|
||||
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<String, Any?> =
|
||||
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
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -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() },
|
||||
|
||||
+1
-1
@@ -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() },
|
||||
|
||||
+161
@@ -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<NormalizedRelayUrl> = snapshot?.dead ?: emptySet()
|
||||
|
||||
/** Relays with a recent successful open, from any monitor whose records we hold. */
|
||||
fun liveSet(): Set<NormalizedRelayUrl> = 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
|
||||
}
|
||||
}
|
||||
+313
@@ -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<NormalizedRelayUrl, Observation>()
|
||||
|
||||
// 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<String, Long>()
|
||||
private val noticeSamples = ConcurrentMap<String, Long>()
|
||||
private val authChallenges = ConcurrentMap<String, Long>()
|
||||
|
||||
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<Observation> =
|
||||
seen
|
||||
.snapshot()
|
||||
.values
|
||||
.filter { it.unreported }
|
||||
.onEach { it.unreported = false }
|
||||
|
||||
/** Every relay ever observed, whether or not it has changed. */
|
||||
fun all(): Collection<Observation> = 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<String, Any?> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
+48
@@ -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<RelayObserver.Observation>,
|
||||
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,
|
||||
|
||||
+255
@@ -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(""))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user