From baf3c41a44fa0ec7525c719d9f76f4be5982afb7 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Sun, 12 Jul 2026 05:57:52 +0200 Subject: [PATCH] refactor(providers): share coalesced load scheduling --- lib/providers/discover_provider.dart | 39 +----- lib/providers/libraries_provider.dart | 62 ++------- lib/utils/coalesced_load_coordinator.dart | 86 ++++++++++++ test/providers/discover_provider_test.dart | 28 ++++ test/providers/libraries_provider_test.dart | 33 +++++ .../coalesced_load_coordinator_test.dart | 127 ++++++++++++++++++ 6 files changed, 294 insertions(+), 81 deletions(-) create mode 100644 lib/utils/coalesced_load_coordinator.dart create mode 100644 test/utils/coalesced_load_coordinator_test.dart diff --git a/lib/providers/discover_provider.dart b/lib/providers/discover_provider.dart index 27724590..277800fa 100644 --- a/lib/providers/discover_provider.dart +++ b/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(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 get _fullyLoadedServerIds => _loadedOnDeckServerIds.intersection(_loadedHubServerIds); - Future? _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 _pendingDeltaServerIds = {}; + late final CoalescedLoadCoordinator _loadCoordinator; Future? _systemShelfSyncFuture; List? _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 load() { if (isDisposed) return Future.value(); - _hasPendingLoad = true; - return _ensureLoadLoop(); - } - - Future _ensureLoadLoop() { - if (isDisposed) return Future.value(); - return _inFlightLoad ??= _runLoadLoop().whenComplete(() { - if (!isDisposed) _inFlightLoad = null; - }); - } - - Future _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.of(_pendingDeltaServerIds); - _pendingDeltaServerIds.clear(); - await _loadDeltaOnce(ids); - } - } + return _loadCoordinator.requestFull(); } Future _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(); } diff --git a/lib/providers/libraries_provider.dart b/lib/providers/libraries_provider.dart index 01836ca7..533980c5 100644 --- a/lib/providers/libraries_provider.dart +++ b/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( + 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? _inFlightLoad; + late final CoalescedLoadCoordinator _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 _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 _pendingDeltaServerIds = {}; - /// Unmodifiable list of all libraries (ordered) List 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 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 _load() { if (isDisposed) return Future.value(); - _hasPendingLoad = true; - return _ensureLoadLoop(); - } - - Future _ensureLoadLoop() { - if (isDisposed) return Future.value(); - return _inFlightLoad ??= _runLoadLoop().whenComplete(() { - if (!isDisposed) _inFlightLoad = null; - }); - } - - Future _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.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(); } diff --git a/lib/utils/coalesced_load_coordinator.dart b/lib/utils/coalesced_load_coordinator.dart new file mode 100644 index 00000000..9d42952a --- /dev/null +++ b/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 { + factory CoalescedLoadCoordinator({ + required Future Function() onFull, + required Future Function(Set) onDelta, + }) => CoalescedLoadCoordinator._(onFull, onDelta); + + CoalescedLoadCoordinator._(this._onFull, this._onDelta); + + final Future Function() _onFull; + final Future Function(Set) _onDelta; + + final Set _pendingDelta = {}; + Future? _inFlight; + bool _pendingFull = false; + bool _disposed = false; + + Future requestFull() { + if (_disposed) return Future.value(); + _pendingFull = true; + return _ensureDrain(); + } + + Future requestDelta(Iterable values) { + if (_disposed) return Future.value(); + _pendingDelta.addAll(values); + if (_pendingDelta.isEmpty) return _inFlight ?? Future.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 _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(); + 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 _drain() async { + while ((_pendingFull || _pendingDelta.isNotEmpty) && !_disposed) { + if (_pendingFull) { + _pendingFull = false; + _pendingDelta.clear(); + await _onFull(); + } else { + final values = Set.of(_pendingDelta); + _pendingDelta.clear(); + await _onDelta(values); + } + } + } +} diff --git a/test/providers/discover_provider_test.dart b/test/providers/discover_provider_test.dart index c98dd8dc..61eef901 100644 --- a/test/providers/discover_provider_test.dart +++ b/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(); + aggregation.onDeckGate = gate.future; + aggregation.hubGate = gate.future; + aggregation.onDeckStarted = Completer(); + aggregation.hubStarted = Completer(); + 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 []; diff --git a/test/providers/libraries_provider_test.dart b/test/providers/libraries_provider_test.dart index 17a250c2..de4793d0 100644 --- a/test/providers/libraries_provider_test.dart +++ b/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(); + 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(); diff --git a/test/utils/coalesced_load_coordinator_test.dart b/test/utils/coalesced_load_coordinator_test.dart new file mode 100644 index 00000000..253e7dab --- /dev/null +++ b/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(), Completer()]; + var fullCalls = 0; + final coordinator = CoalescedLoadCoordinator( + 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.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(); + final passes = >[]; + final coordinator = CoalescedLoadCoordinator( + 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(); + final passes = []; + final coordinator = CoalescedLoadCoordinator( + 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(); + var fullCalls = 0; + final coordinator = CoalescedLoadCoordinator( + 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(); + var fullCalls = 0; + final coordinator = CoalescedLoadCoordinator( + 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); + }); +}