From 60b8629a27b07de0df894516c81dee69af94addf Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 4 Jun 2026 18:36:07 +0000 Subject: [PATCH] refactor(dm): page NIP-17 gift-wrap history per relay, like NIP-04 The gift-wrap history loader was round-based: loadMore asked every active relay together and the next page only went out after the whole round settled, so one slow or auth-walled relay throttled the cadence and fast relays idled until the laggards finished. The per-conversation NIP-04 loader already pages each relay independently; this brings NIP-17 to the same model. Now a single loadMore opens one window spanning the whole walk, and each relay continues itself the instant it EOSEs a non-empty page (pager.beginRound([relay]) + invalidateFilters, which the sub layer diffs so only the advanced relay re-REQs). Fast relays race to the bottom while slow ones catch up in the background. The WindowLoadTracker switches to tracksReqSends=true with an onAbandoned handler that marks silent/ unreachable relays stalled (kept open, still trying) rather than giving up on them - exhaustion comes from the window settling via the tracker's silence + connect-grace backstops, which also removes the need for the old no-progress retry loop. Drop the now-dead loadEverything/autoLoadAll (no callers; with independent paging one loadMore already walks each relay to its bottom). Pin the history floor per window to keep un-advanced relays' filters stable across the per-EOSE invalidateFilters, and keep reachedBack monotonic over all relays. --- .../AccountGiftWrapsHistoryEoseManager.kt | 307 +++++++++--------- 1 file changed, 157 insertions(+), 150 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsHistoryEoseManager.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsHistoryEoseManager.kt index 1cc766027a..9dd5d66bb9 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsHistoryEoseManager.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsHistoryEoseManager.kt @@ -41,7 +41,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 @@ -50,15 +49,22 @@ import java.util.concurrent.ConcurrentHashMap /** * Loads the account's NIP-17 gift-wrap **history** — everything older than the one-week live tail - * ([AccountGiftWrapsEoseManager]) — by **`until`+`limit` paging, per relay**. + * ([AccountGiftWrapsEoseManager]) — by **`until`+`limit` paging, per relay, independently**. * - * Idle until a screen calls [loadMore]. Each round asks every not-yet-empty relay for [PAGE_LIMIT] - * gift wraps older than its own cursor (no `since`, so gaps are skipped). When a relay answers an - * empty page with EOSE it is done; otherwise its cursor advances below the oldest wrap it sent. The - * limit is **not** trusted as a stop signal (a relay may cap results on its own) — only an empty page - * is. The whole history is [exhausted] once a full round advances no relay at all (every relay - * empty-EOSE'd or only answered CLOSED), which is the gap-proof "nothing more is reachable" signal the - * old time-slice model couldn't produce. + * There are no lock-step rounds: 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). Fast + * relays race to the bottom of the history in back-to-back pages while slow / auth-walled relays catch + * up at their own pace in the background — none holds the others back. This mirrors the per-conversation + * NIP-04 loader ([com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.privateDM.datasource.ChatroomNip04HistorySubAssembler]). + * + * A relay is *done* once it answers an empty page (nothing older). A relay that won't answer (auth + * CLOSE, unreachable, silent) is marked *stalled* for the logs but keeps its subscription open and keeps + * trying. The [loadingMore] spinner reflects whether anything is still actively advancing across one + * window spanning the whole walk; it clears — and [exhausted] flips — once every relay is either done or + * stalled (the [WindowLoadTracker] settles silent / unreachable relays via its silence + connect-grace + * backstops), without waiting on the slow ones beyond that. */ class AccountGiftWrapsHistoryEoseManager( client: INostrClient, @@ -66,75 +72,79 @@ class AccountGiftWrapsHistoryEoseManager( ) : PerUserEoseManager(client, allKeys) { override fun user(key: AccountQueryState) = key.account.userProfile() + // Per-relay cursors, keyed by account pubkey so switching accounts preserves each one's progress. private val pager = UntilLimitPager() // Users that have requested history at least once (else the manager stays idle, issuing no REQ). private val started = ConcurrentHashMap.newKeySet() - // The relays the in-flight round asked, per user — used to tally the round on completion. - private val askedRelays = ConcurrentHashMap>() - - // The account behind each user pubkey, captured on subscribe so [loadMore] (UI thread) can read - // the DM relay list without the key. + // The account behind each user pubkey, captured on subscribe so [loadMore] (UI thread) can read the + // DM relay list without the key. private val accounts = ConcurrentHashMap() + // 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>() + // This manager is shared across logged-in accounts (one singleton coordinator), so the single - // display flows below must follow whichever account is currently active. Per-account paging - // cursors live in [pager], so switching away and back preserves progress; [exhaustedByUser] lets - // the display flow repoint accurately on switch instead of leaking the previous account's state. + // display flows below must follow whichever account is currently active. Per-account paging cursors + // live in [pager], so switching away and back preserves progress; [exhaustedByUser] lets the display + // flow repoint accurately on switch instead of leaking the previous account's state. @Volatile private var activeUser: HexKey? = null private val exhaustedByUser = ConcurrentHashMap() - // No-progress guard: the relay set the last round asked and how many events it returned. If a fresh - // loadMore would ask the same relays and the last round brought nothing (all CLOSED / unanswered), - // we skip it rather than busy-retry — the pool re-auths on the open subscription and its EOSE will - // advance/finish them. Cleared by onEose (any EOSE means something changed). - @Volatile - private var lastAskedActive: Set = emptySet() + private val windowLoad = WindowLoadTracker("giftwrap.history", tracksReqSends = true, onAbandoned = ::onRelaysStalled) - @Volatile - private var lastRoundEventCount = -1 + // 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 windowLoad = WindowLoadTracker("giftwrap.history") - val loadingMore: StateFlow = windowLoad.loading - - // True once a full round advanced no relay — nothing older is reachable. + // True once the window settles (every relay done or stalled) — nothing more is reachable right now. private val _exhausted = MutableStateFlow(false) val exhausted: StateFlow = _exhausted.asStateFlow() - // Status surfaced to the loading card: how many relays the current page is asking, and the oldest - // point paging has reached (epoch seconds, the deepest cursor). + // Status surfaced to the loading card: how many relays are still being paged, and the oldest point + // paging has reached (epoch seconds, the deepest cursor). private val _relayCount = MutableStateFlow(0) val relayCount: StateFlow = _relayCount.asStateFlow() private val _reachedBack = MutableStateFlow(null) val reachedBack: StateFlow = _reachedBack.asStateFlow() - // Account scope for the watchdog / round collector. Volatile: written on IO (newSub), read on UI. + // Account scope for the done collector. Volatile: written on IO (newSub), read on UI. @Volatile private var scope: CoroutineScope? = null @Volatile - private var roundJob: Job? = null + private var doneJob: Job? = null - // Backoff retry after a round that made no progress but isn't exhausted (relays failed to answer - // cleanly — cannot-connect / CLOSE during a connect storm — rather than empty-EOSE'ing). Without it - // the no-progress guard would never re-fire and a cold, empty feed would stay on the spinner 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 - // The user whose round is in flight, read by the round collector on completion. + // 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 - // "Load entire history" mode: keep paging to the end without waiting for more scrolling. + // 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 // History starts just below the live tail's one-week floor and pages backward from there. private fun startUntil() = TimeUtils.now() - AccountGiftWrapsEoseManager.LIVE_TAIL_SECONDS + private fun floor() = windowFloor.takeIf { it != 0L } ?: startUntil() + private fun daysAgo(epochSeconds: Long) = (TimeUtils.now() - epochSeconds) / TimeUtils.ONE_DAY override fun updateFilter( @@ -142,115 +152,82 @@ class AccountGiftWrapsHistoryEoseManager( 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 relays = key.account.dmRelays.flow.value val active = pager.activeRelays(user.pubkeyHex, relays).toSet() - askedRelays[user.pubkeyHex] = active - windowLoad.setExpectedRelays(active) if (active.isEmpty()) return emptyList() DmRelayLog.log("giftwrap.history", key.account) - Log.d(TAG) { "[giftwrap.history] REQ ${active.size} relay(s) ${active.map { it.url }}, limit=$PAGE_LIMIT (until ${daysAgo(pager.untilFor(user.pubkeyHex, active.first(), startUntil()))}d…)" } + Log.d(TAG) { "[giftwrap.history] REQ ${active.size} relay(s) ${active.map { it.url }}, limit=$PAGE_LIMIT (until ${daysAgo(pager.untilFor(user.pubkeyHex, active.first(), floor()))}d…)" } return active.flatMap { relay -> filterGiftWrapsToPubkey( relay = relay, pubkey = user.pubkeyHex, since = null, - until = pager.untilFor(user.pubkeyHex, relay, startUntil()), + until = pager.untilFor(user.pubkeyHex, relay, floor()), limit = PAGE_LIMIT, ) } } - /** Requests the next backward page from every relay that still has older history. No-op if exhausted. */ + /** Starts (or resumes) per-relay paging of the gift-wrap history. Idempotent: safe to call again. */ fun loadMore(user: User) { - if (_exhausted.value) { - Log.d(TAG) { "[giftwrap.history] loadMore ignored — exhausted" } - return - } val account = accounts[user.pubkeyHex] ?: return started.add(user.pubkeyHex) - val active = pager.activeRelays(user.pubkeyHex, account.dmRelays.flow.value) - if (active.isEmpty()) { + val allRelays = account.dmRelays.flow.value + if (allRelays.isEmpty()) return + if (pager.activeRelays(user.pubkeyHex, allRelays).isEmpty()) { + // Everything already paged to the bottom. exhaustedByUser[user.pubkeyHex] = true _exhausted.value = true return } - val activeSet = active.toSet() - if (activeSet == lastAskedActive && lastRoundEventCount == 0) { - // The same relays just returned nothing (e.g. all CLOSED, auth pending). The pool retries - // auth on the open subscription and its EOSE finishes them (see markExhaustedIfAllDone), so - // don't hammer with an identical round. onEose clears this gate when anything changes. - Log.d(TAG) { "[giftwrap.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, 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, account.dmRelays.flow.value, startUntil()) - Log.d(TAG) { "[giftwrap.history] loadMore → ${active.size} active relay(s)" } + _exhausted.value = false + DmRelayLog.log("giftwrap.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(allRelays.toSet()) } + updateStatus(user) + Log.d(TAG) { "[giftwrap.history] paging ${allRelays.size} relay(s) independently: ${allRelays.map { it.url }}" } invalidateFilters() } - /** Pages to the very end: each completed round auto-issues the next until the history is exhausted. */ - fun loadEverything(user: User) { - if (_exhausted.value) return - autoLoadAll = true - Log.d(TAG) { "[giftwrap.history] loadEverything — paging to the end" } - loadMore(user) - } - - // Emits the round tally and the exhausted decision when the in-flight load settles. Exhausted ONLY - // when every relay has returned an empty page + EOSE (pager.done): a relay that merely CLOSED (e.g. - // auth-required, before its post-auth retry) or never answered is NOT finished, so we don't call it - // "all caught up" — we keep loading it. In load-all mode, keep paging until that's true. - 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 allRelays = accounts[user.pubkeyHex]?.dmRelays?.flow?.value ?: emptySet() - 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(TAG) { "[giftwrap.history] round done: $count event(s), exhausted=$exhaustedNow" } - if (autoLoadAll && !exhaustedNow) { - loadMore(user) - } else if (!exhaustedNow && count == 0) { - // No progress and not exhausted: the relays failed to answer cleanly - // (cannot-connect / CLOSE) rather than empty-EOSE'ing. Retry after a - // backoff so a transient failure recovers — paced so a fast-CLOSE - // (rate-limited) relay isn't hammered. Stops once exhausted or loading. - retryJob?.cancel() - retryJob = - scope.launch { - delay(NO_PROGRESS_RETRY_MS) - if (!_exhausted.value && !windowLoad.loading.value) { - lastAskedActive = emptySet() - Log.d(TAG) { "[giftwrap.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 @@ -258,14 +235,41 @@ class AccountGiftWrapsHistoryEoseManager( } } - // Flips to exhausted only once every relay has returned an empty page + EOSE (all done). Sets true - // only — the false transitions belong to loadMore / the round collector. Safe off the round path. - private fun markExhaustedIfAllDone(user: User) { - val allRelays = accounts[user.pubkeyHex]?.dmRelays?.flow?.value ?: return - if (allRelays.isNotEmpty() && pager.activeRelays(user.pubkeyHex, allRelays).isEmpty()) { - exhaustedByUser[user.pubkeyHex] = true - if (activeUser == user.pubkeyHex) _exhausted.value = true - } + // 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(TAG) { "[giftwrap.history] ${relay.url} stalled — $reason (kept open, still trying)" } + } + + private fun updateStatus(user: User) { + val relays = accounts[user.pubkeyHex]?.dmRelays?.flow?.value ?: emptySet() + // "Asking N relays" on the status card: the ones still being paged (done relays have dropped out). + _relayCount.value = pager.activeRelays(user.pubkeyHex, relays).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, relays, 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 relays = accounts[user.pubkeyHex]?.dmRelays?.flow?.value ?: return + val done = relays.filter { pager.isDone(user.pubkeyHex, it) }.map { it.url } + val stillTrying = relays.filterNot { pager.isDone(user.pubkeyHex, it) }.map { it.url } + Log.d(TAG) { "[giftwrap.history] settled — done=$done still-trying=$stillTrying" } } override fun newSub(key: AccountQueryState): Subscription { @@ -278,8 +282,6 @@ class AccountGiftWrapsHistoryEoseManager( _exhausted.value = exhaustedByUser[user.pubkeyHex] ?: false _relayCount.value = 0 _reachedBack.value = null - lastAskedActive = emptySet() - lastRoundEventCount = -1 } return requestNewSubscription(historyListener(user, key)) } @@ -289,6 +291,13 @@ class AccountGiftWrapsHistoryEoseManager( key: AccountQueryState, ): SubscriptionListener = object : SubscriptionListener { + override fun onSubscriptionStarted( + relay: String, + forFilters: List, + ) { + windowLoad.onReqSent(relay) + } + override fun onEvent( event: Event, isLive: Boolean, @@ -297,21 +306,27 @@ class AccountGiftWrapsHistoryEoseManager( ) { 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(TAG) { "[giftwrap.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) - // An EOSE means this relay changed (finished, or delivered a page) — clear the - // no-progress gate so the next loadMore can continue, even off the round path. - lastRoundEventCount = -1 - // A post-auth empty EOSE can land after the round already settled on the earlier CLOSED; - // flip to exhausted the moment this finishes the last relay, not only at round end. - markExhaustedIfAllDone(user) + updateStatus(user) + invalidateFilters() } override fun onClosed( @@ -319,12 +334,11 @@ class AccountGiftWrapsHistoryEoseManager( relay: NormalizedRelayUrl, forFilters: List?, ) { - // CLOSED (e.g. auth-required) is not "empty": don't mark the relay done — it may answer - // after the auth handshake. It just settles the load so the spinner can clear. But if it - // keeps rejecting us (auth we can't satisfy), the pager eventually gives up on it; once - // that's the last blocker, exhaustion can complete. + // A relay (e.g. an author's) 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( @@ -332,12 +346,8 @@ class AccountGiftWrapsHistoryEoseManager( message: String, forFilters: List?, ) { - // Cannot-connect is also "no answer": count it toward give-up like a CLOSE, so a relay - // that's unreachable (down / blocked) doesn't keep exhaustion false forever — otherwise a - // cold, empty feed would retry it endlessly and stay on the spinner. Once it's given up, - // exhaustion can complete and the screen resolves (to the loaded rooms, or empty + retry). windowLoad.onRelaySettled(relay) - if (pager.onClosed(user.pubkeyHex, relay)) markExhaustedIfAllDone(user) + markStalled(user.pubkeyHex, relay, "cannot connect: $message") } } @@ -348,8 +358,5 @@ class AccountGiftWrapsHistoryEoseManager( // relay allows it. A relay returning fewer is treated as its own cap, NOT as "nothing more" — // only an empty page + EOSE ends a relay. 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 } }