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

fix: don't declare a window load done while relays are still connecting

Browse Source 
A cold boot trace showed `[giftwrap] load done: idle` firing with 0 events at
+3s while the DM relays had not even connected yet (nos.lol first connected at
+14s). The idle watchdog could not tell "quiet because the relays answered"
from "quiet because nothing has connected", so a slow boot looked finished with
an empty result — and with the Messages screen open that false "done, 0 events"
would trip the auto-fill into widening the window over and over.

WindowLoadTracker now only arms the idle path after the first event or EOSE
(sawActivity). Before that first sign of life, the load can only end via a
generous no-response bound (30s) or the absolute cap, so a still-connecting
boot stays "loading" instead of falsely completing empty. The clean paths are
unchanged: relays that EOSE complete via "all relays", and a stream that starts
then quiets still completes via "idle".

Also makes the gift-wrap load-summary collector a singleton: the tracker is
shared across accounts, so launching it per newSub double-logged every summary
when a second account was logged in. Per-load counters are now reset
synchronously at load start (beginWindowLoad) rather than on the collector's
rising edge, so no in-flight event is counted against the wrong load.

https://claude.ai/code/session_01B1fmmmX8JjQWH3amMLdvcW
This commit is contained in:
Claude
2026-06-01 22:24:16 +00:00
parent 923ad500a7
commit 5fb0cc9dd4
2 changed files with 81 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/WindowLoadTracker.kt
+38 -8
View File
@@ -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")
}
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip59GiftWraps/AccountGiftWrapsEoseManager.kt
+43 -23
View File
@@ -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(