fix(llc, core): coalesce queryChannels on flaky-network reconnects (#2652)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: xsahil03x

packages/stream_chat/CHANGELOG.md

@@ -4,6 +4,9 @@
 
 - Fixed `StreamChatClient.queryDrafts` not forwarding the `filter` argument to the API.
 
+- Coalesced concurrent `queryChannels` calls with identical parameters via an in-flight cache.
+  Rapid reconnect bursts no longer fire multiple duplicate requests in parallel.
+
 ✅ Added
 
 - Added `SortedListX` and `ListX` extensions on `List` for keyed merge, sorted insert/upsert and
@@ -13,6 +16,10 @@
   requested number. No-op when the channel isn't up to date. Prefer `StreamChannel.pruneOldest`
   from `stream_chat_flutter_core` when available — it also resets top-pagination.
 
+- Added `StreamChatClient.recoverStateOnReconnect` setter (default `true`). Controls whether the
+  client re-queries active channels on WebSocket reconnect. Set to `false` if your app handles its
+  own state recovery — e.g. via channel-list controllers.
+
 🚀 Performance
 
 - Faster channel state updates, especially for read receipts, reactions, and other partial-state

packages/stream_chat/lib/src/client/client.dart

@@ -37,6 +37,7 @@ import 'package:stream_chat/src/core/models/poll_vote.dart';
 import 'package:stream_chat/src/core/models/push_preference.dart';
 import 'package:stream_chat/src/core/models/thread.dart';
 import 'package:stream_chat/src/core/models/user.dart';
+import 'package:stream_chat/src/core/util/in_flight_cache.dart';
 import 'package:stream_chat/src/core/util/utils.dart';
 import 'package:stream_chat/src/db/chat_persistence_client.dart';
 import 'package:stream_chat/src/event_type.dart';
@@ -88,7 +89,8 @@ class StreamChatClient {
         StreamAttachmentFileUploader.new,
     Iterable<Interceptor>? chatApiInterceptors,
     HttpClientAdapter? httpClientAdapter,
-  }) {
+    bool recoverStateOnReconnect = true,
+  }) : _recoverStateOnReconnect = recoverStateOnReconnect {
     logger.info('Initiating new StreamChatClient');
 
     final options = StreamHttpClientOptions(
@@ -192,6 +194,21 @@ class StreamChatClient {
   /// The retry policy options getter
   RetryPolicy get retryPolicy => _retryPolicy;
 
+  /// Whether the client should automatically refresh local state from the
+  /// server when the WebSocket connection recovers.
+  ///
+  /// When `true` (default), the client re-queries the active channels on
+  /// reconnect (capped at 30, ordered by `state.channels.keys`). The set of
+  /// state recovered on reconnect may grow in the future to cover threads,
+  /// reminders, etc.
+  ///
+  /// Setting this to `false` disables that client-level recovery. Consumers
+  /// that opt out are responsible for refreshing their own state when the
+  /// [EventType.connectionRecovered] event fires — for example, by re-running
+  /// their channel list query.
+  set recoverStateOnReconnect(bool value) => _recoverStateOnReconnect = value;
+  bool _recoverStateOnReconnect;
+
   /// By default the Chat client will write all messages with level Warn or
   /// Error to stdout.
   ///
@@ -530,13 +547,17 @@ class StreamChatClient {
       // connection recovered
       final cids = [...state.channels.keys.toSet()];
       if (cids.isNotEmpty) {
-        await queryChannelsOnline(
-          filter: Filter.in_('cid', cids),
-          paginationParams: const PaginationParams(limit: 30),
-        );
-
         // Sync the persistence client if available
         if (persistenceEnabled) await sync(cids: cids);
+
+        // Recover the channels that were active before the connection was lost,
+        // only if the client is configured to do so.
+        if (_recoverStateOnReconnect) {
+          await queryChannelsOnline(
+            filter: Filter.in_('cid', cids),
+            paginationParams: const PaginationParams(limit: 30),
+          );
+        }
       }
 
       handleEvent(Event(
@@ -613,7 +634,7 @@ class StreamChatClient {
     });
   }
 
-  final _queryChannelsStreams = <String, Future<List<Channel>>>{};
+  final _queryChannelsCache = InFlightCache<String, List<Channel>>();
 
   /// Requests channels with a given query.
   Stream<List<Channel>> queryChannels({
@@ -643,19 +664,7 @@ class StreamChatClient {
       paginationParams,
     ]);
 
-    // Return results from cache if available
-    if (_queryChannelsStreams.containsKey(hash)) {
-      try {
-        yield await _queryChannelsStreams[hash]!;
-        return;
-      } catch (e, stk) {
-        logger.severe('Error retrieving cached query results', e, stk);
-        // Cache is invalid, continue with fresh query
-        _queryChannelsStreams.remove(hash);
-      }
-    }
-
-    // Get offline results first
+    // Per-caller offline emit — local persistence, not coalesced.
     var offlineChannels = <Channel>[];
     try {
       offlineChannels = await queryChannelsOffline(
@@ -671,31 +680,30 @@ class StreamChatClient {
     }
 
     try {
-      final newQueryChannelsFuture = queryChannelsOnline(
-        filter: filter,
-        sort: channelStateSort,
-        state: state,
-        watch: watch,
-        presence: presence,
-        memberLimit: memberLimit,
-        messageLimit: messageLimit,
-        paginationParams: paginationParams,
-        waitForConnect: waitForConnect,
-      ).timeout(
-        const Duration(seconds: 30),
-        onTimeout: () {
-          logger.warning('Online channel query timed out');
-          throw TimeoutException('Channel query timed out');
-        },
-      ).whenComplete(() {
-        // Always clean up cache reference when done
-        _queryChannelsStreams.remove(hash);
-      });
-
-      // Store the future in cache
-      _queryChannelsStreams[hash] = newQueryChannelsFuture;
-
-      yield await newQueryChannelsFuture;
+      // Coalesce concurrent identical online queries — concurrent callers
+      // share both success and failure outcomes. See [InFlightCache] for
+      // the lifecycle details.
+      final result = await _queryChannelsCache.run(
+        hash,
+        () => queryChannelsOnline(
+          filter: filter,
+          sort: channelStateSort,
+          state: state,
+          watch: watch,
+          presence: presence,
+          memberLimit: memberLimit,
+          messageLimit: messageLimit,
+          paginationParams: paginationParams,
+          waitForConnect: waitForConnect,
+        ).timeout(
+          const Duration(seconds: 30),
+          onTimeout: () {
+            logger.warning('Online channel query timed out');
+            throw TimeoutException('Channel query timed out');
+          },
+        ),
+      );
+      yield result;
     } catch (e, stk) {
       logger.severe('Error querying channels online', e, stk);
       // Only rethrow if we have no channels to show the user

packages/stream_chat/lib/src/core/util/in_flight_cache.dart

@@ -0,0 +1,39 @@
+import 'dart:async';
+
+/// Coalesces concurrent identical async calls by sharing a single in-flight
+/// [Future] keyed by [K].
+///
+/// When [run] is called and no [Future] is in flight for [key], the [work]
+/// closure runs and its [Future] is cached. Concurrent callers passing the
+/// same [key] receive the cached [Future] and share its eventual outcome —
+/// both success and failure. Sequential callers arriving after the [Future]
+/// settles see an empty cache and start fresh.
+///
+/// Sharing failures is intentional: falling through to a fresh call on
+/// error would defeat the dedup precisely when it matters most (e.g., a
+/// rate-limit storm), turning one rejected request into N.
+///
+/// Mirrors the `DistinctChatApi` / `DistinctCall` design used across the
+/// Android Stream Chat SDK.
+class InFlightCache<K, V> {
+  final _inFlight = <K, Future<V>>{};
+
+  /// Returns the in-flight [Future] for [key] if one exists, otherwise runs
+  /// [work], caches its [Future], and frees the slot when it settles.
+  Future<V> run(K key, Future<V> Function() work) {
+    if (_inFlight[key] case final existing?) return existing;
+
+    // [Future.sync] runs [work] synchronously on the current microtask and
+    // wraps any synchronous throw in a rejected [Future] — so concurrent
+    // callers in the same tick share the same single invocation.
+    final future = Future.sync(work);
+    _inFlight[key] = future;
+
+    // The cleanup chain runs regardless of outcome. The trailing [ignore]
+    // silences unhandled-error reporting on the chained future since
+    // [whenComplete] forwards the original error rather than handling it.
+    future.whenComplete(() => _inFlight.remove(key)).ignore();
+
+    return future;
+  }
+}