From 0a29046982d767d65bf6dff973ce6c88457936dc Mon Sep 17 00:00:00 2001 From: zarzet <42882290+zarzet@users.noreply.github.com> Date: Tue, 6 Oct 2026 13:09:54 +0700 Subject: [PATCH] perf(downloads): bound scheduler waiters and prevent overlapping retries --- lib/providers/download_queue_provider.dart | 110 ++++--------- lib/services/download_scheduler.dart | 89 +++++++++++ test/download_scheduler_test.dart | 176 +++++++++++++++++++++ 3 files changed, 297 insertions(+), 78 deletions(-) create mode 100644 lib/services/download_scheduler.dart create mode 100644 test/download_scheduler_test.dart diff --git a/lib/providers/download_queue_provider.dart b/lib/providers/download_queue_provider.dart index dd27b1f4..9695371b 100644 --- a/lib/providers/download_queue_provider.dart +++ b/lib/providers/download_queue_provider.dart @@ -27,6 +27,7 @@ import 'package:spotiflac_android/services/download_metadata_resolver.dart'; import 'package:spotiflac_android/services/download_metadata_embedding.dart'; import 'package:spotiflac_android/services/download_saf_file_replacer.dart'; import 'package:spotiflac_android/services/download_connectivity_policy.dart'; +import 'package:spotiflac_android/services/download_scheduler.dart'; import 'package:spotiflac_android/services/platform_bridge.dart'; import 'package:spotiflac_android/services/download_request_payload.dart'; import 'package:spotiflac_android/services/download_motion_artwork_source.dart'; @@ -1827,86 +1828,39 @@ class DownloadQueueNotifier extends Notifier { } Future _runQueueLoop() async { - final activeDownloads = >{}; - _startMultiProgressPolling(); - - while (true) { - if (state.isPaused) { - if (activeDownloads.isEmpty) { - _log.d('Queue is paused and no active download remains'); - break; - } - await Future.any([ - Future.wait(activeDownloads.values), - Future.delayed(_queueSchedulingInterval), - ]); - continue; - } - - final maxConcurrent = ref - .read(settingsProvider) - .concurrentDownloads - .clamp(1, 3); - if (activeDownloads.length >= maxConcurrent) { - // Keep pause/settings changes responsive without rescanning the full - // queue while every worker slot is already occupied. - await Future.any([ - Future.any(activeDownloads.values), - Future.delayed(_queueSchedulingInterval), - ]); - continue; - } - - final availableSlots = maxConcurrent - activeDownloads.length; - final queuedItems = state.items - .where( - (item) => - item.status == DownloadStatus.queued && - !_pausePendingItemIds.contains(item.id), - ) - .take(availableSlots) - .toList(growable: false); - - if (queuedItems.isEmpty && activeDownloads.isEmpty) { - _log.d('No more items to process'); - break; - } - - for (final item in queuedItems) { - if (state.isPaused) break; - - updateItemStatus(item.id, DownloadStatus.downloading); - - final future = _downloadSingleItem(item).whenComplete(() { - activeDownloads.remove(item.id); - PlatformBridge.clearItemProgress(item.id).catchError((_) {}); - }); - - activeDownloads[item.id] = future; - _log.d('Started download: ${item.track.name}'); - } - - if (activeDownloads.isNotEmpty) { - await Future.any([ - Future.any(activeDownloads.values), - Future.delayed(_queueSchedulingInterval), - ]); - } else { - await Future.delayed(_queueSchedulingInterval); - } + try { + await DownloadScheduler( + queuedItems: () => state.items.where( + (item) => + item.status == DownloadStatus.queued && + !_pausePendingItemIds.contains(item.id), + ), + idOf: (item) => item.id, + isPaused: () => state.isPaused, + concurrency: () => ref.read(settingsProvider).concurrentDownloads, + checkInterval: _queueSchedulingInterval, + start: (item) { + updateItemStatus(item.id, DownloadStatus.downloading); + _log.d('Started download: ${item.track.name}'); + return _downloadSingleItem(item); + }, + onFinished: (id) => + PlatformBridge.clearItemProgress(id).catchError((_) {}), + ).run(); + _log.d( + state.isPaused + ? 'Queue is paused and no active download remains' + : 'No more items to process', + ); + } finally { + _stopProgressPolling(); + final remainingIds = state.items.map((item) => item.id).toSet(); + _locallyCancelledItemIds.removeWhere((id) => !remainingIds.contains(id)); + _pausePendingItemIds.removeWhere((id) => !remainingIds.contains(id)); + _verificationRetryGuard.retainItems(remainingIds); + _rateLimitRetriedItemIds.removeWhere((id) => !remainingIds.contains(id)); } - - if (activeDownloads.isNotEmpty) { - await Future.wait(activeDownloads.values); - } - - _stopProgressPolling(); - final remainingIds = state.items.map((item) => item.id).toSet(); - _locallyCancelledItemIds.removeWhere((id) => !remainingIds.contains(id)); - _pausePendingItemIds.removeWhere((id) => !remainingIds.contains(id)); - _verificationRetryGuard.retainItems(remainingIds); - _rateLimitRetriedItemIds.removeWhere((id) => !remainingIds.contains(id)); } } diff --git a/lib/services/download_scheduler.dart b/lib/services/download_scheduler.dart new file mode 100644 index 00000000..9d99586f --- /dev/null +++ b/lib/services/download_scheduler.dart @@ -0,0 +1,89 @@ +import 'dart:async'; + +/// Runs live queue candidates with bounded concurrency. A single timed waiter +/// observes completions instead of attaching listeners to active work each tick. +class DownloadScheduler { + const DownloadScheduler({ + required this.queuedItems, + required this.idOf, + required this.isPaused, + required this.concurrency, + required this.start, + this.onFinished, + this.checkInterval = const Duration(milliseconds: 250), + }); + + final Iterable Function() queuedItems; + final String Function(T) idOf; + final bool Function() isPaused; + final int Function() concurrency; + final Future Function(T) start; + final void Function(String)? onFinished; + final Duration checkInterval; + + Future run() async { + final active = {}; + (Object, StackTrace)? failure; + Completer? waiter; + Timer? timer; + void wake() { + timer?.cancel(); + timer = null; + final pending = waiter; + waiter = null; + pending?.complete(); + } + + Future waitForChange() { + final pending = waiter = Completer(); + timer = Timer(checkInterval, wake); + return pending.future; + } + + void failed(Object error, StackTrace stack) => failure ??= (error, stack); + void launch(T item, String id) { + active.add(id); + unawaited( + Future.sync( + () => start(item), + ).then((_) {}, onError: failed).whenComplete(() { + active.remove(id); + try { + onFinished?.call(id); + } catch (error, stack) { + failed(error, stack); + } + wake(); + }), + ); + } + + try { + while (true) { + if (failure == null && !isPaused()) { + final limit = concurrency().clamp(1, 3); + if (active.length < limit) { + for (final item in queuedItems()) { + if (isPaused() || active.length >= limit) break; + final id = idOf(item); + // Retry can requeue an item before its previous attempt unwinds. + if (!active.contains(id)) launch(item, id); + } + } + } + if (active.isEmpty) break; + await waitForChange(); + } + } catch (error, stack) { + failed(error, stack); + while (active.isNotEmpty) { + await waitForChange(); + } + } finally { + wake(); + } + if (failure case (final error, final stack)) { + Error.throwWithStackTrace(error, stack); + } + } +} diff --git a/test/download_scheduler_test.dart b/test/download_scheduler_test.dart new file mode 100644 index 00000000..400237a8 --- /dev/null +++ b/test/download_scheduler_test.dart @@ -0,0 +1,176 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:spotiflac_android/services/download_scheduler.dart'; + +class _Queue { + final queued = ['a', 'b', 'c', 'd']; + final started = []; + final finished = []; + final work = >{}; + int limit = 2; + int selections = 0; + bool paused = false; + + DownloadScheduler get scheduler => DownloadScheduler( + queuedItems: () { + selections++; + return List.of(queued); + }, + idOf: (id) => id, + isPaused: () => paused, + concurrency: () => limit, + start: (id) { + queued.remove(id); + started.add(id); + return (work[id] = Completer()).future; + }, + onFinished: finished.add, + ); + + void complete(String id) => work[id]!.complete(); +} + +void main() { + testWidgets('fills free slots without scanning when every slot is occupied', ( + tester, + ) async { + final queue = _Queue(); + final run = queue.scheduler.run(); + await tester.pump(); + expect(queue.started, ['a', 'b']); + await tester.pump(const Duration(seconds: 30)); + expect(queue.selections, 1); + queue.complete('b'); + await tester.pump(); + expect(queue.started, ['a', 'b', 'c']); + queue.complete('a'); + await tester.pump(); + expect(queue.started, ['a', 'b', 'c', 'd']); + queue.complete('c'); + queue.complete('d'); + await tester.pump(); + await run; + expect(queue.finished, ['b', 'a', 'c', 'd']); + }); + + testWidgets( + 'pause drains active downloads and leaves queued work for resume', + (tester) async { + final queue = _Queue(); + var done = false; + final run = queue.scheduler.run().then((_) => done = true); + await tester.pump(); + queue.paused = true; + await tester.pump(const Duration(milliseconds: 250)); + queue.complete('a'); + await tester.pump(); + expect(done, false); + expect(queue.started, ['a', 'b']); + queue.complete('b'); + await tester.pump(); + await run; + expect(queue.queued, ['c', 'd']); + queue.paused = false; + final resumed = queue.scheduler.run(); + await tester.pump(); + queue.complete('c'); + queue.complete('d'); + await tester.pump(); + await resumed; + }, + ); + + testWidgets('retry waits for its active attempt before starting again', ( + tester, + ) async { + final queue = _Queue(); + final run = queue.scheduler.run(); + await tester.pump(); + queue.queued.insert(0, 'a'); + queue.complete('b'); + await tester.pump(); + expect(queue.started, ['a', 'b', 'c']); + expect(queue.queued, ['a', 'd']); + queue.complete('a'); + await tester.pump(); + expect(queue.started, ['a', 'b', 'c', 'a']); + queue.complete('c'); + await tester.pump(); + queue.complete('a'); + queue.complete('d'); + await tester.pump(); + await run; + }); + + testWidgets('uses live ordering and changes concurrency within the bounds', ( + tester, + ) async { + final queue = _Queue()..limit = 1; + final run = queue.scheduler.run(); + await tester.pump(); + queue.queued + ..clear() + ..addAll(['d', 'e', 'c']); + queue.limit = 99; + await tester.pump(const Duration(milliseconds: 250)); + expect(queue.started, ['a', 'd', 'e']); + queue.limit = 0; + queue.complete('d'); + queue.complete('e'); + await tester.pump(); + expect(queue.started, ['a', 'd', 'e']); + queue.complete('a'); + await tester.pump(); + expect(queue.started, ['a', 'd', 'e', 'c']); + queue.complete('c'); + await tester.pump(); + await run; + }); + + testWidgets( + 'observes failure and drains other active work before returning', + (tester) async { + final queue = _Queue(); + var done = false; + final error = StateError('Download failed'); + final run = queue.scheduler.run().whenComplete(() => done = true); + final assertion = expectLater(run, throwsA(same(error))); + await tester.pump(); + queue.work['a']!.completeError(error); + await tester.pump(); + expect(done, false); + expect(queue.started, ['a', 'b']); + queue.complete('b'); + await tester.pump(); + await assertion; + expect(queue.finished, ['a', 'b']); + expect(queue.queued, ['c', 'd']); + }, + ); + + testWidgets( + 'synchronous task failures release their slot without a timer leak', + (tester) async { + final run = DownloadScheduler( + queuedItems: () => ['a'], + idOf: (id) => id, + isPaused: () => false, + concurrency: () => 1, + start: (_) => throw StateError('Startup failed'), + ).run(); + final assertion = expectLater(run, throwsStateError); + await tester.pump(); + await assertion; + }, + ); + + testWidgets('does not start work when the queue is already paused', ( + tester, + ) async { + final queue = _Queue()..paused = true; + await queue.scheduler.run(); + expect(queue.started, isEmpty); + expect(queue.selections, 0); + }); +}