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

fix: stop one bad request or one silent relay from stalling all synchronization

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

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

@@ -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<ActiveWallet?>,
@@ -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<press.mantra.compose.database.model.NostrEvent>.toNegentropy(): Negentropy {
val storage = StorageVector()
val seen = HashSet<String>(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.