fix(providers): stop loads after disposal

2026-07-12 08:42:22 +02:00
parent 5876a602c3
commit 2ebc1e250d
4 changed files with 198 additions and 18 deletions
@@ -139,7 +139,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
  /// Once a full pass has loaded, only the genuinely new servers are fetched
  /// and merged in; already-loaded servers are not refetched.
  Future<void> syncToOnlineServers(Set<String> onlineServerIds) {
-    if (onlineServerIds.isEmpty || isProfileBinding()) return Future<void>.value();
+    if (isDisposed || onlineServerIds.isEmpty || isProfileBinding()) return Future<void>.value();
    if (_onDeckState == DiscoverLoadState.loaded &&
        _hubsState == DiscoverLoadState.loaded &&
        _fullyLoadedServerIds.containsAll(onlineServerIds)) {
@@ -155,11 +155,17 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
  /// the in-flight pass plus at most one trailing pass (so a request that
  /// 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() => _inFlightLoad ??= _runLoadLoop().whenComplete(() => _inFlightLoad = null);
+  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) {
@@ -180,6 +186,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
    // kicked off during build (the screen's initState) doesn't mark
    // listening widgets dirty mid-build.
    await null;
+    if (isDisposed) return;
    appLogger.d('DiscoverProvider: loading content from all servers');
    _onDeckState = DiscoverLoadState.loading;
    _hubsState = DiscoverLoadState.loading;
@@ -197,6 +204,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
      _lastSeenHiddenKeys = Set.of(_hiddenLibraries.hiddenLibraryKeys);

      final settings = await SettingsService.getInstance();
+      if (isDisposed) return;
      final useGlobalHubs = settings.read(SettingsService.useGlobalHubs);
      final aggregation = _multiServer.aggregationService;

@@ -266,8 +274,8 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
      _loadedHubServerIds = fetchedHubs.succeededServerIds;
      safeNotifyListeners();
    } catch (e) {
-      appLogger.e('Failed to load discover content', error: e);
      if (isDisposed) return;
+      appLogger.e('Failed to load discover content', error: e);
      _errorMessage = e.toString();
      _onDeckState = DiscoverLoadState.error;
      _hubsState = DiscoverLoadState.error;
@@ -292,6 +300,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
      if (isDisposed) return;

      final settings = await SettingsService.getInstance();
+      if (isDisposed) return;
      final useGlobalHubs = settings.read(SettingsService.useGlobalHubs);
      final aggregation = _multiServer.aggregationService;

@@ -347,6 +356,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
      safeNotifyListeners();
      unawaited(_syncSystemShelf(_onDeck));
    } catch (e) {
+      if (isDisposed) return;
      // Keep the loaded state — stale rows beat an error flash.
      appLogger.w('DiscoverProvider: delta load failed for $ids', error: e);
    }
@@ -592,6 +602,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
  /// coalesce: a sync that arrives while one is in flight queues exactly one
  /// follow-up pass with the latest items.
  Future<void> _syncSystemShelf(List<MediaItem> onDeck) async {
+    if (isDisposed) return;
    _pendingSystemShelfItems = List<MediaItem>.unmodifiable(onDeck);
    if (_systemShelfSyncFuture != null) {
      await _systemShelfSyncFuture;
@@ -612,6 +623,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin

        try {
          final settings = await SettingsService.getInstance();
+          if (isDisposed) return;
          final syncableOnDeck = onDeck
              .where((item) {
                final serverId = item.serverId;
@@ -628,7 +640,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
        }
      }
    } finally {
-      _systemShelfSyncFuture = null;
+      if (!isDisposed) _systemShelfSyncFuture = null;
    }
  }

@@ -647,6 +659,8 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin
    _watchStateSubscription = null;
    _deletionSubscription?.cancel();
    _deletionSubscription = null;
+    _hasPendingLoad = false;
+    _pendingDeltaServerIds.clear();
    _pendingSystemShelfItems = null;
    super.dispose();
  }
@@ -83,6 +83,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
  /// Initialize the provider with the aggregation service.
  /// This should be called after server connection is established.
  void initialize(DataAggregationService service) {
+    if (isDisposed) return;
    _aggregationService = service;
  }

@@ -101,7 +102,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
  /// Once a full pass has loaded, only the genuinely new servers are fetched
  /// and merged in; already-loaded servers are not refetched.
  Future<void> syncToOnlineServers(Set<String> onlineServerIds) {
-    if (_aggregationService == null || onlineServerIds.isEmpty) return Future<void>.value();
+    if (isDisposed || _aggregationService == null || onlineServerIds.isEmpty) return Future<void>.value();
    if (_loadState == LibrariesLoadState.loaded && _loadedServerIds.containsAll(onlineServerIds)) {
      return Future<void>.value();
    }
@@ -121,21 +122,27 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
  /// [_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.
  Future<void> _load() {
+    if (isDisposed) return Future<void>.value();
    _hasPendingLoad = true;
    return _ensureLoadLoop();
  }

-  Future<void> _ensureLoadLoop() => _inFlightLoad ??= _runLoadLoop().whenComplete(() => _inFlightLoad = null);
+  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) {
+    while ((_hasPendingLoad || _pendingDeltaServerIds.isNotEmpty) && !isDisposed) {
      if (_hasPendingLoad) {
        _hasPendingLoad = false;
        _pendingDeltaServerIds.clear(); // a full pass covers every server
        final succeeded = await _loadLibrariesInternal();
-        // Stop on failure so a persistently failing fetch can't hot-loop; the
-        // next server-status emission re-drives the sync.
-        if (!succeeded) break;
+        // 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();
@@ -149,12 +156,14 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
  /// the current list and leave the ids un-loaded, so the next status
  /// emission retries them.
  Future<void> _loadDelta(Set<String> serverIds) async {
+    if (isDisposed) return;
    // A full pass may have covered these ids while they sat in the queue.
    final ids = serverIds.difference(_loadedServerIds);
    if (ids.isEmpty) return;

    try {
      final result = await _aggregationService!.getMediaLibrariesFromAllServers(serverIds: ids);
+      if (isDisposed) return;
      final fresh = result.libraries;

      final merged = [
@@ -162,7 +171,12 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
          if (!ids.contains(lib.serverId)) lib,
        ...fresh,
      ];
-      final storage = _storageService ??= await StorageService.getInstance();
+      var storage = _storageService;
+      if (storage == null) {
+        storage = await StorageService.getInstance();
+        if (isDisposed) return;
+        _storageService = storage;
+      }
      _libraries = _applyLibraryOrder(merged, storage.getLibraryOrder());
      // Union *succeeded* ids only, so a server whose fetch failed is retried
      // on the next status emission instead of being cached as loaded.
@@ -171,12 +185,14 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
      appLogger.i('LibrariesProvider: merged ${fresh.length} libraries from $ids');
      safeNotifyListeners();
    } catch (e, stackTrace) {
+      if (isDisposed) return;
      appLogger.e('LibrariesProvider: delta load failed for $ids', error: e, stackTrace: stackTrace);
    }
  }

  /// Returns `true` on a successful load, `false` on error.
  Future<bool> _loadLibrariesInternal() async {
+    if (isDisposed) return false;
    if (_aggregationService == null) {
      appLogger.w('LibrariesProvider: Cannot load libraries - not initialized');
      return false;
@@ -200,6 +216,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
      // The aggregation service converts Plex-typed responses to MediaLibrary
      // internally; Jellyfin clients return MediaLibrary natively.
      final result = await _aggregationService!.getMediaLibrariesFromAllServers();
+      if (isDisposed) return false;

      // A pass in which zero servers succeeded is never authoritative — it
      // must not replace existing data, and it may only commit "loaded,
@@ -224,7 +241,12 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
      }

      // Apply saved library order
-      final storage = _storageService ??= await StorageService.getInstance();
+      var storage = _storageService;
+      if (storage == null) {
+        storage = await StorageService.getInstance();
+        if (isDisposed) return false;
+        _storageService = storage;
+      }
      final savedOrder = storage.getLibraryOrder();
      final orderedLibraries = _applyLibraryOrder(result.libraries, savedOrder);

@@ -242,6 +264,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
      safeNotifyListeners();
      return true;
    } catch (e, stackTrace) {
+      if (isDisposed) return false;
      appLogger.e('LibrariesProvider: Failed to load libraries', error: e, stackTrace: stackTrace);
      // A refresh that fails over an existing list keeps the last good data and
      // `loaded` state instead of blanking to an error screen; the next status
@@ -256,6 +279,7 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi

  /// Refresh libraries by reloading from the connected servers.
  Future<void> refresh() async {
+    if (isDisposed) return;
    if (_aggregationService == null) {
      appLogger.w('LibrariesProvider: Cannot refresh - not initialized');
      return;
@@ -265,19 +289,28 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi

  /// Update the library order and persist it.
  Future<void> updateLibraryOrder(List<MediaLibrary> orderedLibraries) async {
+    if (isDisposed) return;
    _libraries = List.from(orderedLibraries);
    safeNotifyListeners();

    // Save the new order
-    final storage = _storageService ??= await StorageService.getInstance();
+    var storage = _storageService;
+    if (storage == null) {
+      storage = await StorageService.getInstance();
+      if (isDisposed) return;
+      _storageService = storage;
+    }
+    if (isDisposed) return;
    final libraryKeys = orderedLibraries.map((lib) => lib.globalKey).toList();
    await storage.saveLibraryOrder(libraryKeys);

+    if (isDisposed) return;
    appLogger.d('LibrariesProvider: Updated library order');
  }

  /// Clear all library data (for profile switch or logout).
  void clear() {
+    if (isDisposed) return;
    _libraries = [];
    _loadState = LibrariesLoadState.initial;
    _errorMessage = null;
@@ -291,6 +324,8 @@ class LibrariesProvider extends ChangeNotifier with DisposableChangeNotifierMixi
  @override
  void dispose() {
    _multiServer?.removeOnlineServersListener(syncToOnlineServers);
+    _hasPendingLoad = false;
+    _pendingDeltaServerIds.clear();
    super.dispose();
  }

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:flutter_test/flutter_test.dart';
 import 'package:plezy/media/ids.dart';
 import 'package:plezy/media/media_backend.dart';
@@ -70,6 +72,10 @@ class _FakeAggregationService extends DataAggregationService {
  Set<String> hubCancelledServerIds = const {};
  List<MediaItem> Function() onDeckResult = () => const [];
  List<MediaHub> Function() hubsResult = () => const [];
+  Future<void>? onDeckGate;
+  Future<void>? hubGate;
+  Completer<void>? onDeckStarted;
+  Completer<void>? hubStarted;

  @override
  Future<OnDeckAggregationResult> getOnDeckFromAllServers({
@@ -79,6 +85,10 @@ class _FakeAggregationService extends DataAggregationService {
  }) async {
    onDeckCalls++;
    lastOnDeckServerIds = serverIds;
+    final started = onDeckStarted;
+    if (started != null && !started.isCompleted) started.complete();
+    final gate = onDeckGate;
+    if (gate != null) await gate;
    final items = onDeckResult();
    return (
      items: limit != null && items.length > limit ? items.sublist(0, limit) : items,
@@ -97,6 +107,10 @@ class _FakeAggregationService extends DataAggregationService {
  }) async {
    hubCalls++;
    lastHubsServerIds = serverIds;
+    final started = hubStarted;
+    if (started != null && !started.isCompleted) started.complete();
+    final gate = hubGate;
+    if (gate != null) await gate;
    return (
      hubs: hubsResult(),
      succeededServerIds: hubSucceededServerIds ?? serverIds ?? const {'server_1'},
@@ -177,6 +191,57 @@ void main() {
    expect(aggregation.hubCalls, 2);
  });

+  test('a failed in-flight pass still runs one coalesced trailing pass', () async {
+    final gate = Completer<void>();
+    aggregation.onDeckGate = gate.future;
+    aggregation.onDeckStarted = Completer<void>();
+    aggregation.onDeckResult = () {
+      if (aggregation.onDeckCalls == 1) throw Exception('first pass failed');
+      return [_item('recovered')];
+    };
+    aggregation.hubsResult = () => [_hub('hub-1')];
+
+    final first = provider.load();
+    await aggregation.onDeckStarted!.future;
+    final coalesced = provider.load();
+    gate.complete();
+    await Future.wait([first, coalesced]);
+
+    expect(aggregation.onDeckCalls, 2);
+    expect(aggregation.hubCalls, 2);
+    expect(provider.onDeck.map((item) => item.id), ['recovered']);
+    expect(provider.errorMessage, isNull);
+  });
+
+  test('dispose during an in-flight coalesced load prevents trailing work and commits', () async {
+    final scoped = DiscoverProvider(multiServer, hiddenLibraries, libraries, isProfileBinding: () => isBinding);
+    final gate = Completer<void>();
+    aggregation.onDeckGate = gate.future;
+    aggregation.hubGate = gate.future;
+    aggregation.onDeckStarted = Completer<void>();
+    aggregation.hubStarted = Completer<void>();
+    aggregation.onDeckResult = () => [_item('late')];
+    aggregation.hubsResult = () => [_hub('late-hub')];
+
+    final first = scoped.load();
+    await Future.wait([aggregation.onDeckStarted!.future, aggregation.hubStarted!.future]);
+    final coalesced = scoped.load();
+    scoped.dispose();
+    gate.complete();
+    await Future.wait([first, coalesced]);
+
+    expect(aggregation.onDeckCalls, 1);
+    expect(aggregation.hubCalls, 1);
+    expect(scoped.onDeck, isEmpty);
+    expect(scoped.hubs, isEmpty);
+    expect(scoped.loadGeneration, 0);
+
+    await scoped.load();
+    await scoped.syncToOnlineServers({'server_1'});
+    expect(aggregation.onDeckCalls, 1);
+    expect(aggregation.hubCalls, 1);
+  });
+
  test('limits the preview row and probes for more', () async {
    aggregation.onDeckResult = () => [for (var i = 0; i < 30; i++) _item('item-$i')];

@@ -31,7 +31,7 @@ MediaLibrary _serverLib(ServerId serverId, String id, String title) =>
 /// hold `fetchLibraries` open to exercise the mid-load race; setting [error]
 /// makes `fetchLibraries` throw, simulating a (possibly transient) failure.
 class _FakeClient implements MediaServerClient {
-  _FakeClient({required this.serverId, this.libraries = const [], this.gate});
+  _FakeClient({required this.serverId, this.libraries = const [], this.gate, this.errorForCall});

  @override
  final ServerId serverId;
@@ -40,6 +40,7 @@ class _FakeClient implements MediaServerClient {

  final List<MediaLibrary> libraries;
  final Future<void>? gate;
+  final Object? Function(int call)? errorForCall;

  /// When non-null, [fetchLibraries] throws this instead of returning. Mutable
  /// so a test can fail a fetch once and then let it recover.
@@ -52,7 +53,8 @@ class _FakeClient implements MediaServerClient {
    fetchLibrariesCalls++;
    final pending = gate;
    if (pending != null) await pending;
-    if (error != null) throw error!;
+    final fetchError = errorForCall?.call(fetchLibrariesCalls) ?? error;
+    if (fetchError != null) throw fetchError;
    return libraries;
  }

@@ -154,13 +156,15 @@ void main() {
      p.dispose();
    });

-    test('safeNotifyListeners after dispose is a no-op', () async {
+    test('mutating methods after dispose are no-ops', () async {
      final p = LibrariesProvider();
      p.dispose();
-      // Post-dispose clear / updateLibraryOrder must not throw — the provider
-      // uses `safeNotifyListeners` which swallows post-dispose firings.
      p.clear();
      await p.updateLibraryOrder([_lib('1', serverId: ServerId('srv'))]);
+
+      expect(p.libraries, isEmpty);
+      final storage = await StorageService.getInstance();
+      expect(storage.getLibraryOrder(), isNull);
    });
  });

@@ -346,6 +350,68 @@ void main() {
      manager.dispose();
    });

+    test('a coalesced call gets its trailing pass after the in-flight pass fails', () async {
+      final manager = MultiServerManager();
+      final gate = Completer<void>();
+      final clientA = _FakeClient(
+        serverId: ServerId('A'),
+        libraries: [_serverLib(ServerId('A'), '1', 'Movies A')],
+        gate: gate.future,
+        errorForCall: (call) => call == 1
+            ? MediaServerHttpException(type: MediaServerHttpErrorType.cancelled, message: 'first pass cancelled')
+            : null,
+      );
+      manager.debugRegisterClientForTesting(clientA);
+      final p = LibrariesProvider()..initialize(DataAggregationService(manager));
+
+      final first = p.loadLibraries();
+      final coalesced = p.loadLibraries();
+      expect(clientA.fetchLibrariesCalls, 1);
+
+      gate.complete();
+      await Future.wait([first, coalesced]);
+
+      expect(clientA.fetchLibrariesCalls, 2);
+      expect(p.hasLoaded, isTrue);
+      expect(p.libraries.map((library) => library.title), ['Movies A']);
+
+      p.dispose();
+      manager.dispose();
+    });
+
+    test('dispose during an in-flight coalesced load prevents trailing work and commits', () async {
+      final manager = MultiServerManager();
+      final gate = Completer<void>();
+      final clientA = _FakeClient(
+        serverId: ServerId('A'),
+        libraries: [_serverLib(ServerId('A'), '1', 'Movies A')],
+        gate: gate.future,
+      );
+      manager.debugRegisterClientForTesting(clientA);
+      final p = LibrariesProvider()..initialize(DataAggregationService(manager));
+      var notifications = 0;
+      p.addListener(() => notifications++);
+
+      final first = p.loadLibraries();
+      final coalesced = p.loadLibraries();
+      expect(clientA.fetchLibrariesCalls, 1);
+      expect(notifications, 1, reason: 'the initial loading state was published before disposal');
+
+      p.dispose();
+      gate.complete();
+      await Future.wait([first, coalesced]);
+
+      expect(clientA.fetchLibrariesCalls, 1, reason: 'the queued trailing pass was discarded');
+      expect(p.libraries, isEmpty, reason: 'the completed fetch was not committed after disposal');
+      expect(notifications, 1);
+
+      await p.loadLibraries();
+      await p.syncToOnlineServers({'A'});
+      expect(clientA.fetchLibrariesCalls, 1, reason: 'post-dispose entry points are no-ops');
+
+      manager.dispose();
+    });
+
    test('clear() resets tracking so the next sync reloads', () async {
      final manager = MultiServerManager();
      final clientA = _FakeClient(serverId: ServerId('A'), libraries: [_serverLib(ServerId('A'), '1', 'Movies A')]);