From abfea4e7b6841389c56bc966c8609d6d6ea5cae5 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 21:48:11 +0000 Subject: [PATCH 1/9] feat(quartz): add high-level negentropy sync-and-download accessory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `INostrClient.negentropySync` / `negentropySyncAsFlow`, a high-level NIP-77 accessory that downloads every event a relay holds matching a `Filter` and delivers each (deduped) through `onEvent` — mirroring the existing `fetchAllPages` accessory so downstream apps stop hand-rolling the `NegentropyManager` dance. It encapsulates the parts that make raw negentropy painful: - reconciles the relay's matched set (empty local set) via NegentropyManager - downloads the resulting ids through a bounded pool of concurrent REQs (`maxConcurrentReqs` subs of `fetchBatch` ids each, refilled on EOSE) - handles the relay-side cap (strfry `max_sync_events`, `NEG-ERR "blocked: too many query results"`) by splitting the filter into adaptive created_at windows; a minimal window that still can't reconcile (or a relay that doesn't speak NIP-77) falls back to `fetchAllPages` and reports it via `NegentropySyncResult.fellBackToPaging` - caps delivery at `maxEvents`, dedupes through a single consumer, and tears down all subscriptions + the neg session on completion/cancel Scope is controlled entirely by the `Filter` (per maintainer guidance the caller-supplied local-id delta interface is dropped in favour of a custom Filter), so the common call is one line. To drive NEG-OPEN on a single connection, add `INostrClient.getOrCreateRelay(url)` (default throws; NostrClient delegates to the pool). Because NEG-OPEN is a one-shot command that — unlike a REQ — is never replayed on reconnect, the accessory connects and waits for the relay to be ready before opening the session. Tests (quartz jvmAndroidTest, in-process relay): full download, maxEvents cap, clean teardown / no leaked subs, the Flow variant, and window-split + paging fallback against a relay that rejects the full reconcile. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../nip01Core/relay/client/INostrClient.kt | 12 + .../nip01Core/relay/client/NostrClient.kt | 2 + .../NostrClientNegentropySyncAsFlowExt.kt | 84 ++++ .../NostrClientNegentropySyncExt.kt | 466 ++++++++++++++++++ .../relay/NostrClientNegentropySyncTest.kt | 166 +++++++ 5 files changed, 730 insertions(+) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncAsFlowExt.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt 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/NostrClientNegentropySyncAsFlowExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncAsFlowExt.kt new file mode 100644 index 0000000000..298435ec1c --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncAsFlowExt.kt @@ -0,0 +1,84 @@ +/* + * 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.awaitClose +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.callbackFlow + +/** + * Flow form of [negentropySync]: runs the sync while collected and emits the + * accumulated events as a growing list on each new arrival, then completes when + * the sync finishes (mirroring [com.vitorpamplona.quartz.nip01Core.relay.client.reqs.fetchAsFlow]). + * Cancelling the collector cancels the sync and tears down its subscriptions via + * [awaitClose]. + * + * See [negentropySync] for the meaning of every parameter. + */ +fun INostrClient.negentropySyncAsFlow( + relay: NormalizedRelayUrl, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + timeoutMs: Long = 30_000L, +): Flow> = + callbackFlow { + var current = listOf() + + negentropySync( + relay = relay, + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + timeoutMs = timeoutMs, + ) { event -> + current = current + event + trySend(current) + } + + close() + + awaitClose { } + } + +fun INostrClient.negentropySyncAsFlow( + relay: String, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + timeoutMs: Long = 30_000L, +): Flow> = + negentropySyncAsFlow( + relay = RelayUrlNormalizer.normalize(relay), + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + timeoutMs = timeoutMs, + ) 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..6a9a092259 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt @@ -0,0 +1,466 @@ +/* + * 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.reqs.SubscriptionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId +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.INegentropyListener +import com.vitorpamplona.quartz.nip77Negentropy.NegentropyManager +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.launch +import kotlinx.coroutines.withTimeoutOrNull +import kotlin.coroutines.coroutineContext +import kotlin.math.min + +/** + * Outcome of a [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). + * @property fellBackToPaging `true` if at least one window could not be + * reconciled by negentropy (relay rejected it, e.g. strfry's + * `max_sync_events`, or did not support NIP-77) and was downloaded via + * [fetchAllPages] instead. + */ +class NegentropySyncResult( + val needCount: Int, + val haveCount: Int, + val downloaded: Int, + val windows: Int, + val fellBackToPaging: Boolean, +) + +/** + * 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 via a + * [NegentropyManager], accumulating the ids the relay has (`needIds`) across + * rounds until completion. + * 2. Downloads those ids through at most [maxConcurrentReqs] concurrent `REQ` + * subscriptions of [fetchBatch] ids each, refilling as each `EOSE` arrives, + * so a huge set never opens thousands of subs at 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 is halved and retried; a minimal window that + * still cannot reconcile (or a relay that does not speak NIP-77) falls back to + * [fetchAllPages] for that window and sets [NegentropySyncResult.fellBackToPaging]. + * + * Scope is controlled entirely by [filter] — narrow it (kinds, authors, `since`, + * tags, …) to download a slice instead of everything. The accessory keeps memory + * bounded by reconciling and downloading one window at a time and by capping the + * delivered set at [maxEvents]. + * + * Coroutine-cancellable: on completion, cancel, or reaching [maxEvents] all `REQ` + * subscriptions are unsubscribed and the negentropy session is closed and its + * listener removed, so nothing leaks. + * + * @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 timeoutMs max wait for a single reconcile round or download page's `EOSE`. + * @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, + timeoutMs: Long = 30_000L, + onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, + onEvent: (Event) -> Unit, +): NegentropySyncResult { + var need = 0 + var windows = 0 + var fellBack = false + + var downloaded = 0 + val seen = HashSet() + + coroutineScope { + // Single funnel for every delivered event (from REQ batches AND from the + // paging fallback), so dedup + the maxEvents cap + onEvent run on one + // coroutine even though the relay reader threads produce concurrently. + val events = Channel(Channel.UNLIMITED) + + val producer = + launch { + try { + syncWindow( + relay = relay, + filter = filter, + timeoutMs = timeoutMs, + fetchBatch = fetchBatch, + maxConcurrentReqs = maxConcurrentReqs, + onWindow = { windows++ }, + onPaged = { fellBack = true }, + // 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.trySend(it) }, + ) + } finally { + events.close() + } + } + + for (event in events) { + if (seen.add(event.id)) { + 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() + } + + return NegentropySyncResult( + needCount = need, + haveCount = 0, + downloaded = downloaded, + windows = windows, + fellBackToPaging = fellBack, + ) +} + +suspend fun INostrClient.negentropySync( + relay: String, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + timeoutMs: Long = 30_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, + timeoutMs = timeoutMs, + 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. + */ +private suspend fun INostrClient.syncWindow( + relay: NormalizedRelayUrl, + filter: Filter, + timeoutMs: Long, + fetchBatch: Int, + maxConcurrentReqs: Int, + onWindow: () -> Unit, + onPaged: () -> Unit, + onNeed: (Int) -> Unit, + deliver: (Event) -> Unit, +) { + coroutineContext.ensureActive() + + when (val outcome = reconcileWindow(relay, filter, timeoutMs)) { + is ReconcileOutcome.Ids -> { + onWindow() + onNeed(outcome.needIds.size) + downloadIds(relay, outcome.needIds, fetchBatch, maxConcurrentReqs, timeoutMs, deliver) + } + + 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 can't help — + // page it. Paging is bounded by created_at cursors, not the + // negentropy cap, so it always terminates. + onWindow() + onPaged() + fetchAllPages(relay, listOf(filter), timeoutMs, onEvent = deliver) + } else { + val mid = lo + (hi - lo) / 2 + syncWindow(relay, filter.copy(since = lo, until = mid), timeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onPaged, onNeed, deliver) + syncWindow(relay, filter.copy(since = mid + 1, until = hi), timeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onPaged, onNeed, deliver) + } + } + + is ReconcileOutcome.Failed -> { + // Relay doesn't speak NIP-77, disconnected, or timed out mid-reconcile. + // Fall back to paging for this window rather than giving up. + onWindow() + onPaged() + fetchAllPages(relay, listOf(filter), timeoutMs, onEvent = deliver) + } + } +} + +private sealed interface ReconcileOutcome { + /** Reconciliation completed; [needIds] are the ids the relay has that we lack. */ + class Ids( + val needIds: List, + ) : ReconcileOutcome + + /** Relay rejected the set as too large (strfry `max_sync_events`). */ + object Overflow : ReconcileOutcome + + /** Reconciliation could not complete (no NIP-77 support, disconnect, timeout). */ + object Failed : ReconcileOutcome +} + +/** + * Drives one NIP-77 reconciliation of [filter] against an EMPTY local set, so the + * resulting `needIds` are every id the relay holds for that filter. Registers a + * [NegentropyManager], sends `NEG-OPEN`, and walks the rounds until completion, + * error, or [timeoutMs]. Always closes the session and removes the listener. + */ +private suspend fun INostrClient.reconcileWindow( + relay: NormalizedRelayUrl, + filter: Filter, + timeoutMs: Long, +): ReconcileOutcome { + val relayClient = getOrCreateRelay(relay) + val subId = newSubId() + val signals = Channel(Channel.UNLIMITED) + val needIds = ArrayList() + + val listener = + object : INegentropyListener { + override fun onHaveIds( + relay: NormalizedRelayUrl, + subId: String, + haveIds: List, + ) { + // Empty local set: there is nothing the relay can lack. Ignore. + } + + override fun onNeedIds( + relay: NormalizedRelayUrl, + subId: String, + needIds: List, + ) { + signals.trySend(NegSignal.Need(needIds)) + } + + override fun onComplete( + relay: NormalizedRelayUrl, + subId: String, + ) { + signals.trySend(NegSignal.Complete) + } + + override fun onError( + relay: NormalizedRelayUrl, + subId: String, + reason: String, + ) { + signals.trySend(NegSignal.Error(reason)) + } + } + + val manager = NegentropyManager(listener) + addConnectionListener(manager) + 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. `sendOrConnectAndSync` on a cold relay only + // kicks off the connect and silently drops the command, so we must connect + // and wait until the relay is ready before opening the session. + relayClient.connect() + val connected = + withTimeoutOrNull(timeoutMs) { + connectedRelaysFlow().first { relay in it } + } + if (connected == null) return ReconcileOutcome.Failed + + manager.startSync(relayClient, subId, filter, localEvents = emptyList()) + + val outcome = + withTimeoutOrNull(timeoutMs) { + while (true) { + when (val signal = signals.receive()) { + is NegSignal.Need -> needIds.addAll(signal.ids) + is NegSignal.Complete -> return@withTimeoutOrNull ReconcileOutcome.Ids(needIds) + is NegSignal.Error -> + return@withTimeoutOrNull if (isOverflow(signal.reason)) { + ReconcileOutcome.Overflow + } else { + ReconcileOutcome.Failed + } + } + } + @Suppress("UNREACHABLE_CODE") + ReconcileOutcome.Failed + } + + return outcome ?: ReconcileOutcome.Failed + } finally { + manager.closeSync(relayClient, subId) + removeConnectionListener(manager) + signals.close() + } +} + +private sealed interface NegSignal { + class Need( + val ids: List, + ) : NegSignal + + object Complete : NegSignal + + class Error( + val reason: String, + ) : NegSignal +} + +/** + * 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) + +/** + * Downloads [ids] from [relay] through at most [maxConcurrentReqs] concurrent + * `REQ`s of [fetchBatch] ids each. A fixed worker pool drains a batch queue, so at + * most [maxConcurrentReqs] subscriptions are ever open at once, each refilled as + * its `EOSE` arrives. + */ +private suspend fun INostrClient.downloadIds( + relay: NormalizedRelayUrl, + ids: List, + fetchBatch: Int, + maxConcurrentReqs: Int, + timeoutMs: Long, + deliver: (Event) -> Unit, +) { + if (ids.isEmpty()) return + + val batches = Channel>(Channel.UNLIMITED) + val chunks = ids.chunked(fetchBatch) + chunks.forEach { batches.trySend(it) } + batches.close() + + val workers = min(maxConcurrentReqs.coerceAtLeast(1), chunks.size) + + coroutineScope { + repeat(workers) { + launch { + for (batch in batches) { + coroutineContext.ensureActive() + fetchByIds(relay, batch, timeoutMs, deliver) + } + } + } + } +} + +/** One `REQ` for [batch] ids; delivers each event, returns on `EOSE`/close/timeout. */ +private suspend fun INostrClient.fetchByIds( + relay: NormalizedRelayUrl, + batch: List, + timeoutMs: Long, + deliver: (Event) -> Unit, +) { + val subId = newSubId() + val done = Channel(Channel.CONFLATED) + + val listener = + object : SubscriptionListener { + override fun onEvent( + event: Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + deliver(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) + withTimeoutOrNull(timeoutMs) { + done.receive() + } + } finally { + unsubscribe(subId) + done.close() + } +} + +/** Seconds: a window this small that still overflows is paged instead of split. */ +private const val MIN_WINDOW_SECONDS = 1L 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..20e573244c --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt @@ -0,0 +1,166 @@ +/* + * 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.negentropySync +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncAsFlow +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.last +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.Test +import kotlin.test.assertEquals +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") + assertFalse(result.fellBackToPaging) + } + + @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 flowVariantEmitsAllEvents() = + runBlocking { + defaultRelay.preload(SyntheticEvents.batch(12, kind = 1)) + + val last = + withTimeout(20_000) { + client + .negentropySyncAsFlow( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + ).last() + } + + assertEquals(12, last.size) + assertEquals(12, last.map { it.id }.toSet().size) + } + + /** + * A relay that caps negentropy below the matched-set size (strfry's + * `max_sync_events`) and whose events all share one `created_at`, so no + * `created_at` window can separate them. The sync must still download every + * event — by splitting down to a minimal window and then falling back to + * [com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages]. + */ + @Test + fun windowSplitAndPagingFallbackOnOverflow() = + 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:7780/") + // 10 events, all at the same created_at: created_at windowing can + // never split them, so the minimal window still overflows the cap. + 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.negentropySync( + relay = url, + filter = Filter(kinds = listOf(1)), + ) { got.add(it) } + } + + assertEquals(10, got.map { it.id }.toSet().size, "all events downloaded despite the cap") + assertEquals(10, result.downloaded) + assertTrue(result.fellBackToPaging, "the over-cap minimal window should page") + assertTrue(result.windows > 1, "the set must be split into multiple created_at windows") + } finally { + client.disconnect() + scope.cancel() + hub.close() + } + } +} From 21c714177c3a79a0ce48dc8174bd94fff45cadf1 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 22:03:29 +0000 Subject: [PATCH 2/9] refactor(quartz): stream events from the negentropy flow variant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the cumulative `negentropySyncAsFlow(): Flow>` with `negentropySyncEvents(): Flow`, which emits each event individually as it arrives. Rebuilding an ever-growing list per event was O(events²) in both CPU and memory and pointless for a bulk download; the stream stays O(1) in memory and hands the caller raw events to collect however they like. Events are buffered with Channel.UNLIMITED because negentropySync delivers through a non-suspending callback — a bounded buffer would drop events when the collector lags. Callers can apply their own buffer/conflate/ collectLatest downstream. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- ... => NostrClientNegentropySyncEventsExt.kt} | 35 +++++++++++-------- .../relay/NostrClientNegentropySyncTest.kt | 16 ++++----- 2 files changed, 28 insertions(+), 23 deletions(-) rename quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/{NostrClientNegentropySyncAsFlowExt.kt => NostrClientNegentropySyncEventsExt.kt} (71%) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncAsFlowExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncEventsExt.kt similarity index 71% rename from quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncAsFlowExt.kt rename to quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncEventsExt.kt index 298435ec1c..2360d8f7d7 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncAsFlowExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncEventsExt.kt @@ -25,30 +25,36 @@ 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 /** - * Flow form of [negentropySync]: runs the sync while collected and emits the - * accumulated events as a growing list on each new arrival, then completes when - * the sync finishes (mirroring [com.vitorpamplona.quartz.nip01Core.relay.client.reqs.fetchAsFlow]). - * Cancelling the collector cancels the sync and tears down its subscriptions via - * [awaitClose]. + * 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.negentropySyncAsFlow( +fun INostrClient.negentropySyncEvents( relay: NormalizedRelayUrl, filter: Filter, maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, timeoutMs: Long = 30_000L, -): Flow> = +): Flow = callbackFlow { - var current = listOf() - negentropySync( relay = relay, filter = filter, @@ -57,24 +63,23 @@ fun INostrClient.negentropySyncAsFlow( fetchBatch = fetchBatch, timeoutMs = timeoutMs, ) { event -> - current = current + event - trySend(current) + trySend(event) } close() awaitClose { } - } + }.buffer(Channel.UNLIMITED) -fun INostrClient.negentropySyncAsFlow( +fun INostrClient.negentropySyncEvents( relay: String, filter: Filter, maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, timeoutMs: Long = 30_000L, -): Flow> = - negentropySyncAsFlow( +): Flow = + negentropySyncEvents( relay = RelayUrlNormalizer.normalize(relay), filter = filter, maxEvents = maxEvents, 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 index 20e573244c..a87d2b6593 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt @@ -27,7 +27,7 @@ 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.negentropySync -import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncAsFlow +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncEvents import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings @@ -35,7 +35,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel -import kotlinx.coroutines.flow.last +import kotlinx.coroutines.flow.toList import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout import kotlin.test.Test @@ -107,21 +107,21 @@ class NostrClientNegentropySyncTest : RelayClientTest() { } @Test - fun flowVariantEmitsAllEvents() = + fun flowVariantStreamsEachEvent() = runBlocking { defaultRelay.preload(SyntheticEvents.batch(12, kind = 1)) - val last = + val events = withTimeout(20_000) { client - .negentropySyncAsFlow( + .negentropySyncEvents( relay = defaultRelayUrl, filter = Filter(kinds = listOf(1)), - ).last() + ).toList() } - assertEquals(12, last.size) - assertEquals(12, last.map { it.id }.toSet().size) + assertEquals(12, events.size) + assertEquals(12, events.map { it.id }.toSet().size) } /** From 3fdf418177b794449785487e892f770e8f919992 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 22:21:02 +0000 Subject: [PATCH 3/9] feat(quartz): make negentropy paging fallback the caller's choice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per review: negentropySync should not silently switch transports. Plain created_at paging is heavier and non-delta, and a caller who reached for negentropy may prefer to know it failed (try another relay, narrow the filter, abort) rather than get a surprise paged download. So: - negentropySync is now negentropy-only. created_at windowing on the relay's max_sync_events cap stays automatic (it's still negentropy), but a window that genuinely can't be reconciled — minimal window still over the cap, or a relay with no NIP-77 support / disconnect / timeout — now throws the typed NegentropySyncException (reason OVER_MAX_SYNC_EVENTS or UNAVAILABLE, carrying the failing window) instead of paging. Dropped NegentropySyncResult.fellBackToPaging. - Added negentropySyncOrFetch (+ negentropySyncOrFetchEvents Flow form) as the ergonomic "try negentropy, else page" combinator: runs negentropySync and, on NegentropySyncException, falls back to fetchAllPages over the same filter, deduping by id across both phases and honoring maxEvents. Returns NegentropyOrFetchResult so callers can see which path ran and why. Callers now choose explicitly: negentropySync to handle failure themselves, negentropySyncOrFetch for automatic paging fallback. Tests: over-cap relay with spread timestamps succeeds via windowing alone; over-cap minimal window throws (and the caller can page); orFetch pages on failure and uses negentropy when it works. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../accessories/NegentropySyncException.kt | 69 +++++++ .../NostrClientNegentropySyncEventsExt.kt | 52 +++++ .../NostrClientNegentropySyncExt.kt | 186 ++++++++++++++---- .../relay/NostrClientNegentropySyncTest.kt | 132 +++++++++++-- 4 files changed, 384 insertions(+), 55 deletions(-) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropySyncException.kt 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/NostrClientNegentropySyncEventsExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncEventsExt.kt index 2360d8f7d7..37945e1d9c 100644 --- 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 @@ -87,3 +87,55 @@ fun INostrClient.negentropySyncEvents( fetchBatch = fetchBatch, timeoutMs = timeoutMs, ) + +/** + * 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, + timeoutMs: Long = 30_000L, +): Flow = + callbackFlow { + negentropySyncOrFetch( + relay = relay, + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + timeoutMs = timeoutMs, + ) { event -> + trySend(event) + } + + close() + + awaitClose { } + }.buffer(Channel.UNLIMITED) + +fun INostrClient.negentropySyncOrFetchEvents( + relay: String, + filter: Filter, + maxEvents: Int = 0, + maxConcurrentReqs: Int = 8, + fetchBatch: Int = 500, + timeoutMs: Long = 30_000L, +): Flow = + negentropySyncOrFetchEvents( + relay = RelayUrlNormalizer.normalize(relay), + filter = filter, + maxEvents = maxEvents, + maxConcurrentReqs = maxConcurrentReqs, + fetchBatch = fetchBatch, + timeoutMs = timeoutMs, + ) 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 index 6a9a092259..bbe5b0bac3 100644 --- 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 @@ -41,7 +41,7 @@ import kotlin.coroutines.coroutineContext import kotlin.math.min /** - * Outcome of a [negentropySync] run. + * 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 @@ -51,17 +51,12 @@ import kotlin.math.min * @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). - * @property fellBackToPaging `true` if at least one window could not be - * reconciled by negentropy (relay rejected it, e.g. strfry's - * `max_sync_events`, or did not support NIP-77) and was downloaded via - * [fetchAllPages] instead. */ class NegentropySyncResult( val needCount: Int, val haveCount: Int, val downloaded: Int, val windows: Int, - val fellBackToPaging: Boolean, ) /** @@ -77,19 +72,25 @@ class NegentropySyncResult( * so a huge set never opens thousands of subs at 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 is halved and retried; a minimal window that - * still cannot reconcile (or a relay that does not speak NIP-77) falls back to - * [fetchAllPages] for that window and sets [NegentropySyncResult.fellBackToPaging]. + * 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. The accessory keeps memory * bounded by reconciling and downloading one window at a time and by capping the * delivered set at [maxEvents]. * - * Coroutine-cancellable: on completion, cancel, or reaching [maxEvents] all `REQ` - * subscriptions are unsubscribed and the negentropy session is closed and its - * listener removed, so nothing leaks. + * 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). @@ -113,15 +114,14 @@ suspend fun INostrClient.negentropySync( ): NegentropySyncResult { var need = 0 var windows = 0 - var fellBack = false var downloaded = 0 val seen = HashSet() coroutineScope { - // Single funnel for every delivered event (from REQ batches AND from the - // paging fallback), so dedup + the maxEvents cap + onEvent run on one - // coroutine even though the relay reader threads produce concurrently. + // Single funnel for every delivered event, so dedup + the maxEvents cap + + // onEvent run on one coroutine even though the relay reader threads produce + // concurrently. val events = Channel(Channel.UNLIMITED) val producer = @@ -134,7 +134,6 @@ suspend fun INostrClient.negentropySync( fetchBatch = fetchBatch, maxConcurrentReqs = maxConcurrentReqs, onWindow = { windows++ }, - onPaged = { fellBack = true }, // Only accumulate here; progress is reported from the single // consumer loop below so the user callback is never invoked // from two coroutines at once. @@ -165,7 +164,6 @@ suspend fun INostrClient.negentropySync( haveCount = 0, downloaded = downloaded, windows = windows, - fellBackToPaging = fellBack, ) } @@ -190,10 +188,113 @@ suspend fun INostrClient.negentropySync( 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, + timeoutMs: Long = 30_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, + timeoutMs = timeoutMs, + 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. + val pageFilter = if (maxEvents > 0) filter.copy(limit = maxEvents) else filter + fetchAllPages(relay, listOf(pageFilter), timeoutMs) { 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, + timeoutMs: Long = 30_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, + timeoutMs = timeoutMs, + 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, @@ -202,7 +303,6 @@ private suspend fun INostrClient.syncWindow( fetchBatch: Int, maxConcurrentReqs: Int, onWindow: () -> Unit, - onPaged: () -> Unit, onNeed: (Int) -> Unit, deliver: (Event) -> Unit, ) { @@ -219,26 +319,28 @@ private suspend fun INostrClient.syncWindow( val lo = filter.since ?: 0L val hi = filter.until ?: TimeUtils.now() if (hi - lo <= MIN_WINDOW_SECONDS) { - // A minimal window that still overflows: negentropy can't help — - // page it. Paging is bounded by created_at cursors, not the - // negentropy cap, so it always terminates. - onWindow() - onPaged() - fetchAllPages(relay, listOf(filter), timeoutMs, onEvent = deliver) + // 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), timeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onPaged, onNeed, deliver) - syncWindow(relay, filter.copy(since = mid + 1, until = hi), timeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onPaged, onNeed, deliver) + syncWindow(relay, filter.copy(since = lo, until = mid), timeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onNeed, deliver) + syncWindow(relay, filter.copy(since = mid + 1, until = hi), timeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onNeed, deliver) } } - is ReconcileOutcome.Failed -> { - // Relay doesn't speak NIP-77, disconnected, or timed out mid-reconcile. - // Fall back to paging for this window rather than giving up. - onWindow() - onPaged() - fetchAllPages(relay, listOf(filter), timeoutMs, onEvent = deliver) - } + is ReconcileOutcome.Failed -> + throw NegentropySyncException( + relay = relay, + window = filter, + reason = NegentropySyncException.Reason.UNAVAILABLE, + detail = outcome.detail, + ) } } @@ -251,8 +353,10 @@ private sealed interface ReconcileOutcome { /** Relay rejected the set as too large (strfry `max_sync_events`). */ object Overflow : ReconcileOutcome - /** Reconciliation could not complete (no NIP-77 support, disconnect, timeout). */ - object Failed : ReconcileOutcome + /** Reconciliation could not complete; [detail] says why. */ + class Failed( + val detail: String, + ) : ReconcileOutcome } /** @@ -318,7 +422,7 @@ private suspend fun INostrClient.reconcileWindow( withTimeoutOrNull(timeoutMs) { connectedRelaysFlow().first { relay in it } } - if (connected == null) return ReconcileOutcome.Failed + if (connected == null) return ReconcileOutcome.Failed("could not connect within ${timeoutMs}ms") manager.startSync(relayClient, subId, filter, localEvents = emptyList()) @@ -332,15 +436,15 @@ private suspend fun INostrClient.reconcileWindow( return@withTimeoutOrNull if (isOverflow(signal.reason)) { ReconcileOutcome.Overflow } else { - ReconcileOutcome.Failed + ReconcileOutcome.Failed(signal.reason) } } } @Suppress("UNREACHABLE_CODE") - ReconcileOutcome.Failed + ReconcileOutcome.Failed("unreachable") } - return outcome ?: ReconcileOutcome.Failed + return outcome ?: ReconcileOutcome.Failed("reconcile timed out after ${timeoutMs}ms") } finally { manager.closeSync(relayClient, subId) removeConnectionListener(manager) 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 index a87d2b6593..2c0fec7d0d 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt @@ -26,8 +26,11 @@ 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 @@ -40,6 +43,7 @@ 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 @@ -64,7 +68,6 @@ class NostrClientNegentropySyncTest : RelayClientTest() { assertEquals(0, result.haveCount) assertEquals(20, result.downloaded) assertEquals(1, result.windows, "small set reconciles in a single window") - assertFalse(result.fellBackToPaging) } @Test @@ -126,23 +129,21 @@ class NostrClientNegentropySyncTest : RelayClientTest() { /** * A relay that caps negentropy below the matched-set size (strfry's - * `max_sync_events`) and whose events all share one `created_at`, so no - * `created_at` window can separate them. The sync must still download every - * event — by splitting down to a minimal window and then falling back to - * [com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages]. + * `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 windowSplitAndPagingFallbackOnOverflow() = + 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:7780/") - // 10 events, all at the same created_at: created_at windowing can - // never split them, so the minimal window still overflows the cap. - val events = (1..10).map { SyntheticEvents.fakeEvent(idSeed = it, kind = 1, createdAt = 1000L) } - hub.getOrCreate(url).preload(events) + 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 = @@ -153,9 +154,8 @@ class NostrClientNegentropySyncTest : RelayClientTest() { ) { got.add(it) } } - assertEquals(10, got.map { it.id }.toSet().size, "all events downloaded despite the cap") - assertEquals(10, result.downloaded) - assertTrue(result.fellBackToPaging, "the over-cap minimal window should page") + 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() @@ -163,4 +163,108 @@ class NostrClientNegentropySyncTest : RelayClientTest() { 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) + } } From e5f43904f2e7d656b699de1a9a569ac299c9dc48 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 23:12:21 +0000 Subject: [PATCH 4/9] perf(quartz): stream negentropy sync in bounded memory for huge windows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rework the negentropy download path from "reconcile fully → then download" into a single back-pressured streaming pipeline so peak memory is independent of the window size — built for multi-million-event syncs. - reconcileStreaming drives the NIP-77 rounds directly (instead of via NegentropyManager) and hands each round's ids to a bounded id-queue *before* acking the next round, so the relay's id stream is paced to the downloader. - Ids flow id-queue → bounded download worker pool → bounded delivery channel; a slow consumer back-pressures the whole chain. The full id list is never materialised. - Drop the global event-dedup set (was O(set) ~ hundreds of MB at 4M): NIP-77 yields a distinct id set, so each event is requested once. Keep only a tiny per-batch dedup (bounded by fetchBatch) to absorb a relay replaying a REQ. - Pin the relay with a never-matching keep-alive subscription for the sync's duration: a NEG-OPEN isn't a REQ, so during a reconcile round the pool would otherwise see the relay as unwanted and disconnect it mid-sync. - Document that timeoutMs must accommodate the relay's first-frame snapshot latency on huge sets (a real strfry took ~73s for an unbounded kind:0 set). Validated against wss://wot.grapevine.network: streamed 30k kind:0 events with heap bounded at ~30-90 MB (not growing with the set) and zero duplicates. New unit test forces many small reconcile frames to exercise the multi-round / back-pressure path; existing windowing/cap/fallback tests still green. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../NostrClientNegentropySyncExt.kt | 361 ++++++++++-------- .../relay/NostrClientNegentropySyncTest.kt | 39 ++ 2 files changed, 248 insertions(+), 152 deletions(-) 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 index bbe5b0bac3..2d1b85b0cf 100644 --- 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 @@ -23,18 +23,23 @@ 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.INegentropyListener -import com.vitorpamplona.quartz.nip77Negentropy.NegentropyManager +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.coroutines.coroutineContext @@ -64,12 +69,17 @@ class NegentropySyncResult( * 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 via a - * [NegentropyManager], accumulating the ids the relay has (`needIds`) across - * rounds until completion. + * 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, - * so a huge set never opens thousands of subs at once. + * 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 @@ -82,9 +92,8 @@ class NegentropySyncResult( * 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. The accessory keeps memory - * bounded by reconciling and downloading one window at a time and by capping the - * delivered set at [maxEvents]. + * 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 @@ -99,6 +108,11 @@ class NegentropySyncResult( * it at or below the relay's per-connection subscription cap. * @param fetchBatch ids per download `REQ`. * @param timeoutMs max wait for a single reconcile round or download page's `EOSE`. + * The relay builds its whole negentropy snapshot before the FIRST round responds, + * which is O(matched set) — for a multi-million-event filter that first response can + * take a minute or more (a real strfry took ~73s for an unbounded kind:0 set), so + * raise this for very large syncs. It is only a ceiling — download REQs return on + * their `EOSE`, so a generous value costs nothing in the common case. * @param onProgress optional `(needSoFar, downloaded)` ticks as work proceeds. * @param onEvent called once per distinct event, on the relay reader thread. */ @@ -114,49 +128,57 @@ suspend fun INostrClient.negentropySync( ): NegentropySyncResult { var need = 0 var windows = 0 - var downloaded = 0 - val seen = HashSet() - coroutineScope { - // Single funnel for every delivered event, so dedup + the maxEvents cap + - // onEvent run on one coroutine even though the relay reader threads produce - // concurrently. - val events = Channel(Channel.UNLIMITED) + // 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, - timeoutMs = timeoutMs, - 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.trySend(it) }, - ) - } finally { - events.close() + val producer = + launch { + try { + syncWindow( + relay = relay, + filter = filter, + timeoutMs = timeoutMs, + 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) { - if (seen.add(event.id)) { + 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() + // 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( @@ -304,16 +326,12 @@ private suspend fun INostrClient.syncWindow( maxConcurrentReqs: Int, onWindow: () -> Unit, onNeed: (Int) -> Unit, - deliver: (Event) -> Unit, + deliver: suspend (Event) -> Unit, ) { coroutineContext.ensureActive() - when (val outcome = reconcileWindow(relay, filter, timeoutMs)) { - is ReconcileOutcome.Ids -> { - onWindow() - onNeed(outcome.needIds.size) - downloadIds(relay, outcome.needIds, fetchBatch, maxConcurrentReqs, timeoutMs, deliver) - } + when (val outcome = downloadWindow(relay, filter, timeoutMs, fetchBatch, maxConcurrentReqs, onNeed, deliver)) { + is ReconcileOutcome.Complete -> onWindow() is ReconcileOutcome.Overflow -> { val lo = filter.since ?: 0L @@ -345,10 +363,8 @@ private suspend fun INostrClient.syncWindow( } private sealed interface ReconcileOutcome { - /** Reconciliation completed; [needIds] are the ids the relay has that we lack. */ - class Ids( - val needIds: List, - ) : 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 @@ -360,108 +376,115 @@ private sealed interface ReconcileOutcome { } /** - * Drives one NIP-77 reconciliation of [filter] against an EMPTY local set, so the - * resulting `needIds` are every id the relay holds for that filter. Registers a - * [NegentropyManager], sends `NEG-OPEN`, and walks the rounds until completion, - * error, or [timeoutMs]. Always closes the session and removes the listener. + * 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.reconcileWindow( +private suspend fun INostrClient.reconcileStreaming( relay: NormalizedRelayUrl, filter: Filter, timeoutMs: Long, + fetchBatch: Int, + onNeed: (Int) -> Unit, + sendBatch: suspend (List) -> Unit, ): ReconcileOutcome { + val targetUrl = relay val relayClient = getOrCreateRelay(relay) val subId = newSubId() - val signals = Channel(Channel.UNLIMITED) - val needIds = ArrayList() + 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) val listener = - object : INegentropyListener { - override fun onHaveIds( - relay: NormalizedRelayUrl, - subId: String, - haveIds: List, + object : RelayConnectionListener { + override fun onIncomingMessage( + relay: IRelayClient, + msgStr: String, + msg: Message, ) { - // Empty local set: there is nothing the relay can lack. Ignore. + 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 onNeedIds( - relay: NormalizedRelayUrl, - subId: String, - needIds: List, - ) { - signals.trySend(NegSignal.Need(needIds)) - } - - override fun onComplete( - relay: NormalizedRelayUrl, - subId: String, - ) { - signals.trySend(NegSignal.Complete) - } - - override fun onError( - relay: NormalizedRelayUrl, - subId: String, - reason: String, - ) { - signals.trySend(NegSignal.Error(reason)) + override fun onDisconnected(relay: IRelayClient) { + if (relay.url == targetUrl) incoming.trySend(NegFrame.Err("closed: relay disconnected")) } } - val manager = NegentropyManager(listener) - addConnectionListener(manager) + 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. `sendOrConnectAndSync` on a cold relay only - // kicks off the connect and silently drops the command, so we must connect - // and wait until the relay is ready before opening the session. + // NEG-OPEN is never resent, so we must connect and wait until the relay is + // ready before sending it. relayClient.connect() val connected = withTimeoutOrNull(timeoutMs) { - connectedRelaysFlow().first { relay in it } + connectedRelaysFlow().first { targetUrl in it } } if (connected == null) return ReconcileOutcome.Failed("could not connect within ${timeoutMs}ms") - manager.startSync(relayClient, subId, filter, localEvents = emptyList()) + relayClient.sendIfConnected(session.open()) - val outcome = - withTimeoutOrNull(timeoutMs) { - while (true) { - when (val signal = signals.receive()) { - is NegSignal.Need -> needIds.addAll(signal.ids) - is NegSignal.Complete -> return@withTimeoutOrNull ReconcileOutcome.Ids(needIds) - is NegSignal.Error -> - return@withTimeoutOrNull if (isOverflow(signal.reason)) { - ReconcileOutcome.Overflow - } else { - ReconcileOutcome.Failed(signal.reason) - } + while (true) { + // Time only the wait for the relay's next frame — never our own + // back-pressured streaming of the previous frame's ids. + val frame = + withTimeoutOrNull(timeoutMs) { incoming.receive() } + ?: return ReconcileOutcome.Failed("reconcile round timed out after ${timeoutMs}ms") + + 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 } } - @Suppress("UNREACHABLE_CODE") - ReconcileOutcome.Failed("unreachable") } - - return outcome ?: ReconcileOutcome.Failed("reconcile timed out after ${timeoutMs}ms") + } } finally { - manager.closeSync(relayClient, subId) - removeConnectionListener(manager) - signals.close() + relayClient.sendIfConnected(session.close()) + removeConnectionListener(listener) + incoming.close() } } -private sealed interface NegSignal { - class Need( - val ids: List, - ) : NegSignal +private sealed interface NegFrame { + class Msg( + val payload: String, + ) : NegFrame - object Complete : NegSignal - - class Error( + class Err( val reason: String, - ) : NegSignal + ) : NegFrame } /** @@ -476,49 +499,71 @@ private fun isOverflow(reason: String): Boolean = reason.startsWith("blocked", ignoreCase = true) /** - * Downloads [ids] from [relay] through at most [maxConcurrentReqs] concurrent - * `REQ`s of [fetchBatch] ids each. A fixed worker pool drains a batch queue, so at - * most [maxConcurrentReqs] subscriptions are ever open at once, each refilled as - * its `EOSE` arrives. + * 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.downloadIds( +private suspend fun INostrClient.downloadWindow( relay: NormalizedRelayUrl, - ids: List, + filter: Filter, + timeoutMs: Long, fetchBatch: Int, maxConcurrentReqs: Int, - timeoutMs: Long, - deliver: (Event) -> Unit, -) { - if (ids.isEmpty()) return - - val batches = Channel>(Channel.UNLIMITED) - val chunks = ids.chunked(fetchBatch) - chunks.forEach { batches.trySend(it) } - batches.close() - - val workers = min(maxConcurrentReqs.coerceAtLeast(1), chunks.size) - + onNeed: (Int) -> Unit, + deliver: suspend (Event) -> Unit, +): ReconcileOutcome = coroutineScope { - repeat(workers) { - launch { - for (batch in batches) { - coroutineContext.ensureActive() - fetchByIds(relay, batch, timeoutMs, deliver) + 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, timeoutMs)) { + deliver(event) + } + } } } - } - } -} -/** One `REQ` for [batch] ids; delivers each event, returns on `EOSE`/close/timeout. */ + val outcome = + reconcileStreaming(relay, filter, timeoutMs, 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, timeoutMs: Long, - deliver: (Event) -> Unit, -) { +): List { val subId = newSubId() val done = Channel(Channel.CONFLATED) + val collected = ArrayList(batch.size) + val seen = HashSet(batch.size) val listener = object : SubscriptionListener { @@ -528,7 +573,7 @@ private suspend fun INostrClient.fetchByIds( relay: NormalizedRelayUrl, forFilters: List?, ) { - deliver(event) + if (seen.add(event.id)) collected.add(event) } override fun onEose( @@ -564,7 +609,19 @@ private suspend fun INostrClient.fetchByIds( unsubscribe(subId) done.close() } + return collected } -/** Seconds: a window this small that still overflows is paged instead of split. */ +/** 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) 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 index 2c0fec7d0d..04734e6162 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt @@ -127,6 +127,45 @@ class NostrClientNegentropySyncTest : RelayClientTest() { 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` From ed5c25e2b1d2b1768ead2d3a56810a2a23d49162 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 1 Jul 2026 00:42:33 +0000 Subject: [PATCH 5/9] fix(quartz): fresh subId per page in fetchAllPages (was truncating large results) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fetchAllPages reused a single subscription id across all pages (unsubscribe + immediately re-subscribe the same id). On a real relay that caps REQ results, the rapid same-id CLOSE→REQ races on the wire: in-flight events from the previous page's REQ bleed into the next page's listener. Those stale events carry a created_at above the freshly-lowered `until`, so `match()` rejects them, the page ends with pageCount == 0, and the whole loop breaks — silently truncating the download. Observed against wss://wot.grapevine.network: a full kind:0 download (~3.55M events, per a concurrent negentropy sync) stopped at 89,500. A controlled diagnosis paging the same data with a fresh subId per page vs a shared subId reproduced it exactly: shared stalled at ~95k with in-page duplicates and events above `until`; fresh advanced cleanly with no duplicates. After the fix, the real-relay fetchAllPages sails past the old stall (100k+ and counting). Fix: allocate the subId inside the paging loop so each page is an independent subscription. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../client/accessories/NostrClientFetchAllPagesExt.kt | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt index 8bacb7b383..22cbcdd495 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,11 +64,17 @@ suspend fun INostrClient.fetchAllPages( // Track how many matching events each filter has received so far. val matchCountPerFilter = IntArray(filters.size) - val subId = newSubId() - while (true) { coroutineContext.ensureActive() + // A fresh subscription id per page. Reusing one id across pages + // (unsubscribe + immediately re-subscribe the same id) races on the wire: + // in-flight events from the previous page's REQ bleed into the next page's + // listener. Those stale events carry a `created_at` above the new `until`, + // so `match()` rejects them, the page ends with `pageCount == 0`, and the + // whole download terminates early — silently truncating large results. + val subId = newSubId() + val pagedFilters = if (until == null) { filters From f357b445c3906b22f7cb054fa6b6fa34c8bfd976 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 1 Jul 2026 02:30:19 +0000 Subject: [PATCH 6/9] fix(quartz): serialize PoolRequests state machine to kill shared-sub double-REQ race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A subscription id is driven from two threads at once: the app thread (the subscribe/unsubscribe path) and every relay's socket-reader thread (an EOSE that triggers an auto-resend). Both read the subscription state and both can decide "the filters changed, send a REQ", but the decision (read state) and the send (mark state SENT in onSent) were not atomic. So the reader could observe 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 duplicate EOSEs and events, or — if a CLOSE interleaves — an empty result that silently truncates a paged download. This is what intermittently broke fetchAllPages on large sets (fixed at the call site in ed5c25e2 by using a fresh sub id per page); this commit fixes the underlying race in the relay-client layer, which could equally corrupt any subscription that spans multiple relays (several reader threads mutate the same RequestSubscriptionState maps concurrently). The fix: - Add a tiny non-reentrant spin lock (withStateLock, same AtomicBoolean primitive BasicRelayClient uses) guarding every access to the subscription state machine. Listener callbacks and socket sends stay OUTSIDE the lock — they re-enter this class via onSent, so holding it across them would deadlock. - Fold the send decision into decideCommandLocked, which runs under the lock and pre-marks the state SENT (+ filters) the moment it decides to send a REQ. A concurrent decider then sees SENT/updated filters and declines, so exactly one REQ is ever produced. Verified with a deterministic A/B repro that pins the exact interleaving open: pre-fix 300/300 episodes produced a duplicate REQ; post-fix 0/300 (max one REQ per episode). Kept as PoolRequestsConcurrencyTest. Full relay test suite passes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../relay/client/pool/PoolRequests.kt | 226 ++++++++++++------ .../relay/PoolRequestsConcurrencyTest.kt | 146 +++++++++++ 2 files changed, 302 insertions(+), 70 deletions(-) create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt 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/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", + ) + } + } +} From 23722969fd41b8a3c1438278567364e33fffef91 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 1 Jul 2026 13:00:05 +0000 Subject: [PATCH 7/9] refactor(quartz): reuse one subscription id across fetchAllPages pages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Revert the fresh-subId-per-page workaround (ed5c25e2) now that the underlying double-REQ race is fixed at the root in PoolRequests. Relays cap the number of concurrent subscriptions per connection, so a single reused id — opened per page with the page's `until`, closed before the next page — keeps the whole download to one subscription slot instead of churning through a distinct id each page. Safe because the pool now serializes the "send a REQ" decision: after a page's EOSE, the auto-resend and the loop's unsubscribe+resubscribe can no longer both fire a REQ for the same id (guarded by PoolRequestsConcurrencyTest). Unit suites (negentropy paging scenarios, subscriptions) pass; full-scale real-relay verification to follow before push. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../NostrClientFetchAllPagesExt.kt | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt index 22cbcdd495..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,17 +64,24 @@ 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) { coroutineContext.ensureActive() - // A fresh subscription id per page. Reusing one id across pages - // (unsubscribe + immediately re-subscribe the same id) races on the wire: - // in-flight events from the previous page's REQ bleed into the next page's - // listener. Those stale events carry a `created_at` above the new `until`, - // so `match()` rejects them, the page ends with `pageCount == 0`, and the - // whole download terminates early — silently truncating large results. - val subId = newSubId() - val pagedFilters = if (until == null) { filters From 55e945a5ed1322fa74fc5370ebd82ea55ff513a2 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 1 Jul 2026 14:02:57 +0000 Subject: [PATCH 8/9] feat(quartz): idle watchdog for negentropySync instead of a fixed per-round timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fixed `timeoutMs` (default 30s) applied to every reconcile round, but the FIRST round on a large relay is a legitimate long silence while the relay builds its whole negentropy fingerprint — observed at ~63-68s for a 3.5M-event kind:0 set on a real relay. So the old default spuriously failed big syncs with UNAVAILABLE ("reconcile round timed out"), even though the relay was working fine and would have answered seconds later. Replace it with `idleTimeoutMs` (default 120s): the maximum time the relay may go COMPLETELY SILENT before giving up. It resets on every message the relay sends — each NIP-77 round and 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. Because the watchdog is fed by a connection-level listener that sees all of the relay's traffic, download activity extends the reconcile deadline and vice versa. `idleTimeoutMs = 0` disables it entirely (run until the socket drops); the initial connect and each download batch keep their own finite bounds so an unreachable relay or a single stuck batch still can't hang the pipeline. Liveness of a dead/half-open socket does not depend on this: the WebSocket keep-alive (ping/pong) detects it and the disconnect is already turned into a clean NEG-ERR abort. Verified against wss://wot.grapevine.network: the first round took 68.4s (the old 30s default would have thrown UNAVAILABLE) and the sync sailed through it under the 120s idle window. Negentropy unit suite passes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../NostrClientNegentropySyncEventsExt.kt | 16 +- .../NostrClientNegentropySyncExt.kt | 162 ++++++++++++++---- 2 files changed, 135 insertions(+), 43 deletions(-) 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 index 37945e1d9c..4c064003a6 100644 --- 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 @@ -52,7 +52,7 @@ fun INostrClient.negentropySyncEvents( maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, - timeoutMs: Long = 30_000L, + idleTimeoutMs: Long = 120_000L, ): Flow = callbackFlow { negentropySync( @@ -61,7 +61,7 @@ fun INostrClient.negentropySyncEvents( maxEvents = maxEvents, maxConcurrentReqs = maxConcurrentReqs, fetchBatch = fetchBatch, - timeoutMs = timeoutMs, + idleTimeoutMs = idleTimeoutMs, ) { event -> trySend(event) } @@ -77,7 +77,7 @@ fun INostrClient.negentropySyncEvents( maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, - timeoutMs: Long = 30_000L, + idleTimeoutMs: Long = 120_000L, ): Flow = negentropySyncEvents( relay = RelayUrlNormalizer.normalize(relay), @@ -85,7 +85,7 @@ fun INostrClient.negentropySyncEvents( maxEvents = maxEvents, maxConcurrentReqs = maxConcurrentReqs, fetchBatch = fetchBatch, - timeoutMs = timeoutMs, + idleTimeoutMs = idleTimeoutMs, ) /** @@ -104,7 +104,7 @@ fun INostrClient.negentropySyncOrFetchEvents( maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, - timeoutMs: Long = 30_000L, + idleTimeoutMs: Long = 120_000L, ): Flow = callbackFlow { negentropySyncOrFetch( @@ -113,7 +113,7 @@ fun INostrClient.negentropySyncOrFetchEvents( maxEvents = maxEvents, maxConcurrentReqs = maxConcurrentReqs, fetchBatch = fetchBatch, - timeoutMs = timeoutMs, + idleTimeoutMs = idleTimeoutMs, ) { event -> trySend(event) } @@ -129,7 +129,7 @@ fun INostrClient.negentropySyncOrFetchEvents( maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, - timeoutMs: Long = 30_000L, + idleTimeoutMs: Long = 120_000L, ): Flow = negentropySyncOrFetchEvents( relay = RelayUrlNormalizer.normalize(relay), @@ -137,5 +137,5 @@ fun INostrClient.negentropySyncOrFetchEvents( maxEvents = maxEvents, maxConcurrentReqs = maxConcurrentReqs, fetchBatch = fetchBatch, - timeoutMs = timeoutMs, + 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 index 2d1b85b0cf..689eeafdb6 100644 --- 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 @@ -42,8 +42,11 @@ import kotlinx.coroutines.flow.first import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeoutOrNull +import kotlin.concurrent.atomics.AtomicReference +import kotlin.concurrent.atomics.ExperimentalAtomicApi import kotlin.coroutines.coroutineContext import kotlin.math.min +import kotlin.time.TimeSource /** * Outcome of a successful [negentropySync] run. @@ -107,12 +110,18 @@ class NegentropySyncResult( * @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 timeoutMs max wait for a single reconcile round or download page's `EOSE`. - * The relay builds its whole negentropy snapshot before the FIRST round responds, - * which is O(matched set) — for a multi-million-event filter that first response can - * take a minute or more (a real strfry took ~73s for an unbounded kind:0 set), so - * raise this for very large syncs. It is only a ceiling — download REQs return on - * their `EOSE`, so a generous value costs nothing in the common case. + * @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. */ @@ -122,7 +131,7 @@ suspend fun INostrClient.negentropySync( maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, - timeoutMs: Long = 30_000L, + idleTimeoutMs: Long = 120_000L, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: (Event) -> Unit, ): NegentropySyncResult { @@ -151,7 +160,7 @@ suspend fun INostrClient.negentropySync( syncWindow( relay = relay, filter = filter, - timeoutMs = timeoutMs, + idleTimeoutMs = idleTimeoutMs, fetchBatch = fetchBatch, maxConcurrentReqs = maxConcurrentReqs, onWindow = { windows++ }, @@ -195,7 +204,7 @@ suspend fun INostrClient.negentropySync( maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, - timeoutMs: Long = 30_000L, + idleTimeoutMs: Long = 120_000L, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: (Event) -> Unit, ): NegentropySyncResult = @@ -205,7 +214,7 @@ suspend fun INostrClient.negentropySync( maxEvents = maxEvents, maxConcurrentReqs = maxConcurrentReqs, fetchBatch = fetchBatch, - timeoutMs = timeoutMs, + idleTimeoutMs = idleTimeoutMs, onProgress = onProgress, onEvent = onEvent, ) @@ -248,7 +257,7 @@ suspend fun INostrClient.negentropySyncOrFetch( maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, - timeoutMs: Long = 30_000L, + idleTimeoutMs: Long = 120_000L, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: (Event) -> Unit, ): NegentropyOrFetchResult { @@ -274,15 +283,17 @@ suspend fun INostrClient.negentropySyncOrFetch( maxEvents = maxEvents, maxConcurrentReqs = maxConcurrentReqs, fetchBatch = fetchBatch, - timeoutMs = timeoutMs, + 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. + // 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 - fetchAllPages(relay, listOf(pageFilter), timeoutMs) { event -> + 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) @@ -295,7 +306,7 @@ suspend fun INostrClient.negentropySyncOrFetch( maxEvents: Int = 0, maxConcurrentReqs: Int = 8, fetchBatch: Int = 500, - timeoutMs: Long = 30_000L, + idleTimeoutMs: Long = 120_000L, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: (Event) -> Unit, ): NegentropyOrFetchResult = @@ -305,7 +316,7 @@ suspend fun INostrClient.negentropySyncOrFetch( maxEvents = maxEvents, maxConcurrentReqs = maxConcurrentReqs, fetchBatch = fetchBatch, - timeoutMs = timeoutMs, + idleTimeoutMs = idleTimeoutMs, onProgress = onProgress, onEvent = onEvent, ) @@ -321,7 +332,7 @@ suspend fun INostrClient.negentropySyncOrFetch( private suspend fun INostrClient.syncWindow( relay: NormalizedRelayUrl, filter: Filter, - timeoutMs: Long, + idleTimeoutMs: Long, fetchBatch: Int, maxConcurrentReqs: Int, onWindow: () -> Unit, @@ -330,7 +341,7 @@ private suspend fun INostrClient.syncWindow( ) { coroutineContext.ensureActive() - when (val outcome = downloadWindow(relay, filter, timeoutMs, fetchBatch, maxConcurrentReqs, onNeed, deliver)) { + when (val outcome = downloadWindow(relay, filter, idleTimeoutMs, fetchBatch, maxConcurrentReqs, onNeed, deliver)) { is ReconcileOutcome.Complete -> onWindow() is ReconcileOutcome.Overflow -> { @@ -347,8 +358,8 @@ private suspend fun INostrClient.syncWindow( ) } else { val mid = lo + (hi - lo) / 2 - syncWindow(relay, filter.copy(since = lo, until = mid), timeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onNeed, deliver) - syncWindow(relay, filter.copy(since = mid + 1, until = hi), timeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onNeed, deliver) + 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) } } @@ -388,7 +399,7 @@ private sealed interface ReconcileOutcome { private suspend fun INostrClient.reconcileStreaming( relay: NormalizedRelayUrl, filter: Filter, - timeoutMs: Long, + idleTimeoutMs: Long, fetchBatch: Int, onNeed: (Int) -> Unit, sendBatch: suspend (List) -> Unit, @@ -402,13 +413,28 @@ private suspend fun INostrClient.reconcileStreaming( // 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)) @@ -426,22 +452,32 @@ private suspend fun INostrClient.reconcileStreaming( // 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. + // 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(timeoutMs) { + withTimeoutOrNull(connectBound) { connectedRelaysFlow().first { targetUrl in it } } - if (connected == null) return ReconcileOutcome.Failed("could not connect within ${timeoutMs}ms") + if (connected == null) return ReconcileOutcome.Failed("could not connect within ${connectBound}ms") relayClient.sendIfConnected(session.open()) while (true) { - // Time only the wait for the relay's next frame — never our own - // back-pressured streaming of the previous frame's ids. + // 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 = - withTimeoutOrNull(timeoutMs) { incoming.receive() } - ?: return ReconcileOutcome.Failed("reconcile round timed out after ${timeoutMs}ms") + 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 -> @@ -508,7 +544,7 @@ private fun isOverflow(reason: String): Boolean = private suspend fun INostrClient.downloadWindow( relay: NormalizedRelayUrl, filter: Filter, - timeoutMs: Long, + idleTimeoutMs: Long, fetchBatch: Int, maxConcurrentReqs: Int, onNeed: (Int) -> Unit, @@ -527,7 +563,7 @@ private suspend fun INostrClient.downloadWindow( launch { for (batch in idBatches) { coroutineContext.ensureActive() - for (event in fetchByIds(relay, batch, timeoutMs)) { + for (event in fetchByIds(relay, batch, idleTimeoutMs)) { deliver(event) } } @@ -535,7 +571,7 @@ private suspend fun INostrClient.downloadWindow( } val outcome = - reconcileStreaming(relay, filter, timeoutMs, fetchBatch, onNeed) { batch -> + reconcileStreaming(relay, filter, idleTimeoutMs, fetchBatch, onNeed) { batch -> idBatches.send(batch) } @@ -558,13 +594,20 @@ private suspend fun INostrClient.downloadWindow( private suspend fun INostrClient.fetchByIds( relay: NormalizedRelayUrl, batch: List, - timeoutMs: Long, + 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( @@ -573,6 +616,7 @@ private suspend fun INostrClient.fetchByIds( relay: NormalizedRelayUrl, forFilters: List?, ) { + clock.bump() if (seen.add(event.id)) collected.add(event) } @@ -602,9 +646,7 @@ private suspend fun INostrClient.fetchByIds( try { subscribe(subId, mapOf(relay to listOf(Filter(ids = batch))), listener) - withTimeoutOrNull(timeoutMs) { - done.receive() - } + done.receiveWithinIdle(clock, batchIdleMs) } finally { unsubscribe(subId) done.close() @@ -625,3 +667,53 @@ private const val DELIVERY_BUFFER = 256 * 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. Thread + * safe: bumped from relay reader threads, read from the driver coroutine. + */ +@OptIn(ExperimentalAtomicApi::class) +private class IdleClock { + private val last = AtomicReference(TimeSource.Monotonic.markNow()) + + fun bump() { + last.store(TimeSource.Monotonic.markNow()) + } + + fun elapsedMs(): Long = last.load().elapsedNow().inWholeMilliseconds +} + +/** + * 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. + } +} From 322e33678aad16316b9b8b5f211cbabe484c245b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 1 Jul 2026 14:18:31 +0000 Subject: [PATCH 9/9] perf(quartz): make the negentropy idle watchdog allocation-free MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit IdleClock.bump() is called for every message the relay sends — the connection listener bumps it per event, so a multi-million-event download bumped it millions of times. It stored a ValueTimeMark into an AtomicReference, and since the value class boxes when used as a generic type argument, every bump allocated a heap object. That is needless GC pressure on the hottest path (and battery/jank on Android). Replace it with a single monotonic base mark taken once (stored unboxed) plus a @Volatile Long of nanos-since-start updated on each bump — zero allocation per bump, and only visibility (not atomicity) is needed since each relay's bumps come from its single reader thread and the driver only reads. Behavior is unchanged; negentropy + concurrency suites pass. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../NostrClientNegentropySyncExt.kt | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) 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 index 689eeafdb6..a5b218cf76 100644 --- 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 @@ -42,8 +42,7 @@ import kotlinx.coroutines.flow.first import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeoutOrNull -import kotlin.concurrent.atomics.AtomicReference -import kotlin.concurrent.atomics.ExperimentalAtomicApi +import kotlin.concurrent.Volatile import kotlin.coroutines.coroutineContext import kotlin.math.min import kotlin.time.TimeSource @@ -680,18 +679,26 @@ 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. Thread - * safe: bumped from relay reader threads, read from the driver coroutine. + * 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. */ -@OptIn(ExperimentalAtomicApi::class) private class IdleClock { - private val last = AtomicReference(TimeSource.Monotonic.markNow()) + private val start = TimeSource.Monotonic.markNow() + + @Volatile + private var lastNanos = 0L fun bump() { - last.store(TimeSource.Monotonic.markNow()) + lastNanos = start.elapsedNow().inWholeNanoseconds } - fun elapsedMs(): Long = last.load().elapsedNow().inWholeMilliseconds + fun elapsedMs(): Long = (start.elapsedNow().inWholeNanoseconds - lastNanos) / 1_000_000 } /**