From 11d496effc5c2b41e96172b6bba0bc94243bc7b9 Mon Sep 17 00:00:00 2001 From: zarzet <42882290+zarzet@users.noreply.github.com> Date: Wed, 7 Oct 2026 04:28:07 +0700 Subject: [PATCH] fix(queue): merge startup additions before persistence and processing --- lib/main.dart | 7 +- lib/providers/download_queue_provider.dart | 130 +++++++++--- test/download_queue_startup_flush_test.dart | 216 +++++++++++++++++++- 3 files changed, 328 insertions(+), 25 deletions(-) diff --git a/lib/main.dart b/lib/main.dart index 107ce22b..44d09bd3 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -441,7 +441,12 @@ class _EagerInitializationState extends ConsumerState // 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. diff --git a/lib/providers/download_queue_provider.dart b/lib/providers/download_queue_provider.dart index 0b115768..fc76699c 100644 --- a/lib/providers/download_queue_provider.dart +++ b/lib/providers/download_queue_provider.dart @@ -349,7 +349,10 @@ class DownloadQueueNotifier extends Notifier { int _failedInSession = 0; int _queueItemSequence = 0; bool _isLoaded = false; - final Completer _queueRestored = Completer(); + Completer _queueRestored = Completer(); + bool _queueStartupPending = false; + bool _queueStorageReady = true; + bool? _startupPausedOverride; bool _foregroundResumeScheduled = false; bool _iosBackgroundExecutionExpired = false; StreamSubscription>? _iosBackgroundExpirationSubscription; @@ -423,23 +426,59 @@ class DownloadQueueNotifier extends Notifier { _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(); 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 { // 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 { } 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 { Future _loadQueueFromStorage() async { if (_isLoaded || !ref.mounted) return; - _isLoaded = true; try { final restorePaused = await _queuePersistence.loadUserPaused(); @@ -560,19 +608,27 @@ class DownloadQueueNotifier extends Notifier { 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 { 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 { } } - 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 { bool persistAcrossRestarts = true, Set 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 { } 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 { } Future _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 { diff --git a/test/download_queue_startup_flush_test.dart b/test/download_queue_startup_flush_test.dart index 08d90230..a01b9092 100644 --- a/test/download_queue_startup_flush_test.dart +++ b/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.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.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.delayed(const Duration(milliseconds: 400)); + earlyDisposed.dispose(); + await disposeFlush; + release.complete(); + await Future.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()), + ); + release.complete(); + await failedFlush; + await Future.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()), + ); + 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()), + ); + 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(); }, );