mirror of
https://github.com/zarzet/SpotiFLAC-Mobile.git
synced 2026-10-21 15:02:21 +02:00
perf(queue): bound SQLite transfers and move large codecs off UI
This commit is contained in:
zarzet committed
1 parent
4d0fe68f17
commit
1094ed72f5
8 files changed
+2081
-115
No files matched your search
@@ -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<RestoredDownloadQueue> _inlineRestore(
|
||||
List<Map<String, dynamic>> rows,
|
||||
) async {
|
||||
final saved = <String, DownloadQueueSavedItem>{};
|
||||
final pending = <DownloadItem>[];
|
||||
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<String, dynamic>.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<List<String>> _inlineEncode(List<DownloadItem> items) async => [
|
||||
for (final item in items) encodeDownloadQueueItemForPersistence(item),
|
||||
];
|
||||
|
||||
Future<void> _inlineWrite(
|
||||
Database database,
|
||||
List<Map<String, dynamic>> 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<DownloadItem> _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<String, dynamic> toJson() {
|
||||
if (Isolate.current.debugName != expectedIsolate) {
|
||||
throw StateError('Small queue serialization left its owning isolate');
|
||||
}
|
||||
return original.toJson();
|
||||
}
|
||||
}
|
||||
|
||||
class _Measurement<T> {
|
||||
_Measurement(this.value, this.elapsedMs, this.maxGapMs, this.ticks);
|
||||
final T value;
|
||||
final double elapsedMs;
|
||||
final double maxGapMs;
|
||||
final int ticks;
|
||||
|
||||
Map<String, Object> toJson() => {
|
||||
'elapsed_ms': elapsedMs,
|
||||
'max_ui_gap_ms': maxGapMs,
|
||||
'ui_ticks': ticks,
|
||||
};
|
||||
}
|
||||
|
||||
Future<_Measurement<T>> _measure<T>(Future<T> 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<void>.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 = <String, Map<String, dynamic>>{};
|
||||
int writes = 0;
|
||||
|
||||
Future<List<Map<String, dynamic>>> load() async => rows.values.toList();
|
||||
|
||||
Future<void> apply({
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Map<String, dynamic> _queueReport(
|
||||
IntegrationTestWidgetsFlutterBinding binding,
|
||||
) {
|
||||
binding.reportData ??= {};
|
||||
return binding.reportData!.putIfAbsent(
|
||||
'queue_workers',
|
||||
() => <String, dynamic>{},
|
||||
)
|
||||
as Map<String, dynamic>;
|
||||
}
|
||||
|
||||
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 = <Map<String, dynamic>>[
|
||||
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 = <String, dynamic>{
|
||||
'count': items.length,
|
||||
'mode': kDebugMode ? 'debug' : 'profile/release',
|
||||
'payload_utf16_units': rows.fold<int>(
|
||||
0,
|
||||
(total, row) => total + (row['item_json'] as String).length,
|
||||
),
|
||||
'restore': <Map<String, Object>>[],
|
||||
'encode': <Map<String, Object>>[],
|
||||
'small': <Map<String, Object>>[],
|
||||
};
|
||||
// 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<RestoredDownloadQueue> baselineRestore;
|
||||
late _Measurement<RestoredDownloadQueue> workerRestore;
|
||||
late _Measurement<List<String>> baselineEncode;
|
||||
late _Measurement<List<String>> 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<Map<String, Object>>).add({
|
||||
'sample': sample,
|
||||
'baseline': baselineRestore.toJson(),
|
||||
'worker': workerRestore.toJson(),
|
||||
});
|
||||
(report['encode'] as List<Map<String, Object>>).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<Map<String, Object>>).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 = <Object>[];
|
||||
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 = <Map<String, Object>>[];
|
||||
for (var sample = 0; sample < 5; sample++) {
|
||||
final index = original.length - 1;
|
||||
items = List<DownloadItem>.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 = <Map<String, Object>>[];
|
||||
for (var sample = 0; sample < 5; sample++) {
|
||||
Future<RestoredDownloadQueue> 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<RestoredDownloadQueue> before;
|
||||
late _Measurement<List<DownloadItem>> 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 = <Map<String, dynamic>>[
|
||||
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 = <Map<String, Object>>[];
|
||||
for (var sample = 0; sample < 5; sample++) {
|
||||
Future<void> optimized() =>
|
||||
AppStateDatabase.instance.applyPendingDownloadQueueChanges(
|
||||
upserts: upserts,
|
||||
deletedIds: const [],
|
||||
databaseOverride: database,
|
||||
);
|
||||
late _Measurement<void> before;
|
||||
late _Measurement<void> 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 = <Map<String, dynamic>>[
|
||||
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<DatabaseException>()),
|
||||
);
|
||||
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);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -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<bool> migrateQueueFromSharedPreferences() async {
|
||||
final prefs = await _prefs;
|
||||
Future<bool> 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<Map<Object?, Object?>>()) {
|
||||
final map = Map<String, dynamic>.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<List<Map<String, dynamic>>> getPendingDownloadQueueRows() async {
|
||||
final db = await database;
|
||||
return db.query(
|
||||
Future<List<Map<String, dynamic>>> 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 <String>[];
|
||||
// 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 = <Map<String, dynamic>>[];
|
||||
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<String, dynamic>.of(row)..remove('__queue_order'));
|
||||
}
|
||||
if (page.length < pageSize) break;
|
||||
}
|
||||
return rows;
|
||||
});
|
||||
}
|
||||
|
||||
Future<void> replacePendingDownloadQueueRows(
|
||||
@@ -412,25 +456,86 @@ class AppStateDatabase {
|
||||
Future<void> applyPendingDownloadQueueChanges({
|
||||
required List<Map<String, dynamic>> upserts,
|
||||
required List<String> 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<void> _writeQueueChanges(
|
||||
Transaction txn,
|
||||
List<Map<String, dynamic>> upserts,
|
||||
List<String> deletedIds,
|
||||
) async {
|
||||
var batch = txn.batch();
|
||||
var pending = 0;
|
||||
Future<void> 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 = <Object?>[];
|
||||
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<String, dynamic> 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<List<Map<String, dynamic>>> getRecentAccessRows({int? limit}) async {
|
||||
final db = await database;
|
||||
return db.query(
|
||||
|
||||
@@ -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<DownloadItem> items;
|
||||
final Map<String, DownloadQueueSavedItem> saved;
|
||||
}
|
||||
|
||||
RestoredDownloadQueue _decodeRows(List<Map<String, dynamic>> rows) {
|
||||
final saved = <String, DownloadQueueSavedItem>{};
|
||||
final pending = <DownloadItem>[];
|
||||
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<String, dynamic>.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<Map<String, dynamic>> 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<RestoredDownloadQueue> decodeDownloadQueueRows(
|
||||
List<Map<String, dynamic>> rows,
|
||||
) async {
|
||||
if (!_largeRows(rows)) return _decodeRows(rows);
|
||||
return _runQueueCodecWorker<RestoredDownloadQueue>(
|
||||
_QueueCodecMode.restore,
|
||||
rows,
|
||||
);
|
||||
}
|
||||
|
||||
List<String> _encodeItems(List<DownloadItem> items) => [
|
||||
for (final item in items) encodeDownloadQueueItemForPersistence(item),
|
||||
];
|
||||
|
||||
bool _largeItems(List<DownloadItem> 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<List<String>> encodeDownloadQueueItems(List<DownloadItem> items) async {
|
||||
if (!_largeItems(items)) return _encodeItems(items);
|
||||
return _runQueueCodecWorker<List<String>>(_QueueCodecMode.persist, items);
|
||||
}
|
||||
|
||||
enum _QueueCodecMode { restore, persist }
|
||||
|
||||
typedef _QueueWorkerRequest = ({SendPort replies, _QueueCodecMode mode});
|
||||
|
||||
Future<T> _runQueueCodecWorker<T>(
|
||||
_QueueCodecMode mode,
|
||||
List<Object> values,
|
||||
) async {
|
||||
final replies = ReceivePort();
|
||||
final ready = Completer<SendPort>();
|
||||
final result = Completer<T>();
|
||||
Completer<void>? 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<Object?> && message.length == 1) {
|
||||
result.complete(message.single as T);
|
||||
} else if (!result.isCompleted) {
|
||||
final error = message is List<Object?> && 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<void>();
|
||||
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 = <DownloadItem>[];
|
||||
final saved = <String, DownloadQueueSavedItem>{};
|
||||
final payloads = <String>[];
|
||||
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<Object?>).cast<Map<String, dynamic>>(),
|
||||
);
|
||||
items.addAll(decoded.items);
|
||||
saved.addAll(decoded.saved);
|
||||
} else {
|
||||
payloads.addAll(
|
||||
_encodeItems((message as List<Object?>).cast<DownloadItem>()),
|
||||
);
|
||||
}
|
||||
request.replies.send(true);
|
||||
}
|
||||
}
|
||||
|
||||
List<Map<String, dynamic>>? _legacyRows((String, String) request) {
|
||||
final (raw, nowIso) = request;
|
||||
final decoded = jsonDecode(raw);
|
||||
if (decoded is! List) return null;
|
||||
final rows = <Map<String, dynamic>>[];
|
||||
for (final entry in decoded.whereType<Map<Object?, Object?>>()) {
|
||||
final map = Map<String, dynamic>.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<List<Map<String, dynamic>>?> 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');
|
||||
}
|
||||
@@ -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<DownloadItem>? _pendingItems;
|
||||
Future<void> _write = Future<void>.value();
|
||||
Future<void> _pauseWrite = Future<void>.value();
|
||||
Map<String, ({DownloadItem? item, String? payload})> _saved = {};
|
||||
Map<String, DownloadQueueSavedItem> _saved = {};
|
||||
|
||||
static Future<List<Map<String, dynamic>>> _loadRows() async {
|
||||
final database = AppStateDatabase.instance;
|
||||
@@ -78,41 +63,16 @@ class DownloadQueuePersistence {
|
||||
.catchError(onError);
|
||||
}
|
||||
|
||||
Future<List<DownloadItem>> restore() async {
|
||||
final rows = await loadRows();
|
||||
_saved.clear();
|
||||
final pending = <DownloadItem>[];
|
||||
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<String, dynamic>.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<List<DownloadItem>> 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<void>((_) {}, onError: (Object _, StackTrace _) {});
|
||||
return restored;
|
||||
}
|
||||
|
||||
void schedule() {
|
||||
@@ -140,20 +100,34 @@ class DownloadQueuePersistence {
|
||||
Future<void> _enqueue(List<DownloadItem> items) {
|
||||
return _write = _write
|
||||
.then((_) async {
|
||||
final nextSaved = <String, ({DownloadItem? item, String? payload})>{};
|
||||
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 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,
|
||||
|
||||
@@ -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<String, dynamic> 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<String, dynamic> toJson() => throw StateError('fixture codec failure');
|
||||
}
|
||||
|
||||
String? _encodedIsolate(String payload) {
|
||||
final decoded = jsonDecode(payload) as Map<String, dynamic>;
|
||||
final track = decoded['track'] as Map<String, dynamic>;
|
||||
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<RemoteError>().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 = <Map<String, dynamic>>[
|
||||
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<String, dynamic>;
|
||||
final track = decoded['track'] as Map<String, dynamic>;
|
||||
expect(track['comment'], item.track.comment);
|
||||
});
|
||||
}
|
||||
@@ -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<SqliteProcessDatabase> 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<SharedPreferences> 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);
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -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 = <Object>[];
|
||||
final started = Completer<void>();
|
||||
final release = Completer<void>();
|
||||
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<void>();
|
||||
final release = Completer<void>();
|
||||
var items = [_item('latest')];
|
||||
var disposed = false;
|
||||
final errors = <Object>[];
|
||||
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,
|
||||
|
||||
@@ -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 = <int>[];
|
||||
final batchSizes = <int>[];
|
||||
final valueSizes = <int>[];
|
||||
final statements = <String>[];
|
||||
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<String, Object?> values, {
|
||||
String? nullColumnHack,
|
||||
ConflictAlgorithm? conflictAlgorithm,
|
||||
}) {
|
||||
size++;
|
||||
batch.insert(
|
||||
table,
|
||||
values,
|
||||
nullColumnHack: nullColumnHack,
|
||||
conflictAlgorithm: conflictAlgorithm,
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
void delete(String table, {String? where, List<Object?>? whereArgs}) {
|
||||
size++;
|
||||
batch.delete(table, where: where, whereArgs: whereArgs);
|
||||
}
|
||||
|
||||
@override
|
||||
Future<List<Object?>> 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<List<Map<String, Object?>>> rawQuery(
|
||||
String sql, [
|
||||
List<Object?>? 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<int> rawInsert(String sql, [List<Object?>? 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<List<Map<String, Object?>>> query(
|
||||
String table, {
|
||||
bool? distinct,
|
||||
List<String>? columns,
|
||||
String? where,
|
||||
List<Object?>? 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<T> transaction<T>(
|
||||
Future<T> 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<SqliteProcessDatabase> 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 = <Map<String, dynamic>>[
|
||||
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 = <Map<String, dynamic>>[
|
||||
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 = <Map<String, dynamic>>[
|
||||
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 = <Map<String, dynamic>>[
|
||||
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,
|
||||
);
|
||||
},
|
||||
);
|
||||
}
|
||||
Reference in new issue
Block a user