fix(queue): merge startup additions before persistence and processing

lib/main.dart

@@ -441,7 +441,12 @@ class _EagerInitializationState extends ConsumerState<EagerInitialization>
       // any debounced download-queue persistence reaches disk.
       if (ref.exists(downloadQueueProvider)) {
         unawaited(
-          ref.read(downloadQueueProvider.notifier).flushQueuePersistence(),
+          ref
+              .read(downloadQueueProvider.notifier)
+              .flushQueuePersistence()
+              .catchError((Object error) {
+                _log.e('Failed to flush background download queue: $error');
+              }),
         );
       }
       // Backgrounded: release idle native runtimes and connections.

lib/providers/download_queue_provider.dart

@@ -349,7 +349,10 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
   int _failedInSession = 0;
   int _queueItemSequence = 0;
   bool _isLoaded = false;
-  final Completer<void> _queueRestored = Completer<void>();
+  Completer<void> _queueRestored = Completer<void>();
+  bool _queueStartupPending = false;
+  bool _queueStorageReady = true;
+  bool? _startupPausedOverride;
   bool _foregroundResumeScheduled = false;
   bool _iosBackgroundExecutionExpired = false;
   StreamSubscription<List<String>>? _iosBackgroundExpirationSubscription;
@@ -423,23 +426,59 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
       _queuePersistence.dispose();
     });
 
+    _queueStorageReady = false;
+    _startQueueRestore(initializeOutputDir: true);
+    return const DownloadQueueState();
+  }
+
+  void _startQueueRestore({bool initializeOutputDir = false}) {
+    if (_queueStartupPending || _queueStorageReady || !ref.mounted) return;
+    _queueStartupPending = true;
+    final restored = _queueRestored = Completer<void>();
     Future.microtask(() async {
       try {
         if (!ref.mounted) return;
         updateSettings(ref.read(settingsProvider));
-        try {
-          await _initOutputDir();
-        } catch (error, stack) {
-          _log.e('Failed to initialize download folder: $error', error, stack);
+        if (initializeOutputDir) {
+          try {
+            await _initOutputDir();
+          } catch (error, stack) {
+            _log.e(
+              'Failed to initialize download folder: $error',
+              error,
+              stack,
+            );
+          }
         }
         await _loadQueueFromStorage();
       } catch (error, stack) {
         _log.e('Failed to initialize download queue: $error', error, stack);
       } finally {
-        if (!_queueRestored.isCompleted) _queueRestored.complete();
+        if (ref.mounted && _queueStorageReady) {
+          // Native adoption can publish an older paused snapshot after a user
+          // has changed this choice while its platform calls were awaiting.
+          final paused = _startupPausedOverride;
+          if (paused != null && state.isPaused != paused) {
+            if (paused && state.isProcessing) {
+              pauseQueue(persistAcrossRestarts: false);
+            } else if (!paused) {
+              resumeQueue();
+            } else {
+              state = state.copyWith(isPaused: true);
+            }
+          }
+          _startupPausedOverride = null;
+        }
+        _queueStartupPending = false;
+        if (ref.mounted && _queueStorageReady) {
+          _saveQueueToStorage();
+          if (!state.isProcessing && !state.isPaused && state.queuedCount > 0) {
+            Future.microtask(_processQueue);
+          }
+        }
+        if (!restored.isCompleted) restored.complete();
       }
     });
-    return const DownloadQueueState();
   }
 
   /// Flush any debounced queue persistence to disk immediately. Called when
@@ -450,13 +489,19 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
     // Startup publishes restored items before completing this gate. Capturing
     // the initial empty state while hydration is in flight would delete the
     // restored rows when the persistence writer reaches that empty snapshot.
+    if (!_queueStorageReady) _startQueueRestore();
     await _queueRestored.future;
     if (!ref.mounted) return;
+    if (!_queueStorageReady) {
+      throw StateError('Download queue restoration failed; retry the flush');
+    }
     await _queuePersistence.flush();
   }
 
-  void _persistUserPausedQueue(bool paused) =>
-      _queuePersistence.setUserPaused(paused);
+  void _persistUserPausedQueue(bool paused) {
+    if (_queueStartupPending) _startupPausedOverride = paused;
+    _queuePersistence.setUserPaused(paused);
+  }
 
   /// Restarts a queue that was deliberately left pending because Android did
   /// not allow a foreground service to be launched while the app was hidden.
@@ -536,7 +581,11 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
     } else if (!state.isProcessing) {
       state = state.copyWith(isPaused: true, currentDownload: null);
     }
-    unawaited(flushQueuePersistence());
+    unawaited(
+      flushQueuePersistence().catchError((Object error) {
+        _log.e('Failed to flush expired iOS queue: $error');
+      }),
+    );
     // The native expiration callback cancels Go synchronously before this
     // asynchronous event reaches Dart. If foregrounding won that race, still
     // requeue the cancelled item, then immediately let the running queue loop
@@ -548,7 +597,6 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
 
   Future<void> _loadQueueFromStorage() async {
     if (_isLoaded || !ref.mounted) return;
-    _isLoaded = true;
 
     try {
       final restorePaused = await _queuePersistence.loadUserPaused();
@@ -560,19 +608,27 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
         return service == item.service ? item : item.copyWith(service: service);
       }).toList();
 
+      final normalizedPendingItems = _normalizeRestoredQueueIds(pendingItems);
+      state = state.copyWith(
+        // Additions remain visible immediately while the database and codec
+        // worker are awaiting. Append them without deduplicating track IDs:
+        // distinct requests/quality variants are deliberately separate items.
+        items: [...normalizedPendingItems, ...state.items],
+        isPaused:
+            _startupPausedOverride ??
+            (normalizedPendingItems.isNotEmpty && restorePaused),
+      );
+      _isLoaded = true;
+      _queueStorageReady = true;
+      _saveQueueToStorage();
       if (pendingItems.isEmpty) {
-        if (restorePaused) _persistUserPausedQueue(false);
+        if (restorePaused && _startupPausedOverride == null) {
+          _queuePersistence.setUserPaused(false);
+        }
         _log.d('No pending items to restore');
         await _queuePersistence.flush();
         return;
       }
-
-      final normalizedPendingItems = _normalizeRestoredQueueIds(pendingItems);
-      state = state.copyWith(
-        items: normalizedPendingItems,
-        isPaused: restorePaused,
-      );
-      _saveQueueToStorage();
       _log.i(
         'Restored ${normalizedPendingItems.length} pending items from storage',
       );
@@ -580,9 +636,7 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
         return;
       }
       if (!ref.mounted) return;
-      if (!restorePaused) {
-        Future.microtask(() => _processQueue());
-      } else {
+      if (state.isPaused) {
         _log.i('Restored queue in user-paused state');
       }
     } catch (e) {
@@ -590,7 +644,16 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
     }
   }
 
-  void _saveQueueToStorage() => _queuePersistence.schedule();
+  void _saveQueueToStorage() {
+    // A partial startup snapshot must never enter the writer's debounce or
+    // dispose path: it would otherwise delete the restored IDs later on.
+    if (!_queueStorageReady) {
+      _startQueueRestore();
+      return;
+    }
+    if (_queueStartupPending) return;
+    _queuePersistence.schedule();
+  }
 
   bool _isSafMode(AppSettings settings) {
     return Platform.isAndroid &&
@@ -1157,6 +1220,14 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
     bool persistAcrossRestarts = true,
     Set<String> nativeCancelledItemIds = const {},
   }) {
+    if (_queueStartupPending) {
+      _startupPausedOverride = true;
+      if (!state.isProcessing) {
+        state = state.copyWith(isPaused: true, currentDownload: null);
+        if (persistAcrossRestarts) _persistUserPausedQueue(true);
+        return;
+      }
+    }
     if (state.isProcessing && !state.isPaused) {
       final nativeWorkerActive = _hasActiveAndroidNativeWorker;
       if (nativeWorkerActive) {
@@ -1196,6 +1267,11 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
   }
 
   void resumeQueue() {
+    if (_queueStartupPending || !_queueStorageReady) {
+      _startupPausedOverride = false;
+      _persistUserPausedQueue(false);
+      _startQueueRestore();
+    }
     if (state.isPaused) {
       if (_hasActiveAndroidNativeWorker) {
         PlatformBridge.resumeNativeDownloadWorker().catchError((_) {});
@@ -1509,6 +1585,14 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
   }
 
   Future<void> _processQueue() async {
+    // Wait before acquiring the single-flight processing lock. Restoration
+    // may adopt an existing native run and must be free to finish that work.
+    if (_queueStartupPending || !_queueStorageReady) {
+      _startQueueRestore();
+      await _queueRestored.future;
+      if (!ref.mounted || !_queueStorageReady || state.isPaused) return;
+    }
+    if (!ref.mounted) return;
     if (!_queueProcessingGate.tryEnter()) return;
 
     try {

test/download_queue_startup_flush_test.dart

@@ -12,6 +12,14 @@ import 'package:spotiflac_android/providers/download_queue_provider.dart';
 import 'package:spotiflac_android/providers/settings_provider.dart';
 import 'package:spotiflac_android/services/app_state_database.dart';
 
+const _earlyTrack = Track(
+  id: 'example:early',
+  name: 'Immediate addition',
+  artistName: 'Artist',
+  albumName: 'Album',
+  duration: 180,
+);
+
 class _Settings extends SettingsNotifier {
   @override
   AppSettings build() =>
@@ -21,7 +29,7 @@ class _Settings extends SettingsNotifier {
 void main() {
   TestWidgetsFlutterBinding.ensureInitialized();
   test(
-    'background flush waits for restored provider state and skips a disposed owner',
+    'startup restores, merges and retries only complete owned snapshots',
     () async {
       SharedPreferences.setMockInitialValues({
         'app_state_migrated_queue_to_sqlite_v1': true,
@@ -70,6 +78,7 @@ void main() {
       var queueReads = 0;
       var deletes = 0;
       var writes = 0;
+      var failRead = false;
       messenger.setMockMethodCallHandler(sqlite, (call) async {
         if (call.method == 'openDatabase') return {'id': 1};
         final arguments = call.arguments as Map? ?? {};
@@ -90,6 +99,9 @@ void main() {
           final snapshot = rows.values.toList();
           started.complete();
           await release.future;
+          if (failRead) {
+            throw PlatformException(code: 'sqlite_error', message: 'Try again');
+          }
           const columns = [
             'id',
             'item_json',
@@ -179,6 +191,208 @@ void main() {
       expect([writes, deletes], [0, 0]);
       release.complete();
       await Future<void>.delayed(const Duration(milliseconds: 20));
+
+      // Both public add paths publish immediately, but neither processing nor
+      // the 350 ms debounce may act on this incomplete startup snapshot.
+      started = Completer();
+      release = Completer();
+      final additions = scope();
+      final additionsQueue = additions.read(downloadQueueProvider.notifier);
+      await started.future;
+      final singleId = additionsQueue.addToQueue(
+        items.first.track,
+        '',
+        qualityOverride: 'lossless',
+      );
+      additionsQueue.addMultipleToQueue(
+        [items.first.track, _earlyTrack],
+        '',
+        qualityOverride: 'high',
+        playlistName: 'Immediate playlist',
+        playlistPositions: [7, 9],
+      );
+      final earlyItems = additions.read(downloadQueueProvider).items.toList();
+      expect(earlyItems.length, 3);
+      expect(earlyItems.first.id, singleId);
+      expect(earlyItems.map((item) => item.qualityOverride), [
+        'lossless',
+        'high',
+        'high',
+      ]);
+      final additionsFlush = additionsQueue.flushQueuePersistence();
+      await Future<void>.delayed(const Duration(milliseconds: 400));
+      expect(additions.read(downloadQueueProvider).isProcessing, false);
+      expect(
+        additions.read(downloadQueueProvider).items.map((item) => item.status),
+        everyElement(DownloadStatus.queued),
+      );
+      expect([writes, deletes], [0, 0]);
+      release.complete();
+      await additionsFlush;
+      final merged = additions.read(downloadQueueProvider);
+      expect(merged.items.map((item) => item.id), [
+        ...items.map((item) => item.id),
+        ...earlyItems.map((item) => item.id),
+      ]);
+      for (var index = 0; index < earlyItems.length; index++) {
+        expect(merged.items[items.length + index], same(earlyItems[index]));
+      }
+      expect(merged.items.last.playlistPosition, 9);
+      expect(merged.items.last.fromBatch, true);
+      expect(merged.isPaused, true);
+      expect(rows.keys, merged.items.map((item) => item.id));
+      expect([writes, deletes], [3, 0]);
+      additions.dispose();
+
+      // Even disposal cannot flush an early add over the restored disk rows.
+      final savedIds = rows.keys.toList();
+      started = Completer();
+      release = Completer();
+      final earlyDisposed = scope();
+      final earlyDisposedQueue = earlyDisposed.read(
+        downloadQueueProvider.notifier,
+      );
+      await started.future;
+      earlyDisposedQueue.addToQueue(_earlyTrack, '');
+      final disposeFlush = earlyDisposedQueue.flushQueuePersistence();
+      await Future<void>.delayed(const Duration(milliseconds: 400));
+      earlyDisposed.dispose();
+      await disposeFlush;
+      release.complete();
+      await Future<void>.delayed(const Duration(milliseconds: 20));
+      expect(rows.keys, savedIds);
+      expect([writes, deletes], [3, 0]);
+
+      // A failed restore retains the new UI item and existing disk contents.
+      // A subsequent add starts one fresh attempt shared by concurrent flushes.
+      started = Completer();
+      release = Completer();
+      failRead = true;
+      final recovery = scope();
+      final recoveryQueue = recovery.read(downloadQueueProvider.notifier);
+      await started.future;
+      final firstRecoveryId = recoveryQueue.addToQueue(_earlyTrack, '');
+      recoveryQueue.pauseQueue();
+      final failedFlush = expectLater(
+        recoveryQueue.flushQueuePersistence(),
+        throwsA(isA<StateError>()),
+      );
+      release.complete();
+      await failedFlush;
+      await Future<void>.delayed(const Duration(milliseconds: 400));
+      expect(rows.keys, savedIds);
+      expect([writes, deletes], [3, 0]);
+      expect(
+        recovery.read(downloadQueueProvider).items.single.id,
+        firstRecoveryId,
+      );
+      expect(recovery.read(downloadQueueProvider).isProcessing, false);
+
+      started = Completer();
+      release = Completer();
+      failRead = false;
+      final readsBeforeRetry = queueReads;
+      final secondRecoveryId = recoveryQueue.addToQueue(_earlyTrack, '');
+      await started.future;
+      recoveryQueue.resumeQueue();
+      expect(recovery.read(downloadQueueProvider).isPaused, false);
+      final retryFlushes = Future.wait([
+        recoveryQueue.flushQueuePersistence(),
+        recoveryQueue.flushQueuePersistence(),
+      ]);
+      bool? resumedPublication;
+      final subscription = recovery.listen(downloadQueueProvider, (_, next) {
+        if (resumedPublication == null &&
+            next.items.length == savedIds.length + 2) {
+          resumedPublication = next.isPaused;
+          // Pause immediately after observing the merged publication so the
+          // fixture never starts extension/download work after gate release.
+          recoveryQueue.pauseQueue();
+        }
+      });
+      release.complete();
+      await retryFlushes;
+      subscription.close();
+      expect(
+        resumedPublication,
+        false,
+        reason: 'newer resume overrides the persisted paused preference',
+      );
+      final recovered = recovery.read(downloadQueueProvider);
+      expect(recovered.items.map((item) => item.id), [
+        ...savedIds,
+        firstRecoveryId,
+        secondRecoveryId,
+      ]);
+      expect(
+        recovered.isPaused,
+        true,
+        reason: 'the latest pause while publishing also wins',
+      );
+      expect(rows.keys, recovered.items.map((item) => item.id));
+      expect(queueReads, readsBeforeRetry + 1);
+      expect([writes, deletes], [5, 0]);
+      recovery.dispose();
+
+      // Explicit flush and resume also retry a failed startup without needing
+      // another addition. Failed attempts may not touch the saved queue.
+      final recoveredIds = rows.keys.toList();
+      started = Completer();
+      release = Completer();
+      failRead = true;
+      final explicitRetry = scope();
+      final explicitQueue = explicitRetry.read(downloadQueueProvider.notifier);
+      await started.future;
+      explicitQueue.pauseQueue();
+      final firstFailure = expectLater(
+        explicitQueue.flushQueuePersistence(),
+        throwsA(isA<StateError>()),
+      );
+      release.complete();
+      await firstFailure;
+      expect(rows.keys, recoveredIds);
+      expect([writes, deletes], [5, 0]);
+
+      started = Completer();
+      release = Completer();
+      final flushRetryReads = queueReads;
+      final secondFailure = expectLater(
+        explicitQueue.flushQueuePersistence(),
+        throwsA(isA<StateError>()),
+      );
+      await started.future;
+      release.complete();
+      await secondFailure;
+      expect(queueReads, flushRetryReads + 1);
+      expect(rows.keys, recoveredIds);
+      expect([writes, deletes], [5, 0]);
+
+      started = Completer();
+      release = Completer();
+      failRead = false;
+      final resumeRetryReads = queueReads;
+      explicitQueue.resumeQueue();
+      await started.future;
+      final resumedFlush = explicitQueue.flushQueuePersistence();
+      bool? explicitResumePublication;
+      final resumedSubscription = explicitRetry.listen(downloadQueueProvider, (
+        _,
+        next,
+      ) {
+        if (explicitResumePublication == null &&
+            next.items.length == recoveredIds.length) {
+          explicitResumePublication = next.isPaused;
+          explicitQueue.pauseQueue();
+        }
+      });
+      release.complete();
+      await resumedFlush;
+      resumedSubscription.close();
+      expect(explicitResumePublication, false);
+      expect(queueReads, resumeRetryReads + 1);
+      expect(rows.keys, recoveredIds);
+      expect([writes, deletes], [5, 0]);
+      explicitRetry.dispose();
       await (await AppStateDatabase.instance.database).close();
     },
   );