perf(downloads): reuse committed queue snapshots

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<String?> getApplicationDocumentsPath() async => path;
+}
+
+Future<Map<String, int>> _measure(
+  List<DownloadItem> initial,
+  String change, {
+  required bool legacy,
+}) async {
+  var items = initial;
+  final rows = <String, String>{};
+  var writes = 0;
+  Future<void> applyChanges({
+    required List<Map<String, dynamic>> upserts,
+    required List<String> 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<DownloadItem>.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 = <String, String>{
+      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 = <Map<String, dynamic>>[];
+  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 = <String, List<Map<String, int>>>{
+          '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 = <Object>[];
+      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<DownloadItem>.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);
+    }
+  });
+}

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<void> _write = Future<void>.value();
   Future<void> _pauseWrite = Future<void>.value();
   Map<String, DownloadQueueSavedItem> _saved = {};
+  List<DownloadItem>? _savedSnapshot;
 
   static Future<List<Map<String, dynamic>>> _loadRows() async {
     final database = AppStateDatabase.instance;
@@ -65,6 +67,7 @@ class DownloadQueuePersistence {
 
   Future<List<DownloadItem>> 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<void> _enqueue(List<DownloadItem> items) {
     return _write = _write
         .then((_) async {
-          final nextSaved = <String, DownloadQueueSavedItem>{};
+          // 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 = <String>{};
+          final savedUpdates = <String, DownloadQueueSavedItem>{};
           final upserts = <Map<String, dynamic>>[];
           final now = DateTime.now().toIso8601String();
           final changed = <DownloadItem>[];
@@ -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<DownloadItem> ? items : null;
         })
         .catchError(onError);
   }

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 = <Object>[];
+      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 {

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<DownloadItem> Function() currentItems;
+  final Future<void> Function({
+    required List<Map<String, dynamic>> upserts,
+    required List<String> deletedIds,
+  })
+  applyChanges;
+  Map<String, DownloadQueueSavedItem> _saved = {};
+  Future<void> _write = Future<void>.value();
+  final Future<void> _pauseWrite = Future<void>.value();
+
+  Future<void> flush() async {
+    final items = currentItems();
+    await (_write = _write.then((_) async {
+      final nextSaved = <String, DownloadQueueSavedItem>{};
+      final upserts = <Map<String, dynamic>>[];
+      final now = DateTime.now().toIso8601String();
+      final changed = <DownloadItem>[];
+      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;
+  }
+}