From 62249fc5d52abf30e07349cf9da99d8e6026378e Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 28 Aug 2026 16:09:47 -0400 Subject: [PATCH 1/5] fix(chat): report stream liveness honestly and stop rewinding read pointers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects behind the reported late messages and the unread badge that survives opening a chat. `openBidirectionalStream` retries internally, but `isStreamActive` is only ever set by `activateStream()` on a first response and cleared by an explicit teardown, so it stayed `true` for the whole retry window. Every external recovery trigger reads that flag or the reference's mere existence to decide whether to step in: the 30s heartbeat in `EventStreamDelegate` checks `isStreamActive`, and `open()` from lifecycle resume and network reconnect checks `isConnected`. With a stale `true`, all of them no-op while nothing is delivering, so recovery waited on the loop's own backoff — up to eight attempts and ~91s of sleeps before it gives up and the next heartbeat tick can reopen. The loop now marks the reference down at the top of each attempt, which is what the heartbeat needs to see. `ChatMemberDao.upsert` was a whole-row REPLACE, so a feed sync overwrote `pointers_json` with the server's copy. The READ pointer is advanced locally as messages come into view and reported afterwards, so any sync racing that report — or following a failed one — rewound the pointer and the chat re-reported as unread. Its sibling `ChatMetadataDao.upsert` already avoids this for `latest_event_sequence` and `analytics_counted_through`; `pointers_json` is the same kind of client-owned watermark and now merges by taking the greater value per pointer. The `FullRefresh` branch of the event stream wiped the member rows before re-inserting them, which defeated any merge, so it prunes departed members instead of clearing the table. --- .../internal/delegates/EventStreamDelegate.kt | 6 +- .../app/persistence/dao/ChatMemberDao.kt | 62 +++++++++- .../app/persistence/dao/ChatMemberDaoTest.kt | 106 ++++++++++++++++++ .../sources/ChatMemberDataSource.kt | 24 ++++ .../bidi/BidirectionalStreamReference.kt | 13 +++ .../opencode/internal/bidi/OpenStream.kt | 6 + .../bidi/OpenBidirectionalStreamTest.kt | 51 +++++++++ 7 files changed, 263 insertions(+), 5 deletions(-) create mode 100644 apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt index 13109f4812..88ebd20a25 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt @@ -301,8 +301,10 @@ class EventStreamDelegate @Inject constructor( when (metaUpdate) { is MetadataUpdate.FullRefresh -> { metadataDataSource.upsert(metaUpdate.metadata) - memberDataSource.deleteForChat(metaUpdate.metadata.chatId) - memberDataSource.upsert(metaUpdate.metadata.chatId, metaUpdate.metadata.members) + memberDataSource.replaceMembers( + metaUpdate.metadata.chatId, + metaUpdate.metadata.members, + ) metaUpdate.metadata.lastMessage?.let { msg -> messageDataSource.upsert(metaUpdate.metadata.chatId, listOf(msg)) } diff --git a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt index 7f90333dd7..0e1dbda913 100644 --- a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt +++ b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt @@ -5,6 +5,7 @@ import androidx.room.Insert import androidx.room.OnConflictStrategy import androidx.room.Query import androidx.room.Transaction +import com.flipcash.app.persistence.converters.MessagePointerSerialized import com.flipcash.app.persistence.entities.ChatMemberEntity import com.flipcash.app.persistence.entities.ChatMemberWithProfile import kotlinx.coroutines.flow.Flow @@ -25,10 +26,32 @@ interface ChatMemberDao { fun observeAll(): Flow> @Insert(onConflict = OnConflictStrategy.REPLACE) - suspend fun upsert(entity: ChatMemberEntity) + suspend fun insertOrReplace(entity: ChatMemberEntity) - @Insert(onConflict = OnConflictStrategy.REPLACE) - suspend fun upsert(entities: List) + /** + * Writes server truth for a member, keeping whichever copy of each pointer is further ahead. + * + * Deliberately not a whole-row REPLACE: `pointers_json` carries the READ pointer, which the + * client advances locally the instant a message is seen and only reports afterwards. A feed + * payload is the server's view as of the fetch, so replacing the row rewinds every advance + * the server has not acknowledged yet — and the chat goes back to reporting unread. + */ + @Transaction + suspend fun upsert(entity: ChatMemberEntity) { + val existing = getMember(entity.chatIdHex, entity.userIdHex) + ?: return insertOrReplace(entity) + + insertOrReplace( + entity.copy( + pointersJson = mergePointers(existing.pointersJson, entity.pointersJson), + ) + ) + } + + @Transaction + suspend fun upsert(entities: List) { + for (entity in entities) upsert(entity) + } @Query("SELECT * FROM chat_members WHERE chat_id_hex = :chatIdHex AND user_id_hex = :userIdHex LIMIT 1") suspend fun getMember(chatIdHex: String, userIdHex: String): ChatMemberEntity? @@ -55,6 +78,39 @@ interface ChatMemberDao { @Query("DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex") suspend fun deleteForChat(chatIdHex: String) + /** Drops the members of [chatIdHex] that are no longer in [keepUserIdHexes]. */ + @Query( + "DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex " + + "AND user_id_hex NOT IN (:keepUserIdHexes)" + ) + suspend fun deleteMembersNotIn(chatIdHex: String, keepUserIdHexes: List) + @Query("DELETE FROM chat_members") suspend fun deleteAll() } + +/** + * The pointers a member row should end up holding. + * + * A pointer only ever moves forward on either side — the client bumps its own READ pointer as + * messages are seen, and the server only ever raises the copy it hands back — so taking the + * greater value per (type, member) lets a refresh carry a pointer forward without rewinding a + * local advance it has not been told about yet. + */ +private fun mergePointers( + existing: List?, + incoming: List?, +): List? { + if (existing.isNullOrEmpty()) return incoming + if (incoming.isNullOrEmpty()) return existing + + val merged = existing.associateByTo(LinkedHashMap()) { it.type to it.userIdHex } + for (pointer in incoming) { + val key = pointer.type to pointer.userIdHex + val current = merged[key] + if (current == null || pointer.value > current.value) { + merged[key] = pointer + } + } + return merged.values.toList() +} diff --git a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt new file mode 100644 index 0000000000..e10f42cf81 --- /dev/null +++ b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt @@ -0,0 +1,106 @@ +package com.flipcash.app.persistence.dao + +import android.content.Context +import androidx.room.Room +import androidx.test.core.app.ApplicationProvider +import com.flipcash.app.persistence.FlipcashDatabase +import com.flipcash.app.persistence.converters.MessagePointerSerialized +import com.flipcash.app.persistence.entities.ChatMemberEntity +import kotlinx.coroutines.test.runTest +import org.junit.After +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import kotlin.test.assertEquals +import kotlin.test.assertNull + +/** + * Covers what a feed sync is allowed to overwrite on a member row. `pointers_json` holds the + * READ pointer, which the client advances locally the moment a message is seen and only then + * reports to the server. A feed payload is server truth as of the fetch, so a whole-row replace + * rewinds an advance the server has not acknowledged yet and the chat re-reports as unread. + */ +@RunWith(RobolectricTestRunner::class) +class ChatMemberDaoTest { + + private lateinit var db: FlipcashDatabase + private lateinit var dao: ChatMemberDao + + @Before + fun setUp() { + val context = ApplicationProvider.getApplicationContext() + db = Room.inMemoryDatabaseBuilder(context, FlipcashDatabase::class.java) + .allowMainThreadQueries() + .build() + dao = db.chatMemberDao() + } + + @After + fun tearDown() { + db.close() + } + + private fun readPointer(value: Long, userIdHex: String = SELF_HEX) = + MessagePointerSerialized(type = "READ", userIdHex = userIdHex, value = value) + + private fun member(vararg pointers: MessagePointerSerialized) = ChatMemberEntity( + chatIdHex = CHAT_HEX, + userIdHex = SELF_HEX, + pointersJson = pointers.toList(), + ) + + @Test + fun `upsert does not rewind a locally advanced read pointer`() = runTest { + dao.upsert(member(readPointer(1))) + dao.updatePointers( + CHAT_HEX, + SELF_HEX, + """[{"type":"READ","userIdHex":"$SELF_HEX","value":9,"timestampEpochSeconds":0}]""", + ) + + // A feed sync issued before the advance reached the server carries the old pointer. + dao.upsert(member(readPointer(1))) + + assertEquals(9L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + + @Test + fun `upsert applies a server pointer that is ahead of the local one`() = runTest { + dao.upsert(member(readPointer(1))) + + // Read on another device: the server is ahead and must win. + dao.upsert(member(readPointer(12))) + + assertEquals(12L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + + @Test + fun `upsert keeps a local pointer while applying the other member's`() = runTest { + dao.upsert(member(readPointer(9), readPointer(3, OTHER_HEX))) + + // The server has seen the peer read further, but not yet our own advance to 9. + dao.upsert(member(readPointer(1), readPointer(7, OTHER_HEX))) + + val pointers = dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson.orEmpty() + assertEquals(9L, pointers.single { it.userIdHex == SELF_HEX }.value) + assertEquals(7L, pointers.single { it.userIdHex == OTHER_HEX }.value) + } + + @Test + fun `deleteMembersNotIn drops departed members and leaves the rest intact`() = runTest { + dao.upsert(member(readPointer(9))) + dao.upsert(member(readPointer(4, OTHER_HEX)).copy(userIdHex = OTHER_HEX)) + + dao.deleteMembersNotIn(CHAT_HEX, listOf(SELF_HEX)) + + assertNull(dao.getMember(CHAT_HEX, OTHER_HEX)) + assertEquals(9L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + + private companion object { + const val CHAT_HEX = "aabb" + const val SELF_HEX = "ccdd" + const val OTHER_HEX = "eeff" + } +} diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt index deb9b9470f..85358a3cbf 100644 --- a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt @@ -84,6 +84,30 @@ class ChatMemberDataSource @Inject constructor( dao.updatePointers(chatIdHex, userIdHex, mapper.pointersToJson(merged)) } + /** + * Replaces a chat's membership with [members], keeping the rows that survive instead of + * clearing the table first. A wipe takes each member's pointers with it, and the refresh + * that follows can only restore what the server knew at fetch time — so a read the client + * has advanced locally but not yet reported would be lost. + */ + suspend fun replaceMembers(chatId: ChatId, members: List) { + if (members.isEmpty()) { + deleteForChat(chatId) + return + } + + val database = db ?: return + val hex = mapper.chatIdHex(chatId) + database.withTransaction { + database.userProfileDao().upsertFull(members.map { mapper.toProfileEntity(it) }) + database.chatMemberDao().upsert(members.map { mapper.toEntity(hex, it) }) + database.chatMemberDao().deleteMembersNotIn( + chatIdHex = hex, + keepUserIdHexes = members.map { mapper.userIdHex(it.userId) }, + ) + } + } + suspend fun deleteForChat(chatId: ChatId) { db?.chatMemberDao()?.deleteForChat(mapper.chatIdHex(chatId)) } diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/BidirectionalStreamReference.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/BidirectionalStreamReference.kt index 309cbdf74d..3a22c086b4 100644 --- a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/BidirectionalStreamReference.kt +++ b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/BidirectionalStreamReference.kt @@ -176,4 +176,17 @@ class BidirectionalStreamReference( fun activateStream() { isStreamActive = true } + + /** + * Reports the stream down without tearing the reference down. + * + * [cancel] and [destroy] both cancel [coroutineScope], which on this reference is where + * the reconnect loop itself runs, so neither can express "no live stream right now, still + * retrying". Callers poll [isActive] to decide whether to step in and reopen; without this, + * liveness stays stuck at whatever the last successful activation set and they sit out the + * whole retry window. + */ + fun deactivateStream() { + isStreamActive = false + } } \ No newline at end of file diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt index fcd2beca4b..7db44b65e7 100644 --- a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt +++ b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt @@ -65,6 +65,12 @@ fun openBidirectionalStream( var attempt = 0 while (attempt++ <= maxReconnectAttempts) { + // Nothing is connected until this attempt's first response activates it. The flag + // has to say so for the whole gap — backoff included — because the callers that + // poll it (the chat heartbeat, lifecycle resume) use it to decide whether the + // stream needs reopening, and a stale `true` makes every one of them a no-op. + streamRef.deactivateStream() + val backoffMs = computeBackoffMs(attempt, reconnectDelayMs, maxReconnectDelayMs) if (backoffMs > 0) { delay(backoffMs) diff --git a/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt b/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt index 592ceab1b3..572c13a54d 100644 --- a/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt +++ b/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt @@ -4,6 +4,7 @@ package com.getcode.opencode.internal.bidi import io.grpc.Status import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.flow @@ -11,6 +12,7 @@ import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.runTest import org.junit.Test import kotlin.test.assertEquals +import kotlin.test.assertFalse import kotlin.test.assertTrue import kotlin.time.Duration.Companion.seconds import kotlin.time.TestTimeSource @@ -257,4 +259,53 @@ class OpenBidirectionalStreamTest { streamRef.destroy() } + + /** + * Liveness must describe the stream that exists right now, not the last one that ever + * connected. The loop retries internally, so between a failure and the next activation + * there is no stream — and every external recovery trigger (the chat heartbeat, lifecycle + * resume, network reconnect) reads this flag to decide whether to step in. + * + * Here the stream activates, fails with UNAVAILABLE, and the retry never emits. The + * reference must report inactive while the loop is parked waiting on that attempt. + */ + @Test + fun `stream reports inactive while the reconnect loop is between attempts`() = runTest { + var attemptCount = 0 + + val streamRef = BidirectionalStreamReference(this, "test-stream") + streamRef.retain() + + openBidirectionalStream>( + streamRef = streamRef, + apiCall = { requestFlow -> + attemptCount++ + flow { + requestFlow.first() + if (attemptCount == 1) { + emit("activation") + throw Status.UNAVAILABLE.asRuntimeException() + } + // The reconnect never gets a response — the stream is down. + awaitCancellation() + } + }, + initialRequest = { "req" }, + responseHandler = { _: String, _: (String) -> Unit -> }, + onError = { }, + reconnectOnUnavailable = true, + maxReconnectAttempts = 3, + reconnectDelayMs = 0, + ) + + advanceUntilIdle() + + assertEquals(2, attemptCount, "Should have reconnected once after UNAVAILABLE") + assertFalse( + streamRef.isActive, + "Reference reports a live stream while the loop is still waiting to reconnect" + ) + + streamRef.destroy() + } } From a37210e2c6c42c79bcbfa914aec65facab781f05 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 28 Aug 2026 16:17:53 -0400 Subject: [PATCH 2/5] fix(chat): create the member row when recording a read pointer `ChatMemberDao.updatePointers` was a bare `UPDATE ... WHERE chat_id_hex = ? AND user_id_hex = ?`, which matches nothing when the member row is absent, so the pointer was dropped without a trace. Both callers can reach it in that state: a read is written the moment a message is on screen, which can beat the feed sync that writes the membership, and the event stream applies pointer updates for members the feed has not written yet. It is now `advancePointer`, which inserts the row when it is missing and otherwise carries the member's other pointers over. The read-merge-write moved into the DAO with it, so a feed sync landing between the read and the write can no longer be lost. --- .../app/persistence/dao/ChatMemberDao.kt | 27 +++++++++++++++++-- .../app/persistence/dao/ChatMemberDaoTest.kt | 25 +++++++++++++---- .../sources/ChatMemberDataSource.kt | 22 +++++++-------- 3 files changed, 56 insertions(+), 18 deletions(-) diff --git a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt index 0e1dbda913..dbe065eac6 100644 --- a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt +++ b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt @@ -72,8 +72,31 @@ interface ChatMemberDao { ) suspend fun getChatIdForMember(userIdHex: String, chatType: String): String? - @Query("UPDATE chat_members SET pointers_json = :pointersJson WHERE chat_id_hex = :chatIdHex AND user_id_hex = :userIdHex") - suspend fun updatePointers(chatIdHex: String, userIdHex: String, pointersJson: String) + /** + * Records [pointer] for [userIdHex] in [chatIdHex], creating the member row if it is not + * there yet. + * + * The row is not a given at this point. A read is written the moment a message is on screen, + * and a pointer update can arrive off the event stream for a member the feed has not written + * yet — the bare `UPDATE` this replaces matched nothing in either case and dropped the + * pointer silently. Only the member's pointer of the same type is displaced; the rest carry + * over. + */ + @Transaction + suspend fun advancePointer( + chatIdHex: String, + userIdHex: String, + pointer: MessagePointerSerialized, + ) { + val existing = getMember(chatIdHex, userIdHex)?.pointersJson.orEmpty() + insertOrReplace( + ChatMemberEntity( + chatIdHex = chatIdHex, + userIdHex = userIdHex, + pointersJson = existing.filterNot { it.type == pointer.type } + pointer, + ) + ) + } @Query("DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex") suspend fun deleteForChat(chatIdHex: String) diff --git a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt index e10f42cf81..f457efb0cb 100644 --- a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt +++ b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt @@ -53,11 +53,7 @@ class ChatMemberDaoTest { @Test fun `upsert does not rewind a locally advanced read pointer`() = runTest { dao.upsert(member(readPointer(1))) - dao.updatePointers( - CHAT_HEX, - SELF_HEX, - """[{"type":"READ","userIdHex":"$SELF_HEX","value":9,"timestampEpochSeconds":0}]""", - ) + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(9)) // A feed sync issued before the advance reached the server carries the old pointer. dao.upsert(member(readPointer(1))) @@ -87,6 +83,25 @@ class ChatMemberDaoTest { assertEquals(7L, pointers.single { it.userIdHex == OTHER_HEX }.value) } + @Test + fun `advancePointer records a read for a member that has not synced yet`() = runTest { + // The chat is open and a message is on screen before the feed writes the membership. + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(4)) + + assertEquals(4L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + + @Test + fun `advancePointer leaves the member's other pointers alone`() = runTest { + dao.upsert(member(readPointer(2), MessagePointerSerialized("DELIVERED", SELF_HEX, 5))) + + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(6)) + + val pointers = dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson.orEmpty() + assertEquals(6L, pointers.single { it.type == "READ" }.value) + assertEquals(5L, pointers.single { it.type == "DELIVERED" }.value) + } + @Test fun `deleteMembersNotIn drops departed members and leaves the rest intact`() = runTest { dao.upsert(member(readPointer(9))) diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt index 85358a3cbf..728da06ef3 100644 --- a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt @@ -69,19 +69,19 @@ class ChatMemberDataSource @Inject constructor( } } + /** + * Records [pointer] for its member in [chatId], whether or not that member has synced yet. + * + * The read-merge-write runs inside the DAO so a feed sync landing between the two halves + * cannot be lost. + */ suspend fun updatePointers(chatId: ChatId, pointer: MessagePointer) { val dao = db?.chatMemberDao() ?: return - val chatIdHex = mapper.chatIdHex(chatId) - val userIdHex = mapper.userIdHex(pointer.userId) - - val existing = dao.getMember(chatIdHex, userIdHex) - val existingPointers = existing?.pointersJson ?: emptyList() - - val merged = existingPointers - .filter { it.type != pointer.type.name } - .plus(mapper.pointerSerialized(pointer)) - - dao.updatePointers(chatIdHex, userIdHex, mapper.pointersToJson(merged)) + dao.advancePointer( + chatIdHex = mapper.chatIdHex(chatId), + userIdHex = mapper.userIdHex(pointer.userId), + pointer = mapper.pointerSerialized(pointer), + ) } /** From b28287da9791831e26a4cd45f5a12c4f564d35d5 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 28 Aug 2026 16:25:49 -0400 Subject: [PATCH 3/5] fix(chat): only ever move a read pointer forward MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `advancePointer` was last-write-wins per pointer type, which left the stream path able to do what the feed sync no longer can. `EventStreamDelegate` applies `pointerUpdates` as they arrive, and one of those is the server's copy of the member's own pointer, which can sit behind a local advance that has not been reported yet — applying it put the chat back to unread. It now keeps whichever value is further ahead, the same rule `mergePointers` uses on a feed sync, so both writers agree. --- .../flipcash/app/persistence/dao/ChatMemberDao.kt | 14 ++++++++++---- .../app/persistence/dao/ChatMemberDaoTest.kt | 10 ++++++++++ .../persistence/sources/ChatMemberDataSource.kt | 3 ++- 3 files changed, 22 insertions(+), 5 deletions(-) diff --git a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt index dbe065eac6..45f52955ef 100644 --- a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt +++ b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt @@ -73,14 +73,17 @@ interface ChatMemberDao { suspend fun getChatIdForMember(userIdHex: String, chatType: String): String? /** - * Records [pointer] for [userIdHex] in [chatIdHex], creating the member row if it is not - * there yet. + * Moves [userIdHex]'s pointer of [pointer]'s type forward in [chatIdHex], creating the member + * row if it is not there yet. The member's other pointers carry over. * * The row is not a given at this point. A read is written the moment a message is on screen, * and a pointer update can arrive off the event stream for a member the feed has not written * yet — the bare `UPDATE` this replaces matched nothing in either case and dropped the - * pointer silently. Only the member's pointer of the same type is displaced; the rest carry - * over. + * pointer silently. + * + * Forward only, on the same reasoning as [mergePointers]: the stream echoes a member's + * pointer as the server last saw it, which can be behind a local advance that has not been + * reported yet, and applying it would put the chat back to unread. */ @Transaction suspend fun advancePointer( @@ -89,6 +92,9 @@ interface ChatMemberDao { pointer: MessagePointerSerialized, ) { val existing = getMember(chatIdHex, userIdHex)?.pointersJson.orEmpty() + val current = existing.firstOrNull { it.type == pointer.type } + if (current != null && current.value >= pointer.value) return + insertOrReplace( ChatMemberEntity( chatIdHex = chatIdHex, diff --git a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt index f457efb0cb..a95c33805d 100644 --- a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt +++ b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt @@ -102,6 +102,16 @@ class ChatMemberDaoTest { assertEquals(5L, pointers.single { it.type == "DELIVERED" }.value) } + @Test + fun `advancePointer does not lower a pointer that is already ahead`() = runTest { + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(9)) + + // The stream echoes our own READ pointer as the server last saw it, behind the local one. + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(4)) + + assertEquals(9L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + @Test fun `deleteMembersNotIn drops departed members and leaves the rest intact`() = runTest { dao.upsert(member(readPointer(9))) diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt index 728da06ef3..089f609ebf 100644 --- a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt @@ -70,7 +70,8 @@ class ChatMemberDataSource @Inject constructor( } /** - * Records [pointer] for its member in [chatId], whether or not that member has synced yet. + * Moves [pointer]'s member forward in [chatId], whether or not that member has synced yet. + * A pointer already ahead of [pointer] stays where it is. * * The read-merge-write runs inside the DAO so a feed sync landing between the two halves * cannot be lost. From 26a4df1604dff3996d499261db0667cf4713f4c7 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 28 Aug 2026 16:32:52 -0400 Subject: [PATCH 4/5] fix(chat): re-report a read the server never took MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Advancing the read pointer writes locally and then reports to the server, and the report is fire-and-forget: ChatViewModel drops the Result and nothing retries. The local pointer survives a failed report — the member row keeps whichever value is further ahead — so the two copies can stay apart indefinitely. This device's badge clears while the server still considers the chat unread, which keeps sending pushes for messages already read and shows them unread on every other device and after a reinstall. A feed payload carries the server's own copy of the pointer, so the sync is where the two can be compared. When the stored pointer is ahead of the fetched one, the sync emits ReadPointerUnreported and the coordinator re-sends the advance. Rather than an in-place retry, this rides the existing sync triggers — login, foreground, network reconnect, heartbeat — so a report lost to a process death is still recovered. The re-report deliberately skips advanceReadPointer: the local write and the message-received analytics already happened when the message was seen. --- .../chat/internal/RealChatCoordinator.kt | 2 + .../internal/delegates/FeedSyncDelegate.kt | 37 +++++ .../internal/delegates/MessagingDelegate.kt | 22 +++ .../chat/FeedSyncReadPointerReportTest.kt | 137 ++++++++++++++++++ 4 files changed, 198 insertions(+) create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncReadPointerReportTest.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 535b70adc3..1028a3518b 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 @@ -145,6 +145,8 @@ class RealChatCoordinator @Inject constructor( messagingDelegate.loadMessages(event.chatId) is FeedSyncDelegate.Event.DeltaSyncNeeded -> eventStreamDelegate.performDeltaSync(event.chatId) + is FeedSyncDelegate.Event.ReadPointerUnreported -> + messagingDelegate.reportReadPointer(event.chatId, event.messageId) // Arrives after every catch-up item above it, because this is one sequential // collector over a FIFO channel. Anything reading chat history as evidence — // the wallet's "send a tip" milestone — waits for this rather than for the diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt index 60fe42e866..cf5cd9f282 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt @@ -17,6 +17,7 @@ import com.flipcash.shared.chat.FeedOperations import com.flipcash.shared.chat.FeedSyncState import com.flipcash.shared.chat.internal.ChatStateHolder import com.flipcash.services.user.UserManager +import com.getcode.opencode.model.core.ID import com.getcode.utils.TraceType import com.getcode.utils.trace import kotlinx.coroutines.CoroutineScope @@ -71,6 +72,12 @@ class FeedSyncDelegate @Inject constructor( */ data class DeltaSyncNeeded(val chatId: ChatId) : Event + /** + * The client's own READ pointer for [chatId] is ahead of the copy the feed just returned, + * so the server never took the advance. The consumer re-reports [messageId]. + */ + data class ReadPointerUnreported(val chatId: ChatId, val messageId: Long) : Event + /** * Emitted last by every successful sync, after any catch-up above it. * @@ -184,6 +191,32 @@ class FeedSyncDelegate @Inject constructor( feedObserverJob = null } + /** + * Emits [Event.ReadPointerUnreported] when the stored READ pointer for [chat] is ahead of the + * one the feed just returned. + * + * A read is written locally the moment the message is on screen and reported to the server + * afterwards, and nothing retries a report that fails. The local pointer survives — the member + * row keeps whichever value is further ahead — so the two copies can disagree indefinitely, + * leaving every other device and the pushes this one receives treating the chat as unread. A + * feed payload is the server's own copy, so the sync is where they can be compared. + */ + private suspend fun reportUnreportedRead(chat: ChatMetadata, selfId: ID) { + val server = chat.members + .firstOrNull { it.userId == selfId } + ?.pointers + ?.firstOrNull { it.type == PointerType.READ } + ?.value + ?: 0L + + // Read after the merge above, so this is max(local, server): ahead of `server` only when + // a local advance never reached it. + val local = memberDataSource.getSelfReadPointer(chat.chatId, selfId) + if (local > server) { + _events.send(Event.ReadPointerUnreported(chat.chatId, local)) + } + } + private suspend fun buildFeedFromDb( metadataEntities: List, membersByChat: Map>, @@ -229,7 +262,11 @@ class FeedSyncDelegate @Inject constructor( stateHolder.update { it.copy(feedSyncState = FeedSyncState.Synced) } trace(tag = TAG, message = "Feed synced: ${chats.size} chats", type = TraceType.Process) + val selfId = userManager.accountId + for (chat in chats) { + if (selfId != null) reportUnreportedRead(chat, selfId) + // The applied cursor, not the presence of messages, is what says whether a // transcript was ever pulled: the loop above persists each chat's last-message // preview, so "has messages" is true for nearly every chat in a feed the client 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 4babb09c8c..01d2a08fab 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 @@ -28,6 +28,8 @@ import com.flipcash.shared.chat.MessagingOperations import com.flipcash.shared.chat.internal.ChatStateHolder import com.flipcash.services.user.UserManager import com.getcode.opencode.model.core.ID +import com.getcode.utils.TraceType +import com.getcode.utils.trace import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.distinctUntilChanged @@ -235,6 +237,22 @@ class MessagingDelegate @Inject constructor( return messagingController.advancePointer(chatId, PointerType.READ, messageId) } + /** + * Re-sends a READ pointer the server never took, without touching the local copy or the + * analytics that go with a genuine read — both already happened when the message was seen. + */ + internal suspend fun reportReadPointer(chatId: ChatId, messageId: Long) { + messagingController.advancePointer(chatId, PointerType.READ, messageId) + .onFailure { + trace( + tag = TAG, + message = "Re-reporting read pointer $messageId failed", + type = TraceType.Error, + error = it, + ) + } + } + override suspend fun markAsRead(chatId: ChatId): Result { val messageId = stateHolder.current.feed .firstOrNull { it.chatId == chatId } @@ -292,4 +310,8 @@ class MessagingDelegate @Inject constructor( } // endregion + + private companion object { + const val TAG = "MessagingDelegate" + } } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncReadPointerReportTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncReadPointerReportTest.kt new file mode 100644 index 0000000000..7a87628675 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncReadPointerReportTest.kt @@ -0,0 +1,137 @@ +package com.flipcash.shared.chat + +import com.flipcash.app.persistence.sources.ChatMemberDataSource +import com.flipcash.app.persistence.sources.ChatMessageDataSource +import com.flipcash.app.persistence.sources.ChatMetadataDataSource +import com.flipcash.services.controllers.ChatController +import com.flipcash.services.models.chat.ChatFeedPage +import com.flipcash.services.models.chat.ChatId +import com.flipcash.services.models.chat.ChatMember +import com.flipcash.services.models.chat.ChatMetadata +import com.flipcash.services.models.chat.ChatType +import com.flipcash.services.models.chat.MessagePointer +import com.flipcash.services.models.chat.PointerType +import com.flipcash.services.models.UserProfile +import com.flipcash.services.user.UserManager +import com.flipcash.shared.chat.internal.ChatStateHolder +import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import org.junit.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue +import kotlin.time.Instant + +/** + * The READ pointer is written locally the moment a message is seen and reported to the server + * afterwards, and nothing retries a report that fails. The local copy therefore survives — the + * member row keeps whichever pointer is further ahead — while the server's stays behind, and every + * other device, plus the pushes this one receives, goes on treating the chat as unread. + * + * A feed payload is the server's own copy, so the sync is where the two can be compared. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class FeedSyncReadPointerReportTest { + + private val selfId = listOf(1, 2, 3) + private val chatId = ChatId("aabbccdd") + + private fun member(pointer: Long) = ChatMember( + userId = selfId, + userProfile = UserProfile.Empty, + pointers = listOf( + MessagePointer( + type = PointerType.READ, + userId = selfId, + value = pointer, + timestamp = Instant.fromEpochSeconds(1_000), + ) + ), + ) + + private class Harness(serverPointer: Long, localPointer: Long, selfId: List, chatId: ChatId) { + val chats = listOf( + ChatMetadata( + chatId = chatId, + type = ChatType.TIP_DM, + members = listOf( + ChatMember( + userId = selfId, + userProfile = UserProfile.Empty, + pointers = listOf( + MessagePointer( + type = PointerType.READ, + userId = selfId, + value = serverPointer, + timestamp = Instant.fromEpochSeconds(1_000), + ) + ), + ) + ), + lastMessage = null, + lastActivity = Instant.fromEpochSeconds(1_000), + ) + ) + + val memberDataSource = mockk(relaxed = true).also { source -> + coEvery { source.getSelfReadPointer(any(), any()) } returns localPointer + } + + val chatController = mockk(relaxed = true).also { controller -> + coEvery { controller.getDmChatFeed(ChatType.CONTACT_DM, any()) } returns + Result.success(ChatFeedPage(emptyList(), null, false)) + coEvery { controller.getDmChatFeed(ChatType.TIP_DM, any()) } returns + Result.success(ChatFeedPage(chats, null, false)) + } + + val delegate = FeedSyncDelegate( + chatController = chatController, + metadataDataSource = mockk(relaxed = true), + messageDataSource = mockk(relaxed = true), + memberDataSource = memberDataSource, + stateHolder = ChatStateHolder(), + userManager = mockk(relaxed = true).also { + every { it.accountId } returns selfId + }, + ) + + fun sync(scope: TestScope): List { + val events = mutableListOf() + delegate.events.onEach { events += it }.launchIn(scope.backgroundScope) + delegate.initialize(scope.backgroundScope) + delegate.syncFeed() + scope.runCurrent() + return events + } + } + + @Test + fun `a read the server never took is re-reported`() = runTest { + val events = Harness(serverPointer = 3, localPointer = 9, selfId = selfId, chatId = chatId) + .sync(this) + + assertEquals( + listOf(FeedSyncDelegate.Event.ReadPointerUnreported(chatId, 9)), + events.filterIsInstance(), + "the local pointer is ahead of the server's, so the advance never landed", + ) + } + + @Test + fun `a read the server already has is not re-reported`() = runTest { + val events = Harness(serverPointer = 9, localPointer = 9, selfId = selfId, chatId = chatId) + .sync(this) + + assertTrue( + events.none { it is FeedSyncDelegate.Event.ReadPointerUnreported }, + "the server is level with the client; there is nothing to report", + ) + } +} From 99486628fe4bf4c53451debf6d433b9ff706c04e Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 28 Aug 2026 16:39:46 -0400 Subject: [PATCH 5/5] fix(chat): reconnect on stream death instead of at the next tick MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stream that fails outside the reconnect loop's retryable set — INTERNAL, UNKNOWN, a ping timeout, or the loop running out of attempts — is not retried by the loop at all. Nothing else was watching for that: the heartbeat polls liveness every 30s, so the connection stayed down for up to a full interval after it died, and every message in that window arrived late. EventStreamingController now emits on streamFailures when it clears the ref for a stream that ended this way, and the heartbeat waits on that signal or the tick, whichever comes first. A pending signal is dropped when a stream is opened, so it cannot wake the supervisor against a stream that has not had a chance to connect yet. Reopening backs off from the second consecutive attempt on, doubling from 1s up to the tick interval, so a stream that fails as soon as it opens cannot spin the loop — the worst case is where every failure used to sit. Liveness is read after that wait rather than before, leaving alone a stream that a lifecycle or network trigger reopened in the meantime. --- .../internal/delegates/EventStreamDelegate.kt | 53 ++++++++- .../shared/chat/EventStreamHeartbeatTest.kt | 106 ++++++++++++++++++ .../controllers/EventStreamingController.kt | 20 +++- 3 files changed, 172 insertions(+), 7 deletions(-) create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamHeartbeatTest.kt diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt index 88ebd20a25..f2123b6bf7 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt @@ -29,17 +29,21 @@ import com.getcode.utils.TraceType import com.getcode.utils.trace import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.firstOrNull import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.receiveAsFlow import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeoutOrNull import javax.inject.Inject import javax.inject.Singleton import kotlin.time.Clock +import kotlin.time.Duration import kotlin.time.Duration.Companion.seconds /** @@ -86,6 +90,8 @@ class EventStreamDelegate @Inject constructor( companion object { private const val TAG = "EventStreamDelegate" private val GAP_FILL_DELAY = 2.seconds + private val HEARTBEAT_INTERVAL = 30.seconds + private val REOPEN_BACKOFF_BASE = 1.seconds } sealed interface Event { @@ -164,22 +170,57 @@ class EventStreamDelegate @Inject constructor( eventStreamingController.close() } + /** + * Supervises the stream: whenever it is found down, syncs the feed and reopens it. + * + * Each pass waits for whichever comes first, a [HEARTBEAT_INTERVAL] tick or + * [EventStreamingController.streamFailures] reporting a stream that ended with nothing + * retrying it. The failure signal is what keeps such a stream from costing a whole interval: + * a status outside the reconnect loop's retryable set, or a ping timeout, is never retried by + * the loop, and the tick was the only thing that noticed. + * + * Reopening backs off from the second consecutive attempt on, so a stream that fails as soon + * as it opens cannot spin the loop. The backoff tops out at [HEARTBEAT_INTERVAL], which is + * where every failure used to wait. + */ internal fun startHeartbeat(onReconnect: () -> Unit) { val scope = scope ?: return stopHeartbeat() heartbeatJob = scope.launch { + var consecutiveReopens = 0 + while (true) { - delay(30.seconds) - if (!eventStreamingController.isStreamActive) { - trace(tag = TAG, message = "Heartbeat: event stream dead, syncing feed and reconnecting", type = TraceType.Process) - onReconnect() - eventStreamingController.close() - open() + // A failures flow that completes means no signal is coming, not that one just + // arrived — waiting out the tick keeps this a supervisor rather than a spin. + val failed = withTimeoutOrNull(HEARTBEAT_INTERVAL) { + eventStreamingController.streamFailures.firstOrNull() ?: awaitCancellation() + } != null + + // Only the failure-driven wake can arrive back to back; the tick already spaces + // itself out. Liveness is read after the wait, so a stream another trigger + // reopened in the meantime is left alone. + if (failed) delay(reopenBackoff(consecutiveReopens)) + + if (eventStreamingController.isStreamActive) { + consecutiveReopens = 0 + continue } + consecutiveReopens++ + + trace(tag = TAG, message = "Event stream down, syncing feed and reconnecting", type = TraceType.Process) + onReconnect() + eventStreamingController.close() + open() } } } + private fun reopenBackoff(consecutiveReopens: Int): Duration = when { + consecutiveReopens <= 0 -> Duration.ZERO + else -> (REOPEN_BACKOFF_BASE * (1 shl (consecutiveReopens - 1).coerceAtMost(5))) + .coerceAtMost(HEARTBEAT_INTERVAL) + } + internal fun stopHeartbeat() { heartbeatJob?.cancel() heartbeatJob = null diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamHeartbeatTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamHeartbeatTest.kt new file mode 100644 index 0000000000..f0d7207c40 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamHeartbeatTest.kt @@ -0,0 +1,106 @@ +package com.flipcash.shared.chat + +import com.flipcash.services.controllers.EventStreamingController +import com.flipcash.shared.chat.internal.ChatStateHolder +import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.emptyFlow +import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.seconds + +/** + * The heartbeat is the only thing that reopens a stream the reconnect loop has stopped retrying — + * a status outside its retryable set, a ping timeout, or its attempts running out. Polling for + * that on a fixed interval costs up to the whole interval in missed messages. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +class EventStreamHeartbeatTest { + + private class Harness { + val failures = Channel(capacity = Channel.CONFLATED) + + val controller = mockk(relaxed = true).also { controller -> + every { controller.streamFailures } returns failures.receiveAsFlow() + every { controller.chatUpdates } returns emptyFlow() + every { controller.blobUpdates } returns emptyFlow() + // Nothing is connected for the whole test: every wake finds the stream down. + every { controller.isStreamActive } returns false + every { controller.isConnected } returns false + } + + val delegate = EventStreamDelegate( + eventStreamingController = controller, + messagingController = mockk(relaxed = true), + metadataDataSource = mockk(relaxed = true), + messageDataSource = mockk(relaxed = true), + memberDataSource = mockk(relaxed = true), + tokenCoordinator = mockk(relaxed = true), + userManager = mockk(relaxed = true), + stateHolder = ChatStateHolder(), + analytics = mockk(relaxed = true), + exchange = mockk(relaxed = true), + ) + + fun start(scope: TestScope) { + delegate.initialize(scope.backgroundScope) + delegate.startHeartbeat { } + scope.runCurrent() + } + } + + @Test + fun `a stream that gives up is reopened without waiting out the tick`() = runTest { + val harness = Harness() + harness.start(this) + + harness.failures.trySend(Unit) + advanceTimeBy(100.milliseconds) + runCurrent() + + verify(exactly = 1) { harness.controller.open(any()) } + } + + @Test + fun `a stream failing as soon as it opens is not reopened in a hot loop`() = runTest { + val harness = Harness() + harness.start(this) + + harness.failures.trySend(Unit) + advanceTimeBy(100.milliseconds) + runCurrent() + + // The reopened stream fails immediately too. + harness.failures.trySend(Unit) + advanceTimeBy(500.milliseconds) + runCurrent() + verify(exactly = 1) { harness.controller.open(any()) } + + advanceTimeBy(1.seconds) + runCurrent() + verify(exactly = 2) { harness.controller.open(any()) } + } + + @Test + fun `a dead stream is still reopened at the tick`() = runTest { + val harness = Harness() + harness.start(this) + + advanceTimeBy(31.seconds) + runCurrent() + + verify(exactly = 1) { harness.controller.open(any()) } + } +} diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt index e8f48019b3..4a0269142e 100644 --- a/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt +++ b/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt @@ -24,6 +24,17 @@ class EventStreamingController @Inject constructor( private val _blobUpdates = Channel(capacity = Channel.UNLIMITED) val blobUpdates: Flow = _blobUpdates.receiveAsFlow() + // Conflated: the signal is "there is no stream any more", which two failures state no better + // than one, and it has to survive being emitted while the consumer is between waits. + private val _streamFailures = Channel(capacity = Channel.CONFLATED) + + /** + * Emits when a stream ends with nothing retrying it — a status outside the reconnect loop's + * retryable set, or the loop exhausting its attempts. Reopening is the consumer's call; this + * only says the connection is gone, so a supervisor need not poll to find out. + */ + val streamFailures: Flow = _streamFailures.receiveAsFlow() + // Guards all reads/writes of [streamRef]. open()/close() are invoked // concurrently from multiple triggers (login, lifecycle onStart, network // reconnect, feature-flag, heartbeat) on the multi-threaded IO scope, so @@ -43,6 +54,10 @@ class EventStreamingController @Inject constructor( return true } + // A pending failure describes the stream being replaced here. Left queued, it would wake + // the supervisor against the new stream before it has had a chance to connect. + while (_streamFailures.tryReceive().isSuccess) Unit + val owner = userManager.accountCluster?.authority?.keyPair ?: run { trace("EventStreamingController: No account cluster, cannot open stream") return false @@ -66,7 +81,10 @@ class EventStreamingController @Inject constructor( // event creates a fresh stream. Only clear if it is still THIS // stream — never null out a newer ref opened after us. synchronized(lock) { - if (streamRef === openedRef) streamRef = null + if (streamRef === openedRef) { + streamRef = null + _streamFailures.trySend(Unit) + } } }, )