From 0ef0a33350d7066bbc18d19de3799564b26b4a67 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Sat, 5 Sep 2026 16:15:51 +0200 Subject: [PATCH] feat(chat): close live subscriptions in the background, catch up on return MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Subscriptions that stay open are only free while the app is on screen. Holding a socket open behind a doze window achieves nothing but battery, and the relay drops the subscription anyway — so this ties them to the app lifecycle and adds the reconciliation that covers the gap. Nothing in this app observed the app lifecycle at all: MainActivity only calls setContent. AppLifecycle is a small singleton holding an isForeground StateFlow, fed from a LifecycleEventObserver in MantraNavHost (ON_START/ON_STOP), and read by LiveSubscriptionManager. A singleton rather than something threaded through the composition because the consumers are not composables — they are application-scoped coroutines started before any screen exists and outliving all of them. It defaults to foreground: on a platform that has not wired the observer up, "always on" is the behaviour that predates this file, and a subscription that never opens is a far worse failure than one that stays open too long. collectLatest over that flow is the entire mechanism. Backgrounding cancels the block holding the subscriptions, and each one's finally sends its CLOSE and releases the retained REQ on the way out — which is also what tells the socket it no longer has a reason to reconnect. On the way back: **Reconnect before asking for anything.** RelayPool.reconnectAll tears every socket down and immediately rebuilds it. Trusting the connection is the mistake here: a socket that was open when the OS suspended the process reports itself connected on the way back while being functionally dead. Re-opening eagerly rather than leaving it to the next send is deliberate — it is what makes this a RE-connection, so retained subscriptions are replayed and any collector still attached from before the gap starts receiving again. That also covers queue requests that were mid-flight when we went away, which would otherwise sit until SUBSCRIPTION_TIMEOUT. **Then a catch-up reconciliation.** A live subscription answers "what is new since I connected"; negentropy answers "what do you have that I don't". Coming back from a gap is exactly the question only the second can answer — `limit` on the re-opened subscriptions is a window, not a guarantee. queueCatchUpSynchronization queues the same two negentropy requests ChatRoomListViewModel queues on open: gift wraps p-tagged to us, and group events h-tagged with every group we are in. Deliberately the same filter shape as the screen's, down to limit=50. A negentropy request is stored under a hash of its filter, so an identical shape collapses into one row instead of queueing the same reconciliation twice while both callers exist. The value itself barely matters — the negentropy pump drops `limit` outright and it only survives into the plain-REQ fallback. The room-to-group-id rule (has MLS state, not left, not deleted, sorted) now lives in one place, since the catch-up and the subscription reconcile have to agree on what "a group we are in" means. Not covered here: connectivity changes. A network switch mid-foreground is still only noticed by the socket's own reconnect loop, which handles the common case but cannot know the network changed underneath it. That wants a platform connectivity observer, and is its own change. Co-Authored-By: Claude Opus 5 --- .../press/mantra/compose/AppLifecycle.kt | 32 +++++ .../managers/LiveSubscriptionManager.kt | 128 ++++++++++++++++-- .../compose/network/relays/RelayPool.kt | 21 +++ .../network/relays/RelaysSocketManager.kt | 3 + .../ui/composable/navigation/MantraNavHost.kt | 20 +++ .../ui/view/model/SynchronizationViewModel.kt | 2 + 6 files changed, 192 insertions(+), 14 deletions(-) create mode 100644 composeApp/src/commonMain/kotlin/press/mantra/compose/AppLifecycle.kt diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/AppLifecycle.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/AppLifecycle.kt new file mode 100644 index 00000000..7f833932 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/AppLifecycle.kt @@ -0,0 +1,32 @@ +package press.mantra.compose + +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow + +/** + * Whether the app is in the foreground, for the parts of the app that hold network + * connections open. + * + * A singleton rather than something threaded through the composition, because the consumers + * are not composables: they are long-lived coroutines at application scope, started before + * any screen exists and outliving all of them. + * + * Defaults to foreground. Nothing here can observe a platform that has not wired + * [enteredForeground]/[enteredBackground] up, and on such a platform "always on" is the + * behaviour that predates this file — a subscription that never opens is a far worse failure + * than one that stays open too long. + */ +object AppLifecycle { + private val _isForeground = MutableStateFlow(true) + + val isForeground: StateFlow = _isForeground.asStateFlow() + + fun enteredForeground() { + _isForeground.value = true + } + + fun enteredBackground() { + _isForeground.value = false + } +} diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt index a37fd240..04145fc1 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt @@ -19,6 +19,8 @@ import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.debounce import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.map @@ -26,7 +28,9 @@ import kotlinx.coroutines.flow.transformWhile import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.withContext +import press.mantra.compose.database.model.NegentropySynchronizeRequest import press.mantra.compose.database.model.NostrEvent +import press.mantra.compose.database.model.types.SynchronizationFilter import press.mantra.compose.network.relays.RelaysSocketManager import press.mantra.compose.network.relays.isRelayBackPressure import press.mantra.compose.network.sockets.NostrIncomingMessage @@ -62,6 +66,7 @@ class LiveSubscriptionManager( private val relaysSocketManager: RelaysSocketManager, private val nostrRepository: NostrRepository, private val chatRepository: ChatRepository, + private val isForeground: StateFlow, ) { private val logger = Logger.withTag(TAG) @@ -112,6 +117,17 @@ class LiveSubscriptionManager( private val INITIAL_REOPEN_DELAY = 5.seconds private val MAX_REOPEN_DELAY = 5.minutes + + /** + * Matches what `ChatRoomListViewModel` queued, deliberately: the id a negentropy + * request is stored under is a hash of its filter, so an identical filter shape + * collapses into the same row rather than queueing the same reconciliation twice + * while both callers exist. + * + * The value itself barely matters — the negentropy pump drops `limit` outright, and it + * only survives into the plain-REQ fallback. + */ + private const val CATCH_UP_LIMIT = 50 } /** @@ -129,9 +145,33 @@ class LiveSubscriptionManager( * active wallet, so a wallet switch tears every subscription down and the new wallet's * call builds its own. */ - suspend fun observe(keyPair: KeyPair): Unit = coroutineScope { + suspend fun observe(keyPair: KeyPair) { val publicKey = keyPair.pubKey.toHexKey() - logger.i("Opening live subscriptions for $publicKey") + + // collectLatest is the whole lifecycle mechanism: going to the background cancels the + // block below, and every subscription's `finally` sends its CLOSE on the way out. + isForeground.collectLatest { foreground -> + if (!foreground) { + logger.i("Backgrounded; live subscriptions closed") + return@collectLatest + } + + runWhileForeground(publicKey = publicKey, keyPair = keyPair) + } + } + + private suspend fun runWhileForeground(publicKey: HexKey, keyPair: KeyPair): Unit = coroutineScope { + logger.i("Foregrounded; opening live subscriptions for $publicKey") + + // Before anything is asked for. A socket that was open when the OS suspended us + // reports itself connected on the way back while being functionally dead, and the + // relay dropped our subscriptions long ago. + runCatching { relaysSocketManager.reconnectAll() } + .onFailure { logger.w(throwable = it) { "Reconnect on foreground failed" } } + + // Live subscriptions cover the window we are online for; this covers the gap we were + // not. Neither subsumes the other. + launch(Dispatchers.IO) { queueCatchUpSynchronization(publicKey) } Relays.DefaultDMRelayList.forEach { relay -> launch(Dispatchers.IO) { @@ -147,6 +187,62 @@ class LiveSubscriptionManager( launch(Dispatchers.IO) { followGroupMembership(publicKey = publicKey, keyPair = keyPair) } } + /** + * Reconciles what we hold against what the relays hold, for the time we were away. + * + * A live subscription answers "what is new since I connected"; negentropy answers "what do + * you have that I don't". Coming back from the background is exactly the question only the + * second one can answer, and `limit` on the re-opened subscriptions is a window, not a + * guarantee. + */ + private suspend fun queueCatchUpSynchronization(publicKey: HexKey) { + val groupIds = liveGroupIds(publicKey) + + val giftWrapFilter = SynchronizationFilter( + kinds = arrayOf(GiftWrapEvent.KIND), + tags = mapOf("p" to listOf(publicKey)), + limit = CATCH_UP_LIMIT, + ) + + val requests = mutableListOf() + + Relays.DefaultDMRelayList.forEach { relay -> + requests += NegentropySynchronizeRequest( + id = NegentropySynchronizeRequest.computeId( + relayURL = relay.url, + synchronizationFilter = giftWrapFilter, + ), + purpose = "chat", + synchronizationFilter = giftWrapFilter, + relayURL = relay.url, + level = 0, + ) + + if (groupIds.isEmpty()) return@forEach + + val groupSynchronizationFilter = SynchronizationFilter( + kinds = arrayOf(GroupEvent.KIND), + tags = mapOf("h" to groupIds), + ) + requests += NegentropySynchronizeRequest( + id = NegentropySynchronizeRequest.computeId( + relayURL = relay.url, + synchronizationFilter = groupSynchronizationFilter, + ), + purpose = "mlsMessages", + synchronizationFilter = groupSynchronizationFilter, + relayURL = relay.url, + level = 0, + ) + } + + logger.i("Queueing ${requests.size} catch-up reconciliation(s) over ${groupIds.size} group(s)") + nostrRepository.queueNegentropySynchronizeRequest(requests) + } + + private suspend fun liveGroupIds(publicKey: HexKey): List = + chatRepository.getChatRoomListByPublicKey(publicKey).toLiveGroupIds() + /** * Every gift wrap addressed to us: direct messages, and the Marmot Welcome events that * make us a member of a group. @@ -195,18 +291,7 @@ class LiveSubscriptionManager( val subscriptions = mutableMapOf() chatRepository.observeChatRoomListByPublicKey(publicKey) - .map { rooms -> - rooms - // An MLS group is a room with group state; a NIP-17 room has none and is - // served by the gift wrap subscription instead. A room we have left or - // deleted keeps its history locally but must stop pulling new messages. - .filter { it.chatRoom.mlsGroupState != null } - .filter { it.chatRoom.leftGroupAt == null && it.chatRoom.deletedAt == null } - .map { it.chatRoom.id } - .sorted() - } - // Sorted first so that the same membership in a different row order is the same - // value, and a re-emit that changes nothing costs nothing. + .map { rooms -> rooms.toLiveGroupIds() } .distinctUntilChanged() .debounce(GROUP_CHANGE_DEBOUNCE) .collect { groupIds -> @@ -284,6 +369,21 @@ class LiveSubscriptionManager( } } + /** + * An MLS group is a room with group state; a NIP-17 room has none and is served by the + * gift wrap subscription instead. A room we have left or deleted keeps its history + * locally but must stop pulling new messages. + * + * Sorted so the same membership in a different row order is the same value, and a + * re-emit that changes nothing costs nothing downstream. + */ + private fun List.toLiveGroupIds() = + this + .filter { it.chatRoom.mlsGroupState != null } + .filter { it.chatRoom.leftGroupAt == null && it.chatRoom.deletedAt == null } + .map { it.chatRoom.id } + .sorted() + private fun subscriptionKey(relayUrl: String, subId: String) = "$relayUrl|$subId" private fun subscriptionIndex(key: String) = 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 73a08734..d4a1be20 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 @@ -166,6 +166,27 @@ class RelayPool( fun hasRelays() = relays.isNotEmpty() + /** + * Tears every socket down and immediately builds it again. + * + * For coming back from the background, where trusting the connection is the mistake: a + * socket that was open when the OS suspended the process reports itself connected on the + * way back while being functionally dead, and the relay has long since dropped the + * subscriptions it was carrying. + * + * Re-opening here rather than leaving it to the next send is deliberate — it is what makes + * this a RE-connection, so the retained subscriptions are replayed and any collector still + * attached from before the gap starts receiving again. + */ + suspend fun reconnectAll() { + socketClients.forEach { client -> + updateRelayStatus(url = client.socketUrl, connected = false) + runCatching { client.close() } + runCatching { client.ensureSocketConnectionOrThrow() } + .onFailure { logger.w(throwable = it) { "Could not re-open ${client.socketUrl}" } } + } + } + suspend fun tryConnectingToRelay(url: String) { runCatching { socketClients.find { it.socketUrl == url }?.ensureSocketConnectionOrThrow() diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt index 256828de..8b18428f 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt @@ -126,6 +126,9 @@ class RelaysSocketManager( ) } + /** @see RelayPool.reconnectAll */ + suspend fun reconnectAll() = relayPool.reconnectAll() + fun tryConnectingToAllRelays() { relayPool.relays.forEach { scope.launch { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/navigation/MantraNavHost.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/navigation/MantraNavHost.kt index 9f9d612e..1a6cd763 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/navigation/MantraNavHost.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/navigation/MantraNavHost.kt @@ -3,10 +3,14 @@ package press.mantra.compose.ui.composable.navigation import androidx.compose.foundation.layout.fillMaxSize import androidx.compose.material3.Surface import androidx.compose.runtime.Composable +import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.remember import androidx.compose.ui.Modifier import androidx.compose.ui.graphics.Color +import press.mantra.compose.AppLifecycle +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.LifecycleEventObserver import androidx.lifecycle.compose.LocalLifecycleOwner import androidx.lifecycle.viewmodel.compose.viewModel import androidx.navigation.NavHostController @@ -219,6 +223,22 @@ fun MantraNavHost( // TODO: Produce a synchronization UI element... + // The one place in the app that knows whether it is on screen. Everything that holds a + // relay connection open reads AppLifecycle rather than a lifecycle owner, because those + // consumers are application-scoped coroutines that outlive any composition. + DisposableEffect(lifecycleOwner) { + val observer = LifecycleEventObserver { _, event -> + when (event) { + Lifecycle.Event.ON_START -> AppLifecycle.enteredForeground() + Lifecycle.Event.ON_STOP -> AppLifecycle.enteredBackground() + else -> Unit + } + } + + lifecycleOwner.lifecycle.addObserver(observer) + onDispose { lifecycleOwner.lifecycle.removeObserver(observer) } + } + LaunchedEffect(lifecycleOwner) { navigationViewModel.navigationUIState.collect { state -> when (state) { 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 146fbc81..fab0fb71 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 @@ -8,6 +8,7 @@ import press.mantra.compose.database.model.BroadcastNostrEventRequest import press.mantra.compose.database.model.SynchronizeNostrEventRequest import press.mantra.compose.database.model.types.SynchronizationFilter import press.mantra.compose.network.dto.toRelayDTO +import press.mantra.compose.AppLifecycle import press.mantra.compose.managers.LiveSubscriptionManager import press.mantra.compose.network.relays.RelayPool import press.mantra.compose.network.relays.RelaysSocketManager @@ -81,6 +82,7 @@ class SynchronizationViewModel( relaysSocketManager = relaysSocketManager, nostrRepository = nostrRepository, chatRepository = chatRepository, + isForeground = AppLifecycle.isForeground, ) companion object {