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 c8fb74f8de1..da6eb2ca1df 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. @@ -302,14 +424,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) @@ -317,7 +458,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( @@ -432,7 +593,12 @@ 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. 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 @@ -452,26 +618,45 @@ 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 + // 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, - 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" + ) + } } /** @@ -480,9 +665,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 } @@ -491,9 +680,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..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 @@ -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,531 @@ 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 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) + 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 +960,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 +1065,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 60fff4fcbde..a6eda94ca2a 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 @@ -120,42 +120,45 @@ class ConversationListUpdaterUnreadTest { } @Test - fun `marking read releases a pending unread, so the server's read state applies`() { - stored = stored.copy(lastReadMessage = MARKER, unreadMessages = 3) - updater.markPendingUnread(INTERNAL_CONVERSATION_ID) + fun `marking read releases a pending unread, so the server's read state applies`() = + runTest { + stored = stored.copy(lastReadMessage = MARKER, unreadMessages = 3) + updater.markPendingUnread(INTERNAL_CONVERSATION_ID) - updater.switchToPendingMarkAsRead(INTERNAL_CONVERSATION_ID, READ_UP_TO_NEWEST) + updater.switchToPendingMarkAsRead(INTERNAL_CONVERSATION_ID, READ_UP_TO_NEWEST) - assertEquals(READ_UP_TO_NEWEST, updater.pendingReadMarker(INTERNAL_CONVERSATION_ID)) - val read = preserve(serverLastRead = READ_UP_TO_NEWEST, serverUnread = 0) - assertEquals(READ_UP_TO_NEWEST, read.lastReadMessage) - assertEquals(0, read.unreadMessages) - } + assertEquals(READ_UP_TO_NEWEST, updater.pendingReadMarker(INTERNAL_CONVERSATION_ID)) + val read = preserve(serverLastRead = READ_UP_TO_NEWEST, serverUnread = 0) + assertEquals(READ_UP_TO_NEWEST, read.lastReadMessage) + assertEquals(0, read.unreadMessages) + } @Test - fun `marking read without a marker, as in a federated conversation, still releases a pending unread`() { - stored = stored.copy(lastReadMessage = MARKER, unreadMessages = 3) - updater.markPendingUnread(INTERNAL_CONVERSATION_ID) + fun `marking read without a marker, as in a federated conversation, still releases a pending unread`() = + runTest { + stored = stored.copy(lastReadMessage = MARKER, unreadMessages = 3) + updater.markPendingUnread(INTERNAL_CONVERSATION_ID) - updater.switchToPendingMarkAsRead(INTERNAL_CONVERSATION_ID, lastReadMessage = null) + updater.switchToPendingMarkAsRead(INTERNAL_CONVERSATION_ID, lastReadMessage = null) - assertNull(updater.pendingReadMarker(INTERNAL_CONVERSATION_ID)) - assertEquals(0, preserve(serverLastRead = READ_UP_TO_NEWEST, serverUnread = 0).unreadMessages) - } + assertNull(updater.pendingReadMarker(INTERNAL_CONVERSATION_ID)) + assertEquals(0, preserve(serverLastRead = READ_UP_TO_NEWEST, serverUnread = 0).unreadMessages) + } @Test - fun `marking unread drops a pending read marker and guards the unread state instead`() { - stored = stored.copy(lastReadMessage = MARKER, unreadMessages = 3) - updater.markPendingReadMarker(INTERNAL_CONVERSATION_ID, READ_UP_TO_NEWEST) + fun `marking unread drops a pending read marker and guards the unread state instead`() = + runTest { + stored = stored.copy(lastReadMessage = MARKER, unreadMessages = 3) + updater.markPendingReadMarker(INTERNAL_CONVERSATION_ID, READ_UP_TO_NEWEST) - updater.switchToPendingMarkAsUnread(INTERNAL_CONVERSATION_ID) + updater.switchToPendingMarkAsUnread(INTERNAL_CONVERSATION_ID) - assertNull(updater.pendingReadMarker(INTERNAL_CONVERSATION_ID)) - // a response that still reports the conversation read does not overwrite the local unread state - val unread = preserve(serverLastRead = READ_UP_TO_NEWEST, serverUnread = 0) - assertEquals(MARKER, unread.lastReadMessage) - assertEquals(3, unread.unreadMessages) - } + assertNull(updater.pendingReadMarker(INTERNAL_CONVERSATION_ID)) + // a response that still reports the conversation read does not overwrite the local unread state + val unread = preserve(serverLastRead = READ_UP_TO_NEWEST, serverUnread = 0) + assertEquals(MARKER, unread.lastReadMessage) + assertEquals(3, unread.unreadMessages) + } @Test fun `reading in the chat releases a mark as unread from the list`() = @@ -183,12 +186,14 @@ class ConversationListUpdaterUnreadTest { assertEquals(4, preserve(serverLastRead = MARKER, serverUnread = 4).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) }