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..dd0bac444e 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 @@ -69,9 +70,12 @@ 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 +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch import kotlinx.coroutines.selects.select import kotlinx.coroutines.withTimeoutOrNull import okhttp3.OkHttpClient @@ -479,6 +483,67 @@ 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 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. + * 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 { + // 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. 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)) { + seen.add(event.id) + 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..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 @@ -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 * @@ -51,40 +52,69 @@ 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, 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. `--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/paging cap for a fetch (either path) 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`: 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 (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, limit, timeoutMs) + return fetchByCode(dataDir, it, effectiveLimit ?: DEFAULT_LIMIT, timeoutMs) } - 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 -> 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 events = + val received = + if (paginate) { + ctx.drainAllPages(relays.associateWith { listOf(filter) }, timeoutMs) + } else { + ctx.drain(relays.associateWith { listOf(filter) }, timeoutMs) + } + val ordered = received .asSequence() .map { it.second } .distinctBy { it.id } .sortedByDescending { it.createdAt } - .take(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()) } .toList() 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..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 @@ -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,25 +32,49 @@ 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 + * 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. - * @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, @@ -64,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 @@ -86,25 +117,39 @@ 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. - val remainingFilters = - pagedFilters.filterIndexed { index, filter -> - val limit = filter.limit - limit == null || matchCountPerFilter[index] < limit + // 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 activeFilters above and breaks with no + // REQ, so firing this earlier would report a page that never happens. + if (until != null) onNewPage?.invoke(until) val doneChannel = Channel(Channel.CONFLATED) - 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 = @@ -115,20 +160,38 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - // Check if the relay is returning what we asked before moving forward + 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, + // 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 - for (i in pagedFilters.indices) { - val limit = pagedFilters[i].limit - if ((limit == null || matchCountPerFilter[i] < limit) && pagedFilters[i].match(event)) { - matchCountPerFilter[i]++ + var advancesCursor = false + 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 } } if (atLeastOne) { onEvent(event) - pageCount++ - if (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) + } } } } @@ -157,7 +220,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() @@ -170,12 +233,41 @@ 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. + // 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 = nextUntil } return totalEvents 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/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt new file mode 100644 index 0000000000..32c1418c8a --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/utils/SeenIds.kt @@ -0,0 +1,173 @@ +/* + * 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). + * + * **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. + */ +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. + */ + 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)) + } + + /** + * 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, + ): 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 + } + } + + fun reset() { + allocate(1 shl INITIAL_POW2) + zeroSeen = false + } + + 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/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]) + } +} 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..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 @@ -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,124 @@ 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() + } + } + + /** + * 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() + } + } } 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..aa619046c0 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/SeenIdsTest.kt @@ -0,0 +1,100 @@ +/* + * 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")) + } + + @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()) + } +}