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..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 @@ -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,29 @@ 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 - but only once the refresh + // succeeded, otherwise the skipped events would become unrecoverable. logger.e { "[performSync] failed (too many events to sync): $result" } + 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 + return refresh.value } 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 +374,43 @@ 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 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, 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 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 refresh.value } /** @@ -407,9 +473,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 +506,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 +518,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 +682,83 @@ 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. + * + * 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 refreshed, or a failure when any request failed. + */ + private suspend fun refreshActiveWatchedChannels( + cidsToExclude: Set = emptySet(), + ): Result> { + 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()) { + 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( + 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" } + failed.set(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) + } + } + // 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 returned by the server; cids.size: ${missedCids.size}" } + } + logger.v { "[refreshActiveWatchedChannels] refreshedCids.size: ${refreshedCids.size}" } + return when (val error = failed.get()) { + null -> Result.Success(refreshedCids) + else -> Result.Failure(error) + } + } + 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..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 @@ -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,237 @@ 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) + } + } + + @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. + */ + 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 */ @@ -367,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() @@ -1194,8 +1432,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,