From 37d715830b68639392f6098ba6d1b1a7f87a7622 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Sun, 9 Aug 2026 04:24:35 +0000 Subject: [PATCH 1/2] Let a drained paged walk close the leg below it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A paged band records the events it SAW, never the range it asked for, so `legs()` can only ever say "walked this far" and keeps re-asking the leg below the floor. Against a relay whose corpus for one kind simply starts later than the others' that leg is unclosable: it comes back empty every cycle, an empty fetch earns no band, so the floor never moves. Measured on a live mirror of five NIP-65 indexers, three were in that state — kind 10002 re-walked from the beginning of time to Feb 2023 forever, because relay lists did not exist before then. The missing fact is why a page ended. `fetchAllPages` treated all three terminal signals as one bare `Unit`, so an empty page could not be told apart from silence or a CLOSED. It now carries a PageEnd, and reports `onDrained` only for the one ending that proves absence: an EOSE on a page that returned nothing, with no filter capped by its `limit` and no `search` filter in play (both stop the walk short of the corpus). An idle timeout is silence, not an answer, and recording it would durably claim coverage the relay never served. A callback rather than a richer return type: ~25 call sites across quartz, geode and downstream use the `Int`, and none should have to change to learn a fact they do not want. It follows `onNewPage`'s shape. `SyncCoverage.record` takes `drained` and marks the kinds that produced evidence complete — which required completeness to move from Band onto Span. It could not stay on the band: once kinds diverge, `legs()` hands each group its own ask, so a walk that drained `kinds: [10002]` proves nothing about kind 0, and a band-level flag set from that leg would claim both. That is the same over-claim per-kind spans exist to prevent, one level up. `Band.complete` stays as a DERIVED all-kinds-complete, so both state files keep writing the flag a pre-per-kind reader expects, and read it back as every span's default. A kind the walk never saw at all still earns nothing: there is no interval to anchor a claim to, and inventing one would be the over-claim again. Tests: five in NostrClientFetchAllPagesDrainTest pinning EOSE-empty vs silence vs CLOSED vs cannot-connect vs a fulfilled limit, and five in SyncCoverageTest for per-kind completeness, widening, and the deeper-floor escape hatch that a drain must not defeat. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016TNy5BsU9NErXYa3UNGTeJ --- .../geode/mirror/SyncCoverageFile.kt | 20 +- .../NostrClientFetchAllPagesExt.kt | 77 +++++- .../relay/client/accessories/SyncCoverage.kt | 78 ++++-- .../client/accessories/SyncCoverageTest.kt | 120 ++++++++- .../NostrClientFetchAllPagesDrainTest.kt | 230 ++++++++++++++++++ 5 files changed, 495 insertions(+), 30 deletions(-) create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt index ea5f9aad21..a13fa57420 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt @@ -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 { + 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) }, ) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt index f7b4834515..e08d432ee1 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt @@ -32,6 +32,24 @@ 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*. [onDrained] needs the + * difference: an empty page is proof the relay has nothing older only when the + * relay actually said so. + */ +private enum class PageEnd { + /** 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, +} + /** * Downloads all pages of events matching [filters] from a single [relay] using * paginated `until` cursors. @@ -87,6 +105,19 @@ 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. + * @param onDrained Called at most once, just before returning, when the walk ended + * because the relay served everything [filters] match at-or-below the starting + * `until` — an empty page confirmed by an EOSE. That is the difference between + * "the relay has nothing older" and "the relay stopped answering", which the + * `Int` return cannot express: a caller recording sync coverage may treat the + * range below the oldest event it saw as verified-empty ONLY in the first case. + * Every other ending — an idle timeout, a CLOSED, a failed connect, a fulfilled + * [Filter.limit] — leaves it unfired, because none of them prove absence. + * + * A callback rather than a richer return type on purpose: this function has + * ~25 call sites across quartz, geode and downstream repos that use the `Int`, + * and none of them should have to change to learn a fact they do not want. It + * follows the shape [onNewPage] already set. * @return Total number of distinct events delivered across all pages. */ suspend fun INostrClient.fetchAllPages( @@ -94,10 +125,12 @@ suspend fun INostrClient.fetchAllPages( filters: List, idleTimeoutMs: Long = 30_000L, onNewPage: ((Long) -> Unit)? = null, + onDrained: (() -> Unit)? = null, onEvent: suspend (Event) -> Unit, ): Int { var until: Long? = null var totalEvents = 0 + var drained = false // Track how many matching events each filter has received so far. val matchCountPerFilter = IntArray(filters.size) @@ -155,7 +188,7 @@ suspend fun INostrClient.fetchAllPages( // REQ, so firing this earlier would report a page that never happens. if (until != null) onNewPage?.invoke(until) - val doneChannel = Channel(Channel.CONFLATED) + val doneChannel = Channel(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 +202,12 @@ suspend fun INostrClient.fetchAllPages( var pageMinTs = Long.MAX_VALUE val idsAtPageMin = HashSet() + // 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: PageEnd? = null + try { val listener = object : SubscriptionListener { @@ -241,7 +280,7 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - doneChannel.trySend(Unit) + doneChannel.trySend(PageEnd.EOSE) } override fun onClosed( @@ -249,7 +288,7 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - doneChannel.trySend(Unit) + doneChannel.trySend(PageEnd.CLOSED) } override fun onCannotConnect( @@ -257,7 +296,7 @@ suspend fun INostrClient.fetchAllPages( message: String, forFilters: List?, ) { - doneChannel.trySend(Unit) + doneChannel.trySend(PageEnd.CANNOT_CONNECT) } } @@ -266,7 +305,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 +316,26 @@ 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 + } + drained = pageEnd == PageEnd.EOSE && !cappedByLimit && filters.none { it.search != null } + break + } if (delivered == 0) { // Every event this page was a boundary-second duplicate; nothing older @@ -312,6 +369,10 @@ suspend fun INostrClient.fetchAllPages( until = nextUntil } + // After the loop, not at the break: every other exit above leaves `drained` + // false, and firing from one place keeps "at most once" true by construction. + if (drained) onDrained?.invoke() + return totalEvents } @@ -320,6 +381,7 @@ suspend fun INostrClient.fetchAllPages( filters: List, idleTimeoutMs: Long = 30_000L, onNewPage: ((Long) -> Unit)? = null, + onDrained: (() -> Unit)? = null, onEvent: suspend (Event) -> Unit, ): Int = fetchAllPages( @@ -327,5 +389,6 @@ suspend fun INostrClient.fetchAllPages( filters = filters, idleTimeoutMs = idleTimeoutMs, onNewPage = onNewPage, + onDrained = onDrained, onEvent = onEvent, ) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt index c8e2e3aed8..03a0ec6d40 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt @@ -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, - 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? = 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 = 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, - 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) } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt index abf1113f02..3ead83ba3c 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt @@ -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) + } } diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt new file mode 100644 index 0000000000..e8797a8d0f --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt @@ -0,0 +1,230 @@ +/* + * 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.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 `onDrained` — the one signal that tells a caller the relay served + * *everything* below where the walk stopped, rather than merely stopping there. + * + * The distinction is invisible in the `Int` return and 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>, + 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) + } + + var drained = false + val total = + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 2_000, + onDrained = { drained = true }, + ) { } + feeder.join() + + assertEquals(2, total) + assertTrue(drained, "an empty page the relay EOSEd is proof there is nothing older") + } + + @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) + } + + var drained = false + val total = + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 200, + onDrained = { drained = true }, + ) { } + feeder.join() + + assertEquals(2, total, "the events already delivered are still kept") + assertFalse(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) + } + + var drained = false + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 2_000, + onDrained = { drained = true }, + ) { } + feeder.join() + + assertFalse(drained, "a CLOSED is the relay declining, not an empty corpus") + } + + @Test + fun aRelayItCannotReachDoesNotDrain() = + runBlocking { + val client = ScriptedClient() + val feeder = + launch { + client.awaitPage(1) + client.listener!!.onCannotConnect(relay, "connection refused", null) + } + + var drained = false + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 2_000, + onDrained = { drained = true }, + ) { } + feeder.join() + + assertFalse(drained, "never got to ask") + } + + @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) + } + + var drained = false + val total = + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1), limit = 2)), + idleTimeoutMs = 2_000, + onDrained = { drained = true }, + ) { } + feeder.join() + + assertEquals(2, total) + assertEquals(1, client.subscribeCount, "the limit was met, so there was no second page") + assertFalse(drained, "a fulfilled limit is the caller stopping, not the corpus ending") + } +} From 5136a97b16c737382a54bab04d6b57b0f1479cda Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Sun, 9 Aug 2026 05:25:06 +0000 Subject: [PATCH 2/2] Return the walk's outcome instead of signalling a drain by callback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `onDrained` was the wrong shape. It reported the one ending a coverage caller happens to need and threw the rest away, so a CLOSED and an idle timeout still arrived indistinguishable from a clean finish — the very conflation this branch set out to remove, just moved one step along. `fetchAllPages` now returns `PagedFetchResult(downloaded, end)`, where `end` names every way the loop can stop: DRAINED, LIMIT_REACHED, IDLE, CLOSED, CANNOT_CONNECT, UNPAGEABLE. `drained` stays as a shorthand on the result so the meaning lives in one place. A caller can no longer ignore the reason by accident, and the two failure endings are now reportable rather than silently swallowed. I argued for the callback on the grounds that ~25 call sites use the `Int`. That was overstated: most call it as a statement and never touch the return. Six needed a `.downloaded`, all mechanical. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016TNy5BsU9NErXYa3UNGTeJ --- .../geode/mirror/MirrorWorker.kt | 2 +- .../NostrClientFetchAllPagesExt.kt | 139 +++++++++++++----- .../NostrClientFetchAllPagesPoolExt.kt | 4 +- .../NostrClientFetchAllPagesDrainTest.kt | 73 +++++---- ...NostrClientFetchAllPagesIdleTimeoutTest.kt | 4 +- .../NostrClientReqBypassingRelayLimitsTest.kt | 6 +- 6 files changed, 145 insertions(+), 83 deletions(-) diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt index e10f2b14bf..b78c0077a9 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt @@ -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 diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt index e08d432ee1..203311e8e4 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt @@ -33,13 +33,13 @@ import kotlinx.coroutines.ensureActive import kotlin.coroutines.coroutineContext /** - * Why one page stopped. The three terminal signals used to be indistinguishable — + * 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*. [onDrained] needs the + * 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 PageEnd { +private enum class PageSignal { /** The relay finished serving its stored events for this REQ. */ EOSE, @@ -50,6 +50,62 @@ private enum class PageEnd { 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. @@ -105,32 +161,25 @@ private enum class PageEnd { * 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. - * @param onDrained Called at most once, just before returning, when the walk ended - * because the relay served everything [filters] match at-or-below the starting - * `until` — an empty page confirmed by an EOSE. That is the difference between - * "the relay has nothing older" and "the relay stopped answering", which the - * `Int` return cannot express: a caller recording sync coverage may treat the - * range below the oldest event it saw as verified-empty ONLY in the first case. - * Every other ending — an idle timeout, a CLOSED, a failed connect, a fulfilled - * [Filter.limit] — leaves it unfired, because none of them prove absence. - * - * A callback rather than a richer return type on purpose: this function has - * ~25 call sites across quartz, geode and downstream repos that use the `Int`, - * and none of them should have to change to learn a fact they do not want. It - * follows the shape [onNewPage] already set. - * @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, filters: List, idleTimeoutMs: Long = 30_000L, onNewPage: ((Long) -> Unit)? = null, - onDrained: (() -> Unit)? = null, onEvent: suspend (Event) -> Unit, -): Int { +): PagedFetchResult { var until: Long? = null var totalEvents = 0 - var drained = false + // 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) @@ -181,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(Channel.CONFLATED) + val doneChannel = Channel(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 @@ -206,7 +266,7 @@ suspend fun INostrClient.fetchAllPages( // [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: PageEnd? = null + var pageEnd: PageSignal? = null try { val listener = @@ -280,7 +340,7 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - doneChannel.trySend(PageEnd.EOSE) + doneChannel.trySend(PageSignal.EOSE) } override fun onClosed( @@ -288,7 +348,7 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - doneChannel.trySend(PageEnd.CLOSED) + doneChannel.trySend(PageSignal.CLOSED) } override fun onCannotConnect( @@ -296,7 +356,7 @@ suspend fun INostrClient.fetchAllPages( message: String, forFilters: List?, ) { - doneChannel.trySend(PageEnd.CANNOT_CONNECT) + doneChannel.trySend(PageSignal.CANNOT_CONNECT) } } @@ -333,7 +393,15 @@ suspend fun INostrClient.fetchAllPages( val limit = filters[i].limit limit != null && matchCountPerFilter[i] >= limit } - drained = pageEnd == PageEnd.EOSE && !cappedByLimit && filters.none { it.search != null } + 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 } @@ -345,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. @@ -369,11 +440,7 @@ suspend fun INostrClient.fetchAllPages( until = nextUntil } - // After the loop, not at the break: every other exit above leaves `drained` - // false, and firing from one place keeps "at most once" true by construction. - if (drained) onDrained?.invoke() - - return totalEvents + return PagedFetchResult(totalEvents, end) } suspend fun INostrClient.fetchAllPages( @@ -381,14 +448,12 @@ suspend fun INostrClient.fetchAllPages( filters: List, idleTimeoutMs: Long = 30_000L, onNewPage: ((Long) -> Unit)? = null, - onDrained: (() -> Unit)? = null, onEvent: suspend (Event) -> Unit, -): Int = +): PagedFetchResult = fetchAllPages( relay = RelayUrlNormalizer.normalize(relay), filters = filters, idleTimeoutMs = idleTimeoutMs, onNewPage = onNewPage, - onDrained = onDrained, onEvent = onEvent, ) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesPoolExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesPoolExt.kt index 7de482b0a4..e59fcc34ed 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesPoolExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesPoolExt.kt @@ -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() } diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt index e8797a8d0f..f2de98b505 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt @@ -23,6 +23,7 @@ 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 @@ -37,14 +38,13 @@ import kotlin.test.assertFalse import kotlin.test.assertTrue /** - * Pins `onDrained` — the one signal that tells a caller the relay served - * *everything* below where the walk stopped, rather than merely stopping there. + * Pins [PagedFetchResult.End] — WHY a walk stopped, which is the half of the + * answer `downloaded` cannot carry. * - * The distinction is invisible in the `Int` return and 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. + * 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. @@ -103,18 +103,17 @@ class NostrClientFetchAllPagesDrainTest { client.listener!!.onEose(relay, null) } - var drained = false - val total = + val result = client.fetchAllPages( relay = relay, filters = listOf(Filter(kinds = listOf(1))), idleTimeoutMs = 2_000, - onDrained = { drained = true }, ) { } feeder.join() - assertEquals(2, total) - assertTrue(drained, "an empty page the relay EOSEd is proof there is nothing older") + 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 @@ -133,18 +132,17 @@ class NostrClientFetchAllPagesDrainTest { client.awaitPage(2) } - var drained = false - val total = + val result = client.fetchAllPages( relay = relay, filters = listOf(Filter(kinds = listOf(1))), idleTimeoutMs = 200, - onDrained = { drained = true }, ) { } feeder.join() - assertEquals(2, total, "the events already delivered are still kept") - assertFalse(drained, "an idle timeout says nothing about what the relay holds") + 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 @@ -164,16 +162,16 @@ class NostrClientFetchAllPagesDrainTest { client.listener!!.onClosed("auth-required: we don't serve that", relay, null) } - var drained = false - client.fetchAllPages( - relay = relay, - filters = listOf(Filter(kinds = listOf(1))), - idleTimeoutMs = 2_000, - onDrained = { drained = true }, - ) { } + val result = + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 2_000, + ) { } feeder.join() - assertFalse(drained, "a CLOSED is the relay declining, not an empty corpus") + assertEquals(PagedFetchResult.End.CLOSED, result.end, "a CLOSED is the relay declining, not an empty corpus") + assertFalse(result.drained) } @Test @@ -186,16 +184,16 @@ class NostrClientFetchAllPagesDrainTest { client.listener!!.onCannotConnect(relay, "connection refused", null) } - var drained = false - client.fetchAllPages( - relay = relay, - filters = listOf(Filter(kinds = listOf(1))), - idleTimeoutMs = 2_000, - onDrained = { drained = true }, - ) { } + val result = + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 2_000, + ) { } feeder.join() - assertFalse(drained, "never got to ask") + assertEquals(PagedFetchResult.End.CANNOT_CONNECT, result.end, "never got to ask") + assertFalse(result.drained) } @Test @@ -213,18 +211,17 @@ class NostrClientFetchAllPagesDrainTest { client.listener!!.onEose(relay, null) } - var drained = false - val total = + val result = client.fetchAllPages( relay = relay, filters = listOf(Filter(kinds = listOf(1), limit = 2)), idleTimeoutMs = 2_000, - onDrained = { drained = true }, ) { } feeder.join() - assertEquals(2, total) + assertEquals(2, result.downloaded) assertEquals(1, client.subscribeCount, "the limit was met, so there was no second page") - assertFalse(drained, "a fulfilled limit is the caller stopping, not the corpus ending") + assertEquals(PagedFetchResult.End.LIMIT_REACHED, result.end, "a fulfilled limit is the caller stopping, not the corpus ending") + assertFalse(result.drained) } } diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesIdleTimeoutTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesIdleTimeoutTest.kt index fb9855a7a6..534f995cdb 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesIdleTimeoutTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesIdleTimeoutTest.kt @@ -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") } diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientReqBypassingRelayLimitsTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientReqBypassingRelayLimitsTest.kt index 844ff29282..96971d4e0c 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientReqBypassingRelayLimitsTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientReqBypassingRelayLimitsTest.kt @@ -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 {