feat(relay): stop re-sending REQs relays structurally refuse

Relays that can't serve a request (a NIP-50 search-only relay pulled into the feed, a write-only relay, a relay whose filter shape is rejected) were being hammered with the same doomed REQs. Observed on a 40s cold start: search.nos.today CLOSED 13-14x ("error: search filter is required") across 6 subscriptions, plus repeated "restricted: does not accept REQs" and "unsupported: too many filters". The reconnect path replayed refused REQs on every reconnect, and many different subscriptions kept hitting the same capability wall. Two complementary, purely-quartz mechanisms (so the app and Amy both benefit with zero wiring): - Per-subscription refusal memory (RequestSubscriptionState + PoolRequests): a filter a relay CLOSES is not replayed to that relay across reconnects until it meaningfully changes or a REQ succeeds (EOSE/event). Never applies to auth-required (the auth subsystem re-signs and replays) or rate-limited (the adaptive limiter spaces it out). - Per-relay capability block (RelayReqRefusals): after 2 refusals, classify a relay SEARCH_ONLY (suppress only non-search filters; genuine search REQs still flow) or NO_READS (suppress all), from narrow substring markers that deliberately avoid auth-conditional messages. A fully-blocked relay is dropped from PoolRequests.desiredRelays so the pool disconnects it, closing the idle socket rather than keeping it open with every REQ suppressed. A SEARCH_ONLY relay stays connected while any subscription carries a search filter for it, so the separate search path is unaffected. Adds UNSUPPORTED to MachineReadablePrefix (relays send "unsupported:"; parse() returned null on it before). Device-verified: search.nos.today now Connecting -> OnOpen -> 2x Closed -> Disconnected; sendit.nosflare.com (NO_READS) -> 3x Closed -> Disconnected; feed event volume unchanged (kind-1 from 17 relays) - no coverage loss. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-08-12 01:07:46 +00:00 · 2026-07-17 14:58:03 -04:00
parent 3d99f10058
commit 10d88afea3
6 changed files with 640 additions and 8 deletions
@@ -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<NormalizedRelayUrl>()
        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<Filter>? = 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<NormalizedRelayUrl>,
+        relay: NormalizedRelayUrl,
+        filters: List<Filter>,
+    ): 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<NormalizedRelayUrl>,
+        relay: NormalizedRelayUrl,
+        reason: String,
+        forFilters: List<Filter>?,
+    ) {
+        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()
@@ -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<NormalizedRelayUrl, Progress>()
+
+    // Relays that have reached [threshold] refusals of one class; NO_READS wins over SEARCH_ONLY.
+    private val blocked = ConcurrentMap<NormalizedRelayUrl, Policy>()
+
+    private val blockedSet = MutableStateFlow<Set<NormalizedRelayUrl>>(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<Set<NormalizedRelayUrl>> = 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<Filter>,
+    ): Boolean =
+        when (blocked[relay]) {
+            Policy.NO_READS -> true
+            Policy.SEARCH_ONLY -> filters.isNotEmpty() && filters.all { it.search.isNullOrEmpty() }
+            null -> false
+        }
+
+    fun blockedRelays(): Map<NormalizedRelayUrl, Policy> = 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",
+            )
+    }
+}
@@ -82,6 +82,48 @@ class RequestSubscriptionState<T> {
     */
    private val lastKnownFilterStates = mutableMapOf<T, List<Filter>>()

+    /**
+     * 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<T, List<Filter>>()
+    private val refusalCounts = mutableMapOf<T, Int>()
+
+    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<Filter>,
+        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<T> {
    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
    }

@@ -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 `"<code>: <message>"` form. */
@@ -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<Command> {
+        pool.onConnecting(relay)
+        val sent = mutableListOf<Command>()
+        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<ReqCmd>().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<ReqCmd>().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<ReqCmd>().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<ReqCmd>().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<Pair<String, Boolean>>() // subId -> did a REQ go out
+
+        fun mountSub(
+            subId: String,
+            filter: List<Filter>,
+        ) {
+            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<ReqCmd>().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<Command>()
+            pool2.syncState(relay) { sent.add(it) }
+            pool2.onIncomingMessage(FakeRelayClient(relay), ClosedMessage("sub", MachineReadablePrefix.RATE_LIMITED.format("slow down")))
+        }
+        pool2.onConnecting(relay)
+        val sent = mutableListOf<Command>()
+        pool2.syncState(relay) { sent.add(it) }
+        assertEquals(1, sent.filterIsInstance<ReqCmd>().size, "rate-limited must keep replaying (the limiter spaces it out)")
+    }
+}
@@ -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")
+    }
+}