From 2d29fc0a37eeecc33037169b5ee9a560d6d66e85 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Sat, 5 Sep 2026 11:27:04 +0200 Subject: [PATCH] fix: reconcile with a relay to completion instead of stopping after one round Negentropy is a multi-round protocol. The initiator opens with fingerprints over its whole set -- 16 buckets, per kmp-negentropy's BUCKETS_IN_MESSAGE -- and the peer answers each bucket either by agreeing (a skip), by listing the ids in that range, or, when the range still holds more than 32 items on its side, by splitting it into 16 finer fingerprints. Only the ranges that come back as id lists produce have/need ids. Everything still under a fingerprint needs another NEG-MSG from us, and reconcile() says so by returning a non-null `msg`; it returns null exactly when there is nothing left to ask about. This client discarded result.msg and never sent a second NEG-MSG. Worse, isTerminalFor() listed NegentropyMessage as terminal, so completeOnSubscriptionEnd ended the flow on the FIRST one -- the collector finished, the finally block sent NEG-CLOSE, and a reconciliation the relay was still in the middle of was abandoned. With 16 buckets a single round tells you almost nothing about a set of any size: for anything past a couple of dozen events the exchange was torn down before it had located most of the difference, and the ids it did find were whichever handful happened to resolve at depth one. The old comment on isTerminalFor described this as a deliberate design ("this client reconciles in a single round"), which is what kept it in place. It is not a design one can choose -- the protocol has no single-round mode. What it produced was a sync that mostly did not sync, hidden behind a diff that was never empty and a REQ fallback that quietly did the real work. ## The loop NegentropyMessage is no longer terminal. The collector feeds each NEG-MSG to reconcile(), accumulates the round's needIds/sendIds, and while `msg` is non-null sends it straight back on the same subscription via the new RelayPool.sendNegentropyMessage. When reconcile() returns null the exchange is over -- a fact only the caller can see, since a relay owes us no EOSE for a NEG session -- so a `transformWhile` on the flow ends the collection there. The predicate reads a flag the collector sets, which works because a flow's downstream collector runs synchronously inside emit(). MAX_NEGENTROPY_ROUNDS caps the ping-pong at 32 in case a peer's ranges never converge; a healthy exchange settles in far fewer, since each round splits the disagreeing ranges 16 ways. ## Acting once, at the end Follow-ups moved out of the per-message branch into applyReconciliation, called after the exchange. Acting per round would have queued a REQ for ids that later rounds were still discovering. It runs outside the try and under NonCancellable so an exchange that is cut short still acts on what it did reconcile rather than discarding the rounds it paid for. Two fixes came with the move: - needIds go out chunked at 500 per REQ. Relays cap the length of a filter's `ids` array (1000 is common) and a first sync can reconcile thousands; a single oversized REQ is answered with a CLOSED, or silently truncated, which loses every id past the cap. Previously all of them went in one filter -- survivable only because one round never found many. - the "do we actually hold this?" check on sendIds is a Set lookup instead of `in` on a List, which was a linear scan per id over the whole local set. Also dropped two logger.d calls that dumped every local event id and every local timestamp on each NEG-MSG. At one line per message that was tolerable; at one per round over a real set it is megabytes of logging on the hot path. Not covered by tests: this is websocket exchange behaviour with a live relay. Verified by compilation and by tracing kmp-negentropy's Negentropy.reconcile against quartz's own NegentropySession, whose documented usage is the same loop ("If processMessage returns a non-null NegMsgCmd, send it back / repeat until a result with a null command"). Co-Authored-By: Claude Opus 5 --- .../compose/network/relays/RelayPool.kt | 13 ++ .../network/relays/RelaysSocketManager.kt | 8 + .../sockets/NostrIncomingMessageExt.kt | 19 +- .../ui/view/model/SynchronizationViewModel.kt | 164 ++++++++++++++---- 4 files changed, 162 insertions(+), 42 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt index 8d385011..0d0b0ad0 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt @@ -12,6 +12,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.displayUrl import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers @@ -225,6 +226,18 @@ class RelayPool( } } + /** + * Sends the next round of a negentropy exchange. The relay answers on the same + * subscription, so the caller keeps collecting the flow it already opened with + * [negentropySync] rather than opening a new one. + */ + suspend fun sendNegentropyMessage(negMsgCmd: NegMsgCmd, relayUrl: String) { + val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() } + ?: throw press.mantra.compose.exceptions.NetworkException("$relayUrl is not connected") + + nostrSocketClient.sendMESSAGE(OptimizedJsonMapper.toJson(negMsgCmd)) + } + suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) { addRelaysIfMissing( setOf( diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt index 7c81a050..fe9bc561 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt @@ -6,6 +6,7 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd import fr.acinq.phoenix.data.ActiveWallet import kotlinx.coroutines.CancellationException @@ -157,6 +158,13 @@ class RelaysSocketManager( ) } + suspend fun sendNegentropyMessage(negMsgCmd: NegMsgCmd, relayUrl: String) { + return relayPool.sendNegentropyMessage( + negMsgCmd = negMsgCmd, + relayUrl = relayUrl + ) + } + suspend fun closeNegentropySync(negCloseCmd: NegCloseCmd, relayUrl: String) { return relayPool.closeNegentropySync( negCloseCmd = negCloseCmd, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt index ce8bbc24..153a6d4e 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt @@ -11,17 +11,22 @@ import kotlinx.coroutines.flow.transformWhile * NOTICE is deliberately absent. It carries no subscription id, so it is * delivered to every collector on the socket — treating it as terminal would * let one relay notice tear down every unrelated subscription at once. + * + * NEG-MSG is absent too, and for a sharper reason. Negentropy is a multi-round + * protocol: the initiator opens with fingerprints over its whole set, and each + * NEG-MSG the relay sends back resolves some ranges into id lists while + * splitting others into finer fingerprints the client must answer with another + * NEG-MSG. Only the ranges that come back as id lists produce have/need ids at + * all, and a message carries 16 buckets — so one round says almost nothing about + * a set of any size. Treating the first NEG-MSG as terminal made every sync a + * single round and abandoned a reconciliation the relay was still in the middle + * of. A negentropy exchange instead ends when `reconcile()` reports no further + * message to send, which only the caller can see (see SynchronizationViewModel). */ fun NostrIncomingMessage.isTerminalFor(id: String): Boolean = (this is NostrIncomingMessage.EoseMessage && subscriptionId == id) || (this is NostrIncomingMessage.ClosedMessage && subscriptionId == id) || - (this is NostrIncomingMessage.NegentropyError && subscriptionId == id) || - // A NEG-MSG ends the exchange because this client reconciles in a single - // round: it diffs once, queues the ids it needs as a plain REQ, schedules - // what the relay is missing, and closes. Waiting for an EOSE that a - // negentropy exchange need not send would hold the subscription open - // forever — and, now that slots are capped, stall the queue behind it. - (this is NostrIncomingMessage.NegentropyMessage && subscriptionId == id) + (this is NostrIncomingMessage.NegentropyError && subscriptionId == id) /** * Completes the flow once the subscription is over, emitting the terminal diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt index 1ef97982..b6ff99fc 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt @@ -24,6 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd import fr.acinq.phoenix.data.ActiveWallet import fr.acinq.phoenix.managers.nostrPrivateKey @@ -43,6 +44,7 @@ import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.timeout +import kotlinx.coroutines.flow.transformWhile import kotlinx.coroutines.launch import kotlinx.coroutines.withContext import kotlinx.coroutines.NonCancellable @@ -85,6 +87,19 @@ class SynchronizationViewModel( */ private val PUBLISH_ATTEMPT_TIMEOUT = (RelayPool.PUBLISH_TIMEOUT * 2).milliseconds + /** + * Backstop for a peer whose ranges never converge. Each round splits the disagreeing + * ranges 16 ways, so a well-behaved exchange over even a very large set settles in a + * handful; this only bounds the pathological case. + */ + private const val MAX_NEGENTROPY_ROUNDS = 32 + + /** + * Ids per fallback REQ. Relays cap the size of a filter's `ids` array (1000 is the + * common limit), and a first sync can reconcile far more than that. + */ + private const val MAX_IDS_PER_REQ = 500 + private val mutex = Mutex() fun factory( @@ -355,16 +370,41 @@ class SynchronizationViewModel( nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest) + // The ids we hold, for deciding which of the relay's "you have what I + // don't" answers we can actually broadcast. A set, because reconcile() can + // report thousands of them and this used to be a list scan per id — and + // taken here rather than inside the coroutine below so that a long + // exchange does not keep every loaded event (content and all) alive for + // its whole life. + val localEventIds = events.mapTo(HashSet(events.size)) { it.id } + launch(Dispatchers.IO) { subscriptionSlots.withPermit { val negCloseCmd = NegCloseCmd( subId = negentropySynchronizeRequest.uuid, ) + // Reconciliation is incremental: each round resolves some ranges into + // ids and splits the rest. Accumulate across rounds and act once, at + // the end — acting per round would queue a REQ per round for ids that + // later rounds are still discovering. + val needIds = LinkedHashSet() + val sendIds = LinkedHashSet() + var rounds = 0 + var reconciliationOver = false + try { relaysSocketManager.negentropySync( negOpenCmd, negentropySynchronizeRequest.relayURL - ).collect { nostrIncomingMessage -> + ).transformWhile { message -> + // The collector below runs inside this emit, so by the time + // the predicate is read it has already recorded whether the + // exchange finished. Without this the flow would only end on + // EOSE/CLOSED/NEG-ERR, and a relay owes us none of those once + // reconciliation completes. + emit(message) + !reconciliationOver + }.collect { nostrIncomingMessage -> when (nostrIncomingMessage) { is NostrIncomingMessage.EventMessage -> { launch(Dispatchers.IO) { @@ -413,48 +453,38 @@ class SynchronizationViewModel( return@collect } is NostrIncomingMessage.NegentropyMessage -> { - logger.d("NegentropyMessage: ${nostrIncomingMessage.negentropyMessage}") + rounds++ + logger.d("NegentropyMessage (round $rounds): ${nostrIncomingMessage.negentropyMessage}") val result = negentropy.reconcile( nostrIncomingMessage.negentropyMessage.hexToByteArray() ) - logger.d("NeedIds: ${result.needIds.map { it.toHexString() }}") - logger.d("SendIds: ${result.sendIds.map { it.toHexString() }}") - logger.d("EventsIds: ${events.map { it.id }}") - logger.d("Timestamp: ${events.map { it.createdAt.epochSeconds }}") + result.needIds.mapTo(needIds) { it.toHexString() } + result.sendIds.mapTo(sendIds) { it.toHexString() } + logger.d("Round $rounds: +${result.needIds.size} need, +${result.sendIds.size} send") - if (result.needIds.isNotEmpty()) { - // Schedule a sync from this relay... - nostrRepository.queueSynchronizeNostrEvent( - listOf( - SynchronizeNostrEventRequest( - purpose = negentropySynchronizeRequest.purpose, - synchronizationFilters = arrayOf( - SynchronizationFilter( - ids = result.needIds.map { it.toHexString() } - .toTypedArray() - ) - ), - relayURL = negentropySynchronizeRequest.relayURL, - level = negentropySynchronizeRequest.level, - ) - ) - ) + val nextMessage = result.msg + // A null message is the library saying "nothing left + // to ask about" — that, not the arrival of the first + // NEG-MSG, is where an exchange is over. The round cap + // is a backstop against a peer that keeps splitting + // ranges forever; reconciliation halves the search + // space each round, so a healthy one is far shorter. + if (nextMessage == null || rounds >= MAX_NEGENTROPY_ROUNDS) { + if (nextMessage != null) { + logger.w("Negentropy with ${negentropySynchronizeRequest.relayURL} did not settle in $MAX_NEGENTROPY_ROUNDS rounds; using what reconciled so far") + } + reconciliationOver = true + return@collect } - val eventIds = events.map { it.id } - val broadcastNostrEventRequests = result.sendIds.filter { it.toHexString() in eventIds }.map { sendId -> - BroadcastNostrEventRequest( - nostrEventId = sendId.toHexString(), - relayURL = negentropySynchronizeRequest.relayURL - ) - } - logger.d("broadcastNostrEventRequests: $broadcastNostrEventRequests") - // Schedule broadcastNostrEventRequests - nostrRepository.rescheduleBroadcastNostrEventRequests( - broadcastNostrEventRequests + relaysSocketManager.sendNegentropyMessage( + NegMsgCmd( + subId = negentropySynchronizeRequest.uuid, + message = nextMessage.toHexString(), + ), + negentropySynchronizeRequest.relayURL ) - return@collect } is NostrIncomingMessage.ClosedMessage -> { @@ -522,6 +552,20 @@ class SynchronizationViewModel( } } } + + // Outside the try, and NonCancellable, so a session that timed out or + // was cancelled mid-exchange still acts on what it did reconcile + // instead of throwing the rounds it paid for away. + withContext(NonCancellable) { + runCatching { + applyReconciliation( + negentropySynchronizeRequest = negentropySynchronizeRequest, + needIds = needIds, + sendIds = sendIds, + localEventIds = localEventIds, + ) + }.onFailure { logger.e("Failed to schedule reconciliation follow-ups", it) } + } } } @@ -533,6 +577,56 @@ class SynchronizationViewModel( } } + /** + * Turns a finished reconciliation into work: fetch what only the relay has, offer what only + * we have. + */ + private suspend fun applyReconciliation( + negentropySynchronizeRequest: press.mantra.compose.database.model.NegentropySynchronizeRequest, + needIds: Set, + sendIds: Set, + localEventIds: Set, + ) { + logger.d("Reconciled with ${negentropySynchronizeRequest.relayURL}: ${needIds.size} to fetch, ${sendIds.size} to offer") + + if (needIds.isNotEmpty()) { + // One REQ per chunk. A first sync can need thousands of ids, and relays cap the + // length of a filter's `ids` array — a single oversized REQ is answered with a + // CLOSED (or silently truncated), which loses every id past the cap. + nostrRepository.queueSynchronizeNostrEvent( + needIds.chunked(MAX_IDS_PER_REQ).map { chunk -> + SynchronizeNostrEventRequest( + purpose = negentropySynchronizeRequest.purpose, + synchronizationFilters = arrayOf( + SynchronizationFilter( + ids = chunk.toTypedArray() + ) + ), + relayURL = negentropySynchronizeRequest.relayURL, + level = negentropySynchronizeRequest.level, + ) + } + ) + } + + // Only offer events we actually hold. `sendIds` is what the relay is missing relative to + // our vector, so everything in it should be local — the guard is against a peer that + // echoes ids we never sent. + val broadcastNostrEventRequests = sendIds.filter { it in localEventIds }.map { sendId -> + BroadcastNostrEventRequest( + nostrEventId = sendId, + relayURL = negentropySynchronizeRequest.relayURL + ) + } + logger.d("broadcastNostrEventRequests: ${broadcastNostrEventRequests.size}") + + if (broadcastNostrEventRequests.isNotEmpty()) { + nostrRepository.rescheduleBroadcastNostrEventRequests( + broadcastNostrEventRequests + ) + } + } + @OptIn(FlowPreview::class) private suspend fun observePendingBroadcastNostrEventRequests( keyPair: KeyPair