From 5c016fc2d44b9bdeaac8ba75fcf602b7aa9dd9c2 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 21:53:48 +0000 Subject: [PATCH] fix(quartz): fetchAllPages must not drop events at page boundaries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fetchAllPages advanced with `until = oldest - 1` (exclusive) and no dedup. That skips any event sharing the boundary second that didn't fit in the page — which happens at *every* page boundary landing inside a second, not just pathological dense ones — silently dropping events. An in-process probe with no second denser than the relay's page cap still lost one event straddling the boundary. Page inclusively now: `until = oldest created_at of the previous page`, and drop the re-fetched boundary events by id. The dedup set is bounded to just the current boundary second (`until` only decreases, so duplicates can only recur there), so memory stays O(one second), never O(total). A single second denser than the relay's page cap can't be drained (its tail is unreachable — no client-side fix; raising the request limit is futile since we already send one above the relay's cap). Once a page yields nothing new we step strictly past that second so paging keeps progressing to older events instead of stalling forever. Tests: boundary-straddle retrieves all 6 (was 5); dense-second-beyond-cap steps past without stalling and still delivers the neighbours. Verified on live relays (strfry / nostr-rs-relay / khatru): ground-truthing each dense internal second against the paginated set shows no gaps, incl. a 36-event second fully retrieved. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt --- .../NostrClientFetchAllPagesExt.kt | 89 ++++++++++++++++--- .../NostrClientReqBypassingRelayLimitsTest.kt | 65 ++++++++++++++ 2 files changed, 141 insertions(+), 13 deletions(-) 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 5b4d4bc854..95b9676efa 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 @@ -21,6 +21,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId @@ -31,20 +32,35 @@ import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.ensureActive import kotlinx.coroutines.withTimeoutOrNull import kotlin.coroutines.coroutineContext -import kotlin.math.min /** * Downloads all pages of events matching [filters] from a single [relay] using * paginated `until` cursors. * - * After EOSE the oldest [Event.createdAt] seen in that page minus one becomes the - * next `until`, and the query repeats until the relay returns no new events. + * Each page after the first repeats the query with `until = oldest created_at of + * the previous page` — **inclusive**, not `oldest - 1`. Advancing exclusively would + * skip any event sharing that boundary second that didn't fit in the page, which + * happens at *every* page boundary that lands inside a second (not just pathological + * "dense" seconds), silently dropping events. Re-fetching the boundary second and + * dropping the events already delivered from it (via [Event.id]) instead retrieves + * the whole boundary. The dedup set is bounded to just the current boundary second — + * `until` only ever decreases, so duplicates can only recur there — so memory stays + * O(one second), never O(total events). * * Event counting is tracked per filter using [Filter.match]. A filter is considered * fulfilled when the number of matching events reaches its [Filter.limit]. Pagination * stops when all filters with limits are fulfilled or when a page returns no events. * Filters without a limit are considered unbounded and only stop on empty pages. * + * The one unavoidable case: a single `created_at` second holding more events than the + * relay returns in a page. The inclusive re-fetch then keeps returning the same page + * and can never advance, so once a page yields nothing new we step strictly past that + * second (`until = boundary - 1`) and continue. If the second was denser than the + * relay's page cap its unreachable tail is lost — there is no client-side fix (raising + * the request `limit` is futile: while paging we already send one above the relay's + * cap, so a larger value is clamped to the same page). Stepping past at least keeps + * the download progressing to older events instead of stalling forever. + * * A `search` ([Filter.search]) filter is the exception: NIP-50 results are ranked by * relevance, not `created_at`, so paging one by a `until` cursor is meaningless — it * would silently turn a top-N search into a time-walk, and never terminate against a @@ -57,8 +73,8 @@ import kotlin.math.min * @param relay The relay to query. * @param filters Filters to apply on every page (the `until` field is overwritten per page). * @param timeoutMs Maximum time to wait for a single page's EOSE before giving up. - * @param onEvent Called for every event received (in page order, after each EOSE). - * @return Total number of events received across all pages. + * @param onEvent Called once for every distinct event delivered, in page order. + * @return Total number of distinct events delivered across all pages. */ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, @@ -73,6 +89,12 @@ suspend fun INostrClient.fetchAllPages( // Track how many matching events each filter has received so far. val matchCountPerFilter = IntArray(filters.size) + // Bounded dedup: ids already delivered at exactly the current boundary second + // (`until`), which the next inclusive page re-fetches. `until` decreases + // monotonically, so a duplicate can only ever be a boundary-second event — + // hence no full-history seen-set, and memory is O(one second)'s worth of ids. + var seenAtBoundary = HashSet() + // One subscription id reused for every page. Each page opens it (with the // page's `until`), waits for EOSE, then closes it before the next page opens // it again — so at most one subscription is ever live and the whole download @@ -122,8 +144,12 @@ suspend fun INostrClient.fetchAllPages( val doneChannel = Channel(Channel.CONFLATED) - var pageCount = 0 + // Captured for the listener: the boundary second we re-fetch this page. + val boundary = until + var received = 0 + var delivered = 0 var pageMinTs = Long.MAX_VALUE + val idsAtPageMin = HashSet() try { val listener = @@ -134,6 +160,11 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { + received++ + // Drop a boundary-second event we already delivered on an + // earlier page (the inclusive re-fetch returns it again). + if (boundary != null && event.createdAt == boundary && event.id in seenAtBoundary) return + // Count this event against every active filter it satisfies // (one event can match more than one). Only a non-search filter // may advance the `until` cursor: a search hit — possibly old, @@ -150,9 +181,17 @@ suspend fun INostrClient.fetchAllPages( } if (atLeastOne) { onEvent(event) - pageCount++ - if (advancesCursor && event.createdAt < pageMinTs) { - pageMinTs = event.createdAt + delivered++ + // Track the oldest advancing second and the ids delivered + // in it — that becomes the next boundary and its dedup set. + if (advancesCursor) { + if (event.createdAt < pageMinTs) { + pageMinTs = event.createdAt + idsAtPageMin.clear() + idsAtPageMin.add(event.id) + } else if (event.createdAt == pageMinTs) { + idsAtPageMin.add(event.id) + } } } } @@ -194,12 +233,36 @@ suspend fun INostrClient.fetchAllPages( doneChannel.close() } - if (pageCount == 0) break + totalEvents += delivered - totalEvents += pageCount + // The relay sent nothing at-or-below `until` → the whole set is drained. + if (received == 0) break - // Advance cursor: next page starts just before the oldest event seen. - until = min((until ?: Long.MAX_VALUE) - 1, pageMinTs - 1) + if (delivered == 0) { + // Every event this page was a boundary-second duplicate; nothing older + // came back. Either the boundary second is exhausted (and there is + // nothing older → the step's next page is empty and we stop) or it is + // denser than the relay's page and keeps refilling it (stuck → the step + // 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 + until = step - 1 + seenAtBoundary = HashSet() + continue + } + + // Only search hits advanced nothing pageable → can't page further. + if (pageMinTs == Long.MAX_VALUE) break + + // Advance inclusively to the oldest second seen, carrying its dedup set: + // still the same boundary → accumulate; a genuinely older one → replace. + if (boundary != null && pageMinTs == boundary) { + seenAtBoundary.addAll(idsAtPageMin) + } else { + seenAtBoundary = idsAtPageMin + } + until = pageMinTs } return totalEvents 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 3b4f476edb..844ff29282 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 @@ -174,4 +174,69 @@ class NostrClientReqBypassingRelayLimitsTest : RelayClientTest() { cappedHub.close() } } + + /** + * A page boundary that lands *inside* a `created_at` second must not drop the + * straddled events. 6 events, 2 per second at t=100/99/98, relay cap 3 — no + * second exceeds the cap, but the 3rd slot splits the t=99 second. Exclusive + * `until = oldest - 1` used to skip the sibling; the inclusive re-fetch + dedup + * must retrieve all 6, once each. + */ + @Test + fun boundaryStraddlingASecondIsFullyRetrieved() = + runBlocking { + val hub = InProcessRelays(defaultPolicy = { LimitsPolicy(RelayLimits(maxLimit = 3, defaultLimit = 3)) }) + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(hub, scope) + try { + hub.getOrCreate(defaultRelayUrl).preload( + listOf(100L, 100L, 99L, 99L, 98L, 98L).mapIndexed { i, ts -> + SyntheticEvents.fakeEvent(idSeed = i + 1, createdAt = ts) + }, + ) + val got = mutableListOf() + client.fetchAllPages(defaultRelayUrl, listOf(Filter(kinds = listOf(1)))) { got.add(it) } + assertEquals(6, got.size, "every event must be retrieved despite a boundary inside a second") + assertEquals(6, got.map { it.id }.toSet().size, "no duplicate deliveries") + } finally { + client.disconnect() + scope.cancel() + hub.close() + } + } + + /** + * A single second denser than the relay's page cap can't be fully drained (its + * tail is unreachable — no client-side fix). Paging MUST still step strictly + * past it, delivering the events newer and older than it, and MUST terminate + * rather than spin re-fetching the same page. Cap 2; A(1000), four events at + * t=999, F(998). + */ + @Test + fun denseSecondBeyondCapIsSteppedPastWithoutStalling() = + runBlocking { + val hub = InProcessRelays(defaultPolicy = { LimitsPolicy(RelayLimits(maxLimit = 2, defaultLimit = 2)) }) + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(hub, scope) + try { + val a = SyntheticEvents.fakeEvent(idSeed = 1, createdAt = 1000L) + val dense = (2..5).map { SyntheticEvents.fakeEvent(idSeed = it, createdAt = 999L) } + val f = SyntheticEvents.fakeEvent(idSeed = 6, createdAt = 998L) + hub.getOrCreate(defaultRelayUrl).preload(listOf(a) + dense + listOf(f)) + + val got = mutableListOf() + client.fetchAllPages(defaultRelayUrl, listOf(Filter(kinds = listOf(1)))) { got.add(it) } + + val ids = got.map { it.id }.toSet() + assertEquals(ids.size, got.size, "no duplicate deliveries") + assertTrue(a.id in ids, "the event newer than the dense second must be retrieved") + assertTrue(f.id in ids, "the event older than the dense second must be retrieved (stepped past)") + assertEquals(2, got.count { it.createdAt == 999L }, "exactly the relay cap of the dense second is reachable") + assertEquals(4, got.size, "A + 2-of-4 dense + F") + } finally { + client.disconnect() + scope.cancel() + hub.close() + } + } }