laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-08 23:54:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(quartz): fetchAllPages must not drop events at page boundaries

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt
This commit is contained in:
Claude
2026-07-06 21:53:48 +00:00
parent aa8412630e
commit 5c016fc2d4
2 changed files with 141 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt
+76 -13
View File
@@ -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<HexKey>()
// 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<Unit>(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<HexKey>()
try {
val listener =
@@ -134,6 +160,11 @@ suspend fun INostrClient.fetchAllPages(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
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
quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientReqBypassingRelayLimitsTest.kt
+65
View File
@@ -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<Event>()
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<Event>()
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()
}
}
}