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/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..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 @@ -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(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 +262,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: PageSignal? = null + try { val listener = object : SubscriptionListener { @@ -241,7 +340,7 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - doneChannel.trySend(Unit) + doneChannel.trySend(PageSignal.EOSE) } override fun onClosed( @@ -249,7 +348,7 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - doneChannel.trySend(Unit) + doneChannel.trySend(PageSignal.CLOSED) } override fun onCannotConnect( @@ -257,7 +356,7 @@ suspend fun INostrClient.fetchAllPages( message: String, forFilters: List?, ) { - 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, 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/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..f2de98b505 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt @@ -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>, + 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) + } +} 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 {