From 2ded1cff6853d34d877159123aefb9f2ffa0e223 Mon Sep 17 00:00:00 2001 From: zarzet <42882290+zarzet@users.noreply.github.com> Date: Thu, 8 Oct 2026 08:47:47 +0700 Subject: [PATCH] perf(downloads): batch queue cancellation and progress cleanup --- .../com/zarz/spotiflac/SelectedCoreBackend.kt | 7 +- .../download_cancel_dispatch_test.dart | 201 ++++++++++++++++++ ios/Runner/CoreBackend.swift | 8 +- lib/providers/download_queue_provider.dart | 3 +- lib/services/platform_bridge.dart | 17 ++ rust_backend/crates/core/src/cancellation.rs | 35 ++- rust_backend/crates/core/src/downloads.rs | 10 + rust_backend/crates/core/src/progress/mod.rs | 117 ++++++++-- rust_backend/crates/mobile/src/progress.rs | 66 ++++++ test/download_retry_batch_test.dart | 122 ++++++++++- 10 files changed, 562 insertions(+), 24 deletions(-) create mode 100644 integration_test/download_cancel_dispatch_test.dart diff --git a/android/app/src/rust/kotlin/com/zarz/spotiflac/SelectedCoreBackend.kt b/android/app/src/rust/kotlin/com/zarz/spotiflac/SelectedCoreBackend.kt index 96158f77..a474698f 100644 --- a/android/app/src/rust/kotlin/com/zarz/spotiflac/SelectedCoreBackend.kt +++ b/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 } diff --git a/integration_test/download_cancel_dispatch_test.dart b/integration_test/download_cancel_dispatch_test.dart new file mode 100644 index 00000000..d9c51964 --- /dev/null +++ b/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 = { + '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.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> 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.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('cancelDownloads', { + 'item_ids': ['untouched', 1], + }), + throwsA(isA()), + ); + 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 = >[]; + final batched = >[]; + 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.delayed(const Duration(milliseconds: 50)); + var dispatchUs = 0; + final result = await measureCollectionOperation(() { + final watch = Stopwatch()..start(); + final Future pending; + if (batch) { + pending = PlatformBridge.cancelDownloads(ids); + } else { + final calls = >[]; + 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++; + }); + } +} diff --git a/ios/Runner/CoreBackend.swift b/ios/Runner/CoreBackend.swift index 1c250dac..aef4e368 100644 --- a/ios/Runner/CoreBackend.swift +++ b/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": diff --git a/lib/providers/download_queue_provider.dart b/lib/providers/download_queue_provider.dart index 44faecfa..d6f4a5aa 100644 --- a/lib/providers/download_queue_provider.dart +++ b/lib/providers/download_queue_provider.dart @@ -1235,8 +1235,9 @@ class DownloadQueueNotifier extends Notifier { _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); diff --git a/lib/services/platform_bridge.dart b/lib/services/platform_bridge.dart index 5133a5a0..65eb8188 100644 --- a/lib/services/platform_bridge.dart +++ b/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 cancelDownloads(Iterable 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 resetDownloadCancel(String itemId) async { diff --git a/rust_backend/crates/core/src/cancellation.rs b/rust_backend/crates/core/src/cancellation.rs index 366b8f79..4636b807 100644 --- a/rust_backend/crates/core/src/cancellation.rs +++ b/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, + ) -> 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] diff --git a/rust_backend/crates/core/src/downloads.rs b/rust_backend/crates/core/src/downloads.rs index c13df218..88530cf6 100644 --- a/rust_backend/crates/core/src/downloads.rs +++ b/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(); diff --git a/rust_backend/crates/core/src/progress/mod.rs b/rust_backend/crates/core/src/progress/mod.rs index 3ff44a16..5b6e3f38 100644 --- a/rust_backend/crates/core/src/progress/mod.rs +++ b/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, + ) -> 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(®istry.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(®istry.snapshot().unwrap()).unwrap(); + assert_eq!(after, json!({"items": {"keep": before["items"]["keep"]}})); + let delta: Value = serde_json::from_str(®istry.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(®istry.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::>(), + [&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(); diff --git a/rust_backend/crates/mobile/src/progress.rs b/rust_backend/crates/mobile/src/progress.rs index c2fdec9f..c9c0d018 100644 --- a/rust_backend/crates/mobile/src/progress.rs +++ b/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) -> Result<(), CancellationError> { + self.inner.cancel_many(&item_ids).map_err(Into::into) + } + pub fn cancel_active_downloads(&self) -> Result, 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(); diff --git a/test/download_retry_batch_test.dart b/test/download_retry_batch_test.dart index 62fef39b..1d69fd54 100644 --- a/test/download_retry_batch_test.dart +++ b/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 = >[]; + var concurrent = 0; + var peak = 0; + messenger.setMockMethodCallHandler(backend, (call) async { + expect(call.method, 'cancelDownloads'); + final chunk = List.from( + (call.arguments as Map)['item_ids'] as List, + ); + calls.add(chunk); + concurrent++; + if (concurrent > peak) peak = concurrent; + await Future.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(); + final release = Completer(); + final done = Completer(); + final calls = >[]; + messenger.setMockMethodCallHandler(backend, (call) async { + expect(call.method, 'cancelDownloads'); + final ids = List.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(); + } + }, + ); + } }