fix(conversations): don't let stale syncs revert a pending read marker

Reading a chat and returning to the conversation list often flipped the
entry to read and back to unread seconds later, sticking there until
the next sync. Log analysis showed a request-ordering race: the
resume-triggered room list sync request leaves before the concurrently
sent read marker reaches the server, so the sync response is computed
against the pre-marker state and the authoritative merge restores the
old unread state - even though the server accepted the marker moments
later.

Read markers written locally are now tracked as pending in the
ChatMessageSyncer singleton until a room list sync confirms their
delivery. While a marker is pending and a sync reports a read state
still behind it, that state is provably stale and the local read state
is kept in the merge. Once the server has caught up (or moved past the
marker, e.g. read further on another device) the marker is released and
the server state applies unchanged - including a lower one, so marking
as unread from another device keeps working. When sending the marker
ultimately fails, ReadMarkerSyncWorker releases it as well, so the
server re-asserts its state at the next sync and the client can never
stay diverged; a process death empties the in-memory pending state with
the same effect.

Assisted-by: Claude Code:claude-fable-5
Signed-off-by: Andy Scherzinger <info@andy-scherzinger.de>

Author: AndyScherzinger

app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt

@@ -404,11 +404,38 @@ class ChatMessageSyncer @Inject constructor(
         )
     }
 
+    /**
+     * Read markers that were written locally but whose delivery the server has not confirmed yet.
+     *
+     * A room list sync request can be answered before a marker sent at the same time reaches the
+     * server, so the sync response carries a provably stale read state — applying it would revert
+     * the conversation entry to unread until the next sync. While a marker is pending, such stale
+     * responses are kept out of the merge; the entry is only released once a sync confirms the
+     * marker (server read state caught up) or sending it ultimately failed, so the server's
+     * authority is restored either way and marking as unread from another device stays possible.
+     */
+    private val pendingReadMarkers = ConcurrentHashMap<String, Int>()
+
+    /**
+     * The locally written but not yet server-confirmed read marker of the conversation, or null.
+     */
+    fun pendingReadMarker(internalConversationId: String): Int? = pendingReadMarkers[internalConversationId]
+
+    /**
+     * Releases a pending read marker: called when a room list sync confirmed it or when sending it
+     * ultimately failed. Only removes [lastReadMessage] itself, so a newer marker written in the
+     * meantime stays pending.
+     */
+    fun clearPendingReadMarker(internalConversationId: String, lastReadMessage: Int) {
+        pendingReadMarkers.remove(internalConversationId, lastReadMessage)
+    }
+
     /**
      * Optimistically writes the user's read state into the conversation entry, so the
      * conversation list reflects it immediately — before (and independent of) the read marker
      * reaching the server. The unread count is recounted from the cached messages above the new
-     * marker. The server stays the authority: the next room list sync re-asserts its state.
+     * marker. The server stays the authority: the next room list sync re-asserts its state, with
+     * [pendingReadMarkers] bridging the window until the marker's delivery is confirmed.
      */
     suspend fun updateLocalReadState(target: SyncTarget, lastReadMessage: Int) {
         val conversation =
@@ -418,6 +445,7 @@ class ChatMessageSyncer @Inject constructor(
             messageId = lastReadMessage.toLong(),
             excludedActorId = target.user.userId!!
         )
+        pendingReadMarkers[target.internalConversationId] = lastReadMessage
         conversationsDao.updateReadState(conversation.internalId, lastReadMessage, unreadMessages)
         Log.d(TAG, "Local read state for room ${target.roomToken}: lastRead=$lastReadMessage, unread=$unreadMessages")
     }

app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt

@@ -185,7 +185,7 @@ class OfflineFirstConversationsRepository @Inject constructor(
 
             dao.syncConversationsForUser(
                 accountId = user.id!!,
-                serverItems = conversationsFromSync,
+                serverItems = preserveReadStateOfPendingMarkers(previousConversations, conversationsFromSync),
                 conversationIdsToDelete = determineLeftConversationIds(previousConversations, conversationsFromSync)
             )
 
@@ -295,6 +295,41 @@ class OfflineFirstConversationsRepository @Inject constructor(
             connectivityManager.restrictBackgroundStatus == ConnectivityManager.RESTRICT_BACKGROUND_STATUS_ENABLED
     }
 
+    /**
+     * Keeps provably stale read states out of the sync merge: a sync response computed before a
+     * concurrently sent read marker reached the server still reports the room as unread, and
+     * applying it would revert the conversation entry until the next sync. While a marker for the
+     * room is pending and the server's read state is still behind it, the local read state is
+     * kept; once the server has caught up (or moved past it, e.g. read further on another device)
+     * the marker is released and the server state applies unchanged — including a lower one, so
+     * marking as unread from another device keeps working.
+     */
+    private fun preserveReadStateOfPendingMarkers(
+        previousConversations: Map<String, ConversationEntity>,
+        conversationsFromSync: List<ConversationEntity>
+    ): List<ConversationEntity> =
+        conversationsFromSync.map { serverItem ->
+            val pendingMarker = chatMessageSyncer.pendingReadMarker(serverItem.internalId)
+                ?: return@map serverItem
+            val previous = previousConversations[serverItem.internalId]
+                ?: return@map serverItem
+
+            if (serverItem.lastReadMessage < pendingMarker) {
+                Log.d(
+                    TAG,
+                    "Keeping local read state for room ${serverItem.token}: server lastRead=" +
+                        "${serverItem.lastReadMessage} is behind the pending read marker $pendingMarker"
+                )
+                serverItem.copy(
+                    lastReadMessage = previous.lastReadMessage,
+                    unreadMessages = previous.unreadMessages
+                )
+            } else {
+                chatMessageSyncer.clearPendingReadMarker(serverItem.internalId, pendingMarker)
+                serverItem
+            }
+        }
+
     private fun determineLeftConversationIds(
         previousConversations: Map<String, ConversationEntity>,
         conversationsFromSync: List<ConversationEntity>

app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt

@@ -21,6 +21,7 @@ import androidx.work.WorkerParameters
 import autodagger.AutoInjector
 import com.nextcloud.talk.application.NextcloudTalkApplication
 import com.nextcloud.talk.application.NextcloudTalkApplication.Companion.sharedApplication
+import com.nextcloud.talk.chat.data.network.ChatMessageSyncer
 import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource
 import com.nextcloud.talk.users.UserManager
 import com.nextcloud.talk.utils.ApiUtils
@@ -34,10 +35,12 @@ import javax.inject.Inject
  *
  * The caller updates the local conversation entry optimistically before enqueuing this worker, so
  * the conversation list reflects the read state immediately. The server stays the authority:
- * every room list sync re-asserts the server's read state over the local entry, so when all
- * attempts fail the client falls back to the server state at the next sync instead of staying
- * diverged. Work is unique per room with [ExistingWorkPolicy.REPLACE], so the newest marker for a
- * room always wins and retries can never ship a stale marker backwards.
+ * every room list sync re-asserts the server's read state over the local entry — guarded by the
+ * pending marker in [ChatMessageSyncer] only while the marker is provably not delivered yet.
+ * When all attempts fail, the pending marker is released so the client falls back to the server
+ * state at the next sync instead of staying diverged. Work is unique per room with
+ * [ExistingWorkPolicy.REPLACE], so the newest marker for a room always wins and retries can never
+ * ship a stale marker backwards.
  */
 @AutoInjector(NextcloudTalkApplication::class)
 class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) :
@@ -49,6 +52,9 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) :
     @Inject
     lateinit var chatNetworkDataSource: ChatNetworkDataSource
 
+    @Inject
+    lateinit var chatMessageSyncer: ChatMessageSyncer
+
     override suspend fun doWork(): Result {
         sharedApplication!!.componentApplication.inject(this)
 
@@ -71,7 +77,7 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) :
         val credentials = user?.let { ApiUtils.getCredentials(it.username, it.token) }
         if (user == null || credentials == null) {
             Log.e(TAG, "No user or credentials found for user id $userId, dropping read marker sync")
-            return Result.failure()
+            return fail(userId, roomToken, lastReadMessage)
         }
 
         val sent = runCatching {
@@ -88,11 +94,21 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) :
             Result.success()
         } else {
             Log.w(TAG, "Sending read marker for room $roomToken failed (attempt ${runAttemptCount + 1})")
-            retryOrFail()
+            retryOrFail(userId, roomToken, lastReadMessage)
         }
     }
 
-    private fun retryOrFail(): Result = if (runAttemptCount < MAX_RUN_ATTEMPTS - 1) Result.retry() else Result.failure()
+    private fun retryOrFail(userId: Long, roomToken: String, lastReadMessage: Int): Result =
+        if (runAttemptCount < MAX_RUN_ATTEMPTS - 1) {
+            Result.retry()
+        } else {
+            fail(userId, roomToken, lastReadMessage)
+        }
+
+    private fun fail(userId: Long, roomToken: String, lastReadMessage: Int): Result {
+        chatMessageSyncer.clearPendingReadMarker("$userId@$roomToken", lastReadMessage)
+        return Result.failure()
+    }
 
     companion object {
         private val TAG: String = ReadMarkerSyncWorker::class.java.simpleName

app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt

@@ -153,6 +153,46 @@ class ConversationListFreshnessIntegrationTest {
         }
     }
 
+    @Test
+    fun `a sync cannot revert the read state while its marker is pending`() {
+        val user = user(withKeepNotificationsCapability = false)
+        val repository = OfflineFirstConversationsRepository(
+            db.conversationsDao(),
+            conversationsNetwork,
+            chatNetwork,
+            networkMonitor,
+            syncer,
+            ApplicationProvider.getApplicationContext()
+        )
+        whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn(
+            // computed before the pending marker 12 reached the server: provably stale
+            Observable.just(listOf(staleServerRoom(lastReadMessage = 10, unreadMessages = 2))),
+            // the server caught up with the marker: confirms it and releases the guard
+            Observable.just(listOf(staleServerRoom(lastReadMessage = 12, unreadMessages = 0))),
+            // marked as unread from another device: must apply now that nothing is pending
+            Observable.just(listOf(staleServerRoom(lastReadMessage = 8, unreadMessages = 4)))
+        )
+
+        runBlocking {
+            seedConversation(lastActivity = 12, lastReadMessage = 10, unreadMessages = 2)
+            syncer.updateLocalReadState(target(), lastReadMessage = 12)
+            assertEquals(12, conversationEntity().lastReadMessage)
+            assertEquals(0, conversationEntity().unreadMessages)
+
+            repository.getRooms(user).join()
+            assertEquals("stale sync must not revert the read marker", 12, conversationEntity().lastReadMessage)
+            assertEquals("stale sync must not revert the unread count", 0, conversationEntity().unreadMessages)
+
+            repository.getRooms(user).join()
+            assertEquals(12, conversationEntity().lastReadMessage)
+            assertEquals(0, conversationEntity().unreadMessages)
+
+            repository.getRooms(user).join()
+            assertEquals("server authority must be restored", 8, conversationEntity().lastReadMessage)
+            assertEquals("marking unread on another device must apply", 4, conversationEntity().unreadMessages)
+        }
+    }
+
     @Test
     fun `room list flow reflects database writes reactively`() {
         val user = user(withKeepNotificationsCapability = false)
@@ -184,6 +224,14 @@ class ConversationListFreshnessIntegrationTest {
         }
     }
 
+    private fun staleServerRoom(lastReadMessage: Int, unreadMessages: Int): Conversation =
+        Conversation(
+            token = ROOM_TOKEN,
+            lastActivity = 12,
+            lastReadMessage = lastReadMessage,
+            unreadMessages = unreadMessages
+        )
+
     private suspend fun seedConversation(lastActivity: Long, lastReadMessage: Int, unreadMessages: Int) {
         val entity = Conversation(
             token = ROOM_TOKEN,