diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/UntilLimitPager.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/UntilLimitPager.kt index 9dce43f344..db33a87959 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/UntilLimitPager.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/UntilLimitPager.kt @@ -145,6 +145,22 @@ class UntilLimitPager { 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, diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt index 97483a755b..378e44fd57 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt @@ -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) -> Unit = {}, ) { private val _loading = MutableStateFlow(true) val loading: StateFlow = _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() + // 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() + 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) { @@ -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 { diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/privateDM/datasource/ChatroomNip04HistorySubAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/privateDM/datasource/ChatroomNip04HistorySubAssembler.kt index 646fbfbad6..981a5b7e03 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/privateDM/datasource/ChatroomNip04HistorySubAssembler.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/privateDM/datasource/ChatroomNip04HistorySubAssembler.kt @@ -69,7 +69,7 @@ class ChatroomNip04HistorySubAssembler( private val started = ConcurrentHashMap.newKeySet() private val askedRelays = ConcurrentHashMap>() - private val windowLoad = WindowLoadTracker("convo.nip04.history") + private val windowLoad = WindowLoadTracker("convo.nip04.history", onAbandoned = ::onRelaysAbandoned) val loadingMore: StateFlow = 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) { + 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, + ) { + windowLoad.onReqSent(relay) + } + override fun onEvent( event: Event, isLive: Boolean, diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/UntilLimitPagerGiveUpTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/UntilLimitPagerGiveUpTest.kt new file mode 100644 index 0000000000..595bc4c2fb --- /dev/null +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/UntilLimitPagerGiveUpTest.kt @@ -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() + + 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() + + // 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() + + 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)) + } +} diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTrackerSilenceTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTrackerSilenceTest.kt new file mode 100644 index 0000000000..b6a699b4f5 --- /dev/null +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTrackerSilenceTest.kt @@ -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>(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() + } +}