diff --git a/integration_test/queue_persistence_snapshot_test.dart b/integration_test/queue_persistence_snapshot_test.dart new file mode 100644 index 00000000..7febfee1 --- /dev/null +++ b/integration_test/queue_persistence_snapshot_test.dart @@ -0,0 +1,246 @@ +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:integration_test/integration_test.dart'; +// ignore: depend_on_referenced_packages +import 'package:path_provider_platform_interface/path_provider_platform_interface.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/services/app_state_database.dart'; +import 'package:spotiflac_android/services/download_queue_persistence.dart'; +import 'package:spotiflac_android/utils/chunked_list.dart'; + +import '../test/support/queue_persistence_baseline.dart'; +import '../test/support/queue_removal_benchmark.dart'; + +class _Paths extends PathProviderPlatform { + _Paths(this.path); + final String path; + @override + Future getApplicationDocumentsPath() async => path; +} + +Future> _measure( + List initial, + String change, { + required bool legacy, +}) async { + var items = initial; + final rows = {}; + var writes = 0; + Future applyChanges({ + required List> upserts, + required List deletedIds, + }) async { + writes++; + for (final id in deletedIds) { + rows.remove(id); + } + for (final row in upserts) { + rows[row['id'] as String] = row['item_json'] as String; + } + } + + final persistence = legacy + ? null + : DownloadQueuePersistence( + currentItems: () => items, + onError: (error) => fail('$error'), + loadRows: () async => [], + applyChanges: applyChanges, + ); + final flush = legacy + ? LegacyQueueSnapshot( + currentItems: () => items, + applyChanges: applyChanges, + ).flush + : persistence!.flush; + try { + await flush(); + writes = 0; + final watch = Stopwatch(); + for (var tick = 0; tick < 20; tick++) { + if (change != 'unchanged') { + final index = change == 'completed' ? 100 + tick : 100 + tick % 3; + final item = items[index]; + items = ChunkedList.from(items).updated({ + index: switch (change) { + 'progress' => item.copyWith( + status: DownloadStatus.downloading, + progress: (tick + 1) / 100, + bytesReceived: tick * 100000, + ), + 'completed' => item.copyWith(status: DownloadStatus.completed), + _ => item.copyWith(playlistName: 'Playlist $tick'), + }, + }); + } + watch.start(); + await flush(); + watch.stop(); + } + final expected = { + for (final item in items) + if (item.status != DownloadStatus.completed && + item.status != DownloadStatus.failed) + item.id: encodeDownloadQueueItemForPersistence(item), + }; + expect(rows, expected); + expect(writes, change == 'unchanged' || change == 'progress' ? 0 : 20); + return { + 'microseconds': watch.elapsedMicroseconds, + 'flushes': 20, + 'storage_calls': writes, + 'durable_rows': rows.length, + }; + } finally { + persistence?.dispose(); + } +} + +void main() { + final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized(); + final cases = >[]; + tearDownAll(() { + binding.reportData ??= {}; + binding.reportData!['queue_persistence_snapshot'] = { + 'platform': Platform.operatingSystem, + 'completed_cases': cases.length, + 'note': + '20 flushes from a warm saved cache, five alternating legacy/current pairs after warming both. Original snapshot/diff pipeline and actual current persistence service use the same codec and in-memory applyChanges sink, with canonical durable rows checked. Setup/initial serialization and state mutations excluded; measures snapshot/diff/encoding CPU, not SQLite I/O or rendered frames.', + 'cases': cases, + }; + }); + for (final count in [1000, 10000]) { + for (final change in ['unchanged', 'progress', 'completed', 'metadata']) { + testWidgets('persist $change changes in a $count-item snapshot', ( + tester, + ) async { + final initial = await queueRemovalFixture(count); + await _measure(initial.items, change, legacy: true); + await _measure(initial.items, change, legacy: false); + final samples = >>{ + 'legacy': [], + 'current': [], + }; + for (var pair = 0; pair < 5; pair++) { + for (final legacy in pair.isEven ? [true, false] : [false, true]) { + samples[legacy ? 'legacy' : 'current']!.add( + await _measure(initial.items, change, legacy: legacy), + ); + } + } + cases.add({ + 'initial_count': count, + 'change': change, + 'samples': samples, + }); + }); + } + } + + testWidgets('private SQLite rolls back, retries and reloads saved snapshots', ( + tester, + ) async { + final root = await Directory.systemTemp.createTemp('queue-snapshot-'); + final originalPaths = PathProviderPlatform.instance; + PathProviderPlatform.instance = _Paths(root.path); + SharedPreferences.setMockInitialValues({ + 'app_state_migrated_queue_to_sqlite_v1': true, + }); + Database? db; + DownloadQueuePersistence? persistence; + try { + final database = AppStateDatabase.instance; + final connection = await database.database; + db = connection; + expect(connection.path, '${root.path}/app_state.db'); + var items = (await queueRemovalFixture(2100)).items; + final errors = []; + var rejectWrite = false; + var attempts = 0; + persistence = DownloadQueuePersistence( + currentItems: () => items, + onError: errors.add, + applyChanges: ({required upserts, required deletedIds}) async { + attempts++; + if (rejectWrite) { + await connection.transaction((txn) async { + for (final id in deletedIds) { + await txn.delete( + 'download_queue_items', + where: 'id = ?', + whereArgs: [id], + ); + } + await txn.insert( + 'download_queue_items', + upserts.first, + conflictAlgorithm: ConflictAlgorithm.replace, + ); + throw StateError('injected transaction failure'); + }); + } + await database.applyPendingDownloadQueueChanges( + upserts: upserts, + deletedIds: deletedIds, + ); + }, + ); + await persistence.flush(); + final before = await database.getPendingDownloadQueueRows(); + expect(before, hasLength(2000)); + final changed = List.of(items); + changed[100] = changed[100].copyWith(status: DownloadStatus.completed); + changed[102] = changed[102].copyWith(playlistName: 'Updated'); + changed.add(changed[103].copyWith(id: 'new-item')); + changed.removeAt(101); + items = ChunkedList.from(changed); + rejectWrite = true; + await persistence.flush(); + expect(errors, hasLength(1)); + expect(await database.getPendingDownloadQueueRows(), before); + rejectWrite = false; + await persistence.flush(); + final expected = { + for (final item in items) + if (item.status != DownloadStatus.completed) + item.id: encodeDownloadQueueItemForPersistence(item), + }; + final after = await database.getPendingDownloadQueueRows(); + expect({for (final row in after) row['id']: row['item_json']}, expected); + expect(after, hasLength(1999)); + await persistence.flush(); + expect(attempts, 3); + + // A storage reload must invalidate the previously committed list identity. + await connection.delete( + 'download_queue_items', + where: 'id = ?', + whereArgs: ['new-item'], + ); + expect(await persistence.restore(), hasLength(1998)); + await persistence.flush(); + expect(attempts, 4); + expect(await database.getPendingDownloadQueueRows(), hasLength(1999)); + final restored = await persistence.restore(); + expect({ + for (final item in restored) + item.id: encodeDownloadQueueItemForPersistence(item), + }, expected); + cases.add({ + 'change': 'private_sqlite_rollback_retry_restore', + 'initial_rows': before.length, + 'final_rows': restored.length, + 'write_attempts': attempts, + 'expected_errors': errors.length, + }); + } finally { + persistence?.dispose(); + await db?.close(); + PathProviderPlatform.instance = originalPaths; + await root.delete(recursive: true); + } + }); +} diff --git a/lib/services/download_queue_persistence.dart b/lib/services/download_queue_persistence.dart index 0cedcd59..7216bdcd 100644 --- a/lib/services/download_queue_persistence.dart +++ b/lib/services/download_queue_persistence.dart @@ -4,6 +4,7 @@ import 'package:shared_preferences/shared_preferences.dart'; import 'package:spotiflac_android/models/download_item.dart'; import 'package:spotiflac_android/services/app_state_database.dart'; import 'package:spotiflac_android/services/download_queue_codec.dart'; +import 'package:spotiflac_android/utils/chunked_list.dart'; export 'package:spotiflac_android/services/download_queue_codec.dart' show downloadQueuePersistenceStatus, encodeDownloadQueueItemForPersistence; @@ -32,6 +33,7 @@ class DownloadQueuePersistence { Future _write = Future.value(); Future _pauseWrite = Future.value(); Map _saved = {}; + List? _savedSnapshot; static Future>> _loadRows() async { final database = AppStateDatabase.instance; @@ -65,6 +67,7 @@ class DownloadQueuePersistence { Future> restore() { final restored = _write.then((_) async { + _savedSnapshot = null; final decoded = await decodeDownloadQueueRows(await loadRows()); _saved = decoded.saved; return decoded.items; @@ -100,7 +103,11 @@ class DownloadQueuePersistence { Future _enqueue(List items) { return _write = _write .then((_) async { - final nextSaved = {}; + // Queue states use immutable storage. Repeated lifecycle/debounce + // flushes of an already committed snapshot need no diff or encoding. + if (identical(items, _savedSnapshot)) return; + final seenIds = {}; + final savedUpdates = {}; final upserts = >[]; final now = DateTime.now().toIso8601String(); final changed = []; @@ -112,12 +119,14 @@ class DownloadQueuePersistence { final saved = _saved[item.id]; final cached = identical(saved?.item, item) && saved?.payload != null; - nextSaved[item.id] = ( - item: item, - payload: cached ? saved!.payload : null, - ); + seenIds.add(item.id); if (!cached) { + savedUpdates[item.id] = (item: item, payload: null); changed.add(item); + } else if (savedUpdates.containsKey(item.id)) { + // Preserve the last non-terminal occurrence for malformed input + // with repeated request IDs, just as the full snapshot did. + savedUpdates[item.id] = saved!; } } final encoded = await encodeDownloadQueueItems(changed); @@ -125,8 +134,8 @@ class DownloadQueuePersistence { final item = changed[index]; final saved = _saved[item.id]; final payload = encoded[index]; - if (identical(nextSaved[item.id]?.item, item)) { - nextSaved[item.id] = (item: item, payload: payload); + if (identical(savedUpdates[item.id]?.item, item)) { + savedUpdates[item.id] = (item: item, payload: payload); } if (saved?.payload == payload) continue; upserts.add({ @@ -138,12 +147,20 @@ class DownloadQueuePersistence { }); } final deletedIds = _saved.keys - .where((id) => !nextSaved.containsKey(id)) + .where((id) => !seenIds.contains(id)) .toList(growable: false); if (upserts.isNotEmpty || deletedIds.isNotEmpty) { await applyChanges(upserts: upserts, deletedIds: deletedIds); } - _saved = nextSaved; + // Keep the last durable cache intact until the complete write has + // succeeded, so failed changes are still retried by the next flush. + for (final id in deletedIds) { + _saved.remove(id); + } + _saved.addAll(savedUpdates); + // Callers outside the notifier may supply a mutable List. Its + // identity alone never proves that its contents stayed unchanged. + _savedSnapshot = items is ChunkedList ? items : null; }) .catchError(onError); } diff --git a/test/download_queue_persistence_test.dart b/test/download_queue_persistence_test.dart index 6ed54cea..a6a441d6 100644 --- a/test/download_queue_persistence_test.dart +++ b/test/download_queue_persistence_test.dart @@ -10,6 +10,7 @@ import 'package:spotiflac_android/models/download_item.dart'; import 'package:spotiflac_android/models/track.dart'; import 'package:spotiflac_android/services/app_state_database.dart'; import 'package:spotiflac_android/services/download_queue_persistence.dart'; +import 'package:spotiflac_android/utils/chunked_list.dart'; DownloadItem _item( String id, { @@ -203,6 +204,111 @@ void main() { }, ); + test( + 'immutable snapshots are retried after failed combined writes', + () async { + final store = _Store(); + final track = _CountingTrack('retained'); + var items = ChunkedList.from([ + _item('retained').copyWith(track: track), + _item('removed'), + ]); + var rejectWrite = false; + final errors = []; + final persistence = DownloadQueuePersistence( + currentItems: () => items, + onError: errors.add, + loadRows: store.load, + applyChanges: ({required upserts, required deletedIds}) async { + if (rejectWrite) throw StateError('disk unavailable'); + await store.apply(upserts: upserts, deletedIds: deletedIds); + }, + ); + addTearDown(persistence.dispose); + await persistence.flush(); + await persistence.flush(); + expect(track.serializations, 1); + expect(store.writes, 1); + final original = items; + items = ChunkedList.from([ + items.first.copyWith(playlistName: 'Updated'), + _item('added'), + ]); + rejectWrite = true; + await persistence.flush(); + expect(errors, hasLength(1)); + expect(store.rows.keys, ['retained', 'removed']); + expect(store.writes, 1); + rejectWrite = false; + await persistence.flush(); + expect(store.rows.keys, ['retained', 'added']); + expect( + store.rows['retained']!['item_json'], + encodeDownloadQueueItemForPersistence(items.first), + ); + final serialized = track.serializations; + await persistence.flush(); + expect(track.serializations, serialized); + expect(store.writes, 2); + items = original; + await persistence.flush(); + expect(store.rows.keys, ['retained', 'removed']); + expect( + store.rows['retained']!['item_json'], + encodeDownloadQueueItemForPersistence(original.first), + ); + expect(store.writes, 3); + }, + ); + + test( + 'in-place mutable list edits are never treated as a saved snapshot', + () async { + final store = _Store(); + final items = [_item('first'), _item('removed')]; + final persistence = DownloadQueuePersistence( + currentItems: () => items, + onError: (error) => fail('$error'), + loadRows: store.load, + applyChanges: store.apply, + ); + addTearDown(persistence.dispose); + await persistence.flush(); + items[0] = items[0].copyWith(playlistName: 'Updated'); + items.removeLast(); + items.add(_item('added')); + await persistence.flush(); + expect(store.rows.keys, ['first', 'added']); + expect( + store.rows['first']!['item_json'], + encodeDownloadQueueItemForPersistence(items.first), + ); + items.clear(); + await persistence.flush(); + expect(store.rows, isEmpty); + expect(store.writes, 3); + }, + ); + + test('restore invalidates the previously saved immutable snapshot', () async { + final store = _Store(); + final items = ChunkedList.from([_item('retained')]); + final persistence = DownloadQueuePersistence( + currentItems: () => items, + onError: (error) => fail('$error'), + loadRows: store.load, + applyChanges: store.apply, + ); + addTearDown(persistence.dispose); + await persistence.flush(); + store.rows.clear(); + store.seed(_item('external')); + expect((await persistence.restore()).single.id, 'external'); + await persistence.flush(); + expect(store.rows.keys, ['retained']); + expect(store.writes, 2); + }); + test( 'failed writes retry and concurrent flushes retain the latest snapshot', () async { diff --git a/test/support/queue_persistence_baseline.dart b/test/support/queue_persistence_baseline.dart new file mode 100644 index 00000000..bde4df53 --- /dev/null +++ b/test/support/queue_persistence_baseline.dart @@ -0,0 +1,65 @@ +import 'package:spotiflac_android/models/download_item.dart'; +import 'package:spotiflac_android/services/download_queue_codec.dart'; + +/// Original snapshot/diff/encode pipeline for the device benchmark. Benchmark +/// callers flush serially and do not invoke restore, scheduling or pause writes. +class LegacyQueueSnapshot { + LegacyQueueSnapshot({required this.currentItems, required this.applyChanges}); + final List Function() currentItems; + final Future Function({ + required List> upserts, + required List deletedIds, + }) + applyChanges; + Map _saved = {}; + Future _write = Future.value(); + final Future _pauseWrite = Future.value(); + + Future flush() async { + final items = currentItems(); + await (_write = _write.then((_) async { + final nextSaved = {}; + final upserts = >[]; + final now = DateTime.now().toIso8601String(); + final changed = []; + for (final item in items) { + if (item.status == DownloadStatus.completed || + item.status == DownloadStatus.failed) { + continue; + } + final saved = _saved[item.id]; + final cached = identical(saved?.item, item) && saved?.payload != null; + nextSaved[item.id] = ( + item: item, + payload: cached ? saved!.payload : null, + ); + if (!cached) changed.add(item); + } + final encoded = await encodeDownloadQueueItems(changed); + for (var index = 0; index < changed.length; index++) { + final item = changed[index]; + final saved = _saved[item.id]; + final payload = encoded[index]; + if (identical(nextSaved[item.id]?.item, item)) { + nextSaved[item.id] = (item: item, payload: payload); + } + if (saved?.payload == payload) continue; + upserts.add({ + 'id': item.id, + 'item_json': payload, + 'status': downloadQueuePersistenceStatus(item.status).name, + 'created_at': item.createdAt.toIso8601String(), + 'updated_at': now, + }); + } + final deletedIds = _saved.keys + .where((id) => !nextSaved.containsKey(id)) + .toList(growable: false); + if (upserts.isNotEmpty || deletedIds.isNotEmpty) { + await applyChanges(upserts: upserts, deletedIds: deletedIds); + } + _saved = nextSaved; + })); + await _pauseWrite; + } +}