From e236b4279d45aa14b488687e81d98ef5f51b5e54 Mon Sep 17 00:00:00 2001 From: Aleksandar Apostolov Date: Wed, 19 Aug 2026 16:14:12 +0200 Subject: [PATCH 1/2] perf(client): skip oversized /sync event replay on reconnect A /sync response can sit below the backend cap and still be too expensive to replay: every event goes through the state and persistence pipeline, holding it long enough to slow down the requests that also need it. Above 250 events, skip the replay and refresh the actively watched channels with queryChannels instead, which carries the latest channel state, messages, members and read state and re-registers the watch. The existing HTTP 400 too-many-events branch now takes the same path instead of refreshing nothing. Channels that are not actively watched wait for a future explicit query. The limit stays internal - it is a performance guard, not a knob. --- .../state/sync/internal/SyncManager.kt | 148 +++++++++++++++-- .../state/internal/SyncManagerTest.kt | 152 ++++++++++++++++++ 2 files changed, 286 insertions(+), 14 deletions(-) diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/sync/internal/SyncManager.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/sync/internal/SyncManager.kt index cc5eee63dd0..bc66ebbf577 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/sync/internal/SyncManager.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/sync/internal/SyncManager.kt @@ -83,6 +83,22 @@ import java.util.concurrent.atomic.AtomicReference private const val QUERIES_TO_RETRY = 3 private const val SYNC_MAX_CIDS = 100 +/** + * Maximum number of events replayed from a single `/sync` response. Above this count, event replay + * is skipped and the actively watched channels are refreshed with `queryChannels` instead. + * + * Deliberately conservative and well below the backend cap: a payload can be accepted by the + * backend and still be too expensive to replay on the device. + */ +private const val SYNC_EVENT_REPLAY_MAX_COUNT = 250 + +/** + * Page size of the `queryChannels` fallback. The API caps `queryChannels` at 30 channels per + * request, so watched cids are refreshed in chunks of this size rather than in [SYNC_MAX_CIDS] + * batches like `/sync` is. + */ +private const val QUERY_CHANNELS_MAX_LIMIT = 30 + /** * This class is responsible to sync messages, reactions and channel data. It tries to sync then, if necessary, * when connection is reestablished or when a health check event happens. @@ -99,6 +115,7 @@ internal class SyncManager( private val userPresence: Boolean, private val isAutomaticSyncOnReconnectEnabled: Boolean, private val syncMaxThreshold: TimeDuration, + private val eventReplayMaxCount: Int = SYNC_EVENT_REPLAY_MAX_COUNT, private val now: () -> Long, private val serverClockOffset: ServerClockOffset, scope: CoroutineScope, @@ -254,8 +271,8 @@ internal class SyncManager( } if (isAutomaticSyncOnReconnectEnabled) { logger.i { "[onConnectionEstablished] performing sync and restoring active channels" } - performSync() - restoreActiveChannels() + val refreshedCids = performSync() + restoreActiveChannels(alreadyRefreshedCids = refreshedCids) } else { logger.i { "[onConnectionEstablished] skipping sync, isAutomaticSyncOnReconnectEnabled=false" } } @@ -285,23 +302,31 @@ internal class SyncManager( logger.i { "[connectionLost] failed: $e" } } - private suspend fun performSync() { + /** + * @return The channel ids refreshed by the `queryChannels` fallback, empty when events were + * replayed normally. + */ + private suspend fun performSync(): Set { logger.i { "[performSync] no args" } val cids = logicRegistry.getActiveChannelsLogic().map { it.cid }.ifEmpty { logger.w { "[performSync] no active cids found" } repos.selectSyncState(currentUserId)?.activeChannelIds ?: emptyList() } - mutex.withLock { + return mutex.withLock { performSync(cids) } } + /** + * @return The channel ids refreshed by the `queryChannels` fallback, empty when events were + * replayed normally. + */ @VisibleForTesting - internal suspend fun performSync(cids: List) { + internal suspend fun performSync(cids: List): Set { logger.d { "[performSync] cids.size: ${cids.size} " } if (cids.isEmpty()) { logger.w { "[performSync] rejected (cids is empty)" } - return + return emptySet() } val syncState = syncState.value ?: repos.selectSyncState(currentUserId) val lastSyncAt = syncState?.lastSyncedAt ?: Date(now()) @@ -314,17 +339,24 @@ internal class SyncManager( chatClient.getSyncHistory(cappedCids, lastSyncAt).await() } if (result.isTooManyEventsToSyncError()) { + // Event replay is impossible here, so refresh the watched surfaces instead. The + // payload is unavailable, hence the timestamp moves to now to avoid retrying forever. logger.e { "[performSync] failed (too many events to sync): $result" } + val refreshedCids = refreshActiveWatchedChannels() updateLastSyncedDate(latestEventDate = Date(now()), rawLatestEventDate = null) - return + return refreshedCids } if (result !is Result.Success) { logger.e { "[performSync] failed($result)" } - return + return emptySet() } val sortedEvents = result.value.sortedBy { it.createdAt } logger.v { "[performSync] succeed; events.size: ${sortedEvents.size}" } + if (sortedEvents.size > eventReplayMaxCount) { + return skipEventReplay(sortedEvents) + } + val latestEvent = sortedEvents.lastOrNull() val latestEventDate = latestEvent?.createdAt ?: Date(now()) val rawLatestEventDate = latestEvent?.rawCreatedAt @@ -337,14 +369,36 @@ internal class SyncManager( } if (sortedEvents.isEmpty()) { logger.w { "[performSync] rejected (no events to emit)" } - return + return emptySet() } if (rawLastSyncAt == rawLatestEventDate) { logger.w { "[performSync] rejected (rawLatestEventDate equals to rawLastSyncAt)" } - return + return emptySet() } events.emit(sortedEvents) logger.v { "[performSync] events emission completed" } + return emptySet() + } + + /** + * Skips the replay of an oversized `/sync` payload. + * + * Replaying this many events would hold the state and persistence pipeline long enough to slow + * down the requests that also need it, so the actively watched channels are refreshed with + * `queryChannels` instead. The last-synced timestamp moves to the newest skipped event, which is + * safe because the visible surfaces were refreshed and the channels that are not actively + * watched wait for a future explicit query. + * + * @return The channel ids that were refreshed. + */ + private suspend fun skipEventReplay(sortedEvents: List): Set { + logger.i { + "[performSync] skipping event replay; events.size: ${sortedEvents.size} exceeds $eventReplayMaxCount" + } + val refreshedCids = refreshActiveWatchedChannels() + val skippedLatestEvent = sortedEvents.last() + updateLastSyncedDate(skippedLatestEvent.createdAt, skippedLatestEvent.rawCreatedAt) + return refreshedCids } /** @@ -407,9 +461,16 @@ internal class SyncManager( logger.e(e) { "[retryFailedEntities] failed: $e" } } - private suspend fun restoreActiveChannels() { + /** + * @param alreadyRefreshedCids Channel ids already brought up to date earlier in the recovery + * flow, e.g. by the `/sync` event replay fallback. They are not queried a second time. + */ + private suspend fun restoreActiveChannels(alreadyRefreshedCids: Set = emptySet()) { val recoverAll = !isFirstConnect.compareAndSet(true, false) - logger.d { "[restoreActiveChannels] recoverAll: $recoverAll" } + logger.d { + "[restoreActiveChannels] recoverAll: $recoverAll, " + + "alreadyRefreshedCids.size: ${alreadyRefreshedCids.size}" + } val allLogics = logicRegistry.getActiveQueryChannelsLogic() val hasGroupedQueries = allLogics.any { it.groupKey() != null } @@ -433,7 +494,7 @@ internal class SyncManager( is Result.Success -> { val updatedCids = result.value logger.v { "[restoreActiveChannels] standardCids.size: ${result.value.size}" } - updateActiveChannels(recoverAll, updatedCids + groupedHandledCids) + updateActiveChannels(recoverAll, updatedCids + groupedHandledCids + alreadyRefreshedCids) } is Result.Failure -> { logger.e { "[restoreActiveChannels] standard query failed: ${result.value}" } @@ -445,7 +506,7 @@ internal class SyncManager( // --- Active Channels created outside of QueryChannels requests if (!hasStandardQueries && !hasGroupedQueries) { // Check for active channels created outside of a QueryChannels requests - updateActiveChannels(recoverAll, cidsToExclude = emptySet()) + updateActiveChannels(recoverAll, cidsToExclude = alreadyRefreshedCids) } } @@ -609,6 +670,65 @@ internal class SyncManager( } } + /** + * Refreshes the actively watched channels through `queryChannels` instead of replaying an + * oversized `/sync` payload event by event. + * + * A `queryChannels` response carries the latest channel state, messages, members, watchers and + * read state, and re-registers the watch for the requested cids, which is what the currently + * visible surfaces need. Channels that are not actively watched are left to a future explicit + * query. + * + * @param cidsToExclude Channel ids already refreshed elsewhere in the recovery flow. + * + * @return The channel ids that were successfully refreshed. + */ + private suspend fun refreshActiveWatchedChannels(cidsToExclude: Set = emptySet()): Set { + val online = clientState.isOnline + val watchedCids = buildSet { + addAll(stateRegistry.getTrackedWatchedChannels()) + logicRegistry.getActiveChannelsLogic().mapTo(this) { it.cid } + } - cidsToExclude + logger.d { + "[refreshActiveWatchedChannels] watchedCids.size: ${watchedCids.size}, online: $online, " + + "cidsToExclude.size: ${cidsToExclude.size}" + } + if (watchedCids.isEmpty() || !online) { + return emptySet() + } + + val refreshedCids = mutableSetOf() + watchedCids.chunked(QUERY_CHANNELS_MAX_LIMIT).forEach { batch -> + val request = QueryChannelsRequest( + filter = Filters.`in`("cid", batch), + offset = 0, + limit = batch.size, + ).apply { presence = userPresence } + logger.v { "[refreshActiveWatchedChannels] request: $request" } + chatClient.queryChannelsInternal(request) + .await() + .onError { + logger.e { "[refreshActiveWatchedChannels] request failed: $it" } + } + .onSuccessSuspend { queryResult -> + val foundChannels = queryResult.channels + foundChannels.forEach { channel -> + ChannelId.fromTypeAndId(channel.type, channel.id) + ?.let(logicRegistry::channel) + ?.updateDataForChannel(channel, channel.messages.size) + } + repos.storeStateForChannels(foundChannels) + foundChannels.mapTo(refreshedCids, Channel::cid) + } + } + val missedCids = watchedCids - refreshedCids + if (missedCids.isNotEmpty()) { + logger.w { "[refreshActiveWatchedChannels] not refreshed cids.size: ${missedCids.size}" } + } + logger.v { "[refreshActiveWatchedChannels] refreshedCids.size: ${refreshedCids.size}" } + return refreshedCids + } + private suspend fun retryChannels() { val cids = repos.selectChannelCidsBySyncNeeded() logger.d { "[retryChannels] cids.size: ${cids.size}" } diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/internal/SyncManagerTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/internal/SyncManagerTest.kt index bd0df8057ee..87495ed0700 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/internal/SyncManagerTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/internal/SyncManagerTest.kt @@ -42,6 +42,7 @@ import io.getstream.chat.android.client.utils.internal.ServerClockOffset import io.getstream.chat.android.client.utils.observable.Disposable import io.getstream.chat.android.core.internal.InternalStreamChatApi import io.getstream.chat.android.core.internal.coroutines.Tube +import io.getstream.chat.android.models.Channel import io.getstream.chat.android.models.ConnectionState import io.getstream.chat.android.models.GroupedChannels import io.getstream.chat.android.models.GroupedChannelsGroup @@ -258,6 +259,155 @@ internal class SyncManagerTest { verify(chatClient).getSyncHistory(eq(expectedCids), any()) } + @Test + fun `performSync skips event replay and refreshes watched channels when the payload is oversized`() = + runTest(testDispatcher) { + /* Given */ + val createdAt = Date() + val rawCreatedAt = streamDateFormatter.format(createdAt) + val watchedChannel = randomChannel(type = "messaging", id = "watched") + givenOversizedSyncPayload(eventCount = 3, createdAt = createdAt, rawCreatedAt = rawCreatedAt) + givenWatchedChannels(cids = setOf(watchedChannel.cid), foundChannels = listOf(watchedChannel)) + + val syncManager = buildSyncManager(eventReplayMaxCount = 2) + + /* When */ + _syncEvents.test { + syncManager.performSync(cids = listOf("1", "2")) + + /* Then */ + expectNoEvents() + } + argumentCaptor { + verify(chatClient).queryChannelsInternal(capture()) + val filter = firstValue.filter as InFilterObject + assertEquals(setOf(watchedChannel.cid), filter.values) + assertEquals(true, firstValue.watch) + assertEquals(true, firstValue.state) + } + verify(repositoryFacade).storeStateForChannels(listOf(watchedChannel)) + // The timestamp moves to the newest skipped event so the same payload is not retried. + assertEquals(createdAt, _syncState.value?.lastSyncedAt) + } + + @Test + fun `performSync replays events when the payload size equals the replay limit`() = runTest(testDispatcher) { + /* Given */ + val createdAt = Date() + val rawCreatedAt = streamDateFormatter.format(createdAt) + val events = givenOversizedSyncPayload(eventCount = 2, createdAt = createdAt, rawCreatedAt = rawCreatedAt) + givenWatchedChannels(cids = setOf(randomCID()), foundChannels = emptyList()) + + val syncManager = buildSyncManager(eventReplayMaxCount = 2) + + _syncEvents.test { + /* When */ + syncManager.performSync(cids = listOf("1", "2")) + + /* Then */ + assertEquals(events, awaitItem()) + } + verify(chatClient, never()).queryChannelsInternal(any()) + } + + @Test + fun `performSync refreshes watched channels when sync fails with too many events`() = runTest(testDispatcher) { + /* Given */ + val watchedChannel = randomChannel(type = "messaging", id = "watched") + val error = Error.NetworkError( + serverErrorCode = ChatErrorCode.VALIDATION_ERROR.code, + message = "Too many events to sync, please use a more recent last_sync_at parameter", + statusCode = 400, + ) + _syncState.value = null + whenever(repositoryFacade.selectSyncState(any())) doReturn null + whenever(chatClient.getSyncHistory(any(), any())) doReturn TestCall(Result.Failure(error)) + whenever(chatClient.getSyncHistory(any(), any())) doReturn TestCall(Result.Failure(error)) + givenWatchedChannels(cids = setOf(watchedChannel.cid), foundChannels = listOf(watchedChannel)) + + val syncManager = buildSyncManager() + + /* When */ + _syncEvents.test { + syncManager.performSync(cids = listOf("1", "2")) + + /* Then */ + expectNoEvents() + } + verify(chatClient).queryChannelsInternal(any()) + verify(repositoryFacade).storeStateForChannels(listOf(watchedChannel)) + } + + @Test + fun `performSync batches the watched channels fallback by the queryChannels page limit`() = + runTest(testDispatcher) { + /* Given */ + val createdAt = Date() + val watchedCids = (1..35).map { "messaging:watched-$it" }.toSet() + givenOversizedSyncPayload( + eventCount = 3, + createdAt = createdAt, + rawCreatedAt = streamDateFormatter.format(createdAt), + ) + givenWatchedChannels(cids = watchedCids, foundChannels = emptyList()) + + val syncManager = buildSyncManager(eventReplayMaxCount = 2) + + /* When */ + syncManager.performSync(cids = listOf("1", "2")) + + /* Then */ + argumentCaptor { + verify(chatClient, times(2)).queryChannelsInternal(capture()) + val requestedCids = allValues.flatMap { (it.filter as InFilterObject).values } + assertEquals(watchedCids, requestedCids.toSet()) + assertEquals(30, allValues.first().limit) + assertEquals(5, allValues.last().limit) + } + } + + /** + * Stubs `/sync` to return [eventCount] identical events, with no previously stored sync state. + */ + private suspend fun givenOversizedSyncPayload( + eventCount: Int, + createdAt: Date, + rawCreatedAt: String, + ): List { + val previousSyncedAt = Date(createdAt.time - 60_000) + val previousSyncState = SyncState( + userId = user.id, + activeChannelIds = emptyList(), + lastSyncedAt = previousSyncedAt, + rawLastSyncedAt = streamDateFormatter.format(previousSyncedAt), + markedAllReadAt = previousSyncedAt, + ) + _syncState.value = previousSyncState + whenever(repositoryFacade.selectSyncState(any())) doReturn previousSyncState + val events = List(eventCount) { + mock { + on(it.createdAt) doReturn createdAt + on(it.rawCreatedAt) doReturn rawCreatedAt + } + } + whenever(chatClient.getSyncHistory(any(), any())) doReturn TestCall(Result.Success(events)) + whenever(chatClient.getSyncHistory(any(), any())) doReturn TestCall(Result.Success(events)) + return events + } + + /** + * Stubs the actively watched channels and the `queryChannels` response the fallback receives. + */ + private fun givenWatchedChannels(cids: Set, foundChannels: List) { + whenever(clientState.isOnline) doReturn true + whenever(stateRegistry.getTrackedWatchedChannels()) doReturn cids + whenever(logicRegistry.getActiveChannelsLogic()) doReturn emptyList() + whenever(logicRegistry.channel(any())) doReturn mock() + whenever(chatClient.queryChannelsInternal(any())) doReturn TestCall( + Result.Success(QueryChannelsResult(channels = foundChannels, predefinedFilter = null)), + ) + } + @Test fun `test sync max threshold for messages`() = runTest(testDispatcher) { /* Given */ @@ -1194,8 +1344,10 @@ internal class SyncManagerTest { private fun TestScope.buildSyncManager( isAutomaticSyncOnReconnectEnabled: Boolean = true, syncMaxThreshold: TimeDuration = TimeDuration.seconds(5), + eventReplayMaxCount: Int = 250, ): SyncManager { return SyncManager( + eventReplayMaxCount = eventReplayMaxCount, currentUserId = user.id, scope = backgroundScope, logicRegistry = logicRegistry, From 7d7555708a6f83d8875e9c60046faa2ba738c4d2 Mon Sep 17 00:00:00 2001 From: Aleksandar Apostolov Date: Thu, 20 Aug 2026 10:46:29 +0200 Subject: [PATCH 2/2] fix(client): keep the sync checkpoint when the fallback refresh fails The oversized-payload and too-many-events paths advanced lastSyncedAt even when the queryChannels fallback failed, so the events they skipped could never be replayed and the watched channels were left stale with no way back. refreshActiveWatchedChannels now reports failure, and both callers keep the previous checkpoint on it - the next reconnect asks for the same range and retries. Having nothing to refresh stays a success: with no watched channels nothing visible can go stale, and failing there would re-fetch the payload on every reconnect forever. A cid the server does not return is not a failure either, since a deleted or invisible channel would pin the checkpoint permanently. --- .../state/sync/internal/SyncManager.kt | 62 +++++++++---- .../state/internal/SyncManagerTest.kt | 88 +++++++++++++++++++ 2 files changed, 134 insertions(+), 16 deletions(-) diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/sync/internal/SyncManager.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/sync/internal/SyncManager.kt index bc66ebbf577..5dbd8462ebe 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/sync/internal/SyncManager.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/sync/internal/SyncManager.kt @@ -339,12 +339,17 @@ internal class SyncManager( chatClient.getSyncHistory(cappedCids, lastSyncAt).await() } if (result.isTooManyEventsToSyncError()) { - // Event replay is impossible here, so refresh the watched surfaces instead. The - // payload is unavailable, hence the timestamp moves to now to avoid retrying forever. + // Event replay is impossible here, so refresh the watched surfaces instead. The payload + // is unavailable, hence the timestamp moves to now - but only once the refresh + // succeeded, otherwise the skipped events would become unrecoverable. logger.e { "[performSync] failed (too many events to sync): $result" } - val refreshedCids = refreshActiveWatchedChannels() + val refresh = refreshActiveWatchedChannels() + if (refresh !is Result.Success) { + logger.e { "[performSync] fallback refresh failed, keeping lastSyncedAt: ${refresh.errorOrNull()}" } + return emptySet() + } updateLastSyncedDate(latestEventDate = Date(now()), rawLatestEventDate = null) - return refreshedCids + return refresh.value } if (result !is Result.Success) { logger.e { "[performSync] failed($result)" } @@ -385,20 +390,27 @@ internal class SyncManager( * * Replaying this many events would hold the state and persistence pipeline long enough to slow * down the requests that also need it, so the actively watched channels are refreshed with - * `queryChannels` instead. The last-synced timestamp moves to the newest skipped event, which is - * safe because the visible surfaces were refreshed and the channels that are not actively - * watched wait for a future explicit query. + * `queryChannels` instead. + * + * The last-synced timestamp only moves to the newest skipped event once that refresh succeeded: + * it is what makes discarding the payload safe. When the refresh fails the checkpoint is kept, + * so the next reconnect asks for the same range again and can retry - at the cost of re-fetching + * a payload we discard, which is a request and a parse rather than a stalled pipeline. * - * @return The channel ids that were refreshed. + * @return The channel ids that were refreshed, empty when the refresh failed. */ private suspend fun skipEventReplay(sortedEvents: List): Set { logger.i { "[performSync] skipping event replay; events.size: ${sortedEvents.size} exceeds $eventReplayMaxCount" } - val refreshedCids = refreshActiveWatchedChannels() + val refresh = refreshActiveWatchedChannels() + if (refresh !is Result.Success) { + logger.e { "[skipEventReplay] fallback refresh failed, keeping lastSyncedAt: ${refresh.errorOrNull()}" } + return emptySet() + } val skippedLatestEvent = sortedEvents.last() updateLastSyncedDate(skippedLatestEvent.createdAt, skippedLatestEvent.rawCreatedAt) - return refreshedCids + return refresh.value } /** @@ -679,11 +691,17 @@ internal class SyncManager( * visible surfaces need. Channels that are not actively watched are left to a future explicit * query. * + * Having nothing to refresh is a success: with no watched channels there is nothing visible that + * can go stale. A failed request is not, and the caller must not advance the last-synced date on + * it, or the events it skipped become unrecoverable. + * * @param cidsToExclude Channel ids already refreshed elsewhere in the recovery flow. * - * @return The channel ids that were successfully refreshed. + * @return The channel ids that were refreshed, or a failure when any request failed. */ - private suspend fun refreshActiveWatchedChannels(cidsToExclude: Set = emptySet()): Set { + private suspend fun refreshActiveWatchedChannels( + cidsToExclude: Set = emptySet(), + ): Result> { val online = clientState.isOnline val watchedCids = buildSet { addAll(stateRegistry.getTrackedWatchedChannels()) @@ -693,10 +711,15 @@ internal class SyncManager( "[refreshActiveWatchedChannels] watchedCids.size: ${watchedCids.size}, online: $online, " + "cidsToExclude.size: ${cidsToExclude.size}" } - if (watchedCids.isEmpty() || !online) { - return emptySet() + if (watchedCids.isEmpty()) { + return Result.Success(emptySet()) + } + if (!online) { + logger.w { "[refreshActiveWatchedChannels] rejected (offline)" } + return Result.Failure(Error.GenericError("Cannot refresh watched channels while offline")) } + val failed = AtomicReference() val refreshedCids = mutableSetOf() watchedCids.chunked(QUERY_CHANNELS_MAX_LIMIT).forEach { batch -> val request = QueryChannelsRequest( @@ -709,6 +732,7 @@ internal class SyncManager( .await() .onError { logger.e { "[refreshActiveWatchedChannels] request failed: $it" } + failed.set(it) } .onSuccessSuspend { queryResult -> val foundChannels = queryResult.channels @@ -721,12 +745,18 @@ internal class SyncManager( foundChannels.mapTo(refreshedCids, Channel::cid) } } + // A cid the server did not return is not a failure: it can be deleted or no longer visible + // to this user, and holding the checkpoint for it would re-fetch the payload on every + // reconnect forever. Only a failed request blocks the checkpoint. val missedCids = watchedCids - refreshedCids if (missedCids.isNotEmpty()) { - logger.w { "[refreshActiveWatchedChannels] not refreshed cids.size: ${missedCids.size}" } + logger.w { "[refreshActiveWatchedChannels] not returned by the server; cids.size: ${missedCids.size}" } } logger.v { "[refreshActiveWatchedChannels] refreshedCids.size: ${refreshedCids.size}" } - return refreshedCids + return when (val error = failed.get()) { + null -> Result.Success(refreshedCids) + else -> Result.Failure(error) + } } private suspend fun retryChannels() { diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/internal/SyncManagerTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/internal/SyncManagerTest.kt index 87495ed0700..aa62d2b1329 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/internal/SyncManagerTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/internal/SyncManagerTest.kt @@ -366,6 +366,88 @@ internal class SyncManagerTest { } } + @Test + fun `performSync keeps lastSyncedAt when the watched channels fallback fails`() = runTest(testDispatcher) { + /* Given */ + val createdAt = Date() + givenOversizedSyncPayload( + eventCount = 3, + createdAt = createdAt, + rawCreatedAt = streamDateFormatter.format(createdAt), + ) + val previousSyncedAt = _syncState.value!!.lastSyncedAt + givenWatchedChannels(cids = setOf(randomCID()), foundChannels = emptyList()) + whenever(chatClient.queryChannelsInternal(any())) doReturn TestCall( + Result.Failure(Error.NetworkError(serverErrorCode = 0, message = "boom", statusCode = 500)), + ) + + val syncManager = buildSyncManager(eventReplayMaxCount = 2) + + /* When */ + syncManager.performSync(cids = listOf("1", "2")) + + /* Then */ + // Advancing here would make the skipped events unrecoverable on the next reconnect. + assertEquals(previousSyncedAt, _syncState.value?.lastSyncedAt) + verify(repositoryFacade, never()).insertSyncState(any()) + } + + @Test + fun `performSync keeps lastSyncedAt when only one fallback batch fails`() = runTest(testDispatcher) { + /* Given */ + val createdAt = Date() + givenOversizedSyncPayload( + eventCount = 3, + createdAt = createdAt, + rawCreatedAt = streamDateFormatter.format(createdAt), + ) + val previousSyncedAt = _syncState.value!!.lastSyncedAt + val foundChannel = randomChannel(type = "messaging", id = "watched-1") + givenWatchedChannels( + cids = (1..35).map { "messaging:watched-$it" }.toSet(), + foundChannels = listOf(foundChannel), + ) + // First batch succeeds, second fails. + whenever(chatClient.queryChannelsInternal(any())) + .doReturn( + TestCall(Result.Success(QueryChannelsResult(channels = listOf(foundChannel), predefinedFilter = null))), + TestCall(Result.Failure(Error.NetworkError(serverErrorCode = 0, message = "boom", statusCode = 500))), + ) + + val syncManager = buildSyncManager(eventReplayMaxCount = 2) + + /* When */ + syncManager.performSync(cids = listOf("1", "2")) + + /* Then */ + verify(chatClient, times(2)).queryChannelsInternal(any()) + assertEquals(previousSyncedAt, _syncState.value?.lastSyncedAt) + } + + @Test + fun `performSync advances lastSyncedAt when there are no watched channels to refresh`() = + runTest(testDispatcher) { + /* Given */ + val createdAt = Date() + givenOversizedSyncPayload( + eventCount = 3, + createdAt = createdAt, + rawCreatedAt = streamDateFormatter.format(createdAt), + ) + // Nothing is being watched, so nothing visible can go stale - this must not be treated + // as a failed refresh, or the oversized payload is re-fetched on every reconnect. + givenWatchedChannels(cids = emptySet(), foundChannels = emptyList()) + + val syncManager = buildSyncManager(eventReplayMaxCount = 2) + + /* When */ + syncManager.performSync(cids = listOf("1", "2")) + + /* Then */ + verify(chatClient, never()).queryChannelsInternal(any()) + assertEquals(createdAt, _syncState.value?.lastSyncedAt) + } + /** * Stubs `/sync` to return [eventCount] identical events, with no previously stored sync state. */ @@ -517,6 +599,12 @@ internal class SyncManagerTest { whenever(repositoryFacade.selectSyncState(any())) doReturn testSyncState whenever(chatClient.getSyncHistory(any(), any())) doReturn TestCall(result) whenever(chatClient.getSyncHistory(any(), any())) doReturn TestCall(result) + // The timestamp only advances once the watched-channels fallback succeeded, so the client + // has to be able to run it. The failure and offline cases are covered separately. + whenever(clientState.isOnline) doReturn true + whenever(chatClient.queryChannelsInternal(any())) doReturn TestCall( + Result.Success(QueryChannelsResult(channels = emptyList(), predefinedFilter = null)), + ) val syncManager = buildSyncManager()