diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt index 7dfa5a5333..eb8c220c7d 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt @@ -27,6 +27,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient 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.MachineReadablePrefix import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command @@ -43,7 +44,23 @@ import kotlinx.coroutines.flow.MutableStateFlow * This code also awaits a subscription to come to EOSE since many relays * have through switching subs while they are processing the past. */ -class PoolRequests { +class PoolRequests( + /** + * How many times a relay may CLOSE the *same* filter (across reconnects) before we + * stop replaying it to that relay. Small so a structurally-refused REQ (e.g. a + * search-only relay CLOSING a plain kinds filter) stops fast, but > 1 so a single + * transient close doesn't silence a legitimate subscription. A meaningful filter + * change or a successful REQ resets the count. + */ + private val maxRefusalsBeforeSuppress: Int = 3, + /** + * Relay-wide capability refusals (a search-only relay refusing every feed REQ, a + * write-only relay refusing all REQs). Complements the per-subscription memory: this + * stops a relay being hammered by many *different* subscriptions it can't serve. + * Exposed so the app can subtract [RelayReqRefusals.blockedFlow] from feed read-sets. + */ + val relayRefusals: RelayReqRefusals = RelayReqRefusals(), +) { /** * Desired subs and listeners * @@ -94,9 +111,18 @@ class PoolRequests { * to new ones or disconnect to old ones if they are not needed anymore */ private fun updateRelays() { + // A relay is wanted only if at least one desired filter would actually be sent to it. + // A relay-wide capability block (NO_READS, or SEARCH_ONLY for a non-search filter) + // drops it here so the pool disconnects — closing the idle socket rather than keeping + // it open with every REQ suppressed. This is per-filter/capability, so a search-only + // relay stays connected as long as some sub carries a search filter for it. val myRelays = mutableSetOf() desiredSubs.forEach { sub, perRelayFilters -> - myRelays.addAll(perRelayFilters.keys) + perRelayFilters.forEach { (relay, filters) -> + if (!relayRefusals.shouldSuppress(relay, filters)) { + myRelays.add(relay) + } + } } if (desiredRelays.value != myRelays) { @@ -259,15 +285,24 @@ class PoolRequests { } is ClosedMessage -> { + // Relay-wide capability tracking is independent of any single sub's lock. Run + // it first so decideCommandLocked below already sees the block; if it newly + // blocked the relay, drop it from the desired set so the pool disconnects. + val newlyBlocked = relayRefusals.onRefused(relay.url, msg.message) + var forFilters: List? = null val cmd = relayState.get(msg.subId)?.let { state -> state.withLock { state.onClosed(relay.url) forFilters = state.lastKnownFilterStates(relay.url) + recordRefusalIfStructural(state, relay.url, msg.message, forFilters) decideCommandLocked(state, msg.subId, relay.url) } } + + if (newlyBlocked) updateRelays() + desiredSubListeners.get(msg.subId)?.onClosed( message = msg.message, relay = relay.url, @@ -307,15 +342,25 @@ class PoolRequests { // // This is a fresh-connection sync (onConnected → onConnecting always // cleared the per-relay state first), so every desired filter is - // (re)sent unconditionally: unlike the change-driven path there is no - // in-flight REQ on this brand-new socket to dedupe against. + // (re)sent: unlike the change-driven path there is no in-flight REQ on + // this brand-new socket to dedupe against. The one exception is a filter + // this relay has structurally refused too many times — replaying it on + // every reconnect is the doomed-REQ loop this guard exists to stop. desiredSubs.forEach { subId, perRelayFilters -> val filters = perRelayFilters[relay] if (!filters.isNullOrEmpty()) { - subState(subId).let { state -> - state.withLock { state.onOpenReq(relay, filters) } - } - sync(ReqCmd(subId, filters)) + val send = + subState(subId).let { state -> + state.withLock { + if (isStructurallyRefused(state, relay, filters)) { + false + } else { + state.onOpenReq(relay, filters) + true + } + } + } + if (send) sync(ReqCmd(subId, filters)) } } } @@ -404,6 +449,10 @@ class PoolRequests { val current = state.currentState(relay) if (current == ReqSubStatus.SENT || current == ReqSubStatus.QUERYING_PAST) { null + } else if (isStructurallyRefused(state, relay, newFilters)) { + // The relay keeps CLOSING this exact filter; stop offering it. A + // meaningful filter change makes isStructurallyRefused false again. + null } else { // Pre-mark SENT + filters so a concurrent decider skips. state.onOpenReq(relay, newFilters) @@ -418,6 +467,67 @@ class PoolRequests { } } + /** + * True once [relay] has CLOSED this exact [filters] shape at least + * [maxRefusalsBeforeSuppress] times without a success in between. Filter equality + * reuses [FiltersChanged.needsToResendRequest], so a mere `since` bump still counts + * as the same refused shape while any real change re-enables the REQ. + * + * MUST be called while holding [state]'s lock. + */ + private fun isStructurallyRefused( + state: RequestSubscriptionState, + relay: NormalizedRelayUrl, + filters: List, + ): Boolean { + if (filters.isEmpty()) return false + // Relay-wide capability block (search-only / no-reads) — independent of which sub. + if (relayRefusals.shouldSuppress(relay, filters)) return true + // Per-subscription: this relay keeps CLOSING this exact filter shape. + if (state.refusalCount(relay) < maxRefusalsBeforeSuppress) return false + val refused = state.refusedFilters(relay) ?: return false + return !FiltersChanged.needsToResendRequest(refused, filters) + } + + /** + * Records a CLOSED as a refusal of [forFilters] when its reason is a structural + * "this relay won't serve this REQ" (unprefixed benign closes and the two + * self-resolving prefixes are ignored): + * - `auth-required` — the auth subsystem re-signs and replays; suppressing it would + * block the legitimate post-auth retry. + * - `rate-limited` — the adaptive limiter spaces the REQ out; it's not doomed. + * - no recognized prefix — lifecycle noise (`Subscription closed`, `not found`, …); + * don't risk silencing a live sub on an unstructured message. + * + * MUST be called while holding [state]'s lock. + */ + private fun recordRefusalIfStructural( + state: RequestSubscriptionState, + relay: NormalizedRelayUrl, + reason: String, + forFilters: List?, + ) { + if (forFilters.isNullOrEmpty()) return + val prefix = MachineReadablePrefix.parse(reason) ?: return + val structural = + when (prefix) { + MachineReadablePrefix.AUTH_REQUIRED, + MachineReadablePrefix.RATE_LIMITED, + MachineReadablePrefix.DUPLICATE, + MachineReadablePrefix.POW, + -> false + MachineReadablePrefix.BLOCKED, + MachineReadablePrefix.INVALID, + MachineReadablePrefix.RESTRICTED, + MachineReadablePrefix.ERROR, + MachineReadablePrefix.UNSUPPORTED, + -> true + } + if (!structural) return + val sameAsLast = state.refusedFilters(relay)?.let { !FiltersChanged.needsToResendRequest(it, forFilters) } ?: false + state.recordRefusal(relay, forFilters, sameAsLast) + } + fun destroy() { relayState.clear() desiredSubs.clear() diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayReqRefusals.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayReqRefusals.kt new file mode 100644 index 0000000000..0ec74ff71d --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayReqRefusals.kt @@ -0,0 +1,168 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay.client.pool + +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.update + +/** + * Per-relay REQ suppression for **relay-wide capability refusals** — a relay that + * won't serve a whole *class* of REQs no matter which subscription sends them. + * + * This is the sibling of the per-subscription refusal memory in + * [com.vitorpamplona.quartz.nip01Core.relay.client.reqs.RequestSubscriptionState]: + * that one stops replaying *one refused filter* across reconnects; this one stops a + * relay being hammered by *many different* subscriptions it structurally can't answer. + * The motivating case: a NIP-50 search-only relay pulled into the general read set + * CLOSES every feed REQ with `error: search filter is required` — 13 CLOSEDs across 6 + * subscriptions on a *single* connection, where per-filter memory can't help because + * every filter differs. + * + * Two capability classes are recognized from the CLOSED/NOTICE text (substring match, + * mirroring [com.vitorpamplona.quartz.nip01Core.relay.client.accessories.AdaptiveRelayLimiter]'s + * marker approach — the discriminating detail is in the human message, not the prefix): + * + * - [Policy.SEARCH_ONLY] — the relay requires a NIP-50 `search` field. Only REQs whose + * every filter lacks `search` are suppressed; a genuine search REQ still goes through, + * so this block is safe to keep for the whole session. + * - [Policy.NO_READS] — the relay serves no REQs at all (write-only / "queries not + * allowed"). Every REQ is suppressed. + * + * A relay is only blocked after [threshold] matching refusals, so a single fluke never + * silences it. The block is session-scoped (the lifetime of the owning + * [com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolRequests]); these are stable + * relay properties, and both policies leave the REQs the relay *does* serve untouched, + * so there is deliberately no auto-clear on the per-event hot path. + * + * Fed from the relay socket-reader threads (via CLOSED frames) and read from the send + * threads, so all state is in [ConcurrentMap]s. + */ +class RelayReqRefusals( + private val threshold: Int = 2, +) { + enum class Policy { + /** The relay only answers REQs that carry a NIP-50 `search` term. */ + SEARCH_ONLY, + + /** The relay answers no REQs at all. */ + NO_READS, + } + + // How far each relay has progressed toward a block: the candidate class and its count. + private val progress = ConcurrentMap() + + // Relays that have reached [threshold] refusals of one class; NO_READS wins over SEARCH_ONLY. + private val blocked = ConcurrentMap() + + private val blockedSet = MutableStateFlow>(emptySet()) + + /** + * The relays currently blocked (any class), as a reactive set. The app subtracts this + * from a feed's read-relay set so a proven-useless relay is dropped entirely — closing + * the idle socket — not merely having its REQs suppressed. Monotonic within a session. + */ + val blockedFlow: StateFlow> = blockedSet + + private class Progress( + val candidate: Policy, + val hits: Int, + ) + + /** + * Feed a CLOSED (or NOTICE) reason for [relay]; escalates to a block once the class + * repeats. Returns true only when this call *newly* blocked the relay, so the caller can + * drop it from the desired-relay set (closing the socket) exactly once. + */ + fun onRefused( + relay: NormalizedRelayUrl, + reason: String, + ): Boolean { + val candidate = classify(reason) ?: return false + // NO_READS is the strictest verdict; once reached, nothing softens it. + if (blocked[relay] == Policy.NO_READS) return false + + val next = + progress.merge(relay, Progress(candidate, 1)) { old, _ -> + if (old.candidate == candidate) Progress(candidate, old.hits + 1) else Progress(candidate, 1) + } + if (next.hits >= threshold) { + val wasBlocked = blocked[relay] != null + blocked[relay] = candidate + if (!wasBlocked) { + blockedSet.update { it + relay } + return true + } + } + return false + } + + /** + * True when [relay] has been shown to refuse the class of REQ that [filters] represents. + * [Policy.SEARCH_ONLY] suppresses only when every filter lacks a `search` term. + */ + fun shouldSuppress( + relay: NormalizedRelayUrl, + filters: List, + ): Boolean = + when (blocked[relay]) { + Policy.NO_READS -> true + Policy.SEARCH_ONLY -> filters.isNotEmpty() && filters.all { it.search.isNullOrEmpty() } + null -> false + } + + fun blockedRelays(): Map = blocked.snapshot() + + private fun classify(reason: String): Policy? { + val t = reason.lowercase() + if (SEARCH_REQUIRED_MARKERS.any { it in t }) return Policy.SEARCH_ONLY + if (NO_READ_MARKERS.any { it in t }) return Policy.NO_READS + return null + } + + companion object { + // The relay only serves NIP-50 search REQs (a plain feed REQ is refused). + private val SEARCH_REQUIRED_MARKERS = + listOf( + "search filter is required", + "search filter required", + "requires a search", + "search query is required", + "search term is required", + ) + + // The relay answers no REQs at all (write-only / queries disabled). Kept narrow + // and unconditional so it never catches an auth-gated "authenticate first" message, + // which is resolved by the auth subsystem, not by giving up on the relay. + private val NO_READ_MARKERS = + listOf( + "does not accept req", + "not accepting req", + "queries not allowed", + "queries are not allowed", + "does not allow queries", + "reqs are not allowed", + ) + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RequestSubscriptionState.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RequestSubscriptionState.kt index 33b7238dec..4bfdd6aa2a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RequestSubscriptionState.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RequestSubscriptionState.kt @@ -82,6 +82,48 @@ class RequestSubscriptionState { */ private val lastKnownFilterStates = mutableMapOf>() + /** + * Refused-filter memory. Unlike [subStates]/[filterStates] above — per-connection + * wire state wiped by [connecting]/[disconnected] — this SURVIVES reconnects on + * purpose: a relay that structurally refuses a filter (a search-only relay CLOSING + * a plain kinds REQ, a relay that "does not accept REQs", "too many filters", …) + * refuses it again on every new socket, so [PoolRequests.syncState] replaying it + * each reconnect is pure waste. [refusalCounts] accumulates repeated refusals of + * the same shape so a one-off (transient) close isn't mistaken for a structural + * one. Cleared on a successful REQ ([onEose]/[onNewEvent]) or when the caller + * observes the desired filter meaningfully changed. + */ + private val refusedFilters = mutableMapOf>() + private val refusalCounts = mutableMapOf() + + fun refusedFilters(reference: T) = refusedFilters[reference] + + fun refusalCount(reference: T) = refusalCounts[reference] ?: 0 + + /** + * Records that [reference] refused [filters]. [sameAsLastRefusal] must be true when + * [filters] matches the previously refused shape (the caller owns that comparison, + * keeping filter-equality policy in one place), so repeated refusals accumulate + * instead of resetting. + */ + fun recordRefusal( + reference: T, + filters: List, + sameAsLastRefusal: Boolean, + ) { + if (sameAsLastRefusal) { + refusalCounts[reference] = refusalCount(reference) + 1 + } else { + refusedFilters[reference] = filters + refusalCounts[reference] = 1 + } + } + + fun clearRefusal(reference: T) { + refusedFilters.remove(reference) + refusalCounts.remove(reference) + } + fun currentFilters() = filterStates fun currentFilters(reference: T) = filterStates[reference] @@ -91,12 +133,17 @@ class RequestSubscriptionState { fun currentState(reference: T) = subStates[reference] fun onNewEvent(reference: T) { + // The relay is serving this REQ (it matched an event), so any past refusal + // no longer applies — let it be tried freely again. + clearRefusal(reference) if (subStates[reference] == ReqSubStatus.SENT) { subStates[reference] = ReqSubStatus.QUERYING_PAST } } fun onEose(reference: T) { + // Reaching EOSE means the relay accepted and finished the REQ; clear any refusal. + clearRefusal(reference) subStates[reference] = ReqSubStatus.LIVE } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MachineReadablePrefix.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MachineReadablePrefix.kt index 6d3abf2937..8582f4361f 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MachineReadablePrefix.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MachineReadablePrefix.kt @@ -58,6 +58,9 @@ enum class MachineReadablePrefix( /** A generic, usually transient, server-side failure. */ ERROR("error"), + + /** The relay does not support this request (filter shape, feature, or kind). */ + UNSUPPORTED("unsupported"), ; /** Builds a reason string in the NIP-01 `": "` form. */ diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequestsRefusalTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequestsRefusalTest.kt new file mode 100644 index 0000000000..2e26da853a --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequestsRefusalTest.kt @@ -0,0 +1,197 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay.client.pool + +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * A relay that structurally refuses a REQ (e.g. CLOSING an over-large filter with + * `unsupported: too many filters`) closes the same REQ on every + * reconnect. [RequestSubscriptionState.onClosed] already blocks an *in-connection* + * resend, but a reconnect clears the per-connection wire state and [PoolRequests.syncState] + * replays every desired filter unconditionally — so the client re-sends the identical + * refused REQ once per reconnect, forever (observed 14x in a 40s cold start). + * + * These pin the refusal memory that survives reconnects: after [PoolRequests] sees the + * same filter refused enough times, `syncState` stops replaying it — until the desired + * filter meaningfully changes (which re-enables it), and never for `auth-required` / + * `rate-limited`, which the auth + adaptive-limiter subsystems resolve on their own. + */ +class PoolRequestsRefusalTest { + private val relay = NormalizedRelayUrl("wss://search.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 plainFilter(kind: Int = 1) = listOf(Filter(kinds = listOf(kind), limit = 10)) + + /** One reconnect cycle: the pool clears wire state, then replays desired filters. */ + private fun reconnectAndSync(pool: PoolRequests): List { + pool.onConnecting(relay) + val sent = mutableListOf() + pool.syncState(relay) { sent.add(it) } + return sent + } + + private fun close( + pool: PoolRequests, + subId: String, + reason: String, + ) = pool.onIncomingMessage(FakeRelayClient(relay), ClosedMessage(subId, reason)) + + @Test + fun stopsReplayingAThriceRefusedFilterAcrossReconnects() { + val pool = PoolRequests(maxRefusalsBeforeSuppress = 3) + pool.addOrUpdate("sub", mapOf(relay to plainFilter()), null) + + // Under the threshold, each reconnect still replays the REQ (giving the relay a chance). + repeat(3) { attempt -> + val sent = reconnectAndSync(pool) + assertEquals(1, sent.filterIsInstance().size, "reconnect #$attempt should replay the REQ") + close(pool, "sub", "unsupported: too many filters") + } + + // Once the same filter has been refused [maxRefusalsBeforeSuppress] times, stop replaying it. + val suppressed = reconnectAndSync(pool) + assertTrue(suppressed.filterIsInstance().isEmpty(), "a thrice-refused filter must not be replayed again") + } + + @Test + fun aMeaningfulFilterChangeReEnablesTheReq() { + val pool = PoolRequests(maxRefusalsBeforeSuppress = 2) + pool.addOrUpdate("sub", mapOf(relay to plainFilter(1)), null) + + repeat(2) { + reconnectAndSync(pool) + close(pool, "sub", "unsupported: too many filters") + } + assertTrue(reconnectAndSync(pool).filterIsInstance().isEmpty(), "refused filter is suppressed") + + // The app changes the subscription's filter (different kind) — the relay may now accept it. + pool.addOrUpdate("sub", mapOf(relay to plainFilter(30023)), null) + assertEquals(1, reconnectAndSync(pool).filterIsInstance().size, "a changed filter must be tried again") + } + + @Test + fun aSearchOnlyRelayStopsReceivingPlainReqsFromEveryNewSubOnOneConnection() { + // The search.nos.today case: 6 different subscriptions, one connection, each a + // distinct plain feed filter the relay CLOSES with `error: search filter is required`. + // Per-filter memory can't help (the filters differ); the relay-wide block must. + val pool = PoolRequests(relayRefusals = RelayReqRefusals(threshold = 2)) + + val sent = mutableListOf>() // subId -> did a REQ go out + + fun mountSub( + subId: String, + filter: List, + ) { + val affected = pool.addOrUpdate(subId, mapOf(relay to filter), null) + var reqSent = false + pool.sendToRelayIfChanged(subId, affected) { _, cmd -> if (cmd is ReqCmd) reqSent = true } + sent.add(subId to reqSent) + close(pool, subId, "error: search filter is required") + } + + mountSub("sub1", plainFilter(1)) + mountSub("sub2", plainFilter(2)) + mountSub("sub3", plainFilter(3)) + mountSub("sub4", plainFilter(4)) + + assertTrue(sent[0].second && sent[1].second, "the first two plain subs are sent (learning the relay is search-only)") + assertTrue(!sent[2].second && !sent[3].second, "after two refusals, further plain subs are not sent to a search-only relay") + } + + @Test + fun aCapabilityBlockedRelayIsDroppedFromDesiredRelays() { + // The socket-closing half: once a relay is capability-blocked and no desired sub can + // use it, it leaves the desired-relay set so the pool disconnects it. + val pool = PoolRequests(relayRefusals = RelayReqRefusals(threshold = 2)) + pool.addOrUpdate("sub", mapOf(relay to plainFilter()), null) + assertTrue(relay in pool.desiredRelays.value, "the relay is wanted before it refuses anything") + + close(pool, "sub", "error: search filter is required") + assertTrue(relay in pool.desiredRelays.value, "one refusal doesn't drop it yet") + + close(pool, "sub", "error: search filter is required") + assertTrue(relay !in pool.desiredRelays.value, "a search-only relay with only plain subs is dropped (socket closes)") + } + + @Test + fun aSearchOnlyRelayStaysWantedWhileASearchSubNeedsIt() { + val pool = PoolRequests(relayRefusals = RelayReqRefusals(threshold = 2)) + pool.addOrUpdate("plain", mapOf(relay to plainFilter()), null) + pool.addOrUpdate("search", mapOf(relay to listOf(Filter(kinds = listOf(1), search = "nostr"))), null) + + close(pool, "plain", "error: search filter is required") + close(pool, "plain", "error: search filter is required") + + assertTrue(relay in pool.desiredRelays.value, "the relay stays wanted: a search sub still has a usable filter for it") + } + + @Test + fun authRequiredAndRateLimitedAreNeverSuppressed() { + val pool = PoolRequests(maxRefusalsBeforeSuppress = 2) + pool.addOrUpdate("sub", mapOf(relay to plainFilter()), null) + + repeat(4) { + reconnectAndSync(pool) + close(pool, "sub", MachineReadablePrefix.AUTH_REQUIRED.format("authenticate first")) + } + assertEquals(1, reconnectAndSync(pool).filterIsInstance().size, "auth-required must keep replaying (auth resolves it)") + + val pool2 = PoolRequests(maxRefusalsBeforeSuppress = 2) + pool2.addOrUpdate("sub", mapOf(relay to plainFilter()), null) + repeat(4) { + pool2.onConnecting(relay) + val sent = mutableListOf() + pool2.syncState(relay) { sent.add(it) } + pool2.onIncomingMessage(FakeRelayClient(relay), ClosedMessage("sub", MachineReadablePrefix.RATE_LIMITED.format("slow down"))) + } + pool2.onConnecting(relay) + val sent = mutableListOf() + pool2.syncState(relay) { sent.add(it) } + assertEquals(1, sent.filterIsInstance().size, "rate-limited must keep replaying (the limiter spaces it out)") + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayReqRefusalsTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayReqRefusalsTest.kt new file mode 100644 index 0000000000..bed37f8339 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayReqRefusalsTest.kt @@ -0,0 +1,107 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay.client.pool + +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class RelayReqRefusalsTest { + private val relay = NormalizedRelayUrl("wss://search.nos.today/") + private val other = NormalizedRelayUrl("wss://relay.example/") + + private fun plain() = listOf(Filter(kinds = listOf(1), limit = 10)) + + private fun search() = listOf(Filter(kinds = listOf(1), search = "bitcoin")) + + @Test + fun searchOnlyBlocksPlainReqsButNotSearchReqsAfterThreshold() { + val refusals = RelayReqRefusals(threshold = 2) + + assertFalse(refusals.shouldSuppress(relay, plain()), "no block before any refusal") + + // A search-only relay CLOSES plain feed REQs from different subscriptions. + refusals.onRefused(relay, "error: search filter is required") + assertFalse(refusals.shouldSuppress(relay, plain()), "one refusal is not enough to block") + refusals.onRefused(relay, "error: search filter is required") + + assertTrue(refusals.shouldSuppress(relay, plain()), "a twice-refused search-only relay blocks plain REQs") + assertFalse(refusals.shouldSuppress(relay, search()), "but a genuine search REQ is still allowed through") + assertFalse(refusals.shouldSuppress(other, plain()), "the block is scoped to the offending relay") + assertEquals(mapOf(relay to RelayReqRefusals.Policy.SEARCH_ONLY), refusals.blockedRelays()) + } + + @Test + fun noReadsBlocksEveryReq() { + val refusals = RelayReqRefusals(threshold = 2) + + refusals.onRefused(relay, "restricted: this relay does not accept REQs") + refusals.onRefused(relay, "restricted: this relay does not accept REQs") + + assertTrue(refusals.shouldSuppress(relay, plain()), "a no-reads relay blocks plain REQs") + assertTrue(refusals.shouldSuppress(relay, search()), "a no-reads relay blocks search REQs too") + } + + @Test + fun authRequiredAndUnknownReasonsNeverBlock() { + val refusals = RelayReqRefusals(threshold = 2) + + // Auth gating is resolved by the auth subsystem, not by giving up on the relay. + repeat(5) { refusals.onRefused(relay, "auth-required: this relay only serves private notes to authenticated authors or recipients") } + repeat(5) { refusals.onRefused(relay, "rate-limited: slow down") } + repeat(5) { refusals.onRefused(relay, "Subscription closed") } + + assertFalse(refusals.shouldSuppress(relay, plain()), "auth/rate/lifecycle reasons must never blanket-block a relay") + assertTrue(refusals.blockedRelays().isEmpty()) + } + + @Test + fun blockedFlowEmitsARelayWhenItBecomesBlocked() { + val refusals = RelayReqRefusals(threshold = 2) + assertTrue(refusals.blockedFlow.value.isEmpty(), "nothing blocked initially") + + refusals.onRefused(relay, "error: search filter is required") + assertTrue(refusals.blockedFlow.value.isEmpty(), "one refusal doesn't publish a block") + + refusals.onRefused(relay, "error: search filter is required") + assertEquals(setOf(relay), refusals.blockedFlow.value, "the relay is published to the flow once blocked") + + // Further refusals of the same relay don't churn the flow to a new (equal) set instance. + val snapshot = refusals.blockedFlow.value + refusals.onRefused(relay, "error: search filter is required") + assertTrue(refusals.blockedFlow.value === snapshot, "an already-blocked relay doesn't re-emit") + } + + @Test + fun aFlippingReasonDoesNotAccumulateToABlock() { + val refusals = RelayReqRefusals(threshold = 3) + + // Alternating unrelated capability complaints shouldn't cross the threshold for either class. + refusals.onRefused(relay, "error: search filter is required") + refusals.onRefused(relay, "queries not allowed") + refusals.onRefused(relay, "error: search filter is required") + + assertFalse(refusals.shouldSuppress(relay, plain()), "no single class reached the threshold") + } +}