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 10b83f83..8d385011 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 @@ -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 { 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) 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 0a05d658..ce8bbc24 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 @@ -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.completeOnSubscriptionEnd(id: String) = + transformWhile { message -> + emit(message) + !message.isTerminalFor(id) + } fun Flow.filterBySubscriptionId(id: String) = filter { 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 b55b444f..1ef97982 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 @@ -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")