From fc91340b4414dc995cb631825a5663f225428e49 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 11 Aug 2026 13:00:43 +0200 Subject: [PATCH] perf(llc): skip replaying oversized `/sync` payloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaying a large `/sync` payload through `handleEvent` holds local persistence and state updates long enough to slow down the regular requests that need them. Payloads over 250 events are no longer replayed. On reconnect the synced channels are re-queried in their place, a page at a time, and `lastSyncAt` only advances once that refresh succeeded — dropping the events is safe when their state has been re-fetched, but advancing past a failed refresh would lose them. Mark-all-read events are still applied, since a channel refresh does not carry read state. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_chat/CHANGELOG.md | 2 + .../stream_chat/lib/src/client/client.dart | 100 ++++++- .../test/src/client/client_test.dart | 283 ++++++++++++++++++ packages/stream_chat/test/src/fakes.dart | 16 + 4 files changed, 388 insertions(+), 13 deletions(-) diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 0205063c4f..f38be28629 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/packages/stream_chat/CHANGELOG.md @@ -4,6 +4,8 @@ - `Channel.translateMessage` now merges the translated message into the channel state, so the translation reaches anything watching the channel without the caller applying the response itself. - Raised minimum Dart SDK to `^3.12.0`. +- `StreamChatClient.sync` now skips replaying oversized `/sync` payloads (over 250 events) to avoid stalling local persistence. On reconnect the synced channels are re-queried in their place before `lastSyncAt` advances. +- `Channel.memberCountStream` is now distinct, so it only emits when the count actually changes. 🐞 Fixed diff --git a/packages/stream_chat/lib/src/client/client.dart b/packages/stream_chat/lib/src/client/client.dart index d174612a16..73b126e5ee 100644 --- a/packages/stream_chat/lib/src/client/client.dart +++ b/packages/stream_chat/lib/src/client/client.dart @@ -616,11 +616,15 @@ class StreamChatClient { // surface as an unhandled crash instead of reaching the app. try { // Sync the persistence client if available - if (persistenceEnabled) await sync(cids: cids); + var channelsRefreshedBySync = false; + if (persistenceEnabled) { + channelsRefreshedBySync = await _sync(cids: cids, refreshChannelsOnSkip: true); + } - // Recover the channels that were active before the connection was lost, - // only if the client is configured to do so. - if (_recoverStateOnReconnect) { + // Recover the channels that were active before the connection was + // lost, only if the client is configured to do so and the sync has + // not already refreshed them. + if (_recoverStateOnReconnect && !channelsRefreshedBySync) { await queryChannelsOnline( filter: Filter.in_('cid', cids), paginationParams: const PaginationParams(limit: 30), @@ -658,34 +662,85 @@ class StreamChatClient { // Lock to make sure only one sync process is running at a time. final _syncLock = Lock(); + // Maximum number of events replayed from a single `/sync` response before + // skipping replay to avoid stalling local persistence on reconnect. + static const _syncEventReplayMaximumEventCount = 250; + + // Maximum number of channels a single `queryChannels` request returns. + static const _channelQueryMaximumPageSize = 30; + /// Get the events missed while offline to sync the offline storage /// Will automatically fetch [cids] and [lastSyncedAt] if [persistenceEnabled] + /// + /// Events from an oversized payload are not replayed. The sync pointer still + /// advances, so callers relying on the replayed state should refresh it + /// themselves. Future sync({List? cids, DateTime? lastSyncAt}) { + return _sync(cids: cids, lastSyncAt: lastSyncAt); + } + + // Runs the sync flow, returning whether the synced channels were refreshed. + // + // Set [refreshChannelsOnSkip] to refresh the synced channels when an + // oversized payload skips event replay, so that their state takes the place + // of the events that were dropped. + Future _sync({ + List? cids, + DateTime? lastSyncAt, + bool refreshChannelsOnSkip = false, + }) { return _syncLock.synchronized(() async { final channels = cids ?? await chatPersistenceClient?.getChannelCids(); - if (channels == null || channels.isEmpty) return; + if (channels == null || channels.isEmpty) return false; final syncAt = lastSyncAt ?? await chatPersistenceClient?.getLastSyncAt(); if (syncAt == null) { logger.info('Fresh sync start: lastSyncAt initialized to now.'); - return chatPersistenceClient?.updateLastSyncAt(DateTime.now()); + await chatPersistenceClient?.updateLastSyncAt(DateTime.timestamp()); + return false; } try { logger.info('Syncing events since $syncAt for channels: $channels'); final res = await _chatApi.general.sync(channels, syncAt); - final events = res.events.sorted( - (a, b) => a.createdAt.compareTo(b.createdAt), - ); + final events = res.events.sorted((a, b) => a.createdAt.compareTo(b.createdAt)); + final updatedSyncAt = events.lastOrNull?.createdAt ?? DateTime.timestamp(); + + // Bail out of oversized event replay. Replaying a large payload through + // [handleEvent] can hold local persistence and state updates long + // enough to slow down regular requests. Refresh the synced channels + // instead, and only advance the sync pointer once that succeeded: + // dropping the events is safe when their state has been re-fetched, + // but advancing past a failed refresh loses them for good. + if (events.length > _syncEventReplayMaximumEventCount) { + logger.info( + 'Skipping replay of ${events.length} events, exceeding the ' + 'limit of $_syncEventReplayMaximumEventCount.', + ); + + if (refreshChannelsOnSkip) await _refreshChannels(channels); + + // A channel refresh does not carry the read state, so keep honouring + // the mark-all-read events instead of losing them with the rest of + // the payload. + for (final event in events) { + if (event.type != EventType.notificationMarkRead) continue; + if (event.cid != null) continue; + handleEvent(event); + } + + await chatPersistenceClient?.updateLastSyncAt(updatedSyncAt); + return refreshChannelsOnSkip; + } for (final event in events) { logger.fine('Syncing event: ${event.type}'); handleEvent(event); } - final updatedSyncAt = events.lastOrNull?.createdAt ?? DateTime.now(); - return await chatPersistenceClient?.updateLastSyncAt(updatedSyncAt); + await chatPersistenceClient?.updateLastSyncAt(updatedSyncAt); + return false; } catch (error, stk) { // If we got a 400 error, it means that either the sync time is too // old or the channel list is too long or too many events need to be @@ -699,18 +754,37 @@ class StreamChatClient { try { await chatPersistenceClient?.flush(); - return await chatPersistenceClient?.updateLastSyncAt(DateTime.now()); + await chatPersistenceClient?.updateLastSyncAt(DateTime.timestamp()); } catch (resetError, resetStk) { logger.warning('Error resetting the persistence client', resetError, resetStk); - return; } + + return false; } logger.warning('Error syncing events', error, stk); + return false; } }); } + // Refreshes the state of the given channels from the server, a page at a + // time so that sets larger than [_channelQueryMaximumPageSize] are covered + // in full rather than truncated to the first page. + Future _refreshChannels(List cids) async { + logger.info('Refreshing ${cids.length} channels'); + + for (final batch in cids.slices(_channelQueryMaximumPageSize)) { + await queryChannelsOnline( + filter: Filter.in_('cid', batch), + paginationParams: PaginationParams(limit: batch.length), + // Fail fast if the connection dropped again: waiting for it here would + // hold the sync lock, blocking the sync the next reconnect starts. + waitForConnect: false, + ); + } + } + final _queryChannelsCache = InFlightCache(); /// Requests channels with a given query. diff --git a/packages/stream_chat/test/src/client/client_test.dart b/packages/stream_chat/test/src/client/client_test.dart index 5d7ae8e9d4..9a8b8ba8b2 100644 --- a/packages/stream_chat/test/src/client/client_test.dart +++ b/packages/stream_chat/test/src/client/client_test.dart @@ -5233,6 +5233,132 @@ void main() { verify(() => api.general.sync(cids, lastSyncAt)).called(1); }); + + test( + '''should replay events and advance lastSyncAt when the payload is within the replay limit''', + () async { + final cids = ['channel1']; + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final fakeClient = FakePersistenceClient( + channelCids: cids, + lastSyncAt: lastSyncAt, + ); + + client.chatPersistenceClient = fakeClient; + final events = List.generate( + 10, + (index) => Event( + type: EventType.messageNew, + cid: 'channel1', + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(cids, lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + final replayed = []; + final sub = client.on(EventType.messageNew).listen(replayed.add); + addTearDown(sub.cancel); + + await client.sync(); + await pumpEventQueue(); + + verify(() => api.general.sync(cids, lastSyncAt)).called(1); + // Within the limit, every event is replayed through the event handler. + expect(replayed, hasLength(events.length)); + // lastSyncAt advances to the newest replayed event date. + expect(await fakeClient.getLastSyncAt(), events.last.createdAt); + }, + ); + + test( + '''should skip replay but advance lastSyncAt when the payload exceeds the replay limit''', + () async { + final cids = ['channel1']; + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final fakeClient = FakePersistenceClient( + channelCids: cids, + lastSyncAt: lastSyncAt, + ); + + client.chatPersistenceClient = fakeClient; + // 251 events exceeds the internal replay limit of 250. + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: 'channel1', + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(cids, lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + final replayed = []; + final sub = client.on(EventType.messageNew).listen(replayed.add); + addTearDown(sub.cancel); + + await client.sync(); + await pumpEventQueue(); + + verify(() => api.general.sync(cids, lastSyncAt)).called(1); + // Replay is skipped; no events are dispatched through the handler. + expect(replayed, isEmpty); + // A direct call refreshes nothing on its own. + verifyZeroInteractions(api.channel); + // lastSyncAt still advances to the newest event date in the skipped + // payload, so the same oversized payload is not retried indefinitely. + expect(await fakeClient.getLastSyncAt(), events.last.createdAt); + }, + ); + + test( + '''should still replay mark-all-read events from a payload that exceeds the replay limit''', + () async { + final cids = ['channel1']; + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final fakeClient = FakePersistenceClient( + channelCids: cids, + lastSyncAt: lastSyncAt, + ); + + client.chatPersistenceClient = fakeClient; + final events = [ + ...List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: 'channel1', + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ), + // A channel-less `notification.mark_read` marks every channel read. + Event( + type: EventType.notificationMarkRead, + createdAt: lastSyncAt.add(const Duration(minutes: 5)), + ), + ]; + when(() => api.general.sync(cids, lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + final replayed = []; + final sub = client.on(EventType.notificationMarkRead).listen(replayed.add); + addTearDown(sub.cancel); + + await client.sync(); + await pumpEventQueue(); + + // A channel refresh does not carry the read state, so these events + // survive the skip. + expect(replayed, hasLength(1)); + }, + ); }); }); @@ -5417,6 +5543,163 @@ void main() { ); }); + // Skipping event replay leaves the state of the synced channels behind, so + // the skip refreshes them itself, whatever this flag is set to. + test('should re-query active channels when the sync skipped event replay', () async { + client = StreamChatClient(apiKey, chatApi: api, ws: ws, recoverStateOnReconnect: false); + await client.connectUser(user, token); + await delay(300); + + const cid = 'messaging:c1'; + final channel = Channel.fromState(client, ChannelState(channel: ChannelModel(cid: cid))); + client.state.addChannels({cid: channel}); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + client.chatPersistenceClient = FakePersistenceClient(channelCids: const [cid], lastSyncAt: lastSyncAt); + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + // 251 events exceeds the internal replay limit of 250. + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: cid, + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(const [cid], lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + clearInteractions(api.channel); + + await simulateReconnect(); + + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', const [cid]), + sort: any(named: 'sort'), + state: any(named: 'state'), + watch: any(named: 'watch'), + presence: any(named: 'presence'), + memberLimit: any(named: 'memberLimit'), + messageLimit: any(named: 'messageLimit'), + paginationParams: const PaginationParams(limit: 1), + ), + ).called(1); + }); + + // Dropping the skipped events is only safe once their state has been + // re-fetched; advancing the pointer past a failed refresh loses them. + test('should keep lastSyncAt when the re-query after a skipped replay fails', () async { + when( + () => api.channel.queryChannels( + filter: any(named: 'filter'), + sort: any(named: 'sort'), + state: any(named: 'state'), + watch: any(named: 'watch'), + presence: any(named: 'presence'), + memberLimit: any(named: 'memberLimit'), + messageLimit: any(named: 'messageLimit'), + paginationParams: any(named: 'paginationParams'), + ), + ).thenThrow(const StreamChatError('You cannot use queryChannels without an active connection.')); + + client = StreamChatClient(apiKey, chatApi: api, ws: ws, recoverStateOnReconnect: false); + await client.connectUser(user, token); + await delay(300); + + const cid = 'messaging:c1'; + final channel = Channel.fromState(client, ChannelState(channel: ChannelModel(cid: cid))); + client.state.addChannels({cid: channel}); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final persistenceClient = FakePersistenceClient(channelCids: const [cid], lastSyncAt: lastSyncAt); + client.chatPersistenceClient = persistenceClient; + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + // 251 events exceeds the internal replay limit of 250. + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: cid, + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(const [cid], lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + await simulateReconnect(); + + expect(await persistenceClient.getLastSyncAt(), lastSyncAt); + }); + + test('should re-query in batches when more channels are active than fit in one page', () async { + client = StreamChatClient(apiKey, chatApi: api, ws: ws, recoverStateOnReconnect: false); + await client.connectUser(user, token); + await delay(300); + + // 31 channels spill over the 30-channel page size into a second request. + final cids = List.generate(31, (index) => 'messaging:c$index'); + client.state.addChannels({ + for (final cid in cids) cid: Channel.fromState(client, ChannelState(channel: ChannelModel(cid: cid))), + }); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + client.chatPersistenceClient = FakePersistenceClient(channelCids: cids, lastSyncAt: lastSyncAt); + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: cids.first, + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(cids, lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + clearInteractions(api.channel); + + await simulateReconnect(); + + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', cids.take(30).toList()), + sort: any(named: 'sort'), + state: any(named: 'state'), + watch: any(named: 'watch'), + presence: any(named: 'presence'), + memberLimit: any(named: 'memberLimit'), + messageLimit: any(named: 'messageLimit'), + paginationParams: const PaginationParams(limit: 30), + ), + ).called(1); + + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', cids.skip(30).toList()), + sort: any(named: 'sort'), + state: any(named: 'state'), + watch: any(named: 'watch'), + presence: any(named: 'presence'), + memberLimit: any(named: 'memberLimit'), + messageLimit: any(named: 'messageLimit'), + paginationParams: const PaginationParams(limit: 1), + ), + ).called(1); + }); + test('should respect runtime toggling via the setter', () async { client = StreamChatClient(apiKey, chatApi: api, ws: ws); await client.connectUser(user, token); diff --git a/packages/stream_chat/test/src/fakes.dart b/packages/stream_chat/test/src/fakes.dart index 6c2794b87c..97a5961e75 100644 --- a/packages/stream_chat/test/src/fakes.dart +++ b/packages/stream_chat/test/src/fakes.dart @@ -98,6 +98,22 @@ class FakePersistenceClient extends Fake implements ChatPersistenceClient { @override Future> getChannelCids() async => _channelCids; + + @override + Future saveChannelQueries({ + required List cids, + Filter? filter, + SortOrder? sort, + String? predefinedFilter, + Filter? resolvedFilter, + SortOrder? resolvedSort, + Map? filterValues, + Map? sortValues, + bool clearQueryCache = false, + }) async {} + + @override + Future updateChannelStates(List channelStates) async {} } class FakeChatApi extends Fake implements StreamChatApi {