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

Merge pull request #3884 from vitorpamplona/claude/paged-drain-signal

Browse Source 
Let a drained paged walk close the sync leg below it
This commit is contained in:
Vitor Pamplona
2026-08-09 10:40:57 -04:00
committed by GitHub
parent 4f41f16db5 5136a97b16
commit cee2a9ea4b
9 changed files with 572 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt
+1 -1
View File
@@ -548,7 +548,7 @@ class MirrorWorker(
// The watchdog matches negentropySync's default rather
// than fetchAllPages' shorter one: a paged catch-up
// sits behind the same slow upstreams.
downloaded += client.fetchAllPages(up.url, listOf(leg), idleTimeoutMs = 120_000L) { observe(it) }
downloaded += client.fetchAllPages(up.url, listOf(leg), idleTimeoutMs = 120_000L) { observe(it) }.downloaded
true
}
paged = paged || legPaged
geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt
+17 -3
View File
@@ -108,7 +108,6 @@ class SyncCoverageFile(
key to
SyncCoverage.Band(
spansOf(o),
o["complete"]?.jsonPrimitive?.boolean ?: false,
o["fullAt"]?.jsonPrimitive?.long ?: 0L,
)
}.toMap(),
@@ -128,17 +127,31 @@ class SyncCoverageFile(
* discarded, and the first paged walk that reports per kind replaces it.
* Dropping it instead would re-download every upstream's corpus once on
* upgrade, which is the cost bands exist to avoid.
*
* Completeness is read the same way, one level down: a span written before
* it was per kind has no `complete` of its own, so it inherits the band's —
* which is precisely what that flag used to mean for every kind at once.
*/
private fun spansOf(o: JsonObject): Map<Int, SyncCoverage.Span> {
val bandComplete = o["complete"]?.jsonPrimitive?.boolean ?: false
o["spans"]?.jsonObject?.let { spans ->
return spans.entries.associate { (kind, v) ->
val span = v.jsonObject
kind.toInt() to SyncCoverage.Span(span.getValue("min").jsonPrimitive.long, span.getValue("max").jsonPrimitive.long)
kind.toInt() to
SyncCoverage.Span(
span.getValue("min").jsonPrimitive.long,
span.getValue("max").jsonPrimitive.long,
span["complete"]?.jsonPrimitive?.boolean ?: bandComplete,
)
}
}
return mapOf(
SyncCoverage.ALL_KINDS to
SyncCoverage.Span(o.getValue("min").jsonPrimitive.long, o.getValue("max").jsonPrimitive.long),
SyncCoverage.Span(
o.getValue("min").jsonPrimitive.long,
o.getValue("max").jsonPrimitive.long,
bandComplete,
),
)
}
@@ -170,6 +183,7 @@ class SyncCoverageFile(
buildJsonObject {
put("min", span.min)
put("max", span.max)
put("complete", span.complete)
},
)
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt
+142 -14
View File
@@ -32,6 +32,80 @@ import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.ensureActive
import kotlin.coroutines.coroutineContext
/**
* Why ONE page stopped. The three terminal signals used to be indistinguishable —
* all of them put a bare `Unit` on the page's channel — and [fetchAllPages] only
* ever asked *whether* a page ended, never *how*. [PagedFetchResult] needs the
* difference: an empty page is proof the relay has nothing older only when the
* relay actually said so.
*/
private enum class PageSignal {
/** The relay finished serving its stored events for this REQ. */
EOSE,
/** The relay ended the subscription itself — auth required, rate limited, policy. */
CLOSED,
/** Never got to ask. */
CANNOT_CONNECT,
}
/**
* What a [fetchAllPages] walk delivered, and why it stopped.
*
* The count alone cannot answer the question that matters to anything recording
* coverage: is there nothing older, or did the relay simply stop giving us more?
* Those look identical from `downloaded`, and a caller that guesses wrong either
* re-walks a corpus forever or claims history it never read.
*/
data class PagedFetchResult(
/** Total number of distinct events delivered across all pages. */
val downloaded: Int,
val end: End,
) {
enum class End {
/**
* A page came back empty and the relay EOSEd it: there is nothing at or
* below the last cursor. The only ending that proves ABSENCE, and so the
* only one a coverage claim may be built on.
*/
DRAINED,
/**
* A [Filter.limit] was fulfilled. The caller bounded the download itself,
* so the walk stopped on its own instruction, not at the end of the
* corpus — nothing below the last event was ever asked for.
*/
LIMIT_REACHED,
/**
* The relay went quiet for `idleTimeoutMs` without ending the page.
* Silence is not an answer; everything delivered so far is still good.
*/
IDLE,
/** The relay ended the subscription — auth required, rate limited, policy. */
CLOSED,
/** Never got to ask. */
CANNOT_CONNECT,
/**
* The walk cannot advance its cursor: only `search` hits came back (NIP-50
* results are relevance-ranked, so they never page), or a first page
* delivered nothing any active filter matched.
*/
UNPAGEABLE,
}
/**
* Shorthand for the one ending that licenses skipping work later. Read this
* rather than comparing to [End.DRAINED] by hand, so the meaning stays in one
* place if the enum grows.
*/
val drained: Boolean get() = end == End.DRAINED
}
/**
* Downloads all pages of events matching [filters] from a single [relay] using
* paginated `until` cursors.
@@ -87,7 +161,11 @@ import kotlin.coroutines.coroutineContext
* bounds this walk is a [Filter.limit] (the documented way to cap a download) or
* cancelling the caller, which the [ensureActive] at the top of each page honors.
* @param onEvent Called once for every distinct event delivered, in page order.
* @return Total number of distinct events delivered across all pages.
* @return What was delivered and WHY the walk stopped — see [PagedFetchResult].
* The reason is part of the answer, not a detail: `downloaded` cannot tell
* "the relay has nothing older" from "the relay stopped answering", and a
* caller recording sync coverage may only treat the range below the oldest
* event it saw as verified-empty in the first case.
*/
suspend fun INostrClient.fetchAllPages(
relay: NormalizedRelayUrl,
@@ -95,9 +173,13 @@ suspend fun INostrClient.fetchAllPages(
idleTimeoutMs: Long = 30_000L,
onNewPage: ((Long) -> Unit)? = null,
onEvent: suspend (Event) -> Unit,
): Int {
): PagedFetchResult {
var until: Long? = null
var totalEvents = 0
// Overwritten by whichever break ends the loop. UNPAGEABLE is the honest
// default: the two breaks that leave it alone are both "the cursor cannot
// advance", and it is the reading that licenses the least.
var end = PagedFetchResult.End.UNPAGEABLE
// Track how many matching events each filter has received so far.
val matchCountPerFilter = IntArray(filters.size)
@@ -148,14 +230,25 @@ suspend fun INostrClient.fetchAllPages(
stillNeedsMore && pageableThisPage
}
if (activeFilters.isEmpty()) break
if (activeFilters.isEmpty()) {
// Every filter either met its limit or is a search that has had its
// one page. The first is the caller stopping the walk; the second
// cannot page at all. Neither is the corpus ending.
end =
if (filters.any { it.limit != null }) {
PagedFetchResult.End.LIMIT_REACHED
} else {
PagedFetchResult.End.UNPAGEABLE
}
break
}
// Announce the page only now that we know it will actually be fetched: a
// search-only filter drops out of activeFilters above and breaks with no
// REQ, so firing this earlier would report a page that never happens.
if (until != null) onNewPage?.invoke(until)
val doneChannel = Channel<Unit>(Channel.CONFLATED)
val doneChannel = Channel<PageSignal>(Channel.CONFLATED)
// Idle watchdog for this page: every arriving event bumps it, so the page's
// timeout measures silence since the relay's most recent message (the same
@@ -169,6 +262,12 @@ suspend fun INostrClient.fetchAllPages(
var pageMinTs = Long.MAX_VALUE
val idsAtPageMin = HashSet<HexKey>()
// How this page ended, read after the wait: null for an idle timeout, which
// [receiveWithinIdle] reports by returning null. Only an EOSE can support a
// drain claim below — silence is not an answer, and a CLOSED is the relay
// declining to give one.
var pageEnd: PageSignal? = null
try {
val listener =
object : SubscriptionListener {
@@ -241,7 +340,7 @@ suspend fun INostrClient.fetchAllPages(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
doneChannel.trySend(Unit)
doneChannel.trySend(PageSignal.EOSE)
}
override fun onClosed(
@@ -249,7 +348,7 @@ suspend fun INostrClient.fetchAllPages(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
doneChannel.trySend(Unit)
doneChannel.trySend(PageSignal.CLOSED)
}
override fun onCannotConnect(
@@ -257,7 +356,7 @@ suspend fun INostrClient.fetchAllPages(
message: String,
forFilters: List<Filter>?,
) {
doneChannel.trySend(Unit)
doneChannel.trySend(PageSignal.CANNOT_CONNECT)
}
}
@@ -266,7 +365,7 @@ suspend fun INostrClient.fetchAllPages(
// Wait for the page's terminal signal (EOSE / CLOSED / cannot-connect),
// giving up only after [idleTimeoutMs] of silence — the wait resets on every
// arriving event, so an actively streaming page is never cut mid-delivery.
doneChannel.receiveWithinIdle(clock, idleTimeoutMs)
pageEnd = doneChannel.receiveWithinIdle(clock, idleTimeoutMs)
unsubscribe(subId)
doneChannel.close()
@@ -277,8 +376,34 @@ suspend fun INostrClient.fetchAllPages(
totalEvents += delivered
// The relay sent nothing at-or-below `until` → the whole set is drained.
if (received == 0) break
// The relay sent nothing at-or-below `until`. Whether that DRAINS the set
// depends on why the page ended and on what was asked:
//
// - only an EOSE proves absence. An idle timeout (`pageEnd == null`) is
// silence and a CLOSED is the relay declining to answer; reading either
// as "nothing older exists" would durably record coverage the relay
// never served, which is the one error a coverage claim must not make.
// - a filter that reached its [Filter.limit] stopped early on the caller's
// own instruction, so nothing below its last event was ever asked for.
// - a `search` filter runs on the first page only, so every page after it
// dropped out never carried it and cannot speak for it.
if (received == 0) {
val cappedByLimit =
filters.indices.any { i ->
val limit = filters[i].limit
limit != null && matchCountPerFilter[i] >= limit
}
end =
when {
pageEnd == PageSignal.CLOSED -> PagedFetchResult.End.CLOSED
pageEnd == PageSignal.CANNOT_CONNECT -> PagedFetchResult.End.CANNOT_CONNECT
pageEnd == null -> PagedFetchResult.End.IDLE
cappedByLimit -> PagedFetchResult.End.LIMIT_REACHED
filters.any { it.search != null } -> PagedFetchResult.End.UNPAGEABLE
else -> PagedFetchResult.End.DRAINED
}
break
}
if (delivered == 0) {
// Every event this page was a boundary-second duplicate; nothing older
@@ -288,14 +413,17 @@ suspend fun INostrClient.fetchAllPages(
// recovers progress, dropping only the second's unreachable tail). Both
// are resolved by stepping strictly past it. `boundary` is null only on
// the first page, which has no dedup and so can't be all-duplicate.
val step = boundary ?: break
val step = boundary ?: break // first page, all-duplicate: impossible, and `end` stays UNPAGEABLE
until = step - 1
seenAtBoundary = HashSet()
continue
}
// Only search hits advanced nothing pageable → can't page further.
if (pageMinTs == Long.MAX_VALUE) break
if (pageMinTs == Long.MAX_VALUE) {
end = PagedFetchResult.End.UNPAGEABLE
break
}
// Advance inclusively to the oldest second seen, carrying its dedup set:
// still the same boundary → accumulate; a genuinely older one → replace.
@@ -312,7 +440,7 @@ suspend fun INostrClient.fetchAllPages(
until = nextUntil
}
return totalEvents
return PagedFetchResult(totalEvents, end)
}
suspend fun INostrClient.fetchAllPages(
@@ -321,7 +449,7 @@ suspend fun INostrClient.fetchAllPages(
idleTimeoutMs: Long = 30_000L,
onNewPage: ((Long) -> Unit)? = null,
onEvent: suspend (Event) -> Unit,
): Int =
): PagedFetchResult =
fetchAllPages(
relay = RelayUrlNormalizer.normalize(relay),
filters = filters,
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesPoolExt.kt
+2 -2
View File
@@ -79,14 +79,14 @@ suspend fun INostrClient.fetchAllPagesFromPool(
launch {
try {
onRelayStart?.invoke(relay)
val total =
val result =
fetchAllPages(
relay = relay,
filters = filtersForRelay,
idleTimeoutMs = idleTimeoutMs,
onNewPage = onNewPage?.let { cb -> { until -> cb(until, relay) } },
) { event -> onEvent(event, relay) }
onRelayComplete?.invoke(relay, total)
onRelayComplete?.invoke(relay, result.downloaded)
} finally {
semaphore.release()
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt
+59 -19
View File
@@ -102,12 +102,27 @@ class SyncCoverage(
}
}
/** A covered `created_at` interval, inclusive at both ends. */
/**
* A covered `created_at` interval, inclusive at both ends.
*
* [complete] is the difference between "we walked this interval" and "there
* is nothing below it". A paged fetch earns the first by seeing events; it
* earns the second only when the relay EOSEs on an empty page, which is the
* one answer that distinguishes an exhausted corpus from a relay that capped
* us or went quiet (see `fetchAllPages`'s `onDrained`). A finished negentropy
* reconcile earns it too, by comparing the whole range at once.
*
* It lives HERE rather than on [Band] because a paged leg does not have to
* cover every kind: once kinds diverge, [legs] hands each group its own ask,
* so a walk that drained `kinds: [10002]` says nothing about kind 0. A
* band-level flag set from such a leg would claim both.
*/
data class Span(
val min: Long,
val max: Long,
val complete: Boolean = false,
) {
fun widen(other: Span) = Span(minOf(min, other.min), maxOf(max, other.max))
fun widen(other: Span) = Span(minOf(min, other.min), maxOf(max, other.max), complete || other.complete)
}
/**
@@ -125,30 +140,36 @@ class SyncCoverage(
* span under [ALL_KINDS] — the same claim as before, correctly scoped to
* the case where it is the only claim available.
*
* [complete] is the difference between "we walked this span" (a paged
* fetch) and "we are in sync below this point" (a finished negentropy
* reconcile, which compared the whole range). Only a complete band may
* skip its older leg. It is a property of the BAND rather than of a span:
* a reconcile compares the filter's whole id set at once, so it either
* covers every kind in it or none.
* Completeness is per kind, on [Span] — see there for why a paged leg
* cannot speak for kinds it did not ask about.
*
* [fullAt] is when the last pass that started from nothing finished — the
* clock for the periodic re-walk.
*/
data class Band(
val spans: Map<Int, Span>,
val complete: Boolean = false,
val fullAt: Long = 0,
) {
/** The outer edges across every kind — for logging and for the file's compatibility fields. */
val minCreatedAt: Long get() = spans.values.minOfOrNull { it.min } ?: 0
val maxCreatedAt: Long get() = spans.values.maxOfOrNull { it.max } ?: 0
/**
* The band-level claim, DERIVED: true only when every kind is complete.
*
* Kept for the state files, which still write a `complete` beside
* `min`/`max` so a build from before per-kind completeness reads them and
* behaves as it always did. `all` rather than `any` is the safe direction
* for that reader: it cannot see the per-kind detail, so it must be told
* the weakest true thing, not the strongest.
*/
val complete: Boolean get() = spans.isNotEmpty() && spans.values.all { it.complete }
/** Widen each kind by its counterpart, keeping kinds only one side knows. */
fun widen(other: Band): Band {
val merged = spans.toMutableMap()
for ((kind, span) in other.spans) merged[kind] = merged[kind]?.widen(span) ?: span
return Band(merged, complete || other.complete, fullAt)
return Band(merged, fullAt)
}
}
@@ -189,7 +210,8 @@ class SyncCoverage(
val kinds = filter.kinds
if (kinds.isNullOrEmpty()) {
// Nothing to split by. One span, exactly as before.
return windows(filter, band.spans[ALL_KINDS], band.complete, floor)
val span = band.spans[ALL_KINDS]
return windows(filter, span, span?.complete == true, floor)
.map { (since, until) -> filter.copy(since = since, until = until) }
}
@@ -205,7 +227,12 @@ class SyncCoverage(
// wider claim for every kind — the behaviour this replaces — and
// self-corrects on the first paged walk that reports per kind.
val span = band.spans[kind] ?: band.spans[ALL_KINDS]
byWindows.getOrPut(windows(filter, span, band.complete, floor)) { mutableListOf() }.add(kind)
// Completeness comes from the SPAN, so a leg that drained one kind
// drops only that kind's older leg. Before this it came from the
// band, and a reconcile was the only thing that could set it — which
// is why a drained paged walk over `kinds: [10002]` used to leave an
// older leg that no future walk could ever close.
byWindows.getOrPut(windows(filter, span, span?.complete == true, floor)) { mutableListOf() }.add(kind)
}
return byWindows.flatMap { (windows, group) ->
// toList(): `group` is the mutable accumulator above, and handing
@@ -266,6 +293,19 @@ class SyncCoverage(
* the sync STARTED — recorded against that instant rather than the newest
* event seen, because "the relay had nothing newer" and "we never asked"
* must not look alike.
*
* [drained] is the paged equivalent, and the only way a paged walk can ever
* claim its older leg is finished. Pass it from `fetchAllPages`'s
* `onDrained` — the relay EOSEd on an empty page, so there is nothing below
* what was seen. Without it a paged band only ever says "walked this far",
* and the leg below it is re-asked every cycle forever: against a relay
* whose corpus for a kind simply starts later than the others', that leg
* returns nothing, records nothing, and so can never close itself.
*
* It marks only the kinds that produced evidence. A kind the walk never saw
* at all has no interval to anchor a claim to, and inventing one would be
* the over-claim [Span] exists to prevent — so it keeps no band and is
* asked again, which is the safe direction.
*/
fun record(
url: NormalizedRelayUrl,
@@ -275,13 +315,14 @@ class SyncCoverage(
paged: Boolean,
reconciledThrough: Long? = null,
observedByKind: Map<Int, Span>? = null,
drained: Boolean = false,
) {
if (reconciledThrough != null) {
// A reconcile compares the filter's whole id set in one pass, so
// the span it earns is the same for every kind the filter names —
// no per-kind evidence needed or possible.
val span = Span(observedMin ?: reconciledThrough, reconciledThrough)
put(url, filter, kindsOf(filter).associateWith { span }, complete = true)
val span = Span(observedMin ?: reconciledThrough, reconciledThrough, complete = true)
put(url, filter, kindsOf(filter).associateWith { span })
return
}
if (!paged) return
@@ -298,7 +339,7 @@ class SyncCoverage(
}
if (plausible.isEmpty()) return
val named = filter.kinds
val spans =
val spans: Map<Int, Span> =
if (named.isNullOrEmpty()) {
// A filter naming no kinds cannot be split, so [legs] reads
// ALL_KINDS and nothing else. Storing what the walk saw per
@@ -321,7 +362,7 @@ class SyncCoverage(
plausible.filterKeys { it in named }
}
if (spans.isEmpty()) return
put(url, filter, spans, complete = false)
put(url, filter, if (drained) spans.mapValues { it.value.copy(complete = true) } else spans)
return
}
@@ -349,7 +390,7 @@ class SyncCoverage(
// range nothing can be in, forever.
if (observedMin == null || observedMax == null) return
if (!isPlausible(observedMin, at) || !isPlausible(observedMax, at)) return
put(url, filter, kinds.associateWith { Span(observedMin, observedMax) }, complete = false)
put(url, filter, kinds.associateWith { Span(observedMin, observedMax, complete = drained) })
}
/** The kinds a band is keyed by: the filter's, or [ALL_KINDS] when it names none. */
@@ -364,9 +405,8 @@ class SyncCoverage(
url: NormalizedRelayUrl,
filter: Filter,
spans: Map<Int, Span>,
complete: Boolean,
) {
val fresh = Band(spans, complete, now())
val fresh = Band(spans, now())
bands.merge(key(url, filter), fresh) { old, new ->
if (isStale(old)) new else old.widen(new)
}
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt
+119 -1
View File
@@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import com.vitorpamplona.quartz.utils.TimeUtils
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertNull
import kotlin.test.assertSame
import kotlin.test.assertTrue
@@ -511,7 +512,6 @@ class SyncCoverageTest {
key to
SyncCoverage.Band(
mapOf(SyncCoverage.ALL_KINDS to SyncCoverage.Span(1_690_000_000L, 1_700_000_000L)),
complete = false,
fullAt = now(),
),
),
@@ -579,4 +579,122 @@ class SyncCoverageTest {
assertEquals(2, c.legs(relay, anyKind).size)
assertEquals(1_690_000_000L, c.legs(relay, anyKind)[0].until)
}
// ---- draining: the leg a paged walk is finally allowed to close ---------
@Test
fun `a drained paged walk stops asking about the past`() {
// THE OTHER HALF OF THE BUG ABOVE. Per-kind spans stopped kind 0 from
// vouching for 30382 — but they left 30382 an older leg that nothing
// could ever close. The relay's corpus for it simply starts later, so
// that leg comes back empty every cycle, records nothing (an empty
// fetch earns no band), and the floor never moves. Forever.
//
// A drain is the missing evidence: the relay EOSEd on an empty page, so
// there IS nothing below what we saw, and the leg is done.
val walked = SyncCoverage()
walked.record(relay, profiles, 1_690_000_000L, 1_700_000_000L, paged = true)
assertEquals(2, walked.legs(relay, profiles).size, "not drained: still asks below the floor")
val drained = SyncCoverage()
drained.record(relay, profiles, 1_690_000_000L, 1_700_000_000L, paged = true, drained = true)
val legs = drained.legs(relay, profiles)
assertEquals(1, legs.size, "drained: only the newer leg is left")
assertEquals(1_700_000_000L, legs[0].since)
assertNull(legs[0].until)
}
@Test
fun `a drained leg closes its own kind and not the others`() {
// Why completeness had to move from the band onto the span. After the
// kinds diverge, legs() hands each group its own ask — so a walk that
// drained `kinds: [30382]` proves nothing whatever about kind 0. A
// band-level flag set from that leg would have claimed both, which is
// the same over-claim per-kind spans exist to prevent, just one level up.
val c = SyncCoverage()
c.record(
relay,
mixed,
null,
null,
paged = true,
observedByKind = mapOf(30382 to SyncCoverage.Span(1_690_000_000L, 1_700_000_000L)),
drained = true,
)
// Kind 0 has no evidence at all here, so it keeps asking for everything.
c.record(
relay,
mixed,
null,
null,
paged = true,
observedByKind = mapOf(0 to SyncCoverage.Span(1_600_000_000L, 1_700_000_000L)),
)
val legs = c.legs(relay, mixed)
assertTrue(!reaches(legs, 30382, 1_650_000_000L), "30382 drained — its past is settled")
assertTrue(reaches(legs, 0, 1_500_000_000L), "kind 0 did not drain, so its older leg stands")
}
@Test
fun `a band is complete only when every kind is`() {
// The band-level flag is derived now, and the state files still write it
// for a reader that predates per-kind completeness. That reader cannot
// see the detail, so it has to be told the weakest true thing.
val c = SyncCoverage()
c.record(
relay,
mixed,
null,
null,
paged = true,
observedByKind = mapOf(0 to SyncCoverage.Span(1_690_000_000L, 1_700_000_000L)),
drained = true,
)
assertTrue(c.band(relay, mixed)!!.complete, "the only kind with a span drained")
c.record(
relay,
mixed,
null,
null,
paged = true,
observedByKind = mapOf(30382 to SyncCoverage.Span(1_695_000_000L, 1_700_000_000L)),
)
assertFalse(c.band(relay, mixed)!!.complete, "…and now one of the two has not")
}
@Test
fun `widening carries a drain forward rather than losing it`() {
// Two legs of one walk against the same relay land in the same span.
// Widening must not let the second, undrained one erase what the first
// proved: the past below the merged floor really was checked.
val c = SyncCoverage()
c.record(relay, profiles, 1_690_000_000L, 1_695_000_000L, paged = true, drained = true)
c.record(relay, profiles, 1_696_000_000L, 1_700_000_000L, paged = true)
assertTrue(
c
.band(relay, profiles)!!
.spans
.getValue(0)
.complete,
)
assertEquals(1, c.legs(relay, profiles).size, "still done with the past")
}
@Test
fun `a deeper floor still re-opens history below a drained band`() {
// A drain says "nothing below what this walk asked for", not "nothing
// below, ever". A caller that now reaches deeper than the band's floor
// gets its older leg back — the same escape hatch a finished reconcile
// has, and the reason `since` is consulted at all.
val c = SyncCoverage()
c.record(relay, profiles, 1_690_000_000L, 1_700_000_000L, paged = true, drained = true)
assertEquals(1, c.legs(relay, profiles).size)
val deeper = c.legs(relay, profiles, floor = 1_600_000_000L)
assertEquals(2, deeper.size, "the caller's floor dropped below the band, so the past re-opens")
assertEquals(1_690_000_000L, deeper[0].until)
}
}
quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt
+227
View File
@@ -0,0 +1,227 @@
/*
* 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
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.PagedFetchResult
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages
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.nip01Core.relay.normalizer.RelayUrlNormalizer
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertTrue
/**
* Pins [PagedFetchResult.End] — WHY a walk stopped, which is the half of the
* answer `downloaded` cannot carry.
*
* It matters enormously to anything recording sync coverage: without it, "the
* relay has nothing older" and "the relay capped us / went quiet / hung up" look
* identical, so a coverage band can never close its oldest leg and re-asks a
* range that will always come back empty, every cycle, forever.
*
* Real-clock ([runBlocking]) for the same reason the idle-timeout suite is: the
* page watchdog is a monotonic clock bumped from the socket reader thread.
*/
class NostrClientFetchAllPagesDrainTest {
/** Captures the subscription listener so the test can play a relay, page by page. */
private class ScriptedClient : INostrClient by EmptyNostrClient() {
@Volatile
var listener: SubscriptionListener? = null
@Volatile
var subscribeCount = 0
override fun subscribe(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
listener: SubscriptionListener?,
) {
subscribeCount++
this.listener = listener
}
/** Block until the walk has opened its [n]th page, so a script can answer it. */
suspend fun awaitPage(n: Int) {
while (subscribeCount < n) delay(2)
}
}
private val relay = RelayUrlNormalizer.normalize("wss://drain.example.com")
private fun event(createdAt: Long) =
Event(
id = createdAt.toString(16).padStart(64, '0'),
pubKey = "f".repeat(64),
createdAt = createdAt,
kind = 1,
tags = emptyArray(),
content = "e$createdAt",
sig = "0".repeat(128),
)
@Test
fun anEmptyPageConfirmedByEoseDrains() =
runBlocking {
val client = ScriptedClient()
val feeder =
launch {
client.awaitPage(1)
client.listener!!.onEvent(event(2000), false, relay, null)
client.listener!!.onEvent(event(1000), false, relay, null)
client.listener!!.onEose(relay, null)
// The second page asks below 1000 and the relay says, with an
// EOSE, that it has nothing. THAT is a drain.
client.awaitPage(2)
client.listener!!.onEose(relay, null)
}
val result =
client.fetchAllPages(
relay = relay,
filters = listOf(Filter(kinds = listOf(1))),
idleTimeoutMs = 2_000,
) { }
feeder.join()
assertEquals(2, result.downloaded)
assertEquals(PagedFetchResult.End.DRAINED, result.end, "an empty page the relay EOSEd is proof there is nothing older")
assertTrue(result.drained)
}
@Test
fun aSilentPageDoesNotDrain() =
runBlocking {
val client = ScriptedClient()
val feeder =
launch {
client.awaitPage(1)
client.listener!!.onEvent(event(2000), false, relay, null)
client.listener!!.onEvent(event(1000), false, relay, null)
client.listener!!.onEose(relay, null)
// Page two: the relay simply stops answering. Silence is not an
// answer — reading it as "nothing older exists" would durably
// record coverage that was never served.
client.awaitPage(2)
}
val result =
client.fetchAllPages(
relay = relay,
filters = listOf(Filter(kinds = listOf(1))),
idleTimeoutMs = 200,
) { }
feeder.join()
assertEquals(2, result.downloaded, "the events already delivered are still kept")
assertEquals(PagedFetchResult.End.IDLE, result.end)
assertFalse(result.drained, "an idle timeout says nothing about what the relay holds")
}
@Test
fun aClosedSubscriptionDoesNotDrain() =
runBlocking {
val client = ScriptedClient()
val feeder =
launch {
client.awaitPage(1)
client.listener!!.onEvent(event(2000), false, relay, null)
client.listener!!.onEvent(event(1000), false, relay, null)
client.listener!!.onEose(relay, null)
// Page two: the relay ends the subscription instead of serving
// it — auth-required, rate limit, policy. It declined to answer,
// which is not the same as answering "nothing".
client.awaitPage(2)
client.listener!!.onClosed("auth-required: we don't serve that", relay, null)
}
val result =
client.fetchAllPages(
relay = relay,
filters = listOf(Filter(kinds = listOf(1))),
idleTimeoutMs = 2_000,
) { }
feeder.join()
assertEquals(PagedFetchResult.End.CLOSED, result.end, "a CLOSED is the relay declining, not an empty corpus")
assertFalse(result.drained)
}
@Test
fun aRelayItCannotReachDoesNotDrain() =
runBlocking {
val client = ScriptedClient()
val feeder =
launch {
client.awaitPage(1)
client.listener!!.onCannotConnect(relay, "connection refused", null)
}
val result =
client.fetchAllPages(
relay = relay,
filters = listOf(Filter(kinds = listOf(1))),
idleTimeoutMs = 2_000,
) { }
feeder.join()
assertEquals(PagedFetchResult.End.CANNOT_CONNECT, result.end, "never got to ask")
assertFalse(result.drained)
}
@Test
fun aFulfilledLimitDoesNotDrain() =
runBlocking {
// The caller bounded the download itself, so the walk stopped on its
// own instruction rather than at the end of the relay's corpus.
// Nothing below the last event was ever asked for.
val client = ScriptedClient()
val feeder =
launch {
client.awaitPage(1)
client.listener!!.onEvent(event(2000), false, relay, null)
client.listener!!.onEvent(event(1000), false, relay, null)
client.listener!!.onEose(relay, null)
}
val result =
client.fetchAllPages(
relay = relay,
filters = listOf(Filter(kinds = listOf(1), limit = 2)),
idleTimeoutMs = 2_000,
) { }
feeder.join()
assertEquals(2, result.downloaded)
assertEquals(1, client.subscribeCount, "the limit was met, so there was no second page")
assertEquals(PagedFetchResult.End.LIMIT_REACHED, result.end, "a fulfilled limit is the caller stopping, not the corpus ending")
assertFalse(result.drained)
}
}
quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesIdleTimeoutTest.kt
+2 -2
View File
@@ -111,7 +111,7 @@ class NostrClientFetchAllPagesIdleTimeoutTest {
) { got.add(it) }
feeder.join()
assertEquals(6, total, "a slowly-but-actively streaming page must never be cropped")
assertEquals(6, total.downloaded, "a slowly-but-actively streaming page must never be cropped")
assertEquals(6, got.size)
assertEquals(1, client.subscribeCount, "the whole stream must arrive in ONE page — a hard deadline would truncate and re-subscribe")
assertEquals(0, pages, "no pagination should be needed")
@@ -145,7 +145,7 @@ class NostrClientFetchAllPagesIdleTimeoutTest {
feeder.join()
val elapsedMs = start.elapsedNow().inWholeMilliseconds
assertEquals(2, total, "events delivered before the stall are kept")
assertEquals(2, total.downloaded, "events delivered before the stall are kept")
assertTrue(elapsedMs >= 300, "must wait out at least one idle window, took ${elapsedMs}ms")
assertTrue(elapsedMs < 5_000, "a stalled page must end promptly after the idle window, took ${elapsedMs}ms")
}
quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientReqBypassingRelayLimitsTest.kt
+3 -3
View File
@@ -67,7 +67,7 @@ class NostrClientReqBypassingRelayLimitsTest : RelayClientTest() {
events.add(event)
}
assertEquals(1000, totalFound)
assertEquals(1000, totalFound.downloaded)
assertEquals(1000, events.size)
events.forEach { event ->
assertEquals(MetadataEvent.KIND, event.kind)
@@ -115,7 +115,7 @@ class NostrClientReqBypassingRelayLimitsTest : RelayClientTest() {
}
}
assertEquals(2500, totalFound)
assertEquals(2500, totalFound.downloaded)
assertEquals(1000, metadataEvents.size)
assertEquals(1500, contactListEvents.size)
}
@@ -165,7 +165,7 @@ class NostrClientReqBypassingRelayLimitsTest : RelayClientTest() {
filters = listOf(Filter(search = "kotlin")),
onNewPage = { searchPages++ },
) { searchEvents.add(it) }
assertEquals(2, searchTotal, "a search filter must be fetched as a single page (the relay's cap)")
assertEquals(2, searchTotal.downloaded, "a search filter must be fetched as a single page (the relay's cap)")
assertEquals(2, searchEvents.size)
assertEquals(0, searchPages, "a search filter must never advance the until cursor")
} finally {