perf(downloads): batch cancellation resets for retry

This commit is contained in:
zarzet committed 2026-10-08 06:55:26 +07:00
1 parent e0bdcb5134
commit c67d2a2dd2
8 files changed
+643 -19

No files matched your search

@@ -880,6 +880,14 @@ internal object RustCoreBackend : CoreBackend {
environment.downloadState().use { state -> state.allProgress() }
}
"cleanupConnections" -> current.environment().use { it.cleanupConnections(); null }
"resetDownloadCancels" -> {
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) }
}
null
}
"clearItemProgress", "cancelDownload", "resetDownloadCancel" -> current.environment().use { environment ->
environment.downloadState().use { state ->
when (method) {
@@ -0,0 +1,186 @@
import 'dart:convert';
import 'dart:io';
import 'dart:typed_data';
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';
Uint8List _wave() {
final bytes = Uint8List(1644);
final data = ByteData.sublistView(bytes);
bytes.setRange(0, 4, ascii.encode('RIFF'));
data.setUint32(4, bytes.length - 8, Endian.little);
bytes.setRange(8, 16, ascii.encode('WAVEfmt '));
data.setUint32(16, 16, Endian.little);
data.setUint16(20, 1, Endian.little);
data.setUint16(22, 1, Endian.little);
data.setUint32(24, 8000, Endian.little);
data.setUint32(28, 16000, Endian.little);
data.setUint16(32, 2, Endian.little);
data.setUint16(34, 16, Endian.little);
bytes.setRange(36, 40, ascii.encode('data'));
data.setUint32(40, bytes.length - 44, Endian.little);
return bytes;
}
Future<void> _seed(List<String> ids) async {
// Keep setup bounded too; Java GC from thousands of simultaneous channel
// calls must not dominate whichever measured operation happens next.
for (var start = 0; start < ids.length; start += 64) {
await Future.wait(
ids.skip(start).take(64).map(PlatformBridge.cancelDownload),
);
}
}
void main() {
final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized();
final report = <String, dynamic>{
'platform': Platform.operatingSystem,
'note':
'Real native cancellation bridge, five alternating old/new pairs per size. Bounded cancellation seeding and a 50ms settle are excluded. Measures reset dispatch latency and caller heartbeat, not download speed or FPS. iOS debug is correctness only.',
};
var completed = 0;
late Directory root;
late Directory output;
setUpAll(() async {
root = await Directory.systemTemp.createTemp('retry-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],
);
final implementations = await PlatformBridge.getBackendImplementations();
expect(implementations['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) {
file.writeBytes(outputPath, ${jsonEncode(base64Encode(_wave()))});
return { success: true, file_path: outputPath, actual_extension: '.wav' };
}
});
''');
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!['retry_dispatch'] = report;
});
Future<Map<String, dynamic>> download(String id) =>
PlatformBridge.downloadByStrategy(
payload: DownloadRequestPayload(
itemId: id,
service: 'example.audio',
providerTrackId: id,
trackName: id,
artistName: 'Artist',
albumName: 'Album',
outputDir: output.path,
filenameFormat: '{title}',
outputExt: '.wav',
embedMetadata: false,
embedLyrics: false,
useExtensions: true,
useFallback: false,
),
);
testWidgets('reset selected idle flags across chunk boundaries', (
tester,
) async {
final ids = [for (var i = 0; i < 260; i++) 'selected-$i', '音楽'];
await _seed([...ids, 'unselected']);
expect(
await PlatformBridge.resetDownloadCancels([...ids, ids.first, 'missing']),
{...ids, 'missing'},
);
for (final id in [ids.first, ids[255], ids[256], ids[259], '音楽']) {
final response = await download(id);
expect(response['success'], true, reason: response.toString());
expect(
await File(response['file_path'] as String).readAsBytes(),
_wave(),
);
}
final cancelled = await download('unselected');
expect(cancelled['success'], false);
expect(cancelled['error'].toString().toLowerCase(), contains('cancel'));
expect(await File('${output.path}/unselected.wav').exists(), false);
completed++;
});
testWidgets('malformed native batch does not reset any flags', (
tester,
) async {
const channel = MethodChannel('com.zarz.spotiflac/backend');
await PlatformBridge.cancelDownload('invalid-batch');
await expectLater(
channel.invokeMethod<void>('resetDownloadCancels', {
'item_ids': ['invalid-batch', 1],
}),
throwsA(isA<PlatformException>()),
);
final cancelled = await download('invalid-batch');
expect(cancelled['success'], false);
expect(cancelled['error'].toString().toLowerCase(), contains('cancel'));
completed++;
});
for (final count in [32, 1024, 2048]) {
testWidgets('reset $count idle cancellation flags', (tester) async {
final ids = [for (var i = 0; i < count; i++) 'retry-probe-$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]) {
await _seed(ids);
await Future<void>.delayed(const Duration(milliseconds: 50));
final result = await measureCollectionOperation<Set<String>>(
() async {
if (batch) return PlatformBridge.resetDownloadCancels(ids);
await Future.wait(ids.map(PlatformBridge.resetDownloadCancel));
return ids.toSet();
},
);
expect(result.value, ids.toSet());
(batch ? batched : baseline).add(result.timing);
}
}
report['ids_$count'] = {
'baseline': baseline,
'batched': batched,
'baseline_calls': count,
'batched_calls': (count / 256).ceil(),
};
completed++;
});
}
}
+6
View File
@@ -824,6 +824,12 @@ 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":
guard let itemIds = args["item_ids"] as? [String] else {
throw failure("Expected item_ids string list")
}
try current.environment().downloadState().resetDownloadCancels(itemIds: itemIds)
return nil
case "getExtensionPendingAuth": return try current.getExtensionPendingAuthJson(extensionId: string("extension_id"))
case "completeExtensionSessionGrant":
try completeSessionGrant(current, id: string("extension_id"), grant: string("grant"))
+2 -13
View File
@@ -1406,19 +1406,8 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
}
_log.i('Retrying ${failedIds.length} failed download(s)');
final resetResults = await Future.wait(
failedIds.map((id) async {
try {
await PlatformBridge.resetDownloadCancel(id);
return id;
} catch (e) {
_log.w('Failed to reset cancel flag for $id: $e');
return null;
}
}),
);
final retryableIds = resetResults.whereType<String>().toSet();
if (retryableIds.isEmpty) return;
final retryableIds = await PlatformBridge.resetDownloadCancels(failedIds);
if (!ref.mounted || retryableIds.isEmpty) return;
_locallyCancelledItemIds.removeAll(retryableIds);
_pausePendingItemIds.removeAll(retryableIds);
+24
View File
@@ -880,6 +880,30 @@ class PlatformBridge {
await _channel.invokeMethod('resetDownloadCancel', {'item_id': itemId});
}
/// Returns only IDs whose native reset succeeded. Bound each request so a
/// large retry selection does not flood the channel or hold the native lock.
static Future<Set<String>> resetDownloadCancels(
Iterable<String> itemIds,
) async {
final ids = itemIds.toSet().toList(growable: false);
final reset = <String>{};
for (var start = 0; start < ids.length; start += 256) {
final end = start + 256 < ids.length ? start + 256 : ids.length;
final chunk = ids.sublist(start, end);
try {
await _channel.invokeMethod('resetDownloadCancels', {
'item_ids': chunk,
});
reset.addAll(chunk);
} catch (error) {
_log.w(
'Failed to reset cancel flags for ${chunk.length} downloads: $error',
);
}
}
return reset;
}
/// iOS only: run a verification/OAuth page inside ASWebAuthenticationSession.
/// The session intercepts the callback scheme in-process, so the flow
/// completes even where the app's URL scheme is not registered with the OS
+88 -6
View File
@@ -176,14 +176,25 @@ impl CancellationRegistry {
}
pub fn reset_if_idle(&self, id: &str) -> Result<(), CancellationError> {
self.reset_many_if_idle(std::iter::once(id))
}
/// Clear only the selected idle sentinels under one registry lock. Leases
/// from a still-unwinding attempt keep their cancellation state intact.
pub fn reset_many_if_idle<'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 state
.entries
.get(id)
.is_some_and(|entry| entry.references == 0)
{
state.entries.remove(id);
for id in ids {
if state
.entries
.get(id)
.is_some_and(|entry| entry.references == 0)
{
state.entries.remove(id);
}
}
Ok(())
}
@@ -335,6 +346,77 @@ mod tests {
use std::sync::atomic::AtomicUsize;
use std::task::Wake;
#[test]
fn batch_reset_preserves_active_leases_and_unselected_cancellation() {
let registry = CancellationRegistry::new(CancellationDomain::Download);
let active = registry.acquire("active").unwrap();
let second = registry.acquire("active").unwrap();
for id in ["active", "idle", "keep", "音楽"] {
registry.cancel(id).unwrap();
}
registry
.reset_many_if_idle(["active", "idle", "idle", "missing", "", "音楽"])
.unwrap();
assert!(active.is_cancelled().unwrap());
assert!(second.is_cancelled().unwrap());
assert!(registry.is_cancelled("keep").unwrap());
assert!(!registry.is_cancelled("idle").unwrap());
assert!(!registry.is_cancelled("音楽").unwrap());
active.release();
registry.reset_many_if_idle(["active"]).unwrap();
assert!(second.is_cancelled().unwrap());
second.release();
let retry = registry.acquire("active").unwrap();
assert!(!retry.is_cancelled().unwrap());
drop(active);
drop(second);
assert!(!retry.is_cancelled().unwrap());
registry.cancel("active").unwrap();
assert!(retry.is_cancelled().unwrap());
}
#[test]
fn batch_reset_matches_individual_resets() {
let baseline = CancellationRegistry::new(CancellationDomain::Download);
let batch = CancellationRegistry::new(CancellationDomain::Download);
let ids: Vec<_> = (0..300).map(|i| format!("item-{i}")).collect();
let mut held = Vec::new();
for (index, id) in ids.iter().enumerate() {
if index % 3 == 0 {
held.push((baseline.acquire(id).unwrap(), batch.acquire(id).unwrap()));
}
baseline.cancel(id).unwrap();
batch.cancel(id).unwrap();
}
let selected: Vec<_> = ids.iter().step_by(2).map(String::as_str).collect();
for id in &selected {
baseline.reset_if_idle(id).unwrap();
}
batch.reset_many_if_idle(selected).unwrap();
for id in ids {
assert_eq!(baseline.is_cancelled(&id), batch.is_cancelled(&id));
}
for (old, new) in held {
assert_eq!(old.is_cancelled(), new.is_cancelled());
}
}
#[test]
fn closed_registry_rejects_batch_including_empty_input() {
let registry = CancellationRegistry::new(CancellationDomain::Download);
registry.cancel("keep").unwrap();
registry.shutdown();
assert_eq!(
registry.reset_many_if_idle(["keep"]),
Err(CancellationError::RegistryClosed)
);
assert_eq!(
registry.reset_many_if_idle([]),
Err(CancellationError::RegistryClosed)
);
assert!(registry.shared.state.lock().unwrap().entries.is_empty());
}
#[test]
fn external_waker_observes_changes_without_holding_the_registry_lock() {
#[derive(Default)]
@@ -57,6 +57,13 @@ impl DownloadState {
.map_err(Into::into)
}
pub fn reset_download_cancels(&self, item_ids: Vec<String>) -> Result<(), CancellationError> {
self.inner
.cancellation
.reset_many_if_idle(item_ids.iter().map(String::as_str))
.map_err(Into::into)
}
pub fn is_cancelled(&self, item_id: String) -> Result<bool, CancellationError> {
self.inner
.cancellation
@@ -119,6 +126,39 @@ impl Default for DownloadState {
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn reset_batch_uses_the_same_download_owner_and_keeps_active_cancelled() {
let state = DownloadState::new();
let active = state.acquire("active".into()).unwrap();
for id in ["active", "idle", "keep"] {
state.cancel_download(id.into()).unwrap();
}
state
.reset_download_cancels(vec!["active".into(), "idle".into()])
.unwrap();
assert!(active.is_cancelled().unwrap());
assert!(!state.is_cancelled("idle".into()).unwrap());
assert!(state.is_cancelled("keep".into()).unwrap());
active.release();
assert!(
!state
.acquire("active".into())
.unwrap()
.is_cancelled()
.unwrap()
);
state.shutdown();
assert!(matches!(
state.reset_download_cancels(vec!["keep".into()]),
Err(CancellationError::RegistryClosed)
));
}
}
#[derive(uniffi::Object)]
pub struct DownloadProgressSubscription {
inner: ProgressSubscription,
+289
View File
@@ -0,0 +1,289 @@
import 'dart:async';
import 'package:flutter/services.dart';
import 'package:flutter_riverpod/flutter_riverpod.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:shared_preferences/shared_preferences.dart';
import 'package:sqflite/sqflite.dart';
import 'package:spotiflac_android/models/download_item.dart';
import 'package:spotiflac_android/models/settings.dart';
import 'package:spotiflac_android/models/track.dart';
import 'package:spotiflac_android/providers/download_queue_provider.dart';
import 'package:spotiflac_android/providers/settings_provider.dart';
import 'package:spotiflac_android/services/platform_bridge.dart';
class _Settings extends SettingsNotifier {
@override
AppSettings build() =>
const AppSettings(downloadDirectory: '/fixture/output');
}
class _Queue extends DownloadQueueNotifier {
_Queue(this._items);
final List<DownloadItem> _items;
@override
DownloadQueueState build() {
super.build();
// An existing processing run owns scheduling. These tests exercise the
// retry mutation without launching audio downloads from a test fixture.
return const DownloadQueueState().copyWith(
items: _items,
isProcessing: true,
isPaused: true,
);
}
}
DownloadItem _item(String id, DownloadStatus status, DownloadErrorType? type) =>
DownloadItem(
id: id,
track: Track(
id: id,
name: id,
artistName: 'Artist',
albumName: 'Album',
duration: 180,
),
service: 'example',
status: status,
createdAt: DateTime.utc(2026),
error: 'Old error',
errorType: type,
filePath: '/old/$id',
progress: .5,
speedMBps: 1,
bytesReceived: 100,
bytesTotal: 200,
qualityOverride: 'HIGH',
playlistName: 'Playlist',
playlistPosition: 7,
fromBatch: true,
networkDownloadFolder: 'smb://example.test/music',
);
void main() {
final messenger =
TestWidgetsFlutterBinding.ensureInitialized().defaultBinaryMessenger;
const backend = MethodChannel('com.zarz.spotiflac/backend');
const sqlite = MethodChannel('com.tekartik.sqflite');
const paths = MethodChannel('plugins.flutter.io/path_provider');
final previousFactory = databaseFactoryOrNull;
setUpAll(() => databaseFactory = databaseFactorySqflitePlugin);
tearDownAll(() => databaseFactory = previousFactory);
setUp(() {
SharedPreferences.setMockInitialValues({
'app_state_migrated_queue_to_sqlite_v1': true,
});
messenger.setMockMethodCallHandler(
paths,
(_) async => '/fixture/documents',
);
messenger.setMockMethodCallHandler(sqlite, (call) async {
if (call.method == 'openDatabase') return {'id': 1};
if (call.method != 'query') return null;
final sql = (call.arguments as Map)['sql'] as String;
return sql.toLowerCase().contains('pragma user_version')
? {
'columns': ['user_version'],
'rows': [
[4],
],
}
: {'columns': <String>[], 'rows': <List<Object?>>[]};
});
});
tearDown(() {
for (final channel in [backend, sqlite, paths]) {
messenger.setMockMethodCallHandler(channel, null);
}
});
test(
'bridge snapshots, deduplicates and bounds requests, retaining successful chunks',
() async {
final ids = [for (var i = 0; i < 600; i++) 'id-$i', 'id-0'];
final original = ids.toList();
final calls = <List<String>>[];
var concurrent = 0;
var peak = 0;
messenger.setMockMethodCallHandler(backend, (call) async {
expect(call.method, 'resetDownloadCancels');
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 == 'id-256') throw PlatformException(code: 'fixture');
return null;
});
final reset = PlatformBridge.resetDownloadCancels(ids);
ids.clear();
expect(await reset, {
...original.take(256),
...original.skip(512).take(88),
});
expect(calls.map((chunk) => chunk.length), [256, 256, 88]);
expect(calls.expand((chunk) => chunk), original.take(600));
expect(peak, 1);
expect(await PlatformBridge.resetDownloadCancels([]), isEmpty);
expect(calls, hasLength(3));
},
);
for (final networkOnly in [false, true]) {
for (final failFirst in [false, true]) {
test(
'retry keeps filter and failed chunks (network: $networkOnly, fail: $failFirst)',
() async {
final initial = [
for (var i = 0; i < 600; i++)
_item(
'$i',
i.isEven ? DownloadStatus.failed : DownloadStatus.skipped,
i % 3 == 0
? DownloadErrorType.notFound
: DownloadErrorType.network,
),
_item(
'active',
DownloadStatus.downloading,
DownloadErrorType.network,
),
_item('complete', DownloadStatus.completed, null),
];
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 requested = <String>[];
final succeeded = <String>{};
messenger.setMockMethodCallHandler(backend, (call) async {
expect(call.method, 'resetDownloadCancels');
final ids = List<String>.from(
(call.arguments as Map)['item_ids'] as List,
);
final first = requested.isEmpty;
requested.addAll(ids);
if (failFirst && first) throw PlatformException(code: 'fixture');
succeeded.addAll(ids);
return null;
});
try {
await queue.retryAllFailed(networkOnly: networkOnly);
expect(
requested,
initial
.take(600)
.where(
(item) =>
!networkOnly ||
item.errorType == DownloadErrorType.network,
)
.map((item) => item.id),
);
final state = container.read(downloadQueueProvider);
expect(state.isPaused, false);
expect(state.isProcessing, true);
expect(
state.items.map((item) => item.id),
initial.map((item) => item.id),
);
for (var i = 0; i < initial.length; i++) {
final before = initial[i];
final after = state.items[i];
if (!succeeded.contains(before.id)) {
expect(after, same(before));
continue;
}
expect(after.status, DownloadStatus.queued);
expect([
after.error,
after.errorType,
after.filePath,
], everyElement(isNull));
expect([
after.progress,
after.speedMBps,
after.bytesReceived,
after.bytesTotal,
], everyElement(0));
expect(after.track, same(before.track));
expect(after.createdAt, before.createdAt);
expect(after.qualityOverride, before.qualityOverride);
expect(after.playlistName, before.playlistName);
expect(after.playlistPosition, before.playlistPosition);
expect(after.fromBatch, before.fromBatch);
expect(after.networkDownloadFolder, before.networkDownloadFolder);
}
await queue.flushQueuePersistence();
} finally {
container.dispose();
}
},
);
}
}
for (final outcome in ['failed', 'changed', 'disposed']) {
test('pending reset respects $outcome outcome', () async {
final initial = [
_item('a', DownloadStatus.failed, DownloadErrorType.network),
_item('b', DownloadStatus.skipped, DownloadErrorType.network),
];
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 before = container.read(downloadQueueProvider);
final started = Completer<void>();
final release = Completer<void>();
messenger.setMockMethodCallHandler(backend, (call) async {
if (call.method != 'resetDownloadCancels') return null;
started.complete();
await release.future;
if (outcome == 'failed') throw PlatformException(code: 'fixture');
return null;
});
final pending = queue.retryAllFailed();
await started.future;
if (outcome == 'disposed') {
container.dispose();
} else if (outcome == 'changed') {
queue.updateItemStatus(
'a',
DownloadStatus.completed,
filePath: '/finished/a',
);
queue.dismissItem('b');
}
release.complete();
await pending;
if (outcome == 'disposed') return;
final state = container.read(downloadQueueProvider);
if (outcome == 'failed') {
expect(state, same(before));
expect(state.isPaused, true);
} else {
expect(state.items.single.id, 'a');
expect(state.items.single.status, DownloadStatus.completed);
expect(state.items.single.filePath, '/finished/a');
}
await queue.flushQueuePersistence();
container.dispose();
});
}
}