From e5f43904f2e7d656b699de1a9a569ac299c9dc48 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 23:12:21 +0000 Subject: [PATCH] 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`