mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-12 09:13:23 +00:00
refactor(dm): extract per-relay paging primitives to quartz + add BackwardRelayPager
Step 1+2 of generalizing the DM history pagination into a reusable toolkit. Move the four transport-agnostic primitives out of the amethyst module into quartz's jvmAndroid source set, new package `nip01Core.relay.client.paging`: UntilLimitPager, PerRelayLoadTracker, WindowLoadTracker, RelayPagingProgress. They were already pure Kotlin (no Android deps); jvmAndroid keeps their java.util.concurrent / @Synchronized concurrency without a KMP-atomics rewrite, while making them visible to amethyst, desktop, and quartz's jvmAndroidTest (geode in-process relay) for the integration tests to come. Add BackwardRelayPager<K>: the generic per-relay backward-pagination engine that collapses the ~80%-identical pager+tracker+status+exhausted bookkeeping the three DM history loaders each reimplement. It owns the cursors, in-flight + silence tracking, stalled set, pinned floor, and the display StateFlows (relayProgress / exhausted / reachedBack / relayCount / stalledCount); the caller supplies only the filter builder, the subscription wiring, and a relaysFor(key) lookup. Not yet wired into the managers — that swap is step 3. Pure relocation + new component; no behavior change. UntilLimitPagerTest and WindowLoadTrackerSilenceTest stay in amethyst (they use JUnit) with explicit imports added, and both still pass against the relocated classes.
This commit is contained in:
Claude
+331
@@ -0,0 +1,331 @@
|
||||
/*
|
||||
* 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 com.vitorpamplona.quartz.utils.Log
|
||||
import com.vitorpamplona.quartz.utils.TimeUtils
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* Reusable **per-relay backward pagination** engine: pages a set of relays back through history,
|
||||
* **one page at a time, per relay, on demand**, by `until`+`limit` ([UntilLimitPager]) — with each
|
||||
* relay advancing independently the moment *it* settles, never paced by the slowest one. This is the
|
||||
* generic core extracted from the DM history loaders (gift-wrap, conversation NIP-04, rooms-list
|
||||
* NIP-04), which were ~80% identical; any feed that wants demand-driven, gap-proof, per-relay history
|
||||
* paging can build one of these instead of re-deriving the cursor/stall/exhausted bookkeeping.
|
||||
*
|
||||
* What it owns: the per-relay cursors ([UntilLimitPager]), the in-flight + silence tracking
|
||||
* ([PerRelayLoadTracker]), the stalled-relay set, the per-key "exhausted" memo, the session-pinned
|
||||
* history floor, and the display [StateFlow]s ([relayProgress], [exhausted], [reachedBack],
|
||||
* [relayCount], [stalledCount]).
|
||||
*
|
||||
* What it does NOT own (the caller supplies these — they are protocol- and framework-specific):
|
||||
* - **Building the actual REQ filters.** The caller reads [armedRelays] + [requestedUntilFor] and
|
||||
* assembles its own `RelayBasedFilter`s (the kinds / authors / `#p` tags differ per feed).
|
||||
* - **The subscription lifecycle.** The caller wires its `INostrClient` subscription and forwards
|
||||
* relay callbacks here via [onEvent] / [onEose] / [onClosed] / [onCannotConnect], then re-issues
|
||||
* its filter (e.g. `invalidateFilters()`) after [advance] / [advanceAll] return true.
|
||||
* - **Which relays a key fans out to.** Supplied once as [relaysFor]; the engine reads it whenever it
|
||||
* needs the active key's relay set (status recompute, exhaustion, membership checks).
|
||||
*
|
||||
* ### Keying ([K])
|
||||
* State is partitioned by an opaque key [K] — e.g. an account pubkey, or `(account, conversation)` —
|
||||
* so several independently-paged scopes can share one engine without leaking cursors across them, and
|
||||
* switching the on-screen scope just repoints the display flows ([activate]) instead of resetting
|
||||
* progress. Exactly one key is "active" (its state is mirrored into the display flows) at a time.
|
||||
*
|
||||
* ### Done vs stalled (read [exhausted] with care)
|
||||
* A relay is **done** once it answers an empty page (gap-proof: nothing older). A relay that won't
|
||||
* answer right now — auth CLOSE, unreachable, or silent past the tracker's window — is flagged
|
||||
* **stalled** but kept (its subscription stays open; re-[advance] retries it). [exhausted] flips true
|
||||
* once every relay is *done or stalled* — "nothing more reachable right now", which is NOT the same as
|
||||
* "fully caught up". Callers that render a terminal state should split on [stalledCount]: `exhausted &&
|
||||
* stalledCount == 0` is genuinely caught up; `exhausted && stalledCount > 0` stopped early and may be
|
||||
* missing messages.
|
||||
*
|
||||
* Not internally synchronized beyond the primitives it composes; intended to be driven from one owning
|
||||
* scope with relay callbacks serialized per relay (as the relay IO layer delivers them).
|
||||
*/
|
||||
class BackwardRelayPager<K>(
|
||||
// Short label for the DMPagination logs (e.g. "giftwrap.history", "convo.nip04.history").
|
||||
private val name: String,
|
||||
// Asked of every relay per page; large on purpose (a whole band in one page), and caps per-request
|
||||
// volume. A relay returning fewer is its own cap, NOT exhaustion — only an empty page ends a relay.
|
||||
val pageLimit: Int = DEFAULT_PAGE_LIMIT,
|
||||
// How far below "now" the history floor sits — paging starts here and walks backward. Defaults to
|
||||
// the one-week live-tail boundary: everything newer is the always-on tail's job.
|
||||
private val liveTailSeconds: Long = DEFAULT_LIVE_TAIL_SECONDS,
|
||||
// The relay set a key currently fans out to. Read on every status/exhaustion recompute, so it must
|
||||
// reflect the key's live relay list. Null/empty means "no relays known yet" (no-op).
|
||||
private val relaysFor: (K) -> Collection<NormalizedRelayUrl>?,
|
||||
) {
|
||||
private val pager = UntilLimitPager<K>()
|
||||
private val loadTracker = PerRelayLoadTracker(name, onSilenced = ::onSilenced)
|
||||
|
||||
// Relays not currently advancing for a key (auth CLOSE / unreachable / silent). Kept (not given up)
|
||||
// and surfaced as stalled; they resume if the key re-advances them.
|
||||
private val stalledRelays = ConcurrentHashMap<K, MutableSet<NormalizedRelayUrl>>()
|
||||
|
||||
// Per-key exhausted memo, so a backgrounded key keeps its terminal state and switching back to it
|
||||
// restores the right flag instead of flashing "loading".
|
||||
private val exhaustedByKey = ConcurrentHashMap<K, Boolean>()
|
||||
|
||||
// History starts just below the live-tail floor and pages backward. Pinned per key 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.
|
||||
private val pinnedFloor = ConcurrentHashMap<K, Long>()
|
||||
|
||||
// The key whose state is currently mirrored into the display flows (the one on screen). A background
|
||||
// key's late EOSE still advances its cursors in [pager] but must not overwrite the display flows.
|
||||
@Volatile
|
||||
private var activeKey: K? = null
|
||||
|
||||
/** True while any relay is mid-page. Starts false (an idle engine isn't "loading"). */
|
||||
val loadingMore: StateFlow<Boolean> = loadTracker.loading
|
||||
|
||||
private val _exhausted = MutableStateFlow(false)
|
||||
|
||||
/** Nothing more reachable right now: every relay is done or stalled. See class doc — not "caught up". */
|
||||
val exhausted: StateFlow<Boolean> = _exhausted.asStateFlow()
|
||||
|
||||
private val _relayCount = MutableStateFlow(0)
|
||||
|
||||
/** Relays currently fetching a page (for an "asking N relays" status line). */
|
||||
val relayCount: StateFlow<Int> = _relayCount.asStateFlow()
|
||||
|
||||
private val _stalledCount = MutableStateFlow(0)
|
||||
|
||||
/** Not-done relays that can't be reached right now (auth CLOSE / unreachable / silent). */
|
||||
val stalledCount: StateFlow<Int> = _stalledCount.asStateFlow()
|
||||
|
||||
private val _reachedBack = MutableStateFlow<Long?>(null)
|
||||
|
||||
/** Oldest `createdAt` reached across all relays (the deepest cursor), or null before any delivery. */
|
||||
val reachedBack: StateFlow<Long?> = _reachedBack.asStateFlow()
|
||||
|
||||
private val _relayProgress = MutableStateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>>(emptyMap())
|
||||
|
||||
/** Per-relay window position (reached / done / stalled) — the data on-screen reach markers render. */
|
||||
val relayProgress: StateFlow<Map<NormalizedRelayUrl, RelayPagingProgress>> = _relayProgress.asStateFlow()
|
||||
|
||||
/** The session-pinned floor for [key] — where its paging starts (just below the live tail). */
|
||||
fun floorFor(key: K): Long = pinnedFloor.getOrPut(key) { TimeUtils.now() - liveTailSeconds }
|
||||
|
||||
// --- Filter building support: the caller assembles the actual REQ from these. ---
|
||||
|
||||
/** Relays of [key] that have been advanced (armed) and aren't done — i.e. that should carry a REQ. */
|
||||
fun armedRelays(
|
||||
key: K,
|
||||
relays: Collection<NormalizedRelayUrl>,
|
||||
): List<NormalizedRelayUrl> = pager.armedRelays(key, relays)
|
||||
|
||||
/** The `until` [relay]'s next page should carry for [key] (null if it isn't armed). */
|
||||
fun requestedUntilFor(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
): Long? = pager.requestedUntilFor(key, relay)
|
||||
|
||||
// --- Demand-driven advance (the caller re-issues its filter when these return true). ---
|
||||
|
||||
/** Steps a single [relay] to its next, older page for [key]. @return true if it actually advanced. */
|
||||
fun advance(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
scope: CoroutineScope,
|
||||
): Boolean {
|
||||
if (!arm(key, relay, scope)) return false
|
||||
if (activeKey == key) _exhausted.value = false
|
||||
updateStatus(key)
|
||||
return true
|
||||
}
|
||||
|
||||
/** Steps every not-done, not-in-flight relay of [key] one page. For a scope too small to scroll. */
|
||||
fun advanceAll(
|
||||
key: K,
|
||||
scope: CoroutineScope,
|
||||
): Boolean {
|
||||
val relays = relaysFor(key) ?: return false
|
||||
var any = false
|
||||
relays.forEach { if (arm(key, it, scope)) any = true }
|
||||
if (any) {
|
||||
if (activeKey == key) _exhausted.value = false
|
||||
updateStatus(key)
|
||||
}
|
||||
return any
|
||||
}
|
||||
|
||||
// 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 recompute status — the caller batches.
|
||||
private fun arm(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
scope: CoroutineScope,
|
||||
): Boolean {
|
||||
val relays = relaysFor(key) ?: return false
|
||||
if (relay !in relays) return false
|
||||
if (loadTracker.isInFlight(relay)) return false
|
||||
if (!pager.advance(key, relay, floorFor(key))) return false
|
||||
stalledRelays[key]?.remove(relay)
|
||||
loadTracker.bind(scope)
|
||||
loadTracker.onAdvance(relay)
|
||||
return true
|
||||
}
|
||||
|
||||
// --- Subscription callbacks: the owner forwards these from its SubscriptionListener. ---
|
||||
|
||||
/** Records one delivered event for [relay] (a sign of life + a page tally entry). */
|
||||
fun onEvent(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
createdAt: Long,
|
||||
) {
|
||||
loadTracker.onActivity()
|
||||
pager.onEvent(key, relay, createdAt)
|
||||
stalledRelays[key]?.remove(relay)
|
||||
}
|
||||
|
||||
/** Finalizes [relay]'s page on EOSE. @return true if this EOSE is the one that marked it done. */
|
||||
fun onEose(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
): Boolean {
|
||||
stalledRelays[key]?.remove(relay)
|
||||
pager.onEose(key, relay)
|
||||
loadTracker.onSettled(relay)
|
||||
val done = pager.isDone(key, relay)
|
||||
updateStatus(key)
|
||||
recomputeExhausted(key)
|
||||
return done
|
||||
}
|
||||
|
||||
/** [relay] rejected the REQ (e.g. auth-required): settle it and flag it stalled (kept, retryable). */
|
||||
fun onClosed(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
message: String,
|
||||
) {
|
||||
loadTracker.onSettled(relay)
|
||||
markStalled(key, relay, "CLOSED: $message")
|
||||
updateStatus(key)
|
||||
recomputeExhausted(key)
|
||||
}
|
||||
|
||||
/** [relay] is unreachable right now: settle it and flag it stalled (kept, retryable). */
|
||||
fun onCannotConnect(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
message: String,
|
||||
) {
|
||||
loadTracker.onSettled(relay)
|
||||
markStalled(key, relay, "cannot connect: $message")
|
||||
updateStatus(key)
|
||||
recomputeExhausted(key)
|
||||
}
|
||||
|
||||
// The tracker's silence watchdog fired: the still-pending relays went quiet after their REQ. Flag the
|
||||
// active key's of them stalled (kept) so the window can settle instead of hanging on a dead relay.
|
||||
private fun onSilenced(relays: Set<NormalizedRelayUrl>) {
|
||||
val key = activeKey ?: return
|
||||
relays.forEach { markStalled(key, it, "no response (silence timeout)") }
|
||||
updateStatus(key)
|
||||
recomputeExhausted(key)
|
||||
}
|
||||
|
||||
private fun markStalled(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
reason: String,
|
||||
) {
|
||||
val firstTime = stalledRelays.getOrPut(key) { ConcurrentHashMap.newKeySet() }.add(relay)
|
||||
if (firstTime) Log.d(TAG) { "[$name] ${relay.url} stalled — $reason (kept, advance to retry)" }
|
||||
}
|
||||
|
||||
// --- Display-flow management. ---
|
||||
|
||||
/**
|
||||
* Repoints the display flows to [key] (call on subscribe / when the on-screen scope changes), then
|
||||
* refreshes them. A no-op repoint (same key) just refreshes. Cursors in [pager] are untouched, so a
|
||||
* previously-paged key restores its progress instead of restarting.
|
||||
*/
|
||||
fun activate(key: K) {
|
||||
if (activeKey != key) {
|
||||
activeKey = key
|
||||
loadTracker.reset()
|
||||
_exhausted.value = exhaustedByKey[key] ?: false
|
||||
_relayCount.value = 0
|
||||
_stalledCount.value = 0
|
||||
_reachedBack.value = null
|
||||
_relayProgress.value = emptyMap()
|
||||
}
|
||||
updateStatus(key)
|
||||
}
|
||||
|
||||
/** Recomputes the display flows from [key]'s cursors. No-op when [key] is not the active key. */
|
||||
fun updateStatus(key: K) {
|
||||
if (activeKey != key) return
|
||||
val relays = relaysFor(key) ?: emptySet()
|
||||
_relayCount.value = loadTracker.count()
|
||||
val floor = floorFor(key)
|
||||
_reachedBack.value = pager.deepestReached(key, relays, floor)
|
||||
val stalled = stalledRelays[key] ?: emptySet()
|
||||
_stalledCount.value = relays.count { it in stalled && !pager.isDone(key, it) }
|
||||
_relayProgress.value =
|
||||
relays.associateWith { relay ->
|
||||
RelayPagingProgress(
|
||||
reachedUntil = pager.reachedUntilFor(key, relay, floor),
|
||||
done = pager.isDone(key, relay),
|
||||
stalled = relay in stalled && !pager.isDone(key, 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(key: K) {
|
||||
val relays = relaysFor(key) ?: return
|
||||
if (relays.isEmpty()) return
|
||||
val stalled = stalledRelays[key] ?: emptySet()
|
||||
val pending = relays.any { !pager.isDone(key, it) && it !in stalled }
|
||||
val ex = !pending
|
||||
val was = exhaustedByKey[key] ?: false
|
||||
exhaustedByKey[key] = ex
|
||||
if (ex && !was) {
|
||||
val done = relays.filter { pager.isDone(key, it) }.map { it.url }
|
||||
val stuck = relays.filter { it in stalled && !pager.isDone(key, it) }.map { it.url }
|
||||
Log.d(TAG) { "[$name] window settled (nothing more reachable) — done=$done stalled=$stuck" }
|
||||
}
|
||||
if (activeKey == key) _exhausted.value = ex
|
||||
}
|
||||
|
||||
companion object {
|
||||
private const val TAG = "DMPagination"
|
||||
|
||||
const val DEFAULT_PAGE_LIMIT = 10000
|
||||
|
||||
// One week — matches the DM live-tail floor (everything newer is the always-on tail's job).
|
||||
const val DEFAULT_LIVE_TAIL_SECONDS = 7L * TimeUtils.ONE_DAY
|
||||
}
|
||||
}
|
||||
+170
@@ -0,0 +1,170 @@
|
||||
/*
|
||||
* 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 com.vitorpamplona.quartz.utils.Log
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* Tracks which relays currently have a demand-driven history page **in flight**, so the loading card
|
||||
* can show a spinner while any relay is fetching and clear it the moment they've all answered (or
|
||||
* parked). Unlike [WindowLoadTracker] this has no notion of a "window" or "round" — relays are advanced
|
||||
* one page at a time, independently, by their on-screen markers, so completion is simply "nothing in
|
||||
* flight."
|
||||
*
|
||||
* A single backstop covers a relay that accepts a REQ and then goes silent (auth-walled / dead): if
|
||||
* nothing has been heard from ANY in-flight relay for [silenceMs], the still-pending relays are dropped
|
||||
* from the in-flight set (so the spinner clears) and reported to [onSilenced] so the owner can flag them
|
||||
* stalled. Relays that answer with CLOSED / cannot-connect are settled directly by the owner and don't
|
||||
* need the watchdog.
|
||||
*/
|
||||
class PerRelayLoadTracker(
|
||||
private val name: String,
|
||||
private val silenceMs: Long = 15_000L,
|
||||
private val onSilenced: (Set<NormalizedRelayUrl>) -> Unit = {},
|
||||
) {
|
||||
private val _loading = MutableStateFlow(false)
|
||||
val loading: StateFlow<Boolean> = _loading.asStateFlow()
|
||||
|
||||
private val inFlight = ConcurrentHashMap.newKeySet<NormalizedRelayUrl>()
|
||||
|
||||
@Volatile
|
||||
private var lastActivityMs = 0L
|
||||
|
||||
@Volatile
|
||||
private var watchdog: Job? = null
|
||||
|
||||
// Delays dropping the spinner so back-to-back pages don't flicker it off between each one.
|
||||
@Volatile
|
||||
private var clearJob: Job? = null
|
||||
|
||||
@Volatile
|
||||
private var scope: CoroutineScope? = null
|
||||
|
||||
fun bind(scope: CoroutineScope) {
|
||||
this.scope = scope
|
||||
}
|
||||
|
||||
fun isInFlight(relay: NormalizedRelayUrl) = inFlight.contains(relay)
|
||||
|
||||
fun count() = inFlight.size
|
||||
|
||||
/** A relay's next page was just requested. Raises the spinner and (re)arms the silence watchdog. */
|
||||
@Synchronized
|
||||
fun onAdvance(relay: NormalizedRelayUrl) {
|
||||
clearJob?.cancel() // a new page is starting — keep the spinner up, no flicker
|
||||
clearJob = null
|
||||
inFlight.add(relay)
|
||||
lastActivityMs = System.currentTimeMillis()
|
||||
_loading.value = true
|
||||
ensureWatchdog()
|
||||
}
|
||||
|
||||
/** A sign of life from a relay (an event). Keeps the silence watchdog from firing. */
|
||||
fun onActivity() {
|
||||
lastActivityMs = System.currentTimeMillis()
|
||||
}
|
||||
|
||||
/**
|
||||
* A relay answered (EOSE / CLOSED / cannot-connect). Drops it from in-flight. When the last one
|
||||
* settles, the spinner is dropped after a short linger rather than immediately, so a relay paging
|
||||
* page-after-page (each page settles then the marker fires the next) keeps a steady spinner instead
|
||||
* of flickering it off for the few ms between pages. The linger is cancelled the moment a new page
|
||||
* starts ([onAdvance]).
|
||||
*/
|
||||
@Synchronized
|
||||
fun onSettled(relay: NormalizedRelayUrl) {
|
||||
lastActivityMs = System.currentTimeMillis()
|
||||
if (inFlight.remove(relay) && inFlight.isEmpty()) scheduleClear()
|
||||
}
|
||||
|
||||
private fun scheduleClear() {
|
||||
clearJob?.cancel()
|
||||
val s = scope
|
||||
if (s == null) {
|
||||
_loading.value = false
|
||||
return
|
||||
}
|
||||
clearJob =
|
||||
s.launch {
|
||||
delay(LOADING_LINGER_MS)
|
||||
synchronized(this@PerRelayLoadTracker) {
|
||||
if (inFlight.isEmpty()) _loading.value = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Drops everything (e.g. account/conversation switched). */
|
||||
@Synchronized
|
||||
fun reset() {
|
||||
inFlight.clear()
|
||||
clearJob?.cancel()
|
||||
clearJob = null
|
||||
_loading.value = false
|
||||
watchdog?.cancel()
|
||||
watchdog = null
|
||||
}
|
||||
|
||||
private fun ensureWatchdog() {
|
||||
if (watchdog?.isActive == true) return
|
||||
val s = scope ?: return
|
||||
watchdog =
|
||||
s.launch {
|
||||
while (isActive) {
|
||||
delay(WATCHDOG_TICK_MS)
|
||||
val silenced =
|
||||
synchronized(this@PerRelayLoadTracker) {
|
||||
if (inFlight.isNotEmpty() && System.currentTimeMillis() - lastActivityMs > silenceMs) {
|
||||
val pending = inFlight.toSet()
|
||||
inFlight.clear()
|
||||
_loading.value = false
|
||||
pending
|
||||
} else {
|
||||
emptySet()
|
||||
}
|
||||
}
|
||||
if (silenced.isNotEmpty()) {
|
||||
Log.d(TAG) { "[$name] silenced (no response ${silenceMs}ms): ${silenced.map { it.url }}" }
|
||||
onSilenced(silenced)
|
||||
}
|
||||
if (inFlight.isEmpty()) break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
companion object {
|
||||
private const val TAG = "DMPagination"
|
||||
private const val WATCHDOG_TICK_MS = 1_000L
|
||||
|
||||
// How long to keep the spinner up after the last page settles, to bridge the gap to the next
|
||||
// back-to-back page so the card doesn't flicker between every page.
|
||||
private const val LOADING_LINGER_MS = 600L
|
||||
}
|
||||
}
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
/** How far back one relay has paged a DM history, for the per-relay progress markers. */
|
||||
data class RelayPagingProgress(
|
||||
// The oldest createdAt this relay has loaded down to (its `until` cursor). The marker sits here and
|
||||
// slides down (older) as the relay pages further back.
|
||||
val reachedUntil: Long,
|
||||
// The relay answered an empty page: it has nothing older, it has reached the bottom of its window.
|
||||
val done: Boolean,
|
||||
// The relay isn't answering right now (auth-walled CLOSE / unreachable / slow). It is NOT abandoned
|
||||
// — its subscription stays open and it keeps trying to catch up — but it isn't currently advancing.
|
||||
val stalled: Boolean,
|
||||
)
|
||||
+188
@@ -0,0 +1,188 @@
|
||||
/*
|
||||
* 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 java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* Backward `until`+`limit` pagination cursor, tracked **independently per relay** (and per [K], e.g.
|
||||
* per account or per conversation), and advanced **on demand** — one page at a time, only when the
|
||||
* owner calls [advance].
|
||||
*
|
||||
* The time-window model can't tell "this relay is empty" from "this is a gap" — a `since`/`until`
|
||||
* slice that returns nothing might just be a quiet stretch with older messages beneath it. Paging by
|
||||
* `until`+`limit` removes that ambiguity: a relay returns its N newest events older than `until`,
|
||||
* **skipping gaps**, so an empty page can only mean there is nothing older.
|
||||
*
|
||||
* Two cursors are kept per relay, deliberately decoupled so a relay never pages further than it was
|
||||
* asked to:
|
||||
* - [requestedUntilFor] — the `until` the relay's REQ currently carries. Moves **only** in [advance].
|
||||
* Leaving it untouched on EOSE is what makes paging demand-driven: a relay that finished a page just
|
||||
* parks at the same filter (no re-REQ) until the owner advances it again.
|
||||
* - reached (see [reachedUntilFor]) — the oldest `created_at` the relay has actually delivered. Moves
|
||||
* on EOSE. This is what the in-stream markers sit at; [advance] starts the next page just below it.
|
||||
*
|
||||
* Stop signal (per relay): an **empty page followed by EOSE** ([onEose] with no events) marks that
|
||||
* relay [done][isDone]. A relay that returns anything — even fewer than the requested limit, since a
|
||||
* relay may cap results below what we asked — is not done.
|
||||
*
|
||||
* Not internally synchronized: per-relay counters are touched on the relay IO threads (one relay's
|
||||
* callbacks are serialized) and read on the owning scope; fields are volatile.
|
||||
*/
|
||||
class UntilLimitPager<K> {
|
||||
private class RelayCursor {
|
||||
// The `until` the REQ carries; null until the relay is first advanced. Moves only in advance().
|
||||
@Volatile var requestedUntil: Long? = null
|
||||
|
||||
// The oldest created_at this relay has delivered; null until its first non-empty page. Moves on
|
||||
// EOSE. The marker sits here and the next page starts just below it.
|
||||
@Volatile var reachedUntil: Long? = null
|
||||
|
||||
// Set once the relay answered an empty page with EOSE: there is nothing older on it.
|
||||
@Volatile var done: Boolean = false
|
||||
|
||||
// Per-page tallies, reset by [advance]: how many events arrived and the oldest among them.
|
||||
@Volatile var pageCount: Int = 0
|
||||
|
||||
@Volatile var pageOldest: Long = Long.MAX_VALUE
|
||||
}
|
||||
|
||||
private val perKey = ConcurrentHashMap<K, ConcurrentHashMap<NormalizedRelayUrl, RelayCursor>>()
|
||||
|
||||
private fun cursorsFor(key: K) = perKey.getOrPut(key) { ConcurrentHashMap() }
|
||||
|
||||
private fun cursor(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
) = cursorsFor(key).getOrPut(relay) { RelayCursor() }
|
||||
|
||||
/** True once [relay] has been [advance]d at least once (so its REQ should be issued). */
|
||||
fun isArmed(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
): Boolean = cursor(key, relay).requestedUntil != null
|
||||
|
||||
/** The `until` [relay]'s REQ currently carries. Only meaningful once [isArmed]. */
|
||||
fun requestedUntilFor(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
): Long? = cursor(key, relay).requestedUntil
|
||||
|
||||
/** The oldest point [relay] has reached (its marker depth), or [start] if it hasn't delivered yet. */
|
||||
fun reachedUntilFor(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
start: Long,
|
||||
): Long = cursor(key, relay).reachedUntil ?: start
|
||||
|
||||
/** True once [relay] answered an empty page with EOSE — nothing older to ask it for. */
|
||||
fun isDone(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
): Boolean = cursor(key, relay).done
|
||||
|
||||
/**
|
||||
* Steps [relay] to its next, older page: points its REQ just below the oldest event it has delivered
|
||||
* (or [start] for its very first page) and clears the page tally. No-op (returns false) if the relay
|
||||
* has already paged to the bottom ([done]). The owner re-issues the REQ after this (invalidateFilters).
|
||||
*/
|
||||
fun advance(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
start: Long,
|
||||
): Boolean {
|
||||
val c = cursor(key, relay)
|
||||
if (c.done) return false
|
||||
c.requestedUntil =
|
||||
if (c.requestedUntil == null) {
|
||||
start
|
||||
} else {
|
||||
(c.reachedUntil ?: start) - 1
|
||||
}
|
||||
c.pageCount = 0
|
||||
c.pageOldest = Long.MAX_VALUE
|
||||
return true
|
||||
}
|
||||
|
||||
/** Records one event for [relay] in the current page. */
|
||||
fun onEvent(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
createdAt: Long,
|
||||
) {
|
||||
val c = cursor(key, relay)
|
||||
c.pageCount++
|
||||
if (createdAt < c.pageOldest) c.pageOldest = createdAt
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalizes [relay] for the page on its EOSE: an empty page marks it [done]; otherwise the reached
|
||||
* cursor drops to the oldest event the page returned. The requested cursor is left alone so the relay
|
||||
* parks until [advance] is called again.
|
||||
*/
|
||||
fun onEose(
|
||||
key: K,
|
||||
relay: NormalizedRelayUrl,
|
||||
) {
|
||||
val c = cursor(key, relay)
|
||||
if (c.pageCount == 0) {
|
||||
c.done = true
|
||||
} else {
|
||||
// The reached cursor must move strictly older every page (the next page asks `until =
|
||||
// reached - 1`). A relay that returns events but none older than we already have — a
|
||||
// misbehaving relay echoing the same newest events — would otherwise pin the cursor and the
|
||||
// on-screen sentinel would re-request the same window forever. Treat that as the bottom.
|
||||
val prev = c.reachedUntil
|
||||
if (prev == null || c.pageOldest < prev) {
|
||||
c.reachedUntil = c.pageOldest
|
||||
} else {
|
||||
c.done = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Relays from [all] that still have older history to ask for: not yet empty-EOSE'd ([done]). */
|
||||
fun activeRelays(
|
||||
key: K,
|
||||
all: Collection<NormalizedRelayUrl>,
|
||||
): List<NormalizedRelayUrl> = all.filterNot { cursor(key, it).done }
|
||||
|
||||
/** Relays from [all] that have been armed (advanced at least once) and are not yet [done]. */
|
||||
fun armedRelays(
|
||||
key: K,
|
||||
all: Collection<NormalizedRelayUrl>,
|
||||
): List<NormalizedRelayUrl> =
|
||||
all.filter {
|
||||
val c = cursor(key, it)
|
||||
c.requestedUntil != null && !c.done
|
||||
}
|
||||
|
||||
/**
|
||||
* The oldest point reached across [relays] — the minimum reached cursor (how far back paging has
|
||||
* gone). Relays that haven't delivered count as [start]. Null when [relays] is empty.
|
||||
*/
|
||||
fun deepestReached(
|
||||
key: K,
|
||||
relays: Collection<NormalizedRelayUrl>,
|
||||
start: Long,
|
||||
): Long? = relays.takeIf { it.isNotEmpty() }?.minOf { cursor(key, it).reachedUntil ?: start }
|
||||
}
|
||||
+319
@@ -0,0 +1,319 @@
|
||||
/*
|
||||
* 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.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.utils.Log
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import kotlin.time.Duration
|
||||
import kotlin.time.Duration.Companion.minutes
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
||||
/**
|
||||
* Tracks when one relay-subscription "window" has finished loading, so callers (the rooms-screen
|
||||
* auto-fill loop) can wait for the WHOLE response instead of declaring victory on the first EOSE.
|
||||
*
|
||||
* A subscription fans a single REQ out to several relays. The first EOSE is a misleading "done"
|
||||
* signal: a fast but near-empty relay can EOSE in milliseconds while the relay that actually holds
|
||||
* the data is still connecting, stuck in an auth handshake, or busy streaming thousands of stored
|
||||
* events. An auto-fill loop driven by the first EOSE — or by a fixed wall-clock timeout — would
|
||||
* widen the window again mid-stream, before the events were even decrypted into rooms, re-issuing
|
||||
* an ever-wider REQ that re-downloads the whole history over and over.
|
||||
*
|
||||
* Completion is therefore **per-relay terminal-state** based, not wall-clock based. A relay is
|
||||
* *settled* once it answers with a terminal signal — an EOSE (stored backfill done), a CLOSED (it
|
||||
* rejected the REQ, e.g. `auth-required`), or a cannot-connect (it is unreachable). The load is done
|
||||
* when **every targeted relay has settled** ([settled] ⊇ [expected]). This is the only signal that
|
||||
* survives the real world: relays connect over a wide spread (tens of seconds on mobile), and some
|
||||
* answer only with CLOSED — a quiet-time heuristic fires in the gap between two relays connecting and
|
||||
* mistakes a half-loaded window for a finished one, which is exactly how a load reports "1 event"
|
||||
* when a hundred are still on the way.
|
||||
*
|
||||
* Three backstops cover misbehaving relays. If every relay we're still waiting on has at least been
|
||||
* *heard from* (any event, EOSE, CLOSED, or cannot-connect) but one streamed events without ever
|
||||
* sending EOSE, an [idleTimeout] of quiet completes the load — the "heard from" gate is what keeps
|
||||
* this from firing in a connection gap. A relay that *received our REQ* ([onReqSent]) but then went
|
||||
* completely silent — no event, no EOSE, no CLOSED — for [silenceTimeout] stops blocking the load: an
|
||||
* auth-walled relay (ditto, paid relays) commonly accepts the REQ and answers nothing, and measuring
|
||||
* from REQ-delivery (not window start) means a slow connect doesn't count against it. Such relays are
|
||||
* reported to [onAbandoned] so the owner can react — drop them from its pager, or keep them and flag
|
||||
* them stalled (the convo history keeps trying). A relay that never even
|
||||
* *receives* its REQ (stuck connecting / reconnecting, so it can neither settle nor go "silent") stops
|
||||
* blocking the round after [connectGrace] from the load start — but it is NOT given up (it may be a
|
||||
* genuinely slow connect), so the owner keeps it and retries it next round. And an [absoluteCap] is
|
||||
* the final ceiling on a window that somehow defeats all of the above.
|
||||
*
|
||||
* The two REQ-aware backstops (silence + connect-grace) only make sense when the owner actually feeds
|
||||
* [onReqSent], so they are gated behind [tracksReqSends]. A tracker that does NOT track REQ sends keeps
|
||||
* the plain settle / idle / cap behavior — otherwise, with an always-empty [reqSentAt], EVERY relay
|
||||
* would look "connect-stalled" after [connectGrace] and the window would complete before its REQs even
|
||||
* went out (e.g. during a slow connect storm), prematurely declaring an empty round done.
|
||||
*/
|
||||
class WindowLoadTracker(
|
||||
// Short label for the DMPagination logs (e.g. "giftwrap", "rooms.nip04", "convo.nip04").
|
||||
private val name: String = "dm",
|
||||
// Whether the owner feeds [onReqSent]; enables the silence + connect-grace backstops. Off by
|
||||
// default so trackers that don't track REQ sends are unaffected by them.
|
||||
private val tracksReqSends: Boolean = false,
|
||||
private val idleTimeout: Duration = 3.seconds,
|
||||
private val silenceTimeout: Duration = 10.seconds,
|
||||
private val connectGrace: Duration = 15.seconds,
|
||||
private val absoluteCap: Duration = 5.minutes,
|
||||
// Invoked when a load finishes with the relays that received a REQ but stayed silent past
|
||||
// [silenceTimeout]. The owner decides what to do — drop them from its pager, or keep them open and
|
||||
// flag them stalled. The tracker itself only stops waiting on them; it does not give them up.
|
||||
private val onAbandoned: (Set<NormalizedRelayUrl>) -> Unit = {},
|
||||
) {
|
||||
private val _loading = MutableStateFlow(true)
|
||||
val loading: StateFlow<Boolean> = _loading.asStateFlow()
|
||||
|
||||
// Relays the current REQ was sent to. Volatile: written on IO (updateFilter), read on the
|
||||
// listener threads and the watchdog.
|
||||
@Volatile
|
||||
private var expected: Set<NormalizedRelayUrl> = emptySet()
|
||||
|
||||
// Relays that have produced any signal at all (event / EOSE / CLOSED / cannot-connect). The idle
|
||||
// backstop only arms once this covers [expected], so a still-connecting relay can't be skipped.
|
||||
private val heardFrom = ConcurrentHashMap.newKeySet<NormalizedRelayUrl>()
|
||||
|
||||
// Relays that reached a terminal signal (EOSE / CLOSED / cannot-connect). When this covers
|
||||
// [expected] the stored backfill is complete on every relay and the load is done.
|
||||
private val settled = ConcurrentHashMap.newKeySet<NormalizedRelayUrl>()
|
||||
|
||||
// When the REQ was actually delivered to each relay (post-connect). The silence backstop measures
|
||||
// from here, not window start, so a slow connect isn't mistaken for a dead relay.
|
||||
private val reqSentAt = ConcurrentHashMap<NormalizedRelayUrl, Long>()
|
||||
|
||||
private var watchdog: Job? = null
|
||||
|
||||
// Incremented on every (re)start so a stale watchdog that wakes right as a new load begins
|
||||
// recognizes it has been superseded and bows out instead of completing the new window.
|
||||
private var generation = 0
|
||||
|
||||
// Wall-clock of the last signal for the current window; the idle backstop completes the window
|
||||
// once this stops advancing for [idleTimeout]. Volatile so the hot per-event path stays lock-free.
|
||||
@Volatile
|
||||
private var lastActivityMs = 0L
|
||||
|
||||
// Wall-clock the current window began; the connect-grace backstop measures from here.
|
||||
@Volatile
|
||||
private var loadStartMs = 0L
|
||||
|
||||
/** Begins a fresh window load: clears the per-relay sets, raises [loading], and arms the watchdog. */
|
||||
@Synchronized
|
||||
fun startLoading(scope: CoroutineScope) {
|
||||
val gen = ++generation
|
||||
expected = emptySet()
|
||||
heardFrom.clear()
|
||||
settled.clear()
|
||||
reqSentAt.clear()
|
||||
val nowMs = System.currentTimeMillis()
|
||||
lastActivityMs = nowMs
|
||||
loadStartMs = nowMs
|
||||
val wasLoading = _loading.value
|
||||
_loading.value = true
|
||||
Log.d(TAG) { "[$name] load start" + if (!wasLoading) "" else " (restart)" }
|
||||
watchdog?.cancel()
|
||||
watchdog =
|
||||
scope.launch {
|
||||
val deadline = System.currentTimeMillis() + absoluteCap.inWholeMilliseconds
|
||||
while (isActive) {
|
||||
delay(IDLE_CHECK_MS)
|
||||
if (!tick(gen, System.currentTimeMillis(), deadline)) break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// One watchdog poll. Returns false (stop polling) when this watchdog has been superseded by a
|
||||
// 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(
|
||||
gen: Int,
|
||||
now: Long,
|
||||
deadline: Long,
|
||||
): Boolean {
|
||||
if (gen != generation || !_loading.value) return false
|
||||
if (expected.isNotEmpty()) {
|
||||
// Once every relay is accounted for — settled, gone silent after its REQ, or stuck before
|
||||
// its REQ even went out — nothing more is coming for this round.
|
||||
if (expected.all { accountedFor(it, now) }) {
|
||||
finish("settled/silent")
|
||||
return false
|
||||
}
|
||||
// Idle backstop: every relay we're still waiting on has at least streamed something (so this
|
||||
// isn't a connection gap) and the stream has gone quiet. Accounted-for relays don't count.
|
||||
val stillWaiting = expected.filterNot { accountedFor(it, now) }
|
||||
if (stillWaiting.all { heardFrom.contains(it) } && now - lastActivityMs >= idleTimeout.inWholeMilliseconds) {
|
||||
finish("idle")
|
||||
return false
|
||||
}
|
||||
}
|
||||
if (now >= deadline) {
|
||||
finish("cap")
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// A relay no longer worth waiting on this round: it reached a terminal signal, went silent after
|
||||
// its REQ, or never even received its REQ within the connect grace.
|
||||
private fun accountedFor(
|
||||
relay: NormalizedRelayUrl,
|
||||
now: Long,
|
||||
): Boolean = settled.contains(relay) || silencedOut(relay, now) || connectStalled(relay, now)
|
||||
|
||||
// A relay that received its REQ but produced no signal at all for [silenceTimeout]. Measured from
|
||||
// REQ-delivery so a slow connect (which has no [reqSentAt] yet) is never counted as silent. These
|
||||
// are reported to [onAbandoned] on finish (accepting a REQ then answering nothing usually means an
|
||||
// auth-walled / dead relay) — but whether to give them up is the owner's call, not the tracker's.
|
||||
private fun silencedOut(
|
||||
relay: NormalizedRelayUrl,
|
||||
now: Long,
|
||||
): Boolean = tracksReqSends && relay !in heardFrom && (reqSentAt[relay]?.let { now - it >= silenceTimeout.inWholeMilliseconds } ?: false)
|
||||
|
||||
// A relay that is still expected but has neither been heard from nor even received its REQ within
|
||||
// [connectGrace] of the load start — i.e. stuck connecting / reconnecting. It stops blocking the
|
||||
// round, but is NOT given up (it may simply be a slow connect): the owner retries it next round.
|
||||
private fun connectStalled(
|
||||
relay: NormalizedRelayUrl,
|
||||
now: Long,
|
||||
): Boolean = tracksReqSends && relay !in heardFrom && !reqSentAt.containsKey(relay) && now - loadStartMs >= connectGrace.inWholeMilliseconds
|
||||
|
||||
/** Records which relays the current REQ was sent to. Completes immediately if there are none. */
|
||||
@Synchronized
|
||||
fun setExpectedRelays(relays: Set<NormalizedRelayUrl>) {
|
||||
expected = relays
|
||||
if (relays.isEmpty()) {
|
||||
finish("no relays")
|
||||
} else if (settled.containsAll(relays)) {
|
||||
finish("all relays")
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Records that the REQ was delivered to [relayUrl] (post-connect). Starts that relay's silence clock.
|
||||
* Ignored for relays outside the current [expected] set (or before it is known).
|
||||
*/
|
||||
@Synchronized
|
||||
fun onReqSent(relayUrl: String) {
|
||||
val relay = expected.firstOrNull { it.url == relayUrl } ?: return
|
||||
reqSentAt.putIfAbsent(relay, System.currentTimeMillis())
|
||||
}
|
||||
|
||||
/** A non-terminal sign of life from [relay] (a stored or live event). Keeps the idle timer alive. */
|
||||
fun onRelayEvent(relay: NormalizedRelayUrl) {
|
||||
heardFrom.add(relay)
|
||||
lastActivityMs = System.currentTimeMillis()
|
||||
}
|
||||
|
||||
/**
|
||||
* A terminal signal from [relay] — EOSE, CLOSED, or cannot-connect. Once every expected relay has
|
||||
* settled the stored backfill is complete and the load finishes.
|
||||
*/
|
||||
@Synchronized
|
||||
fun onRelaySettled(relay: NormalizedRelayUrl) {
|
||||
lastActivityMs = System.currentTimeMillis()
|
||||
heardFrom.add(relay)
|
||||
settled.add(relay)
|
||||
if (expected.isNotEmpty() && settled.containsAll(expected)) finish("all relays")
|
||||
}
|
||||
|
||||
// Idempotent: only the first call after a load actually completes (and logs); later calls no-op.
|
||||
@Synchronized
|
||||
private fun finish(reason: String) {
|
||||
if (!_loading.value) return
|
||||
watchdog?.cancel()
|
||||
watchdog = null
|
||||
// Report the silent relays BEFORE flipping [loading]: the owner reacts to loading=false by
|
||||
// recomputing state from its pager, so its reaction to these has to land first.
|
||||
val abandoned = expected.filterTo(mutableSetOf()) { silencedOut(it, System.currentTimeMillis()) }
|
||||
Log.d(TAG) { "[$name] load done: $reason" + if (abandoned.isEmpty()) "" else " (silent: ${abandoned.map { it.url }})" }
|
||||
if (abandoned.isNotEmpty()) onAbandoned(abandoned)
|
||||
_loading.value = false
|
||||
}
|
||||
|
||||
companion object {
|
||||
private const val TAG = "DMPagination"
|
||||
private const val IDLE_CHECK_MS = 500L
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds the standard [SubscriptionListener] that feeds this tracker. Every event (stored backfill
|
||||
* included) is a non-terminal sign of life from its relay; an EOSE, CLOSED, or cannot-connect settles
|
||||
* that relay. [onEachEvent] is invoked for every event (stored or live) for optional instrumentation;
|
||||
* [forward] carries the EOSE / live-event signal so the owning EOSE manager can record the relay's
|
||||
* timestamp (its usual `newEose`).
|
||||
*/
|
||||
fun WindowLoadTracker.trackingListener(
|
||||
onEachEvent: (Event) -> Unit = {},
|
||||
forward: (NormalizedRelayUrl, List<Filter>?) -> Unit,
|
||||
): SubscriptionListener =
|
||||
object : SubscriptionListener {
|
||||
override fun onEose(
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
onRelaySettled(relay)
|
||||
forward(relay, forFilters)
|
||||
}
|
||||
|
||||
override fun onEvent(
|
||||
event: Event,
|
||||
isLive: Boolean,
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
onRelayEvent(relay)
|
||||
onEachEvent(event)
|
||||
if (isLive) {
|
||||
forward(relay, forFilters)
|
||||
}
|
||||
}
|
||||
|
||||
override fun onClosed(
|
||||
message: String,
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
onRelaySettled(relay)
|
||||
}
|
||||
|
||||
override fun onCannotConnect(
|
||||
relay: NormalizedRelayUrl,
|
||||
message: String,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
onRelaySettled(relay)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user