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<void> sync({List<String>? 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<bool> _sync({
+    List<String>? 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<void> _refreshChannels(List<String> 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<String, QueryChannelsResult>();
 
   /// 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 = <Event>[];
+          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 = <Event>[];
+          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 = <Event>[];
+          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<List<String>> getChannelCids() async => _channelCids;
+
+  @override
+  Future<void> saveChannelQueries({
+    required List<String> cids,
+    Filter? filter,
+    SortOrder<ChannelState>? sort,
+    String? predefinedFilter,
+    Filter? resolvedFilter,
+    SortOrder<ChannelState>? resolvedSort,
+    Map<String, Object?>? filterValues,
+    Map<String, Object?>? sortValues,
+    bool clearQueryCache = false,
+  }) async {}
+
+  @override
+  Future<void> updateChannelStates(List<ChannelState> channelStates) async {}
 }
 
 class FakeChatApi extends Fake implements StreamChatApi {