diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt index 02b4ee74bc..125643864e 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt @@ -83,6 +83,18 @@ interface INostrClient : AutoCloseable { fun removeConnectionListener(listener: RelayConnectionListener) + /** + * Returns the [IRelayClient] for [url], creating and registering it in the + * connection pool if it is not there yet. + * + * Most callers should never need this — [subscribe]/[count]/[publish] manage + * the pool for you. It exists for accessories that must drive a single relay + * directly, such as NIP-77 negentropy (which sends `NEG-OPEN` and walks the + * reconciliation rounds on one connection). The default implementation throws; + * only a real pool-backed client can hand out relay clients. + */ + fun getOrCreateRelay(url: NormalizedRelayUrl): IRelayClient = throw UnsupportedOperationException("This INostrClient does not expose relay clients") + fun getReqFiltersOrNull(subId: String): Map>? fun getCountFiltersOrNull(subId: String): Map>? diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt index 693aa74661..f87b6622d4 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt @@ -342,6 +342,8 @@ class NostrClient( listeners.forEach { it.onCannotConnect(relay, errorMessage) } } + override fun getOrCreateRelay(url: NormalizedRelayUrl): IRelayClient = relayPool.getOrCreateRelay(url) + override fun addConnectionListener(listener: RelayConnectionListener) { listeners = listeners.plus(listener) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropySyncException.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropySyncException.kt new file mode 100644 index 0000000000..f6018a8098 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropySyncException.kt @@ -0,0 +1,69 @@ +/* + * 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.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl + +/** + * Thrown by [negentropySync] when a relay's matched set cannot be reconciled + * through NIP-77 for a given [window]. + * + * The accessory does NOT silently fall back to plain paging — that is a heavier, + * non-delta transport and the choice belongs to the caller. Catch this and decide: + * page the filter yourself with [fetchAllPages], try another relay, narrow the + * filter, or give up. For the common "try negentropy, else page" shape use + * [negentropySyncOrFetch], which does exactly that (with id-dedup) for you. + * + * [window] is the specific `created_at` slice that failed. When negentropy fails + * on the very first reconcile (e.g. the relay does not speak NIP-77) it equals the + * filter you passed; after windowing it is a sub-range. Note that events from + * windows that DID reconcile before this failure may already have been delivered + * to your `onEvent`, so dedupe by event id if you then page the whole filter. + * + * @property relay the relay that could not reconcile. + * @property window the filter slice that failed. + * @property reason machine-readable category — branch on this to recover. + * @property detail the underlying specifics (a relay's `NEG-ERR` text, `timeout`, …). + */ +class NegentropySyncException( + val relay: NormalizedRelayUrl, + val window: Filter, + val reason: Reason, + val detail: String, +) : Exception("NIP-77 sync of $relay failed ($reason): $detail") { + enum class Reason { + /** + * The relay caps negentropy below the matched set and even a minimal + * `created_at` window still exceeds that cap (strfry's `max_sync_events`), + * so negentropy cannot enumerate the window at all. Paging is the only way + * to get these events. + */ + OVER_MAX_SYNC_EVENTS, + + /** + * The relay did not complete reconciliation: no NIP-77 support, a + * non-overflow `NEG-ERR`, a disconnect, or a timeout. [detail] carries the + * specifics. + */ + UNAVAILABLE, + } +} 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 8bacb7b383..9c09bf586d 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 @@ -64,6 +64,19 @@ suspend fun INostrClient.fetchAllPages( // Track how many matching events each filter has received so far. val matchCountPerFilter = IntArray(filters.size) + // 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 + // occupies a single subscription slot on the connection (relays cap the + // number of concurrent subscriptions per connection, so churning through a + // fresh id per page is wasteful). + // + // Reusing the id is safe because the pool serializes the "send a REQ" + // decision: after each page's EOSE, the pool's auto-resend and this loop's + // unsubscribe+resubscribe can no longer both fire a REQ for the same id (see + // PoolRequests.decideCommandLocked / PoolRequestsConcurrencyTest). Without + // that fix the two raced and produced a duplicate REQ — two EOSEs, or an + // empty page that silently truncated large results. val subId = newSubId() while (true) { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncEventsExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncEventsExt.kt new file mode 100644 index 0000000000..4c064003a6 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncEventsExt.kt @@ -0,0 +1,141 @@ +/* + * 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 com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.buffer +import kotlinx.coroutines.flow.callbackFlow + +/** + * Streaming form of [negentropySync]: emits each event **individually** as it + * arrives, then completes when the sync finishes. Nothing is accumulated, so it + * stays O(1) in memory regardless of how many events the relay holds. + * + * Events are buffered with [Channel.UNLIMITED] because [negentropySync] delivers + * them through a non-suspending callback on the relay reader thread: a bounded + * buffer would force the producer to drop events when the collector lags. A slow + * collector therefore lets the buffer grow — apply your own + * [kotlinx.coroutines.flow.buffer]/`conflate`/`collectLatest` downstream if you + * need a different policy. Cancelling the collector cancels the sync and tears + * down its subscriptions via [awaitClose]. + * + * See [negentropySync] for the meaning of every parameter. + */ +fun INostrClient.negentropySyncEvents( + relay: NormalizedRelayUrl, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + idleTimeoutMs: Long = 120_000L, +): Flow = + callbackFlow { + negentropySync( + relay = relay, + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + idleTimeoutMs = idleTimeoutMs, + ) { event -> + trySend(event) + } + + close() + + awaitClose { } + }.buffer(Channel.UNLIMITED) + +fun INostrClient.negentropySyncEvents( + relay: String, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + idleTimeoutMs: Long = 120_000L, +): Flow = + negentropySyncEvents( + relay = RelayUrlNormalizer.normalize(relay), + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + idleTimeoutMs = idleTimeoutMs, + ) + +/** + * Streaming "try negentropy, else page" — the [Flow] form of + * [negentropySyncOrFetch]. Emits each event individually as it arrives from + * whichever transport delivered it, deduped by id across both phases, then + * completes. Unlike [negentropySyncEvents] it never throws on a relay that can't + * reconcile; it pages instead. + * + * See [negentropySyncEvents] for the buffering/backpressure note and + * [negentropySync] for the meaning of every parameter. + */ +fun INostrClient.negentropySyncOrFetchEvents( + relay: NormalizedRelayUrl, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + idleTimeoutMs: Long = 120_000L, +): Flow = + callbackFlow { + negentropySyncOrFetch( + relay = relay, + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + idleTimeoutMs = idleTimeoutMs, + ) { event -> + trySend(event) + } + + close() + + awaitClose { } + }.buffer(Channel.UNLIMITED) + +fun INostrClient.negentropySyncOrFetchEvents( + relay: String, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + idleTimeoutMs: Long = 120_000L, +): Flow = + negentropySyncOrFetchEvents( + relay = RelayUrlNormalizer.normalize(relay), + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + idleTimeoutMs = idleTimeoutMs, + ) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt new file mode 100644 index 0000000000..a5b218cf76 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt @@ -0,0 +1,726 @@ +/* + * 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.core.HexKey +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +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 com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage +import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.ensureActive +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.joinAll +import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeoutOrNull +import kotlin.concurrent.Volatile +import kotlin.coroutines.coroutineContext +import kotlin.math.min +import kotlin.time.TimeSource + +/** + * Outcome of a successful [negentropySync] run. + * + * @property needCount ids the relay had that we lacked (i.e. everything that + * matched [Filter] on the relay — this sync always reconciles against an empty + * local set, so it downloads the full matched set). + * @property haveCount ids we had that the relay lacked. Always `0` here because + * the local set is empty; kept so the result mirrors a full NIP-77 reconcile. + * @property downloaded distinct events actually delivered through `onEvent`. + * @property windows number of `created_at` windows the matched set was split + * into (`1` when the relay reconciled the whole filter in one shot). + */ +class NegentropySyncResult( + val needCount: Int, + val haveCount: Int, + val downloaded: Int, + val windows: Int, +) + +/** + * Downloads every event a single [relay] holds matching [filter], delivering each + * one (deduped by id) through [onEvent]. A high-level wrapper over NIP-77 + * negentropy that hides the parts that make the raw protocol painful to use: + * + * 1. Reconciles the relay's matched set against an empty local set, **streaming** + * the ids the relay has straight into the download pipeline as each NIP-77 + * round arrives — the full id list is never materialised. + * 2. Downloads those ids through at most [maxConcurrentReqs] concurrent `REQ` + * subscriptions of [fetchBatch] ids each, refilling as each `EOSE` arrives. The + * reconciliation, the id queue and event delivery are all back-pressured, so a + * slow consumer throttles the whole chain and **peak memory is bounded by the + * pipeline depth, not by the window size** — a multi-million-event window + * streams through in roughly constant memory. No id/event dedup set is held: + * NIP-77 yields a distinct id set, so each event is requested (and returned) + * exactly once. + * 3. Handles the relay-side cap on negentropy (strfry's `max_sync_events`, + * observed as `NEG-ERR … "blocked: too many query results"`): the [filter] is + * split by `created_at` windows and each window reconciled on its own, a + * window that still overflows being halved and retried. + * + * This method is negentropy-only. It does NOT silently fall back to plain paging: + * if a window genuinely cannot be reconciled — a minimal `created_at` window still + * over the relay's cap, or a relay that does not speak NIP-77 / drops the session / + * times out — it throws [NegentropySyncException] so the caller chooses what to do. + * For the common "try negentropy, else page" shape, use [negentropySyncOrFetch]. + * + * Scope is controlled entirely by [filter] — narrow it (kinds, authors, `since`, + * tags, …) to download a slice instead of everything. [maxEvents] additionally caps + * the delivered set. + * + * Coroutine-cancellable: on completion, cancel, reaching [maxEvents], or a thrown + * [NegentropySyncException], all `REQ` subscriptions are unsubscribed and the + * negentropy session is closed and its listener removed, so nothing leaks. + * + * @throws NegentropySyncException when a window cannot be reconciled via NIP-77. + * + * @param relay the relay to sync from. + * @param filter what to download. A single filter (NEG-OPEN is single-filter). + * @param maxEvents stop after delivering this many distinct events. `0` = unlimited. + * @param maxConcurrentReqs upper bound on simultaneously-open download `REQ`s. Keep + * it at or below the relay's per-connection subscription cap. + * @param fetchBatch ids per download `REQ`. + * @param idleTimeoutMs the idle watchdog: the maximum time the relay may go + * **completely silent** before the sync gives up. It is NOT a per-round deadline — + * it **resets on every message the relay sends** (each NIP-77 round, every download + * `EOSE`/event) and on connect. So a genuinely slow but progressing sync runs for as + * long as it needs: only true silence trips it. This matters because the relay + * builds its whole negentropy snapshot before the FIRST round responds — O(matched + * set), a minute or more for a multi-million-event filter — and that first wait is a + * real silence, so keep this comfortably above the largest expected first-round build. + * A dead/half-open socket does NOT depend on this: the WebSocket keep-alive detects + * it and the disconnect is turned into a clean abort. Pass `0` to disable the + * watchdog entirely and run until the socket drops (download batches keep a finite + * internal idle bound regardless, so a single stuck batch can't hang the pipeline). + * @param onProgress optional `(needSoFar, downloaded)` ticks as work proceeds. + * @param onEvent called once per distinct event, on the relay reader thread. + */ +suspend fun INostrClient.negentropySync( + relay: NormalizedRelayUrl, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + idleTimeoutMs: Long = 120_000L, + onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, + onEvent: (Event) -> Unit, +): NegentropySyncResult { + var need = 0 + var windows = 0 + var downloaded = 0 + + // Pin the relay in the pool's "desired" set for the whole sync. A NEG-OPEN is not + // a REQ, so during a reconcile round (before that window's first download REQ + // exists) the relay would otherwise look unwanted and the pool would disconnect + // it — fatal mid-sync, and frequent when many small windows each have such a gap. + // A never-matching keep-alive subscription holds the connection open without + // delivering anything. + val keepAliveSubId = newSubId() + subscribe(keepAliveSubId, mapOf(relay to listOf(Filter(ids = listOf(KEEP_ALIVE_ID)))), null) + try { + coroutineScope { + // Bounded funnel: every delivered event passes through this one consumer + // (so onEvent + the maxEvents cap run single-threaded) and the bound + // back-pressures the download workers when the consumer can't keep up. + val events = Channel(DELIVERY_BUFFER) + + val producer = + launch { + try { + syncWindow( + relay = relay, + filter = filter, + idleTimeoutMs = idleTimeoutMs, + fetchBatch = fetchBatch, + maxConcurrentReqs = maxConcurrentReqs, + onWindow = { windows++ }, + // Only accumulate here; progress is reported from the + // single consumer loop below so the user callback is never + // invoked from two coroutines at once. + onNeed = { need += it }, + deliver = { events.send(it) }, + ) + } finally { + events.close() + } + } + + for (event in events) { + downloaded++ + onEvent(event) + onProgress?.invoke(need, downloaded) + if (maxEvents in 1..downloaded) break + } + + // If we broke out early (cap reached) the producer may still be working — + // stop it. If the producer finished normally this is a no-op. + producer.cancel() + } + } finally { + unsubscribe(keepAliveSubId) + } + + return NegentropySyncResult( + needCount = need, + haveCount = 0, + downloaded = downloaded, + windows = windows, + ) +} + +suspend fun INostrClient.negentropySync( + relay: String, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + idleTimeoutMs: Long = 120_000L, + onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, + onEvent: (Event) -> Unit, +): NegentropySyncResult = + negentropySync( + relay = RelayUrlNormalizer.normalize(relay), + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + idleTimeoutMs = idleTimeoutMs, + onProgress = onProgress, + onEvent = onEvent, + ) + +/** + * Result of [negentropySyncOrFetch]. + * + * @property downloaded distinct events delivered through `onEvent` (across whichever + * path ran). + * @property pagedFallback `true` if negentropy could not reconcile and the events + * came from [fetchAllPages] instead. + * @property negentropy the negentropy outcome when it succeeded; `null` on fallback. + * @property fallbackCause why negentropy was abandoned; `null` when it succeeded. + */ +class NegentropyOrFetchResult( + val downloaded: Int, + val pagedFallback: Boolean, + val negentropy: NegentropySyncResult?, + val fallbackCause: NegentropySyncException?, +) + +/** + * "Try negentropy, else page." Runs [negentropySync] and, if it throws + * [NegentropySyncException] (relay can't reconcile the set — no NIP-77 support, an + * over-cap minimal window, a disconnect, …), transparently falls back to + * [fetchAllPages] over the same [filter]. + * + * This is the convenience combinator for the common case where you just want the + * events and don't care which transport delivered them. Events are deduped by id + * across both phases, so anything the negentropy attempt already delivered before + * failing is not delivered again by the paging phase. [maxEvents] is honored across + * both phases. + * + * Use [negentropySync] directly if you want to decide the fallback yourself (try + * another relay, narrow the filter, abort, …) instead of always paging. + */ +suspend fun INostrClient.negentropySyncOrFetch( + relay: NormalizedRelayUrl, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + idleTimeoutMs: Long = 120_000L, + onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, + onEvent: (Event) -> Unit, +): NegentropyOrFetchResult { + val seen = HashSet() + var delivered = 0 + + // Shared dedup + cap across both phases. Returns true if the event was new and + // delivered. Both phases run sequentially, so no concurrent access. + fun accept(event: Event): Boolean { + if ((maxEvents <= 0 || delivered < maxEvents) && seen.add(event.id)) { + delivered++ + onEvent(event) + return true + } + return false + } + + return try { + val result = + negentropySync( + relay = relay, + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + idleTimeoutMs = idleTimeoutMs, + onProgress = onProgress, + ) { accept(it) } + NegentropyOrFetchResult(delivered, pagedFallback = false, negentropy = result, fallbackCause = null) + } catch (e: NegentropySyncException) { + // Negentropy couldn't enumerate the set — page the whole filter instead, + // skipping anything the negentropy attempt already delivered. fetchAllPages + // has no "no timeout" mode, so a disabled watchdog maps to a finite page bound. + val pageFilter = if (maxEvents > 0) filter.copy(limit = maxEvents) else filter + val pageTimeoutMs = if (idleTimeoutMs > 0) idleTimeoutMs else DEFAULT_DOWNLOAD_IDLE_MS + fetchAllPages(relay, listOf(pageFilter), pageTimeoutMs) { event -> + if (accept(event)) onProgress?.invoke(delivered, delivered) + } + NegentropyOrFetchResult(delivered, pagedFallback = true, negentropy = null, fallbackCause = e) + } +} + +suspend fun INostrClient.negentropySyncOrFetch( + relay: String, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + idleTimeoutMs: Long = 120_000L, + onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, + onEvent: (Event) -> Unit, +): NegentropyOrFetchResult = + negentropySyncOrFetch( + relay = RelayUrlNormalizer.normalize(relay), + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + idleTimeoutMs = idleTimeoutMs, + onProgress = onProgress, + onEvent = onEvent, + ) + +/** + * Recursively reconciles [filter] for [relay], splitting by `created_at` windows + * whenever the relay rejects the set as too large, and downloading the ids of each + * window as it resolves. Runs on a single coroutine; [deliver] funnels events out. + * + * Throws [NegentropySyncException] for any window negentropy cannot reconcile (a + * minimal window still over the cap, or an unavailable/erroring relay). + */ +private suspend fun INostrClient.syncWindow( + relay: NormalizedRelayUrl, + filter: Filter, + idleTimeoutMs: Long, + fetchBatch: Int, + maxConcurrentReqs: Int, + onWindow: () -> Unit, + onNeed: (Int) -> Unit, + deliver: suspend (Event) -> Unit, +) { + coroutineContext.ensureActive() + + when (val outcome = downloadWindow(relay, filter, idleTimeoutMs, fetchBatch, maxConcurrentReqs, onNeed, deliver)) { + is ReconcileOutcome.Complete -> onWindow() + + is ReconcileOutcome.Overflow -> { + val lo = filter.since ?: 0L + val hi = filter.until ?: TimeUtils.now() + if (hi - lo <= MIN_WINDOW_SECONDS) { + // A minimal window that still overflows: negentropy genuinely can't + // enumerate this slice. Surface it — paging is the caller's call. + throw NegentropySyncException( + relay = relay, + window = filter, + reason = NegentropySyncException.Reason.OVER_MAX_SYNC_EVENTS, + detail = "created_at window [$lo, $hi] still exceeds the relay's max_sync_events", + ) + } else { + val mid = lo + (hi - lo) / 2 + syncWindow(relay, filter.copy(since = lo, until = mid), idleTimeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onNeed, deliver) + syncWindow(relay, filter.copy(since = mid + 1, until = hi), idleTimeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onNeed, deliver) + } + } + + is ReconcileOutcome.Failed -> + throw NegentropySyncException( + relay = relay, + window = filter, + reason = NegentropySyncException.Reason.UNAVAILABLE, + detail = outcome.detail, + ) + } +} + +private sealed interface ReconcileOutcome { + /** Reconciliation completed; every id was streamed to the downloader. */ + object Complete : ReconcileOutcome + + /** Relay rejected the set as too large (strfry `max_sync_events`). */ + object Overflow : ReconcileOutcome + + /** Reconciliation could not complete; [detail] says why. */ + class Failed( + val detail: String, + ) : ReconcileOutcome +} + +/** + * Drives one NIP-77 reconciliation of [filter] against an EMPTY local set, sending + * `NEG-OPEN` and walking the rounds itself (rather than via [NegentropyManager]) so + * it can apply back-pressure: each round's `needIds` are handed to [sendBatch] — + * which suspends while the download queue is full — *before* the next round is + * acked, so the relay's id stream is paced to the downloader and never piles up. + * + * The ids are streamed, not returned; the result is only the terminal outcome. + * Always sends `NEG-CLOSE` and removes the listener on the way out. + */ +private suspend fun INostrClient.reconcileStreaming( + relay: NormalizedRelayUrl, + filter: Filter, + idleTimeoutMs: Long, + fetchBatch: Int, + onNeed: (Int) -> Unit, + sendBatch: suspend (List) -> Unit, +): ReconcileOutcome { + val targetUrl = relay + val relayClient = getOrCreateRelay(relay) + val subId = newSubId() + val session = NegentropySession(subId, filter, localEvents = emptyList()) + + // Reader-thread → driver hand-off. Holds at most one frame: the relay only sends + // the next one once we ack, and we ack only after this round's ids are queued. + val incoming = Channel(Channel.UNLIMITED) + + // Idle watchdog. Bumped on connect and on EVERY message this relay sends — + // including the download REQs' events, since this is a connection-level listener + // that sees all of them — so any progress anywhere in the pipeline pushes the + // reconcile deadline out. Only true silence trips it. + val clock = IdleClock() + + val listener = + object : RelayConnectionListener { + override fun onConnected( + relay: IRelayClient, + pingMillis: Int, + compressed: Boolean, + ) { + if (relay.url == targetUrl) clock.bump() + } + + override fun onIncomingMessage( + relay: IRelayClient, + msgStr: String, + msg: Message, + ) { + if (relay.url == targetUrl) clock.bump() + when (msg) { + is NegMsgMessage -> if (msg.subId == subId) incoming.trySend(NegFrame.Msg(msg.message)) + is NegErrMessage -> if (msg.subId == subId) incoming.trySend(NegFrame.Err(msg.reason)) + else -> Unit + } + } + + override fun onDisconnected(relay: IRelayClient) { + if (relay.url == targetUrl) incoming.trySend(NegFrame.Err("closed: relay disconnected")) + } + } + + addConnectionListener(listener) + try { + // NEG-OPEN is a one-shot command. Unlike a REQ — which the client replays + // from its active-request state every time a relay (re)connects — a dropped + // NEG-OPEN is never resent, so we must connect and wait until the relay is + // ready before sending it. The connect itself keeps a finite bound even when + // the watchdog is disabled, so an unreachable relay can't hang here forever. + relayClient.connect() + val connectBound = if (idleTimeoutMs > 0) idleTimeoutMs else DEFAULT_CONNECT_TIMEOUT_MS + val connected = + withTimeoutOrNull(connectBound) { + connectedRelaysFlow().first { targetUrl in it } + } + if (connected == null) return ReconcileOutcome.Failed("could not connect within ${connectBound}ms") + + relayClient.sendIfConnected(session.open()) + + while (true) { + // Wait for the relay's next frame, giving up only after idleTimeoutMs of + // total silence (the wait resets whenever the relay sends anything — + // another round, or an event on a download REQ). A disconnect arrives as + // an Err frame, so a dead socket ends this promptly regardless. + val frame = + incoming.receiveWithinIdle(clock, idleTimeoutMs) + ?: return ReconcileOutcome.Failed( + if (idleTimeoutMs > 0) { + "relay went silent for ${idleTimeoutMs}ms mid-reconcile" + } else { + "connection closed before reconcile completed" + }, + ) + + when (frame) { + is NegFrame.Err -> + return if (isOverflow(frame.reason)) ReconcileOutcome.Overflow else ReconcileOutcome.Failed(frame.reason) + + is NegFrame.Msg -> { + val result = session.processMessage(frame.payload) + val needIds = result.needIds + if (needIds.isNotEmpty()) { + onNeed(needIds.size) + var i = 0 + while (i < needIds.size) { + val end = min(i + fetchBatch, needIds.size) + // Copy each batch so the frame's full id list can be freed + // as soon as it is chunked; suspends under back-pressure. + sendBatch(ArrayList(needIds.subList(i, end))) + i = end + } + } + val next = result.nextCmd + if (next != null) { + relayClient.sendIfConnected(next) + } else { + return ReconcileOutcome.Complete + } + } + } + } + } finally { + relayClient.sendIfConnected(session.close()) + removeConnectionListener(listener) + incoming.close() + } +} + +private sealed interface NegFrame { + class Msg( + val payload: String, + ) : NegFrame + + class Err( + val reason: String, + ) : NegFrame +} + +/** + * strfry sends `["NEG-ERR", subId, "blocked: too many query results"]` when a + * NEG-OPEN matches more than `relay__negentropy__maxSyncEvents`. Match that + * verbatim, plus a looser contains-check so equivalent wording from other relays + * still triggers the window split rather than aborting. + */ +private fun isOverflow(reason: String): Boolean = + reason == "blocked: too many query results" || + reason.contains("too many", ignoreCase = true) || + reason.startsWith("blocked", ignoreCase = true) + +/** + * Reconciles [filter] and streams its ids straight into a bounded download pool, so + * reconciliation and download overlap and peak memory stays independent of the + * window's size. At most [maxConcurrentReqs] `REQ`s of [fetchBatch] ids are open at + * once; the id queue is bounded so a slow download back-pressures reconciliation. + * Returns the terminal [ReconcileOutcome]; events go out through [deliver]. + */ +private suspend fun INostrClient.downloadWindow( + relay: NormalizedRelayUrl, + filter: Filter, + idleTimeoutMs: Long, + fetchBatch: Int, + maxConcurrentReqs: Int, + onNeed: (Int) -> Unit, + deliver: suspend (Event) -> Unit, +): ReconcileOutcome = + coroutineScope { + val workerCount = maxConcurrentReqs.coerceAtLeast(1) + + // Bounded: when full, reconcileStreaming suspends instead of letting the + // relay's id stream accumulate. This is what keeps memory O(pipeline), not + // O(window). + val idBatches = Channel>(workerCount) + + val workers = + List(workerCount) { + launch { + for (batch in idBatches) { + coroutineContext.ensureActive() + for (event in fetchByIds(relay, batch, idleTimeoutMs)) { + deliver(event) + } + } + } + } + + val outcome = + reconcileStreaming(relay, filter, idleTimeoutMs, fetchBatch, onNeed) { batch -> + idBatches.send(batch) + } + + idBatches.close() + workers.joinAll() + outcome + } + +/** + * One `REQ` for [batch] ids; collects the matching events and returns them on + * `EOSE`/close/timeout. All events for a single relay arrive on its one reader + * thread, so collecting here needs no synchronisation. + * + * Events are deduped *within this batch* (a [HashSet] bounded by the batch size, so + * still O(pipeline) memory). A REQ-by-ids should return each id once, but the client + * may re-send the REQ on a reconnect/filter-sync mid-flight, which makes the relay + * replay the batch; without this the same event would be delivered twice. We rely on + * NIP-77 yielding a distinct id set across batches, so no global dedup is needed. + */ +private suspend fun INostrClient.fetchByIds( + relay: NormalizedRelayUrl, + batch: List, + idleTimeoutMs: Long, +): List { + val subId = newSubId() + val done = Channel(Channel.CONFLATED) + val collected = ArrayList(batch.size) + val seen = HashSet(batch.size) + + // Per-batch idle clock: each event resets it, so a batch that keeps streaming is + // never cut off, but a batch that stalls (relay stops mid-flight) unblocks after + // the idle bound instead of hanging a worker. A download batch always keeps a + // finite bound even when the caller disabled the whole-sync watchdog. + val clock = IdleClock() + val batchIdleMs = if (idleTimeoutMs > 0) idleTimeoutMs else DEFAULT_DOWNLOAD_IDLE_MS + + val listener = + object : SubscriptionListener { + override fun onEvent( + event: Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + clock.bump() + if (seen.add(event.id)) collected.add(event) + } + + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + done.trySend(Unit) + } + + override fun onClosed( + message: String, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + done.trySend(Unit) + } + + override fun onCannotConnect( + relay: NormalizedRelayUrl, + message: String, + forFilters: List?, + ) { + done.trySend(Unit) + } + } + + try { + subscribe(subId, mapOf(relay to listOf(Filter(ids = batch))), listener) + done.receiveWithinIdle(clock, batchIdleMs) + } finally { + unsubscribe(subId) + done.close() + } + return collected +} + +/** Seconds: a window this small that still overflows can't be split further. */ +private const val MIN_WINDOW_SECONDS = 1L + +/** Bounded buffer between the download workers and the single delivery consumer. */ +private const val DELIVERY_BUFFER = 256 + +/** + * A 32-byte id that no real event can have (all `f`s), used only to hold a + * never-matching keep-alive subscription that keeps the relay connected for the + * duration of a sync. Synthetic/real event ids are SHA-256 digests, so this never + * collides with an actual event. + */ +private val KEEP_ALIVE_ID = "f".repeat(64) + +/** + * Finite fallback bounds (ms) for the two waits that must stay bounded even when the + * whole-sync idle watchdog is disabled (`idleTimeoutMs = 0`): the initial connect, + * and each individual download batch. Keeping these finite means an unreachable relay + * or a single stuck batch can never hang the pipeline, while the reconcile rounds + * still honor "run until the socket drops". + */ +private const val DEFAULT_CONNECT_TIMEOUT_MS = 30_000L +private const val DEFAULT_DOWNLOAD_IDLE_MS = 60_000L + +/** + * Monotonic "last activity" marker for the idle watchdog. [bump] on every sign of + * life from the relay; [elapsedMs] reports the silence since the last bump. + * + * [bump] is on the per-event hot path (the connection listener bumps for every + * message the relay sends — millions during a large download), so it must not + * allocate: a single [start] mark is taken once (unboxed field) and each bump only + * writes a `Long` of nanos-since-start into a `@Volatile` field. Reader threads + * write, the driver coroutine reads — visibility is all we need, so a plain volatile + * Long beats boxing a `ValueTimeMark` into an `AtomicReference` on every event. + */ +private class IdleClock { + private val start = TimeSource.Monotonic.markNow() + + @Volatile + private var lastNanos = 0L + + fun bump() { + lastNanos = start.elapsedNow().inWholeNanoseconds + } + + fun elapsedMs(): Long = (start.elapsedNow().inWholeNanoseconds - lastNanos) / 1_000_000 +} + +/** + * Receives the next item, giving up (returning `null`) only after [idleMs] elapse with + * no activity on [clock]. Because [clock] is bumped by *any* relay message — not just + * items on this channel — unrelated progress (e.g. download events arriving during a + * reconcile wait) keeps pushing the deadline out. [idleMs] `<= 0` disables the + * watchdog: it waits until an item arrives (a disconnect is delivered as an item, so + * a dead socket still unblocks it). + */ +private suspend fun Channel.receiveWithinIdle( + clock: IdleClock, + idleMs: Long, +): T? { + if (idleMs <= 0) return receive() + while (true) { + val remaining = idleMs - clock.elapsedMs() + if (remaining <= 0) return null + val item = withTimeoutOrNull(remaining) { receive() } + if (item != null) return item + // Timed out with nothing on this channel. If other activity bumped the clock + // meanwhile, the next `remaining` is positive and we wait again; otherwise it + // is <= 0 on the next iteration and we give up. + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt index 5bf4dff727..8ed3822081 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt @@ -35,6 +35,8 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.flow.MutableStateFlow +import kotlin.concurrent.atomics.AtomicBoolean +import kotlin.concurrent.atomics.ExperimentalAtomicApi /** * Manages relay subscriptions for the entire pool in a way that only @@ -43,6 +45,7 @@ import kotlinx.coroutines.flow.MutableStateFlow * This code also awaits a subscription to come to EOSE since many relays * have through switching subs while they are processing the past. */ +@OptIn(ExperimentalAtomicApi::class) class PoolRequests { /** * Desired subs and listeners @@ -64,6 +67,42 @@ class PoolRequests { fun subState(subId: String): RequestSubscriptionState = relayState.getOrCreate(subId) { RequestSubscriptionState() } + /** + * Serializes every access to the subscription state machine + * ([RequestSubscriptionState]) and the "should I send a REQ?" decision. + * + * A single subscription can span many relays, and each relay's + * socket-reader thread delivers messages into this class concurrently while + * the app thread adds/removes subscriptions — so the plain maps inside + * [RequestSubscriptionState] are written from several threads at once. That + * is both a memory hazard (concurrent map mutation) and, more importantly, + * a logic hazard: the check-then-send in [decideCommandLocked] must be + * atomic, otherwise two threads can both observe "no REQ in flight" and both + * send a REQ for the same sub id. + * + * This is a tiny non-reentrant spin lock (the same [AtomicBoolean] primitive + * used by BasicRelayClient's connecting mutex): the critical sections are a + * handful of map operations, never any I/O. Listener callbacks and the + * actual socket sends are ALWAYS performed outside the lock — they re-enter + * this class through [onSent], so holding the lock across them would + * self-deadlock. + */ + private val stateLock = AtomicBoolean(false) + + private inline fun withStateLock(block: () -> R): R { + while (stateLock.exchange(true)) { + // Another thread holds the lock. Spin-read until it looks free + // (test-and-test-and-set: cheaper on the cache line than hammering + // exchange) then retry the acquisition above. + while (stateLock.load()) { } + } + try { + return block() + } finally { + stateLock.store(false) + } + } + /** * This is called when a sub is added or removed from this class and * should update the desired relay list to get the pool to connect @@ -150,8 +189,10 @@ class PoolRequests { */ fun onConnecting(url: NormalizedRelayUrl) { // Change states to connecting. - relayState.forEach { subId, state -> - state.connecting(url) + withStateLock { + relayState.forEach { subId, state -> + state.connecting(url) + } } } @@ -164,7 +205,9 @@ class PoolRequests { ) { when (cmd) { is ReqCmd -> { - subState(cmd.subId).onOpenReq(relay, cmd.filters) + withStateLock { + subState(cmd.subId).onOpenReq(relay, cmd.filters) + } desiredSubListeners.get(cmd.subId)?.onSubscriptionStarted( relay = relay.url, forFilters = cmd.filters, @@ -172,7 +215,9 @@ class PoolRequests { } is CloseCmd -> { - subState(cmd.subId).onSubscriptionClosed(relay) + withStateLock { + subState(cmd.subId).onSubscriptionClosed(relay) + } desiredSubListeners.get(cmd.subId)?.onSubscriptionClosed( relay = relay.url, ) @@ -189,46 +234,63 @@ class PoolRequests { ) { when (msg) { is EventMessage -> { - val state = relayState.get(msg.subId) - state?.onNewEvent(relay.url) + var isLive = false + var forFilters: List? = null + withStateLock { + val state = relayState.get(msg.subId) + state?.onNewEvent(relay.url) + isLive = state?.currentState(relay.url) == ReqSubStatus.LIVE + forFilters = state?.lastKnownFilterStates(relay.url) + } desiredSubListeners.get(msg.subId)?.onEvent( event = msg.event, - isLive = state?.currentState(relay.url) == ReqSubStatus.LIVE, + isLive = isLive, relay = relay.url, - forFilters = state?.lastKnownFilterStates(relay.url), + forFilters = forFilters, ) } is EoseMessage -> { - val state = relayState.get(msg.subId) - state?.onEose(relay.url) + var forFilters: List? = null + val cmd = + withStateLock { + val state = relayState.get(msg.subId) + state?.onEose(relay.url) + forFilters = state?.lastKnownFilterStates(relay.url) + // Decide (and pre-mark) the resend while still holding the + // lock, so a concurrent subscribe/unsubscribe on the app + // thread can't also decide to send a REQ for this sub. + decideCommandLocked(msg.subId, relay.url) + } desiredSubListeners.get(msg.subId)?.onEose( relay = relay.url, - forFilters = state?.lastKnownFilterStates(relay.url), + forFilters = forFilters, ) // send a newer version when done - sendToRelayIfChanged(msg.subId, relay.url) { cmd -> + if (cmd != null) { relay.sendOrConnectAndSync(cmd) } } is ClosedMessage -> { - val state = relayState.get(msg.subId) - state?.onClosed(relay.url) - + var forFilters: List? = null + val cmd = + withStateLock { + val state = relayState.get(msg.subId) + state?.onClosed(relay.url) + forFilters = state?.lastKnownFilterStates(relay.url) + decideCommandLocked(msg.subId, relay.url) + } desiredSubListeners.get(msg.subId)?.onClosed( message = msg.message, relay = relay.url, - forFilters = state?.lastKnownFilterStates(relay.url), + forFilters = forFilters, ) - // send a newer version when done - sendToRelayIfChanged(msg.subId, relay.url) { cmd -> - // don't send a close if just closed - if (cmd !is CloseCmd) { - relay.sendOrConnectAndSync(cmd) - } + // send a newer version when done, but don't send a close if just closed + if (cmd != null && cmd !is CloseCmd) { + relay.sendOrConnectAndSync(cmd) } } } @@ -238,8 +300,10 @@ class PoolRequests { * When the relay disconnects */ fun onDisconnected(url: NormalizedRelayUrl) { - relayState.forEach { subId, state -> - state.disconnected(url) + withStateLock { + relayState.forEach { subId, state -> + state.disconnected(url) + } } } @@ -262,16 +326,27 @@ class PoolRequests { url: NormalizedRelayUrl, errorMessage: String, ) { - relayState.forEach { subId, state -> - // These are all my subs.. need to figure out which relays have them - val subs = desiredSubs.get(subId) - if (subs != null && url in subs.keys) { - desiredSubListeners.get(subId)?.onCannotConnect( - relay = url, - message = errorMessage, - forFilters = state.lastKnownFilterStates(url), - ) + // Snapshot the affected subs (and their last-known filters) under the + // lock, then notify listeners outside it. + val toNotify = + withStateLock { + val list = mutableListOf?>>() + relayState.forEach { subId, state -> + // These are all my subs.. need to figure out which relays have them + val subs = desiredSubs.get(subId) + if (subs != null && url in subs.keys) { + list.add(subId to state.lastKnownFilterStates(url)) + } + } + list } + + toNotify.forEach { (subId, forFilters) -> + desiredSubListeners.get(subId)?.onCannotConnect( + relay = url, + message = errorMessage, + forFilters = forFilters, + ) } } @@ -281,54 +356,65 @@ class PoolRequests { sync: (NormalizedRelayUrl, Command) -> Unit, ) { relaysToUpdate.forEach { relay -> - sendToRelayIfChanged(subId, relay) { cmd -> - if (cmd is ReqCmd) { - val currentState = relayState.get(subId)?.currentState(relay) - - if (currentState == ReqSubStatus.SENT || currentState == ReqSubStatus.QUERYING_PAST) { - // sending multiple REQs triggers multiple EOSEs back and we then don't know which - // one is which. - } else { - sync(relay, cmd) - } - } else { - sync(relay, cmd) - } + // Decide + pre-mark atomically under the lock, then send outside it. + val cmd = withStateLock { decideCommandLocked(subId, relay) } + if (cmd != null) { + sync(relay, cmd) } } } - fun sendToRelayIfChanged( + /** + * Decides which command (if any) must be sent to [relay] to bring it in line + * with the desired filters for [subId], and — for a REQ — pre-marks the + * subscription state as SENT before returning. + * + * Pre-marking is what makes the check-then-send atomic: a second thread that + * runs this method for the same sub sees the SENT state (or the + * already-updated filters) and declines to send a duplicate REQ. Two REQs on + * one sub id race on the wire and produce duplicate EOSEs/events (or, if a + * CLOSE interleaves, an empty result that silently truncates a paged + * download) — that is the bug this guards against. + * + * MUST be called while holding [withStateLock]. + */ + private fun decideCommandLocked( subId: String, relay: NormalizedRelayUrl, - sync: (Command) -> Unit, - ) { + ): Command? { val state = relayState.get(subId) val oldFilters = state?.currentFilters(relay) val newFilters = desiredSubs.get(subId)?.get(relay) - sendToRelayIfChanged(subId, oldFilters, newFilters, sync) - } - fun sendToRelayIfChanged( - subId: String, - oldFilters: List?, - newFilters: List?, - sync: (Command) -> Unit, - ) { - if (newFilters.isNullOrEmpty()) { - // some relays are not in this sub anymore. Stop their subscriptions - if (!oldFilters.isNullOrEmpty()) { - // only update if the old filters are not already closed. - sync(CloseCmd(subId)) + return when { + newFilters.isNullOrEmpty() -> { + // some relays are not in this sub anymore. Stop their subscriptions + // only if the old filters are not already closed. + if (!oldFilters.isNullOrEmpty()) CloseCmd(subId) else null + } + + oldFilters.isNullOrEmpty() || FiltersChanged.needsToResendRequest(oldFilters, newFilters) -> { + // A REQ is warranted: a brand new sub, or the filters changed + // enough (not just a `since` bump) to need a resend. But if a REQ + // is already in flight, don't send another — multiple REQs on one + // sub id trigger multiple EOSEs and we can no longer tell which + // reply belongs to which REQ. The pending change is picked up + // later by the EOSE handler, which runs this method again once the + // sub reaches LIVE. + val current = state?.currentState(relay) + if (current == ReqSubStatus.SENT || current == ReqSubStatus.QUERYING_PAST) { + null + } else { + // Pre-mark SENT + filters so a concurrent decider skips. + subState(subId).onOpenReq(relay, newFilters) + ReqCmd(subId, newFilters) + } + } + + else -> { + // Filters are effectively the same; nothing to do. + null } - } else if (oldFilters.isNullOrEmpty()) { - // new relays were added. Start a new sub in them - sync(ReqCmd(subId, newFilters)) - } else if (FiltersChanged.needsToResendRequest(oldFilters, newFilters)) { - // filters were changed enough (not only an update in since) to warn a new update - sync(ReqCmd(subId, newFilters)) - } else { - // They are the same don't do anything. } } diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt new file mode 100644 index 0000000000..04734e6162 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt @@ -0,0 +1,309 @@ +/* + * 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.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.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySync +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncEvents +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncOrFetch +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class NostrClientNegentropySyncTest : RelayClientTest() { + @Test + fun fullDownloadDeliversEveryEvent() = + runBlocking { + defaultRelay.preload(SyntheticEvents.batch(20, kind = 1)) + + val got = mutableListOf() + val result = + withTimeout(20_000) { + client.negentropySync( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + ) { got.add(it) } + } + + assertEquals(20, got.size, "every event should be delivered") + assertEquals(20, got.map { it.id }.toSet().size, "no duplicates") + assertEquals(20, result.needCount) + assertEquals(0, result.haveCount) + assertEquals(20, result.downloaded) + assertEquals(1, result.windows, "small set reconciles in a single window") + } + + @Test + fun maxEventsCapsDelivery() = + runBlocking { + defaultRelay.preload(SyntheticEvents.batch(20, kind = 1)) + + val got = mutableListOf() + val result = + withTimeout(20_000) { + client.negentropySync( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + maxEvents = 10, + fetchBatch = 5, + ) { got.add(it) } + } + + assertEquals(10, got.size, "delivery stops at maxEvents") + assertEquals(10, result.downloaded) + } + + @Test + fun cleanTeardownLeavesNoSubscriptions() = + runBlocking { + defaultRelay.preload(SyntheticEvents.batch(15, kind = 1)) + + withTimeout(20_000) { + client.negentropySync( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + fetchBatch = 4, + ) { } + } + + assertTrue( + client.activeRequests(defaultRelayUrl).isEmpty(), + "all download subscriptions must be closed after the sync", + ) + } + + @Test + fun flowVariantStreamsEachEvent() = + runBlocking { + defaultRelay.preload(SyntheticEvents.batch(12, kind = 1)) + + val events = + withTimeout(20_000) { + client + .negentropySyncEvents( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + ).toList() + } + + assertEquals(12, events.size) + assertEquals(12, events.map { it.id }.toSet().size) + } + + /** + * Forces the relay to split its NEG-MSG responses into many small frames + * (`frameSizeLimit` at the library floor) so reconciliation spans many rounds, + * and downloads through a small, bounded pipeline (`fetchBatch`/`maxConcurrentReqs`). + * Exercises the streaming + back-pressure path end to end: every event must still + * be delivered exactly once with nothing accumulated. + */ + @Test + fun multiRoundReconcileStreamsEveryEventThrough() = + runBlocking { + val hub = InProcessRelays(negentropySettings = NegentropySettings(frameSizeLimit = 4096)) + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(hub, scope) + try { + val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7784/") + hub.getOrCreate(url).preload(SyntheticEvents.batch(1500, kind = 1)) + + val got = mutableListOf() + val result = + withTimeout(60_000) { + client.negentropySync( + relay = url, + filter = Filter(kinds = listOf(1)), + fetchBatch = 50, + maxConcurrentReqs = 4, + ) { got.add(it) } + } + + assertEquals(1500, got.size, "every event delivered across many reconcile rounds") + assertEquals(1500, got.map { it.id }.toSet().size, "each exactly once") + assertEquals(1500, result.downloaded) + assertEquals(1500, result.needCount) + } finally { + client.disconnect() + scope.cancel() + hub.close() + } + } + + /** + * A relay that caps negentropy below the matched-set size (strfry's + * `max_sync_events`) but whose events are spread across distinct `created_at` + * values. Windowing alone resolves the cap — each window ends up under it — so + * the sync completes purely via negentropy, no exception, no paging. + */ + @Test + fun overCapWithSpreadTimestampsSucceedsViaWindowing() = + runBlocking { + val hub = InProcessRelays(negentropySettings = NegentropySettings(maxSyncEvents = 3)) + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(hub, scope) + try { + val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7781/") + // 12 events at distinct created_at: windowing can split until each + // window holds <= the cap. + hub.getOrCreate(url).preload(SyntheticEvents.batch(12, kind = 1)) + + val got = mutableListOf() + val result = + withTimeout(60_000) { + client.negentropySync( + relay = url, + filter = Filter(kinds = listOf(1)), + ) { got.add(it) } + } + + assertEquals(12, got.map { it.id }.toSet().size, "all events reconciled via windowing") + assertEquals(12, result.downloaded) + assertTrue(result.windows > 1, "the set must be split into multiple created_at windows") + } finally { + client.disconnect() + scope.cancel() + hub.close() + } + } + + /** + * A relay that caps negentropy below the matched-set size AND whose events all + * share one `created_at`, so no `created_at` window can separate them. Even the + * minimal window stays over the cap, so [negentropySync] cannot reconcile it and + * throws [NegentropySyncException] (reason OVER_MAX_SYNC_EVENTS) rather than + * silently paging — the fallback is the caller's call. + */ + @Test + fun overCapMinimalWindowThrowsInsteadOfPaging() = + runBlocking { + val hub = InProcessRelays(negentropySettings = NegentropySettings(maxSyncEvents = 3)) + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(hub, scope) + try { + val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7782/") + val events = (1..10).map { SyntheticEvents.fakeEvent(idSeed = it, kind = 1, createdAt = 1000L) } + hub.getOrCreate(url).preload(events) + + val thrown = + assertFailsWith { + withTimeout(60_000) { + client.negentropySync( + relay = url, + filter = Filter(kinds = listOf(1)), + ) { } + } + } + assertEquals(NegentropySyncException.Reason.OVER_MAX_SYNC_EVENTS, thrown.reason) + + // And the caller can recover by paging it themselves. + val paged = mutableListOf() + withTimeout(60_000) { + client.fetchAllPages(url, listOf(Filter(kinds = listOf(1)))) { paged.add(it) } + } + assertEquals(10, paged.map { it.id }.toSet().size) + } finally { + client.disconnect() + scope.cancel() + hub.close() + } + } + + /** + * The "try negentropy, else page" combinator: against the same over-cap relay + * where raw [negentropySync] throws, [negentropySyncOrFetch] transparently pages + * and delivers every event, reporting that it fell back. + */ + @Test + fun orFetchPagesWhenNegentropyCannotReconcile() = + runBlocking { + val hub = InProcessRelays(negentropySettings = NegentropySettings(maxSyncEvents = 3)) + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(hub, scope) + try { + val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7783/") + val events = (1..10).map { SyntheticEvents.fakeEvent(idSeed = it, kind = 1, createdAt = 1000L) } + hub.getOrCreate(url).preload(events) + + val got = mutableListOf() + val result = + withTimeout(60_000) { + client.negentropySyncOrFetch( + relay = url, + filter = Filter(kinds = listOf(1)), + ) { got.add(it) } + } + + assertEquals(10, got.map { it.id }.toSet().size, "all events delivered via the paging fallback") + assertEquals(10, result.downloaded) + assertTrue(result.pagedFallback, "it should have fallen back to paging") + assertEquals( + NegentropySyncException.Reason.OVER_MAX_SYNC_EVENTS, + result.fallbackCause?.reason, + ) + } finally { + client.disconnect() + scope.cancel() + hub.close() + } + } + + /** + * On a relay that reconciles fine, [negentropySyncOrFetch] uses negentropy and + * does not page. + */ + @Test + fun orFetchUsesNegentropyWhenItWorks() = + runBlocking { + defaultRelay.preload(SyntheticEvents.batch(8, kind = 1)) + + val got = mutableListOf() + val result = + withTimeout(20_000) { + client.negentropySyncOrFetch( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + ) { got.add(it) } + } + + assertEquals(8, got.size) + assertFalse(result.pagedFallback, "negentropy should have handled it") + assertEquals(8, result.negentropy?.downloaded) + } +} diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt new file mode 100644 index 0000000000..6d21d526e3 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt @@ -0,0 +1,146 @@ +/* + * 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.quartz.nip01Core.relay.client.pool.PoolRequests +import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +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 java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.AtomicInteger +import kotlin.concurrent.thread +import kotlin.test.Test +import kotlin.test.assertEquals + +/** + * Regression guard for the shared-sub-id double-REQ race in [PoolRequests]. + * + * A single subscription id is driven from two threads at once: the app thread + * (the subscribe path, [PoolRequests.sendToRelayIfChanged]) and the relay reader + * thread (an EOSE that triggers an auto-resend, [PoolRequests.onIncomingMessage]). + * The subscription is already LIVE and its desired filters have just changed, so + * both threads independently conclude "the filters changed, send a REQ". + * + * The bug: the decision (read state) and the send (mark state SENT via onSent) + * were not atomic, so the reader could read the pre-send state (filters still on + * the previous value) while the app had already moved the desired filters + * forward — and both would send a REQ for the same sub id. Two REQs on one id + * race on the wire: the relay answers with two EOSEs and duplicate events, or — + * if a CLOSE interleaves — an empty result that silently truncates a paged + * download (this is what broke `fetchAllPages` on large sets). + * + * The fix makes the "should I send a REQ?" decision pre-mark the state + * atomically, so exactly one REQ is ever produced. This test pins the exact + * interleaving the bug needs (app has produced its REQ but not yet run onSent) + * open and asserts only one REQ comes out. + */ +class PoolRequestsConcurrencyTest { + private class FakeRelay( + override val url: NormalizedRelayUrl, + val onCmd: (Command) -> Unit, + ) : IRelayClient { + override fun connect() {} + + override fun needsToReconnect() = false + + override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) {} + + override fun isConnected() = true + + override fun sendOrConnectAndSync(cmd: Command) = onCmd(cmd) + + override fun sendIfConnected(cmd: Command) = onCmd(cmd) + + override fun disconnect() {} + } + + @Test + fun concurrentEoseResendAndSubscribeSendExactlyOneReq() { + val url = RelayUrlNormalizer.normalize("ws://race/") + val subId = "shared-sub" + val filtersA = listOf(Filter(kinds = listOf(1))) + val filtersB = listOf(Filter(kinds = listOf(2))) + val listener = object : SubscriptionListener {} + + // Many episodes so a regression that only sometimes doubles still trips. + repeat(300) { episode -> + val pool = PoolRequests() + val reqBCount = AtomicInteger(0) + + fun countReqB(cmd: Command) { + if (cmd is ReqCmd && cmd.filters == filtersB) reqBCount.incrementAndGet() + } + + val fakeRelay = + FakeRelay(url) { cmd -> + // relay-reader auto-resend send path + countReqB(cmd) + pool.onSent(url, cmd) + } + + // Bring the sub to LIVE with filters A. + val setupRelays = pool.addOrUpdate(subId, mapOf(url to filtersA), listener) + pool.sendToRelayIfChanged(subId, setupRelays) { _, cmd -> pool.onSent(url, cmd) } + pool.onIncomingMessage(fakeRelay, EoseMessage(subId)) + + // The desired filters change to B (e.g. the next page of a paged download). + pool.addOrUpdate(subId, mapOf(url to filtersB), listener) + + val appProducedReq = CountDownLatch(1) + val readerDone = CountDownLatch(1) + + val appThread = + thread { + pool.sendToRelayIfChanged(subId, setOf(url)) { _, cmd -> + countReqB(cmd) + // App has produced its REQ(B); park before onSent so the + // subscription state is not yet advanced — the exact window + // the race needs. + appProducedReq.countDown() + readerDone.await() + pool.onSent(url, cmd) + } + } + + val readerThread = + thread { + appProducedReq.await() + pool.onIncomingMessage(fakeRelay, EoseMessage(subId)) + readerDone.countDown() + } + + appThread.join() + readerThread.join() + + assertEquals( + 1, + reqBCount.get(), + "episode $episode: exactly one REQ must be sent for the changed filters, " + + "never a duplicate from the app + reader race", + ) + } + } +}