refactor(providers): share coalesced load scheduling

lib/providers/discover_provider.dart

@@ -12,6 +12,7 @@ import '../services/settings_service.dart';
 import '../services/data_aggregation_service.dart';
 import '../services/system_shelf_service.dart';
 import '../utils/app_logger.dart';
+import '../utils/coalesced_load_coordinator.dart';
 import '../utils/deletion_notifier.dart';
 import '../utils/global_key_utils.dart';
 import '../utils/media_hub_ordering.dart';
@@ -42,6 +43,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
   static const int _continueWatchingProbeLimit = continueWatchingPreviewLimit + 1;
 
   DiscoverProvider(this._multiServer, this._hiddenLibraries, this._libraries, {required this.isProfileBinding}) {
+    _loadCoordinator = CoalescedLoadCoordinator<String>(onFull: _loadOnce, onDelta: _loadDeltaOnce);
     // Late server connects (reconnect after outage, slow wave) refresh
     // discover the same way they refresh libraries. Removed in [dispose] so a
     // profile switch can't leave a stale listener on the app-global provider.
@@ -103,12 +105,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
 
   Set<String> get _fullyLoadedServerIds => _loadedOnDeckServerIds.intersection(_loadedHubServerIds);
 
-  Future<void>? _inFlightLoad;
-  bool _hasPendingLoad = false;
-
-  /// Newly-online servers queued for a delta pass — fetched and merged
-  /// without repeating the full multi-server fan-out.
-  final Set<String> _pendingDeltaServerIds = {};
+  late final CoalescedLoadCoordinator<String> _loadCoordinator;
 
   Future<void>? _systemShelfSyncFuture;
   List<MediaItem>? _pendingSystemShelfItems;
@@ -147,8 +144,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
     }
     // Nothing (or a failed pass) to merge into yet — run the full load.
     if (_onDeckState != DiscoverLoadState.loaded || _hubsState != DiscoverLoadState.loaded) return load();
-    _pendingDeltaServerIds.addAll(onlineServerIds.difference(_fullyLoadedServerIds));
-    return _ensureLoadLoop();
+    return _loadCoordinator.requestDelta(onlineServerIds.difference(_fullyLoadedServerIds));
   }
 
   /// Full load of Continue Watching + hubs. Concurrent calls coalesce into
@@ -156,29 +152,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
   /// arrives mid-load still observes its own fresh fetch).
   Future<void> load() {
     if (isDisposed) return Future<void>.value();
-    _hasPendingLoad = true;
-    return _ensureLoadLoop();
-  }
-
-  Future<void> _ensureLoadLoop() {
-    if (isDisposed) return Future<void>.value();
-    return _inFlightLoad ??= _runLoadLoop().whenComplete(() {
-      if (!isDisposed) _inFlightLoad = null;
-    });
-  }
-
-  Future<void> _runLoadLoop() async {
-    while ((_hasPendingLoad || _pendingDeltaServerIds.isNotEmpty) && !isDisposed) {
-      if (_hasPendingLoad) {
-        _hasPendingLoad = false;
-        _pendingDeltaServerIds.clear(); // a full pass covers every server
-        await _loadOnce();
-      } else {
-        final ids = Set<String>.of(_pendingDeltaServerIds);
-        _pendingDeltaServerIds.clear();
-        await _loadDeltaOnce(ids);
-      }
-    }
+    return _loadCoordinator.requestFull();
   }
 
   Future<void> _loadOnce() async {
@@ -659,8 +633,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
     _watchStateSubscription = null;
     _deletionSubscription?.cancel();
     _deletionSubscription = null;
-    _hasPendingLoad = false;
-    _pendingDeltaServerIds.clear();
+    _loadCoordinator.dispose();
     _pendingSystemShelfItems = null;
     super.dispose();
   }

lib/providers/libraries_provider.dart

@@ -5,6 +5,7 @@ import '../mixins/disposable_change_notifier_mixin.dart';
 import '../services/data_aggregation_service.dart';
 import '../services/storage_service.dart';
 import '../utils/app_logger.dart';
+import '../utils/coalesced_load_coordinator.dart';
 import 'multi_server_provider.dart';
 
 /// Load state for the libraries provider
@@ -16,6 +17,12 @@ enum LibrariesLoadState { initial, loading, loaded, error }
 class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixin {
   LibrariesProvider({this._storageService, this._multiServer, bool Function()? isProfileBinding})
     : _isProfileBinding = isProfileBinding ?? _neverBinding {
+    _loadCoordinator = CoalescedLoadCoordinator<String>(
+      onFull: () async {
+        await _loadLibrariesInternal();
+      },
+      onDelta: _loadDelta,
+    );
     // Reload libraries when a new server comes online. Servers bind in waves
     // on sign-in / profile switch and slow ones reconnect after the initial
     // load; without this they stay missing from the sidebar until a re-switch
@@ -40,9 +47,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
   LibrariesLoadState _loadState = LibrariesLoadState.initial;
   String? _errorMessage;
 
-  /// Coalesces concurrent `loadLibraries()` calls so two simultaneous callers
-  /// see the same in-flight result instead of racing two separate fetches.
-  Future<void>? _inFlightLoad;
+  late final CoalescedLoadCoordinator<String> _loadCoordinator;
 
   /// Server ids whose library fetch *succeeded* in the current [_libraries], as
   /// reported by [DataAggregationService.getMediaLibrariesFromAllServers].
@@ -53,15 +58,6 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
   /// [syncToOnlineServers].
   Set<String> _loadedServerIds = {};
 
-  /// Set when a (re)load is requested while one is already in flight, so the
-  /// loop runs another pass: a server that comes online *during* a load would
-  /// otherwise be lost to the coalesced [_inFlightLoad].
-  bool _hasPendingLoad = false;
-
-  /// Newly-online servers queued for a delta pass — fetched and merged
-  /// without refetching the already-loaded servers.
-  final Set<String> _pendingDeltaServerIds = {};
-
   /// Unmodifiable list of all libraries (ordered)
   List<MediaLibrary> get libraries => List.unmodifiable(_libraries);
 
@@ -108,8 +104,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
     }
     // Nothing (or a failed pass) to merge into yet — run the full load.
     if (_loadState != LibrariesLoadState.loaded) return _load();
-    _pendingDeltaServerIds.addAll(onlineServerIds.difference(_loadedServerIds));
-    return _ensureLoadLoop();
+    return _loadCoordinator.requestDelta(onlineServerIds.difference(_loadedServerIds));
   }
 
   /// Load libraries from all connected servers, unconditionally. Used by
@@ -117,38 +112,11 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
   /// Applies saved ordering.
   Future<void> loadLibraries() => _load();
 
-  /// Single entry point for every full (re)load. Concurrent callers coalesce
-  /// onto one in-flight pass; a request that arrives mid-pass is replayed by
-  /// [_runLoadLoop] so it isn't masked by that coalescing. Each pass fetches
-  /// whatever is online at fetch time, so no caller needs to specify a target.
+  /// Single entry point for every full (re)load. Each pass fetches whatever is
+  /// online at fetch time, so no caller needs to specify a target.
   Future<void> _load() {
     if (isDisposed) return Future<void>.value();
-    _hasPendingLoad = true;
-    return _ensureLoadLoop();
-  }
-
-  Future<void> _ensureLoadLoop() {
-    if (isDisposed) return Future<void>.value();
-    return _inFlightLoad ??= _runLoadLoop().whenComplete(() {
-      if (!isDisposed) _inFlightLoad = null;
-    });
-  }
-
-  Future<void> _runLoadLoop() async {
-    while ((_hasPendingLoad || _pendingDeltaServerIds.isNotEmpty) && !isDisposed) {
-      if (_hasPendingLoad) {
-        _hasPendingLoad = false;
-        _pendingDeltaServerIds.clear(); // a full pass covers every server
-        final succeeded = await _loadLibrariesInternal();
-        // A failure does not enqueue itself, but a caller that arrived during
-        // the failed pass still owns one coalesced trailing attempt.
-        if (!succeeded && !_hasPendingLoad && _pendingDeltaServerIds.isEmpty) return;
-      } else {
-        final ids = Set<String>.of(_pendingDeltaServerIds);
-        _pendingDeltaServerIds.clear();
-        await _loadDelta(ids);
-      }
-    }
+    return _loadCoordinator.requestFull();
   }
 
   /// Fetch libraries from [serverIds] only (servers that came online after
@@ -315,8 +283,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
     _loadState = LibrariesLoadState.initial;
     _errorMessage = null;
     _loadedServerIds = {};
-    _hasPendingLoad = false;
-    _pendingDeltaServerIds.clear();
+    _loadCoordinator.clearPending();
     safeNotifyListeners();
     appLogger.d('LibrariesProvider: Cleared library data');
   }
@@ -324,8 +291,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
   @override
   void dispose() {
     _multiServer?.removeOnlineServersListener(syncToOnlineServers);
-    _hasPendingLoad = false;
-    _pendingDeltaServerIds.clear();
+    _loadCoordinator.dispose();
     super.dispose();
   }
 

lib/utils/coalesced_load_coordinator.dart

@@ -0,0 +1,86 @@
+import 'dart:async';
+
+/// Coalesces full and targeted delta loads behind one in-flight drain.
+///
+/// A full load takes priority and supersedes every queued delta. Requests made
+/// while a pass is running share the drain future and are replayed as trailing
+/// work. The callbacks own fetch, commit, and failure policy.
+final class CoalescedLoadCoordinator<T> {
+  factory CoalescedLoadCoordinator({
+    required Future<void> Function() onFull,
+    required Future<void> Function(Set<T>) onDelta,
+  }) => CoalescedLoadCoordinator._(onFull, onDelta);
+
+  CoalescedLoadCoordinator._(this._onFull, this._onDelta);
+
+  final Future<void> Function() _onFull;
+  final Future<void> Function(Set<T>) _onDelta;
+
+  final Set<T> _pendingDelta = {};
+  Future<void>? _inFlight;
+  bool _pendingFull = false;
+  bool _disposed = false;
+
+  Future<void> requestFull() {
+    if (_disposed) return Future<void>.value();
+    _pendingFull = true;
+    return _ensureDrain();
+  }
+
+  Future<void> requestDelta(Iterable<T> values) {
+    if (_disposed) return Future<void>.value();
+    _pendingDelta.addAll(values);
+    if (_pendingDelta.isEmpty) return _inFlight ?? Future<void>.value();
+    return _ensureDrain();
+  }
+
+  /// Discards trailing work without interrupting the active callback.
+  void clearPending() {
+    if (_disposed) return;
+    _pendingFull = false;
+    _pendingDelta.clear();
+  }
+
+  /// Prevents new work and discards work queued behind the active callback.
+  void dispose() {
+    _disposed = true;
+    _pendingFull = false;
+    _pendingDelta.clear();
+  }
+
+  Future<void> _ensureDrain() {
+    final active = _inFlight;
+    if (active != null) return active;
+
+    // Install the shared future before invoking a callback so a synchronous,
+    // reentrant request is queued behind this drain rather than starting one.
+    final completer = Completer<void>();
+    final future = completer.future;
+    _inFlight = future;
+    _drain().then(
+      (_) {
+        if (!_disposed && identical(_inFlight, future)) _inFlight = null;
+        completer.complete();
+      },
+      onError: (Object error, StackTrace stackTrace) {
+        if (!_disposed && identical(_inFlight, future)) _inFlight = null;
+        completer.completeError(error, stackTrace);
+      },
+    );
+    return future;
+  }
+
+  Future<void> _drain() async {
+    while ((_pendingFull || _pendingDelta.isNotEmpty) && !_disposed) {
+      if (_pendingFull) {
+        _pendingFull = false;
+        _pendingDelta.clear();
+        await _onFull();
+      } else {
+        final values = Set<T>.of(_pendingDelta);
+        _pendingDelta.clear();
+        await _onDelta(values);
+      }
+    }
+  }
+}

test/providers/discover_provider_test.dart

@@ -634,6 +634,34 @@ void main() {
     expect(aggregation.onDeckCalls, callsAfterDelta);
   });
 
+  test('online-server deltas arriving mid-pass are unioned into one trailing pass', () async {
+    aggregation.onDeckResult = () => [_item('a')];
+    aggregation.hubsResult = () => [_hub('hub-1')];
+    await provider.load();
+    final onDeckCallsBefore = aggregation.onDeckCalls;
+    final hubCallsBefore = aggregation.hubCalls;
+
+    final gate = Completer<void>();
+    aggregation.onDeckGate = gate.future;
+    aggregation.hubGate = gate.future;
+    aggregation.onDeckStarted = Completer<void>();
+    aggregation.hubStarted = Completer<void>();
+    aggregation.onDeckResult = () => [_item('new', serverId: 'server_2')];
+    aggregation.hubsResult = () => [_hub('new-hub', serverId: 'server_2')];
+
+    final firstDelta = provider.syncToOnlineServers({'server_1', 'server_2'});
+    await Future.wait([aggregation.onDeckStarted!.future, aggregation.hubStarted!.future]);
+    final trailingDelta = provider.syncToOnlineServers({'server_1', 'server_2', 'server_3'});
+
+    gate.complete();
+    await Future.wait([firstDelta, trailingDelta]);
+
+    expect(aggregation.onDeckCalls, onDeckCallsBefore + 2);
+    expect(aggregation.hubCalls, hubCallsBefore + 2);
+    expect(aggregation.lastOnDeckServerIds, {'server_3'});
+    expect(aggregation.lastHubsServerIds, {'server_3'});
+  });
+
   test('full load partial hub failure retries hubs without refetching continue watching', () async {
     aggregation.onDeckResult = () => [_item('a')];
     aggregation.hubsResult = () => const [];

test/providers/libraries_provider_test.dart

@@ -350,6 +350,39 @@ void main() {
       manager.dispose();
     });
 
+    test('online-server deltas arriving mid-pass are unioned into one trailing pass', () async {
+      final manager = MultiServerManager();
+      final clientA = _FakeClient(serverId: ServerId('A'), libraries: [_serverLib(ServerId('A'), '1', 'Movies A')]);
+      manager.debugRegisterClientForTesting(clientA);
+      final p = LibrariesProvider()..initialize(DataAggregationService(manager));
+      await p.syncToOnlineServers({'A'});
+
+      final gate = Completer<void>();
+      final clientB = _FakeClient(
+        serverId: ServerId('B'),
+        libraries: [_serverLib(ServerId('B'), '1', 'Shows B')],
+        gate: gate.future,
+      );
+      manager.debugRegisterClientForTesting(clientB);
+      final firstDelta = p.syncToOnlineServers({'A', 'B'});
+      expect(clientB.fetchLibrariesCalls, 1);
+
+      final clientC = _FakeClient(serverId: ServerId('C'), libraries: [_serverLib(ServerId('C'), '1', 'Movies C')]);
+      manager.debugRegisterClientForTesting(clientC);
+      final trailingDelta = p.syncToOnlineServers({'A', 'B', 'C'});
+
+      gate.complete();
+      await Future.wait([firstDelta, trailingDelta]);
+
+      expect(clientA.fetchLibrariesCalls, 1);
+      expect(clientB.fetchLibrariesCalls, 1, reason: 'the trailing delta drops ids committed by the active pass');
+      expect(clientC.fetchLibrariesCalls, 1);
+      expect(p.libraries.map((library) => library.title), containsAll(['Movies A', 'Shows B', 'Movies C']));
+
+      p.dispose();
+      manager.dispose();
+    });
+
     test('a coalesced call gets its trailing pass after the in-flight pass fails', () async {
       final manager = MultiServerManager();
       final gate = Completer<void>();

test/utils/coalesced_load_coordinator_test.dart

@@ -0,0 +1,127 @@
+import 'dart:async';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:plezy/utils/coalesced_load_coordinator.dart';
+
+void main() {
+  test('full requests share one drain and schedule one trailing pass', () async {
+    final gates = [Completer<void>(), Completer<void>()];
+    var fullCalls = 0;
+    final coordinator = CoalescedLoadCoordinator<String>(
+      onFull: () async {
+        final gate = gates[fullCalls++];
+        await gate.future;
+      },
+      onDelta: (_) async {},
+    );
+
+    final first = coordinator.requestFull();
+    final second = coordinator.requestFull();
+    final third = coordinator.requestFull();
+
+    expect(identical(first, second), isTrue);
+    expect(identical(first, third), isTrue);
+    expect(fullCalls, 1);
+
+    gates.first.complete();
+    await Future<void>.delayed(Duration.zero);
+    expect(fullCalls, 2);
+
+    gates.last.complete();
+    await first;
+    expect(fullCalls, 2);
+  });
+
+  test('delta requests union behind the active pass', () async {
+    final gate = Completer<void>();
+    final passes = <Set<String>>[];
+    final coordinator = CoalescedLoadCoordinator<String>(
+      onFull: () async {},
+      onDelta: (values) async {
+        passes.add(values);
+        if (passes.length == 1) await gate.future;
+      },
+    );
+
+    final drain = coordinator.requestDelta({'a'});
+    unawaited(coordinator.requestDelta({'b'}));
+    unawaited(coordinator.requestDelta({'b', 'c'}));
+    expect(passes, [
+      {'a'},
+    ]);
+
+    gate.complete();
+    await drain;
+    expect(passes, [
+      {'a'},
+      {'b', 'c'},
+    ]);
+  });
+
+  test('queued full takes priority and supersedes queued deltas', () async {
+    final gate = Completer<void>();
+    final passes = <Object>[];
+    final coordinator = CoalescedLoadCoordinator<String>(
+      onFull: () async => passes.add('full'),
+      onDelta: (values) async {
+        passes.add(values);
+        if (passes.length == 1) await gate.future;
+      },
+    );
+
+    final drain = coordinator.requestDelta({'a'});
+    unawaited(coordinator.requestDelta({'b'}));
+    unawaited(coordinator.requestFull());
+    gate.complete();
+    await drain;
+
+    expect(passes, [
+      {'a'},
+      'full',
+    ]);
+  });
+
+  test('dispose cancels trailing and future requests', () async {
+    final gate = Completer<void>();
+    var fullCalls = 0;
+    final coordinator = CoalescedLoadCoordinator<String>(
+      onFull: () async {
+        fullCalls++;
+        await gate.future;
+      },
+      onDelta: (_) async {},
+    );
+
+    final drain = coordinator.requestFull();
+    unawaited(coordinator.requestFull());
+    coordinator.dispose();
+    gate.complete();
+    await drain;
+    await coordinator.requestFull();
+    await coordinator.requestDelta({'a'});
+
+    expect(fullCalls, 1);
+  });
+
+  test('clearPending discards trailing work but remains reusable', () async {
+    final gate = Completer<void>();
+    var fullCalls = 0;
+    final coordinator = CoalescedLoadCoordinator<String>(
+      onFull: () async {
+        fullCalls++;
+        if (fullCalls == 1) await gate.future;
+      },
+      onDelta: (_) async {},
+    );
+
+    final drain = coordinator.requestFull();
+    unawaited(coordinator.requestFull());
+    coordinator.clearPending();
+    gate.complete();
+    await drain;
+    expect(fullCalls, 1);
+
+    await coordinator.requestFull();
+    expect(fullCalls, 2);
+  });
+}