perf(downloads): bound scheduler waiters and prevent overlapping retries

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<DownloadQueueState> {
   }
 
   Future<void> _runQueueLoop() async {
-    final activeDownloads = <String, Future<void>>{};
-
     _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<void>.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<void>.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<void>.delayed(_queueSchedulingInterval),
-        ]);
-      } else {
-        await Future<void>.delayed(_queueSchedulingInterval);
-      }
+    try {
+      await DownloadScheduler<DownloadItem>(
+        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));
   }
 }
 

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<T> {
+  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<T> Function() queuedItems;
+  final String Function(T) idOf;
+  final bool Function() isPaused;
+  final int Function() concurrency;
+  final Future<void> Function(T) start;
+  final void Function(String)? onFinished;
+  final Duration checkInterval;
+
+  Future<void> run() async {
+    final active = <String>{};
+    (Object, StackTrace)? failure;
+    Completer<void>? waiter;
+    Timer? timer;
+    void wake() {
+      timer?.cancel();
+      timer = null;
+      final pending = waiter;
+      waiter = null;
+      pending?.complete();
+    }
+
+    Future<void> waitForChange() {
+      final pending = waiter = Completer<void>();
+      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<void>((_) {}, 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);
+    }
+  }
+}

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 = <String>['a', 'b', 'c', 'd'];
+  final started = <String>[];
+  final finished = <String>[];
+  final work = <String, Completer<void>>{};
+  int limit = 2;
+  int selections = 0;
+  bool paused = false;
+
+  DownloadScheduler<String> 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<void>()).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<String>(
+        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);
+  });
+}