From 8da5b803f4b27d7545d7f611a7fd3a59a8a9b584 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Sun, 20 Sep 2026 14:43:57 -0400 Subject: [PATCH] fix(chat): stop a reconnect reopening the event stream in the background Bugsnag 6ab0133d58142669ea781ea5 has the bidirectional event stream reopening 6.5 seconds after `MainActivity#onStop()` and going on to sync the feed, with the app in the background. `TokenCoordinator` cancelled its reserve state stream a second after the same `onStop`. `ChatCoordinator.onStop` already closes the stream and stops the heartbeat, so the reopen comes from a second path. The network-reconnect collector is launched on `scope`, which spans the logged-in session rather than the foreground, and a reconnect landing after `onStop` re-synced the feed and reopened the stream behind the close. It now reads a `foregrounded` flag maintained by `onStart`/`onStop`, checked after the debounce so the answer is the lifecycle state the reconnect would actually act in. A reconnect that lands while backgrounded is dropped rather than deferred, since `onStart` already syncs and reopens. The same breadcrumbs show "Network connected, re-syncing chat feed" firing every few seconds in the foreground, each one skipping an already-connected stream. `distinctUntilChanged` was applied to the whole `NetworkState`, which carries signal strength and connection type alongside `connected`, and a cellular radio reports those every few seconds. Every signal wobble passed through as a fresh reconnect on a connection that never dropped. Mapping to `connected` before de-duplicating collapses them. --- .../chat/internal/RealChatCoordinator.kt | 22 +- .../ChatCoordinatorLifecycleStreamTest.kt | 267 ++++++++++++++++++ 2 files changed, 286 insertions(+), 3 deletions(-) create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorLifecycleStreamTest.kt diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt index b6b1ce97ac..d76f1f2589 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt @@ -41,6 +41,7 @@ import kotlinx.coroutines.flow.filter import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.flatMapLatest import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch import javax.inject.Inject @@ -70,7 +71,8 @@ import kotlin.time.Duration.Companion.seconds * rather than either delegate directly. See [refreshFeed]. * - **Lifecycle methods** — [onStart]/[onStop] are inherently cross-cutting * (stream connect/disconnect, heartbeat start/stop, active-chat save/restore). - * - **Flow observers** — network reconnect re-syncing the chat feed. + * - **Flow observers** — network reconnect re-syncing the chat feed, gated on the + * foreground so a reconnect cannot reopen the stream behind [onStop]. * * Delegates require [initialize] with a shared [CoroutineScope] before use; * this happens in [onUserLoggedIn]. @@ -110,6 +112,10 @@ class RealChatCoordinator @Inject constructor( private var networkObserverJob: Job? = null private var backgroundedActiveChat: ChatId? = null + // Maintained by [onStart]/[onStop]. The network-reconnect collector below lives on [scope], + // which spans the whole session rather than the foreground, so it needs to ask. + private val foregrounded = MutableStateFlow(false) + override val state: StateFlow get() = stateHolder.state @@ -195,11 +201,19 @@ class RealChatCoordinator @Inject constructor( } }.launchIn(scope) + // `map { it.connected }` before `distinctUntilChanged`: [NetworkState] also carries signal + // strength and connection type, which a cellular radio reports every few seconds, so + // de-duplicating the whole state let every signal wobble through as a fresh "reconnect" + // and re-synced the feed on a connection that never dropped. networkObserverJob = cluster.filterNotNull() - .flatMapLatest { networkObserver.state } + .flatMapLatest { networkObserver.state.map { state -> state.connected } } .distinctUntilChanged() - .filter { it.connected } + .filter { it } .debounce(1.seconds) + // Read after the debounce, so the answer is the lifecycle state this reconnect would + // actually act in. A reconnect that lands while backgrounded is dropped rather than + // deferred: [onStart] already syncs and reopens, so nothing is lost by ignoring it. + .filter { foregrounded.value } .onEach { trace(tag = TAG, message = "Network connected, re-syncing chat feed", type = TraceType.Process) syncFeeds() @@ -209,6 +223,7 @@ class RealChatCoordinator @Inject constructor( } override fun onStart(owner: LifecycleOwner) { + foregrounded.value = true backgroundedActiveChat?.let { messagingDelegate.setActiveChatId(it) backgroundedActiveChat = null @@ -224,6 +239,7 @@ class RealChatCoordinator @Inject constructor( } override fun onStop(owner: LifecycleOwner) { + foregrounded.value = false backgroundedActiveChat = stateHolder.current.activeChat messagingDelegate.setActiveChatId(null) eventStreamDelegate.stopHeartbeat() diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorLifecycleStreamTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorLifecycleStreamTest.kt new file mode 100644 index 0000000000..b6effe4c13 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorLifecycleStreamTest.kt @@ -0,0 +1,267 @@ +package com.flipcash.shared.chat + +import com.flipcash.app.core.dispatchers.TestDispatchers +import com.flipcash.app.persistence.sources.ChatMemberDataSource +import com.flipcash.app.persistence.sources.ChatMessageDataSource +import com.flipcash.app.persistence.sources.ChatMetadataDataSource +import com.flipcash.app.persistence.sources.ContactDataSource +import com.flipcash.app.tokens.TokenCoordinator +import com.flipcash.services.controllers.ChatController +import com.flipcash.services.controllers.ChatMessagingController +import com.flipcash.services.controllers.EventStreamingController +import com.flipcash.services.models.chat.ChatUpdate +import com.flipcash.services.user.UserManager +import com.flipcash.shared.chat.internal.ChatIdGenerator +import com.flipcash.shared.chat.internal.ChatStateHolder +import com.flipcash.shared.chat.internal.RealChatCoordinator +import com.flipcash.shared.chat.internal.delegates.DmChatResolverDelegate +import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate +import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate +import com.flipcash.shared.chat.internal.delegates.MessagingDelegate +import com.getcode.utils.network.ConnectionType +import com.getcode.utils.network.NetworkConnectivityListener +import com.getcode.utils.network.NetworkState +import com.getcode.utils.network.SignalStrength +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.emptyFlow +import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.test.TestCoroutineScheduler +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import kotlin.test.assertEquals +import kotlin.time.Duration.Companion.seconds + +/** + * The event stream is a long-lived bidirectional gRPC stream, and background delivery does not + * depend on it — a chat that arrives while the app is away comes in over FCM + * ([com.flipcash.app.notifications.NotificationService]). So the stream belongs to the foreground, + * which is what [RealChatCoordinator.onStop] closing it says. + * + * The network-reconnect collector lives on the session scope rather than the foreground one, and + * it reopened the stream behind that close: a Bugsnag timeline had `onStop` at 17:09:04 and the + * stream back up at 17:09:11, feed sync and all, with the app in the background. + * + * The same collector also fired every few seconds in the foreground, because it de-duplicated the + * whole [NetworkState] — signal strength and connection type included — rather than the one field + * it acts on. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +class ChatCoordinatorLifecycleStreamTest { + + private class FakeNetworkObserver : NetworkConnectivityListener { + val mutableState = MutableStateFlow(NetworkState.Default) + override val state: StateFlow = mutableState.asStateFlow() + override val isConnected: Boolean get() = mutableState.value.connected + override val type: ConnectionType get() = mutableState.value.type + } + + private lateinit var eventStreamingController: EventStreamingController + private var opens = 0 + private lateinit var networkObserver: FakeNetworkObserver + private lateinit var coordinator: RealChatCoordinator + private lateinit var testDispatchers: TestDispatchers + + @Before + fun setUp() { + val userManager = mockk(relaxed = true) + every { userManager.accountId } returns listOf(1, 2, 3) + + eventStreamingController = mockk(relaxed = true) + every { eventStreamingController.chatUpdates } returns + Channel(Channel.UNLIMITED).receiveAsFlow() + // Never "already connected", so every reopen attempt reaches the controller and can be + // counted. isStreamActive keeps the heartbeat a no-op supervisor for the whole test. + every { eventStreamingController.isConnected } returns false + opens = 0 + every { eventStreamingController.open(any()) } answers { opens++; true } + every { eventStreamingController.isStreamActive } returns true + every { eventStreamingController.streamFailures } returns emptyFlow() + + val chatController = mockk(relaxed = true) + coEvery { chatController.getDmChatFeed(any(), any()) } returns + Result.failure(RuntimeException("not needed")) + + val metadataDataSource = mockk(relaxed = true) + val messageDataSource = mockk(relaxed = true) + val memberDataSource = mockk(relaxed = true) + val messagingController = mockk(relaxed = true) + val stateHolder = ChatStateHolder() + + networkObserver = FakeNetworkObserver() + testDispatchers = TestDispatchers(TestCoroutineScheduler()) + + coordinator = RealChatCoordinator( + feedDelegate = FeedSyncDelegate( + chatController = chatController, + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + stateHolder = stateHolder, + userManager = userManager, + ), + eventStreamDelegate = EventStreamDelegate( + eventStreamingController = eventStreamingController, + messagingController = messagingController, + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + tokenCoordinator = mockk(relaxed = true), + userManager = userManager, + stateHolder = stateHolder, + analytics = mockk(relaxed = true), + exchange = mockk(relaxed = true), + ), + dmChatResolverDelegate = DmChatResolverDelegate( + chatIdGenerator = ChatIdGenerator(), + userManager = userManager, + contactDataSource = mockk(relaxed = true), + memberDataSource = memberDataSource, + ), + messagingDelegate = MessagingDelegate( + chatController = chatController, + messagingController = messagingController, + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + notificationManager = mockk(relaxed = true), + userManager = userManager, + stateHolder = stateHolder, + analytics = mockk(relaxed = true), + senderResolver = mockk(relaxed = true), + ), + stateHolder = stateHolder, + draftStore = mockk(relaxed = true), + userManager = userManager, + networkObserver = networkObserver, + dispatchers = testDispatchers, + groupFeedDelegate = mockk(relaxed = true), + ) + } + + // Every test tears the coordinator down BEFORE it asserts, and steps with + // runCurrent()/advanceTimeBy rather than advanceUntilIdle. The heartbeat started in + // onStart is an unbounded timed loop: "until idle" never arrives, and an assertion that + // threw before teardown would leave runTest's own drain spinning on virtual time rather + // than reporting the failure. + private fun connected(strength: SignalStrength, type: ConnectionType = ConnectionType.Wifi) { + networkObserver.mutableState.value = NetworkState(true, strength, type) + } + + private fun disconnected() { + networkObserver.mutableState.value = NetworkState.Default + } + + @Test + fun `signal strength changes while connected do not reopen the stream`() = + runTest(testDispatchers.dispatcher) { + coordinator.onUserLoggedIn(mockk(relaxed = true)) + coordinator.onStart(mockk(relaxed = true)) + runCurrent() + + connected(SignalStrength.Good) + advanceTimeBy(2.seconds) + runCurrent() + + // What a cellular radio reports every few seconds on a connection that never dropped. + // Spaced out rather than set back to back: the observer's state is a conflated + // StateFlow, so wobbles applied in the same instant collapse into one before the + // collector ever sees them, and the churn under test never happens. + listOf( + SignalStrength.Great to ConnectionType.Wifi, + SignalStrength.Strong to ConnectionType.Cellular, + SignalStrength.Poor to ConnectionType.Cellular, + SignalStrength.Good to ConnectionType.Wifi, + ).forEach { (strength, type) -> + connected(strength, type) + advanceTimeBy(3.seconds) + runCurrent() + } + + coordinator.teardown() + + // Login, foreground, and the one real connect. + assertEquals(3, opens) + } + + @Test + fun `a reconnect in the background does not reopen the stream`() = + runTest(testDispatchers.dispatcher) { + coordinator.onUserLoggedIn(mockk(relaxed = true)) + coordinator.onStart(mockk(relaxed = true)) + connected(SignalStrength.Good) + advanceTimeBy(2.seconds) + runCurrent() + + coordinator.onStop(mockk(relaxed = true)) + runCurrent() + + disconnected() + advanceTimeBy(2.seconds) + connected(SignalStrength.Good) + advanceTimeBy(10.seconds) + runCurrent() + + coordinator.teardown() + + // Still only login, foreground, and the connect from before backgrounding. + assertEquals(3, opens) + } + + @Test + fun `a reconnect in the foreground reopens the stream`() = + runTest(testDispatchers.dispatcher) { + coordinator.onUserLoggedIn(mockk(relaxed = true)) + coordinator.onStart(mockk(relaxed = true)) + connected(SignalStrength.Good) + advanceTimeBy(2.seconds) + runCurrent() + + disconnected() + advanceTimeBy(2.seconds) + connected(SignalStrength.Good) + advanceTimeBy(2.seconds) + runCurrent() + + coordinator.teardown() + + assertEquals(4, opens) + } + + /** Backgrounding and then returning still reopens — the gate is the lifecycle, not a latch. */ + @Test + fun `returning to the foreground reopens the stream`() = + runTest(testDispatchers.dispatcher) { + coordinator.onUserLoggedIn(mockk(relaxed = true)) + coordinator.onStart(mockk(relaxed = true)) + connected(SignalStrength.Good) + advanceTimeBy(2.seconds) + runCurrent() + + coordinator.onStop(mockk(relaxed = true)) + disconnected() + advanceTimeBy(2.seconds) + connected(SignalStrength.Good) + advanceTimeBy(10.seconds) + runCurrent() + + coordinator.onStart(mockk(relaxed = true)) + runCurrent() + + coordinator.teardown() + + assertEquals(4, opens) + } +}