mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-09 08:04:45 +00:00
Merge pull request #3489 from vitorpamplona/claude/amy-geode-fetch-sync-compare-0xlsl9
fix(quartz): fetchAllPages paging correctness + amy drainAllPages + SeenIds
This commit is contained in:
Vitor Pamplona
+5
-60
@@ -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<NormalizedRelayUrl>,
|
||||
filters: Map<NormalizedRelayUrl, List<Filter>>,
|
||||
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<Filter>,
|
||||
onNewPage: (Long) -> Unit,
|
||||
onEvent: (Event) -> Unit,
|
||||
): Int = fetchAllPages(relay, filters, RELAY_TIMEOUT_MS, onNewPage, onEvent)
|
||||
}
|
||||
|
||||
+3
-2
@@ -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<NormalizedRelayUrl>,
|
||||
|
||||
@@ -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<NormalizedRelayUrl, List<Filter>>,
|
||||
timeoutMs: Long = 30_000,
|
||||
maxConcurrentRelays: Int = 8,
|
||||
): List<Pair<NormalizedRelayUrl, Event>> {
|
||||
if (filters.isEmpty()) return emptyList()
|
||||
val collected = mutableListOf<Pair<NormalizedRelayUrl, Event>>()
|
||||
// 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<Pair<NormalizedRelayUrl, Event>>(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
|
||||
|
||||
@@ -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 <nevent1…|naddr1…|nprofile1…|npub1…|note1…|name@domain>
|
||||
*
|
||||
@@ -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<String>,
|
||||
): 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()
|
||||
|
||||
|
||||
+118
-26
@@ -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<HexKey>()
|
||||
|
||||
// One subscription id reused for every page. Each page opens it (with the
|
||||
// page's `until`), waits for EOSE, then closes it before the next page opens
|
||||
// it again — so at most one subscription is ever live and the whole download
|
||||
@@ -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<Unit>(Channel.CONFLATED)
|
||||
|
||||
var pageCount = 0
|
||||
// Captured for the listener: the boundary second we re-fetch this page.
|
||||
val boundary = until
|
||||
var received = 0
|
||||
var delivered = 0
|
||||
var pageMinTs = Long.MAX_VALUE
|
||||
val idsAtPageMin = HashSet<HexKey>()
|
||||
|
||||
try {
|
||||
val listener =
|
||||
@@ -115,20 +160,38 @@ suspend fun INostrClient.fetchAllPages(
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
// 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
|
||||
|
||||
+93
@@ -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<NormalizedRelayUrl, List<Filter>>,
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<String>` 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
|
||||
}
|
||||
}
|
||||
+80
@@ -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<Pair<NormalizedRelayUrl, Event>>())
|
||||
val completed = ConcurrentHashMap<NormalizedRelayUrl, Int>()
|
||||
|
||||
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])
|
||||
}
|
||||
}
|
||||
+129
@@ -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<Event>()
|
||||
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<Event>()
|
||||
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<Event>()
|
||||
client.fetchAllPages(defaultRelayUrl, listOf(Filter(kinds = listOf(1)))) { got.add(it) }
|
||||
assertEquals(6, got.size, "every event must be retrieved despite a boundary inside a second")
|
||||
assertEquals(6, got.map { it.id }.toSet().size, "no duplicate deliveries")
|
||||
} finally {
|
||||
client.disconnect()
|
||||
scope.cancel()
|
||||
hub.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A single second denser than the relay's page cap can't be fully drained (its
|
||||
* tail is unreachable — no client-side fix). Paging MUST still step strictly
|
||||
* past it, delivering the events newer and older than it, and MUST terminate
|
||||
* rather than spin re-fetching the same page. Cap 2; A(1000), four events at
|
||||
* t=999, F(998).
|
||||
*/
|
||||
@Test
|
||||
fun denseSecondBeyondCapIsSteppedPastWithoutStalling() =
|
||||
runBlocking {
|
||||
val hub = InProcessRelays(defaultPolicy = { LimitsPolicy(RelayLimits(maxLimit = 2, defaultLimit = 2)) })
|
||||
val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
|
||||
val client = NostrClient(hub, scope)
|
||||
try {
|
||||
val a = SyntheticEvents.fakeEvent(idSeed = 1, createdAt = 1000L)
|
||||
val dense = (2..5).map { SyntheticEvents.fakeEvent(idSeed = it, createdAt = 999L) }
|
||||
val f = SyntheticEvents.fakeEvent(idSeed = 6, createdAt = 998L)
|
||||
hub.getOrCreate(defaultRelayUrl).preload(listOf(a) + dense + listOf(f))
|
||||
|
||||
val got = mutableListOf<Event>()
|
||||
client.fetchAllPages(defaultRelayUrl, listOf(Filter(kinds = listOf(1)))) { got.add(it) }
|
||||
|
||||
val ids = got.map { it.id }.toSet()
|
||||
assertEquals(ids.size, got.size, "no duplicate deliveries")
|
||||
assertTrue(a.id in ids, "the event newer than the dense second must be retrieved")
|
||||
assertTrue(f.id in ids, "the event older than the dense second must be retrieved (stepped past)")
|
||||
assertEquals(2, got.count { it.createdAt == 999L }, "exactly the relay cap of the dense second is reachable")
|
||||
assertEquals(4, got.size, "A + 2-of-4 dense + F")
|
||||
} finally {
|
||||
client.disconnect()
|
||||
scope.cancel()
|
||||
hub.close()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user