From d9d27cd0ea2537e4b498641fb111f0cc3adf6477 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Sat, 5 Sep 2026 11:27:44 +0200 Subject: [PATCH] fix: stop one bad request or one silent relay from stalling all synchronization Two ways the sync queues could stop draining and never recover. ## A request that throws while loading the local set is never retried, and blocks ## every request behind it The pending queue is a single row at a time: SELECT * FROM NegentropySynchronizeRequest WHERE status = 'pending' ORDER BY createdAt ASC, id ASC LIMIT 1 observed through distinctUntilChanged. The pump advances only when the head row changes status, and the negentropy request was marked "sent" AFTER the storage vector was built. StorageVector can throw on the way in -- insert() requires exactly 64 hex characters, and seal() rejects a duplicate (timestamp, id) with "duplicate item inserted". guardPump caught the throw and logged it, which kept the pump alive but left the row at "pending". Nothing else observes that status, the flow will not re-emit an unchanged row, so the request was neither retried nor skipped: it sat at the head of the queue and every negentropy request queued after it waited behind it for the life of the process. The vector build now filters and de-duplicates on the way in -- a row negentropy cannot index is one this device cannot reconcile, and dropping it costs one event's worth of extra transfer where letting it through costs the entire sync -- and the request is claimed either way, so a failure that does get through logs and lets the queue move on. ## A relay that opens a subscription and then goes quiet parks a slot forever Both pumps take a permit from subscriptionSlots (4 across all relays) and hold it for the life of the collection. The collection ends on EOSE, CLOSED or NEG-ERR -- none of which a relay is obliged to send. A negentropy exchange in particular ends when reconcile() says so; if the relay simply stops answering mid-round, nothing completes the flow. Four such subscriptions hold every permit and the queue stops, with no error anywhere: the requests are marked "sent", so the UI's pending count reads zero while nothing is being fetched. Both are now bounded by SUBSCRIPTION_TIMEOUT (120s), which covers the collection itself. The REQ pump is included because it is how negentropy's needIds are actually fetched -- a wedged REQ slot breaks negentropy sync just as directly as a wedged NEG one. Generous rather than tight: cutting a slow but live download short costs a re-fetch next pass, and a REQ can now carry up to 500 ids. The existing NEG-CLOSE/CLOSE in the finally block already runs under NonCancellable, so a timed-out subscription still says goodbye to the relay. Not covered by tests: both failures are timing and Room behaviour on the sync path, neither of which runs under :composeApp:testDebugUnitTest. Verified by compilation and by reading the queue's DAO query against the pump's collection. Co-Authored-By: Claude Opus 5 --- .../ui/view/model/SynchronizationViewModel.kt | 82 +++++++++++++++---- 1 file changed, 65 insertions(+), 17 deletions(-) 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 b6ff99fc..9f8fa9aa 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 @@ -53,6 +53,7 @@ import kotlinx.coroutines.sync.Semaphore import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.sync.withPermit import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.seconds class SynchronizationViewModel( val activeWalletStateFlow: StateFlow, @@ -87,6 +88,15 @@ class SynchronizationViewModel( */ private val PUBLISH_ATTEMPT_TIMEOUT = (RelayPool.PUBLISH_TIMEOUT * 2).milliseconds + /** + * Hard ceiling on one REQ or negentropy exchange. Both hold a subscription slot for + * their whole life, and a relay that opens a subscription and then goes quiet owes us + * no EOSE — without a bound, four such relays park every slot and the queue behind + * them never drains. Generous rather than tight: the cost of cutting a slow but live + * download short is re-fetching it next pass, and a REQ can carry MAX_IDS_PER_REQ ids. + */ + private val SUBSCRIPTION_TIMEOUT = 120.seconds + /** * 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 @@ -100,6 +110,8 @@ class SynchronizationViewModel( */ private const val MAX_IDS_PER_REQ = 500 + private const val NOSTR_EVENT_ID_HEX_LENGTH = 64 + private val mutex = Mutex() fun factory( @@ -239,6 +251,10 @@ class SynchronizationViewModel( launch(Dispatchers.IO) { subscriptionSlots.withPermit { try { + // Bounded for the same reason the negentropy exchange is: a relay + // that answers a REQ with neither EOSE nor CLOSED would otherwise + // hold a subscription slot for the life of the app. + withTimeout(SUBSCRIPTION_TIMEOUT) { relaysSocketManager.query( reqCommand, synchronizeNostrEventRequest.relayURL @@ -290,6 +306,7 @@ class SynchronizationViewModel( } } } + } } catch (e: Throwable) { logger.e("Failed to sync", e) @@ -332,24 +349,18 @@ class SynchronizationViewModel( applyLimits = false ) logger.d("Events: ${events.size}") - val storage = StorageVector().apply { - events.forEach { event -> - // Nostr timestamps — and therefore every timestamp a relay - // puts in its own negentropy vector — are in SECONDS. Feeding - // milliseconds here sorted every local item ~1000x past every - // remote one, so the fingerprint bounds could never match and - // reconciliation degenerated into a full ID transfer. - insert( - timestamp = event.createdAt.epochSeconds, - idHex = event.id - ) - } - seal() - } - val negentropy = Negentropy( - storage, - ) + // The vector is built before the request leaves "pending", and building it + // can throw (a malformed id, a duplicate). The pending queue is a single + // oldest-row-first observation, so a row that throws here and never + // changes status is never retried AND blocks every negentropy request + // behind it for the life of the process. Claim the row either way. + val negentropy = runCatching { events.toNegentropy() } + .onFailure { error -> + logger.e("Failed to build the negentropy vector for ${negentropySynchronizeRequest.id}", error) + nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest) + } + .getOrNull() ?: return@withLock val negOpenCmd = NegOpenCmd( subId = negentropySynchronizeRequest.uuid, @@ -393,6 +404,7 @@ class SynchronizationViewModel( var reconciliationOver = false try { + withTimeout(SUBSCRIPTION_TIMEOUT) { relaysSocketManager.negentropySync( negOpenCmd, negentropySynchronizeRequest.relayURL @@ -535,6 +547,7 @@ class SynchronizationViewModel( } } } + } logger.d("Queried Sync") } catch (e: Throwable) { logger.e("Failed to sync", e) @@ -577,6 +590,41 @@ class SynchronizationViewModel( } } + /** + * Loads the local set into a sealed negentropy vector. + * + * `StorageVector` rejects an id that is not 32 bytes and, on seal, a duplicate item — either + * throws out of the whole request — so the set is filtered and de-duplicated on the way in. + * A row the vector cannot hold is a row this device cannot reconcile; dropping it costs one + * event's worth of extra transfer, where letting it through costs the entire sync. + */ + private fun List.toNegentropy(): Negentropy { + val storage = StorageVector() + val seen = HashSet(size) + + forEach { event -> + if (event.id.length != NOSTR_EVENT_ID_HEX_LENGTH || !event.id.all { it.isHex() }) { + logger.w("Skipping event with an id negentropy cannot index: ${event.id}") + return@forEach + } + if (!seen.add(event.id)) return@forEach + + // Nostr timestamps — and therefore every timestamp a relay puts in its own + // negentropy vector — are in SECONDS. Feeding milliseconds here sorted every local + // item ~1000x past every remote one, so the fingerprint bounds could never match and + // reconciliation degenerated into a full ID transfer. + storage.insert( + timestamp = event.createdAt.epochSeconds, + idHex = event.id + ) + } + storage.seal() + + return Negentropy(storage) + } + + private fun Char.isHex() = this in '0'..'9' || this in 'a'..'f' || this in 'A'..'F' + /** * Turns a finished reconciliation into work: fetch what only the relay has, offer what only * we have.