diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt index 6b2cd4be0b..838dd235d8 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt @@ -50,13 +50,18 @@ import kotlin.time.Duration.Companion.seconds * * So completion is **activity-based**: [loading] stays true until either every expected relay has * EOSE'd, or the event stream has gone quiet for [idleTimeout] (a flood of events keeps resetting - * that timer via [onActivity], so a window that is still streaming is never declared done). An - * [absoluteCap] bounds the wait for pathological relays that dribble forever. + * that timer via [onActivity], so a window that is still streaming is never declared done). The idle + * path only arms **after the first event or EOSE** — a cold boot whose relays have not finished + * connecting yet is silent for reasons that have nothing to do with the data, and declaring it "done" + * there would empty the screen and trip the auto-fill into widening over and over. Until that first + * sign of life, only [noResponseTimeout] (a generous "nothing answered at all" bound) and the + * [absoluteCap] (for pathological relays that dribble forever) can end the load. */ class WindowLoadTracker( // Short label for the DMPagination logs (e.g. "giftwrap", "rooms.nip04", "convo.nip04"). private val name: String = "dm", private val idleTimeout: Duration = 3.seconds, + private val noResponseTimeout: Duration = 30.seconds, private val absoluteCap: Duration = 5.minutes, ) { private val _loading = MutableStateFlow(true) @@ -76,12 +81,24 @@ class WindowLoadTracker( @Volatile private var lastActivityMs = 0L + // When the current load started, and whether anything (event or EOSE) has arrived for it yet. + // Until the first sign of life the idle timer is meaningless (the relays may still be connecting), + // so completion falls back to the longer [noResponseTimeout]. Volatile for the lock-free hot path. + @Volatile + private var loadStartedMs = 0L + + @Volatile + private var sawActivity = false + /** Begins a fresh window load: clears the responded set, raises [loading], and arms the watchdog. */ @Synchronized fun startLoading(scope: CoroutineScope) { val gen = ++generation responded.clear() - lastActivityMs = System.currentTimeMillis() + val now = System.currentTimeMillis() + lastActivityMs = now + loadStartedMs = now + sawActivity = false val wasLoading = _loading.value _loading.value = true Log.d(TAG) { "[$name] load start" + if (!wasLoading) "" else " (restart)" } @@ -97,7 +114,7 @@ class WindowLoadTracker( } // One watchdog poll. Returns false (stop polling) when this watchdog has been superseded by a - // newer load, the window already finished, or the idle/cap deadline is reached. Synchronized so + // newer load, the window already finished, or a completion deadline is reached. Synchronized so // the generation/loading checks and the completion are atomic against startLoading/finish. @Synchronized private fun tick( @@ -106,9 +123,19 @@ class WindowLoadTracker( deadline: Long, ): Boolean { if (gen != generation || !_loading.value) return false - if (now - lastActivityMs >= idleTimeout.inWholeMilliseconds) { - finish("idle") - return false + if (sawActivity) { + // The stream started and then went quiet: the relays are done sending. + if (now - lastActivityMs >= idleTimeout.inWholeMilliseconds) { + finish("idle") + return false + } + } else { + // Nothing has answered yet. Don't mistake a slow cold-boot connect for "done"; only give + // up once even the generous no-response bound has elapsed. + if (now - loadStartedMs >= noResponseTimeout.inWholeMilliseconds) { + finish("no response") + return false + } } if (now >= deadline) { finish("cap") @@ -119,10 +146,12 @@ class WindowLoadTracker( /** * Records that the current window is still actively receiving events (stored OR live). Keeps the - * idle watchdog from completing while a relay is mid-flood. Lock-free: just bumps a timestamp. + * idle watchdog from completing while a relay is mid-flood, and marks that the stream has started + * (arming the idle path). Lock-free: just bumps a timestamp and a flag. */ fun onActivity() { lastActivityMs = System.currentTimeMillis() + sawActivity = true } /** Records which relays the current REQ was sent to. Completes immediately if there are none. */ @@ -140,6 +169,7 @@ class WindowLoadTracker( @Synchronized fun onRelayResponded(relay: NormalizedRelayUrl) { lastActivityMs = System.currentTimeMillis() + sawActivity = true responded.add(relay) if (expected.isNotEmpty() && responded.containsAll(expected)) finish("all relays") } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsEoseManager.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsEoseManager.kt index 1dbd551e81..ad8989a9b0 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsEoseManager.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsEoseManager.kt @@ -98,11 +98,51 @@ class AccountGiftWrapsEoseManager( private val eventsThisLoad = AtomicInteger(0) private val outOfWindowThisLoad = AtomicInteger(0) + // The single tracker is shared across every account, so the summary collector is launched once + // (not per newSub) — otherwise a second logged-in account would double every summary line. + @Volatile + private var summaryJob: Job? = null + private fun countEvent(createdAt: Long) { eventsThisLoad.incrementAndGet() if (createdAt < loadSince) outOfWindowThisLoad.incrementAndGet() } + /** + * Starts a window load and resets the per-load counters in the same breath (synchronously, before + * [WindowLoadTracker.startLoading] raises `loading`, so no in-flight event is counted against the + * wrong load). The summary is emitted by a single collector that logs on each load's falling edge. + */ + private fun beginWindowLoad( + user: User, + scope: CoroutineScope, + ) { + loadSince = windowFor(user).since + eventsThisLoad.set(0) + outOfWindowThisLoad.set(0) + ensureSummaryLogger(scope) + windowLoad.startLoading(scope) + } + + private fun ensureSummaryLogger(scope: CoroutineScope) { + if (summaryJob?.isActive == true) return + summaryJob = + scope.launch { + var wasLoading = false + windowLoad.loading.collect { loading -> + if (!loading && wasLoading) { + val total = eventsThisLoad.get() + val outOfWindow = outOfWindowThisLoad.get() + Log.d(TAG) { + "[giftwrap] load summary: $total event(s), $outOfWindow before floor " + + "(since=$loadSince, ${daysAgo(loadSince)}d back)" + } + } + wasLoading = loading + } + } + } + override fun updateFilter( key: AccountQueryState, since: SincePerRelayMap?, @@ -131,7 +171,7 @@ class AccountGiftWrapsEoseManager( window.loadMore() _exhausted.value = window.isExhausted() Log.d(TAG) { "[giftwrap] loadMore ${daysAgo(before)}d -> ${daysAgo(window.since)}d back (exhausted=${_exhausted.value})" } - scope?.let { windowLoad.startLoading(it) } + scope?.let { beginWindowLoad(user, it) } invalidateFilters() } @@ -142,7 +182,7 @@ class AccountGiftWrapsEoseManager( window.loadAll() _exhausted.value = true Log.d(TAG) { "[giftwrap] loadEverything — full history (${daysAgo(window.since)}d back)" } - scope?.let { windowLoad.startLoading(it) } + scope?.let { beginWindowLoad(user, it) } invalidateFilters() } @@ -152,7 +192,7 @@ class AccountGiftWrapsEoseManager( override fun newSub(key: AccountQueryState): Subscription { val user = user(key) scope = key.account.scope - windowLoad.startLoading(key.account.scope) + beginWindowLoad(user, key.account.scope) userJobMap[user]?.forEach { it.cancel() } userJobMap[user] = listOf( @@ -160,26 +200,6 @@ class AccountGiftWrapsEoseManager( key.account.dmRelays.flow .collectLatest { invalidateFilters() } }, - // Resets the per-load counters when a load begins and logs the tally when it ends, so - // the trail shows how many events each widen pulled and how many were below the floor. - key.account.scope.launch { - var wasLoading = false - windowLoad.loading.collect { loading -> - if (loading && !wasLoading) { - loadSince = windowFor(user).since - eventsThisLoad.set(0) - outOfWindowThisLoad.set(0) - } else if (!loading && wasLoading) { - val total = eventsThisLoad.get() - val outOfWindow = outOfWindowThisLoad.get() - Log.d(TAG) { - "[giftwrap] load summary: $total event(s), $outOfWindow before floor " + - "(since=$loadSince, ${daysAgo(loadSince)}d back)" - } - } - wasLoading = loading - } - }, ) return requestNewSubscription(