From 54c8cefe6906f0893174bae7c72d0ff19e766fac Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 19:58:11 +0000 Subject: [PATCH 1/8] fix(quartz): don't time-walk a search filter in fetchAllPages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NIP-50 search results are ranked by relevance, not created_at, so paging a search filter by an `until` cursor silently degrades a top-N search into a full time-walk of the corpus — and never terminates against a relay that runs FTS over its whole corpus regardless of `until`. fetchAllPages now queries a `search` filter on its first page only: it is dropped from every later page and its hits neither advance nor drag back the `until` cursor that co-resident non-search filters page with. onNewPage also moves below the empty-page break so it never announces a page that isn't fetched. Adds a test proving a search filter returns a single relay page while a plain filter over the same capped relay still pages through the set. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt --- .../NostrClientFetchAllPagesExt.kt | 37 +++++++++-- .../NostrClientReqBypassingRelayLimitsTest.kt | 64 +++++++++++++++++++ 2 files changed, 95 insertions(+), 6 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 9c09bf586d..8015467c41 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 @@ -45,6 +45,15 @@ import kotlin.math.min * 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. * + * 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 + * relay that runs FTS over its whole corpus regardless of `until`. So a search filter + * is queried on the FIRST page only; it is then dropped from every later page and its + * hits never advance (nor drag back) the `until` cursor other filters page with. Give + * it a `limit` to bound that single page; without one you get the relay's default page + * of top hits. + * * @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. @@ -86,21 +95,30 @@ suspend fun INostrClient.fetchAllPages( if (until == null) { filters } else { - onNewPage?.invoke(until) filters.map { it.copy(until = until) } } - // Only include filters that still need more events. + // Only include filters that still need more events. A `search` filter is + // relevance-ranked, not `created_at`-ordered, so it is queried on the first + // page only (until == null) and dropped from every later page — paging it by + // `until` would corrupt its ordering and can never terminate. val remainingFilters = pagedFilters.filterIndexed { index, filter -> val limit = filter.limit - limit == null || matchCountPerFilter[index] < limit + val stillNeedsMore = limit == null || matchCountPerFilter[index] < limit + val pageable = until == null || filter.search == null + stillNeedsMore && pageable } if (remainingFilters.isEmpty()) break + // Announce the page only now that we know it will actually be fetched: a + // search-only filter drops out of remainingFilters 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) var pageCount = 0 @@ -117,17 +135,24 @@ suspend fun INostrClient.fetchAllPages( ) { // Check if the relay is returning what we asked before moving forward var atLeastOne = false + // Only a paginating (non-search) filter may advance the `until` + // cursor. A search filter's hits — possibly old, relevance-ranked + // — must not drag the cursor back, or the next page would skip + // events a co-resident normal filter still needs. + var advancesCursor = false for (i in pagedFilters.indices) { - val limit = pagedFilters[i].limit - if ((limit == null || matchCountPerFilter[i] < limit) && pagedFilters[i].match(event)) { + val filter = pagedFilters[i] + val limit = filter.limit + if ((limit == null || matchCountPerFilter[i] < limit) && filter.match(event)) { matchCountPerFilter[i]++ atLeastOne = true + if (filter.search == null) advancesCursor = true } } if (atLeastOne) { onEvent(event) pageCount++ - if (event.createdAt < pageMinTs) { + if (advancesCursor && event.createdAt < pageMinTs) { pageMinTs = event.createdAt } } 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 dac066c916..3b4f476edb 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 @@ -20,17 +20,26 @@ */ package com.vitorpamplona.quartz.nip01Core.relay +import com.vitorpamplona.geode.InProcessRelays import com.vitorpamplona.geode.fixtures.SyntheticEvents import com.vitorpamplona.geode.testing.RelayClientTest import com.vitorpamplona.geode.testing.preload import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.LimitsPolicy +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RelayLimits import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel import kotlinx.coroutines.runBlocking import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertTrue class NostrClientReqBypassingRelayLimitsTest : RelayClientTest() { @Test @@ -110,4 +119,59 @@ class NostrClientReqBypassingRelayLimitsTest : RelayClientTest() { assertEquals(1000, metadataEvents.size) assertEquals(1500, contactListEvents.size) } + + /** + * A `search` filter is relevance-ranked, not `created_at`-ordered, so + * `fetchAllPages` must fetch only its FIRST page and never advance the + * `until` cursor — otherwise a NIP-50 top-N search silently degrades into a + * full time-walk of the corpus. A plain (non-search) filter over the same + * capped relay is the control: it *does* page through everything, proving + * the per-REQ cap is real and pagination is actually happening. + */ + @Test + fun searchFilterIsFetchedAsSingleRelevancePageNotTimeWalked() = + runBlocking { + // A relay that returns at most 2 events per REQ (defaultLimit fills in + // for a filter that gives no limit), so an unbounded filter must + // paginate to drain a larger set. + val cappedHub = InProcessRelays(defaultPolicy = { LimitsPolicy(RelayLimits(maxLimit = 2, defaultLimit = 2)) }) + val cappedScope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val cappedClient = NostrClient(cappedHub, cappedScope) + try { + // Five kind-1 notes, distinct created_at (via idSeed), all matching + // the FTS term "kotlin". + cappedHub.getOrCreate(defaultRelayUrl).preload( + (1..5).map { SyntheticEvents.fakeEvent(idSeed = it, content = "kotlin note $it") }, + ) + + // Control: a non-search filter drains all five across pages and + // advances the cursor (onNewPage fires) — the cap is real. + val controlEvents = mutableListOf() + var controlPages = 0 + cappedClient.fetchAllPages( + relay = defaultRelayUrl, + filters = listOf(Filter(kinds = listOf(1))), + onNewPage = { controlPages++ }, + ) { controlEvents.add(it) } + assertEquals(5, controlEvents.size, "non-search filter must page through the whole set") + assertTrue(controlPages > 0, "non-search filter must advance the until cursor across pages") + + // Search filter: only the first relevance-ranked page is fetched. + val searchEvents = mutableListOf() + var searchPages = 0 + val searchTotal = + cappedClient.fetchAllPages( + relay = defaultRelayUrl, + 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, searchEvents.size) + assertEquals(0, searchPages, "a search filter must never advance the until cursor") + } finally { + cappedClient.disconnect() + cappedScope.cancel() + cappedHub.close() + } + } } From 590731f3561a4a079b7ec07aa7cbb529598d3b0c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 20:20:57 +0000 Subject: [PATCH 2/8] feat(cli): add Context.drainAllPages + shared fetchAllPagesFromPool accessory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Amy's one-shot queries all go through Context.drain, a single REQ drained to EOSE — so a relay that caps its REQ response (strfry's per-REQ limit, ~500) silently truncates the result with no way to page past it. Extract the per-relay fetchAllPages fan-out that already lived privately in EventSync into a reusable quartz accessory, fetchAllPagesFromPool: a sliding-window pool (maxConcurrentRelays) that paginates each relay on its own `until` cursor, tags every event with its source relay, and does not dedup across relays. EventSync now delegates to it (its private downloadPool/ downloadFromRelay are deleted — no behavior change: perRelayFilters is already ordered by and complete over the relay list). Add Context.drainAllPages, the paged sibling of drain: same verify+store and per-relay tagging, but fully draining sets larger than one REQ. Wire it into `amy fetch` behind --paginate/--all (filter mode only), pushing the limit into the filter so paging stays bounded. sync (NIP-77) and fetch stay separate interfaces. Tests: fetchAllPagesFromPool fan-out/tagging/no-cross-relay-dedup. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt --- .../loggedIn/relays/eventsync/EventSync.kt | 65 +------------ .../wallet/wizard/CashuWalletDiscovery.kt | 5 +- .../com/vitorpamplona/amethyst/cli/Context.kt | 50 ++++++++++ .../amethyst/cli/commands/FetchCommand.kt | 20 +++- .../NostrClientFetchAllPagesPoolExt.kt | 93 +++++++++++++++++++ .../relay/NostrClientFetchAllPagesPoolTest.kt | 80 ++++++++++++++++ 6 files changed, 249 insertions(+), 64 deletions(-) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesPoolExt.kt create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesPoolTest.kt diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt index e7c962442d..d589c6de78 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt @@ -27,7 +27,7 @@ import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync.EventSync. 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.accessories.fetchAllPages +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPagesFromPool import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message @@ -46,10 +46,7 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.update -import kotlinx.coroutines.isActive import kotlinx.coroutines.launch -import kotlinx.coroutines.supervisorScope -import kotlinx.coroutines.sync.Semaphore import java.util.concurrent.ConcurrentHashMap import kotlin.coroutines.cancellation.CancellationException @@ -503,9 +500,10 @@ class EventSync( clientBuilder().use { client -> client.addConnectionListener(okListener) try { - client.downloadFromPool( - relays = relaysToProcess, + client.fetchAllPagesFromPool( filters = perRelayFilters, + timeoutMs = RELAY_TIMEOUT_MS, + maxConcurrentRelays = MAX_CONCURRENT_RELAYS, onNewPage = { until, sourceRelay -> _liveActivity.value.runningRelays[sourceRelay] ?.pageUntil @@ -564,7 +562,7 @@ class EventSync( ) } }, - onRelayComplete = { relay -> + onRelayComplete = { relay, _ -> _liveActivity.update { val newCompleted = it.runningRelays[relay] it.copy( @@ -615,57 +613,4 @@ class EventSync( } } } - - /** - * Maintains a sliding window of up to [MAX_CONCURRENT_RELAYS] active relay workers. - * As soon as one relay finishes (all pages exhausted), the next relay from [relays] - * starts immediately — no waiting for an entire batch to drain. - * - * [onEvent] receives the event and the URL of the relay it came from. - */ - private suspend fun INostrClient.downloadFromPool( - relays: List, - filters: Map>, - onNewPage: (Long, NormalizedRelayUrl) -> Unit, - onEvent: (Event, NormalizedRelayUrl) -> Unit, - onRelayStart: (NormalizedRelayUrl) -> Unit, - onRelayComplete: (NormalizedRelayUrl) -> Unit, - ) { - val semaphore = Semaphore(MAX_CONCURRENT_RELAYS) - supervisorScope { - for (relay in relays) { - if (!isActive) break - semaphore.acquire() - launch { - try { - onRelayStart(relay) - filters[relay]?.let { filtersForRelay -> - downloadFromRelay( - relay = relay, - filters = filtersForRelay, - onNewPage = { onNewPage(it, relay) }, - onEvent = { onEvent(it, relay) }, - ) - } ?: 0 - onRelayComplete(relay) - } finally { - semaphore.release() - } - } - } - } - } - - /** - * Fetches all pages from a single [relay] using paginated `until` cursors. - * Delegates to the Quartz [downloadFromRelay] extension. - * - * @return total number of events received across all pages. - */ - private suspend fun INostrClient.downloadFromRelay( - relay: NormalizedRelayUrl, - filters: List, - onNewPage: (Long) -> Unit, - onEvent: (Event) -> Unit, - ): Int = fetchAllPages(relay, filters, RELAY_TIMEOUT_MS, onNewPage, onEvent) } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/wallet/wizard/CashuWalletDiscovery.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/wallet/wizard/CashuWalletDiscovery.kt index 45ad8720f2..a68f4e356b 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/wallet/wizard/CashuWalletDiscovery.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/wallet/wizard/CashuWalletDiscovery.kt @@ -177,8 +177,9 @@ class CashuWalletDiscovery( /** * Sliding-window relay crawl: keeps up to [MAX_CONCURRENT_RELAYS] relays - * paginating at once, starting the next as soon as one finishes. Mirrors - * EventSync.downloadFromPool but only collects (no republish). + * paginating at once, starting the next as soon as one finishes. Same shape as + * the shared `fetchAllPagesFromPool` accessory, but with one filter list for + * every relay and collect-only (no per-relay tagging, no republish). */ private suspend fun INostrClient.crawlPool( relays: List, diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt index dffb600206..f62e624a1e 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt @@ -42,6 +42,7 @@ import com.vitorpamplona.quartz.nip01Core.crypto.verify import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPagesFromPool import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirmDetailed import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId @@ -72,6 +73,8 @@ import com.vitorpamplona.quartz.nip87Ecash.recommendation.MintRecommendationEven import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch import kotlinx.coroutines.selects.select import kotlinx.coroutines.withTimeoutOrNull import okhttp3.OkHttpClient @@ -479,6 +482,53 @@ class Context( return collected } + /** + * Like [drain], but paginates every relay to completion via + * [fetchAllPagesFromPool] instead of stopping at the first EOSE — so a query + * larger than a relay's per-`REQ` cap (strfry's `limit`, ~500) is fully + * retrieved instead of silently truncated. Each relay is walked on its own + * `until` cursor, up to [maxConcurrentRelays] at once, and every event still + * funnels through [verifyAndStore]; the result is tagged by relay exactly like + * [drain] (and, like [drain], is NOT deduped across relays — callers dedup by + * id). + * + * Bound the work with the filters' `limit`: each relay pages until it reaches + * the limit, so an unbounded filter pages that relay's entire matching history. + * A `search` filter is fetched as a single relevance-ranked page (see + * [com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages]). + */ + suspend fun drainAllPages( + filters: Map>, + timeoutMs: Long = 30_000, + maxConcurrentRelays: Int = 8, + ): List> { + if (filters.isEmpty()) return emptyList() + val collected = mutableListOf>() + // fetchAllPages' onEvent can't suspend, but verifyAndStore does — bridge + // through a channel and verify+store single-threaded in one consumer so the + // store writes stay serialized (same shape as `drain`). + val eventChannel = Channel>(UNLIMITED) + coroutineScope { + val consumer = + launch { + for ((relay, event) in eventChannel) { + if (verifyAndStore(event)) collected.add(relay to event) + } + } + try { + client.fetchAllPagesFromPool( + filters = filters, + timeoutMs = timeoutMs, + maxConcurrentRelays = maxConcurrentRelays, + ) { event, relay -> eventChannel.trySend(relay to event) } + } finally { + eventChannel.close() + } + consumer.join() + } + return collected + } + /** * Publish [request] to [relays], then wait for the FIRST event matching [responseFilter] * — a live reply that arrives after our own EOSE, which [drain] would miss (it returns at diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt index 2adffc664c..109b347f26 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt @@ -36,7 +36,8 @@ import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent /** * `amy fetch [--kind …] [--author …] [--id …] [--tag …] [--since/--until TS] - * [--limit N] [--search TEXT] [--relay URL[,URL…]] [--timeout SECS]` + * [--limit N] [--search TEXT] [--relay URL[,URL…]] [--timeout SECS] + * [--paginate]` * * amy fetch * @@ -53,6 +54,12 @@ import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent * * Results are deduplicated by id, sorted newest-first, capped at `--limit` * (default 100), and emitted as full event JSON under an `events` array. + * + * By default (filter mode) a single `REQ` is drained to EOSE, so a relay that + * caps its response (strfry's per-`REQ` `limit`, ~500) truncates the result. + * `--paginate` (alias `--all`) instead walks each relay page-by-page on `until` + * cursors up to `--limit`, fully draining sets larger than one `REQ` — the + * multi-relay [Context.drainAllPages] path. Code mode is always single-shot. */ object FetchCommand { suspend fun run( @@ -71,13 +78,22 @@ object FetchCommand { } val filter = RawEventSupport.buildFilter(args) + // --paginate/--all walks each relay past its per-REQ cap (strfry's ~500) + // by following `until` cursors, bounded by `limit`; default stops at the + // first EOSE like nak's `req`. + val paginate = args.bool("paginate") || args.bool("all") Context.open(dataDir).use { ctx -> ctx.prepare() val relays = RawEventSupport.queryTargets(ctx, args) if (relays.isEmpty()) return Output.error("no_relays", "no relays available; pass --relay or run `amy relay add`") - val received = ctx.drain(relays.associateWith { listOf(filter) }, timeoutMs) + val received = + if (paginate) { + ctx.drainAllPages(relays.associateWith { listOf(filter.copy(limit = limit)) }, timeoutMs) + } else { + ctx.drain(relays.associateWith { listOf(filter) }, timeoutMs) + } val events = received .asSequence() 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 new file mode 100644 index 0000000000..c6d4ffeca2 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesPoolExt.kt @@ -0,0 +1,93 @@ +/* + * 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.client.accessories + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlinx.coroutines.supervisorScope +import kotlinx.coroutines.sync.Semaphore + +/** + * Fans [fetchAllPages] out across every relay in [filters] — the multi-relay form + * of the single-relay downloader. Each relay is paginated independently on its own + * `until` cursor, with at most [maxConcurrentRelays] relays in flight at once: as + * soon as one drains (all pages exhausted) the next from [filters] starts, so the + * concurrency window stays full instead of waiting for a whole batch to finish. + * + * Every delivered event is tagged with the relay it came from. **No cross-relay + * dedup is done here** — the same event id can arrive from several relays, exactly + * like a fan-out `REQ`; dedup downstream if you need a distinct set. [onEvent] runs + * on the delivering relay's reader thread and must not suspend (bridge through a + * channel if your sink suspends). + * + * Relays are isolated by [supervisorScope]: one relay throwing — or all its pages + * timing out — fails only that relay's branch, never the others. A relay that can't + * connect simply yields zero events (its first page EOSEs empty) and completes + * normally. Cancelling the caller cancels every branch. + * + * @param filters per-relay filter lists; the key set is the relays queried, in + * iteration order (pass a [LinkedHashMap]/`associateWith` result to control it). + * A `search` filter is fetched as a single relevance page — see [fetchAllPages]. + * @param timeoutMs per-page EOSE timeout handed to each relay's [fetchAllPages]. + * @param maxConcurrentRelays upper bound on relays paginating at once (≥ 1). + * @param onNewPage optional `(until, relay)` tick before each non-first page. + * @param onRelayStart optional hook fired as each relay's download begins. + * @param onRelayComplete optional `(relay, totalEvents)` hook fired when a relay + * drains (or errors out to an empty first page). + * @param onEvent called once per delivered event with its source relay. + */ +suspend fun INostrClient.fetchAllPagesFromPool( + filters: Map>, + timeoutMs: Long = 30_000L, + maxConcurrentRelays: Int = 8, + onNewPage: ((until: Long, relay: NormalizedRelayUrl) -> Unit)? = null, + onRelayStart: ((relay: NormalizedRelayUrl) -> Unit)? = null, + onRelayComplete: ((relay: NormalizedRelayUrl, totalEvents: Int) -> Unit)? = null, + onEvent: (event: Event, relay: NormalizedRelayUrl) -> Unit, +) { + if (filters.isEmpty()) return + val semaphore = Semaphore(maxConcurrentRelays.coerceAtLeast(1)) + supervisorScope { + for ((relay, filtersForRelay) in filters) { + if (!isActive) break + semaphore.acquire() + launch { + try { + onRelayStart?.invoke(relay) + val total = + fetchAllPages( + relay = relay, + filters = filtersForRelay, + timeoutMs = timeoutMs, + onNewPage = onNewPage?.let { cb -> { until -> cb(until, relay) } }, + ) { event -> onEvent(event, relay) } + onRelayComplete?.invoke(relay, total) + } finally { + semaphore.release() + } + } + } + } +} diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesPoolTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesPoolTest.kt new file mode 100644 index 0000000000..a3c962882e --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesPoolTest.kt @@ -0,0 +1,80 @@ +/* + * 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.geode.fixtures.SyntheticEvents +import com.vitorpamplona.geode.testing.RelayClientTest +import com.vitorpamplona.geode.testing.preload +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPagesFromPool +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.runBlocking +import java.util.Collections +import java.util.concurrent.ConcurrentHashMap +import kotlin.test.Test +import kotlin.test.assertEquals + +class NostrClientFetchAllPagesPoolTest : RelayClientTest() { + /** + * The pool fans [fetchAllPagesFromPool] across relays, tags each event with the + * relay it came from, and — like a fan-out REQ — does NOT dedup across relays: + * an event held by two relays is delivered twice, once per source. + */ + @Test + fun fansOutPerRelayTagsSourceAndDoesNotDedupAcrossRelays() = + runBlocking { + val relayAUrl = RelayUrlNormalizer.normalize("ws://relay-a/") + val relayBUrl = RelayUrlNormalizer.normalize("ws://relay-b/") + + // One shared event lives on BOTH relays; the rest are relay-exclusive. + val shared = SyntheticEvents.fakeEvent(idSeed = 1) + val onlyA = (2..4).map { SyntheticEvents.fakeEvent(idSeed = it) } + val onlyB = (5..9).map { SyntheticEvents.fakeEvent(idSeed = it) } + hub.getOrCreate(relayAUrl).preload(onlyA + shared) + hub.getOrCreate(relayBUrl).preload(onlyB + shared) + + // onEvent runs on each relay's reader thread → collect thread-safely. + val received = Collections.synchronizedList(mutableListOf>()) + val completed = ConcurrentHashMap() + + client.fetchAllPagesFromPool( + filters = + linkedMapOf( + relayAUrl to listOf(Filter(kinds = listOf(1))), + relayBUrl to listOf(Filter(kinds = listOf(1))), + ), + onRelayComplete = { relay, total -> completed[relay] = total }, + ) { event, relay -> received.add(relay to event) } + + val fromA = received.filter { it.first == relayAUrl } + val fromB = received.filter { it.first == relayBUrl } + // A: 3 exclusive + shared = 4 ; B: 5 exclusive + shared = 6. + assertEquals(4, fromA.size, "every event must be tagged with relay A") + assertEquals(6, fromB.size, "every event must be tagged with relay B") + // The shared id arrives once per relay — not deduped across relays. + assertEquals(2, received.count { it.second.id == shared.id }, "shared event must arrive from both relays") + // onRelayComplete reports each relay's fetched total. + assertEquals(4, completed[relayAUrl]) + assertEquals(6, completed[relayBUrl]) + } +} From aa8412630e6d2c914135280b11207fe84daf852f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 20:42:34 +0000 Subject: [PATCH 3/8] refactor(quartz): collapse fetchAllPages to a single active-filter list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The search-single-page logic left two lists with different roles: the listener counted matches over the full `pagedFilters` (including a search filter already dropped from paging) while the subscription only sent `remainingFilters`. That worked — the dropped filter's count was unused and `advancesCursor` kept its hits off the cursor — but it read as if a non-subscribed filter still mattered. Collapse to one `activeFilters` list (index + filter) that is both what we subscribe and what the listener iterates, so counting can't drift from what was asked. Behavior is identical; the multi-filter and search tests still pass. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt --- .../NostrClientFetchAllPagesExt.kt | 45 +++++++++---------- 1 file changed, 22 insertions(+), 23 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 8015467c41..5b4d4bc854 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 @@ -100,22 +100,23 @@ suspend fun INostrClient.fetchAllPages( } } - // Only include filters that still need more events. A `search` filter is - // relevance-ranked, not `created_at`-ordered, so it is queried on the first - // page only (until == null) and dropped from every later page — paging it by - // `until` would corrupt its ordering and can never terminate. - val remainingFilters = - pagedFilters.filterIndexed { index, filter -> - val limit = filter.limit - val stillNeedsMore = limit == null || matchCountPerFilter[index] < limit - val pageable = until == null || filter.search == null - stillNeedsMore && pageable + // The filters actually queried this page, each kept with its index into + // matchCountPerFilter. A filter drops out once it has its limit's worth of + // events; a `search` filter additionally runs on the FIRST page only + // (until == null), because relevance-ranked results can't be paged by a + // created_at cursor. The listener below iterates this SAME list, so what we + // count always matches what we subscribed for. + val activeFilters = + pagedFilters.withIndex().filter { (index, filter) -> + val stillNeedsMore = filter.limit == null || matchCountPerFilter[index] < filter.limit + val pageableThisPage = until == null || filter.search == null + stillNeedsMore && pageableThisPage } - if (remainingFilters.isEmpty()) break + if (activeFilters.isEmpty()) break // Announce the page only now that we know it will actually be fetched: a - // search-only filter drops out of remainingFilters above and breaks with no + // 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) @@ -133,18 +134,16 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - // Check if the relay is returning what we asked before moving forward + // 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, + // relevance-ranked — must not drag the cursor back and make the + // next page skip events a co-resident normal filter still needs. var atLeastOne = false - // Only a paginating (non-search) filter may advance the `until` - // cursor. A search filter's hits — possibly old, relevance-ranked - // — must not drag the cursor back, or the next page would skip - // events a co-resident normal filter still needs. var advancesCursor = false - for (i in pagedFilters.indices) { - val filter = pagedFilters[i] - val limit = filter.limit - if ((limit == null || matchCountPerFilter[i] < limit) && filter.match(event)) { - matchCountPerFilter[i]++ + for ((index, filter) in activeFilters) { + if (matchCountPerFilter[index] < (filter.limit ?: Int.MAX_VALUE) && filter.match(event)) { + matchCountPerFilter[index]++ atLeastOne = true if (filter.search == null) advancesCursor = true } @@ -182,7 +181,7 @@ suspend fun INostrClient.fetchAllPages( } } - subscribe(subId, mapOf(relay to remainingFilters), listener) + subscribe(subId, mapOf(relay to activeFilters.map { it.value }), listener) withTimeoutOrNull(timeoutMs) { doneChannel.receive() From 5c016fc2d44b9bdeaac8ba75fcf602b7aa9dd9c2 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 21:53:48 +0000 Subject: [PATCH 4/8] 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() + } + } } From cb493a22911474ecf569dd685de0835628c5d475 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 22:38:37 +0000 Subject: [PATCH 5/8] =?UTF-8?q?feat(quartz):=20add=20SeenIds=20=E2=80=94?= =?UTF-8?q?=20a=20memory-lean=20event-id=20dedup=20filter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run-scoped "already seen this id" filter for large, mostly-duplicate id streams (a broad relay walk re-receiving the same event from many relays). Keys on the first 128 bits of the id, sliced straight out of the hex with Hex.readLong (table lookups, no parse, no allocation), in one open-addressed LongArray — ~16 bytes/entry and the 64-char String is never retained, so tens of millions of ids cost ~1 GB instead of a HashSet's ~6 GB. add() is O(1) and synchronized. Lives in the jvmAndroid source set (uses @Synchronized; a 40M-id walk is a server-side concern). Ports the caller's implementation with the parseUnsignedLong hot path swapped for Hex.readLong (~45-70 ns/op cheaper). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt --- .../com/vitorpamplona/quartz/utils/SeenIds.kt | 145 ++++++++++++++++++ .../vitorpamplona/quartz/utils/SeenIdsTest.kt | 88 +++++++++++ 2 files changed, 233 insertions(+) create mode 100644 quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt new file mode 100644 index 0000000000..df952a74ac --- /dev/null +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt @@ -0,0 +1,145 @@ +/* + * 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.utils + +/** + * A memory-lean "already seen this event id" filter for large, mostly-duplicate id + * streams — e.g. a broad relay walk that re-receives the same widely-mirrored event + * from dozens of relays. [add] drops a duplicate the moment it arrives, before any + * expensive per-event work (signature verification, a store existence check). + * + * Event ids are SHA-256 hashes — uniform random 256-bit values — so their first 128 + * bits (two longs of the 32-byte id) are themselves a perfect hash. Keying on those, + * the odds of two distinct ids colliding across tens of millions of events is ~1e-22, + * so it never wrongly skips a real event (unlike a Bloom filter). The first 128 bits + * are sliced straight out of the hex string with [Hex.readLong] — table lookups and + * shifts, no text parsing and no allocation. + * + * Backed by one open-addressed [LongArray] (two longs per slot, `(0,0)` = empty), so + * there are NO per-entry objects and the 64-char id [String] is never retained: tens + * of millions of ids cost ~16 bytes each (~1 GB at 40M) instead of the ~6 GB a + * `HashSet` of 64-char hex would. [add] is O(1) and `@Synchronized`; the lock + * is held for nanoseconds, so concurrent producers contend little. + * + * Not unbounded-safe on its own: call [reset] between passes (or whenever the working + * set should be forgotten) so a long-running process can't grow the table forever. + */ +class SeenIds( + initialSlotsPow2: Int = INITIAL_POW2, +) { + private var mask = 0 + private var table = LongArray(0) + private var count = 0 // non-zero-key entries held in [table] + private var zeroSeen = false // the (0,0) key, tracked apart from the empty sentinel + private var resizeAt = 0 + + init { + allocate(1 shl initialSlotsPow2) + } + + private fun allocate(slots: Int) { + table = LongArray(slots * 2) + mask = slots - 1 + resizeAt = (slots * LOAD).toInt() + count = 0 + } + + /** + * Records [idHex] (a 64-char hex event id); returns true if it is NEW this pass + * (the caller should process it), false if already seen (the caller should skip + * it). A too-short/malformed id returns true — it flows through and downstream + * verification drops it — rather than risk collapsing distinct ids. + */ + @Synchronized + fun add(idHex: String): Boolean { + // Slice the first 128 bits straight to two longs via Hex's table-lookup + // reader — no hex text parsing, no allocation. A string too short to slice + // can't be a real 32-byte id, so let it through (verification drops it). + if (idHex.length < 32) return true + return addKey(Hex.readLong(idHex, 0), Hex.readLong(idHex, 16)) + } + + private fun addKey( + hi: Long, + lo: Long, + ): Boolean { + if (hi == 0L && lo == 0L) { + // (0,0) is [table]'s empty sentinel, so this one key is tracked apart. + if (zeroSeen) return false + zeroSeen = true + return true + } + if (count >= resizeAt) grow() + var i = (mix(hi, lo).toInt() and mask) + while (true) { + val s = i * 2 + val h = table[s] + val l = table[s + 1] + if (h == 0L && l == 0L) { + table[s] = hi + table[s + 1] = lo + count++ + return true + } + if (h == hi && l == lo) return false + i = (i + 1) and mask + } + } + + private fun grow() { + val old = table + allocate((mask + 1) shl 1) // resets count; zeroSeen is untouched + var j = 0 + while (j < old.size) { + val h = old[j] + val l = old[j + 1] + if (h != 0L || l != 0L) addKey(h, l) + j += 2 + } + } + + @Synchronized + fun reset() { + allocate(1 shl INITIAL_POW2) + zeroSeen = false + } + + @Synchronized + fun size() = count + if (zeroSeen) 1 else 0 + + // Ids are already uniform, but avalanche the two halves so the low bits used for + // the slot index don't correlate with any particular byte of the hash. + private fun mix( + hi: Long, + lo: Long, + ): Long { + var h = hi xor (lo * -0x61c8864680b583ebL) + h = h xor (h ushr 32) + h *= -0x7ee3623a03d3f7d7L + h = h xor (h ushr 29) + return h + } + + companion object { + private const val LOAD = 0.7 + private const val INITIAL_POW2 = 20 + } +} diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt new file mode 100644 index 0000000000..3ac4dc5788 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt @@ -0,0 +1,88 @@ +/* + * 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.utils + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class SeenIdsTest { + // Vary the FIRST 128 bits (the keyed part): value in the high 16 hex chars, zero tail. + private fun id(i: Int) = i.toLong().toString(16).padStart(16, '0') + "0".repeat(48) + + @Test + fun `first sight is new, repeats are skipped`() { + val seen = SeenIds() + assertTrue(seen.add(id(1)), "first time is new") + assertFalse(seen.add(id(1)), "same id is a duplicate") + assertFalse(seen.add(id(1)), "still a duplicate") + assertTrue(seen.add(id(2)), "a different id is new") + assertEquals(2, seen.size()) + } + + @Test + fun `reset forgets everything`() { + val seen = SeenIds() + seen.add(id(7)) + assertFalse(seen.add(id(7))) + seen.reset() + assertTrue(seen.add(id(7)), "after reset the id is new again") + assertEquals(1, seen.size()) + } + + @Test + fun `holds many distinct ids across resizes, with exact dedup`() { + // Start tiny so it must grow several times. + val seen = SeenIds(initialSlotsPow2 = 4) + val n = 50_000 + repeat(n) { assertTrue(seen.add(id(it)), "id $it should be new") } + assertEquals(n, seen.size()) + // Every one is now a duplicate. + repeat(n) { assertFalse(seen.add(id(it)), "id $it should be a duplicate") } + assertEquals(n, seen.size(), "duplicates don't grow the set") + } + + @Test + fun `only the first 128 bits key the id (differ past 32 hex chars still dedups)`() { + // Same first 128 bits, different tail -> treated as the same (documented tradeoff, ~1e-22 in practice). + val seen = SeenIds() + val prefix = "%032x".format(42) + assertTrue(seen.add(prefix + "0".repeat(32))) + assertFalse(seen.add(prefix + "f".repeat(32)), "same 128-bit prefix collapses") + } + + @Test + fun `the all-zero 128-bit-prefix key (the empty sentinel) is deduped correctly`() { + val seen = SeenIds() + assertTrue(seen.add("0".repeat(64)), "all-zero id is new the first time") + assertFalse(seen.add("0".repeat(64)), "and a duplicate the second") + assertTrue(seen.add(id(5)), "a normal id still works alongside it") + assertEquals(2, seen.size()) + } + + @Test + fun `a malformed id is let through, not skipped`() { + val seen = SeenIds() + assertTrue(seen.add("not-hex"), "malformed -> flows to verify") + assertTrue(seen.add("short")) + } +} From 338c9a41afacd8fef6af46ffdc790641ca06f16e Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 22:52:46 +0000 Subject: [PATCH 6/8] refactor(quartz): make SeenIds single-writer, move to commonMain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop @Synchronized from add/reset/size: SeenIds is now documented as single-writer (not thread-safe). Callers dedup across concurrent relay producers by funneling events into one consumer that owns the instance — the one-consumer ingest pattern used elsewhere — which keeps a single global set, stays lock-free, and lets resize run without coordination. With the JVM-only @Synchronized gone the class is pure common Kotlin (LongArray + Hex.readLong), so it moves from the jvmAndroid source set to commonMain and is now available on every target. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt --- .../com/vitorpamplona/quartz/utils/SeenIds.kt | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) rename quartz/src/{jvmAndroid => commonMain}/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt (88%) diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt similarity index 88% rename from quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt rename to quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt index df952a74ac..51b8a1ec97 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt @@ -36,8 +36,16 @@ package com.vitorpamplona.quartz.utils * Backed by one open-addressed [LongArray] (two longs per slot, `(0,0)` = empty), so * there are NO per-entry objects and the 64-char id [String] is never retained: tens * of millions of ids cost ~16 bytes each (~1 GB at 40M) instead of the ~6 GB a - * `HashSet` of 64-char hex would. [add] is O(1) and `@Synchronized`; the lock - * is held for nanoseconds, so concurrent producers contend little. + * `HashSet` of 64-char hex would. [add] is O(1). + * + * **Not thread-safe — single-writer.** [add] mutates the table (and may resize it) and + * [reset] replaces it, so every call must come from one thread. To dedup across many + * concurrent relay producers, funnel their events into a single consumer that owns the + * SeenIds (the one-consumer ingest pattern used elsewhere in this library): that keeps + * one global set while staying single-writer, and the resize never has to coordinate. + * Giving each producer its own instance is also lock-free, but then dedups only + * *within* that producer, not across them. If you truly need concurrent writers, guard + * it yourself. * * Not unbounded-safe on its own: call [reset] between passes (or whenever the working * set should be forgotten) so a long-running process can't grow the table forever. @@ -68,7 +76,6 @@ class SeenIds( * it). A too-short/malformed id returns true — it flows through and downstream * verification drops it — rather than risk collapsing distinct ids. */ - @Synchronized fun add(idHex: String): Boolean { // Slice the first 128 bits straight to two longs via Hex's table-lookup // reader — no hex text parsing, no allocation. A string too short to slice @@ -116,13 +123,11 @@ class SeenIds( } } - @Synchronized fun reset() { allocate(1 shl INITIAL_POW2) zeroSeen = false } - @Synchronized fun size() = count + if (zeroSeen) 1 else 0 // Ids are already uniform, but avalanche the two halves so the low bits used for From e0ebc8fad69b86029e3d816c03be15ad49b4cc0b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 23:31:09 +0000 Subject: [PATCH 7/8] feat(cli): dedup drainAllPages via SeenIds; unbounded amy fetch --paginate Two changes to the paginated fetch path: - Cross-relay dedup before verify. drainAllPages' single consumer now runs a SeenIds filter: the same widely-mirrored event arrives once per relay, and the repeats are dropped BEFORE the expensive Schnorr verify + store instead of after (they were only trimmed by FetchCommand's distinctBy). An id is marked seen only once it verifies, so a forged copy (valid id, bad sig) delivered first can't suppress the genuine one from another relay. Adds SeenIds.contains (peek without recording) for that check-then-add. - `amy fetch --paginate` no longer forces a --limit. With --limit N it still pages up to N per relay; WITHOUT --limit it drains the whole filter unbounded (the filter's null limit flows straight through). Plain (non-paginate) fetch still trims to the default 100. Verified live: unbounded --paginate over a ~20-min nos.lol firehose window returns 406 (all unique, 3s) vs the old 100 cap; --limit 50 caps at 50; default caps at 100; cross-relay fetch stays count==uniq. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt --- .../com/vitorpamplona/amethyst/cli/Context.kt | 22 +++++++-- .../amethyst/cli/commands/FetchCommand.kt | 48 ++++++++++++------- .../com/vitorpamplona/quartz/utils/SeenIds.kt | 23 +++++++++ .../vitorpamplona/quartz/utils/SeenIdsTest.kt | 12 +++++ 4 files changed, 84 insertions(+), 21 deletions(-) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt index f62e624a1e..bcda8dbcfc 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt @@ -70,6 +70,7 @@ import com.vitorpamplona.quartz.nip61Nutzaps.info.NutzapInfoEvent import com.vitorpamplona.quartz.nip61Nutzaps.nutzap.NutzapEvent import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent import com.vitorpamplona.quartz.nip87Ecash.recommendation.MintRecommendationEvent +import com.vitorpamplona.quartz.utils.SeenIds import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED @@ -487,10 +488,13 @@ class Context( * [fetchAllPagesFromPool] instead of stopping at the first EOSE — so a query * larger than a relay's per-`REQ` cap (strfry's `limit`, ~500) is fully * retrieved instead of silently truncated. Each relay is walked on its own - * `until` cursor, up to [maxConcurrentRelays] at once, and every event still - * funnels through [verifyAndStore]; the result is tagged by relay exactly like - * [drain] (and, like [drain], is NOT deduped across relays — callers dedup by - * id). + * `until` cursor, up to [maxConcurrentRelays] at once, and every event funnels + * through [verifyAndStore]; the result is tagged by the relay that first + * delivered it. Unlike [drain], it IS deduped across relays: the same + * widely-mirrored event arrives once per relay, and the repeats are dropped by a + * [SeenIds] filter BEFORE the expensive verify+store — an id is marked seen only + * after it verifies, so a forged copy (valid id, bad signature) delivered first + * can't suppress the genuine one from another relay. * * Bound the work with the filters' `limit`: each relay pages until it reaches * the limit, so an unbounded filter pages that relay's entire matching history. @@ -511,8 +515,16 @@ class Context( coroutineScope { val consumer = launch { + // One writer → SeenIds' single-writer contract holds. Skip a + // cross-relay duplicate before verifying it; mark it seen only once + // it verifies so a bad-sig copy can't pre-empt a good one. + val seen = SeenIds() for ((relay, event) in eventChannel) { - if (verifyAndStore(event)) collected.add(relay to event) + if (seen.contains(event.id)) continue + if (verifyAndStore(event)) { + seen.add(event.id) + collected.add(relay to event) + } } } try { diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt index 109b347f26..d2222a7962 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt @@ -52,35 +52,42 @@ import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent * NIP-65 write relays, exactly how the app downloads an event/profile from * a shared link. This is nak's `fetch` (nip19-hint resolution). * - * Results are deduplicated by id, sorted newest-first, capped at `--limit` - * (default 100), and emitted as full event JSON under an `events` array. + * Results are deduplicated by id, sorted newest-first, and emitted as full event + * JSON under an `events` array. * - * By default (filter mode) a single `REQ` is drained to EOSE, so a relay that - * caps its response (strfry's per-`REQ` `limit`, ~500) truncates the result. - * `--paginate` (alias `--all`) instead walks each relay page-by-page on `until` - * cursors up to `--limit`, fully draining sets larger than one `REQ` — the - * multi-relay [Context.drainAllPages] path. Code mode is always single-shot. + * By default (filter mode) a single `REQ` is drained to EOSE — so a relay that + * caps its response (strfry's per-`REQ` `limit`, ~500) truncates the result — and + * the output is trimmed to the newest `--limit` (default 100). `--paginate` (alias + * `--all`) instead walks each relay page-by-page on `until` cursors via the + * multi-relay [Context.drainAllPages] path: with `--limit N` it pages up to N per + * relay, and WITHOUT `--limit` it drains the whole filter unbounded (mind broad + * filters — that can be a lot). Code mode is always single-shot. */ object FetchCommand { + /** Output cap for a plain (non-`--paginate`) fetch when `--limit` is omitted. */ + private const val DEFAULT_LIMIT = 100 + suspend fun run( dataDir: DataDir, rest: Array, ): Int { val args = Args(rest) - val limit = args.flag("limit")?.toIntOrNull() ?: 100 - if (limit <= 0) return Output.error("bad_args", "--limit must be > 0") + // `--limit` is optional. When absent, `--paginate` drains the whole filter + // (unbounded) while a plain fetch still trims to DEFAULT_LIMIT. + val explicitLimit = args.flag("limit")?.toIntOrNull() + if (explicitLimit != null && explicitLimit <= 0) return Output.error("bad_args", "--limit must be > 0") val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 8L) * 1000 // Code mode: a nip19/nip05 positional resolves its own relays via the // outbox model rather than using a hand-built filter. args.positionalOrNull(0)?.takeIf { looksLikeCode(it) }?.let { - return fetchByCode(dataDir, it, limit, timeoutMs) + return fetchByCode(dataDir, it, explicitLimit ?: DEFAULT_LIMIT, timeoutMs) } + // buildFilter already carries `--limit` (or null) as the filter's limit, so + // the paginate path uses it verbatim: bounded per relay with --limit, or a + // full drain of the filter without one. val filter = RawEventSupport.buildFilter(args) - // --paginate/--all walks each relay past its per-REQ cap (strfry's ~500) - // by following `until` cursors, bounded by `limit`; default stops at the - // first EOSE like nak's `req`. val paginate = args.bool("paginate") || args.bool("all") Context.open(dataDir).use { ctx -> @@ -90,17 +97,26 @@ object FetchCommand { val received = if (paginate) { - ctx.drainAllPages(relays.associateWith { listOf(filter.copy(limit = limit)) }, timeoutMs) + ctx.drainAllPages(relays.associateWith { listOf(filter) }, timeoutMs) } else { ctx.drain(relays.associateWith { listOf(filter) }, timeoutMs) } - val events = + val ordered = received .asSequence() .map { it.second } .distinctBy { it.id } .sortedByDescending { it.createdAt } - .take(limit) + // An explicit --limit always caps the output. Without it, a single-REQ + // fetch trims to DEFAULT_LIMIT; a --paginate drain returns everything. + val capped = + when { + explicitLimit != null -> ordered.take(explicitLimit) + paginate -> ordered + else -> ordered.take(DEFAULT_LIMIT) + } + val events = + capped .map { Output.mapper.readTree(it.toJson()) } .toList() diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt index 51b8a1ec97..32c1418c8a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt @@ -84,6 +84,29 @@ class SeenIds( return addKey(Hex.readLong(idHex, 0), Hex.readLong(idHex, 16)) } + /** + * Whether [idHex] has already been [add]ed this pass, WITHOUT recording it. A + * too-short/malformed id is never "seen" (returns false) — the mirror of [add] + * letting it through. Use this to check-then-conditionally-add, e.g. to mark an + * id seen only after it verifies (so a forged copy sharing a valid id can't + * pre-empt the genuine one). + */ + fun contains(idHex: String): Boolean { + if (idHex.length < 32) return false + val hi = Hex.readLong(idHex, 0) + val lo = Hex.readLong(idHex, 16) + if (hi == 0L && lo == 0L) return zeroSeen + var i = (mix(hi, lo).toInt() and mask) + while (true) { + val s = i * 2 + val h = table[s] + val l = table[s + 1] + if (h == 0L && l == 0L) return false + if (h == hi && l == lo) return true + i = (i + 1) and mask + } + } + private fun addKey( hi: Long, lo: Long, diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt index 3ac4dc5788..aa619046c0 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt @@ -85,4 +85,16 @@ class SeenIdsTest { assertTrue(seen.add("not-hex"), "malformed -> flows to verify") assertTrue(seen.add("short")) } + + @Test + fun `contains peeks without recording`() { + val seen = SeenIds() + assertFalse(seen.contains(id(9)), "never added -> not contained") + assertFalse(seen.contains(id(9)), "peeking does not add it") + assertEquals(0, seen.size(), "contains must not grow the set") + assertTrue(seen.add(id(9))) + assertTrue(seen.contains(id(9)), "now contained") + assertFalse(seen.contains("short"), "malformed is never contained") + assertEquals(1, seen.size()) + } } From e8f4c5f8068953b328a2634b80986d935c867ad3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 23:57:40 +0000 Subject: [PATCH 8/8] refactor(cli): align fetch default limit across paths; harden paging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit follow-ups before merge: - amy fetch default limit is now the same on both paths: absent --limit → 100 for plain AND --paginate (previously --paginate silently meant "unbounded"). `--limit 0` is the explicit opt-in to drain everything (unbounded); negative is rejected. The effective limit is carried on the filter so both paths agree. - drainAllPages sizes its SeenIds for CLI-scale fetches (initialSlotsPow2 = 12, ~64 KB) instead of the large-walk default (~16 MB eagerly allocated per fetch); it grows if an unbounded drain needs it. - fetchAllPages clamps the inclusive advance to `min(pageMinTs, boundary)` so a misbehaving relay that answers with an event past the requested `until` can't push the cursor upward — the boundary dedup and termination rely on `until` never increasing. No-op for honest relays (they only return events ≤ until). Verified live: default and --paginate both cap at 100; --limit 50 → 50; --limit 0 --paginate drains the full window (>100); paging tests + SeenIds tests still pass. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_015YEbdqCRPkszkGCoi89RMt --- .../com/vitorpamplona/amethyst/cli/Context.kt | 7 ++- .../amethyst/cli/commands/FetchCommand.kt | 52 +++++++++---------- .../NostrClientFetchAllPagesExt.kt | 9 +++- 3 files changed, 37 insertions(+), 31 deletions(-) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt index bcda8dbcfc..dd0bac444e 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt @@ -517,8 +517,11 @@ class Context( launch { // One writer → SeenIds' single-writer contract holds. Skip a // cross-relay duplicate before verifying it; mark it seen only once - // it verifies so a bad-sig copy can't pre-empt a good one. - val seen = SeenIds() + // it verifies so a bad-sig copy can't pre-empt a good one. Start + // small (CLI fetches are typically hundreds of events); it grows if + // an unbounded drain needs it, rather than eagerly taking the + // large-walk default table. + val seen = SeenIds(initialSlotsPow2 = 12) for ((relay, event) in eventChannel) { if (seen.contains(event.id)) continue if (verifyAndStore(event)) { diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt index d2222a7962..9126779780 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt @@ -52,19 +52,20 @@ import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent * NIP-65 write relays, exactly how the app downloads an event/profile from * a shared link. This is nak's `fetch` (nip19-hint resolution). * - * Results are deduplicated by id, sorted newest-first, and emitted as full event - * JSON under an `events` array. + * Results are deduplicated by id, sorted newest-first, capped at `--limit` (default + * 100 — the SAME on both paths), and emitted as full event JSON under an `events` + * array. `--limit 0` removes the cap entirely. * - * By default (filter mode) a single `REQ` is drained to EOSE — so a relay that - * caps its response (strfry's per-`REQ` `limit`, ~500) truncates the result — and - * the output is trimmed to the newest `--limit` (default 100). `--paginate` (alias - * `--all`) instead walks each relay page-by-page on `until` cursors via the - * multi-relay [Context.drainAllPages] path: with `--limit N` it pages up to N per - * relay, and WITHOUT `--limit` it drains the whole filter unbounded (mind broad - * filters — that can be a lot). Code mode is always single-shot. + * By default (filter mode) a single `REQ` is drained to EOSE — so a relay that caps + * its response (strfry's per-`REQ` `limit`, ~500) truncates the result. `--paginate` + * (alias `--all`) instead walks each relay page-by-page on `until` cursors via the + * multi-relay [Context.drainAllPages] path, fully draining sets larger than one + * `REQ`. Both honor the same limit: `--limit N` returns the newest N, absent is 100, + * and `--limit 0` is unbounded — combined with `--paginate` that drains the entire + * filter, so mind broad filters. Code mode is always single-shot. */ object FetchCommand { - /** Output cap for a plain (non-`--paginate`) fetch when `--limit` is omitted. */ + /** Output/paging cap for a fetch (either path) when `--limit` is omitted. */ private const val DEFAULT_LIMIT = 100 suspend fun run( @@ -72,22 +73,25 @@ object FetchCommand { rest: Array, ): Int { val args = Args(rest) - // `--limit` is optional. When absent, `--paginate` drains the whole filter - // (unbounded) while a plain fetch still trims to DEFAULT_LIMIT. + // `--limit`: omitted → DEFAULT_LIMIT on BOTH the plain and --paginate paths; + // `0` → unbounded (drain everything — only useful with --paginate); negative + // → error. `effectiveLimit == null` means "no cap". val explicitLimit = args.flag("limit")?.toIntOrNull() - if (explicitLimit != null && explicitLimit <= 0) return Output.error("bad_args", "--limit must be > 0") + if (explicitLimit != null && explicitLimit < 0) return Output.error("bad_args", "--limit must be >= 0 (0 = unbounded)") + val effectiveLimit: Int? = if (explicitLimit == 0) null else (explicitLimit ?: DEFAULT_LIMIT) val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 8L) * 1000 // Code mode: a nip19/nip05 positional resolves its own relays via the - // outbox model rather than using a hand-built filter. + // outbox model rather than using a hand-built filter. It fetches a single + // entity, so it just uses the (positive) default cap. args.positionalOrNull(0)?.takeIf { looksLikeCode(it) }?.let { - return fetchByCode(dataDir, it, explicitLimit ?: DEFAULT_LIMIT, timeoutMs) + return fetchByCode(dataDir, it, effectiveLimit ?: DEFAULT_LIMIT, timeoutMs) } - // buildFilter already carries `--limit` (or null) as the filter's limit, so - // the paginate path uses it verbatim: bounded per relay with --limit, or a - // full drain of the filter without one. - val filter = RawEventSupport.buildFilter(args) + // Carry the effective limit on the filter so both paths agree: --paginate + // pages each relay up to it (or fully drains the filter when unbounded) and a + // plain fetch asks the relay for that many. + val filter = RawEventSupport.buildFilter(args).copy(limit = effectiveLimit) val paginate = args.bool("paginate") || args.bool("all") Context.open(dataDir).use { ctx -> @@ -107,14 +111,8 @@ object FetchCommand { .map { it.second } .distinctBy { it.id } .sortedByDescending { it.createdAt } - // An explicit --limit always caps the output. Without it, a single-REQ - // fetch trims to DEFAULT_LIMIT; a --paginate drain returns everything. - val capped = - when { - explicitLimit != null -> ordered.take(explicitLimit) - paginate -> ordered - else -> ordered.take(DEFAULT_LIMIT) - } + // effectiveLimit caps the output on both paths; null (--limit 0) is uncapped. + val capped = if (effectiveLimit != null) ordered.take(effectiveLimit) else ordered val events = capped .map { Output.mapper.readTree(it.toJson()) } 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 95b9676efa..18a5502a6c 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 @@ -257,12 +257,17 @@ suspend fun INostrClient.fetchAllPages( // 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) { + // Clamp to `boundary` so a misbehaving relay that answers with an event past + // the requested `until` can't push the cursor UPWARD — the boundary dedup and + // termination both rely on `until` never increasing. Honest relays only + // return events at-or-below `until`, so this is a no-op for them. + val nextUntil = if (boundary != null) minOf(pageMinTs, boundary) else pageMinTs + if (boundary != null && nextUntil == boundary) { seenAtBoundary.addAll(idsAtPageMin) } else { seenAtBoundary = idsAtPageMin } - until = pageMinTs + until = nextUntil } return totalEvents