diff --git a/integration_test/download_queue_workers_test.dart b/integration_test/download_queue_workers_test.dart new file mode 100644 index 00000000..10840336 --- /dev/null +++ b/integration_test/download_queue_workers_test.dart @@ -0,0 +1,597 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:isolate'; + +import 'package:flutter/foundation.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:integration_test/integration_test.dart'; +import 'package:sqflite/sqflite.dart'; +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_codec.dart'; +import 'package:spotiflac_android/services/download_queue_persistence.dart'; + +// Baseline is the synchronous restore loop before the worker change. Keep its +// parsing, canonical-payload comparison, and corrupt-row behavior comparable. +Future _inlineRestore( + List> rows, +) async { + final saved = {}; + final pending = []; + for (final row in rows) { + final id = row['id']?.toString() ?? ''; + saved[id] = (item: null, payload: null); + final payload = row['item_json']; + if (id.isEmpty || payload is! String || payload.isEmpty) continue; + try { + final decoded = jsonDecode(payload); + if (decoded is! Map) continue; + var item = DownloadItem.fromJson(Map.from(decoded)); + final status = downloadQueuePersistenceStatus(item.status); + saved[id] = ( + item: item, + payload: + row['status']?.toString() == status.name && + payload == encodeDownloadQueueItemForPersistence(item) + ? payload + : null, + ); + if (item.status != status) { + item = item.copyWith(status: status, progress: 0); + } + if (status == DownloadStatus.queued || status == DownloadStatus.skipped) { + pending.add(item); + } + } catch (_) { + continue; + } + } + return RestoredDownloadQueue(items: pending, saved: saved); +} + +Future> _inlineEncode(List items) async => [ + for (final item in items) encodeDownloadQueueItemForPersistence(item), +]; + +Future _inlineWrite( + Database database, + List> upserts, +) => database.transaction((txn) async { + final batch = txn.batch(); + for (final row in upserts) { + batch.insert( + 'download_queue_items', + row, + conflictAlgorithm: ConflictAlgorithm.replace, + ); + } + await batch.commit(noResult: true); +}); + +List _fixtureItems(int count) => List.generate(count, (index) { + final status = index % 7 == 0 + ? DownloadStatus.skipped + : index % 3 == 0 + ? DownloadStatus.downloading + : DownloadStatus.queued; + return DownloadItem( + id: 'queue:$index', + track: Track( + id: 'example:$index', + name: 'Song $index — 日本語', + artistName: 'Artist ${index % 99}', + albumName: 'Album ${index ~/ 12}', + albumArtist: 'Album artist ${index % 99}', + duration: 180, + trackNumber: index % 12 + 1, + discNumber: 1, + totalDiscs: 1, + releaseDate: '2026-10-07', + genre: 'Rock', + label: 'Example label', + copyright: '© Example', + composer: 'Composer $index', + comment: List.filled(24, 'Metadata $index — 日本語').join(' '), + coverUrl: 'https://example.test/art/${index ~/ 12}.jpg', + availability: const ServiceAvailability(deezer: true), + ), + service: 'extension.example', + status: status, + progress: .7, + speedMBps: 5, + bytesReceived: 70, + bytesTotal: 100, + preparationStage: 'decrypt', + createdAt: DateTime.utc(2026).add(Duration(seconds: index)), + qualityOverride: 'HI_RES', + playlistName: 'Large playlist', + playlistPosition: index + 1, + fromBatch: true, + preserveQualityVariant: true, + networkDownloadFolder: 'Music/Albums', + ); +}); + +class _InlineProbeTrack extends Track { + _InlineProbeTrack(this.original, this.expectedIsolate) + : super( + id: original.id, + name: original.name, + artistName: original.artistName, + albumName: original.albumName, + duration: original.duration, + albumArtist: original.albumArtist, + comment: original.comment, + coverUrl: original.coverUrl, + genre: original.genre, + label: original.label, + copyright: original.copyright, + composer: original.composer, + releaseDate: original.releaseDate, + availability: original.availability, + ); + final Track original; + final String? expectedIsolate; + + @override + Map toJson() { + if (Isolate.current.debugName != expectedIsolate) { + throw StateError('Small queue serialization left its owning isolate'); + } + return original.toJson(); + } +} + +class _Measurement { + _Measurement(this.value, this.elapsedMs, this.maxGapMs, this.ticks); + final T value; + final double elapsedMs; + final double maxGapMs; + final int ticks; + + Map toJson() => { + 'elapsed_ms': elapsedMs, + 'max_ui_gap_ms': maxGapMs, + 'ui_ticks': ticks, + }; +} + +Future<_Measurement> _measure(Future Function() operation) async { + final heartbeat = Stopwatch()..start(); + var previous = 0; + var maximum = 0; + var ticks = 0; + final timer = Timer.periodic(const Duration(milliseconds: 2), (_) { + final current = heartbeat.elapsedMicroseconds; + final gap = current - previous; + if (gap > maximum) maximum = gap; + previous = current; + ticks++; + }); + await Future.delayed(const Duration(milliseconds: 8)); + previous = heartbeat.elapsedMicroseconds; + maximum = 0; + ticks = 0; + final stopwatch = Stopwatch()..start(); + try { + final value = await operation(); + stopwatch.stop(); + final operationTicks = ticks; + final trailingGap = heartbeat.elapsedMicroseconds - previous; + if (trailingGap > maximum) maximum = trailingGap; + return _Measurement( + value, + stopwatch.elapsedMicroseconds / 1000, + maximum / 1000, + operationTicks, + ); + } finally { + timer.cancel(); + } +} + +class _QueueStore { + final rows = >{}; + int writes = 0; + + Future>> load() async => rows.values.toList(); + + Future apply({ + 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; + } + } +} + +Map _queueReport( + IntegrationTestWidgetsFlutterBinding binding, +) { + binding.reportData ??= {}; + return binding.reportData!.putIfAbsent( + 'queue_workers', + () => {}, + ) + as Map; +} + +void main() { + final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized(); + + testWidgets( + 'large queue codecs preserve payloads and release the UI isolate', + (tester) async { + final items = _fixtureItems(12000); + final rows = >[ + for (final item in items) + { + 'id': item.id, + 'status': item.status.name, + 'item_json': jsonEncode(item.toJson()), + }, + {'id': 'broken', 'status': 'queued', 'item_json': '{broken'}, + ]; + final report = { + 'count': items.length, + 'mode': kDebugMode ? 'debug' : 'profile/release', + 'payload_utf16_units': rows.fold( + 0, + (total, row) => total + (row['item_json'] as String).length, + ), + 'restore': >[], + 'encode': >[], + 'small': >[], + }; + // Warm both codecs before alternating the measured order to reduce bias. + await _inlineRestore(rows.sublist(0, 64)); + await decodeDownloadQueueRows(rows.sublist(0, 64)); + await _inlineEncode(items.sublist(0, 64)); + await encodeDownloadQueueItems(items.sublist(0, 64)); + for (var sample = 0; sample < 5; sample++) { + late _Measurement baselineRestore; + late _Measurement workerRestore; + late _Measurement> baselineEncode; + late _Measurement> workerEncode; + if (sample.isEven) { + baselineRestore = await _measure(() => _inlineRestore(rows)); + workerRestore = await _measure(() => decodeDownloadQueueRows(rows)); + baselineEncode = await _measure(() => _inlineEncode(items)); + workerEncode = await _measure(() => encodeDownloadQueueItems(items)); + } else { + workerEncode = await _measure(() => encodeDownloadQueueItems(items)); + baselineEncode = await _measure(() => _inlineEncode(items)); + workerRestore = await _measure(() => decodeDownloadQueueRows(rows)); + baselineRestore = await _measure(() => _inlineRestore(rows)); + } + expect(workerEncode.value, baselineEncode.value); + expect(workerEncode.ticks, greaterThan(0)); + expect(workerRestore.ticks, greaterThan(0)); + expect( + workerRestore.value.items.map((item) => item.id), + baselineRestore.value.items.map((item) => item.id), + ); + expect( + workerRestore.value.saved.keys, + baselineRestore.value.saved.keys, + ); + for (var index = 0; index < items.length; index++) { + final workerItem = workerRestore.value.items[index]; + final baselineItem = baselineRestore.value.items[index]; + expect( + encodeDownloadQueueItemForPersistence(workerItem), + encodeDownloadQueueItemForPersistence(baselineItem), + ); + expect( + workerRestore.value.saved[workerItem.id]!.payload, + baselineRestore.value.saved[baselineItem.id]!.payload, + ); + if (workerItem.status == DownloadStatus.queued && + items[index].status == DownloadStatus.queued) { + expect( + identical( + workerRestore.value.saved[workerItem.id]!.item, + workerItem, + ), + isTrue, + ); + } + } + expect(workerRestore.value.saved['broken'], ( + item: null, + payload: null, + )); + (report['restore'] as List>).add({ + 'sample': sample, + 'baseline': baselineRestore.toJson(), + 'worker': workerRestore.toJson(), + }); + (report['encode'] as List>).add({ + 'sample': sample, + 'baseline': baselineEncode.toJson(), + 'worker': workerEncode.toJson(), + }); + } + for (final size in [1, 16, 63]) { + final small = items.sublist(0, size); + final guarded = small + .map( + (item) => item.copyWith( + track: _InlineProbeTrack(item.track, Isolate.current.debugName), + ), + ) + .toList(growable: false); + final baseline = await _measure(() => _inlineEncode(small)); + final optimized = await _measure( + () => encodeDownloadQueueItems(guarded), + ); + expect(optimized.value, baseline.value); + (report['small'] as List>).add({ + 'count': size, + 'baseline': baseline.toJson(), + 'optimized': optimized.toJson(), + }); + } + _queueReport(binding)['codecs'] = report; + // ignore: avoid_print + print('QUEUE_WORKER_REPORT ${jsonEncode(report)}'); + }, + ); + + testWidgets( + 'large queue persistence keeps sparse progress and write retries', + (tester) async { + final original = _fixtureItems(2100); + var items = original; + final store = _QueueStore(); + final errors = []; + var failFirstWrite = true; + final persistence = DownloadQueuePersistence( + currentItems: () => items, + onError: errors.add, + loadRows: store.load, + applyChanges: ({required upserts, required deletedIds}) async { + if (failFirstWrite) { + failFirstWrite = false; + throw StateError('Temporary storage failure'); + } + await store.apply(upserts: upserts, deletedIds: deletedIds); + }, + ); + await persistence.flush(); + expect(store.rows, isEmpty); + await persistence.flush(); + expect(store.rows.length, original.length); + expect(errors, hasLength(1)); + final sparse = >[]; + for (var sample = 0; sample < 5; sample++) { + final index = original.length - 1; + items = List.of(items) + ..[index] = items[index].copyWith( + status: DownloadStatus.downloading, + progress: sample / 5, + bytesReceived: sample * 100, + speedMBps: sample + .5, + ); + final measured = await _measure(persistence.flush); + sparse.add(measured.toJson()); + } + expect(store.writes, 1); + final restored = await persistence.restore(); + expect(restored.length, items.length); + for (var index = 0; index < original.length; index++) { + expect( + encodeDownloadQueueItemForPersistence(restored[index]), + encodeDownloadQueueItemForPersistence(original[index]), + ); + } + items = []; + await persistence.flush(); + expect(store.rows, isEmpty); + persistence.dispose(); + _queueReport(binding)['sparse'] = { + 'count': original.length, + 'samples': sparse, + }; + }, + ); + + testWidgets('production queue restore uses bounded real sqflite reads', ( + tester, + ) async { + final directory = await Directory.systemTemp.createTemp('queue-workers-'); + final database = await openDatabase( + '${directory.path}/queue.db', + version: 4, + onCreate: (db, _) async { + await db.execute(''' + CREATE TABLE download_queue_items ( + id TEXT PRIMARY KEY, + item_json TEXT NOT NULL, + status TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + '''); + await db.execute( + 'CREATE INDEX idx_download_queue_items_created ON download_queue_items(created_at ASC)', + ); + await db.execute( + 'CREATE INDEX idx_download_queue_items_status ON download_queue_items(status)', + ); + }, + ); + try { + var items = _fixtureItems(12000); + await database.transaction((txn) async { + final batch = txn.batch(); + for (final item in items) { + batch.insert('download_queue_items', { + 'id': item.id, + 'item_json': encodeDownloadQueueItemForPersistence(item), + 'status': downloadQueuePersistenceStatus(item.status).name, + 'created_at': item.createdAt.toIso8601String(), + 'updated_at': '2026-10-07', + }); + } + batch.insert('download_queue_items', { + 'id': 'broken', + 'item_json': '{broken', + 'status': 'queued', + 'created_at': '2027-01-01', + 'updated_at': '2026-10-07', + }); + await batch.commit(noResult: true); + }); + final persistence = DownloadQueuePersistence( + currentItems: () => items, + onError: (error) => fail('$error'), + loadRows: () => AppStateDatabase.instance.getPendingDownloadQueueRows( + databaseOverride: database, + ), + applyChanges: ({required upserts, required deletedIds}) => + AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: upserts, + deletedIds: deletedIds, + databaseOverride: database, + ), + ); + final samples = >[]; + for (var sample = 0; sample < 5; sample++) { + Future baseline() async => _inlineRestore( + await database.query( + 'download_queue_items', + where: 'status IN (?, ?, ?, ?)', + whereArgs: ['queued', 'downloading', 'finalizing', 'skipped'], + orderBy: 'created_at ASC, rowid ASC', + ), + ); + late _Measurement before; + late _Measurement> after; + if (sample.isEven) { + before = await _measure(baseline); + after = await _measure(persistence.restore); + } else { + after = await _measure(persistence.restore); + before = await _measure(baseline); + } + expect( + after.value.map((item) => item.id), + before.value.items.map((item) => item.id), + ); + expect(after.value.length, items.length); + expect(after.ticks, greaterThan(0)); + samples.add({ + 'sample': sample, + 'baseline': before.toJson(), + 'worker': after.toJson(), + }); + items = after.value; + } + final payloads = await encodeDownloadQueueItems(items); + final upserts = >[ + for (var index = 0; index < items.length; index++) + { + 'id': items[index].id, + 'item_json': payloads[index], + 'status': downloadQueuePersistenceStatus(items[index].status).name, + 'created_at': items[index].createdAt.toIso8601String(), + 'updated_at': '2026-10-07', + }, + ]; + final writeSamples = >[]; + for (var sample = 0; sample < 5; sample++) { + Future optimized() => + AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: upserts, + deletedIds: const [], + databaseOverride: database, + ); + late _Measurement before; + late _Measurement after; + if (sample.isEven) { + before = await _measure(() => _inlineWrite(database, upserts)); + after = await _measure(optimized); + } else { + after = await _measure(optimized); + before = await _measure(() => _inlineWrite(database, upserts)); + } + writeSamples.add({ + 'sample': sample, + 'baseline': before.toJson(), + 'bounded': after.toJson(), + }); + } + await persistence.flush(); + expect( + await database.query( + 'download_queue_items', + where: 'id = ?', + whereArgs: ['broken'], + ), + isEmpty, + ); + final persisted = await database.query( + 'download_queue_items', + orderBy: 'created_at ASC, rowid ASC', + ); + expect(persisted.length, items.length); + for (var index = 0; index < items.length; index++) { + expect( + persisted[index]['item_json'], + encodeDownloadQueueItemForPersistence(items[index]), + ); + } + await database.execute(''' + CREATE TRIGGER reject_queue BEFORE INSERT ON download_queue_items + WHEN NEW.id = 'reject' BEGIN SELECT RAISE(ABORT, 'fixture rejection'); END + '''); + final failing = >[ + for (var index = 0; index < 600; index++) + {...upserts[index], 'id': index == 599 ? 'reject' : 'new:$index'}, + ]; + await expectLater( + AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: failing, + deletedIds: items.take(600).map((item) => item.id).toList(), + databaseOverride: database, + ), + throwsA(isA()), + ); + expect( + (await database.rawQuery( + 'SELECT COUNT(*) AS n FROM download_queue_items', + )).single['n'], + items.length, + ); + expect( + (await database.rawQuery( + "SELECT COUNT(*) AS n FROM download_queue_items WHERE id LIKE 'new:%'", + )).single['n'], + 0, + ); + await database.execute('DROP TRIGGER reject_queue'); + persistence.dispose(); + final report = _queueReport(binding); + report['sqflite_restore'] = {'count': items.length, 'samples': samples}; + report['sqflite_write'] = { + 'count': items.length, + 'samples': writeSamples, + }; + // ignore: avoid_print + print('QUEUE_SQLITE_REPORT ${jsonEncode(report['sqflite_restore'])}'); + // ignore: avoid_print + print('QUEUE_SQLITE_WRITE_REPORT ${jsonEncode(report['sqflite_write'])}'); + } finally { + await database.close(); + await directory.delete(recursive: true); + } + }); +} diff --git a/lib/services/app_state_database.dart b/lib/services/app_state_database.dart index d8283d29..c86847ad 100644 --- a/lib/services/app_state_database.dart +++ b/lib/services/app_state_database.dart @@ -1,7 +1,9 @@ import 'dart:convert'; +import 'package:flutter/foundation.dart'; import 'package:shared_preferences/shared_preferences.dart'; import 'package:sqflite/sqflite.dart'; +import 'package:spotiflac_android/services/download_queue_codec.dart'; import 'package:spotiflac_android/services/sqlite_helpers.dart' as sqlite; import 'package:spotiflac_android/utils/logger.dart'; @@ -187,8 +189,11 @@ class AppStateDatabase { }; } - Future migrateQueueFromSharedPreferences() async { - final prefs = await _prefs; + Future migrateQueueFromSharedPreferences({ + @visibleForTesting SharedPreferences? preferencesOverride, + @visibleForTesting Database? databaseOverride, + }) async { + final prefs = preferencesOverride ?? await _prefs; if (prefs.getBool(_queueMigrationKey) == true) { return false; } @@ -200,44 +205,15 @@ class AppStateDatabase { } try { - final decoded = jsonDecode(raw); - if (decoded is! List) { + final nowIso = DateTime.now().toIso8601String(); + final rows = await prepareLegacyDownloadQueueRows(raw, nowIso); + if (rows == null) { await prefs.setBool(_queueMigrationKey, true); return false; } - final nowIso = DateTime.now().toIso8601String(); - final db = await database; - await db.transaction((txn) async { - final batch = txn.batch(); - for (final entry in decoded.whereType>()) { - final map = Map.from(entry); - final id = map['id'] as String?; - if (id == null || id.isEmpty) continue; - - final status = map['status'] as String? ?? 'queued'; - if (status != 'queued' && status != 'downloading') { - continue; - } - - if (status == 'downloading') { - map['status'] = 'queued'; - map['progress'] = 0.0; - map['speedMBps'] = 0.0; - map['bytesReceived'] = 0; - } - - final createdAt = map['createdAt'] as String? ?? nowIso; - batch.insert(_queueTable, { - 'id': id, - 'item_json': jsonEncode(map), - 'status': 'queued', - 'created_at': createdAt, - 'updated_at': nowIso, - }, conflictAlgorithm: ConflictAlgorithm.replace); - } - await batch.commit(noResult: true); - }); + final db = databaseOverride ?? await database; + await db.transaction((txn) => _writeQueueChanges(txn, rows, const [])); await prefs.setBool(_queueMigrationKey, true); _log.i('Migrated legacy queue data to SQLite'); @@ -312,14 +288,82 @@ class AppStateDatabase { } } - Future>> getPendingDownloadQueueRows() async { - final db = await database; - return db.query( + Future>> getPendingDownloadQueueRows({ + @visibleForTesting Database? databaseOverride, + }) async { + final db = databaseOverride ?? await database; + const pageSize = 256; + const statuses = ['queued', 'downloading', 'finalizing', 'skipped']; + // Ordinary queues still need one query. Keep the channel response bounded + // before deciding whether a transaction/keyset scan is necessary. + final first = await db.query( _queueTable, where: 'status IN (?, ?, ?, ?)', - whereArgs: ['queued', 'downloading', 'finalizing', 'skipped'], + whereArgs: statuses, orderBy: 'created_at ASC, rowid ASC', + limit: pageSize + 1, ); + if (first.length <= pageSize) return first; + // Reread from one coherent snapshot: concurrent inserts, replacements, or + // status changes cannot skip/duplicate entries between page boundaries. + return db.transaction((txn) async { + final index = await txn.rawQuery( + ''' + SELECT sqlite_version() AS version + FROM sqlite_master + WHERE type = 'index' AND name = ? AND tbl_name = ? + ''', + ['idx_${_queueTable}_created', _queueTable], + ); + final indexed = index.isNotEmpty; + final version = indexed + ? index.single['version'].toString().split('.') + : const []; + // Row-value comparisons arrived in SQLite 3.15; Android API 24 may still + // use 3.9. Its range fallback also retains the ordered created_at index. + final tupleSeek = + version.length >= 2 && + (int.tryParse(version[0]) ?? 0) >= 3 && + ((int.tryParse(version[0]) ?? 0) > 3 || + (int.tryParse(version[1]) ?? 0) >= 15); + final rows = >[]; + String? lastCreated; + int? lastRowId; + while (true) { + final cursor = lastCreated == null + ? '' + : tupleSeek + ? 'AND (created_at, rowid) > (?, ?)' + : '${indexed ? 'AND created_at >= ? ' : ''}AND (created_at > ? OR (created_at = ? AND rowid > ?))'; + final page = await txn.rawQuery( + ''' + SELECT *, rowid AS __queue_order + FROM $_queueTable ${indexed ? 'INDEXED BY idx_${_queueTable}_created' : ''} + WHERE status IN (?, ?, ?, ?) + $cursor + ORDER BY created_at ASC, rowid ASC + LIMIT ? + ''', + [ + ...statuses, + if (lastCreated != null) ...[ + lastCreated, + if (!tupleSeek) ...[if (indexed) lastCreated, lastCreated], + lastRowId, + ], + pageSize, + ], + ); + if (page.isEmpty) break; + lastCreated = page.last['created_at'] as String; + lastRowId = page.last['__queue_order'] as int; + for (final row in page) { + rows.add(Map.of(row)..remove('__queue_order')); + } + if (page.length < pageSize) break; + } + return rows; + }); } Future replacePendingDownloadQueueRows( @@ -412,25 +456,86 @@ class AppStateDatabase { Future applyPendingDownloadQueueChanges({ required List> upserts, required List deletedIds, + @visibleForTesting Database? databaseOverride, }) async { if (upserts.isEmpty && deletedIds.isEmpty) return; - final db = await database; - await db.transaction((txn) async { - final batch = txn.batch(); - for (final id in deletedIds) { - batch.delete(_queueTable, where: 'id = ?', whereArgs: [id]); - } - for (final row in upserts) { - batch.insert( - _queueTable, - row, - conflictAlgorithm: ConflictAlgorithm.replace, - ); - } - await batch.commit(noResult: true); - }); + final db = databaseOverride ?? await database; + await db.transaction((txn) => _writeQueueChanges(txn, upserts, deletedIds)); } + static Future _writeQueueChanges( + Transaction txn, + List> upserts, + List deletedIds, + ) async { + var batch = txn.batch(); + var pending = 0; + Future flushBatch() async { + await batch.commit(noResult: true); + batch = txn.batch(); + pending = 0; + } + + // Bound method-channel encoding for a large queue while retaining one + // transaction: a later failed chunk rolls back earlier chunks as well. + for (final id in deletedIds) { + batch.delete(_queueTable, where: 'id = ?', whereArgs: [id]); + if (++pending == 256) await flushBatch(); + } + for (var index = 0; index < upserts.length;) { + final row = upserts[index]; + if (upserts.length >= 64 && _canonicalQueueRow(row)) { + if (pending > 0) await flushBatch(); + final arguments = []; + final start = index; + // Five bindings per row: 192 rows stay below older SQLite's 999 limit. + while (index < upserts.length && + index - start < 192 && + _canonicalQueueRow(upserts[index])) { + final value = upserts[index++]; + arguments.addAll([ + value['id'], + value['item_json'], + value['status'], + value['created_at'], + value['updated_at'], + ]); + } + final count = index - start; + await txn.rawInsert( + count == 192 + ? _fullQueueInsertStatement + : _queueInsertStatement(count), + arguments, + ); + continue; + } + batch.insert( + _queueTable, + row, + conflictAlgorithm: ConflictAlgorithm.replace, + ); + index++; + if (++pending == 256) await flushBatch(); + } + if (pending > 0) await batch.commit(noResult: true); + } + + static bool _canonicalQueueRow(Map row) => + row.length == 5 && + row.containsKey('id') && + row.containsKey('item_json') && + row.containsKey('status') && + row.containsKey('created_at') && + row.containsKey('updated_at'); + + static final String _fullQueueInsertStatement = _queueInsertStatement(192); + + static String _queueInsertStatement(int rows) => + 'INSERT OR REPLACE INTO $_queueTable ' + '(id, item_json, status, created_at, updated_at) VALUES ' + '${List.filled(rows, '(?, ?, ?, ?, ?)').join(', ')}'; + Future>> getRecentAccessRows({int? limit}) async { final db = await database; return db.query( diff --git a/lib/services/download_queue_codec.dart b/lib/services/download_queue_codec.dart new file mode 100644 index 00000000..a89ac7a8 --- /dev/null +++ b/lib/services/download_queue_codec.dart @@ -0,0 +1,283 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:isolate'; + +import 'package:flutter/foundation.dart'; +import 'package:spotiflac_android/models/download_item.dart'; + +typedef DownloadQueueSavedItem = ({DownloadItem? item, String? payload}); + +const _workerItemThreshold = 64; +const _workerPayloadThreshold = 64 * 1024; +const _workerChunkSize = 1024; + +DownloadStatus downloadQueuePersistenceStatus(DownloadStatus status) => + switch (status) { + DownloadStatus.downloading || + DownloadStatus.finalizing => DownloadStatus.queued, + _ => status, + }; + +/// Transfer samples are transient; only restart-relevant state reaches SQLite. +String encodeDownloadQueueItemForPersistence(DownloadItem item) => jsonEncode( + item.toJson() + ..['status'] = downloadQueuePersistenceStatus(item.status).name + ..['progress'] = 0.0 + ..['speedMBps'] = 0.0 + ..['bytesReceived'] = 0 + ..['bytesTotal'] = 0 + ..['preparationStage'] = '', +); + +class RestoredDownloadQueue { + RestoredDownloadQueue({required this.items, required this.saved}); + + final List items; + final Map saved; +} + +RestoredDownloadQueue _decodeRows(List> rows) { + final saved = {}; + final pending = []; + for (final row in rows) { + final id = row['id']?.toString() ?? ''; + // Retain corrupt IDs so the next successful flush deletes those rows. + saved[id] = (item: null, payload: null); + final payload = row['item_json']; + if (id.isEmpty || payload is! String || payload.isEmpty) continue; + try { + final decoded = jsonDecode(payload); + if (decoded is! Map) continue; + var item = DownloadItem.fromJson(Map.from(decoded)); + final status = downloadQueuePersistenceStatus(item.status); + saved[id] = ( + item: item, + payload: + row['status']?.toString() == status.name && + payload == encodeDownloadQueueItemForPersistence(item) + ? payload + : null, + ); + if (item.status != status) { + item = item.copyWith(status: status, progress: 0); + } + if (status == DownloadStatus.queued || status == DownloadStatus.skipped) { + pending.add(item); + } + } catch (_) { + continue; + } + } + return RestoredDownloadQueue(items: pending, saved: saved); +} + +bool _largeRows(List> rows) { + if (rows.length >= _workerItemThreshold) return true; + var length = 0; + for (final row in rows) { + final payload = row['item_json']; + if (payload is String) length += payload.length; + if (length >= _workerPayloadThreshold) return true; + } + return false; +} + +/// Keep ordinary queues inline. Large JSON/model work runs off the UI isolate; +/// bounded requests avoid copying an unbounded queue when starting a worker. +Future decodeDownloadQueueRows( + List> rows, +) async { + if (!_largeRows(rows)) return _decodeRows(rows); + return _runQueueCodecWorker( + _QueueCodecMode.restore, + rows, + ); +} + +List _encodeItems(List items) => [ + for (final item in items) encodeDownloadQueueItemForPersistence(item), +]; + +bool _largeItems(List items) { + if (items.length >= _workerItemThreshold) return true; + var length = 0; + for (final item in items) { + final track = item.track; + // A single long comment/URL can be expensive too. Inspect existing strings + // without invoking toJson on the UI isolate or allocating payload maps. + length += + item.id.length + + item.service.length + + (item.filePath?.length ?? 0) + + (item.error?.length ?? 0) + + (item.qualityOverride?.length ?? 0) + + (item.playlistName?.length ?? 0) + + item.networkDownloadFolder.length + + track.id.length + + track.name.length + + track.artistName.length + + track.albumName.length + + (track.albumArtist?.length ?? 0) + + (track.artistId?.length ?? 0) + + (track.albumId?.length ?? 0) + + (track.coverUrl?.length ?? 0) + + (track.headerVideoUrl?.length ?? 0) + + (track.isrc?.length ?? 0) + + (track.previewUrl?.length ?? 0) + + (track.releaseDate?.length ?? 0) + + (track.deezerId?.length ?? 0) + + (track.source?.length ?? 0) + + (track.albumType?.length ?? 0) + + (track.composer?.length ?? 0) + + (track.genre?.length ?? 0) + + (track.label?.length ?? 0) + + (track.copyright?.length ?? 0) + + (track.comment?.length ?? 0) + + (track.itemType?.length ?? 0) + + (track.audioQuality?.length ?? 0) + + (track.audioModes?.length ?? 0) + + (track.upc?.length ?? 0) + + (track.availability?.tidalUrl?.length ?? 0) + + (track.availability?.qobuzUrl?.length ?? 0) + + (track.availability?.amazonUrl?.length ?? 0) + + (track.availability?.deezerUrl?.length ?? 0) + + (track.availability?.deezerId?.length ?? 0); + if (length >= _workerPayloadThreshold) return true; + } + return false; +} + +/// Only cache misses enter this codec. A sparse progress change in a large +/// queue keeps the inexpensive inline path and leaves untouched tracks alone. +Future> encodeDownloadQueueItems(List items) async { + if (!_largeItems(items)) return _encodeItems(items); + return _runQueueCodecWorker>(_QueueCodecMode.persist, items); +} + +enum _QueueCodecMode { restore, persist } + +typedef _QueueWorkerRequest = ({SendPort replies, _QueueCodecMode mode}); + +Future _runQueueCodecWorker( + _QueueCodecMode mode, + List values, +) async { + final replies = ReceivePort(); + final ready = Completer(); + final result = Completer(); + Completer? acknowledgement; + result.future.ignore(); + final subscription = replies.listen((Object? message) { + if (message is SendPort) { + ready.complete(message); + } else if (message == true) { + acknowledgement?.complete(); + } else if (message is List && message.length == 1) { + result.complete(message.single as T); + } else if (!result.isCompleted) { + final error = message is List && message.length == 2 + ? RemoteError(message[0].toString(), message[1].toString()) + : StateError('Download queue worker exited without a result'); + if (!ready.isCompleted) ready.completeError(error); + result.completeError(error); + if (acknowledgement case final pending? when !pending.isCompleted) { + pending.completeError(error); + } + } + }); + Isolate? worker; + try { + worker = await Isolate.spawn( + _queueCodecWorker, + (replies: replies.sendPort, mode: mode), + onError: replies.sendPort, + onExit: replies.sendPort, + debugName: mode == _QueueCodecMode.restore + ? 'download-queue-restore' + : 'download-queue-persist', + ); + final commands = await ready.future; + for (var offset = 0; offset < values.length; offset += _workerChunkSize) { + final end = (offset + _workerChunkSize).clamp(0, values.length); + final received = Completer(); + acknowledgement = received; + commands.send(values.sublist(offset, end)); + await received.future; + } + commands.send(null); + return await result.future; + } finally { + worker?.kill(priority: Isolate.immediate); + await subscription.cancel(); + replies.close(); + } +} + +void _queueCodecWorker(_QueueWorkerRequest request) async { + final commands = ReceivePort(); + final items = []; + final saved = {}; + final payloads = []; + request.replies.send(commands.sendPort); + await for (final Object? message in commands) { + if (message == null) { + commands.close(); + Isolate.exit(request.replies, [ + request.mode == _QueueCodecMode.restore + ? RestoredDownloadQueue(items: items, saved: saved) + : payloads, + ]); + } + if (request.mode == _QueueCodecMode.restore) { + final decoded = _decodeRows( + (message as List).cast>(), + ); + items.addAll(decoded.items); + saved.addAll(decoded.saved); + } else { + payloads.addAll( + _encodeItems((message as List).cast()), + ); + } + request.replies.send(true); + } +} + +List>? _legacyRows((String, String) request) { + final (raw, nowIso) = request; + final decoded = jsonDecode(raw); + if (decoded is! List) return null; + final rows = >[]; + for (final entry in decoded.whereType>()) { + final map = Map.from(entry); + final id = map['id'] as String?; + if (id == null || id.isEmpty) continue; + final status = map['status'] as String? ?? 'queued'; + if (status != 'queued' && status != 'downloading') continue; + if (status == 'downloading') { + map['status'] = 'queued'; + map['progress'] = 0.0; + map['speedMBps'] = 0.0; + map['bytesReceived'] = 0; + } + rows.add({ + 'id': id, + 'item_json': jsonEncode(map), + 'status': 'queued', + 'created_at': map['createdAt'] as String? ?? nowIso, + 'updated_at': nowIso, + }); + } + return rows; +} + +/// Legacy queue maps may contain fields unknown to today's model. Preserve +/// them rather than hydrating DownloadItem, while moving large JSON work off UI. +Future>?> prepareLegacyDownloadQueueRows( + String raw, + String nowIso, +) async { + final request = (raw, nowIso); + if (raw.length < _workerPayloadThreshold) return _legacyRows(request); + return compute(_legacyRows, request, debugLabel: 'download-queue-migration'); +} diff --git a/lib/services/download_queue_persistence.dart b/lib/services/download_queue_persistence.dart index bec723cd..0cedcd59 100644 --- a/lib/services/download_queue_persistence.dart +++ b/lib/services/download_queue_persistence.dart @@ -1,27 +1,12 @@ import 'dart:async'; -import 'dart:convert'; 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'; -DownloadStatus downloadQueuePersistenceStatus(DownloadStatus status) => - switch (status) { - DownloadStatus.downloading || - DownloadStatus.finalizing => DownloadStatus.queued, - _ => status, - }; - -/// Transfer samples are transient; only restart-relevant state reaches SQLite. -String encodeDownloadQueueItemForPersistence(DownloadItem item) => jsonEncode( - item.toJson() - ..['status'] = downloadQueuePersistenceStatus(item.status).name - ..['progress'] = 0.0 - ..['speedMBps'] = 0.0 - ..['bytesReceived'] = 0 - ..['bytesTotal'] = 0 - ..['preparationStage'] = '', -); +export 'package:spotiflac_android/services/download_queue_codec.dart' + show downloadQueuePersistenceStatus, encodeDownloadQueueItemForPersistence; /// Owns durable queue snapshots, including corrupt-row cleanup and write order. /// A failed write leaves the previous snapshot intact so a later flush retries. @@ -46,7 +31,7 @@ class DownloadQueuePersistence { List? _pendingItems; Future _write = Future.value(); Future _pauseWrite = Future.value(); - Map _saved = {}; + Map _saved = {}; static Future>> _loadRows() async { final database = AppStateDatabase.instance; @@ -78,41 +63,16 @@ class DownloadQueuePersistence { .catchError(onError); } - Future> restore() async { - final rows = await loadRows(); - _saved.clear(); - final pending = []; - for (final row in rows) { - final id = row['id']?.toString() ?? ''; - // Retain corrupt IDs so the next successful flush deletes those rows. - _saved[id] = (item: null, payload: null); - final payload = row['item_json']; - if (id.isEmpty || payload is! String || payload.isEmpty) continue; - try { - final decoded = jsonDecode(payload); - if (decoded is! Map) continue; - var item = DownloadItem.fromJson(Map.from(decoded)); - final status = downloadQueuePersistenceStatus(item.status); - _saved[id] = ( - item: item, - payload: - row['status']?.toString() == status.name && - payload == encodeDownloadQueueItemForPersistence(item) - ? payload - : null, - ); - if (item.status != status) { - item = item.copyWith(status: status, progress: 0); - } - if (status == DownloadStatus.queued || - status == DownloadStatus.skipped) { - pending.add(item); - } - } catch (_) { - continue; - } - } - return pending; + Future> restore() { + final restored = _write.then((_) async { + final decoded = await decodeDownloadQueueRows(await loadRows()); + _saved = decoded.saved; + return decoded.items; + }); + // A restore is ordered with durable writes now that decoding can yield to + // a worker. Its error still reaches the caller without poisoning retries. + _write = restored.then((_) {}, onError: (Object _, StackTrace _) {}); + return restored; } void schedule() { @@ -140,20 +100,34 @@ class DownloadQueuePersistence { Future _enqueue(List items) { return _write = _write .then((_) async { - final nextSaved = {}; + 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 payload = - identical(saved?.item, item) && saved?.payload != null - ? saved!.payload! - : encodeDownloadQueueItemForPersistence(item); - nextSaved[item.id] = (item: item, payload: payload); + 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, diff --git a/test/download_queue_codec_test.dart b/test/download_queue_codec_test.dart new file mode 100644 index 00000000..6d4de2f3 --- /dev/null +++ b/test/download_queue_codec_test.dart @@ -0,0 +1,227 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:isolate'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:spotiflac_android/models/download_item.dart'; +import 'package:spotiflac_android/models/track.dart'; +import 'package:spotiflac_android/services/download_queue_codec.dart'; + +DownloadItem _item( + int index, { + DownloadStatus status = DownloadStatus.queued, +}) => DownloadItem( + id: 'queue:$index', + track: Track( + id: 'example:$index', + name: 'Song $index', + artistName: 'Artist 日本語', + albumName: 'Album', + duration: 1000, + comment: 'Exact metadata — $index', + availability: const ServiceAvailability(deezer: true), + ), + service: 'extension.example', + status: status, + createdAt: DateTime.utc(2026).add(Duration(seconds: index)), + networkDownloadFolder: 'Albums', + qualityOverride: 'HI_RES', + preserveQualityVariant: true, + playlistName: 'Playlist', + playlistPosition: index + 1, + fromBatch: true, +); + +class _IsolateProbeTrack extends Track { + const _IsolateProbeTrack({super.comment}) + : super( + id: 'example:probe', + name: 'Song', + artistName: 'Artist', + albumName: 'Album', + duration: 1000, + ); + + @override + Map toJson() => { + ...super.toJson(), + 'encoded_on': Isolate.current.debugName, + }; +} + +class _ThrowingTrack extends Track { + const _ThrowingTrack() + : super( + id: 'example:failure', + name: 'Song', + artistName: 'Artist', + albumName: 'Album', + duration: 1000, + ); + + @override + Map toJson() => throw StateError('fixture codec failure'); +} + +String? _encodedIsolate(String payload) { + final decoded = jsonDecode(payload) as Map; + final track = decoded['track'] as Map; + return track['encoded_on'] as String?; +} + +void main() { + test( + 'ordinary serialization stays inline; a large batch uses a worker', + () async { + final item = _item(0).copyWith(track: const _IsolateProbeTrack()); + final inline = await encodeDownloadQueueItems([item]); + final worker = await encodeDownloadQueueItems(List.filled(64, item)); + expect(_encodedIsolate(inline.single), Isolate.current.debugName); + for (final payload in worker) { + expect(_encodedIsolate(payload), 'download-queue-persist'); + } + }, + ); + + test('chunked serialization preserves exact bytes and order', () async { + final items = List.generate( + 2100, + (index) => + _item( + index, + status: DownloadStatus.values[index % DownloadStatus.values.length], + ).copyWith( + progress: .7, + speedMBps: 8, + bytesReceived: 70, + bytesTotal: 100, + preparationStage: 'metadata', + ), + ); + expect( + await encodeDownloadQueueItems(items), + items.map(encodeDownloadQueueItemForPersistence).toList(), + ); + }); + + test( + 'a failed later worker chunk releases its wait and allows retry', + () async { + final items = List.generate(2100, (index) => _item(index)); + items[1500] = items[1500].copyWith(track: const _ThrowingTrack()); + await expectLater( + encodeDownloadQueueItems(items).timeout(const Duration(seconds: 10)), + throwsA( + isA().having( + (error) => error.toString(), + 'message', + contains('_ThrowingTrack'), + ), + ), + ); + items[1500] = _item(1500); + expect( + await encodeDownloadQueueItems(items), + items.map(encodeDownloadQueueItemForPersistence).toList(), + ); + }, + ); + + test( + 'chunked restore keeps repair IDs, statuses, ordering, and cache identity', + () async { + final items = List.generate(2100, (index) => _item(index)); + final rows = >[ + for (final item in items) + { + 'id': item.id, + 'status': item.status.name, + 'item_json': encodeDownloadQueueItemForPersistence(item), + }, + {'id': 'corrupt', 'status': 'queued', 'item_json': '{broken'}, + {'id': 'invalid', 'status': 'queued', 'item_json': '[]'}, + {'id': '', 'status': 'queued', 'item_json': '{}'}, + { + 'id': 'active', + 'status': 'downloading', + 'item_json': jsonEncode( + _item( + 2100, + status: DownloadStatus.downloading, + ).copyWith(id: 'active', progress: .9).toJson(), + ), + }, + { + 'id': 'terminal', + 'status': 'completed', + 'item_json': jsonEncode( + _item( + 2101, + status: DownloadStatus.completed, + ).copyWith(id: 'terminal').toJson(), + ), + }, + ]; + var eventLoopResumed = false; + Timer.run(() => eventLoopResumed = true); + final restored = await decodeDownloadQueueRows(rows); + expect(eventLoopResumed, isTrue); + expect(restored.items.map((item) => item.id), [ + ...items.map((item) => item.id), + 'active', + ]); + expect(restored.items.last.status, DownloadStatus.queued); + expect(restored.items.last.progress, 0); + for (var index = 0; index < items.length; index++) { + final item = restored.items[index]; + final saved = restored.saved[item.id]!; + expect(identical(saved.item, item), isTrue); + expect( + saved.payload, + encodeDownloadQueueItemForPersistence(items[index]), + ); + expect(item.track.comment, items[index].track.comment); + expect(item.track.availability!.deezer, isTrue); + } + expect(restored.saved['corrupt'], (item: null, payload: null)); + expect(restored.saved['invalid'], (item: null, payload: null)); + expect(restored.saved[''], (item: null, payload: null)); + expect(restored.saved['active']!.payload, isNull); + expect( + restored.saved['terminal']!.item!.status, + DownloadStatus.completed, + ); + }, + ); + + test( + 'a single large payload also yields JSON hydration to a worker', + () async { + final item = _item(0).copyWith( + track: _item(0).track.copyWith(comment: List.filled(70000, 'x').join()), + ); + var eventLoopResumed = false; + Timer.run(() => eventLoopResumed = true); + final restored = await decodeDownloadQueueRows([ + { + 'id': item.id, + 'status': item.status.name, + 'item_json': encodeDownloadQueueItemForPersistence(item), + }, + ]); + expect(eventLoopResumed, isTrue); + expect(restored.items.single.track.comment, item.track.comment); + }, + ); + + test('a single large metadata field serializes in a worker', () async { + final item = _item(0).copyWith( + track: _IsolateProbeTrack(comment: List.filled(70000, 'x').join()), + ); + final encoded = await encodeDownloadQueueItems([item]); + expect(_encodedIsolate(encoded.single), 'download-queue-persist'); + final decoded = jsonDecode(encoded.single) as Map; + final track = decoded['track'] as Map; + expect(track['comment'], item.track.comment); + }); +} diff --git a/test/download_queue_legacy_migration_test.dart b/test/download_queue_legacy_migration_test.dart new file mode 100644 index 00000000..0c9028c4 --- /dev/null +++ b/test/download_queue_legacy_migration_test.dart @@ -0,0 +1,200 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:spotiflac_android/services/app_state_database.dart'; +import 'package:spotiflac_android/services/download_queue_codec.dart'; + +import 'support/sqlite_process_database.dart'; + +const _marker = 'app_state_migrated_queue_to_sqlite_v1'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + Future fixture() async { + final database = await SqliteProcessDatabase.open(); + addTearDown(database.close); + await database.execute(''' + CREATE TABLE download_queue_items ( + id TEXT PRIMARY KEY, + item_json TEXT NOT NULL, + status TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + '''); + return database; + } + + Future preferences(String raw) { + SharedPreferences.setMockInitialValues({'download_queue': raw}); + return SharedPreferences.getInstance(); + } + + test( + 'legacy migration preserves unknown fields and original normalization', + () async { + final database = await fixture(); + final queued = { + 'id': 'queued', + 'createdAt': '2026-01-01', + 'future_field': {'unknown': 'preserved'}, + }; + final downloading = { + 'id': 'active', + 'status': 'downloading', + 'progress': .8, + 'speedMBps': 3, + 'bytesReceived': 80, + 'bytesTotal': 100, + 'preparationStage': 'legacy-stage', + }; + final raw = jsonEncode([ + queued, + downloading, + {'id': 'terminal', 'status': 'completed'}, + {'id': 'skipped', 'status': 'skipped'}, + {'id': '', 'status': 'queued'}, + null, + 1, + 'ignored', + ]); + final prefs = await preferences(raw); + expect( + await AppStateDatabase.instance.migrateQueueFromSharedPreferences( + preferencesOverride: prefs, + databaseOverride: database, + ), + isTrue, + ); + final rows = await database.query('download_queue_items'); + expect(rows.map((row) => row['id']), ['queued', 'active']); + expect(rows.first['item_json'], jsonEncode(queued)); + expect(rows.first['created_at'], '2026-01-01'); + expect( + rows.last['item_json'], + jsonEncode({ + ...downloading, + 'status': 'queued', + 'progress': 0.0, + 'speedMBps': 0.0, + 'bytesReceived': 0, + }), + ); + expect(rows.last['created_at'], rows.last['updated_at']); + expect(rows.every((row) => row['status'] == 'queued'), isTrue); + expect(prefs.getBool(_marker), isTrue); + expect(prefs.getString('download_queue'), raw); + expect( + await AppStateDatabase.instance.migrateQueueFromSharedPreferences( + preferencesOverride: prefs, + databaseOverride: database, + ), + isFalse, + ); + }, + ); + + test( + 'malformed legacy data keeps its retry marker while non-list data finishes', + () async { + final database = await fixture(); + for (final raw in ['{broken', '[{"id":"valid"},{"id":5}]']) { + final prefs = await preferences(raw); + expect( + await AppStateDatabase.instance.migrateQueueFromSharedPreferences( + preferencesOverride: prefs, + databaseOverride: database, + ), + isFalse, + ); + expect(prefs.getBool(_marker), isNull); + expect(prefs.getString('download_queue'), raw); + expect(await database.query('download_queue_items'), isEmpty); + } + for (final raw in ['', '{}']) { + final prefs = await preferences(raw); + expect( + await AppStateDatabase.instance.migrateQueueFromSharedPreferences( + preferencesOverride: prefs, + databaseOverride: database, + ), + isFalse, + ); + expect(prefs.getBool(_marker), isTrue); + } + }, + ); + + test( + 'large legacy preparation yields to a worker and preserves exact maps', + () async { + final comment = List.filled(5000, 'x').join(); + final entries = [ + for (var index = 0; index < 30; index++) + {'id': '$index', 'status': 'queued', 'comment': comment}, + ]; + var eventLoopResumed = false; + Timer.run(() => eventLoopResumed = true); + final rows = await prepareLegacyDownloadQueueRows( + jsonEncode(entries), + 'now', + ); + expect(eventLoopResumed, isTrue); + expect(rows!.map((row) => row['item_json']), entries.map(jsonEncode)); + expect(rows.every((row) => row['created_at'] == 'now'), isTrue); + }, + ); + + test( + 'a failed large legacy import rolls back rows and leaves preferences retryable', + () async { + final database = await fixture(); + await database.insert('download_queue_items', { + 'id': 'existing', + 'item_json': '{}', + 'status': 'queued', + 'created_at': 'before', + 'updated_at': 'before', + }); + await database.execute(''' + CREATE TRIGGER reject_queue BEFORE INSERT ON download_queue_items + WHEN NEW.id = 'reject' BEGIN SELECT RAISE(ABORT, 'fixture rejection'); END + '''); + final comment = List.filled(70000, 'x').join(); + final raw = jsonEncode([ + {'id': 'first', 'status': 'queued', 'comment': comment}, + {'id': 'reject', 'status': 'queued'}, + ]); + final prefs = await preferences(raw); + expect( + await AppStateDatabase.instance.migrateQueueFromSharedPreferences( + preferencesOverride: prefs, + databaseOverride: database, + ), + isFalse, + ); + expect( + (await database.query('download_queue_items')).map((row) => row['id']), + ['existing'], + ); + expect(prefs.getBool(_marker), isNull); + expect(prefs.getString('download_queue'), raw); + await database.execute('DROP TRIGGER reject_queue'); + expect( + await AppStateDatabase.instance.migrateQueueFromSharedPreferences( + preferencesOverride: prefs, + databaseOverride: database, + ), + isTrue, + ); + expect( + (await database.query('download_queue_items')).map((row) => row['id']), + ['existing', 'first', 'reject'], + ); + expect(prefs.getBool(_marker), isTrue); + }, + ); +} diff --git a/test/download_queue_persistence_test.dart b/test/download_queue_persistence_test.dart index f53e5965..6ed54cea 100644 --- a/test/download_queue_persistence_test.dart +++ b/test/download_queue_persistence_test.dart @@ -329,6 +329,118 @@ void main() { expect(await persistence.loadUserPaused(), isFalse); }); + test( + 'large worker snapshots retry and retain the latest ordered write', + () async { + final store = _Store(); + var items = List.generate(2100, (index) => _item('initial:$index')); + final latest = List.generate( + 1100, + (index) => _item('latest:$index', status: DownloadStatus.skipped), + ); + final errors = []; + final started = Completer(); + final release = Completer(); + var attempts = 0; + final persistence = DownloadQueuePersistence( + currentItems: () => items, + onError: errors.add, + loadRows: store.load, + applyChanges: ({required upserts, required deletedIds}) async { + attempts++; + if (attempts == 1) throw StateError('disk unavailable'); + if (attempts == 2) { + started.complete(); + await release.future; + } + await store.apply(upserts: upserts, deletedIds: deletedIds); + }, + ); + await persistence.flush(); + expect(errors, hasLength(1)); + expect(store.rows, isEmpty); + final initialWrite = persistence.flush(); + await started.future; + items = latest; + final latestWrite = persistence.flush(); + persistence.dispose(); + release.complete(); + await Future.wait([initialWrite, latestWrite]); + expect(store.rows.keys, latest.map((item) => item.id)); + for (final item in latest) { + expect( + store.rows[item.id]!['item_json'], + encodeDownloadQueueItemForPersistence(item), + ); + } + await persistence.flush(); + expect(attempts, 3); + expect(errors, hasLength(1)); + }, + ); + + test( + 'restore worker cannot be overtaken by a flush or pending disposal', + () async { + final store = _Store(); + final old = List.generate(2100, (index) => _item('old:$index')); + for (final item in old) { + store.seed(item); + } + final loaded = Completer(); + final release = Completer(); + var items = [_item('latest')]; + var disposed = false; + final errors = []; + final persistence = DownloadQueuePersistence( + currentItems: () { + if (disposed) throw StateError('provider disposed'); + return items; + }, + onError: errors.add, + loadRows: () async { + final rows = await store.load(); + loaded.complete(); + await release.future; + return rows; + }, + applyChanges: store.apply, + ); + final restored = persistence.restore(); + await loaded.future; + final flushed = persistence.flush(); + persistence.schedule(); + persistence.dispose(); + disposed = true; + items = []; + expect(store.writes, 0); + release.complete(); + expect( + (await restored).map((item) => item.id), + old.map((item) => item.id), + ); + await flushed; + expect(store.rows.keys, ['latest']); + expect(errors, isEmpty); + }, + ); + + test( + 'a failed restore propagates without poisoning later durable writes', + () async { + final store = _Store(); + final persistence = DownloadQueuePersistence( + currentItems: () => [_item('latest')], + onError: (error) => fail('$error'), + loadRows: () async => throw StateError('database unavailable'), + applyChanges: store.apply, + ); + await expectLater(persistence.restore(), throwsStateError); + await persistence.flush(); + expect(store.rows.keys, ['latest']); + }, + ); + test('SQLite restores cancelled rows in creation order for retry', () async { SharedPreferences.setMockInitialValues({ 'app_state_migrated_queue_to_sqlite_v1': true, diff --git a/test/download_queue_read_test.dart b/test/download_queue_read_test.dart new file mode 100644 index 00000000..164046cd --- /dev/null +++ b/test/download_queue_read_test.dart @@ -0,0 +1,468 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:sqflite/sqflite.dart'; +import 'package:spotiflac_android/services/app_state_database.dart'; + +import 'support/sqlite_process_database.dart'; + +class _Reads { + int transactions = 0; + final sizes = []; + final batchSizes = []; + final valueSizes = []; + final statements = []; + String? legacyVersion; +} + +class _RecordingBatch implements Batch { + _RecordingBatch(this.batch, this.reads); + final Batch batch; + final _Reads reads; + int size = 0; + + @override + void insert( + String table, + Map values, { + String? nullColumnHack, + ConflictAlgorithm? conflictAlgorithm, + }) { + size++; + batch.insert( + table, + values, + nullColumnHack: nullColumnHack, + conflictAlgorithm: conflictAlgorithm, + ); + } + + @override + void delete(String table, {String? where, List? whereArgs}) { + size++; + batch.delete(table, where: where, whereArgs: whereArgs); + } + + @override + Future> commit({ + bool? exclusive, + bool? noResult, + bool? continueOnError, + }) { + reads.batchSizes.add(size); + return batch.commit( + exclusive: exclusive, + noResult: noResult, + continueOnError: continueOnError, + ); + } + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +class _RecordingTransaction implements Transaction { + _RecordingTransaction(this.transaction, this.reads); + final Transaction transaction; + final _Reads reads; + + @override + Batch batch() => _RecordingBatch(transaction.batch(), reads); + + @override + Future>> rawQuery( + String sql, [ + List? arguments, + ]) async { + final rows = await transaction.rawQuery(sql, arguments); + if (sql.contains('sqlite_master')) { + if (reads.legacyVersion != null && rows.isNotEmpty) { + return [ + {'version': reads.legacyVersion}, + ]; + } + } else { + reads.sizes.add(rows.length); + reads.statements.add(sql); + } + return rows; + } + + @override + Future rawInsert(String sql, [List? arguments]) { + reads.valueSizes.add((arguments?.length ?? 0) ~/ 5); + return transaction.rawInsert(sql, arguments); + } + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +class _RecordingDatabase implements Database { + _RecordingDatabase(this.database, this.reads); + @override + final Database database; + final _Reads reads; + + @override + Future>> query( + String table, { + bool? distinct, + List? columns, + String? where, + List? whereArgs, + String? groupBy, + String? having, + String? orderBy, + int? limit, + int? offset, + }) async { + final rows = await database.query( + table, + distinct: distinct, + columns: columns, + where: where, + whereArgs: whereArgs, + groupBy: groupBy, + having: having, + orderBy: orderBy, + limit: limit, + offset: offset, + ); + reads.sizes.add(rows.length); + return rows; + } + + @override + Future transaction( + Future Function(Transaction txn) action, { + bool? exclusive, + }) { + reads.transactions++; + return database.transaction( + (txn) => action(_RecordingTransaction(txn, reads)), + exclusive: exclusive, + ); + } + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + setUp(() => SharedPreferences.setMockInitialValues({})); + + Future fixture(int count) async { + final database = await SqliteProcessDatabase.open(); + addTearDown(database.close); + await database.execute(''' + CREATE TABLE download_queue_items ( + id TEXT PRIMARY KEY, + item_json TEXT NOT NULL, + status TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + '''); + await database.execute( + 'CREATE INDEX idx_download_queue_items_created ON download_queue_items(created_at ASC)', + ); + await database.execute( + 'CREATE INDEX idx_download_queue_items_status ON download_queue_items(status)', + ); + await database.transaction((txn) async { + for (var index = 0; index < count; index++) { + await txn.insert('download_queue_items', { + 'id': 'queue:${count - index}', + 'item_json': '{"fixture":$index}', + 'status': 'queued', + // Identical timestamps exercise rowid tie-breaking across pages. + 'created_at': index % 3 == 0 ? '' : '2026-10-07T00:00:00.000Z', + 'updated_at': '2026-10-07', + }); + } + }); + return database; + } + + test( + 'small queues keep a single bounded query without a transaction', + () async { + final database = await fixture(256); + for (final size in [0, 1, 16, 63, 256]) { + await database.execute( + 'UPDATE download_queue_items SET status = CASE WHEN rowid <= ? THEN ? ELSE ? END', + [size, 'queued', 'completed'], + ); + final reads = _Reads(); + final rows = await AppStateDatabase.instance + .getPendingDownloadQueueRows( + databaseOverride: _RecordingDatabase(database, reads), + ); + expect(rows, hasLength(size)); + expect(reads.sizes, [size]); + expect(reads.transactions, 0); + expect(rows.every((row) => !row.containsKey('__queue_order')), isTrue); + } + }, + ); + + test( + 'large pages preserve SQLite order, ties, statuses, and exact row data', + () async { + final database = await fixture(2100); + await database.execute(''' + UPDATE download_queue_items SET status = + CASE rowid % 6 + WHEN 0 THEN 'completed' + WHEN 1 THEN 'failed' + WHEN 2 THEN 'queued' + WHEN 3 THEN 'skipped' + WHEN 4 THEN 'downloading' + ELSE 'finalizing' + END + '''); + final expected = await database.query( + 'download_queue_items', + where: 'status IN (?, ?, ?, ?)', + whereArgs: ['queued', 'downloading', 'finalizing', 'skipped'], + orderBy: 'created_at ASC, rowid ASC', + ); + final reads = _Reads(); + final actual = await AppStateDatabase.instance + .getPendingDownloadQueueRows( + databaseOverride: _RecordingDatabase(database, reads), + ); + expect(actual, expected); + expect(actual, hasLength(1400)); + expect(reads.transactions, 1); + expect(reads.sizes.first, 257); + expect(reads.sizes.skip(1).every((size) => size <= 256), isTrue); + expect(reads.sizes.skip(1).reduce((a, b) => a + b), actual.length); + expect(actual.map((row) => row['id']).toSet().length, actual.length); + }, + ); + + test( + 'bulk changes bound batches but keep delete/upsert order and one transaction', + () async { + final database = await fixture(700); + final existing = await database.query( + 'download_queue_items', + orderBy: 'rowid ASC', + ); + final deleted = existing + .take(550) + .map((row) => row['id'] as String) + .toList(); + final upserts = >[ + for (var index = 0; index < 700; index++) + { + 'id': 'new:$index', + 'item_json': '{"value":$index}', + 'status': 'queued', + 'created_at': '2026-10-07', + 'updated_at': '2026-10-07', + }, + ]; + final reads = _Reads(); + await AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: upserts, + deletedIds: deleted, + databaseOverride: _RecordingDatabase(database, reads), + ); + expect(reads.transactions, 1); + expect(reads.batchSizes, [256, 256, 38]); + expect(reads.valueSizes, [192, 192, 192, 124]); + final actual = await database.query( + 'download_queue_items', + orderBy: 'rowid ASC', + ); + expect(actual, [...existing.skip(550), ...upserts]); + }, + ); + + test( + 'older SQLite and databases without the created index retain row parity', + () async { + for (final withoutIndex in [false, true]) { + final database = await fixture(700); + if (withoutIndex) { + await database.execute('DROP INDEX idx_download_queue_items_created'); + } + final expected = await database.query( + 'download_queue_items', + orderBy: 'created_at ASC, rowid ASC', + ); + final reads = _Reads()..legacyVersion = '3.9.2'; + final actual = await AppStateDatabase.instance + .getPendingDownloadQueueRows( + databaseOverride: _RecordingDatabase(database, reads), + ); + expect(actual, expected); + expect(reads.sizes, [257, 256, 256, 188]); + expect(reads.statements.last.contains('(created_at, rowid) >'), false); + expect(reads.statements.last.contains('INDEXED BY'), !withoutIndex); + expect( + reads.statements.last.contains('AND created_at >= ?'), + !withoutIndex, + ); + } + }, + ); + + test( + 'small writes retain one ordinary batch without VALUES statements', + () async { + final database = await fixture(63); + final rows = await database.query('download_queue_items'); + final reads = _Reads(); + await AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: rows, + deletedIds: const [], + databaseOverride: _RecordingDatabase(database, reads), + ); + expect(reads.transactions, 1); + expect(reads.batchSizes, [63]); + expect(reads.valueSizes, isEmpty); + expect(await database.query('download_queue_items'), rows); + }, + ); + + test( + 'VALUES chunks match replacement order and mixed raw maps exactly', + () async { + final actualDatabase = await fixture(3); + final oracleDatabase = await fixture(3); + for (final database in [actualDatabase, oracleDatabase]) { + await database.execute( + 'ALTER TABLE download_queue_items ADD COLUMN fixture_extra TEXT DEFAULT NULL', + ); + } + final upserts = >[ + for (var index = 0; index < 700; index++) + { + // Duplicates within and across 192-row statement boundaries must + // match the former sequential INSERT OR REPLACE rowid behavior. + 'updated_at': '2026-10-07', + 'created_at': '2026-10-07', + 'status': 'queued', + 'item_json': '{"value":$index}', + 'id': index % 192 < 2 ? 'duplicate' : 'new:$index', + if (index == 383) 'fixture_extra': 'raw metadata 日本語', + }, + ]; + const deletions = ['queue:2', 'duplicate']; + await oracleDatabase.transaction((txn) async { + for (final id in deletions) { + await txn.delete( + 'download_queue_items', + where: 'id = ?', + whereArgs: [id], + ); + } + for (final row in upserts) { + await txn.insert( + 'download_queue_items', + row, + conflictAlgorithm: ConflictAlgorithm.replace, + ); + } + }); + final reads = _Reads(); + await AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: upserts, + deletedIds: deletions, + databaseOverride: _RecordingDatabase(actualDatabase, reads), + ); + expect( + await actualDatabase.query( + 'download_queue_items', + orderBy: 'rowid ASC', + ), + await oracleDatabase.query( + 'download_queue_items', + orderBy: 'rowid ASC', + ), + ); + expect(reads.transactions, 1); + expect(reads.batchSizes, [2, 1]); + expect(reads.valueSizes, [192, 191, 192, 124]); + }, + ); + + test( + 'invalid raw maps roll back earlier canonical VALUES statements', + () async { + final database = await fixture(3); + final before = await database.query('download_queue_items'); + final upserts = >[ + for (var index = 0; index < 500; index++) + { + 'id': 'new:$index', + 'item_json': '{}', + 'status': 'queued', + 'created_at': '2026-10-07', + 'updated_at': '2026-10-07', + if (index == 499) 'unknown_column': 'cannot persist', + }, + ]; + await expectLater( + AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: upserts, + deletedIds: ['queue:1'], + databaseOverride: database, + ), + throwsStateError, + ); + expect(await database.query('download_queue_items'), before); + }, + ); + + test( + 'a late failed chunk rolls back preceding deletions and inserted chunks', + () async { + final database = await fixture(600); + final before = await database.query( + 'download_queue_items', + orderBy: 'rowid ASC', + ); + await database.execute(''' + CREATE TRIGGER reject_queue BEFORE INSERT ON download_queue_items + WHEN NEW.id = 'reject' BEGIN SELECT RAISE(ABORT, 'fixture rejection'); END + '''); + final upserts = >[ + for (var index = 0; index < 600; index++) + { + 'id': index == 599 ? 'reject' : 'new:$index', + 'item_json': '{}', + 'status': 'queued', + 'created_at': '2026-10-07', + 'updated_at': '2026-10-07', + }, + ]; + await expectLater( + AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: upserts, + deletedIds: before.map((row) => row['id'] as String).toList(), + databaseOverride: database, + ), + throwsStateError, + ); + expect( + await database.query('download_queue_items', orderBy: 'rowid ASC'), + before, + ); + await database.execute('DROP TRIGGER reject_queue'); + await AppStateDatabase.instance.applyPendingDownloadQueueChanges( + upserts: upserts, + deletedIds: before.map((row) => row['id'] as String).toList(), + databaseOverride: database, + ); + expect( + await database.query('download_queue_items', orderBy: 'rowid ASC'), + upserts, + ); + }, + ); +}