mantra/mantra-kmp
1
0
Fork 0
Code Issues 2 Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

fix: reconcile with a relay to completion instead of stopping after one round

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
2026-09-05 11:27:04 +02:00
parent 661a5caa17
commit 2d29fc0a37
4 changed files with 162 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt
View File

@@ -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(

8
composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt
View File

@@ -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,

19
composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt
View File

@@ -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

164
composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt
View File

@@ -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<String>()
val sendIds = LinkedHashSet<String>()
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<String>,
sendIds: Set<String>,
localEventIds: Set<String>,
) {
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