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

fix: stop the sync pumps opening unbounded relay subscriptions

Browse Source 
Relays were answering with "too many concurrent REQs" and the emulator
log was full of it. Two independent defects, compounding.

## The subscription flow never completed

RelayPool.queryAsFlow returned a filtered view of the socket's hot
`incomingMessages`, with the terminating operator commented out:

    return this.incomingMessages
        .filterBySubscriptionId(id = subscriptionId)
    //  .transformWhileEventsAreIncoming()

A filter over a hot flow has no terminal event, so every collector
started for a sync request stayed alive for the life of the app --
accumulating one per request ever made, long after EOSE and CLOSE had
been sent. Neither pump's EOSE branch ended collection either;
`return@collect` only ends handling of the message in hand, as the
CLOSED branch's own comment already noted.

That is also why the log *flooded* rather than merely warning.
`filterBySubscriptionId` admits NoticeMessage on every subscription id
(a NOTICE carries none), so a single "too many concurrent REQs" notice
was delivered to every accumulated collector and logged once per
collector. Volume grew as notices x live collectors.

## Nothing bounded how many were open

Both pumps mark the request "sent" and then launch the subscription
detached:

    nostrRepository.negentropySynchronizeRequestProcessed(request)
    launch(Dispatchers.IO) { relaysSocketManager.negentropySync(...).collect { ... } }

The DAO query is `WHERE status = :status ... LIMIT 1`, so flipping the
row changes the head row, Room re-emits, and the collector body runs for
the next request while the previous subscription is still open. The
mutex covers only the setup block and publishSlots guards publishes, not
REQs, so the number of simultaneously open REQ/NEG-OPEN subscriptions
was bounded only by backlog depth.

## And back-pressure amplified itself

The negentropy ClosedMessage branch -- CLOSED being exactly what a relay
sends when refusing for too many concurrent REQs -- answered by queuing
the request again as a plain REQ. Each refusal therefore produced
another subscription. That branch also skipped the close its EOSE and
NEG-MSG siblings performed, leaking a slot precisely when the slot was
most needed.

## The fix, in the order it has to be applied

1. sockets/NostrIncomingMessageExt.kt gains isTerminalFor() and
   completeOnSubscriptionEnd(), which emits the terminal message and
   then completes. NOTICE is deliberately not terminal: with no
   subscription id it reaches every collector on the socket, so treating
   it as terminal would tear down every unrelated subscription at once.
2. RelayPool.queryAsFlow applies it, replacing the commented-out call.
3. SynchronizationViewModel gains subscriptionSlots =
   Semaphore(MAX_CONCURRENT_SUBSCRIPTIONS = 4), acquired inside each
   pump's launch before the socket call, so the backlog still drains but
   queues on the semaphore rather than opening all at once.
4. Both pumps close in a `finally` under NonCancellable, replacing the
   hand-rolled closes in the EOSE and NEG-MSG branches, so CLOSED and
   NEG-ERR exits close too.
5. The negentropy CLOSED branch consults isBackPressure() -- NIP-01's
   `rate-limited:` prefix plus the free-text forms relays actually send
   -- and declines to retry, instead of answering back-pressure by
   opening another subscription.

Order is load-bearing: capping slots before the flow could complete
would have deadlocked the pump on permits that never came back.

## A negentropy assumption that would have deadlocked it anyway

Capping slots nearly stalled the queue on a wrong assumption about what
ends a negentropy exchange. EOSE does not: the NegentropyMessage branch
reconciles ONCE, queues the ids it needs as a plain REQ, schedules what
the relay is missing, and stops -- this client does single-round
reconciliation. Waiting on an EOSE the exchange need not send would have
held all four slots forever. NegentropyMessage is therefore terminal
too, with the reasoning recorded at isTerminalFor().

Worth knowing separately, and left alone here: single-round
reconciliation may not converge on large sets, since negentropy is
normally iterative. A large divergence is closed by the plain-REQ
fallback rather than by negentropy itself.

Live collectors go from one per sync request ever made to at most four.
MAX_CONCURRENT_SUBSCRIPTIONS is the dial if sync feels slow -- relays
commonly allow around 20 per connection, so there is headroom.

Verified:
  ./gradlew :composeApp:compileCommonMainKotlinMetadata
  ./gradlew :composeApp:compileDebugKotlinAndroid

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
2026-08-29 20:04:06 +02:00
parent e1b79ad842
commit 0d6cefbe21
3 changed files with 122 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,6 +4,7 @@ import androidx.annotation.VisibleForTesting
import press.mantra.compose.network.dto.toRelayDTO
import press.mantra.compose.network.sockets.filterByEventId
import press.mantra.compose.network.sockets.filterBySubscriptionId
import press.mantra.compose.network.sockets.completeOnSubscriptionEnd
import co.touchlab.kermit.Logger
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
@@ -277,7 +278,11 @@ class RelayPool(
private suspend fun press.mantra.compose.network.sockets.NostrSocketClient.queryAsFlow(subscriptionId: String): Flow<press.mantra.compose.network.sockets.NostrIncomingMessage> {
return this.incomingMessages
.filterBySubscriptionId(id = subscriptionId)
// .transformWhileEventsAreIncoming()
// The socket's incomingMessages is a hot flow, so a filtered view of it
// never completes on its own. Ending it at EOSE/CLOSED/NEG-ERR is what
// lets a caller's collector finish, its subscription slot be released,
// and the relay-side subscription actually be closed.
.completeOnSubscriptionEnd(id = subscriptionId)
}
@OptIn(FlowPreview::class)

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

@@ -2,6 +2,43 @@ package press.mantra.compose.network.sockets
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.transformWhile
/**
* True for a message that ends a subscription: the relay will send nothing more
* under this id.
*
* 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.
*/
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)
/**
* Completes the flow once the subscription is over, emitting the terminal
* message first so callers still see the EOSE/CLOSED/NEG-ERR that ended it.
*
* Without this the returned flow is a filtered view of the socket's hot
* `incomingMessages` and so never completes: every collector started for a sync
* request stayed alive for the lifetime of the app, accumulating one per request
* ever made. Because a NOTICE is admitted on every id, each relay notice was then
* logged once per accumulated collector — which is what turned an occasional
* "too many concurrent REQs" into a flooded log.
*/
fun Flow<NostrIncomingMessage>.completeOnSubscriptionEnd(id: String) =
transformWhile { message ->
emit(message)
!message.isTerminalFor(id)
}
fun Flow<NostrIncomingMessage>.filterBySubscriptionId(id: String) =
filter {

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

@@ -44,6 +44,8 @@ import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.timeout
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withLock
@@ -70,6 +72,13 @@ class SynchronizationViewModel(
private const val MAX_CONCURRENT_PUBLISHES = 8
/**
* Well under the ~20-per-connection cap relays commonly enforce, and applied
* across all relays rather than per relay, which keeps the emulator's socket
* count sane too.
*/
private const val MAX_CONCURRENT_SUBSCRIPTIONS = 4
/**
* Hard ceiling on a single publish attempt, covering the suspend calls that open the
* socket as well as the response flow, so a request can never sit in "processing".
@@ -105,6 +114,18 @@ class SynchronizationViewModel(
/** Caps how many publishes may be in flight while the queue drains a backlog. */
private val publishSlots = Semaphore(MAX_CONCURRENT_PUBLISHES)
/**
* Caps how many REQ/NEG-OPEN subscriptions may be open at once.
*
* The queue advances the moment a request is marked "sent", so without this the
* pump opened a subscription per pending row with nothing awaiting the previous
* one — relays answered with "too many concurrent REQs" and, because a NOTICE is
* delivered to every collector on the socket, logged it once per open collector.
* Requests queue on this semaphore instead, so the backlog still drains, just not
* all at once.
*/
private val subscriptionSlots = Semaphore(MAX_CONCURRENT_SUBSCRIPTIONS)
/**
* Keeps one bad request from killing the pump that is draining the queue. Mirrors
* `NotaryViewModel.guardNotarization`.
@@ -119,6 +140,21 @@ class SynchronizationViewModel(
}
}
/**
* Whether a relay's CLOSED/NOTICE reason means "you are asking for too much"
* rather than "I cannot serve this". NIP-01 gives `rate-limited:` as the
* machine-readable prefix; the free-text forms are what relays actually send.
*/
private fun isBackPressure(reason: String?): Boolean {
val text = reason?.lowercase() ?: return false
return text.startsWith("rate-limited") ||
text.contains("rate limit") ||
text.contains("too many") ||
text.contains("concurrent") ||
text.contains("slow down")
}
init {
scope.launch {
activeWalletStateFlow.collectLatest { activeWallet ->
@@ -186,6 +222,7 @@ class SynchronizationViewModel(
nostrRepository.synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest)
launch(Dispatchers.IO) {
subscriptionSlots.withPermit {
try {
relaysSocketManager.query(
reqCommand,
@@ -221,14 +258,8 @@ class SynchronizationViewModel(
}
is NostrIncomingMessage.EoseMessage -> {
logger.d("Sync request has been successfully processed (${synchronizeNostrEventRequest.relayURL}): $nostrIncomingMessage")
val closeCommand = CloseCmd(
subId = synchronizeNostrEventRequest.id,
)
relaysSocketManager.closeQuery(
closeCommand,
synchronizeNostrEventRequest.relayURL
)
// EOSE ends the flow (completeOnSubscriptionEnd);
// the finally below sends the CLOSE.
}
is NostrIncomingMessage.ClosedMessage -> {
// The relay ended the subscription on its side (auth
@@ -247,7 +278,21 @@ class SynchronizationViewModel(
} catch (e: Throwable) {
logger.e("Failed to sync", e)
} finally {
// Close on every exit, not just the EOSE branch. A
// subscription the relay never terminates would otherwise
// hold a slot on both sides for the life of the app.
// NonCancellable so a cancelled pump still says goodbye.
withContext(NonCancellable) {
runCatching {
relaysSocketManager.closeQuery(
CloseCmd(subId = synchronizeNostrEventRequest.id),
synchronizeNostrEventRequest.relayURL
)
}
}
}
}
}
}
}
@@ -311,11 +356,11 @@ class SynchronizationViewModel(
nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest)
launch(Dispatchers.IO) {
subscriptionSlots.withPermit {
val negCloseCmd = NegCloseCmd(
subId = negentropySynchronizeRequest.uuid,
)
try {
val negCloseCmd = NegCloseCmd(
subId = negentropySynchronizeRequest.uuid,
)
relaysSocketManager.negentropySync(
negOpenCmd,
negentropySynchronizeRequest.relayURL
@@ -351,11 +396,6 @@ class SynchronizationViewModel(
is NostrIncomingMessage.EoseMessage -> {
logger.d("Sync request has been successfully processed (${negentropySynchronizeRequest.relayURL}): $nostrIncomingMessage")
relaysSocketManager.closeNegentropySync(
negCloseCmd,
negentropySynchronizeRequest.relayURL
)
return@collect
}
is NostrIncomingMessage.NegentropyError -> {
@@ -415,11 +455,6 @@ class SynchronizationViewModel(
broadcastNostrEventRequests
)
relaysSocketManager.closeNegentropySync(
negCloseCmd,
negentropySynchronizeRequest.relayURL
)
return@collect
}
is NostrIncomingMessage.ClosedMessage -> {
@@ -430,7 +465,14 @@ class SynchronizationViewModel(
// (`return@collect` ends handling of this message,
// matching the EOSE and NEG-ERR branches.)
logger.w("Relay closed negentropy subscription (${negentropySynchronizeRequest.relayURL}): ${nostrIncomingMessage.message}")
if (negentropySynchronizeRequest.purpose != "mlsMessages") {
// Falling back to a plain REQ is right when the
// relay cannot serve negentropy, and wrong when it
// is telling us to ease off: answering back-pressure
// by opening another subscription is what turned one
// refusal into a flood of them.
if (isBackPressure(nostrIncomingMessage.message)) {
logger.w("Relay ${negentropySynchronizeRequest.relayURL} is rate limiting; not retrying this request")
} else if (negentropySynchronizeRequest.purpose != "mlsMessages") {
nostrRepository.queueSynchronizeNostrEvent(
listOf(
negentropySynchronizeRequest.toSynchronizeNostrEventRequest()
@@ -466,7 +508,21 @@ class SynchronizationViewModel(
logger.d("Queried Sync")
} catch (e: Throwable) {
logger.e("Failed to sync", e)
} finally {
// Close on every exit. The EOSE and NEG-MSG branches used
// to do this by hand while CLOSED and NEG-ERR did not, so a
// subscription leaked every time a relay refused one —
// exactly the case that needs the slot back most.
withContext(NonCancellable) {
runCatching {
relaysSocketManager.closeNegentropySync(
negCloseCmd,
negentropySynchronizeRequest.relayURL
)
}
}
}
}
}
logger.i("After launch")