From 1c7e073b480dcdca657f2c82ffab1161556150f7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 5 Jun 2026 16:23:30 +0000 Subject: [PATCH] refactor(dm): wire the three history managers onto BackwardRelayPager + tests Step 3+tests of the pagination generalization. The gift-wrap, conversation NIP-04, and rooms-list NIP-04 history managers each reimplemented the same per-relay cursor / in-flight / stall / exhausted / display-flow bookkeeping; they now delegate all of it to the shared BackwardRelayPager and keep only what is genuinely theirs: building the protocol's REQ filters, the relaysFor lookup, and forwarding subscription callbacks. ~550 lines of duplicated logic removed; the public API (loadingMore/exhausted/relayCount/stalledCount/ reachedBack/relayProgress, advance/advanceAll) is unchanged, so the UI is untouched. Behaviour-preserving. Tests (quartz jvmAndroidTest): - BackwardRelayPagerTest drives the engine's callbacks directly and pins the logic that backed the bugs in this branch: empty page -> done -> caught up; a CLOSED / cannot-connect relay -> stalled -> exhausted-but-INCOMPLETE (stalledCount > 0, not "all caught up"); re-advance clears a stall; the reached cursor is the deepest across relays; a done relay won't re-advance; advanceAll arms only not-done relays; switching the active key repoints the display flows and restores a backgrounded key's terminal state. - UntilLimitPagingRelayTest drives a real NostrClient against the in-process relay (geode) to pin the wire contract the design rests on: a backward until+limit walk returns each event exactly once (no re-download), newest first, capped at the limit, with an empty page + EOSE as the gap-proof stop. --- .../AccountGiftWrapsHistoryEoseManager.kt | 200 +++------------- .../ChatroomNip04HistorySubAssembler.kt | 180 ++------------ .../ChatroomListNip04HistorySubAssembler.kt | 182 +++----------- .../client/paging/BackwardRelayPagerTest.kt | 224 ++++++++++++++++++ .../paging/UntilLimitPagingRelayTest.kt | 100 ++++++++ 5 files changed, 405 insertions(+), 481 deletions(-) create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/BackwardRelayPagerTest.kt create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/UntilLimitPagingRelayTest.kt 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 0752d5b9b7..0b96b1d755 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 @@ -30,9 +30,8 @@ import com.vitorpamplona.amethyst.service.relays.SincePerRelayMap import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient -import com.vitorpamplona.quartz.nip01Core.relay.client.paging.PerRelayLoadTracker +import com.vitorpamplona.quartz.nip01Core.relay.client.paging.BackwardRelayPager import com.vitorpamplona.quartz.nip01Core.relay.client.paging.RelayPagingProgress -import com.vitorpamplona.quartz.nip01Core.relay.client.paging.UntilLimitPager import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription @@ -40,9 +39,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quartz.utils.TimeUtils -import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow -import kotlinx.coroutines.flow.asStateFlow import java.util.concurrent.ConcurrentHashMap /** @@ -55,9 +52,10 @@ import java.util.concurrent.ConcurrentHashMap * (see the rooms-list / conversation feed views). So a spam-dense relay never floods: the user has to * scroll through its messages to pull more, and nothing is fetched while its marker is off screen. * - * A relay is *done* once it answers an empty page; one that won't answer (auth CLOSE, unreachable, or - * silent past the load tracker's window) is flagged *stalled* but kept. [exhausted] flips once every - * relay is either done or stalled — nothing more is reachable right now. + * The per-relay cursor / stall / exhaustion bookkeeping lives in the shared [BackwardRelayPager]; this + * class only builds the gift-wrap REQ filters and forwards relay callbacks into the pager. A relay is + * *done* once it answers an empty page; one that won't answer (auth CLOSE, unreachable, or silent) is + * flagged *stalled* but kept. [exhausted] flips once every relay is either done or stalled. */ class AccountGiftWrapsHistoryEoseManager( client: INostrClient, @@ -65,53 +63,22 @@ class AccountGiftWrapsHistoryEoseManager( ) : PerUserEoseManager(client, allKeys) { override fun user(key: AccountQueryState) = key.account.userProfile() - // Per-relay demand-driven cursors, keyed by account pubkey so switching accounts preserves progress. - private val pager = UntilLimitPager() - - // The account behind each user pubkey, captured on subscribe so the UI-thread API can read the DM - // relay list without the key. + // The account behind each user pubkey, captured on subscribe so the pager's relaysFor lookup and the + // advance() API can read the DM relay list (and the account scope) without the key. private val accounts = ConcurrentHashMap() - // Relays not currently advancing for a user (auth CLOSE / unreachable / silent). Kept (not given up) - // and surfaced as stalled in the markers; they resume if the user re-advances them. - private val stalledRelays = ConcurrentHashMap>() + // Per-relay demand-driven paging, keyed by account pubkey so switching accounts preserves progress. + private val pager = + BackwardRelayPager("giftwrap.history") { pk -> + accounts[pk]?.dmRelays?.flow?.value + } - // 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() - - private val loadTracker = PerRelayLoadTracker("giftwrap.history", onSilenced = ::onRelaysSilenced) - val loadingMore: StateFlow = loadTracker.loading - - private val _exhausted = MutableStateFlow(false) - val exhausted: StateFlow = _exhausted.asStateFlow() - - // Relays currently fetching a page (for the "asking N relays" status line). - private val _relayCount = MutableStateFlow(0) - val relayCount: StateFlow = _relayCount.asStateFlow() - - // Relays that aren't done but can't be reached right now (auth CLOSE / unreachable / silent). Surfaced - // on the paused card as "waiting on N relays" — they aren't in-flight, so [relayCount] wouldn't show them. - private val _stalledCount = MutableStateFlow(0) - val stalledCount: StateFlow = _stalledCount.asStateFlow() - - private val _reachedBack = MutableStateFlow(null) - val reachedBack: StateFlow = _reachedBack.asStateFlow() - - // Per-relay window limits — where each relay has paged to, done/stalled — the data the on-screen - // markers render and drive their advance from. - private val _relayProgress = MutableStateFlow>(emptyMap()) - val relayProgress: StateFlow> = _relayProgress.asStateFlow() - - // History starts just below the live tail's one-week floor and pages backward from there. Pinned per - // account for the session: it must NOT drift forward on every recompute, or an un-delivered relay's - // marker (which sits at this floor) would keep changing and re-trigger its on-screen sentinel. The - // live tail covers everything newer than the floor. - private val pinnedFloor = ConcurrentHashMap() - - private fun startUntil(pk: HexKey) = pinnedFloor.getOrPut(pk) { TimeUtils.now() - AccountGiftWrapsEoseManager.LIVE_TAIL_SECONDS } + val loadingMore: StateFlow = pager.loadingMore + val exhausted: StateFlow = pager.exhausted + val relayCount: StateFlow = pager.relayCount + val stalledCount: StateFlow = pager.stalledCount + val reachedBack: StateFlow = pager.reachedBack + val relayProgress: StateFlow> = pager.relayProgress private fun daysAgo(epochSeconds: Long) = (TimeUtils.now() - epochSeconds) / TimeUtils.ONE_DAY @@ -130,8 +97,8 @@ class AccountGiftWrapsHistoryEoseManager( DmRelayLog.log("giftwrap.history", key.account) return armed.flatMap { relay -> val until = pager.requestedUntilFor(user.pubkeyHex, relay) ?: return@flatMap emptyList() - Log.d(TAG) { "[giftwrap.history] REQ ${relay.url} until ${daysAgo(until)}d, limit=$PAGE_LIMIT" } - filterGiftWrapsToPubkey(relay = relay, pubkey = user.pubkeyHex, since = null, until = until, limit = PAGE_LIMIT) + Log.d(TAG) { "[giftwrap.history] REQ ${relay.url} until ${daysAgo(until)}d, limit=${pager.pageLimit}" } + filterGiftWrapsToPubkey(relay = relay, pubkey = user.pubkeyHex, since = null, until = until, limit = pager.pageLimit) } } @@ -140,116 +107,25 @@ class AccountGiftWrapsHistoryEoseManager( user: User, relay: NormalizedRelayUrl, ) { - if (arm(user, relay)) { - _exhausted.value = false - updateStatus(user) - invalidateFilters() - } + val account = accounts[user.pubkeyHex] ?: return + if (pager.advance(user.pubkeyHex, relay, account.scope)) invalidateFilters() } /** Steps every not-done, not-in-flight relay one page. For the empty/initial boundary (nothing to scroll). */ fun advanceAll(user: User) { val account = accounts[user.pubkeyHex] ?: return - var any = false - account.dmRelays.flow.value - .forEach { if (arm(user, it)) any = true } - if (any) { + if (pager.advanceAll(user.pubkeyHex, account.scope)) { Log.d(TAG) { "[giftwrap.history] advanceAll (empty-feed bootstrap)" } - _exhausted.value = false - updateStatus(user) invalidateFilters() } } - // Moves one relay's cursor to its next page and marks it in-flight. Returns false if it can't advance - // (unknown relay, already fetching, or already done). Does NOT invalidate — the caller batches that. - private fun arm( - user: User, - relay: NormalizedRelayUrl, - ): Boolean { - val account = accounts[user.pubkeyHex] ?: return false - if (relay !in account.dmRelays.flow.value) return false - if (loadTracker.isInFlight(relay)) return false - if (!pager.advance(user.pubkeyHex, relay, startUntil(user.pubkeyHex))) return false - stalledRelays[user.pubkeyHex]?.remove(relay) - loadTracker.bind(account.scope) - loadTracker.onAdvance(relay) - return true - } - - private fun onRelaysSilenced(relays: Set) { - val pk = activeUser ?: return - relays.forEach { markStalled(pk, it, "no response (silence timeout)") } - accounts[pk]?.userProfile()?.let { - updateStatus(it) - recomputeExhausted(it) - } - } - - 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, advance to retry)" } - } - - private fun updateStatus(user: User) { - // The display flows are singletons shown for the foreground account; a background account's late - // EOSE must not overwrite them (its cursors still advance in the pager). - if (activeUser != user.pubkeyHex) return - val relays = accounts[user.pubkeyHex]?.dmRelays?.flow?.value ?: emptySet() - _relayCount.value = loadTracker.count() - val start = startUntil(user.pubkeyHex) - _reachedBack.value = pager.deepestReached(user.pubkeyHex, relays, start) - val stalled = stalledRelays[user.pubkeyHex] ?: emptySet() - _stalledCount.value = relays.count { it in stalled && !pager.isDone(user.pubkeyHex, it) } - _relayProgress.value = - relays.associateWith { relay -> - RelayPagingProgress( - reachedUntil = pager.reachedUntilFor(user.pubkeyHex, relay, start), - done = pager.isDone(user.pubkeyHex, relay), - stalled = relay in stalled && !pager.isDone(user.pubkeyHex, relay), - ) - } - } - - // Exhausted once every relay is either done (empty page) or stalled (unreachable) — nothing more is - // reachable right now. A merely parked relay (more to load, just not advancing) keeps this false. - private fun recomputeExhausted(user: User) { - val relays = accounts[user.pubkeyHex]?.dmRelays?.flow?.value ?: return - if (relays.isEmpty()) return - val stalled = stalledRelays[user.pubkeyHex] ?: emptySet() - val pending = relays.any { !pager.isDone(user.pubkeyHex, it) && it !in stalled } - val ex = !pending - val was = exhaustedByUser[user.pubkeyHex] ?: false - exhaustedByUser[user.pubkeyHex] = ex - if (ex && !was) { - val done = relays.filter { pager.isDone(user.pubkeyHex, it) }.map { it.url } - val stuck = relays.filter { it in stalled && !pager.isDone(user.pubkeyHex, it) }.map { it.url } - Log.d(TAG) { "[giftwrap.history] window settled (nothing more reachable) — done=$done stalled=$stuck" } - } - if (activeUser == user.pubkeyHex) _exhausted.value = ex - } - override fun newSub(key: AccountQueryState): Subscription { val user = user(key) accounts[user.pubkeyHex] = key.account - loadTracker.bind(key.account.scope) - if (activeUser != user.pubkeyHex) { - activeUser = user.pubkeyHex - // Account switched: repoint the shared display flows to this account's own state. - loadTracker.reset() - _exhausted.value = exhaustedByUser[user.pubkeyHex] ?: false - _relayCount.value = 0 - _stalledCount.value = 0 - _reachedBack.value = null - _relayProgress.value = emptyMap() - } - // Populate the per-relay markers (all relays at the floor, not done) so the UI can render their - // window-limit sentinels and pull the first page when they come into view. - updateStatus(user) + // Repoint the shared display flows to this account and populate the per-relay markers (all relays + // at the floor, not done) so the UI can render their sentinels and pull the first page on view. + pager.activate(user.pubkeyHex) return requestNewSubscription(historyListener(user, key)) } @@ -264,25 +140,18 @@ class AccountGiftWrapsHistoryEoseManager( relay: NormalizedRelayUrl, forFilters: List?, ) { - loadTracker.onActivity() 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) - loadTracker.onSettled(relay) - if (pager.isDone(user.pubkeyHex, relay)) { + if (pager.onEose(user.pubkeyHex, relay)) { Log.d(TAG) { "[giftwrap.history] ${relay.url} reached the bottom (done)" } } // No auto-advance: the relay parks here until its marker asks for the next page. newEose(key, relay, TimeUtils.now(), forFilters) - updateStatus(user) - recomputeExhausted(user) } override fun onClosed( @@ -290,10 +159,7 @@ class AccountGiftWrapsHistoryEoseManager( relay: NormalizedRelayUrl, forFilters: List?, ) { - loadTracker.onSettled(relay) - markStalled(user.pubkeyHex, relay, "CLOSED: $message") - updateStatus(user) - recomputeExhausted(user) + pager.onClosed(user.pubkeyHex, relay, message) } override fun onCannotConnect( @@ -301,19 +167,11 @@ class AccountGiftWrapsHistoryEoseManager( message: String, forFilters: List?, ) { - loadTracker.onSettled(relay) - markStalled(user.pubkeyHex, relay, "cannot connect: $message") - updateStatus(user) - recomputeExhausted(user) + pager.onCannotConnect(user.pubkeyHex, relay, message) } } companion object { private const val TAG = "DMPagination" - - // Asked of every relay per page. Large on purpose: we want a whole band in one page where the - // 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 } } 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 71ceebcd16..6244c2218c 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 @@ -22,14 +22,12 @@ package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.privateDM.datasource import com.vitorpamplona.amethyst.service.relayClient.eoseManagers.DmRelayLog import com.vitorpamplona.amethyst.service.relayClient.eoseManagers.PerUserAndFollowListEoseManager -import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip59GiftWraps.AccountGiftWrapsEoseManager import com.vitorpamplona.amethyst.service.relays.SincePerRelayMap import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient -import com.vitorpamplona.quartz.nip01Core.relay.client.paging.PerRelayLoadTracker +import com.vitorpamplona.quartz.nip01Core.relay.client.paging.BackwardRelayPager import com.vitorpamplona.quartz.nip01Core.relay.client.paging.RelayPagingProgress -import com.vitorpamplona.quartz.nip01Core.relay.client.paging.UntilLimitPager import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription @@ -38,10 +36,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip17Dm.base.ChatroomKey import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quartz.utils.TimeUtils -import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow -import kotlinx.coroutines.flow.asStateFlow -import java.util.concurrent.ConcurrentHashMap /** * Loads older NIP-04 DMs (kind 4) for one conversation by **`until`+`limit` paging, per relay, on @@ -49,9 +44,10 @@ import java.util.concurrent.ConcurrentHashMap * for that relay asks ([advance]); otherwise it parks. Nothing is walked proactively — a relay pages * only while its marker is visible and keeps paging while it stays visible. * - * A relay is *done* once it answers an empty page; one that won't answer (auth CLOSE, unreachable, or - * silent past the load tracker's window) is flagged *stalled* but kept. [exhausted] flips once every - * relay is either done or stalled. + * The per-relay cursor / stall / exhaustion bookkeeping lives in the shared [BackwardRelayPager]; this + * class only builds the (per-relay scoped) NIP-04 REQ filters and forwards relay callbacks into it. A + * relay is *done* once it answers an empty page; one that won't answer (auth CLOSE, unreachable, or + * silent) is flagged *stalled* but kept. [exhausted] flips once every relay is either done or stalled. */ class ChatroomNip04HistorySubAssembler( client: INostrClient, @@ -66,47 +62,22 @@ class ChatroomNip04HistorySubAssembler( private fun convoKey(key: ChatroomQueryState) = ConvoKey(user(key).pubkeyHex, key.room) - private val pager = UntilLimitPager() + // The conversation's relay set for a key, resolved via the outbox model (per-relay scoped). + private fun relaysFor(pk: ConvoKey): Collection? = allKeys().firstOrNull { convoKey(it) == pk }?.let { nip04DMRelays(it.room.users, it.account)?.all } - private val stalledRelays = ConcurrentHashMap>() + private val pager = BackwardRelayPager("convo.nip04.history", relaysFor = ::relaysFor) - private val loadTracker = PerRelayLoadTracker("convo.nip04.history", onSilenced = ::onRelaysSilenced) - val loadingMore: StateFlow = loadTracker.loading - - private val _exhausted = MutableStateFlow(false) - val exhausted: StateFlow = _exhausted.asStateFlow() - - private val _relayCount = MutableStateFlow(0) - val relayCount: StateFlow = _relayCount.asStateFlow() - - // Not-done relays that can't be reached right now — shown as "waiting on N relays" on the paused card. - private val _stalledCount = MutableStateFlow(0) - val stalledCount: StateFlow = _stalledCount.asStateFlow() - - private val _reachedBack = MutableStateFlow(null) - val reachedBack: StateFlow = _reachedBack.asStateFlow() - - private val _relayProgress = MutableStateFlow>(emptyMap()) - val relayProgress: StateFlow> = _relayProgress.asStateFlow() - - // Shared across accounts/conversations (singleton coordinator): repoint the display flows to the - // conversation now on screen. Cursors live in [pager]. - @Volatile - private var activeConvo: ConvoKey? = null - private val exhaustedByConvo = ConcurrentHashMap() - - // Pinned per conversation for the session — must not drift forward, or an un-delivered relay's marker - // would keep moving and re-trigger its sentinel. See AccountGiftWrapsHistoryEoseManager. - private val pinnedFloor = ConcurrentHashMap() - - private fun startUntil(pk: ConvoKey) = pinnedFloor.getOrPut(pk) { TimeUtils.now() - AccountGiftWrapsEoseManager.LIVE_TAIL_SECONDS } + val loadingMore: StateFlow = pager.loadingMore + val exhausted: StateFlow = pager.exhausted + val relayCount: StateFlow = pager.relayCount + val stalledCount: StateFlow = pager.stalledCount + val reachedBack: StateFlow = pager.reachedBack + val relayProgress: StateFlow> = pager.relayProgress override fun user(key: ChatroomQueryState) = key.account.userProfile() override fun list(key: ChatroomQueryState) = key.listId - private fun relaysFor(pk: ConvoKey): Nip04DmRelays? = allKeys().firstOrNull { convoKey(it) == pk }?.let { nip04DMRelays(it.room.users, it.account) } - override fun updateFilter( key: ChatroomQueryState, since: SincePerRelayMap?, @@ -125,7 +96,7 @@ class ChatroomNip04HistorySubAssembler( toMeRelays = relays.toMeRelays.filterKeys { it in armed }, fromMeRelays = relays.fromMeRelays.filterKeys { it in armed }, ) - return filterNip04DMsHistory(key.account, scoped, PAGE_LIMIT) { relay -> + return filterNip04DMsHistory(key.account, scoped, pager.pageLimit) { relay -> pager.requestedUntilFor(pk, relay) } } @@ -133,110 +104,24 @@ class ChatroomNip04HistorySubAssembler( /** Steps a single [relay] to its next, older page for the open conversation(s). Driven by its marker. */ fun advance(relay: NormalizedRelayUrl) { var any = false - allKeys().forEach { if (arm(it, relay)) any = true } - if (any) { - _exhausted.value = false - updateStatus() - invalidateFilters() - } + allKeys().forEach { if (pager.advance(convoKey(it), relay, it.account.scope)) any = true } + if (any) invalidateFilters() } /** Steps every not-done, not-in-flight relay one page. For a thread too short to scroll. */ fun advanceAll() { var any = false - allKeys().forEach { key -> - val relays = nip04DMRelays(key.room.users, key.account) ?: return@forEach - relays.all.forEach { if (arm(key, it)) any = true } - } + allKeys().forEach { if (pager.advanceAll(convoKey(it), it.account.scope)) any = true } if (any) { Log.d("DMPagination") { "[convo.nip04.history] advanceAll (empty-thread bootstrap)" } - _exhausted.value = false - updateStatus() invalidateFilters() } } - private fun arm( - key: ChatroomQueryState, - relay: NormalizedRelayUrl, - ): Boolean { - val relays = nip04DMRelays(key.room.users, key.account) ?: return false - if (relay !in relays.all) return false - val pk = convoKey(key) - if (loadTracker.isInFlight(relay)) return false - if (!pager.advance(pk, relay, startUntil(pk))) return false - stalledRelays[pk]?.remove(relay) - loadTracker.bind(key.account.scope) - loadTracker.onAdvance(relay) - return true - } - - private fun onRelaysSilenced(relays: Set) { - val pk = activeConvo ?: return - relays.forEach { markStalled(pk, it, "no response (silence timeout)") } - updateStatus() - recomputeExhausted() - } - - private fun markStalled( - pk: ConvoKey, - relay: NormalizedRelayUrl, - reason: String, - ) { - val firstTime = stalledRelays.getOrPut(pk) { ConcurrentHashMap.newKeySet() }.add(relay) - if (firstTime) Log.d("DMPagination") { "[convo.nip04.history] ${relay.url} stalled — $reason (kept, advance to retry)" } - } - - private fun updateStatus() { - val pk = activeConvo ?: return - val relays = relaysFor(pk) ?: return - _relayCount.value = loadTracker.count() - val start = startUntil(pk) - _reachedBack.value = pager.deepestReached(pk, relays.all, start) - val stalled = stalledRelays[pk] ?: emptySet() - _stalledCount.value = relays.all.count { it in stalled && !pager.isDone(pk, it) } - _relayProgress.value = - relays.all.associateWith { relay -> - RelayPagingProgress( - reachedUntil = pager.reachedUntilFor(pk, relay, start), - done = pager.isDone(pk, relay), - stalled = relay in stalled && !pager.isDone(pk, relay), - ) - } - } - - private fun recomputeExhausted() { - val pk = activeConvo ?: return - val relays = relaysFor(pk) ?: return - if (relays.all.isEmpty()) return - val stalled = stalledRelays[pk] ?: emptySet() - val pending = relays.all.any { !pager.isDone(pk, it) && it !in stalled } - val ex = !pending - val was = exhaustedByConvo[pk] ?: false - exhaustedByConvo[pk] = ex - if (ex && !was) { - val done = relays.all.filter { pager.isDone(pk, it) }.map { it.url } - val stuck = relays.all.filter { it in stalled && !pager.isDone(pk, it) }.map { it.url } - Log.d("DMPagination") { "[convo.nip04.history] window settled (nothing more reachable) — done=$done stalled=$stuck" } - } - if (activeConvo == pk) _exhausted.value = ex - } - override fun newSub(key: ChatroomQueryState): Subscription { - val pk = convoKey(key) - loadTracker.bind(key.account.scope) - if (activeConvo != pk) { - activeConvo = pk - loadTracker.reset() - _exhausted.value = exhaustedByConvo[pk] ?: false - _relayCount.value = 0 - _stalledCount.value = 0 - _reachedBack.value = null - _relayProgress.value = emptyMap() - } - // Populate the per-relay markers (all relays at the floor, not done) so the UI can render their - // window-limit sentinels and pull the first page when they come into view. - updateStatus() + // Repoint the shared display flows to this conversation and populate the per-relay markers (all + // relays at the floor, not done) so the UI can render their sentinels and pull the first page. + pager.activate(convoKey(key)) return requestNewSubscription(historyListener(key)) } @@ -249,24 +134,17 @@ class ChatroomNip04HistorySubAssembler( relay: NormalizedRelayUrl, forFilters: List?, ) { - loadTracker.onActivity() pager.onEvent(pk, relay, event.createdAt) - stalledRelays[pk]?.remove(relay) } override fun onEose( relay: NormalizedRelayUrl, forFilters: List?, ) { - stalledRelays[pk]?.remove(relay) - pager.onEose(pk, relay) - loadTracker.onSettled(relay) - if (pager.isDone(pk, relay)) { + if (pager.onEose(pk, relay)) { Log.d("DMPagination") { "[convo.nip04.history] ${relay.url} reached the bottom (done)" } } newEose(key, relay, TimeUtils.now(), forFilters) - updateStatus() - recomputeExhausted() } override fun onClosed( @@ -274,10 +152,7 @@ class ChatroomNip04HistorySubAssembler( relay: NormalizedRelayUrl, forFilters: List?, ) { - loadTracker.onSettled(relay) - markStalled(pk, relay, "CLOSED: $message") - updateStatus() - recomputeExhausted() + pager.onClosed(pk, relay, message) } override fun onCannotConnect( @@ -285,15 +160,8 @@ class ChatroomNip04HistorySubAssembler( message: String, forFilters: List?, ) { - loadTracker.onSettled(relay) - markStalled(pk, relay, "cannot connect: $message") - updateStatus() - recomputeExhausted() + pager.onCannotConnect(pk, relay, message) } } } - - companion object { - private const val PAGE_LIMIT = 10000 - } } 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 c325a676db..a05b75eae7 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 @@ -24,14 +24,12 @@ import com.vitorpamplona.amethyst.model.Account import com.vitorpamplona.amethyst.model.User import com.vitorpamplona.amethyst.service.relayClient.eoseManagers.DmRelayLog import com.vitorpamplona.amethyst.service.relayClient.eoseManagers.PerUserEoseManager -import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip59GiftWraps.AccountGiftWrapsEoseManager import com.vitorpamplona.amethyst.service.relays.SincePerRelayMap import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient -import com.vitorpamplona.quartz.nip01Core.relay.client.paging.PerRelayLoadTracker +import com.vitorpamplona.quartz.nip01Core.relay.client.paging.BackwardRelayPager import com.vitorpamplona.quartz.nip01Core.relay.client.paging.RelayPagingProgress -import com.vitorpamplona.quartz.nip01Core.relay.client.paging.UntilLimitPager import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription @@ -39,9 +37,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quartz.utils.TimeUtils -import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow -import kotlinx.coroutines.flow.asStateFlow import java.util.concurrent.ConcurrentHashMap /** @@ -50,49 +46,31 @@ import java.util.concurrent.ConcurrentHashMap * ([com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip59GiftWraps.AccountGiftWrapsHistoryEoseManager]), * across the account's home (outbox, *from me*) + DM (inbox, *to me*) relays. Each relay advances one * page when its on-screen window-limit marker asks ([advance]); otherwise it parks. Nothing is walked - * proactively. + * proactively. The per-relay cursor / stall / exhaustion bookkeeping lives in [BackwardRelayPager]. */ class ChatroomListNip04HistorySubAssembler( client: INostrClient, allKeys: () -> Set, ) : PerUserEoseManager(client, allKeys) { - private val pager = UntilLimitPager() private val accounts = ConcurrentHashMap() - private val stalledRelays = ConcurrentHashMap>() - - @Volatile - private var activeUser: HexKey? = null - private val exhaustedByUser = ConcurrentHashMap() - - private val loadTracker = PerRelayLoadTracker("rooms.nip04.history", onSilenced = ::onRelaysSilenced) - val loadingMore: StateFlow = loadTracker.loading - - private val _exhausted = MutableStateFlow(false) - val exhausted: StateFlow = _exhausted.asStateFlow() - - private val _relayCount = MutableStateFlow(0) - val relayCount: StateFlow = _relayCount.asStateFlow() - - // Not-done relays that can't be reached right now — shown as "waiting on N relays" on the paused card. - private val _stalledCount = MutableStateFlow(0) - val stalledCount: StateFlow = _stalledCount.asStateFlow() - - private val _reachedBack = MutableStateFlow(null) - val reachedBack: StateFlow = _reachedBack.asStateFlow() - - private val _relayProgress = MutableStateFlow>(emptyMap()) - val relayProgress: StateFlow> = _relayProgress.asStateFlow() - - // Pinned per account for the session — must not drift forward, or an un-delivered relay's marker - // would keep moving and re-trigger its sentinel. See AccountGiftWrapsHistoryEoseManager. - private val pinnedFloor = ConcurrentHashMap() - - private fun startUntil(pk: HexKey) = pinnedFloor.getOrPut(pk) { TimeUtils.now() - AccountGiftWrapsEoseManager.LIVE_TAIL_SECONDS } - - override fun user(key: ChatroomListState) = key.account.userProfile() private fun allRelays(account: Account) = (account.homeRelays.flow.value + account.dmRelays.flow.value).toSet() + // Paged across the account's own home (outbox) + DM (inbox) relays, keyed by account pubkey. + private val pager = + BackwardRelayPager("rooms.nip04.history") { pk -> + accounts[pk]?.let { allRelays(it) } + } + + val loadingMore: StateFlow = pager.loadingMore + val exhausted: StateFlow = pager.exhausted + val relayCount: StateFlow = pager.relayCount + val stalledCount: StateFlow = pager.stalledCount + val reachedBack: StateFlow = pager.reachedBack + val relayProgress: StateFlow> = pager.relayProgress + + override fun user(key: ChatroomListState) = key.account.userProfile() + override fun updateFilter( key: ChatroomListState, since: SincePerRelayMap?, @@ -107,8 +85,8 @@ class ChatroomListNip04HistorySubAssembler( return armed.flatMap { relay -> val until = pager.requestedUntilFor(user.pubkeyHex, relay) ?: return@flatMap emptyList() buildList { - if (relay in homeRelays) add(filterNip04DMsFromMe(user, relay, since = null, until = until, limit = PAGE_LIMIT)) - if (relay in dmRelays) add(filterNip04DMsToMe(user, relay, since = null, until = until, limit = PAGE_LIMIT)) + if (relay in homeRelays) add(filterNip04DMsFromMe(user, relay, since = null, until = until, limit = pager.pageLimit)) + if (relay in dmRelays) add(filterNip04DMsToMe(user, relay, since = null, until = until, limit = pager.pageLimit)) } } } @@ -118,112 +96,25 @@ class ChatroomListNip04HistorySubAssembler( user: User, relay: NormalizedRelayUrl, ) { - if (arm(user, relay)) { - _exhausted.value = false - updateStatus(user) - invalidateFilters() - } + val account = accounts[user.pubkeyHex] ?: return + if (pager.advance(user.pubkeyHex, relay, account.scope)) invalidateFilters() } /** Steps every not-done, not-in-flight relay one page. For the empty/initial boundary (nothing to scroll). */ fun advanceAll(user: User) { val account = accounts[user.pubkeyHex] ?: return - var any = false - allRelays(account).forEach { if (arm(user, it)) any = true } - if (any) { + if (pager.advanceAll(user.pubkeyHex, account.scope)) { Log.d("DMPagination") { "[rooms.nip04.history] advanceAll (empty-feed bootstrap)" } - _exhausted.value = false - updateStatus(user) invalidateFilters() } } - private fun arm( - user: User, - relay: NormalizedRelayUrl, - ): Boolean { - val account = accounts[user.pubkeyHex] ?: return false - if (relay !in allRelays(account)) return false - if (loadTracker.isInFlight(relay)) return false - if (!pager.advance(user.pubkeyHex, relay, startUntil(user.pubkeyHex))) return false - stalledRelays[user.pubkeyHex]?.remove(relay) - loadTracker.bind(account.scope) - loadTracker.onAdvance(relay) - return true - } - - private fun onRelaysSilenced(relays: Set) { - val pk = activeUser ?: return - relays.forEach { markStalled(pk, it, "no response (silence timeout)") } - accounts[pk]?.userProfile()?.let { - updateStatus(it) - recomputeExhausted(it) - } - } - - 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, advance to retry)" } - } - - private fun updateStatus(user: User) { - // The display flows are singletons shown for the foreground account; a background account's late - // EOSE must not overwrite them (its cursors still advance in the pager). - if (activeUser != user.pubkeyHex) return - val account = accounts[user.pubkeyHex] - val relays = account?.let { allRelays(it) } ?: emptySet() - _relayCount.value = loadTracker.count() - val start = startUntil(user.pubkeyHex) - _reachedBack.value = pager.deepestReached(user.pubkeyHex, relays, start) - val stalled = stalledRelays[user.pubkeyHex] ?: emptySet() - _stalledCount.value = relays.count { it in stalled && !pager.isDone(user.pubkeyHex, it) } - _relayProgress.value = - relays.associateWith { relay -> - RelayPagingProgress( - reachedUntil = pager.reachedUntilFor(user.pubkeyHex, relay, start), - done = pager.isDone(user.pubkeyHex, relay), - stalled = relay in stalled && !pager.isDone(user.pubkeyHex, relay), - ) - } - } - - private fun recomputeExhausted(user: User) { - val account = accounts[user.pubkeyHex] ?: return - val relays = allRelays(account) - if (relays.isEmpty()) return - val stalled = stalledRelays[user.pubkeyHex] ?: emptySet() - val pending = relays.any { !pager.isDone(user.pubkeyHex, it) && it !in stalled } - val ex = !pending - val was = exhaustedByUser[user.pubkeyHex] ?: false - exhaustedByUser[user.pubkeyHex] = ex - if (ex && !was) { - val done = relays.filter { pager.isDone(user.pubkeyHex, it) }.map { it.url } - val stuck = relays.filter { it in stalled && !pager.isDone(user.pubkeyHex, it) }.map { it.url } - Log.d("DMPagination") { "[rooms.nip04.history] window settled (nothing more reachable) — done=$done stalled=$stuck" } - } - if (activeUser == user.pubkeyHex) _exhausted.value = ex - } - override fun newSub(key: ChatroomListState): Subscription { val user = user(key) accounts[user.pubkeyHex] = key.account - loadTracker.bind(key.account.scope) - if (activeUser != user.pubkeyHex) { - activeUser = user.pubkeyHex - loadTracker.reset() - _exhausted.value = exhaustedByUser[user.pubkeyHex] ?: false - _relayCount.value = 0 - _stalledCount.value = 0 - _reachedBack.value = null - _relayProgress.value = emptyMap() - } - // Populate the per-relay markers (all relays at the floor, not done) so the UI can render their - // window-limit sentinels and pull the first page when they come into view. - updateStatus(user) + // Repoint the shared display flows to this account and populate the per-relay markers (all relays + // at the floor, not done) so the UI can render their sentinels and pull the first page on view. + pager.activate(user.pubkeyHex) return requestNewSubscription(historyListener(user, key)) } @@ -238,24 +129,17 @@ class ChatroomListNip04HistorySubAssembler( relay: NormalizedRelayUrl, forFilters: List?, ) { - loadTracker.onActivity() 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) - loadTracker.onSettled(relay) - if (pager.isDone(user.pubkeyHex, relay)) { + if (pager.onEose(user.pubkeyHex, relay)) { Log.d("DMPagination") { "[rooms.nip04.history] ${relay.url} reached the bottom (done)" } } newEose(key, relay, TimeUtils.now(), forFilters) - updateStatus(user) - recomputeExhausted(user) } override fun onClosed( @@ -263,10 +147,7 @@ class ChatroomListNip04HistorySubAssembler( relay: NormalizedRelayUrl, forFilters: List?, ) { - loadTracker.onSettled(relay) - markStalled(user.pubkeyHex, relay, "CLOSED: $message") - updateStatus(user) - recomputeExhausted(user) + pager.onClosed(user.pubkeyHex, relay, message) } override fun onCannotConnect( @@ -274,14 +155,7 @@ class ChatroomListNip04HistorySubAssembler( message: String, forFilters: List?, ) { - loadTracker.onSettled(relay) - markStalled(user.pubkeyHex, relay, "cannot connect: $message") - updateStatus(user) - recomputeExhausted(user) + pager.onCannotConnect(user.pubkeyHex, relay, message) } } - - companion object { - private const val PAGE_LIMIT = 10000 - } } diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/BackwardRelayPagerTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/BackwardRelayPagerTest.kt new file mode 100644 index 0000000000..8e233a956f --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/BackwardRelayPagerTest.kt @@ -0,0 +1,224 @@ +/* + * 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.paging + +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 kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +/** + * State-machine tests for [BackwardRelayPager]: drive its relay callbacks directly (no network) and + * assert the cursor / done / stalled / exhausted bookkeeping — the logic that backed the "All caught up + * while messages missing" and the stalled-vs-done bugs. The relay's own `until`+`limit`+EOSE wire + * behaviour is covered separately against the in-process relay in `UntilLimitPagingRelayTest`. + */ +class BackwardRelayPagerTest { + private val r1 = NormalizedRelayUrl("wss://r1.example/") + private val r2 = NormalizedRelayUrl("wss://r2.example/") + private val r3 = NormalizedRelayUrl("wss://r3.example/") + private val key = "acct" + + private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + + @AfterTest + fun tearDown() { + scope.cancel() + } + + private fun pagerOf(vararg relays: NormalizedRelayUrl): BackwardRelayPager = BackwardRelayPager("test") { relays.toList() }.also { it.activate(key) } + + @Test + fun firstPageRequestsTheFloorAndAnEmptyPageIsCaughtUp() { + val p = pagerOf(r1) + assertFalse(p.exhausted.value) + + assertTrue(p.advance(key, r1, scope)) + // The very first page asks `until = floor`. + assertEquals(p.floorFor(key), p.requestedUntilFor(key, r1)) + + // Empty page + EOSE → that relay is done; the only relay is done → genuinely caught up. + assertTrue(p.onEose(key, r1)) + assertTrue( + p.relayProgress.value + .getValue(r1) + .done, + ) + assertTrue(p.exhausted.value) + assertEquals(0, p.stalledCount.value) + } + + @Test + fun nonEmptyPageMovesTheCursorThenBottomsOut() { + val p = pagerOf(r1) + p.advance(key, r1, scope) + + // A page of three events; the oldest is 80, so the reached cursor drops to 80 (not done). + p.onEvent(key, r1, 100) + p.onEvent(key, r1, 80) + p.onEvent(key, r1, 90) + assertFalse(p.onEose(key, r1)) + assertFalse( + p.relayProgress.value + .getValue(r1) + .done, + ) + assertEquals(80L, p.reachedBack.value) + assertFalse(p.exhausted.value) + + // The next page must start strictly below the oldest reached (80 → until 79). + assertTrue(p.advance(key, r1, scope)) + assertEquals(79L, p.requestedUntilFor(key, r1)) + + // Empty page now → done → caught up. + assertTrue(p.onEose(key, r1)) + assertTrue(p.exhausted.value) + assertEquals(0, p.stalledCount.value) + } + + @Test + fun aStalledRelayMakesExhaustionIncompleteNotCaughtUp() { + val p = pagerOf(r1, r2) + p.advance(key, r1, scope) + p.advance(key, r2, scope) + + // r1 genuinely bottoms out; r2 is still pending, so not exhausted yet. + p.onEose(key, r1) + assertFalse(p.exhausted.value) + + // r2 auth-walls the REQ → stalled (kept, not done). + p.onClosed(key, r2, "auth-required") + assertTrue( + p.relayProgress.value + .getValue(r2) + .stalled, + ) + assertFalse( + p.relayProgress.value + .getValue(r2) + .done, + ) + + // Every relay is now done-or-stalled → exhausted, but it is INCOMPLETE: one relay unreachable. + assertTrue(p.exhausted.value) + assertEquals(1, p.stalledCount.value) + } + + @Test + fun cannotConnectAlsoStalls() { + val p = pagerOf(r1) + p.advance(key, r1, scope) + p.onCannotConnect(key, r1, "offline") + assertTrue( + p.relayProgress.value + .getValue(r1) + .stalled, + ) + assertTrue(p.exhausted.value) + assertEquals(1, p.stalledCount.value) + } + + @Test + fun reAdvancingAStalledRelayClearsTheStallAndUnExhausts() { + val p = pagerOf(r1) + p.advance(key, r1, scope) + p.onClosed(key, r1, "auth-required") + assertTrue(p.exhausted.value) + assertEquals(1, p.stalledCount.value) + + // Retrying it re-arms the relay: no longer stalled, no longer exhausted. + assertTrue(p.advance(key, r1, scope)) + assertFalse( + p.relayProgress.value + .getValue(r1) + .stalled, + ) + assertFalse(p.exhausted.value) + assertEquals(0, p.stalledCount.value) + } + + @Test + fun reachedBackIsTheDeepestCursorAcrossRelays() { + val p = pagerOf(r1, r2) + p.advance(key, r1, scope) + p.advance(key, r2, scope) + + p.onEvent(key, r1, 500) + p.onEose(key, r1) // r1 reached 500 + + p.onEvent(key, r2, 300) + p.onEose(key, r2) // r2 reached 300 + + // Deepest = the oldest point any relay has reached. + assertEquals(300L, p.reachedBack.value) + } + + @Test + fun aDoneRelayWillNotAdvanceAgain() { + val p = pagerOf(r1) + p.advance(key, r1, scope) + p.onEose(key, r1) // empty → done + assertFalse(p.advance(key, r1, scope)) + } + + @Test + fun advanceAllArmsEveryNotDoneRelay() { + val p = pagerOf(r1, r2, r3) + // r2 already finished; advanceAll should arm only r1 and r3. + p.advance(key, r2, scope) + p.onEose(key, r2) + + assertTrue(p.advanceAll(key, scope)) + assertEquals(setOf(r1, r3), p.armedRelays(key, listOf(r1, r2, r3)).toSet()) + } + + @Test + fun switchingActiveKeyRepointsTheDisplayFlows() { + val keyA = "a" + val keyB = "b" + val relaysByKey = mapOf(keyA to listOf(r1), keyB to listOf(r2)) + val p = BackwardRelayPager("test") { relaysByKey[it] } + + p.activate(keyA) + p.advance(keyA, r1, scope) + p.onClosed(keyA, r1, "auth-required") // A: exhausted + 1 stalled + assertTrue(p.exhausted.value) + assertEquals(1, p.stalledCount.value) + + // Switching to a fresh key B repoints the flows to B's own state: nothing stalled, and its + // reach sits at B's floor (no history fetched yet — the markers start at the live-tail boundary). + p.activate(keyB) + assertFalse(p.exhausted.value) + assertEquals(0, p.stalledCount.value) + assertEquals(p.floorFor(keyB), p.reachedBack.value) + + // Switching back to A restores its remembered terminal state. + p.activate(keyA) + assertTrue(p.exhausted.value) + assertEquals(1, p.stalledCount.value) + } +} diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/UntilLimitPagingRelayTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/UntilLimitPagingRelayTest.kt new file mode 100644 index 0000000000..d422ec580a --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/UntilLimitPagingRelayTest.kt @@ -0,0 +1,100 @@ +/* + * 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.paging + +import com.vitorpamplona.geode.fixtures.SyntheticEvents +import com.vitorpamplona.geode.testing.RelayClientTest +import com.vitorpamplona.geode.testing.collectUntilEose +import com.vitorpamplona.geode.testing.preload +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * Pins down the relay-side contract the whole [UntilLimitPager] / [BackwardRelayPager] design rests on, + * against the in-process relay: a backward `until`+`limit` walk returns each event **exactly once** + * (no re-download), in **newest-first** capped pages, and an **empty page + EOSE** is the gap-proof + * stop. If a relay ever stopped honouring this (e.g. oldest-first, or ignoring `until`), these break — + * which is exactly the signal the pager's correctness depends on. + */ +class UntilLimitPagingRelayTest : RelayClientTest() { + @Test + fun backwardUntilLimitWalkCoversEveryEventOnceAndStopsOnEmptyPage() = + runBlocking { + // 250 regular events, createdAt 1..250 (distinct pubkeys so none collapse). + defaultRelay.preload(SyntheticEvents.batch(TOTAL, kind = KIND)) + + val seenIds = mutableSetOf() + var totalReceived = 0 + var pages = 0 + var until: Long? = null + + while (pages < SAFETY_CAP) { + val (events, eose) = + client.collectUntilEose( + defaultRelayUrl, + Filter(kinds = listOf(KIND), until = until, limit = LIMIT), + ) + assertTrue(eose, "every page must end with EOSE") + + if (events.isEmpty()) break // gap-proof stop: empty page = nothing older + + pages++ + assertTrue(events.size <= LIMIT, "page must respect the limit") + // Newest-first + cursor honoured: nothing newer than the cursor leaks into a later page. + until?.let { cursor -> assertTrue(events.all { it.createdAt <= cursor }, "page must be older than the cursor") } + + events.forEach { e: Event -> + seenIds.add(e.id) + totalReceived++ + } + until = events.minOf { it.createdAt } - 1 + } + + // No re-download: total delivered equals the corpus, and every id is distinct. + assertEquals(TOTAL, totalReceived, "no event should be delivered twice across pages") + assertEquals(TOTAL, seenIds.size, "every event fetched exactly once") + // 250 / 100 → 100 + 100 + 50, then an empty page stops the walk. + assertEquals(3, pages) + } + + @Test + fun anEmptyRelayAnswersOneEmptyPageWithEose() = + runBlocking { + val (events, eose) = + client.collectUntilEose( + defaultRelayUrl, + Filter(kinds = listOf(KIND), until = null, limit = LIMIT), + ) + assertTrue(eose) + assertEquals(0, events.size) + } + + companion object { + private const val KIND = 1 + private const val TOTAL = 250 + private const val LIMIT = 100 + private const val SAFETY_CAP = 10 + } +}