perf(downloads): batch queue cancellation and progress cleanup

android/app/src/rust/kotlin/com/zarz/spotiflac/SelectedCoreBackend.kt

@@ -880,11 +880,14 @@ internal object RustCoreBackend : CoreBackend {
                 environment.downloadState().use { state -> state.allProgress() }
             }
             "cleanupConnections" -> current.environment().use { it.cleanupConnections(); null }
-            "resetDownloadCancels" -> {
+            "resetDownloadCancels", "cancelDownloads" -> {
                 val values = args["item_ids"] as? List<*> ?: error("Expected item_ids list")
                 val itemIds = values.map { it as? String ?: error("Expected string item ID") }
                 current.environment().use { environment ->
-                    environment.downloadState().use { it.resetDownloadCancels(itemIds) }
+                    environment.downloadState().use { state ->
+                        if (method == "cancelDownloads") state.cancelDownloads(itemIds)
+                        else state.resetDownloadCancels(itemIds)
+                    }
                 }
                 null
             }

integration_test/download_cancel_dispatch_test.dart

@@ -0,0 +1,201 @@
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:flutter/services.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:integration_test/integration_test.dart';
+import 'package:spotiflac_android/services/download_request_payload.dart';
+import 'package:spotiflac_android/services/platform_bridge.dart';
+
+import '../test/support/library_collections_benchmark.dart';
+
+void main() {
+  final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized();
+  final report = <String, dynamic>{
+    'platform': Platform.operatingSystem,
+    'note':
+        'Five alternating old/new pairs, queued IDs without active transfers. Measures synchronous dispatch, total native cancel/clear completion and caller heartbeat, not UI frames or download throughput. Reset setup and 50ms settle excluded. iOS debug is correctness only.',
+  };
+  late Directory root;
+  late Directory output;
+  var completed = 0;
+  setUpAll(() async {
+    root = await Directory.systemTemp.createTemp('cancel-dispatch-');
+    final sources = await Directory('${root.path}/extensions').create();
+    final data = await Directory('${root.path}/data').create();
+    output = await Directory('${root.path}/output').create();
+    await PlatformBridge.initExtensionSystem(
+      sources.path,
+      data.path,
+      masterKey: base64Encode(List<int>.filled(32, 1)),
+      lyricsProviders: const [],
+      lyricsFetchOptions: const {},
+      allowedDirectories: [output.path],
+    );
+    expect(
+      (await PlatformBridge.getBackendImplementations())['downloads'],
+      'rust',
+    );
+    final provider = await Directory('${sources.path}/example.audio').create();
+    await File('${provider.path}/manifest.json').writeAsString(
+      jsonEncode({
+        'name': 'example.audio',
+        'version': '1',
+        'description': 'Generic cancellation fixture',
+        'type': ['download_provider'],
+        'permissions': {'file': true},
+        'skipMetadataEnrichment': true,
+      }),
+    );
+    await File('${provider.path}/index.js').writeAsString('''
+registerExtension({
+  checkAvailability() { return { available: true, track_id: 'track' }; },
+  download(trackId, quality, outputPath) {
+    const name = outputPath.slice(outputPath.lastIndexOf('/') + 1);
+    file.writeBytes(${jsonEncode('${output.path}/')} + name + '.started', 'c3RhcnRlZA==');
+    if (name.startsWith('active.')) utils.sleep(30000);
+    return { success: false, error: 'fixture finished' };
+  }
+});
+''');
+    await PlatformBridge.loadExtensionsFromDir(sources.path);
+    await PlatformBridge.setExtensionEnabled('example.audio', true);
+  });
+  tearDownAll(() async {
+    await PlatformBridge.cleanupExtensions();
+    await root.delete(recursive: true);
+    report['completed_cases'] = completed;
+    binding.reportData ??= {};
+    binding.reportData!['cancel_dispatch'] = report;
+  });
+
+  Future<Map<String, dynamic>> download(String id, {String? title}) =>
+      PlatformBridge.downloadByStrategy(
+        payload: DownloadRequestPayload(
+          itemId: id,
+          service: 'example.audio',
+          providerTrackId: id,
+          trackName: title ?? id,
+          artistName: 'Artist',
+          albumName: 'Album',
+          outputDir: output.path,
+          filenameFormat: '{title}',
+          outputExt: '.flac',
+          embedMetadata: false,
+          embedLyrics: false,
+          useExtensions: true,
+          useFallback: false,
+        ),
+      );
+
+  testWidgets(
+    'batch stops active work and queued attempts, preserving other IDs',
+    (tester) async {
+      final pending = download('active');
+      final marker = File('${output.path}/active.flac.started');
+      final deadline = DateTime.now().add(const Duration(seconds: 10));
+      while (!await marker.exists() && DateTime.now().isBefore(deadline)) {
+        await Future<void>.delayed(const Duration(milliseconds: 10));
+      }
+      expect(await marker.exists(), true);
+      expect(
+        (await PlatformBridge.getAllDownloadProgress())['items'],
+        contains('active'),
+      );
+      await PlatformBridge.cancelDownloads([
+        'active',
+        for (var i = 0; i < 600; i++) 'queued-$i',
+        'active',
+        '',
+      ]);
+      final cancelled = await pending.timeout(const Duration(seconds: 5));
+      expect(cancelled['error'].toString().toLowerCase(), contains('cancel'));
+      expect(
+        (await PlatformBridge.getAllDownloadProgress())['items'],
+        isNot(contains('active')),
+      );
+      for (final index in [0, 254, 255, 256, 599]) {
+        expect(
+          (await download('queued-$index'))['error'].toString().toLowerCase(),
+          contains('cancel'),
+        );
+        expect(
+          await File('${output.path}/queued-$index.flac.started').exists(),
+          false,
+        );
+      }
+      expect(
+        (await download('unselected'))['error'],
+        contains('fixture finished'),
+      );
+      expect(
+        await File('${output.path}/unselected.flac.started').exists(),
+        true,
+      );
+      expect(await PlatformBridge.resetDownloadCancels(['active']), {'active'});
+      expect(
+        (await download('active', title: 'retry'))['error'],
+        contains('fixture finished'),
+      );
+      expect(await File('${output.path}/retry.flac.started').exists(), true);
+      completed++;
+    },
+  );
+
+  testWidgets('invalid native batch leaves IDs uncancelled', (tester) async {
+    const channel = MethodChannel('com.zarz.spotiflac/backend');
+    await expectLater(
+      channel.invokeMethod<void>('cancelDownloads', {
+        'item_ids': ['untouched', 1],
+      }),
+      throwsA(isA<PlatformException>()),
+    );
+    expect(
+      (await download('untouched'))['error'],
+      contains('fixture finished'),
+    );
+    completed++;
+  });
+
+  for (final count in [32, 1024, 2048]) {
+    testWidgets('cancel and clear $count queued IDs', (tester) async {
+      final ids = [for (var i = 0; i < count; i++) 'cancel-$count-$i'];
+      final baseline = <Map<String, int>>[];
+      final batched = <Map<String, int>>[];
+      for (var sample = 0; sample < 5; sample++) {
+        for (final batch in sample.isEven ? [false, true] : [true, false]) {
+          expect(await PlatformBridge.resetDownloadCancels(ids), ids.toSet());
+          await Future<void>.delayed(const Duration(milliseconds: 50));
+          var dispatchUs = 0;
+          final result = await measureCollectionOperation<void>(() {
+            final watch = Stopwatch()..start();
+            final Future<void> pending;
+            if (batch) {
+              pending = PlatformBridge.cancelDownloads(ids);
+            } else {
+              final calls = <Future<void>>[];
+              for (final id in ids) {
+                calls.add(PlatformBridge.cancelDownload(id));
+                calls.add(PlatformBridge.clearItemProgress(id));
+              }
+              pending = Future.wait(calls);
+            }
+            dispatchUs = watch.elapsedMicroseconds;
+            return pending;
+          });
+          (batch ? batched : baseline).add({
+            ...result.timing,
+            'dispatch_us': dispatchUs,
+          });
+        }
+      }
+      report['ids_$count'] = {
+        'baseline': baseline,
+        'batched': batched,
+        'old_calls': count * 2,
+        'new_calls': (count / 256).ceil(),
+      };
+      completed++;
+    });
+  }
+}

ios/Runner/CoreBackend.swift

@@ -824,11 +824,15 @@ final class RustCoreBackend: CoreBackend {
         case "clearItemProgress": try current.environment().downloadState().clearItemProgress(itemId: string("item_id")); return nil
         case "cancelDownload": try current.environment().downloadState().cancelDownload(itemId: string("item_id")); return nil
         case "resetDownloadCancel": try current.environment().downloadState().resetDownloadCancel(itemId: string("item_id")); return nil
-        case "resetDownloadCancels":
+        case "resetDownloadCancels", "cancelDownloads":
             guard let itemIds = args["item_ids"] as? [String] else {
                 throw failure("Expected item_ids string list")
             }
-            try current.environment().downloadState().resetDownloadCancels(itemIds: itemIds)
+            if method == "cancelDownloads" {
+                try current.environment().downloadState().cancelDownloads(itemIds: itemIds)
+            } else {
+                try current.environment().downloadState().resetDownloadCancels(itemIds: itemIds)
+            }
             return nil
         case "getExtensionPendingAuth": return try current.getExtensionPendingAuthJson(extensionId: string("extension_id"))
         case "completeExtensionSessionGrant":

lib/providers/download_queue_provider.dart

@@ -1235,8 +1235,9 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
       _pausePendingItemIds.addAll(activeIds);
       _locallyCancelledItemIds.addAll(activeIds);
       for (final id in activeIds) {
-        _requestNativeCancel(id);
+        _verificationWaitCoordinator.cancelItem(id);
       }
+      unawaited(PlatformBridge.cancelDownloads(activeIds));
     }
 
     state = state.copyWith(items: [], isPaused: false, currentDownload: null);

lib/services/platform_bridge.dart

@@ -874,6 +874,23 @@ class PlatformBridge {
     await _channel.invokeMethod('cancelDownload', {'item_id': itemId});
   }
 
+  /// Cancels the selected attempts and clears their native progress. Failures
+  /// are best effort, as for individual queue cancellation; later chunks still
+  /// run, without issuing thousands of concurrent channel requests.
+  static Future<void> cancelDownloads(Iterable<String> itemIds) async {
+    final ids = itemIds.toSet().toList(growable: false);
+    for (var start = 0; start < ids.length; start += 256) {
+      final end = start + 256 < ids.length ? start + 256 : ids.length;
+      try {
+        await _channel.invokeMethod('cancelDownloads', {
+          'item_ids': ids.sublist(start, end),
+        });
+      } catch (error) {
+        _log.w('Failed to cancel ${end - start} downloads: $error');
+      }
+    }
+  }
+
   /// Drops a stale pre-registered cancel flag for an item with no active
   /// download, so a user-initiated retry does not abort instantly.
   static Future<void> resetDownloadCancel(String itemId) async {

rust_backend/crates/core/src/cancellation.rs

@@ -134,16 +134,27 @@ impl CancellationRegistry {
     }
 
     pub fn cancel(&self, id: &str) -> Result<(), CancellationError> {
+        self.cancel_many(std::iter::once(id))
+    }
+
+    pub fn cancel_many<'a>(
+        &self,
+        ids: impl IntoIterator<Item = &'a str>,
+    ) -> Result<(), CancellationError> {
         let mut state = self.shared.state.lock().expect("cancellation state lock");
         state.check_open()?;
-        if !id.is_empty() {
+        let mut changed = false;
+        for id in ids.into_iter().filter(|id| !id.is_empty()) {
             state
                 .entries
                 .entry(id.to_owned())
                 .or_default()
                 .cancelled
                 .store(true, Ordering::Release);
-            drop(state);
+            changed = true;
+        }
+        drop(state);
+        if changed {
             self.shared.notify();
         }
         Ok(())
@@ -448,20 +459,34 @@ mod tests {
         assert_eq!(observer.calls.load(Ordering::Relaxed), 1);
         registry.cancel("").unwrap();
         assert_eq!(observer.calls.load(Ordering::Relaxed), 1);
+        registry
+            .cancel_many(["first", "idle", "idle", "", "音楽"])
+            .unwrap();
+        assert_eq!(observer.calls.load(Ordering::Relaxed), 2);
+        assert!(registry.acquire("idle").unwrap().is_cancelled().unwrap());
+        assert!(registry.is_cancelled("音楽").unwrap());
+        assert!(!second.is_cancelled().unwrap());
+        assert!(!anonymous.is_cancelled().unwrap());
+        // Dropping the temporary idle lease also notifies its release.
+        let before = observer.calls.load(Ordering::Relaxed);
         assert_eq!(registry.cancel_active().unwrap(), ["first", "second"]);
         assert!(second.is_cancelled().unwrap());
         assert!(!anonymous.is_cancelled().unwrap());
-        assert_eq!(observer.calls.load(Ordering::Relaxed), 2);
+        assert_eq!(observer.calls.load(Ordering::Relaxed), before + 1);
         second.release();
         second.release();
         assert_eq!(second.check_active(), Err(CancellationError::LeaseReleased));
-        assert_eq!(observer.calls.load(Ordering::Relaxed), 3);
+        assert_eq!(observer.calls.load(Ordering::Relaxed), before + 2);
         registry.shutdown();
         assert_eq!(
             anonymous.check_active(),
             Err(CancellationError::RegistryClosed)
         );
-        assert_eq!(observer.calls.load(Ordering::Relaxed), 4);
+        assert_eq!(observer.calls.load(Ordering::Relaxed), before + 3);
+        assert_eq!(
+            registry.cancel_many(["later"]),
+            Err(CancellationError::RegistryClosed)
+        );
     }
 
     #[test]

rust_backend/crates/core/src/downloads.rs

@@ -142,6 +142,16 @@ impl DownloadState {
         Ok(ids)
     }
 
+    /// The application's bulk cancel/clear operation. Cancellation includes
+    /// queued sentinels; progress cleanup is best effort even during shutdown.
+    pub fn cancel_many(&self, ids: &[String]) -> Result<(), CancellationError> {
+        let cancelled = self
+            .cancellation
+            .cancel_many(ids.iter().map(String::as_str));
+        let _ = self.progress.remove_many(ids.iter().map(String::as_str));
+        cancelled
+    }
+
     pub fn shutdown(&self) {
         self.cancellation.shutdown();
         self.progress.shutdown();

rust_backend/crates/core/src/progress/mod.rs

@@ -118,9 +118,14 @@ impl State {
     }
 
     fn next(&mut self, changed: &Condvar) -> i64 {
+        let seq = self.advance();
+        changed.notify_all();
+        seq
+    }
+
+    fn advance(&mut self) -> i64 {
         self.seq = self.seq.wrapping_add(1);
         self.dirty = true;
-        changed.notify_all();
         self.seq
     }
 
@@ -300,21 +305,36 @@ impl ProgressRegistry {
     }
 
     pub fn remove(&self, id: &str) -> Result<(), ProgressError> {
+        self.remove_many(std::iter::once(id))
+    }
+
+    pub fn remove_many<'a>(
+        &self,
+        ids: impl IntoIterator<Item = &'a str>,
+    ) -> Result<(), ProgressError> {
         let mut state = self.shared.state.lock().expect("progress state lock");
         state.check()?;
-        if state.items.remove(id).is_some() {
-            let seq = state.next(&self.shared.changed);
-            state.removed.insert(id.to_owned(), seq);
-            if state.removed.len() > 512 {
-                let mut revisions: Vec<_> = state.removed.values().copied().collect();
-                revisions.sort_unstable();
-                let cutoff = revisions[revisions.len() / 2];
-                state.removed.retain(|_, revision| *revision > cutoff);
-                state.reset = state.reset.max(cutoff);
+        let mut changed = false;
+        for id in ids {
+            if state.items.remove(id).is_some() {
+                let seq = state.advance();
+                state.removed.insert(id.to_owned(), seq);
+                if state.removed.len() > 512 {
+                    let mut revisions: Vec<_> = state.removed.values().copied().collect();
+                    revisions.sort_unstable();
+                    let cutoff = revisions[revisions.len() / 2];
+                    state.removed.retain(|_, revision| *revision > cutoff);
+                    state.reset = state.reset.max(cutoff);
+                }
+                changed = true;
             }
+            // Even removing a missing ID invalidates the snapshot cache in Go.
+            state.dirty = true;
+        }
+        drop(state);
+        if changed {
+            self.shared.changed.notify_all();
         }
-        // Even removing a missing ID invalidates the snapshot cache in Go.
-        state.dirty = true;
         Ok(())
     }
 
@@ -377,6 +397,79 @@ impl Drop for ProgressRegistry {
     }
 }
 
+#[cfg(test)]
+mod batch_tests {
+    use super::*;
+    use serde_json::{Value, json};
+
+    #[test]
+    fn batch_removal_preserves_revisions_tombstones_and_cached_survivors() {
+        let registry = ProgressRegistry::new();
+        for id in ["a", "b", "keep", ""] {
+            registry.start(id).unwrap();
+        }
+        registry.set_progress("keep", 0.5, 20, 40).unwrap();
+        let before: Value = serde_json::from_str(&registry.snapshot().unwrap()).unwrap();
+        let seq = registry.shared.state.lock().unwrap().seq;
+        registry
+            .remove_many(["a", "b", "a", "missing", ""])
+            .unwrap();
+        let after: Value = serde_json::from_str(&registry.snapshot().unwrap()).unwrap();
+        assert_eq!(after, json!({"items": {"keep": before["items"]["keep"]}}));
+        let delta: Value = serde_json::from_str(&registry.wait_delta(seq, 1).unwrap()).unwrap();
+        assert_eq!(delta, json!({"seq":seq + 3, "removed":["", "a", "b"]}));
+        registry.remove_many(["a", "missing"]).unwrap();
+        assert_eq!(registry.shared.state.lock().unwrap().seq, seq + 3);
+        registry.start("a").unwrap();
+        let delta: Value = serde_json::from_str(&registry.wait_delta(seq, 1).unwrap()).unwrap();
+        assert_eq!(delta["removed"], json!(["", "b"]));
+        assert_eq!(delta["items"]["a"]["status"], "preparing");
+    }
+
+    #[test]
+    fn pruning_resets_old_cursors_and_retains_newer_removals() {
+        let registry = ProgressRegistry::new();
+        let ids: Vec<_> = (0..514).map(|i| format!("item-{i:04}")).collect();
+        for id in &ids {
+            registry.start(id).unwrap();
+        }
+        registry
+            .remove_many(ids.iter().take(513).map(String::as_str))
+            .unwrap();
+        let state = registry.shared.state.lock().unwrap();
+        assert_eq!(state.seq, 1027);
+        assert_eq!(state.reset, 771);
+        assert_eq!(state.removed.len(), 256);
+        let older: Value = serde_json::from_str(&state.delta(770)).unwrap();
+        assert_eq!(older["reset"], true);
+        assert_eq!(
+            older["items"]
+                .as_object()
+                .unwrap()
+                .keys()
+                .collect::<Vec<_>>(),
+            [&ids[513]]
+        );
+        let newer: Value = serde_json::from_str(&state.delta(771)).unwrap();
+        assert_eq!(newer["removed"], json!(ids[257..513]));
+        assert!(newer.get("reset").is_none());
+    }
+
+    #[test]
+    fn batch_wakes_subscribers_and_closed_registry_rejects_empty_batch() {
+        let registry = Arc::new(ProgressRegistry::new());
+        registry.start("a").unwrap();
+        registry.start("b").unwrap();
+        let listener = registry.subscribe().unwrap();
+        let waiting = std::thread::spawn(move || listener.wait_delta(2, 1000).unwrap());
+        registry.remove_many(["a", "b"]).unwrap();
+        let delta: Value = serde_json::from_str(&waiting.join().unwrap()).unwrap();
+        assert_eq!(delta, json!({"seq":4,"removed":["a","b"]}));
+        registry.shutdown();
+        assert_eq!(registry.remove_many([]), Err(ProgressError::Closed));
+    }
+}
+
 fn mark_downloading(item: &mut ItemProgress) {
     item.is_downloading = true;
     item.status = "downloading".into();

rust_backend/crates/mobile/src/progress.rs

@@ -46,6 +46,10 @@ impl DownloadState {
         self.inner.cancel(&item_id).map_err(Into::into)
     }
 
+    pub fn cancel_downloads(&self, item_ids: Vec<String>) -> Result<(), CancellationError> {
+        self.inner.cancel_many(&item_ids).map_err(Into::into)
+    }
+
     pub fn cancel_active_downloads(&self) -> Result<Vec<String>, CancellationError> {
         self.inner.cancel_active().map_err(Into::into)
     }
@@ -130,6 +134,68 @@ impl Default for DownloadState {
 mod tests {
     use super::*;
 
+    #[test]
+    fn cancel_batch_keeps_other_attempts_and_clears_selected_progress() {
+        let state = DownloadState::new();
+        let active = state.acquire("active".into()).unwrap();
+        let keep = state.acquire("keep".into()).unwrap();
+        let anonymous = state.acquire("".into()).unwrap();
+        for id in ["active", "queued", "keep", ""] {
+            state.init_item_progress(id.into()).unwrap();
+        }
+        state
+            .cancel_downloads(vec![
+                "active".into(),
+                "queued".into(),
+                "queued".into(),
+                "".into(),
+            ])
+            .unwrap();
+        assert!(active.is_cancelled().unwrap());
+        assert!(!keep.is_cancelled().unwrap());
+        assert!(!anonymous.is_cancelled().unwrap());
+        assert!(
+            state
+                .acquire("queued".into())
+                .unwrap()
+                .is_cancelled()
+                .unwrap()
+        );
+        let progress: serde_json::Value =
+            serde_json::from_str(&state.all_progress().unwrap()).unwrap();
+        assert_eq!(progress["items"].as_object().unwrap().len(), 1);
+        assert!(progress["items"].get("keep").is_some());
+        state.reset_download_cancels(vec!["active".into()]).unwrap();
+        assert!(active.is_cancelled().unwrap());
+        active.release();
+        assert!(
+            !state
+                .acquire("active".into())
+                .unwrap()
+                .is_cancelled()
+                .unwrap()
+        );
+        state.shutdown();
+        assert!(matches!(
+            state.cancel_downloads(vec!["late".into()]),
+            Err(CancellationError::RegistryClosed)
+        ));
+    }
+
+    #[test]
+    fn cancel_batch_keeps_best_effort_cleanup_when_one_registry_is_closed() {
+        let state = DownloadState::new();
+        state.init_item_progress("a".into()).unwrap();
+        state.inner.cancellation.shutdown();
+        assert!(state.cancel_downloads(vec!["a".into()]).is_err());
+        assert_eq!(state.all_progress().unwrap(), "{\"items\":{}}");
+        let state = DownloadState::new();
+        let active = state.acquire("a".into()).unwrap();
+        state.inner.progress.shutdown();
+        state.cancel_downloads(vec!["a".into()]).unwrap();
+        assert!(active.is_cancelled().unwrap());
+    }
+
     #[test]
     fn reset_batch_uses_the_same_download_owner_and_keeps_active_cancelled() {
         let state = DownloadState::new();

test/download_retry_batch_test.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 
 import 'package:flutter/services.dart';
+import 'package:flutter_local_notifications/flutter_local_notifications.dart';
 import 'package:flutter_riverpod/flutter_riverpod.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:shared_preferences/shared_preferences.dart';
@@ -68,10 +69,17 @@ void main() {
   const backend = MethodChannel('com.zarz.spotiflac/backend');
   const sqlite = MethodChannel('com.tekartik.sqflite');
   const paths = MethodChannel('plugins.flutter.io/path_provider');
+  const notifications = MethodChannel(
+    'dexterous.com/flutter/local_notifications',
+  );
   final previousFactory = databaseFactoryOrNull;
-  setUpAll(() => databaseFactory = databaseFactorySqflitePlugin);
+  setUpAll(() {
+    databaseFactory = databaseFactorySqflitePlugin;
+    AndroidFlutterLocalNotificationsPlugin.registerWith();
+  });
   tearDownAll(() => databaseFactory = previousFactory);
   setUp(() {
+    messenger.setMockMethodCallHandler(notifications, (_) async => null);
     SharedPreferences.setMockInitialValues({
       'app_state_migrated_queue_to_sqlite_v1': true,
     });
@@ -94,7 +102,7 @@ void main() {
     });
   });
   tearDown(() {
-    for (final channel in [backend, sqlite, paths]) {
+    for (final channel in [backend, sqlite, paths, notifications]) {
       messenger.setMockMethodCallHandler(channel, null);
     }
   });
@@ -286,4 +294,114 @@ void main() {
       container.dispose();
     });
   }
+
+  test(
+    'bulk cancel snapshots IDs, bounds dispatch and continues after failure',
+    () async {
+      final ids = [for (var i = 0; i < 600; i++) 'cancel-$i', 'cancel-0'];
+      final expected = ids.take(600).toList();
+      final calls = <List<String>>[];
+      var concurrent = 0;
+      var peak = 0;
+      messenger.setMockMethodCallHandler(backend, (call) async {
+        expect(call.method, 'cancelDownloads');
+        final chunk = List<String>.from(
+          (call.arguments as Map)['item_ids'] as List,
+        );
+        calls.add(chunk);
+        concurrent++;
+        if (concurrent > peak) peak = concurrent;
+        await Future<void>.delayed(Duration.zero);
+        concurrent--;
+        if (chunk.first == 'cancel-256') {
+          throw PlatformException(code: 'fixture');
+        }
+        return null;
+      });
+      final cancelled = PlatformBridge.cancelDownloads(ids);
+      ids.clear();
+      await cancelled;
+      expect(calls.expand((chunk) => chunk), expected);
+      expect(calls.map((chunk) => chunk.length), [256, 256, 88]);
+      expect(peak, 1);
+      await PlatformBridge.cancelDownloads([]);
+      expect(calls, hasLength(3));
+    },
+  );
+
+  for (final dispose in [false, true]) {
+    test(
+      'clearAll publishes immediately and bounds native work (dispose: $dispose)',
+      () async {
+        final initial = [
+          for (var i = 0; i < 600; i++)
+            _item('$i', switch (i % 3) {
+              0 => DownloadStatus.queued,
+              1 => DownloadStatus.downloading,
+              _ => DownloadStatus.finalizing,
+            }, null).copyWith(networkDownloadFolder: ''),
+          for (final status in [
+            DownloadStatus.completed,
+            DownloadStatus.failed,
+            DownloadStatus.skipped,
+          ])
+            _item(
+              status.name,
+              status,
+              null,
+            ).copyWith(networkDownloadFolder: ''),
+        ];
+        final container = ProviderContainer(
+          overrides: [
+            settingsProvider.overrideWith(_Settings.new),
+            downloadQueueProvider.overrideWith(() => _Queue(initial)),
+          ],
+        );
+        final queue = container.read(downloadQueueProvider.notifier);
+        queue.pauseQueue(persistAcrossRestarts: false);
+        await queue.flushQueuePersistence();
+        final started = Completer<void>();
+        final release = Completer<void>();
+        final done = Completer<void>();
+        final calls = <List<String>>[];
+        messenger.setMockMethodCallHandler(backend, (call) async {
+          expect(call.method, 'cancelDownloads');
+          final ids = List<String>.from(
+            (call.arguments as Map)['item_ids'] as List,
+          );
+          calls.add(ids);
+          if (calls.length == 1) {
+            started.complete();
+            await release.future;
+            throw PlatformException(code: 'fixture');
+          }
+          if (ids.last == '599') done.complete();
+          return null;
+        });
+        queue.clearAll();
+        expect(container.read(downloadQueueProvider).items, isEmpty);
+        expect(container.read(downloadQueueProvider).isPaused, false);
+        expect(container.read(downloadQueueProvider).currentDownload, isNull);
+        await started.future;
+        if (dispose) container.dispose();
+        release.complete();
+        await done.future;
+        expect(
+          calls.expand((ids) => ids),
+          initial.take(600).map((item) => item.id),
+        );
+        expect(calls.map((ids) => ids.length), [256, 256, 88]);
+        if (!dispose) {
+          // Late progress for removed work and a repeated Clear all must not
+          // recreate rows or issue cancellation for completed/failed/skipped IDs.
+          queue.updateProgress('0', 0.9);
+          queue.clearAll();
+          expect(container.read(downloadQueueProvider).items, isEmpty);
+          expect(calls, hasLength(3));
+          await queue.flushQueuePersistence();
+          container.dispose();
+        }
+      },
+    );
+  }
 }