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 db33a87959..3eb54055b1 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 @@ -35,9 +35,10 @@ import java.util.concurrent.ConcurrentHashMap * Stop signal (per relay): an **empty page followed by EOSE** ([onEose] with no events) marks that * relay [done][isDone]. A relay that returns anything — even fewer than the requested limit, since a * relay may cap results below what we asked — is *not* done; its cursor advances to one second below - * the oldest event it sent and it is asked again. Globally, the owner decides "exhausted" from - * [roundEventCount]: a whole round that advanced no relay (every relay empty-EOSE'd or only answered - * CLOSED) means nothing more is reachable. + * the oldest event it sent and the owner asks it again (typically right away, per relay). Relays a + * relay can't be read from (CLOSED / unreachable / silent) are the owner's concern, not the cursor's: + * the owner flags them stalled and lets its [WindowLoadTracker] settle the load so they can't block + * exhaustion forever. * * Not internally synchronized: per-relay counters are touched on the relay IO threads (one relay's * callbacks are serialized) and read on the owning scope after the load settles; fields are volatile. @@ -50,14 +51,6 @@ class UntilLimitPager { // Set once the relay answered an empty page with EOSE: there is nothing older on it. @Volatile var done: Boolean = false - // Set once the relay has rejected us [GIVE_UP_AFTER_CLOSES] times in a row without ever - // answering (CLOSED, e.g. "auth-required" for authors we can't authenticate). It is NOT done — - // we just can't read its window — but it must be excluded so it doesn't block exhaustion forever. - @Volatile var givenUp: Boolean = false - - // Consecutive CLOSEDs since this relay last actually answered (event or EOSE). Reset on contact. - @Volatile var closedStreak: Int = 0 - // Per-round tallies, reset by [beginRound]: how many events arrived and the oldest among them. @Volatile var roundCount: Int = 0 @@ -103,7 +96,6 @@ class UntilLimitPager { createdAt: Long, ) { val c = cursor(key, relay) - c.closedStreak = 0 c.roundCount++ if (createdAt < c.roundOldest) c.roundOldest = createdAt } @@ -118,7 +110,6 @@ class UntilLimitPager { relay: NormalizedRelayUrl, ) { val c = cursor(key, relay) - c.closedStreak = 0 if (c.roundCount == 0) { c.done = true } else { @@ -126,59 +117,11 @@ class UntilLimitPager { } } - /** - * Records a CLOSED (rejection) from [relay]. After [GIVE_UP_AFTER_CLOSES] in a row with no answer in - * between — i.e. the relay keeps rejecting us and auth can't fix it — the relay is [given up][givenUp] - * so it stops blocking exhaustion. Returns true if this CLOSED tipped it into given-up. - */ - fun onClosed( - key: K, - relay: NormalizedRelayUrl, - ): Boolean { - val c = cursor(key, relay) - if (c.givenUp || c.done) return false - c.closedStreak++ - if (c.closedStreak >= GIVE_UP_AFTER_CLOSES) { - c.givenUp = true - return true - } - 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, - relays: Collection, - ): Int = relays.sumOf { cursor(key, it).roundCount } - - /** - * Relays from [all] that still have older history to ask for: not yet empty-EOSE'd ([done]) and not - * abandoned as unreadable ([givenUp]). - */ + /** Relays from [all] that still have older history to ask for: not yet empty-EOSE'd ([done]). */ fun activeRelays( key: K, all: Collection, - ): List = - all.filterNot { - val c = cursor(key, it) - c.done || c.givenUp - } + ): List = all.filterNot { cursor(key, it).done } /** * The oldest point reached across [relays] — the minimum cursor (how far back paging has gone). @@ -189,10 +132,4 @@ class UntilLimitPager { relays: Collection, start: Long, ): Long? = relays.takeIf { it.isNotEmpty() }?.minOf { cursor(key, it).until ?: start } - - companion object { - // Consecutive CLOSEDs (with no answer in between) before a relay is abandoned as unreadable. - // Allows for the pool's auth handshake + a retry or two before concluding auth can't succeed. - private const val GIVE_UP_AFTER_CLOSES = 3 - } } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/rooms/datasource/ChatroomListNip04HistorySubAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/rooms/datasource/ChatroomListNip04HistorySubAssembler.kt index 248860f18e..608f90286b 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/rooms/datasource/ChatroomListNip04HistorySubAssembler.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/rooms/datasource/ChatroomListNip04HistorySubAssembler.kt @@ -40,7 +40,6 @@ import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quartz.utils.TimeUtils import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job -import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow @@ -48,11 +47,20 @@ import kotlinx.coroutines.launch import java.util.concurrent.ConcurrentHashMap /** - * Loads older NIP-04 DMs (kind 4) for the rooms list by `until`+`limit` paging, per relay — the same - * gap-proof model as [com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip59GiftWraps.AccountGiftWrapsHistoryEoseManager], - * but for kind 4 (exact timestamps, no margin) across the account's home + DM relays. Idle until - * [loadMore]; a relay is done on an empty page + EOSE; the whole history is [exhausted] once a round - * advances no relay. + * Loads older NIP-04 DMs (kind 4) for the rooms list by `until`+`limit` paging, **per relay, + * independently** — the same model as the per-conversation loader + * ([com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.privateDM.datasource.ChatroomNip04HistorySubAssembler]) + * and the gift-wrap history loader + * ([com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip59GiftWraps.AccountGiftWrapsHistoryEoseManager]), + * but account-wide across the home (outbox, *from me*) + DM (inbox, *to me*) relays. + * + * Idle until [loadMore]. A single [loadMore] kicks off every relay that still has older history, and + * from then on each relay drives its own pages off its own cursor, continuing the instant it EOSEs a + * non-empty page ([onEose] → `pager.beginRound([relay])` + `invalidateFilters`; the subscription layer + * diffs per relay, so re-issuing only re-REQs the relay whose cursor moved). A relay is *done* on an + * empty page + EOSE; one that won't answer (auth CLOSE, unreachable, silent) is marked *stalled* but + * kept open. The whole history is [exhausted] once the window settles — every relay done or stalled — + * via the [WindowLoadTracker]'s silence + connect-grace backstops. */ class ChatroomListNip04HistorySubAssembler( client: INostrClient, @@ -60,25 +68,26 @@ class ChatroomListNip04HistorySubAssembler( ) : PerUserEoseManager(client, allKeys) { private val pager = UntilLimitPager() private val started = ConcurrentHashMap.newKeySet() - private val askedRelays = ConcurrentHashMap>() private val accounts = ConcurrentHashMap() - // Shared across accounts (singleton coordinator): repoint the display flows to the active account - // on switch instead of leaking the previous one's exhausted/mark state. Cursors live in [pager]. + // Relays currently not advancing for a user (auth CLOSE / unreachable / silent). Tracked only for the + // logs; these relays are NOT given up — they keep their subscription and keep trying to catch up. + private val stalledRelays = ConcurrentHashMap>() + + // Shared across accounts (singleton coordinator): repoint the display flows to the active account on + // switch instead of leaking the previous one's state. Cursors live in [pager]. @Volatile private var activeUser: HexKey? = null private val exhaustedByUser = ConcurrentHashMap() - // No-progress guard: skip re-issuing an identical round that brought nothing (see the gift-wrap - // history manager's twin); cleared by onEose. - @Volatile - private var lastAskedActive: Set = emptySet() + private val windowLoad = WindowLoadTracker("rooms.nip04.history", tracksReqSends = true, onAbandoned = ::onRelaysStalled) - @Volatile - private var lastRoundEventCount = -1 - - private val windowLoad = WindowLoadTracker("rooms.nip04.history") - val loadingMore: StateFlow = windowLoad.loading + // Exposed instead of windowLoad.loading directly: that flow starts `true` (it assumes a load is in + // flight from construction). Wired straight through, its `true` would wedge the scroll-driven loader + // — whose gate is `!loading` — so the first loadMore could never fire. This starts false and only + // goes true once paging actually begins (mirrored from windowLoad by the done collector). + private val _loadingMore = MutableStateFlow(false) + val loadingMore: StateFlow = _loadingMore.asStateFlow() private val _exhausted = MutableStateFlow(false) val exhausted: StateFlow = _exhausted.asStateFlow() @@ -93,21 +102,30 @@ class ChatroomListNip04HistorySubAssembler( private var scope: CoroutineScope? = null @Volatile - private var roundJob: Job? = null + private var doneJob: Job? = null - // Backoff retry after a no-progress, not-exhausted round (relays failed to answer cleanly rather - // than empty-EOSE'ing), so a transient connect-storm failure recovers instead of stalling forever. + // The user whose window is in flight, read by the done collector when it settles. @Volatile - private var retryJob: Job? = null + private var windowUser: User? = null + // Whether a paging window is currently running. Tracked ourselves rather than read from + // windowLoad.loading (which starts `true` before any window exists), so the first loadMore actually + // starts the window instead of mistaking the construction-time `true` for an in-flight one. @Volatile - private var lastRoundUser: User? = null + private var windowActive = false + // The history floor (live-tail boundary) pinned for the current window. startUntil() is `now − 1w`, + // which drifts forward in real time — if it were recomputed per assembly, an un-advanced relay's + // filter (until = floor) would change every time ANY relay's EOSE triggers invalidateFilters, + // re-REQing relays that haven't moved. Pinning it per window keeps those filters stable so only a + // relay whose cursor genuinely advanced is re-REQed. @Volatile - private var autoLoadAll = false + private var windowFloor = 0L private fun startUntil() = TimeUtils.now() - AccountGiftWrapsEoseManager.LIVE_TAIL_SECONDS + private fun floor() = windowFloor.takeIf { it != 0L } ?: startUntil() + override fun user(key: ChatroomListState) = key.account.userProfile() override fun updateFilter( @@ -115,104 +133,80 @@ class ChatroomListNip04HistorySubAssembler( since: SincePerRelayMap?, ): List? { val user = user(key) - if (!key.account.isWriteable() || user.pubkeyHex !in started) { - windowLoad.setExpectedRelays(emptySet()) - return emptyList() - } + if (!key.account.isWriteable() || user.pubkeyHex !in started) return emptyList() + + // Every relay that still has older history to ask for, each at its own cursor. A relay whose + // cursor advanced since the last assembly re-REQs its next page; one still mid-page keeps its open + // REQ; a done relay drops out (its REQ closes). This is what lets relays run independently. val homeRelays = key.account.homeRelays.flow.value val dmRelays = key.account.dmRelays.flow.value val active = pager.activeRelays(user.pubkeyHex, (homeRelays + dmRelays).toSet()).toSet() - askedRelays[user.pubkeyHex] = active - windowLoad.setExpectedRelays(active) if (active.isEmpty()) return emptyList() DmRelayLog.log("rooms.nip04.history", key.account) Log.d("DMPagination") { "[rooms.nip04.history] REQ ${active.size} relay(s), limit=$PAGE_LIMIT fromMe(outbox)=${homeRelays.filter { it in active }.map { it.url }} toMe(inbox)=${dmRelays.filter { it in active }.map { it.url }}" } return homeRelays.filter { it in active }.map { - filterNip04DMsFromMe(user, it, since = null, until = pager.untilFor(user.pubkeyHex, it, startUntil()), limit = PAGE_LIMIT) + filterNip04DMsFromMe(user, it, since = null, until = pager.untilFor(user.pubkeyHex, it, floor()), limit = PAGE_LIMIT) } + dmRelays.filter { it in active }.map { - filterNip04DMsToMe(user, it, since = null, until = pager.untilFor(user.pubkeyHex, it, startUntil()), limit = PAGE_LIMIT) + filterNip04DMsToMe(user, it, since = null, until = pager.untilFor(user.pubkeyHex, it, floor()), limit = PAGE_LIMIT) } } - /** Requests the next backward page from every relay that still has older NIP-04 history. */ + /** Starts (or resumes) per-relay paging of the NIP-04 history. Idempotent: safe to call again. */ fun loadMore(user: User) { - if (_exhausted.value) return val account = accounts[user.pubkeyHex] ?: return started.add(user.pubkeyHex) val all = (account.homeRelays.flow.value + account.dmRelays.flow.value).toSet() - val active = pager.activeRelays(user.pubkeyHex, all) - if (active.isEmpty()) { + if (all.isEmpty()) return + if (pager.activeRelays(user.pubkeyHex, all).isEmpty()) { + // Everything already paged to the bottom. exhaustedByUser[user.pubkeyHex] = true _exhausted.value = true return } - val activeSet = active.toSet() - if (activeSet == lastAskedActive && lastRoundEventCount == 0) { - Log.d("DMPagination") { "[rooms.nip04.history] loadMore skipped — no progress on the same relays" } - return - } - lastAskedActive = activeSet - pager.beginRound(user.pubkeyHex, active) - lastRoundUser = user - _relayCount.value = active.size - // Over ALL relays (incl. finished ones) so "reached back to X" stays monotonic and doesn't jump - // to a newer date when the deepest relay drops out of the active set. - _reachedBack.value = pager.deepestUntil(user.pubkeyHex, all, startUntil()) - Log.d("DMPagination") { "[rooms.nip04.history] loadMore → ${active.size} active relay(s)" } + _exhausted.value = false + DmRelayLog.log("rooms.nip04.history", account) + windowUser = user scope?.let { - ensureRoundCollector(it) - windowLoad.startLoading(it) + ensureDoneCollector(it) + // One window spanning the whole per-relay pagination: it settles a relay only on that relay's + // empty-EOSE (done) or when it goes silent/stalled, never on a mid-history page, so the + // spinner tracks "is anything still advancing" rather than any single page. Start it only if + // none is running — a re-entrant loadMore (the scroll loader re-firing mid-pagination) must + // not reset the window and forget the relays that already finished. + if (!windowActive) { + windowActive = true + windowFloor = startUntil() + // Populate the relay count BEFORE raising the spinner, so the status card never renders a + // "loading from 0 relays" frame between loadingMore flipping true and the first progress. + updateStatus(user) + _loadingMore.value = true + windowLoad.startLoading(it) + } + windowLoad.setExpectedRelays(all) } + updateStatus(user) + Log.d("DMPagination") { "[rooms.nip04.history] paging ${all.size} relay(s) independently: ${all.map { it.url }}" } invalidateFilters() } - /** Pages to the end: each completed round auto-issues the next until exhausted. */ - fun loadEverything(user: User) { - if (_exhausted.value) return - autoLoadAll = true - loadMore(user) - } - - private fun ensureRoundCollector(scope: CoroutineScope) { - if (roundJob?.isActive == true) return - roundJob = + // Mirrors the window's loading state into [_loadingMore] and, when it settles (every relay done or + // stalled), flips [exhausted] and clears [windowActive] so the next loadMore can start a fresh window. + private fun ensureDoneCollector(scope: CoroutineScope) { + if (doneJob?.isActive == true) return + doneJob = scope.launch { var wasLoading = false windowLoad.loading.collect { loading -> + _loadingMore.value = loading && windowActive if (!loading && wasLoading) { - val user = lastRoundUser - if (user != null) { - val asked = askedRelays[user.pubkeyHex] ?: emptySet() - val count = pager.roundEventCount(user.pubkeyHex, asked) - lastRoundEventCount = count - val account = accounts[user.pubkeyHex] - val allRelays = account?.let { (it.homeRelays.flow.value + it.dmRelays.flow.value).toSet() } ?: emptySet() - // Exhausted ONLY when every relay returned an empty page + EOSE; CLOSED / - // unanswered relays are not finished, so keep loading them. - val exhaustedNow = allRelays.isNotEmpty() && pager.activeRelays(user.pubkeyHex, allRelays).isEmpty() - exhaustedByUser[user.pubkeyHex] = exhaustedNow - _exhausted.value = exhaustedNow - // Over ALL relays (incl. finished ones) so the "reached back" date is monotonic. - _reachedBack.value = pager.deepestUntil(user.pubkeyHex, allRelays, startUntil()) - Log.d("DMPagination") { "[rooms.nip04.history] round done: $count event(s), exhausted=$exhaustedNow" } - if (autoLoadAll && !exhaustedNow) { - loadMore(user) - } else if (!exhaustedNow && count == 0) { - // No progress and not exhausted: relays failed to answer cleanly rather - // than empty-EOSE'ing. Retry after a backoff so a transient failure - // recovers, paced so a rate-limited relay isn't hammered. - retryJob?.cancel() - retryJob = - scope.launch { - delay(NO_PROGRESS_RETRY_MS) - if (!_exhausted.value && !windowLoad.loading.value) { - lastAskedActive = emptySet() - Log.d("DMPagination") { "[rooms.nip04.history] retry after no-progress round" } - loadMore(user) - } - } - } + windowActive = false + _loadingMore.value = false + windowUser?.let { user -> + exhaustedByUser[user.pubkeyHex] = true + if (activeUser == user.pubkeyHex) _exhausted.value = true + updateStatus(user) + logSettleSummary(user) } } wasLoading = loading @@ -220,36 +214,71 @@ class ChatroomListNip04HistorySubAssembler( } } + // WindowLoadTracker reports relays that accepted a REQ then went silent, or never got their REQ out. + // We do NOT give up on them (they may simply be slow and need to catch up) — we just record them as + // stalled for the logs and let them keep their open subscription. + private fun onRelaysStalled(relays: Set) { + started.forEach { pk -> relays.forEach { markStalled(pk, it, "no response (silence/connect timeout)") } } + } + + // Records [relay] as not currently advancing for [pk] and logs it once (the first time it stalls in + // this window). The relay is kept — it kept its subscription and keeps trying to catch up. + private fun markStalled( + pk: HexKey, + relay: NormalizedRelayUrl, + reason: String, + ) { + val firstTime = stalledRelays.getOrPut(pk) { ConcurrentHashMap.newKeySet() }.add(relay) + if (firstTime) Log.d("DMPagination") { "[rooms.nip04.history] ${relay.url} stalled — $reason (kept open, still trying)" } + } + + private fun updateStatus(user: User) { + val account = accounts[user.pubkeyHex] + val all = account?.let { (it.homeRelays.flow.value + it.dmRelays.flow.value).toSet() } ?: emptySet() + // "Asking N relays" on the status card: the ones still being paged (done relays have dropped out). + _relayCount.value = pager.activeRelays(user.pubkeyHex, all).size + // Over ALL relays, not just the still-active ones: a relay that finished keeps its deep cursor, so + // "reached back to X" stays monotonic instead of jumping back to a newer date when the deepest + // relay drops out of the active set. + _reachedBack.value = pager.deepestUntil(user.pubkeyHex, all, floor()) + } + + // A one-line breakdown of where each relay landed when the window settles — the snapshot to reach for + // when history didn't load tomorrow: who reached the bottom vs. who is still being retried. + private fun logSettleSummary(user: User) { + val account = accounts[user.pubkeyHex] ?: return + val all = (account.homeRelays.flow.value + account.dmRelays.flow.value).toSet() + val done = all.filter { pager.isDone(user.pubkeyHex, it) }.map { it.url } + val stillTrying = all.filterNot { pager.isDone(user.pubkeyHex, it) }.map { it.url } + Log.d("DMPagination") { "[rooms.nip04.history] settled — done=$done still-trying=$stillTrying" } + } + override fun newSub(key: ChatroomListState): Subscription { val user = user(key) scope = key.account.scope accounts[user.pubkeyHex] = key.account if (activeUser != user.pubkeyHex) { activeUser = user.pubkeyHex + // Account switched: repoint the shared display flows to this account's own state. _exhausted.value = exhaustedByUser[user.pubkeyHex] ?: false _relayCount.value = 0 _reachedBack.value = null - lastAskedActive = emptySet() - lastRoundEventCount = -1 } return requestNewSubscription(historyListener(user, key)) } - // Flips to exhausted only once every relay has returned an empty page + EOSE. Sets true only. - private fun markExhaustedIfAllDone(user: User) { - val account = accounts[user.pubkeyHex] ?: return - val allRelays = (account.homeRelays.flow.value + account.dmRelays.flow.value).toSet() - if (allRelays.isNotEmpty() && pager.activeRelays(user.pubkeyHex, allRelays).isEmpty()) { - exhaustedByUser[user.pubkeyHex] = true - if (activeUser == user.pubkeyHex) _exhausted.value = true - } - } - private fun historyListener( user: User, key: ChatroomListState, ): SubscriptionListener = object : SubscriptionListener { + override fun onSubscriptionStarted( + relay: String, + forFilters: List, + ) { + windowLoad.onReqSent(relay) + } + override fun onEvent( event: Event, isLive: Boolean, @@ -258,17 +287,27 @@ class ChatroomListNip04HistorySubAssembler( ) { windowLoad.onRelayEvent(relay) pager.onEvent(user.pubkeyHex, relay, event.createdAt) + stalledRelays[user.pubkeyHex]?.remove(relay) } override fun onEose( relay: NormalizedRelayUrl, forFilters: List?, ) { + stalledRelays[user.pubkeyHex]?.remove(relay) pager.onEose(user.pubkeyHex, relay) - windowLoad.onRelaySettled(relay) + if (pager.isDone(user.pubkeyHex, relay)) { + // Reached the bottom on this relay: settle it for the spinner, nothing more to ask. + windowLoad.onRelaySettled(relay) + Log.d("DMPagination") { "[rooms.nip04.history] ${relay.url} reached the bottom (done)" } + } else { + // This page had events: reset only this relay's tally and let it continue to its next + // page immediately, independent of every other relay. + pager.beginRound(user.pubkeyHex, listOf(relay)) + } newEose(key, relay, TimeUtils.now(), forFilters) - lastRoundEventCount = -1 - markExhaustedIfAllDone(user) + updateStatus(user) + invalidateFilters() } override fun onClosed( @@ -276,8 +315,11 @@ class ChatroomListNip04HistorySubAssembler( relay: NormalizedRelayUrl, forFilters: List?, ) { + // A relay may demand auth we can't satisfy and CLOSE. It's stalled, not done — keep its + // subscription so the pool can re-auth and it can catch up — but don't let it hold the + // spinner. windowLoad.onRelaySettled(relay) - if (pager.onClosed(user.pubkeyHex, relay)) markExhaustedIfAllDone(user) + markStalled(user.pubkeyHex, relay, "CLOSED: $message") } override fun onCannotConnect( @@ -285,18 +327,12 @@ class ChatroomListNip04HistorySubAssembler( message: String, forFilters: List?, ) { - // Count cannot-connect toward give-up like a CLOSE, so an unreachable relay doesn't keep - // exhaustion false forever (otherwise a cold, empty feed retries it endlessly and stays on - // the spinner). Once given up, exhaustion completes and the screen resolves. windowLoad.onRelaySettled(relay) - if (pager.onClosed(user.pubkeyHex, relay)) markExhaustedIfAllDone(user) + markStalled(user.pubkeyHex, relay, "cannot connect: $message") } } companion object { private const val PAGE_LIMIT = 10000 - - // Backoff before retrying a no-progress, not-exhausted round (transient relay failure). - private const val NO_PROGRESS_RETRY_MS = 5_000L } } 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 deleted file mode 100644 index 595bc4c2fb..0000000000 --- a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/UntilLimitPagerGiveUpTest.kt +++ /dev/null @@ -1,69 +0,0 @@ -/* - * Copyright (c) 2025 Vitor Pamplona - * - * Permission is hereby granted, free of charge, to any person obtaining a copy of - * this software and associated documentation files (the "Software"), to deal in - * the Software without restriction, including without limitation the rights to use, - * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the - * Software, and to permit persons to whom the Software is furnished to do so, - * subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS - * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR - * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN - * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION - * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. - */ -package com.vitorpamplona.amethyst.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)) - } -}