Skip oversized /sync event replay and refresh watched channels

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<String> {
         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<String>) {
+    internal suspend fun performSync(cids: List<String>): Set<String> {
         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<ChatEvent>): Set<String> {
+        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<String> = 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<String> = emptySet(),
+    ): Result<Set<String>> {
+        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<Error>()
+        val refreshedCids = mutableSetOf<String>()
+        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}" }