From e0ebc8fad69b86029e3d816c03be15ad49b4cc0b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 23:31:09 +0000 Subject: [PATCH] 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()) + } }