fix: give up on DM relays that accept a REQ but never answer

The convo NIP-04 history spinner stayed up for minutes against 9 relays even when the conversation was fully loaded. Auth-walled relays (ditto, nostr.wine, …) accept the REQ (success=true) and then send nothing — no event, no EOSE, no CLOSED. WindowLoadTracker only completed once every relay reached a terminal signal, an idle gate that required hearing from *all* relays, or the 5-minute cap; a silent relay armed none of those, so only the cap freed the spinner. The pager likewise kept the silent relay 'active' every round, so the count never dropped and exhaustion never completed. Add a silence backstop keyed off onSubscriptionStarted (REQ delivered, post-connect — so a slow connect isn't mistaken for a dead relay): a relay that received its REQ but stays silent past silenceTimeout (10s) no longer blocks completion and is reported via onAbandoned, which the convo assembler uses to giveUp() the relay in its pager so it leaves the active set and lets exhaustion finish. finish() applies the give-up before flipping loading, so the round collector recomputes exhaustion after the silent relays are dropped. Tests cover the pager give-up/exhaustion and the tracker's silence + connection-gap behavior. https://claude.ai/code/session_01B1fmmmX8JjQWH3amMLdvcW
2026-08-12 01:07:46 +00:00 · 2026-06-02 22:10:16 +00:00
parent 0430fd176c
commit 3cb6613cd7
5 changed files with 256 additions and 12 deletions
@@ -145,6 +145,22 @@ class UntilLimitPager<K> {
        return false
    }

+    /**
+     * Abandons [relay] for [key] when it accepted our REQ but never answered (no event, EOSE, or CLOSED
+     * within the silence window). Like [givenUp] via [onClosed], it is excluded from [activeRelays] so a
+     * silent relay can't block exhaustion forever — but a relay that already finished cleanly ([done])
+     * is left alone. Returns true if this abandoned a relay that wasn't already done/given-up.
+     */
+    fun giveUp(
+        key: K,
+        relay: NormalizedRelayUrl,
+    ): Boolean {
+        val c = cursor(key, relay)
+        if (c.done || c.givenUp) return false
+        c.givenUp = true
+        return true
+    }
+
    /** Total events received across [relays] in the round just finished. Zero ⇒ nothing more is reachable. */
    fun roundEventCount(
        key: K,
@@ -58,16 +58,25 @@ import kotlin.time.Duration.Companion.seconds
 * mistakes a half-loaded window for a finished one, which is exactly how a load reports "1 event"
 * when a hundred are still on the way.
 *
- * Two backstops cover misbehaving relays. If every relay has at least been *heard from* (any event,
- * EOSE, CLOSED, or cannot-connect) but one streamed events without ever sending EOSE, an [idleTimeout]
- * of quiet completes the load — the "heard from all" gate is what keeps this from firing in a
- * connection gap. And an [absoluteCap] bounds a relay that connects and then dribbles or hangs forever.
+ * Three backstops cover misbehaving relays. If every relay we're still waiting on has at least been
+ * *heard from* (any event, EOSE, CLOSED, or cannot-connect) but one streamed events without ever
+ * sending EOSE, an [idleTimeout] of quiet completes the load — the "heard from" gate is what keeps
+ * this from firing in a connection gap. A relay that *received our REQ* ([onReqSent]) but then went
+ * completely silent — no event, no EOSE, no CLOSED — for [silenceTimeout] is given up on: an
+ * auth-walled relay (ditto, paid relays) commonly accepts the REQ and answers nothing, and measuring
+ * from REQ-delivery (not window start) means a slow connect doesn't count against it. Such relays are
+ * reported to [onAbandoned] so the owner can drop them from its pager too. And an [absoluteCap] bounds
+ * a relay that hangs in the connect itself, before any REQ is even sent.
 */
 class WindowLoadTracker(
    // Short label for the DMPagination logs (e.g. "giftwrap", "rooms.nip04", "convo.nip04").
    private val name: String = "dm",
    private val idleTimeout: Duration = 3.seconds,
+    private val silenceTimeout: Duration = 10.seconds,
    private val absoluteCap: Duration = 5.minutes,
+    // Invoked with the relays that received a REQ but stayed silent past [silenceTimeout] when a load
+    // finishes — the owner gives up on them in its pager so they stop blocking future rounds.
+    private val onAbandoned: (Set<NormalizedRelayUrl>) -> Unit = {},
 ) {
    private val _loading = MutableStateFlow(true)
    val loading: StateFlow<Boolean> = _loading.asStateFlow()
@@ -85,6 +94,10 @@ class WindowLoadTracker(
    // [expected] the stored backfill is complete on every relay and the load is done.
    private val settled = ConcurrentHashMap.newKeySet<NormalizedRelayUrl>()

+    // When the REQ was actually delivered to each relay (post-connect). The silence backstop measures
+    // from here, not window start, so a slow connect isn't mistaken for a dead relay.
+    private val reqSentAt = ConcurrentHashMap<NormalizedRelayUrl, Long>()
+
    private var watchdog: Job? = null

    // Incremented on every (re)start so a stale watchdog that wakes right as a new load begins
@@ -103,6 +116,7 @@ class WindowLoadTracker(
        expected = emptySet()
        heardFrom.clear()
        settled.clear()
+        reqSentAt.clear()
        lastActivityMs = System.currentTimeMillis()
        val wasLoading = _loading.value
        _loading.value = true
@@ -128,11 +142,21 @@ class WindowLoadTracker(
        deadline: Long,
    ): Boolean {
        if (gen != generation || !_loading.value) return false
-        // Every relay has spoken and the stream has gone quiet: a relay that streamed without ever
-        // EOSE'ing is done. The "heard from all" gate keeps this from firing in a connection gap.
-        if (expected.isNotEmpty() && heardFrom.containsAll(expected) && now - lastActivityMs >= idleTimeout.inWholeMilliseconds) {
-            finish("idle")
-            return false
+        if (expected.isNotEmpty()) {
+            // A relay is accounted for once it reached a terminal signal, or it received our REQ and
+            // then stayed completely silent past [silenceTimeout] (an auth-walled / dead relay). Once
+            // every relay is accounted for, nothing more is coming.
+            if (expected.all { settled.contains(it) || silencedOut(it, now) }) {
+                finish("settled/silent")
+                return false
+            }
+            // Idle backstop: every relay we're still waiting on has at least streamed something (so this
+            // isn't a connection gap) and the stream has gone quiet. Silenced/settled relays don't count.
+            val stillWaiting = expected.filterNot { settled.contains(it) || silencedOut(it, now) }
+            if (stillWaiting.all { heardFrom.contains(it) } && now - lastActivityMs >= idleTimeout.inWholeMilliseconds) {
+                finish("idle")
+                return false
+            }
        }
        if (now >= deadline) {
            finish("cap")
@@ -141,6 +165,14 @@ class WindowLoadTracker(
        return true
    }

+    // A relay that received its REQ but produced no signal at all for [silenceTimeout]. Measured from
+    // REQ-delivery so a slow connect (or a still-connecting relay, which has no [reqSentAt]) is never
+    // counted as silent.
+    private fun silencedOut(
+        relay: NormalizedRelayUrl,
+        now: Long,
+    ): Boolean = relay !in heardFrom && (reqSentAt[relay]?.let { now - it >= silenceTimeout.inWholeMilliseconds } ?: false)
+
    /** Records which relays the current REQ was sent to. Completes immediately if there are none. */
    @Synchronized
    fun setExpectedRelays(relays: Set<NormalizedRelayUrl>) {
@@ -152,6 +184,16 @@ class WindowLoadTracker(
        }
    }

+    /**
+     * Records that the REQ was delivered to [relayUrl] (post-connect). Starts that relay's silence clock.
+     * Ignored for relays outside the current [expected] set (or before it is known).
+     */
+    @Synchronized
+    fun onReqSent(relayUrl: String) {
+        val relay = expected.firstOrNull { it.url == relayUrl } ?: return
+        reqSentAt.putIfAbsent(relay, System.currentTimeMillis())
+    }
+
    /** A non-terminal sign of life from [relay] (a stored or live event). Keeps the idle timer alive. */
    fun onRelayEvent(relay: NormalizedRelayUrl) {
        heardFrom.add(relay)
@@ -174,10 +216,14 @@ class WindowLoadTracker(
    @Synchronized
    private fun finish(reason: String) {
        if (!_loading.value) return
-        _loading.value = false
        watchdog?.cancel()
        watchdog = null
-        Log.d(TAG) { "[$name] load done: $reason" }
+        // Give up the silent relays BEFORE flipping [loading]: the owner's round collector reacts to
+        // loading=false by recomputing exhaustion from its pager, so the give-up has to land first.
+        val abandoned = expected.filterTo(mutableSetOf()) { silencedOut(it, System.currentTimeMillis()) }
+        Log.d(TAG) { "[$name] load done: $reason" + if (abandoned.isEmpty()) "" else " (gave up on silent ${abandoned.map { it.url }})" }
+        if (abandoned.isNotEmpty()) onAbandoned(abandoned)
+        _loading.value = false
    }

    companion object {
@@ -69,7 +69,7 @@ class ChatroomNip04HistorySubAssembler(
    private val started = ConcurrentHashMap.newKeySet<ConvoKey>()
    private val askedRelays = ConcurrentHashMap<ConvoKey, Set<NormalizedRelayUrl>>()

-    private val windowLoad = WindowLoadTracker("convo.nip04.history")
+    private val windowLoad = WindowLoadTracker("convo.nip04.history", onAbandoned = ::onRelaysAbandoned)
    val loadingMore: StateFlow<Boolean> = windowLoad.loading

    private val _exhausted = MutableStateFlow(false)
@@ -233,6 +233,18 @@ class ChatroomNip04HistorySubAssembler(
        return requestNewSubscription(historyListener(key))
    }

+    // A relay accepted the REQ but never answered (auth-walled / dead): drop it from every open
+    // conversation's pager so it stops blocking the relay count and exhaustion on the next round. May
+    // complete exhaustion right away if it was the last relay still holding a thread open.
+    private fun onRelaysAbandoned(relays: Set<NormalizedRelayUrl>) {
+        var gaveUp = false
+        started.forEach { pk ->
+            val asked = askedRelays[pk] ?: return@forEach
+            relays.forEach { if (it in asked && pager.giveUp(pk, it)) gaveUp = true }
+        }
+        if (gaveUp) markExhaustedIfAllDone()
+    }
+
    // Flips to exhausted only once every open conversation's relays have all returned an empty page +
    // EOSE. Sets true only — false transitions belong to loadMore / the round collector.
    private fun markExhaustedIfAllDone() {
@@ -252,6 +264,13 @@ class ChatroomNip04HistorySubAssembler(
    private fun historyListener(key: ChatroomQueryState): SubscriptionListener {
        val pk = convoKey(key)
        return object : SubscriptionListener {
+            override fun onSubscriptionStarted(
+                relay: String,
+                forFilters: List<Filter>,
+            ) {
+                windowLoad.onReqSent(relay)
+            }
+
            override fun onEvent(
                event: Event,
                isLive: Boolean,
@@ -0,0 +1,69 @@
+/*
+ * 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.service.relayClient.eoseManagers
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class UntilLimitPagerGiveUpTest {
+    private val mine = NormalizedRelayUrl("wss://vitor.nostr1.com/")
+    private val silent = NormalizedRelayUrl("wss://relay.ditto.pub/")
+    private val all = listOf(mine, silent)
+
+    @Test
+    fun givenUpRelayLeavesTheActiveSet() {
+        val pager = UntilLimitPager<String>()
+
+        assertEquals(all, pager.activeRelays("k", all))
+
+        assertTrue("first give-up takes effect", pager.giveUp("k", silent))
+        assertEquals(listOf(mine), pager.activeRelays("k", all))
+
+        assertFalse("giving up twice is a no-op", pager.giveUp("k", silent))
+    }
+
+    @Test
+    fun givingUpEveryRelayExhaustsTheKey() {
+        val pager = UntilLimitPager<String>()
+
+        // mine pages to empty cleanly; the silent relay never answers and is given up.
+        pager.beginRound("k", all)
+        pager.onEose("k", mine) // empty page + EOSE => done
+        pager.giveUp("k", silent)
+
+        assertTrue("no relay left to ask", pager.activeRelays("k", all).isEmpty())
+    }
+
+    @Test
+    fun aRelayThatAlreadyFinishedIsNotMarkedGivenUp() {
+        val pager = UntilLimitPager<String>()
+
+        pager.beginRound("k", listOf(mine))
+        pager.onEose("k", mine) // done
+
+        // A late silence sweep must not "give up" a relay that already finished cleanly.
+        assertFalse(pager.giveUp("k", mine))
+        assertTrue(pager.isDone("k", mine))
+    }
+}
@@ -0,0 +1,94 @@
+/*
+ * 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.service.relayClient.eoseManagers
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import java.util.concurrent.atomic.AtomicReference
+import kotlin.time.Duration.Companion.milliseconds
+
+/**
+ * Real-time (not virtual-time) tests: the tracker's watchdog reads the wall clock, so a short
+ * [silenceTimeout] with real delays is the honest way to exercise the silence backstop.
+ */
+class WindowLoadTrackerSilenceTest {
+    private val good = NormalizedRelayUrl("wss://vitor.nostr1.com/")
+    private val silent = NormalizedRelayUrl("wss://relay.ditto.pub/")
+
+    @Test
+    fun silentRelayDoesNotBlockTheLoadAndIsReportedAsAbandoned() =
+        runBlocking {
+            val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+            val abandoned = AtomicReference<Set<NormalizedRelayUrl>>(emptySet())
+            val tracker =
+                WindowLoadTracker(
+                    name = "test",
+                    silenceTimeout = 50.milliseconds,
+                    onAbandoned = { abandoned.set(it) },
+                )
+
+            tracker.startLoading(scope)
+            tracker.setExpectedRelays(setOf(good, silent))
+            // Both received the REQ; only the good relay answers (an EOSE settles it).
+            tracker.onReqSent(good.url)
+            tracker.onReqSent(silent.url)
+            tracker.onRelaySettled(good)
+
+            // The good relay is settled and the silent one trips the silence backstop, so the load
+            // completes without ever hearing from the silent relay.
+            withTimeout(3000) { tracker.loading.first { !it } }
+
+            assertEquals(setOf(silent), abandoned.get())
+            scope.cancel()
+        }
+
+    @Test
+    fun aSilentRelayThatNeverGotAReqStillBlocksUntilItSettles() =
+        runBlocking {
+            val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+            val tracker = WindowLoadTracker(name = "test", silenceTimeout = 50.milliseconds)
+
+            tracker.startLoading(scope)
+            tracker.setExpectedRelays(setOf(good, silent))
+            tracker.onReqSent(good.url)
+            tracker.onRelaySettled(good)
+            // `silent` is still connecting: no onReqSent, so the silence clock never starts and the
+            // load must stay open (a connection gap must not be mistaken for a dead relay).
+
+            Thread.sleep(400) // well past silenceTimeout
+            assertTrue("still loading while a relay has not even been sent its REQ", tracker.loading.value)
+
+            // Once it connects, gets its REQ, and stays silent, the backstop then completes the load.
+            tracker.onReqSent(silent.url)
+            withTimeout(3000) { tracker.loading.first { !it } }
+
+            scope.cancel()
+        }
+}