From cd91b398ffa21aa47e4729eb2080af09587884ea Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Sun, 20 Sep 2026 19:47:35 -0400 Subject: [PATCH] fix(chat): recover a message left sending by an app kill MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kill the app while a chat message is sending and the row is stranded. A send is marked failed only from the `onFailure` of the coroutine that issued it, so a dead process leaves `status = 'SENDING'` on disk with nothing left to move it. The bubble spins — retry is offered for `FAILED`, not `SENDING` — until the next refresh reaches `upsertAndClearPending` and deletes the row, taking the user's text with it and offering no way back. `deleteAllPending` is scoped to `SENDING` and spares `FAILED`, so flipping the orphans once at startup is enough to make them survive every later refresh, at which point they pick up the retry affordance that already exists. The retry re-sends under the stored `pending_client_id_hex`, the same `client_message_id` the server dedupes on, so it replaces the original rather than duplicating it. The sweep is awaited in `onUserLoggedIn` rather than launched from an `initialize(scope)` hook like the sibling delegates'. Those hooks are field assignments, and the refresh is itself a `launch`; a launched sweep would race the very refresh it has to precede, and a sweep that lands second finds nothing to do. Covered by DAO tests for the three states that matter — swept, then surviving the refresh that would have deleted it, and a sent message left alone — and by a coordinator test pinning the order. The ordering test was checked in both directions: moving the call after `syncFeeds()` fails it. --- .../chat/internal/RealChatCoordinator.kt | 5 + .../internal/delegates/MessagingDelegate.kt | 21 ++ .../chat/InterruptedSendRecoveryTest.kt | 194 ++++++++++++++++++ .../app/persistence/dao/ChatMessageDao.kt | 20 ++ .../app/persistence/dao/ChatMessageDaoTest.kt | 41 ++++ .../sources/ChatMessageDataSource.kt | 8 + 6 files changed, 289 insertions(+) create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/InterruptedSendRecoveryTest.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..5560f3783a 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 @@ -129,6 +129,11 @@ class RealChatCoordinator @Inject constructor( feedDelegate.initialize(scope) eventStreamDelegate.initialize(scope) groupFeedDelegate.initialize(scope) + // Before the first refresh, and awaited rather than launched: a refresh that carries a + // self-authored message deletes this chat's `SENDING` rows, so a send a previous process + // was killed mid-flight has to be marked failed first or it is destroyed instead of + // becoming retryable. See [MessagingDelegate.recoverInterruptedSends]. + messagingDelegate.recoverInterruptedSends() feedDelegate.observeFeedFromDb() syncFeeds() eventStreamDelegate.open() diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt index 7aea5900b4..3ac3f10b7c 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt @@ -500,6 +500,27 @@ class MessagingDelegate @Inject constructor( } } + /** + * Fails the sends a previous process left in flight, so the transcript offers to retry them + * instead of spinning on a send that nothing is still attempting. + * + * A send is marked failed from the `onFailure` of the coroutine that issued it ([sendMessage], + * [retryMessage]). Kill the app mid-send and that coroutine dies with the process, leaving a + * `SENDING` row on disk that nothing will ever move — until the first refresh deletes it and the + * user's text with it. + * + * Must complete before the first refresh of this session. The refresh path + * ([com.flipcash.app.persistence.sources.ChatMessageDataSource.upsert] on a self-authored + * message) deletes `SENDING` rows and spares `FAILED` ones, so sweeping first makes them + * durable and sweeping second does nothing at all. That is why this is `suspend` and awaited by + * [com.flipcash.shared.chat.internal.RealChatCoordinator.onUserLoggedIn] rather than launched + * from an `initialize(scope)` hook like the other delegates' — the refresh is itself a `launch`, + * and two coroutines would leave the order to chance. + */ + internal suspend fun recoverInterruptedSends() { + messageDataSource.failInterruptedSends() + } + internal suspend fun clear() { pendingMutations.value = emptyMap() metadataDataSource.clear() diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/InterruptedSendRecoveryTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/InterruptedSendRecoveryTest.kt new file mode 100644 index 0000000000..fc369a3483 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/InterruptedSendRecoveryTest.kt @@ -0,0 +1,194 @@ +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.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.NetworkConnectivityListener +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.test.TestCoroutineScheduler +import kotlinx.coroutines.test.TestScope +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.assertTrue +import kotlin.time.Duration.Companion.milliseconds + +/** + * A send the app was killed in the middle of has to be marked failed before the session's first + * refresh, or the refresh deletes it. + * + * Nothing in-process survives the kill: a send is marked failed from the `onFailure` of the + * coroutine that issued it, so the row is left `SENDING` on disk. The first refresh that carries a + * self-authored message runs `deleteAllPending`, which is scoped to `SENDING` — so the sweep either + * runs first and makes the row durable, or runs second and finds nothing. There is no third + * outcome, which is why the order is worth a test rather than a comment. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +class InterruptedSendRecoveryTest { + + private val chatUpdatesChannel = Channel(capacity = Channel.UNLIMITED) + + private lateinit var coordinator: RealChatCoordinator + private lateinit var messageDataSource: ChatMessageDataSource + private lateinit var testDispatchers: TestDispatchers + + /** Set when the sweep returns; read by the feed fetch to report what it raced. */ + private var sweepFinished = false + private var feedFetchSawFinishedSweep: Boolean? = null + + @Before + fun setUp() { + val userManager = mockk(relaxed = true) + every { userManager.accountId } returns listOf(1, 2, 3) + + val eventStreamingController = mockk(relaxed = true) + every { eventStreamingController.chatUpdates } returns chatUpdatesChannel.receiveAsFlow() + every { eventStreamingController.isConnected } returns true + every { eventStreamingController.isStreamActive } returns true + + val chatController = mockk(relaxed = true) + coEvery { chatController.getDmChatFeed(any(), any()) } coAnswers { + feedFetchSawFinishedSweep = sweepFinished + Result.failure(RuntimeException("feed contents are not what this test is about")) + } + + testDispatchers = TestDispatchers(TestCoroutineScheduler()) + + val stateHolder = ChatStateHolder() + val memberDataSource = mockk(relaxed = true) + val messagingController = mockk(relaxed = true) + val metadataDataSource = mockk(relaxed = true) + messageDataSource = mockk(relaxed = true) + // The sweep is a Room write, so it is modelled with a suspension point. Without one the + // test cannot tell the fix from the bug: the refresh is a `launch`, so on a single-threaded + // test dispatcher it could not start before a sweep that never suspends had already + // returned, and a sweep placed after it would still appear to have gone first. + coEvery { messageDataSource.failInterruptedSends() } coAnswers { + delay(1) + sweepFinished = true + } + + val feedDelegate = FeedSyncDelegate( + chatController = chatController, + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + stateHolder = stateHolder, + userManager = userManager, + ) + + val 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), + ) + + val 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), + ) + + val dmChatResolverDelegate = DmChatResolverDelegate( + chatIdGenerator = ChatIdGenerator(), + userManager = userManager, + contactDataSource = mockk(relaxed = true), + memberDataSource = memberDataSource, + ) + + coordinator = RealChatCoordinator( + feedDelegate = feedDelegate, + eventStreamDelegate = eventStreamDelegate, + dmChatResolverDelegate = dmChatResolverDelegate, + messagingDelegate = messagingDelegate, + stateHolder = stateHolder, + draftStore = mockk(relaxed = true), + userManager = userManager, + networkObserver = mockk(relaxed = true), + dispatchers = testDispatchers, + groupFeedDelegate = mockk(relaxed = true), + ) + } + + /** + * Signs in, lets the login work settle, and tears the coordinator down however [block] ends. + * + * The `finally` is load-bearing, for the reason the other coordinator tests give: logging in + * starts a heartbeat that is a `while (true)` of delays on the coordinator's own scope, which + * is not [TestScope.backgroundScope] and so outlives the body. [runTest] drains the scheduler + * once the body returns, so a failed assertion that skipped the teardown would leave that loop + * spinning virtual time and bury the failure in a run that never ends. + * + * The advance is bounded for the same reason — `advanceUntilIdle` would never come back. + */ + private suspend fun TestScope.loggedIn(block: () -> Unit) { + try { + coordinator.onUserLoggedIn(mockk(relaxed = true)) + advanceTimeBy(10.milliseconds) + runCurrent() + block() + } finally { + coordinator.teardown() + } + } + + @Test + fun `login sweeps interrupted sends before it fetches the feed`() = + runTest(testDispatchers.dispatcher) { + loggedIn { + assertTrue( + feedFetchSawFinishedSweep == true, + "The feed fetch ran while the sweep was still in flight, so a refresh " + + "carrying a self-authored message could delete a SENDING row before it " + + "had been marked failed.", + ) + } + } + + @Test + fun `the sweep runs once per login`() = runTest(testDispatchers.dispatcher) { + loggedIn { + coVerify(exactly = 1) { messageDataSource.failInterruptedSends() } + } + } +} diff --git a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt index f8151989a0..049ab513c6 100644 --- a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt +++ b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt @@ -123,6 +123,26 @@ interface ChatMessageDao { @Query("DELETE FROM chat_messages WHERE chat_id_hex = :chatIdHex AND status = 'SENDING'") suspend fun deleteAllPending(chatIdHex: String) + /** + * Fails every row still marked `SENDING`, in every chat. + * + * A send is only ever marked failed from the `onFailure` of the coroutine that issued it, so a + * process killed mid-send leaves its row `SENDING` on disk with nothing left alive to move it. + * The transcript shows that row spinning — retry is offered for `FAILED`, not `SENDING` — until + * the next refresh reaches [upsertAndClearPending] and deletes it, taking the user's text with + * it and offering no way to get it back. + * + * Run once per process before the first refresh. [deleteAllPending] is scoped to `SENDING` and + * spares `FAILED`, so a row swept here survives every later refresh and picks up the retry + * affordance that already exists. `pending_client_id_hex` is untouched, and it is what the retry + * re-sends under — the same `client_message_id` the server dedupes on, so the re-send replaces + * the original rather than duplicating it. + * + * Deliberately not scoped to a chat: the caller runs at login, before it knows which chats exist. + */ + @Query("UPDATE chat_messages SET status = 'FAILED' WHERE status = 'SENDING'") + suspend fun failInterruptedSends() + /** * The optimistic row is written before the server has stamped the message, so every * server-assigned field is written here — `event_sequence` included. Leaving it at the pending diff --git a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt index 1ce9cccfeb..73dbe19eb9 100644 --- a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt +++ b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt @@ -183,6 +183,47 @@ class ChatMessageDaoTest { assertEquals(2, dao.countNewerThan(CHAT_HEX, 3 * 1_000)) } + /** + * Killing the app mid-send leaves the row `SENDING` with nothing alive to move it: failure is + * only ever written from the `onFailure` of the coroutine that issued the send. The sweep is + * what a fresh process has instead of that coroutine. + */ + @Test + fun `the sweep fails a send a dead process left in flight`() = runTest { + dao.upsert(pending("half-sent")) + + dao.failInterruptedSends() + + val stored = dao.getByClientId(CHAT_HEX, CLIENT_HEX)!! + assertEquals(MessageStatus.FAILED, stored.status) + // The retry re-sends under this id, so losing it would lose the affordance with it. + assertEquals(CLIENT_HEX, stored.pendingClientIdHex) + } + + /** + * The point of failing the row rather than leaving it pending. `deleteAllPending` — which every + * refresh carrying a self-authored message runs — is scoped to `SENDING`, so a swept row is + * durable where a pending one is deleted along with the text the user typed. + */ + @Test + fun `a swept row survives the refresh that would have deleted it`() = runTest { + dao.upsert(pending("half-sent")) + dao.failInterruptedSends() + + dao.upsertAndClearPending(CHAT_HEX, listOf(text(1, "from the server"))) + + assertEquals(MessageStatus.FAILED, dao.getByClientId(CHAT_HEX, CLIENT_HEX)?.status) + } + + @Test + fun `the sweep leaves a sent message alone`() = runTest { + dao.upsert(text(1, "already sent")) + + dao.failInterruptedSends() + + assertEquals(MessageStatus.SENT, dao.getMessage(CHAT_HEX, 1)?.status) + } + private companion object { const val CHAT_HEX = "aabb" const val OTHER_HEX = "ccdd" diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt index 4c54d20ad6..b5b4de0853 100644 --- a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt @@ -2,6 +2,7 @@ package com.flipcash.app.persistence.sources import androidx.paging.PagingSource import com.flipcash.app.persistence.FlipcashDatabase +import com.flipcash.app.persistence.dao.ChatMessageDao import com.flipcash.app.persistence.entities.ChatMessageEntity import com.flipcash.app.persistence.sources.mapper.chat.ChatEntityMapper import com.flipcash.services.models.chat.ChatId @@ -216,6 +217,13 @@ class ChatMessageDataSource @Inject constructor( ) } + /** + * Fails the sends a previous process left in flight. See [ChatMessageDao.failInterruptedSends]. + */ + suspend fun failInterruptedSends() { + db?.chatMessageDao()?.failInterruptedSends() + } + suspend fun retryPending(chatId: ChatId, pendingClientIdHex: String): ClientMessageId { val clientMessageId = mapper.clientMessageIdFromHex(pendingClientIdHex) db?.chatMessageDao()?.updatePendingStatus(