From c575c6e1bbd0f64ae639bb99a1cee8d12b37f610 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Sun, 11 Oct 2026 15:34:11 +0000 Subject: [PATCH 1/2] fix(conversations): do not apply server responses older than the stored entry A room list sync and a single room refresh both replace the whole conversation entry. The one that was requested earlier but answered later overwrote the newer entry with stale activity, read state, call state, archive flag and so on, and a full sync that was requested before a room was refreshed deleted the room again. The pending local state guards only cover what the user changed, not the order of two server responses. Remember per room the start of the request behind the newest state written to its entry, and leave out of the merge, and out of the deletions, a response whose request started before that. The start is taken from System.nanoTime: it only orders requests inside the process. A background catch-up writes only the last message, its activity and the unread count, so it marks just that group of fields, with the start of the request that fetched the message. A response that is older than the group but newer than the entry applies, and keeps the stored values of the group. The unread count follows from the last read message: when the response moved the marker, the count is derived again from it, and taken from the response where that is not possible. Reading the stored entries, the merge, the write and the mark happen under one lock per account, and the write and the mark cannot be cancelled in between. Assisted-by: Claude-Code:claude-opus-5-5 Signed-off-by: Krainov Gleb --- .../chat/data/network/ChatMessageSyncer.kt | 46 +- .../talk/chat/domain/ChatPullResult.kt | 4 +- .../data/network/ConversationListUpdater.kt | 240 +++++++- .../OfflineFirstConversationsRepository.kt | 83 ++- ...onversationListFreshnessIntegrationTest.kt | 517 +++++++++++++++++- .../network/ConversationListUpdaterPinTest.kt | 12 +- .../ConversationListUpdaterUnreadTest.kt | 6 +- ...OfflineFirstConversationsRepositoryTest.kt | 15 +- 8 files changed, 843 insertions(+), 80 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt index 5a9b74eeb4f..bb076e740a6 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt @@ -97,7 +97,8 @@ class ChatMessageSyncer @Inject constructor( * * [newestPersistedMessage] is the newest persisted message that qualifies as a conversation's * last message (system messages that never change a conversation's preview are skipped). It - * feeds the conversation list update after a background catch-up. + * feeds the conversation list update after a background catch-up. [newestMessageRequestedAtNanos] + * is the [System.nanoTime] at which the request that fetched it was started. * * [visibleMessageIds] are the ids of all persisted messages of this round that render as their * own bubble in an open chat (see [CHAT_HIDDEN_SYSTEM_MESSAGE_TYPES] — a different, broader @@ -115,7 +116,8 @@ class ChatMessageSyncer @Inject constructor( val persistedMessageCount: Int = 0, val syncFailed: Boolean = false, val newestPersistedMessage: ChatMessageDto? = null, - val visibleMessageIds: List = emptyList() + val visibleMessageIds: List = emptyList(), + val newestMessageRequestedAtNanos: Long? = null ) /** @@ -507,7 +509,12 @@ class ChatMessageSyncer @Inject constructor( oldestPersistedMessageId = anchorOutcome.oldestPersistedMessageId, persistedMessageCount = anchorOutcome.persistedMessageCount + backlogOutcome.persistedMessageCount, syncFailed = backlogOutcome.syncFailed, - newestPersistedMessage = backlogOutcome.newestPersistedMessage ?: anchorOutcome.newestPersistedMessage + newestPersistedMessage = backlogOutcome.newestPersistedMessage ?: anchorOutcome.newestPersistedMessage, + newestMessageRequestedAtNanos = if (backlogOutcome.newestPersistedMessage != null) { + backlogOutcome.newestMessageRequestedAtNanos + } else { + anchorOutcome.newestMessageRequestedAtNanos + } ) } @@ -539,6 +546,7 @@ class ChatMessageSyncer @Inject constructor( var oldestPersisted: Long? = null var newestPersisted: Long? = null var newestPersistedMessage: ChatMessageDto? = null + var newestMessageRequestedAtNanos: Long? = null repeat(MAX_BACKLOG_ROUNDS) { val fieldMap = buildFieldMap( @@ -558,7 +566,10 @@ class ChatMessageSyncer @Inject constructor( totalCount += roundOutcome.persistedMessageCount oldestPersisted = oldestPersisted ?: roundOutcome.oldestPersistedMessageId newestPersisted = roundOutcome.newestPersistedMessageId ?: newestPersisted - newestPersistedMessage = roundOutcome.newestPersistedMessage ?: newestPersistedMessage + if (roundOutcome.newestPersistedMessage != null) { + newestPersistedMessage = roundOutcome.newestPersistedMessage + newestMessageRequestedAtNanos = roundOutcome.newestMessageRequestedAtNanos + } } val caughtUp = !roundOutcome.persistedNewMessages || roundOutcome.persistedMessageCount < limit @@ -570,7 +581,8 @@ class ChatMessageSyncer @Inject constructor( oldestPersistedMessageId = oldestPersisted, persistedMessageCount = totalCount, syncFailed = roundOutcome.syncFailed, - newestPersistedMessage = newestPersistedMessage + newestPersistedMessage = newestPersistedMessage, + newestMessageRequestedAtNanos = newestMessageRequestedAtNanos ) } anchor = nextAnchor @@ -603,7 +615,12 @@ class ChatMessageSyncer @Inject constructor( oldestPersistedMessageId = fallbackOutcome.oldestPersistedMessageId, persistedMessageCount = fallbackOutcome.persistedMessageCount, syncFailed = fallbackOutcome.syncFailed, - newestPersistedMessage = fallbackOutcome.newestPersistedMessage ?: newestPersistedMessage + newestPersistedMessage = fallbackOutcome.newestPersistedMessage ?: newestPersistedMessage, + newestMessageRequestedAtNanos = if (fallbackOutcome.newestPersistedMessage != null) { + fallbackOutcome.newestMessageRequestedAtNanos + } else { + newestMessageRequestedAtNanos + } ) } @@ -636,7 +653,7 @@ class ChatMessageSyncer @Inject constructor( var totalCount = 0 var oldestPersisted: Long? = null var newestPersisted: Long? = null - var newestPersistedMessage: ChatMessageDto? = null + var newestMessageOutcome: SyncOutcome? = null repeat(MAX_VISIBLE_MESSAGE_ROUNDS) { val roundOutcome = pullAndPersistMessages(target, fieldMap, events) @@ -653,7 +670,7 @@ class ChatMessageSyncer @Inject constructor( } val foundNewVisibleMessage = roundOutcome.visibleMessageIds.any { it != knownAnchorId } if (foundNewVisibleMessage) { - newestPersistedMessage = roundOutcome.newestPersistedMessage + newestMessageOutcome = roundOutcome } val nextAnchor = if (lookIntoFuture) { @@ -670,7 +687,8 @@ class ChatMessageSyncer @Inject constructor( oldestPersistedMessageId = oldestPersisted, persistedMessageCount = totalCount, syncFailed = roundOutcome.syncFailed, - newestPersistedMessage = newestPersistedMessage + newestPersistedMessage = newestMessageOutcome?.newestPersistedMessage, + newestMessageRequestedAtNanos = newestMessageOutcome?.newestMessageRequestedAtNanos ) } @@ -689,7 +707,8 @@ class ChatMessageSyncer @Inject constructor( newestPersistedMessageId = newestPersisted, oldestPersistedMessageId = oldestPersisted, persistedMessageCount = totalCount, - newestPersistedMessage = newestPersistedMessage + newestPersistedMessage = newestMessageOutcome?.newestPersistedMessage, + newestMessageRequestedAtNanos = newestMessageOutcome?.newestMessageRequestedAtNanos ) } @@ -721,6 +740,7 @@ class ChatMessageSyncer @Inject constructor( var attempts = 1 while (attempts < MAX_PULL_ATTEMPTS) { + val requestedAtNanos = System.nanoTime() runCatching { network.pullChatMessages(target.credentials, target.urlForChatting, fieldMap) }.fold( @@ -728,7 +748,8 @@ class ChatMessageSyncer @Inject constructor( val result = when (response.code()) { HTTP_CODE_OK -> ChatPullResult.Success( messages = response.body()?.ocs?.data.orEmpty(), - lastCommonRead = response.headers()["X-Chat-Last-Common-Read"]?.toInt() + lastCommonRead = response.headers()["X-Chat-Last-Common-Read"]?.toInt(), + requestedAtNanos = requestedAtNanos ) HTTP_CODE_NOT_MODIFIED -> ChatPullResult.NotModified HTTP_CODE_PRECONDITION_FAILED -> ChatPullResult.PreconditionFailed @@ -857,7 +878,8 @@ class ChatMessageSyncer @Inject constructor( oldestPersistedMessageId = persistedMessages.minOfOrNull { it.id }, persistedMessageCount = persistedMessages.size, newestPersistedMessage = previewEligibleMessages.maxByOrNull { it.id }, - visibleMessageIds = chatVisibleMessages.map { it.id } + visibleMessageIds = chatVisibleMessages.map { it.id }, + newestMessageRequestedAtNanos = result.requestedAtNanos ) } else { Log.d(TAG, "No new messages to update") diff --git a/app/src/main/java/com/nextcloud/talk/chat/domain/ChatPullResult.kt b/app/src/main/java/com/nextcloud/talk/chat/domain/ChatPullResult.kt index 9a1fdac194f..e60463abffc 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/domain/ChatPullResult.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/domain/ChatPullResult.kt @@ -10,7 +10,9 @@ package com.nextcloud.talk.chat.domain import com.nextcloud.talk.models.json.chat.ChatMessageDto sealed class ChatPullResult { - data class Success(val messages: List, val lastCommonRead: Int?) : ChatPullResult() + /** [requestedAtNanos] is the [System.nanoTime] at which the request that got the response was started. */ + data class Success(val messages: List, val lastCommonRead: Int?, val requestedAtNanos: Long) : + ChatPullResult() object NotModified : ChatPullResult() object PreconditionFailed : ChatPullResult() diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt index 8866b601d44..7b1c2ded859 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt @@ -16,7 +16,10 @@ import com.nextcloud.talk.data.database.dao.ConversationsDao import com.nextcloud.talk.data.database.model.ConversationEntity import com.nextcloud.talk.utils.Undo import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.flow.first +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import java.util.concurrent.ConcurrentHashMap import javax.inject.Inject @@ -49,6 +52,125 @@ class ConversationListUpdater @Inject constructor( */ private val pendingReadMarkers = ConcurrentHashMap() + /** + * A set of fields that one writer sets without writing the whole entry. Such a writer is not + * ordered against a response by the entry mark of [recordServerWrite], as a response may be + * newer than the entry and still older than the fields. + */ + enum class FieldGroup { + /** The last message of the conversation, its activity and the unread count, as a catch-up writes them. */ + PREVIEW + } + + /** + * The [System.nanoTime] at which the request behind the newest state written to a room's entry + * was started: the request of the last room list sync or single room refresh that wrote it. A + * server response replaces the whole entry, so one whose request was started before that + * moment is older than what is stored and is not applied; see [isSuperseded]. The same is kept + * per [FieldGroup] for writers of a part of the entry. In memory only: the clock is monotonic + * within a process, and a process starts with nothing in flight. The marks are never removed, + * they are bounded by the rooms seen in the process, and the mark of a deleted room must stay + * so that a late response does not bring the room back. + */ + private val serverWriteStarts = ConcurrentHashMap() + + private val fieldGroupWriteStarts = ConcurrentHashMap, Long>() + + private val serverWriteLocks = ConcurrentHashMap() + + /** + * Runs [block] holding the lock of [accountId] that orders writes of server state. A sync or + * single room refresh holds it from reading the stored entries through writing its response and + * [recordServerWrite], and a catch-up holds it for its write: otherwise a response could be + * judged against a state that changes before the response is written. + */ + suspend fun withServerWriteLock(accountId: Long, block: suspend () -> T): T = + serverWriteLocks.getOrPut(accountId) { Mutex() }.withLock { block() } + + /** + * Whether the room's entry holds state from a request started after [requestedAtNanos], so a + * response to a request started then is older than the entry. + */ + fun isSuperseded(internalConversationId: String, requestedAtNanos: Long): Boolean = + isNewer(serverWriteStarts[internalConversationId], requestedAtNanos) + + /** + * Whether any write to the room, of the whole entry or of a [FieldGroup], was made from a + * request started after [requestedAtNanos]. Such a write shows that the account has the room. + */ + fun hasNewerWrite(internalConversationId: String, requestedAtNanos: Long): Boolean = + isSuperseded(internalConversationId, requestedAtNanos) || + FieldGroup.entries.any { isNewer(fieldGroupWriteStarts[internalConversationId to it], requestedAtNanos) } + + private fun isNewer(writtenAt: Long?, requestedAtNanos: Long): Boolean = + writtenAt != null && requestedAtNanos - writtenAt < 0 + + /** + * Records that the entries of [internalConversationIds] were written, or deleted, from state + * requested at [requestedAtNanos]; with a [fieldGroup], that only its fields were written. Call + * it after the write, holding [withServerWriteLock]. + */ + fun recordServerWrite( + internalConversationIds: Collection, + requestedAtNanos: Long, + fieldGroup: FieldGroup? = null + ) { + internalConversationIds.forEach { id -> + if (fieldGroup == null) { + serverWriteStarts.merge(id, requestedAtNanos, ::newerOf) + } else { + fieldGroupWriteStarts.merge(id to fieldGroup, requestedAtNanos, ::newerOf) + } + } + } + + private fun newerOf(recorded: Long, new: Long): Long = if (new - recorded > 0) new else recorded + + /** + * Gives the fields of [group] the values of the stored entry: a response older than their + * write must not revert them. [ownUserId] is null when the account has no user id. + */ + private suspend fun keepStored( + group: FieldGroup, + serverItem: ConversationEntity, + stored: ConversationEntity, + ownUserId: String? + ): ConversationEntity = + when (group) { + FieldGroup.PREVIEW -> serverItem.copy( + lastMessage = stored.lastMessage, + lastActivity = stored.lastActivity, + unreadMessages = unreadFollowingMarker(serverItem, stored, ownUserId) + ) + } + + /** + * The unread count is not a server value but follows from the last read message, so a pair of + * the two must come from one state. With the marker of the response unchanged the stored count + * stands; otherwise it is derived again from the marker of the response. Where that is not + * possible the count of the response is taken, for a lower marker the higher of it and the + * stored count: both are lower bounds, the response does not know the new messages and the + * stored count was derived from a higher marker. + */ + private suspend fun unreadFollowingMarker( + serverItem: ConversationEntity, + stored: ConversationEntity, + ownUserId: String? + ): Int { + if (serverItem.lastReadMessage == stored.lastReadMessage) { + return stored.unreadMessages + } + val derived = ownUserId?.let { + deriveUnreadMessagesCount(serverItem.internalId, it, serverItem.lastReadMessage) + } ?: UNREAD_COUNT_UNKNOWN + return when { + derived != UNREAD_COUNT_UNKNOWN -> derived + serverItem.lastReadMessage < stored.lastReadMessage -> + maxOf(serverItem.unreadMessages, stored.unreadMessages) + else -> serverItem.unreadMessages + } + } + /** * Favorite flags applied locally but not yet confirmed by a server response; guarded in the * merge like the pending read markers. @@ -285,14 +407,33 @@ class ConversationListUpdater @Inject constructor( * the local state is kept; once the server confirms it (or moved past it, e.g. read further * or unfavorited on another device after confirmation) the pending change is released and * the server state applies unchanged, so the server stays the authority. + * + * [requestedAtNanos] is the [System.nanoTime] at which the request that produced + * [conversationsFromServer] was started. A room whose entry already holds state from a later + * request is left out of the result, as the response is older than the entry: the pending + * guards above only cover what the user changed, not two server responses that arrive in a + * different order than their requests were made. The fields of a [FieldGroup] written from a + * later request keep their stored values. [ownUserId] is the user of the account, for the + * unread count that has to be derived again; without it the count of the response is taken. */ - fun preservePendingLocalState( + suspend fun preservePendingLocalState( previousConversations: Map, - conversationsFromServer: List + conversationsFromServer: List, + requestedAtNanos: Long, + ownUserId: String? ): List = - conversationsFromServer.map { serverItem -> + conversationsFromServer.mapNotNull { serverItem -> + if (isSuperseded(serverItem.internalId, requestedAtNanos)) { + Log.d( + TAG, + "Dropping a response for room ${serverItem.token}: requested at $requestedAtNanos, " + + "older than the stored entry written from a request at " + + "${serverWriteStarts[serverItem.internalId]}" + ) + return@mapNotNull null + } val previous = previousConversations[serverItem.internalId] - ?: return@map serverItem + ?: return@mapNotNull serverItem var guarded = guardPendingReadMarker(serverItem, previous) guarded = guardPendingUnread(guarded, previous) guarded = guardPendingFavorite(guarded, previous) @@ -300,7 +441,27 @@ class ConversationListUpdater @Inject constructor( guarded = guardPendingHiddenPinnedId(guarded, previous) guarded = guardPendingArchived(guarded, previous) guarded = guardPendingTags(guarded, previous) - guarded + keepNewerFieldGroups(guarded, previous, requestedAtNanos, ownUserId) + } + + private suspend fun keepNewerFieldGroups( + serverItem: ConversationEntity, + previous: ConversationEntity, + requestedAtNanos: Long, + ownUserId: String? + ): ConversationEntity = + FieldGroup.entries.fold(serverItem) { item, group -> + val writtenAt = fieldGroupWriteStarts[item.internalId to group] + if (isNewer(writtenAt, requestedAtNanos)) { + Log.d( + TAG, + "Keeping $group of room ${item.token}: requested at $requestedAtNanos, older than " + + "its write from a request at $writtenAt" + ) + keepStored(group, item, previous, ownUserId) + } else { + item + } } private fun guardPendingReadMarker( @@ -415,7 +576,10 @@ class ConversationListUpdater @Inject constructor( * Concurrency: runs inside the per-room catch-up mutex and writes via a single guarded UPDATE * that only applies when the derived state is newer than the stored one, so a concurrently * finishing room list sync can neither be overwritten with older data nor interleave with a - * read-modify-write. Thread catch-ups don't describe the room itself and are skipped. + * read-modify-write. It also holds [withServerWriteLock] and records its write as + * [FieldGroup.PREVIEW], from the start of the request that fetched the message, so a response + * requested before that request cannot revert these fields. Thread catch-ups don't describe the + * room itself and are skipped. * * Unlike [preservePendingLocalState], this path doesn't merge against a server response, so a * pending read marker can't be guarded the same way — but the marker is still consulted as the @@ -435,26 +599,40 @@ class ConversationListUpdater @Inject constructor( if (target.threadId != null || !outcome.persistedNewMessages || newestMessage == null) { return } - val conversation = - conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return + withServerWriteLock(target.accountId) { + val conversation = + conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() + ?: return@withServerWriteLock - val lastReadMessage = maxOf( - conversation.lastReadMessage, - pendingReadMarker(target.internalConversationId) ?: 0 - ) - val unreadMessages = deriveUnreadMessagesCount(target, lastReadMessage) - val updatedRows = conversationsDao.updateConversationFromCatchUp( - internalId = conversation.internalId, - lastMessageJson = LoganSquare.serialize(newestMessage), - lastActivity = newestMessage.timestamp, - unreadMessages = unreadMessages - ) - Log.d( - TAG, - "Conversation list update for room ${target.roomToken} from catch-up " + - "(lastActivity=${newestMessage.timestamp}, unread=$unreadMessages): " + - if (updatedRows > 0) "applied" else "skipped, stored state is newer" - ) + val lastReadMessage = maxOf( + conversation.lastReadMessage, + pendingReadMarker(target.internalConversationId) ?: 0 + ) + val unreadMessages = + deriveUnreadMessagesCount(target.internalConversationId, target.user.userId!!, lastReadMessage) + // the write and its mark are one step even if the caller gives up meanwhile + val updatedRows = withContext(NonCancellable) { + conversationsDao.updateConversationFromCatchUp( + internalId = conversation.internalId, + lastMessageJson = LoganSquare.serialize(newestMessage), + lastActivity = newestMessage.timestamp, + unreadMessages = unreadMessages + ).also { rows -> + // a response requested before the request of the message can still lack it + if (rows > 0) { + outcome.newestMessageRequestedAtNanos?.let { + recordServerWrite(listOf(conversation.internalId), it, FieldGroup.PREVIEW) + } + } + } + } + Log.d( + TAG, + "Conversation list update for room ${target.roomToken} from catch-up " + + "(lastActivity=${newestMessage.timestamp}, unread=$unreadMessages): " + + if (updatedRows > 0) "applied" else "skipped, stored state is newer" + ) + } } /** @@ -463,9 +641,13 @@ class ConversationListUpdater @Inject constructor( * the last read message. Own messages don't count: sending from another device advances the * server-side read marker, which the locally cached [lastReadMessage] may lag behind. */ - private suspend fun deriveUnreadMessagesCount(target: ChatMessageSyncer.SyncTarget, lastReadMessage: Int): Int { + private suspend fun deriveUnreadMessagesCount( + internalConversationId: String, + ownUserId: String, + lastReadMessage: Int + ): Int { val latestBlock = if (lastReadMessage > 0) { - chatBlocksDao.getLatestChatBlock(target.internalConversationId, null).first() + chatBlocksDao.getLatestChatBlock(internalConversationId, null).first() } else { null } @@ -474,9 +656,9 @@ class ConversationListUpdater @Inject constructor( return if (blockReachesUnreadBoundary) { chatDao.countMessagesNewerThan( - internalConversationId = target.internalConversationId, + internalConversationId = internalConversationId, messageId = lastReadMessage.toLong(), - excludedActorId = target.user.userId!! + excludedActorId = ownUserId ) } else { UNREAD_COUNT_UNKNOWN diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt index a9cd035e0ef..592cd10305f 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt @@ -36,6 +36,7 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow @@ -142,19 +143,33 @@ class OfflineFirstConversationsRepository @Inject constructor( override fun getRoom(user: User, roomToken: String): Job = scope.launch { try { + val requestedAtNanos = System.nanoTime() val model = chatNetworkDataSource.getRoom(user, roomToken) val existingEntity = dao.getConversationForUser(user.id!!, model.token).first() model.hiddenUpcomingEvent = existingEntity?.hiddenUpcomingEvent _conversationFlow.emit(model) - val previous = existingEntity?.let { mapOf(it.internalId to it) }.orEmpty() - val entityList = conversationListUpdater.preservePendingLocalState( - previous, - listOf(model.asEntity()) - ) - try { - dao.upsertConversations(user.id!!, entityList) - } catch (e: SQLiteConstraintException) { - Log.w(TAG, "Skipping conversation upsert for removed account ${user.id}", e) + // the entry is read again under the lock: a sync may have written it since + conversationListUpdater.withServerWriteLock(user.id!!) { + val previous = dao.getConversationForUser(user.id!!, model.token).first() + ?.let { mapOf(it.internalId to it) }.orEmpty() + val entityList = conversationListUpdater.preservePendingLocalState( + previous, + listOf(model.asEntity()), + requestedAtNanos, + user.userId + ) + try { + // the write and its mark are one step even if the caller gives up meanwhile + withContext(NonCancellable) { + dao.upsertConversations(user.id!!, entityList) + conversationListUpdater.recordServerWrite( + entityList.map { it.internalId }, + requestedAtNanos + ) + } + } catch (e: SQLiteConstraintException) { + Log.w(TAG, "Skipping conversation upsert for removed account ${user.id}", e) + } } } catch (e: Exception) { // In case network is offline, the call fails, or getRoom can't resolve a supported @@ -254,12 +269,14 @@ class OfflineFirstConversationsRepository @Inject constructor( // not something the response can be read as having left out val knownBeforeRequest = dao.getConversationIdsForUser(accountId).toSet() + var requestedAtNanos = 0L val roomList = withRetry( retries = NETWORK_FETCH_RETRIES, initialDelayMillis = NETWORK_FETCH_RETRY_INITIAL_DELAY_MS, maxDelayMillis = NETWORK_FETCH_RETRY_MAX_DELAY_MS, retryOn = { !it.isUnauthorized() } ) { + requestedAtNanos = System.nanoTime() network.getRooms(user, user.baseUrl!!, includeStatus, modifiedSince) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) @@ -270,28 +287,44 @@ class OfflineFirstConversationsRepository @Inject constructor( it.asEntity(accountId) } - val previousConversations = dao.getConversationsForUser(accountId).first() - .associateBy { it.internalId } + val previousConversations = conversationListUpdater.withServerWriteLock(accountId) { + val stored = dao.getConversationsForUser(accountId).first() + .associateBy { it.internalId } - val serverItems = if (includeStatus) { - conversationsFromSync - } else { - keepCachedStatus(conversationsFromSync, previousConversations) - } + val serverItems = if (includeStatus) { + conversationsFromSync + } else { + keepCachedStatus(conversationsFromSync, stored) + } - dao.syncConversationsForUser( - accountId = accountId, - serverItems = conversationListUpdater.preservePendingLocalState( - previousConversations, - serverItems - ), - conversationIdsToDelete = if (roomList.wasDelta) { + // a conversation that holds state from a later request is not one this response can delete + val conversationIdsToDelete = if (roomList.wasDelta) { emptyList() } else { - determineLeftConversationIds(previousConversations, conversationsFromSync) + determineLeftConversationIds(stored, conversationsFromSync) .filter { it in knownBeforeRequest } + .filterNot { conversationListUpdater.hasNewerWrite(it, requestedAtNanos) } } - ) + val mergedItems = conversationListUpdater.preservePendingLocalState( + stored, + serverItems, + requestedAtNanos, + user.userId + ) + // the write and its mark are one step even if the worker's time budget runs out meanwhile + withContext(NonCancellable) { + dao.syncConversationsForUser( + accountId = accountId, + serverItems = mergedItems, + conversationIdsToDelete = conversationIdsToDelete + ) + conversationListUpdater.recordServerWrite( + mergedItems.map { it.internalId } + conversationIdsToDelete, + requestedAtNanos + ) + } + stored + } rememberSyncedState(accountId, roomList) diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt index d5f86bc33fd..279f28893a3 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt @@ -35,22 +35,31 @@ import com.nextcloud.talk.utils.ApiUtils import io.reactivex.Observable import io.reactivex.android.plugins.RxAndroidPlugins import io.reactivex.schedulers.Schedulers +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch +import kotlinx.coroutines.runInterruptible import kotlinx.coroutines.runBlocking +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull import org.junit.Assert.assertTrue import org.junit.Before import org.junit.Test import org.junit.runner.RunWith import org.mockito.kotlin.any import org.mockito.kotlin.anyOrNull +import org.mockito.kotlin.doAnswer +import org.mockito.kotlin.doSuspendableAnswer import org.mockito.kotlin.mock +import org.mockito.kotlin.spy import org.mockito.kotlin.whenever import org.mockito.kotlin.wheneverBlocking import org.robolectric.RobolectricTestRunner @@ -65,7 +74,7 @@ import retrofit2.Response */ @RunWith(RobolectricTestRunner::class) @Config(application = Application::class, sdk = [33]) -@Suppress("TooManyFunctions") +@Suppress("TooManyFunctions", "LargeClass") class ConversationListFreshnessIntegrationTest { private lateinit var db: TalkDatabase @@ -364,6 +373,497 @@ class ConversationListFreshnessIntegrationTest { } } + @Test + fun `a room refresh requested before a sync cannot revert the entry the sync wrote`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val refreshRequested = CompletableDeferred() + val refreshAnswer = CompletableDeferred() + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.doSuspendableAnswer { + refreshRequested.complete(Unit) + refreshAnswer.await() + ConversationModel.mapToConversationModel(serverRoom(lastActivity = 12, lastRead = 10, unread = 2), user) + } + val refresh = repository.getRoom(user, ROOM_TOKEN) + refreshRequested.await() + + val roomWithCall = serverRoom(lastActivity = 20, lastRead = 20, unread = 0).apply { hasCall = true } + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())) + .thenReturn(roomList(listOf(roomWithCall))) + repository.getRooms(user).join() + refreshAnswer.complete(Unit) + refresh.join() + + val conversation = conversationEntity() + assertEquals("the older answer must not revert the activity", 20L, conversation.lastActivity) + assertEquals("the older answer must not revert the read marker", 20, conversation.lastReadMessage) + assertEquals("the older answer must not bring back the unread count", 0, conversation.unreadMessages) + assertTrue("the older answer must not drop the call", conversation.hasCall) + } + } + + @Test + fun `a sync requested before a room refresh cannot revert the entry the refresh wrote`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val syncRequested = CountDownLatch(1) + val syncAnswer = CountDownLatch(1) + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())).thenReturn( + Observable.fromCallable { + syncRequested.countDown() + syncAnswer.await() + RoomListResult( + listOf(serverRoom(lastActivity = 12, lastRead = 10, unread = 2)), + modifiedBefore = null, + wasDelta = false + ) + } + ) + val sync = repository.getRooms(user) + runInterruptible { syncRequested.await() } + + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenReturn( + ConversationModel.mapToConversationModel( + serverRoom(lastActivity = 20, lastRead = 20, unread = 0).apply { hasArchived = true }, + user + ) + ) + repository.getRoom(user, ROOM_TOKEN).join() + syncAnswer.countDown() + sync.join() + + val conversation = conversationEntity() + assertEquals("the older sync must not revert the activity", 20L, conversation.lastActivity) + assertEquals("the older sync must not revert the read marker", 20, conversation.lastReadMessage) + assertEquals("the older sync must not bring back the unread count", 0, conversation.unreadMessages) + assertTrue("the older sync must not revert the archive flag", conversation.hasArchived) + } + } + + @Test + fun `a sync requested before a catch-up cannot revert the message the catch-up stored`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val syncRequested = CountDownLatch(1) + val syncAnswer = CountDownLatch(1) + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())).thenReturn( + Observable.fromCallable { + syncRequested.countDown() + syncAnswer.await() + RoomListResult( + listOf(serverRoom(lastActivity = 10, lastRead = 10, unread = 0)), + modifiedBefore = null, + wasDelta = false + ) + } + ) + val sync = repository.getRooms(user) + runInterruptible { syncRequested.await() } + + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(10), message(11), message(12)))) + syncer.catchUpRoom(target(), lastReadMessage = 10, unreadMessages = 2) + syncAnswer.countDown() + sync.join() + + val conversation = conversationEntity() + assertEquals("the older sync must not revert the activity", 12L, conversation.lastActivity) + assertEquals("the older sync must not revert the unread count", 2, conversation.unreadMessages) + assertEquals(12L, LoganSquare.parse(conversation.lastMessage, ChatMessageDto::class.java).id) + } + } + + @Test + fun `a full sync requested before a room was refreshed does not delete it`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val syncRequested = CountDownLatch(1) + val syncAnswer = CountDownLatch(1) + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())).thenReturn( + Observable.fromCallable { + syncRequested.countDown() + syncAnswer.await() + RoomListResult( + listOf(serverRoom(lastActivity = 10, lastRead = 10, unread = 0)), + modifiedBefore = null, + wasDelta = false + ) + } + ) + val sync = repository.getRooms(user) + runInterruptible { syncRequested.await() } + + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenReturn( + ConversationModel.mapToConversationModel( + serverRoom(token = JOINED_ROOM_TOKEN, lastActivity = 15, lastRead = 15, unread = 0), + user + ) + ) + repository.getRoom(user, JOINED_ROOM_TOKEN).join() + syncAnswer.countDown() + sync.join() + + assertEquals( + "a room the sync could not know about must survive it", + 15L, + db.conversationsDao().getConversationForUser(ACCOUNT_ID, JOINED_ROOM_TOKEN).first()?.lastActivity + ) + } + } + + @Test + fun `a room refresh requested before a full sync does not bring back a room the sync deleted`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val refreshRequested = CompletableDeferred() + val refreshAnswer = CompletableDeferred() + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.doSuspendableAnswer { + refreshRequested.complete(Unit) + refreshAnswer.await() + ConversationModel.mapToConversationModel(serverRoom(lastActivity = 10, lastRead = 10, unread = 0), user) + } + val refresh = repository.getRoom(user, ROOM_TOKEN) + refreshRequested.await() + + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())).thenReturn( + roomList(listOf(serverRoom(token = JOINED_ROOM_TOKEN, lastActivity = 15, lastRead = 15, unread = 0))) + ) + repository.getRooms(user).join() + assertNull(db.conversationsDao().getConversationForUser(ACCOUNT_ID, ROOM_TOKEN).first()) + refreshAnswer.complete(Unit) + refresh.join() + + assertNull( + "an answer older than the deletion must not bring the room back", + db.conversationsDao().getConversationForUser(ACCOUNT_ID, ROOM_TOKEN).first() + ) + } + } + + @Test + fun `a room refresh requested after the catch-up request is applied in full`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val catchUpRequested = CompletableDeferred() + val catchUpAnswer = CompletableDeferred() + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) }.doSuspendableAnswer { + catchUpRequested.complete(Unit) + catchUpAnswer.await() + Response.success(overall(message(10), message(11), message(12))) + } + val catchUp = launch(Dispatchers.IO) { + syncer.catchUpRoom(target(), lastReadMessage = 10, unreadMessages = 2) + } + catchUpRequested.await() + + val refreshRequested = CompletableDeferred() + val refreshAnswer = CompletableDeferred() + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.doSuspendableAnswer { + refreshRequested.complete(Unit) + refreshAnswer.await() + ConversationModel.mapToConversationModel( + serverRoom(lastActivity = 12, lastRead = 10, unread = 2).apply { hasCall = true }, + user + ) + } + val refresh = repository.getRoom(user, ROOM_TOKEN) + refreshRequested.await() + catchUpAnswer.complete(Unit) + catchUp.join() + refreshAnswer.complete(Unit) + refresh.join() + + assertTrue("an answer requested after the catch-up request must apply", conversationEntity().hasCall) + } + } + + @Test + fun `a room refresh requested before the catch-up request keeps the fields of the catch-up only`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val refreshRequested = CompletableDeferred() + val refreshAnswer = CompletableDeferred() + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.doSuspendableAnswer { + refreshRequested.complete(Unit) + refreshAnswer.await() + ConversationModel.mapToConversationModel( + serverRoom(lastActivity = 10, lastRead = 10, unread = 0).apply { hasCall = true }, + user + ) + } + val refresh = repository.getRoom(user, ROOM_TOKEN) + refreshRequested.await() + + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(10), message(11), message(12)))) + syncer.catchUpRoom(target(), lastReadMessage = 10, unreadMessages = 2) + refreshAnswer.complete(Unit) + refresh.join() + + val conversation = conversationEntity() + assertEquals("the older answer must not revert the activity", 12L, conversation.lastActivity) + assertEquals("the older answer must not revert the unread count", 2, conversation.unreadMessages) + assertEquals(12L, LoganSquare.parse(conversation.lastMessage, ChatMessageDto::class.java).id) + assertTrue("the rest of the answer is newer than the entry and applies", conversation.hasCall) + } + } + + @Test + fun `a delta sync requested before a room refresh drops that room only`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val syncRequested = CountDownLatch(1) + val syncAnswer = CountDownLatch(1) + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())).thenReturn( + Observable.fromCallable { + syncRequested.countDown() + syncAnswer.await() + RoomListResult( + listOf( + serverRoom(lastActivity = 12, lastRead = 10, unread = 2), + serverRoom(token = JOINED_ROOM_TOKEN, lastActivity = 5, lastRead = 5, unread = 0) + ), + modifiedBefore = null, + wasDelta = true + ) + } + ) + val sync = repository.getRooms(user) + runInterruptible { syncRequested.await() } + + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenReturn( + ConversationModel.mapToConversationModel(serverRoom(lastActivity = 20, lastRead = 20, unread = 0), user) + ) + repository.getRoom(user, ROOM_TOKEN).join() + syncAnswer.countDown() + sync.join() + + assertEquals("the dropped room keeps the newer entry", 20L, conversationEntity().lastActivity) + assertEquals( + "the other rooms of the delta are written", + 5L, + db.conversationsDao().getConversationForUser(ACCOUNT_ID, JOINED_ROOM_TOKEN).first()?.lastActivity + ) + } + } + + @Test + fun `accounts of one server do not order each other's responses`() { + val user = user(withKeepNotificationsCapability = false) + val otherUser = user.copy(id = OTHER_ACCOUNT_ID, userId = "other", username = "other") + val repository = repository() + runBlocking { + db.usersDao().saveUser( + UserEntity(id = OTHER_ACCOUNT_ID, userId = "other", username = "other", baseUrl = BASE_URL) + ) + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val refreshRequested = CompletableDeferred() + val refreshAnswer = CompletableDeferred() + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.doSuspendableAnswer { + refreshRequested.complete(Unit) + refreshAnswer.await() + ConversationModel.mapToConversationModel( + serverRoom(lastActivity = 12, lastRead = 10, unread = 2), + otherUser + ) + } + val refresh = repository.getRoom(otherUser, ROOM_TOKEN) + refreshRequested.await() + + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())) + .thenReturn(roomList(listOf(serverRoom(lastActivity = 20, lastRead = 20, unread = 0)))) + repository.getRooms(user).join() + refreshAnswer.complete(Unit) + refresh.join() + + assertEquals(20L, conversationEntity().lastActivity) + assertEquals( + "an answer for the other account is not older than the entry of this one", + 12L, + db.conversationsDao().getConversationForUser(OTHER_ACCOUNT_ID, ROOM_TOKEN).first()?.lastActivity + ) + } + } + + @Test + fun `a room refresh waits for the write of a sync it is judged against`() { + val user = user(withKeepNotificationsCapability = false) + val mergeReached = CountDownLatch(1) + val refreshDone = CountDownLatch(1) + val updater = spy(conversationListUpdater) + val firstMerge = AtomicBoolean(true) + // the sync is held after it judged its response and before it writes it, the window the lock exists for + doAnswer { invocation -> + val superseded = invocation.callRealMethod() + if (firstMerge.getAndSet(false)) { + mergeReached.countDown() + refreshDone.await(LOCK_WAIT_MILLIS, TimeUnit.MILLISECONDS) + } + superseded + }.whenever(updater).isSuperseded(any(), any()) + val repository = OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + updater, + ArbitraryStorageManager(ArbitraryStoragesRepositoryImpl(db.arbitraryStoragesDao())), + ApplicationProvider.getApplicationContext(), + mock() + ) + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + // a delta sync, so that its first judgement is the one of the merge + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())).thenReturn( + Observable.just( + RoomListResult( + listOf(serverRoom(lastActivity = 12, lastRead = 10, unread = 2)), + modifiedBefore = null, + wasDelta = true + ) + ) + ) + val sync = repository.getRooms(user) + runInterruptible { mergeReached.await() } + + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenReturn( + ConversationModel.mapToConversationModel(serverRoom(lastActivity = 20, lastRead = 20, unread = 0), user) + ) + val refresh = repository.getRoom(user, ROOM_TOKEN) + launch(Dispatchers.IO) { + refresh.join() + refreshDone.countDown() + } + sync.join() + refresh.join() + + assertEquals( + "the refresh was requested after the sync, so its write must come last", + 20L, + conversationEntity().lastActivity + ) + } + } + + @Test + fun `a catch-up during a refresh derives unread again from the marker of the answer`() { + runBlocking { + // messages 11 and 12 were read on another device before the refresh was requested + refreshAcrossCatchUp(storedLastRead = 10, answer = serverRoom(lastActivity = 12, lastRead = 12, unread = 0)) + + val conversation = conversationEntity() + assertEquals(12, conversation.lastReadMessage) + assertEquals("the count follows the marker of the answer", 0, conversation.unreadMessages) + assertEquals("the preview stays the one of the catch-up", 12L, conversation.lastActivity) + } + } + + @Test + fun `a catch-up during a refresh follows a chat marked as unread elsewhere`() { + runBlocking { + refreshAcrossCatchUp(storedLastRead = 12, answer = serverRoom(lastActivity = 12, lastRead = 10, unread = 2)) + + val conversation = conversationEntity() + assertEquals(10, conversation.lastReadMessage) + assertEquals("the count follows the marker of the answer", 2, conversation.unreadMessages) + } + } + + @Test + fun `a catch-up during a refresh keeps its unread count while the marker is the same`() { + runBlocking { + refreshAcrossCatchUp(storedLastRead = 10, answer = serverRoom(lastActivity = 10, lastRead = 10, unread = 0)) + + val conversation = conversationEntity() + assertEquals(10, conversation.lastReadMessage) + assertEquals("the older answer cannot know the new messages", 2, conversation.unreadMessages) + } + } + + @Test + fun `a lowered marker with an unknown count keeps the higher of the stored and the answered count`() { + runBlocking { + // the cached messages start above the marker of the answer, so the count cannot be derived + refreshAcrossCatchUp( + storedLastRead = 12, + answer = serverRoom(lastActivity = 15, lastRead = 10, unread = 2), + storedUnread = 3, + fetched = listOf(14, 15) + ) + + val conversation = conversationEntity() + assertEquals(10, conversation.lastReadMessage) + assertEquals("both counts are lower bounds", 3, conversation.unreadMessages) + } + } + + /** + * Stores a room, requests a refresh, lets a catch-up that fetches messages 10 to 12 write the room + * while the refresh is under way, and then lets the refresh answer with [answer]. + */ + private suspend fun refreshAcrossCatchUp( + storedLastRead: Int, + answer: ConversationDto, + storedUnread: Int = 0, + fetched: List = listOf(10, 11, 12) + ) { + val user = user(withKeepNotificationsCapability = false) + seedConversation(lastActivity = 10, lastReadMessage = storedLastRead, unreadMessages = storedUnread) + val refreshRequested = CompletableDeferred() + val refreshAnswer = CompletableDeferred() + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.doSuspendableAnswer { + refreshRequested.complete(Unit) + refreshAnswer.await() + ConversationModel.mapToConversationModel(answer, user) + } + val refresh = repository().getRoom(user, ROOM_TOKEN) + refreshRequested.await() + + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(ChatOverall(ocs = ChatOCS(meta = null, data = fetched.map { message(it) })))) + syncer.catchUpRoom(target(), lastReadMessage = storedLastRead, unreadMessages = storedUnread) + refreshAnswer.complete(Unit) + refresh.join() + } + + @Test + fun `a response requested after the last write applies unchanged`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenReturn( + ConversationModel.mapToConversationModel(serverRoom(lastActivity = 20, lastRead = 20, unread = 0), user) + ) + repository.getRoom(user, ROOM_TOKEN).join() + + // the server moved on, e.g. the chat was marked as unread on another device + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())) + .thenReturn(roomList(listOf(serverRoom(lastActivity = 20, lastRead = 15, unread = 5)))) + repository.getRooms(user).join() + + assertEquals(15, conversationEntity().lastReadMessage) + assertEquals(5, conversationEntity().unreadMessages) + } + } + @Test fun `room list flow reflects database writes reactively`() { val user = user(withKeepNotificationsCapability = false) @@ -426,6 +926,19 @@ class ConversationListFreshnessIntegrationTest { mock() ) + private fun serverRoom( + lastActivity: Long, + lastRead: Int, + unread: Int, + token: String = ROOM_TOKEN + ): ConversationDto = + ConversationDto( + token = token, + lastActivity = lastActivity, + lastReadMessage = lastRead, + unreadMessages = unread + ) + private fun staleServerRoom( lastReadMessage: Int, unreadMessages: Int, @@ -518,9 +1031,11 @@ class ConversationListFreshnessIntegrationTest { private const val OTHER_ACCOUNT_ID = 2L private const val BASE_URL = "https://server.example.com" private const val ROOM_TOKEN = "room1" + private const val JOINED_ROOM_TOKEN = "room2" private const val INTERNAL_CONVERSATION_ID = "$ACCOUNT_ID@$ROOM_TOKEN" private const val TIMEOUT_MILLIS = 10_000L private const val POLL_INTERVAL_MILLIS = 50L + private const val LOCK_WAIT_MILLIS = 250L } } diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterPinTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterPinTest.kt index 18bdc199183..3aaa2f813fe 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterPinTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterPinTest.kt @@ -109,16 +109,20 @@ class ConversationListUpdaterPinTest { assertNull(preserveHidden(serverHiddenId = null).hiddenPinnedId) } - private fun preserveHidden(serverHiddenId: Long?): ConversationEntity = + private suspend fun preserveHidden(serverHiddenId: Long?): ConversationEntity = updater.preservePendingLocalState( previousConversations = mapOf(stored.internalId to stored), - conversationsFromServer = listOf(stored.copy(hiddenPinnedId = serverHiddenId)) + conversationsFromServer = listOf(stored.copy(hiddenPinnedId = serverHiddenId)), + requestedAtNanos = System.nanoTime(), + ownUserId = "me" ).single() - private fun preserve(serverPinnedId: Long?): ConversationEntity = + private suspend fun preserve(serverPinnedId: Long?): ConversationEntity = updater.preservePendingLocalState( previousConversations = mapOf(stored.internalId to stored), - conversationsFromServer = listOf(stored.copy(lastPinnedId = serverPinnedId)) + conversationsFromServer = listOf(stored.copy(lastPinnedId = serverPinnedId)), + requestedAtNanos = System.nanoTime(), + ownUserId = "me" ).single() private fun givenCachedConversation(pinnedId: Long?) { diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterUnreadTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterUnreadTest.kt index 8a016090fba..a68f0195a5b 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterUnreadTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterUnreadTest.kt @@ -118,12 +118,14 @@ class ConversationListUpdaterUnreadTest { assertEquals(0, readAgain.unreadMessages) } - private fun preserve(serverLastRead: Int, serverUnread: Int): ConversationEntity = + private suspend fun preserve(serverLastRead: Int, serverUnread: Int): ConversationEntity = updater.preservePendingLocalState( previousConversations = mapOf(stored.internalId to stored), conversationsFromServer = listOf( stored.copy(lastReadMessage = serverLastRead, unreadMessages = serverUnread) - ) + ), + requestedAtNanos = System.nanoTime(), + ownUserId = "me" ).single() private fun givenCachedMessagesNewerThanMarker(count: Int) { diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt index 54393008112..438c433c84b 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt @@ -35,6 +35,7 @@ import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking @@ -55,7 +56,6 @@ import org.mockito.kotlin.doSuspendableAnswer import org.mockito.kotlin.eq import org.mockito.kotlin.mock import org.mockito.kotlin.never -import org.mockito.kotlin.stub import org.mockito.kotlin.times import org.mockito.kotlin.verify import org.mockito.kotlin.verifyBlocking @@ -111,8 +111,10 @@ class OfflineFirstConversationsRepositoryTest { stubStoredConversations() whenever(arbitraryStorageManager.getStorageSetting(any(), any(), any())) .thenReturn(Maybe.empty()) - whenever(conversationListUpdater.preservePendingLocalState(any(), any())) + wheneverBlocking { conversationListUpdater.preservePendingLocalState(any(), any(), any(), any()) } .thenAnswer { invocation -> invocation.getArgument>(1) } + wheneverBlocking { conversationListUpdater.withServerWriteLock(any(), any()) } + .doSuspendableAnswer { invocation -> invocation.getArgument Any?>(1).invoke() } repository = OfflineFirstConversationsRepository( dao, @@ -231,9 +233,8 @@ class OfflineFirstConversationsRepositoryTest { runTest { val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 2) whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(room))) - dao.stub { - onBlocking { syncConversationsForUser(any(), any(), any()) } doSuspendableAnswer { awaitCancellation() } - } + // a write already under way is not interrupted by the budget, so the sync is held before it + whenever(dao.getConversationsForUser(ACCOUNT_ID)).thenReturn(flow { awaitCancellation() }) assertNull(repository.syncRooms(user(), roomListTimeoutMillis = ROOM_LIST_TIMEOUT_MILLIS)) verifyBlocking(chatMessageSyncer, never()) { catchUpRoom(any(), any(), anyOrNull(), any()) } @@ -374,7 +375,9 @@ class OfflineFirstConversationsRepositoryTest { val previousCaptor = argumentCaptor>() val serverCaptor = argumentCaptor>() - verify(conversationListUpdater).preservePendingLocalState(previousCaptor.capture(), serverCaptor.capture()) + verifyBlocking(conversationListUpdater) { + preservePendingLocalState(previousCaptor.capture(), serverCaptor.capture(), any(), any()) + } assertEquals(mapOf(previous.internalId to previous), previousCaptor.firstValue) assertEquals(6L, serverCaptor.firstValue.first().lastActivity) } From a2233914530e451da9ebbcc05ace9b218245a630 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Sun, 11 Oct 2026 18:28:06 +0000 Subject: [PATCH 2/2] fix(conversations): do not apply a catch-up older than the stored entry A catch-up is a writer of the conversation entry like a sync or a room refresh, but it was not ordered against them by the start of its request. When a room refresh requested after the catch-up request was written first, the catch-up still ran its UPDATE, which compares only the last activity. A newer response can have a lower activity, for example after the newest message was deleted, and the catch-up then brought back the deleted preview and the unread count. Under the account lock, skip the catch-up when the entry holds state from a request started after the one that fetched its message, with the same isSuperseded check that drops an older response. Assisted-by: Claude-Code:claude-sonnet-5-5 Signed-off-by: Krainov Gleb --- .../data/network/ConversationListUpdater.kt | 11 ++++-- ...onversationListFreshnessIntegrationTest.kt | 34 +++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt index 7b1c2ded859..9966801138f 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt @@ -578,8 +578,10 @@ class ConversationListUpdater @Inject constructor( * finishing room list sync can neither be overwritten with older data nor interleave with a * read-modify-write. It also holds [withServerWriteLock] and records its write as * [FieldGroup.PREVIEW], from the start of the request that fetched the message, so a response - * requested before that request cannot revert these fields. Thread catch-ups don't describe the - * room itself and are skipped. + * requested before that request cannot revert these fields. In turn, a catch-up whose request + * was started before the one behind the stored entry is [isSuperseded] and not applied: the + * activity guard above cannot see that, as a newer response can have a lower activity, e.g. + * after a message was deleted. Thread catch-ups don't describe the room itself and are skipped. * * Unlike [preservePendingLocalState], this path doesn't merge against a server response, so a * pending read marker can't be guarded the same way — but the marker is still consulted as the @@ -603,6 +605,11 @@ class ConversationListUpdater @Inject constructor( val conversation = conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return@withServerWriteLock + // the message was requested before the request of the stored entry, so the entry is newer + if (outcome.newestMessageRequestedAtNanos?.let { isSuperseded(conversation.internalId, it) } == true) { + Log.d(TAG, "Skipping a catch-up for room ${target.roomToken}: older than the stored entry") + return@withServerWriteLock + } val lastReadMessage = maxOf( conversation.lastReadMessage, diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt index 279f28893a3..60b3a25abc4 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt @@ -479,6 +479,40 @@ class ConversationListFreshnessIntegrationTest { } } + @Test + fun `a catch-up requested before a room refresh cannot revert the entry the refresh wrote`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + val catchUpRequested = CompletableDeferred() + val catchUpAnswer = CompletableDeferred() + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) }.doSuspendableAnswer { + catchUpRequested.complete(Unit) + catchUpAnswer.await() + Response.success(overall(message(10), message(11), message(12))) + } + val catchUp = launch { syncer.catchUpRoom(target(), lastReadMessage = 10, unreadMessages = 2) } + catchUpRequested.await() + + // message 12 was deleted meanwhile, so the newer answer has a lower activity than the catch-up + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenReturn( + ConversationModel.mapToConversationModel( + serverRoom(lastActivity = 11, lastRead = 11, unread = 0), + user + ) + ) + repository.getRoom(user, ROOM_TOKEN).join() + catchUpAnswer.complete(Unit) + catchUp.join() + + val conversation = conversationEntity() + assertEquals("the older catch-up must not revert the activity", 11L, conversation.lastActivity) + assertEquals("the older catch-up must not bring back the unread count", 0, conversation.unreadMessages) + assertEquals(11, conversation.lastReadMessage) + } + } + @Test fun `a full sync requested before a room was refreshed does not delete it`() { val user = user(withKeepNotificationsCapability = false)