laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-10 08:27:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(dm): wire the three history managers onto BackwardRelayPager + tests

Browse Source 
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.
This commit is contained in:
Claude
2026-06-05 16:23:30 +00:00
parent 7ef6224c19
commit 1c7e073b48
5 changed files with 405 additions and 481 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsHistoryEoseManager.kt
+29 -171
View File
@@ -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<AccountQueryState>(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<HexKey>()
// 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<HexKey, Account>()
// 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<HexKey, MutableSet<NormalizedRelayUrl>>()
// Per-relay demand-driven paging, keyed by account pubkey so switching accounts preserves progress.
private val pager =
BackwardRelayPager<HexKey>("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<HexKey, Boolean>()
private val loadTracker = PerRelayLoadTracker("giftwrap.history", onSilenced = ::onRelaysSilenced)
val loadingMore: StateFlow<Boolean> = loadTracker.loading
private val _exhausted = MutableStateFlow(false)
val exhausted: StateFlow<Boolean> = _exhausted.asStateFlow()
// Relays currently fetching a page (for the "asking N relays" status line).
private val _relayCount = MutableStateFlow(0)
val relayCount: StateFlow<Int> = _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<Int> = _stalledCount.asStateFlow()
private val _reachedBack = MutableStateFlow<Long?>(null)
val reachedBack: StateFlow<Long?> = _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<Map<NormalizedRelayUrl, RelayPagingProgress>>(emptyMap())
val relayProgress: StateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>> = _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<HexKey, Long>()
private fun startUntil(pk: HexKey) = pinnedFloor.getOrPut(pk) { TimeUtils.now() - AccountGiftWrapsEoseManager.LIVE_TAIL_SECONDS }
val loadingMore: StateFlow<Boolean> = pager.loadingMore
val exhausted: StateFlow<Boolean> = pager.exhausted
val relayCount: StateFlow<Int> = pager.relayCount
val stalledCount: StateFlow<Int> = pager.stalledCount
val reachedBack: StateFlow<Long?> = pager.reachedBack
val relayProgress: StateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>> = 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<NormalizedRelayUrl>) {
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<Filter>?,
) {
loadTracker.onActivity()
pager.onEvent(user.pubkeyHex, relay, event.createdAt)
stalledRelays[user.pubkeyHex]?.remove(relay)
}
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
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<Filter>?,
) {
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<Filter>?,
) {
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
}
}
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/privateDM/datasource/ChatroomNip04HistorySubAssembler.kt
+24 -156
View File
@@ -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<ConvoKey>()
// The conversation's relay set for a key, resolved via the outbox model (per-relay scoped).
private fun relaysFor(pk: ConvoKey): Collection<NormalizedRelayUrl>? = allKeys().firstOrNull { convoKey(it) == pk }?.let { nip04DMRelays(it.room.users, it.account)?.all }
private val stalledRelays = ConcurrentHashMap<ConvoKey, MutableSet<NormalizedRelayUrl>>()
private val pager = BackwardRelayPager<ConvoKey>("convo.nip04.history", relaysFor = ::relaysFor)
private val loadTracker = PerRelayLoadTracker("convo.nip04.history", onSilenced = ::onRelaysSilenced)
val loadingMore: StateFlow<Boolean> = loadTracker.loading
private val _exhausted = MutableStateFlow(false)
val exhausted: StateFlow<Boolean> = _exhausted.asStateFlow()
private val _relayCount = MutableStateFlow(0)
val relayCount: StateFlow<Int> = _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<Int> = _stalledCount.asStateFlow()
private val _reachedBack = MutableStateFlow<Long?>(null)
val reachedBack: StateFlow<Long?> = _reachedBack.asStateFlow()
private val _relayProgress = MutableStateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>>(emptyMap())
val relayProgress: StateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>> = _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<ConvoKey, Boolean>()
// 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<ConvoKey, Long>()
private fun startUntil(pk: ConvoKey) = pinnedFloor.getOrPut(pk) { TimeUtils.now() - AccountGiftWrapsEoseManager.LIVE_TAIL_SECONDS }
val loadingMore: StateFlow<Boolean> = pager.loadingMore
val exhausted: StateFlow<Boolean> = pager.exhausted
val relayCount: StateFlow<Int> = pager.relayCount
val stalledCount: StateFlow<Int> = pager.stalledCount
val reachedBack: StateFlow<Long?> = pager.reachedBack
val relayProgress: StateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>> = 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<NormalizedRelayUrl>) {
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<Filter>?,
) {
loadTracker.onActivity()
pager.onEvent(pk, relay, event.createdAt)
stalledRelays[pk]?.remove(relay)
}
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
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<Filter>?,
) {
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<Filter>?,
) {
loadTracker.onSettled(relay)
markStalled(pk, relay, "cannot connect: $message")
updateStatus()
recomputeExhausted()
pager.onCannotConnect(pk, relay, message)
}
}
}
companion object {
private const val PAGE_LIMIT = 10000
}
}
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/rooms/datasource/ChatroomListNip04HistorySubAssembler.kt
+28 -154
View File
@@ -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<ChatroomListState>,
) : PerUserEoseManager<ChatroomListState>(client, allKeys) {
private val pager = UntilLimitPager<HexKey>()
private val accounts = ConcurrentHashMap<HexKey, Account>()
private val stalledRelays = ConcurrentHashMap<HexKey, MutableSet<NormalizedRelayUrl>>()
@Volatile
private var activeUser: HexKey? = null
private val exhaustedByUser = ConcurrentHashMap<HexKey, Boolean>()
private val loadTracker = PerRelayLoadTracker("rooms.nip04.history", onSilenced = ::onRelaysSilenced)
val loadingMore: StateFlow<Boolean> = loadTracker.loading
private val _exhausted = MutableStateFlow(false)
val exhausted: StateFlow<Boolean> = _exhausted.asStateFlow()
private val _relayCount = MutableStateFlow(0)
val relayCount: StateFlow<Int> = _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<Int> = _stalledCount.asStateFlow()
private val _reachedBack = MutableStateFlow<Long?>(null)
val reachedBack: StateFlow<Long?> = _reachedBack.asStateFlow()
private val _relayProgress = MutableStateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>>(emptyMap())
val relayProgress: StateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>> = _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<HexKey, Long>()
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<HexKey>("rooms.nip04.history") { pk ->
accounts[pk]?.let { allRelays(it) }
}
val loadingMore: StateFlow<Boolean> = pager.loadingMore
val exhausted: StateFlow<Boolean> = pager.exhausted
val relayCount: StateFlow<Int> = pager.relayCount
val stalledCount: StateFlow<Int> = pager.stalledCount
val reachedBack: StateFlow<Long?> = pager.reachedBack
val relayProgress: StateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>> = 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<NormalizedRelayUrl>) {
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<Filter>?,
) {
loadTracker.onActivity()
pager.onEvent(user.pubkeyHex, relay, event.createdAt)
stalledRelays[user.pubkeyHex]?.remove(relay)
}
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
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<Filter>?,
) {
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<Filter>?,
) {
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
}
}
quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/BackwardRelayPagerTest.kt
+224
View File
@@ -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<String> = BackwardRelayPager<String>("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<String>("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)
}
}
quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/UntilLimitPagingRelayTest.kt
+100
View File
@@ -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<String>()
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
}
}