perf(collections): stage large playlist additions off the UI thread

This commit is contained in:
zarzet committed 2026-10-08 03:19:45 +07:00
1 parent aa54c57590
commit 460f8ebc2d
12 files changed
+1284 -53

No files matched your search

@@ -0,0 +1,259 @@
import 'dart:convert';
import 'dart:io';
import 'dart:isolate';
import 'package:crypto/crypto.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/library_collections.dart';
import 'package:spotiflac_android/services/library_collections_database.dart';
import 'package:spotiflac_android/services/platform_bridge.dart';
import 'package:spotiflac_android/services/playlist_batch_writer.dart';
import '../test/support/library_collections_benchmark.dart';
import '../test/support/playlist_batch_benchmark.dart';
Future<String> _digest(Object value) => Isolate.run(
() => sha256.convert(utf8.encode(jsonEncode(value))).toString(),
);
void main() {
final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized();
final report = <String, dynamic>{
'platform': Platform.operatingSystem,
'note': 'Five alternating paired samples. Heartbeat gaps are not FPS.',
};
late Directory root;
late Database db;
final now = DateTime.utc(2026, 10, 8);
final stamp = now.toIso8601String();
var completed = 0;
setUp(() async {
root = await Directory.systemTemp.createTemp('playlist-device-');
final extensions = await Directory('${root.path}/extensions').create();
final data = await Directory('${root.path}/data').create();
final output = await Directory('${root.path}/output').create();
await PlatformBridge.initExtensionSystem(
extensions.path,
data.path,
masterKey: base64Encode(List<int>.filled(32, 1)),
lyricsProviders: const [],
lyricsFetchOptions: const {},
allowedDirectories: [output.path],
);
final implementations = await PlatformBridge.getBackendImplementations();
expect(implementations['extensions'], 'rust');
db = await openDatabase(
'${root.path}/library_collections.db',
version: 2,
onConfigure: (db) async {
await db.execute('PRAGMA foreign_keys=ON');
await db.execute('PRAGMA recursive_triggers=ON');
await db.execute('PRAGMA auto_vacuum=INCREMENTAL');
await db.rawQuery('PRAGMA journal_mode=WAL');
await db.execute('PRAGMA synchronous=NORMAL');
await db.rawQuery('PRAGMA busy_timeout=5000');
},
onCreate: (db, _) async {
await db.execute(
'CREATE TABLE playlists(id TEXT PRIMARY KEY,name TEXT NOT NULL,cover_image_path TEXT,created_at TEXT NOT NULL,updated_at TEXT NOT NULL)',
);
await db.execute(
'CREATE TABLE playlist_tracks(playlist_id TEXT NOT NULL,track_key TEXT NOT NULL,track_json TEXT NOT NULL,added_at TEXT NOT NULL,PRIMARY KEY(playlist_id,track_key),FOREIGN KEY(playlist_id) REFERENCES playlists(id) ON DELETE CASCADE)',
);
await db.execute(
'CREATE INDEX idx_playlist_tracks_playlist_id ON playlist_tracks(playlist_id)',
);
for (final id in ['p', 'other']) {
await db.insert('playlists', {
'id': id,
'name': id,
'created_at': 'old',
'updated_at': 'old',
});
}
},
);
});
tearDown(() async {
await db.close();
await PlatformBridge.cleanupExtensions();
await root.delete(recursive: true);
});
tearDownAll(() {
report['completed_cases'] = completed;
binding.reportData = {'playlist_batch_write': report};
});
Future<void> publish(PreparedPlaylistBatch batch, {String id = 'p'}) async {
if (batch.rowsPath == null) {
await LibraryCollectionsDatabase.writeDatabasePlaylistTracks(
db,
playlistId: id,
playlistUpdatedAt: stamp,
tracks: batch.rows,
);
} else {
await publishPlaylistBatch(
database: db,
playlistId: id,
updatedAt: stamp,
rowsPath: batch.rowsPath!,
expectedCount: batch.entries.length,
temporaryDirectory: root,
);
}
}
// Preserve the pre-optimization single large sqflite batch as the baseline.
Future<void> originalWrite(List<Map<String, String>> rows) =>
db.transaction((txn) async {
final batch = txn.batch();
for (final row in rows) {
batch.insert('playlist_tracks', {
'playlist_id': 'p',
...row,
}, conflictAlgorithm: ConflictAlgorithm.replace);
}
batch.update(
'playlists',
{'updated_at': stamp},
where: 'id=?',
whereArgs: ['p'],
);
await batch.commit(noResult: true);
});
for (final count in [20, 256, 1024, 2048, 12000]) {
testWidgets('playlist publication parity and paired timing ($count)', (
tester,
) async {
final unique = await playlistBatchFixture(count);
final input = [...unique, unique.first, unique.last];
final expected = inlinePlaylistBatch(input, const {}, now);
final expectedDigest = await _digest(expected.rows);
final baseline = <Map<String, int>>[];
final optimized = <Map<String, int>>[];
final playlist = UserPlaylistCollection(
id: 'p',
name: 'p',
createdAt: now,
updatedAt: now,
tracks: const [],
);
for (var sample = 0; sample < 5; sample++) {
for (final candidate in sample.isEven ? [false, true] : [true, false]) {
await db.delete('playlist_tracks');
await db.update('playlists', {'updated_at': 'old'});
final result = await measureCollectionOperation(() async {
late List<CollectionTrackEntry> additions;
if (candidate) {
final batch = await preparePlaylistBatch(
tracks: input,
existingKeys: playlist.trackKeys,
addedAt: now,
temporaryDirectory: root,
);
try {
expect(batch.duplicates, 2);
expect(batch.rowsPath != null, count >= 2048);
await publish(batch);
additions = batch.entries;
} finally {
await batch.dispose();
}
} else {
final batch = inlinePlaylistBatch(input, playlist.trackKeys, now);
await originalWrite(batch.rows);
additions = batch.entries;
}
// Include the same model publication in both timing paths.
return playlist.copyWith(
tracks: [...playlist.tracks, ...additions],
updatedAt: now,
);
});
expect(result.value.trackKeys.length, count);
expect(
result.value.tracks.map((entry) => entry.key),
expected.entries.map((entry) => entry.key),
);
final persisted = await db.rawQuery(
'SELECT track_key,track_json,added_at FROM playlist_tracks ORDER BY added_at ASC,rowid ASC',
);
expect(await _digest(persisted), expectedDigest);
expect(await db.rawQuery('PRAGMA foreign_key_check'), isEmpty);
expect((await db.rawQuery('PRAGMA database_list')).length, 1);
expect(
await db.rawQuery(
'SELECT id,updated_at FROM playlists ORDER BY id',
),
[
{'id': 'other', 'updated_at': 'old'},
{'id': 'p', 'updated_at': stamp},
],
);
expect(
await root
.list()
.where(
(entry) =>
entry.path.contains('playlist-input-') ||
entry.path.contains('playlist-stage-'),
)
.isEmpty,
isTrue,
);
(candidate ? optimized : baseline).add(result.timing);
}
}
report['$count'] = {'baseline': baseline, 'optimized': optimized};
completed++;
binding.reportData = {'playlist_batch_write': report};
});
}
for (final count in [300, 2100]) {
testWidgets('rollback and recovery through actual SQLite ($count)', (
tester,
) async {
final batch = await preparePlaylistBatch(
tracks: await playlistBatchFixture(count),
existingKeys: const {},
addedAt: now,
temporaryDirectory: root,
);
try {
await db.execute(
"CREATE TRIGGER reject_update BEFORE UPDATE ON playlists BEGIN SELECT RAISE(ABORT,'fixture rollback'); END",
);
await expectLater(publish(batch), throwsA(isA<DatabaseException>()));
expect(await db.rawQuery('SELECT * FROM playlist_tracks'), isEmpty);
expect((await db.rawQuery('PRAGMA database_list')).length, 1);
await db.execute('DROP TRIGGER reject_update');
await expectLater(
publish(batch, id: 'missing'),
throwsA(isA<DatabaseException>()),
);
expect(await db.rawQuery('SELECT * FROM playlist_tracks'), isEmpty);
await publish(batch);
expect(
Sqflite.firstIntValue(
await db.rawQuery('SELECT COUNT(*) FROM playlist_tracks'),
),
count,
);
expect(await db.rawQuery('PRAGMA integrity_check'), [
{'integrity_check': 'ok'},
]);
completed++;
} finally {
await batch.dispose();
}
});
}
}
+39 -50
View File
@@ -9,6 +9,7 @@ import 'package:spotiflac_android/models/track.dart';
import 'package:spotiflac_android/services/ffmpeg_service.dart';
import 'package:spotiflac_android/services/library_collections_database.dart';
import 'package:spotiflac_android/services/library_collections_hydration.dart';
import 'package:spotiflac_android/services/playlist_batch_writer.dart';
import 'package:spotiflac_android/utils/logger.dart';
export 'package:spotiflac_android/models/library_collections.dart';
@@ -421,60 +422,48 @@ class LibraryCollectionsNotifier extends Notifier<LibraryCollectionsState> {
}
final now = DateTime.now();
final knownKeys = <String>{...playlist.trackKeys};
final entriesToAdd = <CollectionTrackEntry>[];
var alreadyInPlaylistCount = 0;
for (final track in tracks) {
final key = trackCollectionKey(track);
if (!knownKeys.add(key)) {
alreadyInPlaylistCount++;
continue;
final prepared = await preparePlaylistBatch(
tracks: tracks,
existingKeys: playlist.trackKeys,
addedAt: now,
);
try {
if (!ref.mounted || prepared.entries.isEmpty) {
return PlaylistAddBatchResult(
addedCount: 0,
alreadyInPlaylistCount: prepared.duplicates,
);
}
entriesToAdd.add(
CollectionTrackEntry(key: key, track: track, addedAt: now),
);
}
if (entriesToAdd.isEmpty) {
final rowsPath = prepared.rowsPath;
if (rowsPath != null) {
await _db.upsertPlaylistTracksFile(
playlistId: playlistId,
playlistUpdatedAt: now.toIso8601String(),
rowsPath: rowsPath,
expectedCount: prepared.entries.length,
);
} else {
await _db.upsertPlaylistTracksBatch(
playlistId: playlistId,
playlistUpdatedAt: now.toIso8601String(),
tracks: prepared.rows,
);
}
final changed = _replacePlaylistById(playlistId, (current) {
return current.copyWith(
// Append in playlist order, matching the ASC snapshot ordering.
tracks: [...current.tracks, ...prepared.entries],
updatedAt: now,
);
});
if (changed) _invalidatePlaylistPickerSummaries();
return PlaylistAddBatchResult(
addedCount: 0,
alreadyInPlaylistCount: alreadyInPlaylistCount,
addedCount: changed ? prepared.entries.length : 0,
alreadyInPlaylistCount: prepared.duplicates,
);
} finally {
await prepared.dispose();
}
await _db.upsertPlaylistTracksBatch(
playlistId: playlistId,
playlistUpdatedAt: now.toIso8601String(),
tracks: entriesToAdd
.map(
(entry) => <String, String>{
'track_key': entry.key,
'track_json': jsonEncode(entry.track.toJson()),
'added_at': entry.addedAt.toIso8601String(),
},
)
.toList(growable: false),
);
final changed = _replacePlaylistById(playlistId, (current) {
return current.copyWith(
// Append in playlist order, matching the ASC snapshot ordering.
tracks: [...current.tracks, ...entriesToAdd],
updatedAt: now,
);
});
if (!changed) {
return PlaylistAddBatchResult(
addedCount: 0,
alreadyInPlaylistCount: alreadyInPlaylistCount,
);
}
_invalidatePlaylistPickerSummaries();
return PlaylistAddBatchResult(
addedCount: entriesToAdd.length,
alreadyInPlaylistCount: alreadyInPlaylistCount,
);
});
Future<void> removeTrackFromPlaylist(String playlistId, String trackKey) =>
+37 -3
View File
@@ -8,6 +8,7 @@ import 'package:sqflite/sqflite.dart';
import 'package:spotiflac_android/models/library_collections.dart';
import 'package:spotiflac_android/services/collection_restore_codec.dart';
import 'package:spotiflac_android/services/platform_bridge.dart';
import 'package:spotiflac_android/services/playlist_batch_writer.dart';
import 'package:spotiflac_android/services/sqlite_helpers.dart' as sqlite;
import 'package:spotiflac_android/services/sqlite_native_snapshot.dart';
import 'package:spotiflac_android/utils/logger.dart';
@@ -725,9 +726,24 @@ class LibraryCollectionsDatabase {
required List<Map<String, String>> tracks,
}) async {
if (tracks.isEmpty) return;
final db = await database;
await db.transaction((txn) async {
final batch = txn.batch();
await writeDatabasePlaylistTracks(
await database,
playlistId: playlistId,
playlistUpdatedAt: playlistUpdatedAt,
tracks: tracks,
);
}
static Future<void> writeDatabasePlaylistTracks(
Database db, {
required String playlistId,
required String playlistUpdatedAt,
required List<Map<String, String>> tracks,
}) async {
if (tracks.isEmpty) return;
await sqlite.transactionWithBusyRetry(db, (txn) async {
var batch = txn.batch();
var pending = 0;
for (final track in tracks) {
batch.insert(_tablePlaylistTracks, {
'playlist_id': playlistId,
@@ -735,6 +751,11 @@ class LibraryCollectionsDatabase {
'track_json': track['track_json'],
'added_at': track['added_at'],
}, conflictAlgorithm: ConflictAlgorithm.replace);
if (++pending == 128) {
await batch.commit(noResult: true);
batch = txn.batch();
pending = 0;
}
}
batch.update(
_tablePlaylists,
@@ -746,6 +767,19 @@ class LibraryCollectionsDatabase {
});
}
Future<void> upsertPlaylistTracksFile({
required String playlistId,
required String playlistUpdatedAt,
required String rowsPath,
required int expectedCount,
}) async => publishPlaylistBatch(
database: await database,
playlistId: playlistId,
updatedAt: playlistUpdatedAt,
rowsPath: rowsPath,
expectedCount: expectedCount,
);
Future<void> deletePlaylistTrack({
required String playlistId,
required String trackKey,
+201
View File
@@ -0,0 +1,201 @@
import 'dart:convert';
import 'dart:io';
import 'dart:isolate';
import 'package:path_provider/path_provider.dart';
import 'package:sqflite/sqflite.dart';
import 'package:spotiflac_android/models/library_collections.dart';
import 'package:spotiflac_android/models/track.dart';
import 'package:spotiflac_android/services/platform_bridge.dart';
import 'package:spotiflac_android/services/sqlite_helpers.dart' as sqlite;
import 'package:spotiflac_android/services/sqlite_native_snapshot.dart';
import 'package:spotiflac_android/utils/logger.dart';
// Small selections cost less inline; reserve worker setup for larger lists.
const _workerThreshold = 1024;
const _nativeThreshold = 2048;
var _attachmentSequence = 0;
final _log = AppLogger('PlaylistBatch');
class PreparedPlaylistBatch {
const PreparedPlaylistBatch({
required this.entries,
required this.duplicates,
required this.rows,
this.rowsPath,
this.inputDirectory,
});
final List<CollectionTrackEntry> entries;
final int duplicates;
final List<Map<String, String>> rows;
final String? rowsPath;
final String? inputDirectory;
Future<void> dispose() async {
final directory = inputDirectory;
if (directory != null) await _removeDirectory(directory);
}
}
/// Deduplication and encoding share one worker. Large native batches return a
/// file path instead of transferring thousands of JSON maps back through UI.
Future<PreparedPlaylistBatch> preparePlaylistBatch({
required Iterable<Track> tracks,
required Set<String> existingKeys,
required DateTime addedAt,
bool? useNative,
Directory? temporaryDirectory,
}) async {
final snapshot = tracks.toList(growable: false);
String? inputDirectory;
if (snapshot.length >= _nativeThreshold &&
(useNative ?? PlatformBridge.supportsCoreBackend)) {
inputDirectory =
(await (temporaryDirectory ?? await getTemporaryDirectory()).createTemp(
'playlist-input-',
)).path;
}
try {
if (snapshot.length < _workerThreshold) {
return _prepare(snapshot, existingKeys, addedAt, inputDirectory);
}
return await Isolate.run(
() => _prepare(snapshot, existingKeys, addedAt, inputDirectory),
);
} catch (_) {
if (inputDirectory != null) await _removeDirectory(inputDirectory);
rethrow;
}
}
PreparedPlaylistBatch _prepare(
List<Track> tracks,
Set<String> existingKeys,
DateTime addedAt,
String? inputDirectory,
) {
final keys = <String>{};
final entries = <CollectionTrackEntry>[];
var duplicates = 0;
for (final track in tracks) {
final key = trackCollectionKey(track);
if (existingKeys.contains(key) || !keys.add(key)) {
duplicates++;
} else {
entries.add(
CollectionTrackEntry(key: key, track: track, addedAt: addedAt),
);
}
}
final rows = <Map<String, String>>[];
String? rowsPath;
if (inputDirectory != null && entries.length >= _nativeThreshold) {
rowsPath = '$inputDirectory/rows.ndjson';
final file = File(rowsPath).openSync(mode: FileMode.write);
try {
var buffer = StringBuffer();
for (final entry in entries) {
buffer.writeln(
jsonEncode({'key': entry.key, 'track': entry.track.toJson()}),
);
if (buffer.length >= 64 * 1024) {
file.writeStringSync(buffer.toString());
buffer = StringBuffer();
}
}
if (buffer.isNotEmpty) file.writeStringSync(buffer.toString());
} finally {
file.closeSync();
}
} else {
final timestamp = addedAt.toIso8601String();
for (final entry in entries) {
rows.add({
'track_key': entry.key,
'track_json': jsonEncode(entry.track.toJson()),
'added_at': timestamp,
});
}
}
return PreparedPlaylistBatch(
entries: entries,
duplicates: duplicates,
rows: rows,
rowsPath: rowsPath,
inputDirectory: inputDirectory,
);
}
/// Rust exclusively builds a new private database. Publication and the
/// playlist timestamp stay in one transaction on the existing sqflite owner.
Future<void> publishPlaylistBatch({
required Database database,
required String playlistId,
required String updatedAt,
required String rowsPath,
required int expectedCount,
Future<Map<String, dynamic>> Function(Map<String, dynamic>)? nativeJobRunner,
Directory? temporaryDirectory,
}) async {
final directory = await (temporaryDirectory ?? await getTemporaryDirectory())
.createTemp('playlist-stage-');
final stagedPath = '${directory.path}/playlist_additions.db';
var detached = true;
try {
final result = await (nativeJobRunner ?? PlatformBridge.runNativeDataJob)({
'operation': 'collections_stage_playlist_additions',
'rows_path': rowsPath,
'output_path': stagedPath,
'expected_count': expectedCount,
});
if (result['committed'] != true ||
result['published'] != true ||
result['count'] != expectedCount ||
result['path'] != stagedPath) {
throw const FormatException('Incomplete playlist staging');
}
final alias = 'playlist_additions_${_attachmentSequence++}';
await database.execute('ATTACH DATABASE ? AS $alias', [stagedPath]);
detached = false;
var committed = false;
try {
await sqlite.transactionWithBusyRetry(database, (txn) async {
await txn.execute(
'INSERT OR REPLACE INTO main.playlist_tracks '
'(playlist_id,track_key,track_json,added_at) '
'SELECT ?,track_key,track_json,? FROM $alias.playlist_additions '
'ORDER BY position',
[playlistId, updatedAt],
);
await txn.update(
'playlists',
{'updated_at': updatedAt},
where: 'id = ?',
whereArgs: [playlistId],
);
});
committed = true;
} finally {
try {
await detachNativeSqliteSnapshot(database, alias, stagedPath);
detached = true;
} on NativeSqliteSnapshotDetachException {
if (!committed) rethrow;
_log.w('Playlist saved; retaining staging after failed DETACH');
}
}
} finally {
// This output directory is separate from the input batch so its owner can
// always remove the input even if a failed DETACH retains the SQLite file.
if (detached) await _removeDirectory(directory.path);
}
}
Future<void> _removeDirectory(String path) async {
try {
await Directory(path).delete(recursive: true);
} on FileSystemException catch (error) {
_log.w('Could not remove playlist staging: $error');
}
}
@@ -126,6 +126,9 @@ fn execute(
check,
),
"collections_snapshot" => crate::native_collections::execute(request, check),
"collections_stage_playlist_additions" => {
crate::native_playlist_write::execute(request, check)
}
operation if operation.starts_with("library_") => {
crate::native_library::execute(request, check)
}
+1
View File
@@ -18,6 +18,7 @@ mod native_id3;
mod native_library;
mod native_listing;
mod native_lyrics;
mod native_playlist_write;
mod native_playlists;
mod native_xml;
mod network_tags;
@@ -0,0 +1,210 @@
//! Build an isolated playlist write batch. Only the live database's owner may
//! attach this closed file and publish its rows into the application's tables.
use rusqlite::Connection;
use serde::Deserialize;
use serde_json::{Value, json, value::RawValue};
use std::{
borrow::Cow,
fs::File,
io::{BufRead, BufReader},
path::Path,
};
#[derive(Deserialize)]
struct TrackRow<'a> {
#[serde(borrow)]
key: Cow<'a, str>,
#[serde(borrow)]
track: &'a RawValue,
}
pub(crate) fn execute(
request: &Value,
check: &dyn Fn() -> Result<(), String>,
) -> Result<Value, String> {
check()?;
let input = request["rows_path"]
.as_str()
.ok_or("Missing playlist rows_path")?;
let output = request["output_path"]
.as_str()
.ok_or("Missing playlist output_path")?;
let expected = request["expected_count"]
.as_u64()
.ok_or("Invalid playlist expected_count")?;
let output = Path::new(output);
if !Path::new(input).is_absolute()
|| !output.is_absolute()
|| output.file_name().and_then(|name| name.to_str()) != Some("playlist_additions.db")
{
return Err(
"Playlist staging requires absolute rows_path and private playlist_additions.db".into(),
);
}
let mut input =
BufReader::with_capacity(64 * 1024, File::open(input).map_err(|e| e.to_string())?);
let staged = tempfile::NamedTempFile::new_in(
output
.parent()
.ok_or("Missing playlist staging directory")?,
)
.map_err(|e| e.to_string())?;
// Never open output_path itself: it may already exist. The temporary file
// is exclusively owned by this job and is published without replacement.
let mut db = Connection::open(staged.path()).map_err(|e| e.to_string())?;
db.execute_batch(
"PRAGMA user_version=1;
CREATE TABLE playlist_additions(
position INTEGER PRIMARY KEY,
track_key TEXT NOT NULL UNIQUE,
track_json TEXT NOT NULL
);",
)
.map_err(|e| e.to_string())?;
let transaction = db.transaction().map_err(|e| e.to_string())?;
let mut count = 0_u64;
{
let mut insert = transaction
.prepare("INSERT INTO playlist_additions(track_key,track_json) VALUES (?,?)")
.map_err(|e| e.to_string())?;
let mut line = Vec::new();
loop {
check()?;
line.clear();
if input
.read_until(b'\n', &mut line)
.map_err(|e| e.to_string())?
== 0
{
break;
}
let row: TrackRow<'_> = serde_json::from_slice(&line).map_err(|e| e.to_string())?;
if row.key.is_empty() || !row.track.get().starts_with('{') {
return Err("Invalid playlist track row".into());
}
if count == expected {
return Err("Playlist staging received too many rows".into());
}
// Preserve the original track JSON byte-for-byte instead of
// allocating a metadata map and serializing every field again.
insert
.execute([row.key.as_ref(), row.track.get()])
.map_err(|e| e.to_string())?;
count += 1;
}
}
if count != expected {
return Err(format!(
"Incomplete playlist staging: expected {expected}, got {count}"
));
}
check()?;
transaction.commit().map_err(|e| e.to_string())?;
db.close().map_err(|(_, error)| error.to_string())?;
staged.as_file().sync_all().map_err(|e| e.to_string())?;
check()?;
staged
.persist_noclobber(output)
.map_err(|e| e.to_string())?;
Ok(json!({"path":output,"count":count,"committed":true,"published":true}))
}
#[cfg(test)]
mod tests {
use super::*;
use std::cell::Cell;
#[test]
fn stages_ordered_unicode_rows_and_preserves_raw_track_json() {
let root = tempfile::tempdir().unwrap();
let input = root.path().join("rows.ndjson");
let output = root.path().join("playlist_additions.db");
let raw = r#"{ "name": "音楽 🎵", "unmodeled": {"value":1e-12}, "id":"one" }"#;
std::fs::write(&input, format!("{{\"key\":\"source:one\\n\\\"\",\"track\":{raw}}}\n{{\"key\":\"source:two\",\"track\":{{\"id\":\"two\"}}}}\n")).unwrap();
let result = execute(
&json!({"rows_path":input,"output_path":output,"expected_count":2}),
&|| Ok(()),
)
.unwrap();
assert_eq!(result["count"], 2);
let db = Connection::open(&output).unwrap();
let rows = db
.prepare("SELECT track_key,track_json FROM playlist_additions ORDER BY position")
.unwrap()
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
rows,
[
("source:one\n\"".into(), raw.into()),
("source:two".into(), r#"{"id":"two"}"#.into())
]
);
assert_eq!(
db.pragma_query_value(None, "user_version", |row| row.get::<_, i64>(0))
.unwrap(),
1
);
}
#[test]
fn rejects_invalid_duplicate_truncated_and_cancelled_batches_without_publication() {
let valid = "{\"key\":\"one\",\"track\":{\"id\":\"one\"}}\n";
for (payload, expected, cancel) in [
(valid.to_owned(), 2, false),
(valid.repeat(2), 1, false),
(valid.repeat(2), 2, false),
(format!("{valid}{{bad"), 2, false),
("{\"key\":\"one\",\"track\":null}\n".into(), 1, false),
("{\"key\":\"\",\"track\":{}}\n".into(), 1, false),
(valid.to_owned(), 1, true),
] {
let root = tempfile::tempdir().unwrap();
let input = root.path().join("rows.ndjson");
let output = root.path().join("playlist_additions.db");
std::fs::write(&input, payload).unwrap();
let checks = Cell::new(0);
let result = execute(
&json!({"rows_path":input,"output_path":output,"expected_count":expected}),
&|| {
checks.set(checks.get() + 1);
if cancel && checks.get() == 3 {
Err("cancelled".into())
} else {
Ok(())
}
},
);
assert!(result.is_err());
assert!(!output.exists());
assert_eq!(std::fs::read_dir(root.path()).unwrap().count(), 1);
}
}
#[test]
fn never_replaces_an_existing_destination_or_opens_the_live_database() {
for filename in ["playlist_additions.db", "library_collections.db"] {
let root = tempfile::tempdir().unwrap();
let input = root.path().join("rows.ndjson");
let output = root.path().join(filename);
std::fs::write(&input, "").unwrap();
std::fs::write(&output, "existing database").unwrap();
assert!(
execute(
&json!({"rows_path":input,"output_path":output,"expected_count":0}),
&|| Ok(())
)
.is_err()
);
assert_eq!(
std::fs::read_to_string(output).unwrap(),
"existing database"
);
assert_eq!(std::fs::read_dir(root.path()).unwrap().count(), 2);
}
}
}
@@ -163,6 +163,20 @@ class _CollectionsDatabase implements LibraryCollectionsDatabase {
Future<void> deletePlaylist(String playlistId) =>
_write('delete:playlist:$playlistId', () => playlists.remove(playlistId));
@override
Future<void> upsertPlaylistTracksBatch({
required String playlistId,
required String playlistUpdatedAt,
required List<Map<String, String>> tracks,
}) => _write('batch:$playlistId', () {
final rows = playlistTracks.putIfAbsent(playlistId, () => []);
for (final track in tracks) {
rows.removeWhere((row) => row['track_key'] == track['track_key']);
rows.add({...track, 'playlist_id': playlistId});
}
playlists[playlistId]!['updated_at'] = playlistUpdatedAt;
});
@override
dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
}
@@ -189,6 +203,64 @@ ProviderContainer _container(_CollectionsDatabase db) {
}
void main() {
for (final failFirst in [false, true]) {
test(
'bulk additions publish after persistence and serialize duplicates (failure: $failFirst)',
() async {
final db = _CollectionsDatabase();
db.playlists['playlist'] = {
'id': 'playlist',
'name': 'Playlist',
'created_at': '2026-01-01T00:00:00Z',
'updated_at': '2026-01-01T00:00:00Z',
};
db.releasePlaylistRead.complete();
db.failNextWrite = failFirst;
final container = _container(db);
final notifier = container.read(libraryCollectionsProvider.notifier);
final first = notifier.addTracksToPlaylist('playlist', [
_track('1'),
_track('1'),
_track('2'),
]);
final failure = failFirst ? expectLater(first, throwsStateError) : null;
await db.writeStarted.future;
expect(
container
.read(libraryCollectionsProvider)
.playlistById('playlist')!
.trackCount,
0,
);
final second = notifier.addTracksToPlaylist('playlist', [
_track('2'),
_track('3'),
]);
expect(db.events, ['batch:playlist']);
db.releaseWrite.complete();
if (failure != null) {
await failure;
} else {
final result = await first;
expect(result.addedCount, 2);
expect(result.alreadyInPlaylistCount, 1);
}
final result = await second;
expect(result.addedCount, failFirst ? 2 : 1);
expect(result.alreadyInPlaylistCount, failFirst ? 0 : 1);
final playlist = container
.read(libraryCollectionsProvider)
.playlistById('playlist')!;
expect(
playlist.tracks.map((entry) => entry.track.id),
failFirst ? ['2', '3'] : ['1', '2', '3'],
);
expect(playlist.trackCount, failFirst ? 2 : 3);
expect(db.events, ['batch:playlist', 'batch:playlist']);
},
);
}
test('disposed snapshot read removes native rows before returning', () async {
final directory = await Directory.systemTemp.createTemp(
'collection-read-disposed-',
+59
View File
@@ -0,0 +1,59 @@
import 'dart:convert';
import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
import 'package:spotiflac_android/services/playlist_batch_writer.dart';
import 'support/library_collections_benchmark.dart';
import 'support/playlist_batch_benchmark.dart';
void main() {
test(
'compare playlist batch preparation with the original inline path',
() async {
final tracks = await playlistBatchFixture(12000);
final root = await Directory.systemTemp.createTemp('playlist-benchmark-');
addTearDown(() => root.delete(recursive: true));
final samples = <Map<String, int>>[];
final workers = <Map<String, int>>[];
for (var index = 0; index < 5; index++) {
for (final worker in index.isEven ? [false, true] : [true, false]) {
if (worker) {
final result = await measureCollectionOperation(
() => preparePlaylistBatch(
tracks: tracks,
existingKeys: const {},
addedAt: DateTime.utc(2026, 10, 8),
useNative: true,
temporaryDirectory: root,
),
);
expect(result.value.entries.length, tracks.length);
workers.add(result.timing);
await result.value.dispose();
continue;
}
final result = await measureCollectionOperation(
() => inlinePlaylistBatch(
tracks,
const {},
DateTime.utc(2026, 10, 8),
),
);
expect(result.value.rows.length, tracks.length);
samples.add(result.timing);
}
}
// Host diagnostics only; the device profile benchmark owns speed claims.
// ignore: avoid_print
print(
jsonEncode({
'rows': tracks.length,
'host_inline': samples,
'host_worker': workers,
}),
);
},
skip: !const bool.fromEnvironment('RUN_PLAYLIST_BATCH_BENCHMARK'),
);
}
+210
View File
@@ -0,0 +1,210 @@
import 'dart:convert';
import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
import 'package:sqflite/sqflite.dart';
import 'package:spotiflac_android/services/playlist_batch_writer.dart';
import 'support/sqlite_process_database.dart';
class _DetachFailure implements Database {
_DetachFailure(this.database);
@override
final SqliteProcessDatabase database;
@override
Future<void> execute(String sql, [List<Object?>? args]) async {
if (sql.startsWith('DETACH DATABASE')) throw StateError('detach failed');
await database.execute(sql, args);
}
@override
Future<T> transaction<T>(
Future<T> Function(Transaction) action, {
bool? exclusive,
}) => database.transaction(action, exclusive: exclusive);
@override
dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
}
void main() {
late Directory root;
late SqliteProcessDatabase live;
late String input;
String? retainedStage;
setUp(() async {
root = await Directory.systemTemp.createTemp('playlist-publication-');
live = await SqliteProcessDatabase.open(path: '${root.path}/live.db');
await live.execute('PRAGMA foreign_keys=ON');
await live.execute(
'CREATE TABLE playlists(id TEXT PRIMARY KEY, name TEXT, updated_at TEXT)',
);
await live.execute(
'CREATE TABLE playlist_tracks(playlist_id TEXT NOT NULL, track_key TEXT NOT NULL, track_json TEXT NOT NULL, added_at TEXT NOT NULL, PRIMARY KEY(playlist_id,track_key), FOREIGN KEY(playlist_id) REFERENCES playlists(id))',
);
await live.execute('CREATE TABLE published(track_key TEXT)');
await live.execute(
'CREATE TRIGGER audit AFTER INSERT ON playlist_tracks BEGIN INSERT INTO published VALUES(new.track_key); END',
);
await live.execute("INSERT INTO playlists VALUES('p','Original','old')");
await live.execute("INSERT INTO playlists VALUES('other','Other','old')");
await live.execute(
"INSERT INTO playlist_tracks VALUES('p','existing','{}','old')",
);
await live.execute('DELETE FROM published');
input = '${root.path}/rows.ndjson';
await File(input).writeAsString('');
retainedStage = null;
});
tearDown(() async {
await live.close();
await root.delete(recursive: true);
});
// This fixture implements only the native boundary. SQL publication below
// runs on real SQLite; the native parser itself has Rust and device tests.
Future<Map<String, dynamic>> stage(Map<String, dynamic> request) async {
expect(request['operation'], 'collections_stage_playlist_additions');
expect(request.values, isNot(contains(live.path)));
expect((await live.rawQuery('PRAGMA database_list')).length, 1);
final path = request['output_path'] as String;
retainedStage = path;
final private = await SqliteProcessDatabase.open(path: path);
try {
await private.execute(
'CREATE TABLE playlist_additions(position INTEGER PRIMARY KEY,track_key TEXT UNIQUE,track_json TEXT)',
);
for (final key in ['new-a', 'new-b']) {
await private.execute(
'INSERT INTO playlist_additions(track_key,track_json) VALUES(?,?)',
[
key,
jsonEncode({'id': key, 'name': '音楽 🎵'}),
],
);
}
} finally {
await private.close();
}
return {'committed': true, 'published': true, 'path': path, 'count': 2};
}
Future<void> publish({
Database? database,
String playlist = 'p',
Future<Map<String, dynamic>> Function(Map<String, dynamic>)? runner,
}) => publishPlaylistBatch(
database: database ?? live,
playlistId: playlist,
updatedAt: 'new',
rowsPath: input,
expectedCount: 2,
temporaryDirectory: root,
nativeJobRunner: runner ?? stage,
);
test(
'publishes in order through live triggers and changes only target timestamp',
() async {
await publish();
expect(
(await live.rawQuery(
'SELECT track_key FROM playlist_tracks ORDER BY rowid',
)).map((row) => row['track_key']),
['existing', 'new-a', 'new-b'],
);
expect(await live.rawQuery('SELECT * FROM published'), [
{'track_key': 'new-a'},
{'track_key': 'new-b'},
]);
expect(
await live.rawQuery(
'SELECT id,name,updated_at FROM playlists ORDER BY rowid',
),
[
{'id': 'p', 'name': 'Original', 'updated_at': 'new'},
{'id': 'other', 'name': 'Other', 'updated_at': 'old'},
],
);
expect(await live.rawQuery('PRAGMA foreign_key_check'), isEmpty);
expect((await live.rawQuery('PRAGMA database_list')).length, 1);
expect(await File(retainedStage!).exists(), false);
},
);
test(
'timestamp failure rolls back all inserted tracks and trigger effects',
() async {
await live.execute(
"CREATE TRIGGER fail_update BEFORE UPDATE ON playlists BEGIN SELECT RAISE(ABORT,'denied'); END",
);
await expectLater(publish(), throwsA(isA<StateError>()));
expect(
(await live.rawQuery(
'SELECT track_key FROM playlist_tracks',
)).single['track_key'],
'existing',
);
expect(await live.rawQuery('SELECT * FROM published'), isEmpty);
expect(
(await live.rawQuery(
"SELECT updated_at FROM playlists WHERE id='p'",
)).single['updated_at'],
'old',
);
expect((await live.rawQuery('PRAGMA database_list')).length, 1);
expect(await File(retainedStage!).exists(), false);
},
);
test(
'missing playlist preserves foreign-key enforcement and existing data',
() async {
await expectLater(
publish(playlist: 'missing'),
throwsA(isA<StateError>()),
);
expect(
(await live.rawQuery('SELECT track_key FROM playlist_tracks')).length,
1,
);
expect((await live.rawQuery('PRAGMA database_list')).length, 1);
},
);
test('incomplete native result cannot enter the live transaction', () async {
await expectLater(
publish(
runner: (request) async {
final result = await stage(request);
result['count'] = 1;
return result;
},
),
throwsFormatException,
);
expect(
(await live.rawQuery('SELECT track_key FROM playlist_tracks')).length,
1,
);
expect(await File(retainedStage!).exists(), false);
expect((await live.rawQuery('PRAGMA database_list')).length, 1);
});
test(
'failed detach retains its file without reporting a durable commit as failure',
() async {
await publish(database: _DetachFailure(live));
expect(
(await live.rawQuery('SELECT track_key FROM playlist_tracks')).length,
3,
);
expect(await File(retainedStage!).exists(), true);
expect((await live.rawQuery('PRAGMA database_list')).length, 2);
await File(input).delete();
expect(await File(retainedStage!).exists(), true);
},
);
}
+132
View File
@@ -0,0 +1,132 @@
import 'dart:convert';
import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
import 'package:spotiflac_android/models/track.dart';
import 'package:spotiflac_android/services/playlist_batch_writer.dart';
import 'support/playlist_batch_benchmark.dart';
class _InvalidTrack extends Track {
_InvalidTrack()
: super(
id: 'invalid',
name: 'Invalid',
artistName: '',
albumName: '',
duration: 0,
);
@override
Map<String, dynamic> toJson() => {'value': double.nan};
}
void main() {
late Directory root;
setUp(() async {
root = await Directory.systemTemp.createTemp('playlist-batch-test-');
});
tearDown(() => root.delete(recursive: true));
final now = DateTime.utc(2026, 10, 8);
for (final count in [12, 1024, 2500]) {
test(
'preparation preserves order, metadata and duplicate counts ($count)',
() async {
final fixture = await playlistBatchFixture(count);
final tracks = [...fixture, fixture[1], fixture[3]];
final keys = {'example:0', 'example:1'};
final baseline = inlinePlaylistBatch(tracks, keys, now);
final prepared = await preparePlaylistBatch(
tracks: tracks,
existingKeys: keys,
addedAt: now,
useNative: true,
temporaryDirectory: root,
);
try {
expect(
prepared.entries.map((entry) => entry.key),
baseline.entries.map((entry) => entry.key),
);
expect(prepared.duplicates, baseline.duplicates);
expect(prepared.entries.every((entry) => entry.addedAt == now), true);
expect(keys, {'example:0', 'example:1'});
if (prepared.rowsPath == null) {
expect(prepared.rows, baseline.rows);
} else {
expect(prepared.rows, isEmpty);
final rows = await File(prepared.rowsPath!).readAsLines();
expect(rows.length, baseline.rows.length);
for (var index = 0; index < rows.length; index++) {
final row = jsonDecode(rows[index]) as Map;
expect(row['key'], baseline.rows[index]['track_key']);
expect(
jsonEncode(row['track']),
baseline.rows[index]['track_json'],
);
}
}
} finally {
await prepared.dispose();
}
expect(await root.list().toList(), isEmpty);
},
);
}
test('ISRC wins across providers and first batch metadata wins', () async {
Track track(String id, String source, String isrc) => Track(
id: id,
source: source,
isrc: isrc,
name: id,
artistName: 'Artist',
albumName: 'Album',
duration: 1,
);
final prepared = await preparePlaylistBatch(
tracks: [
track('first', 'one', ' xx001 '),
track('second', 'two', 'XX001'),
track('third', 'two', 'xx002'),
],
existingKeys: {'isrc:XX002'},
addedAt: now,
);
expect(prepared.duplicates, 2);
expect(prepared.entries.single.track.id, 'first');
expect(prepared.entries.single.key, 'isrc:XX001');
});
test('mostly duplicate batches retain the small persistence route', () async {
final fixture = await playlistBatchFixture(12);
final prepared = await preparePlaylistBatch(
tracks: [for (var index = 0; index < 256; index++) ...fixture],
existingKeys: const {},
addedAt: now,
useNative: true,
temporaryDirectory: root,
);
expect(prepared.rowsPath, isNull);
expect(prepared.entries.length, 12);
expect(prepared.duplicates, 3060);
await prepared.dispose();
expect(await root.list().toList(), isEmpty);
});
test('failed encoding removes its partial row file', () async {
final fixture = await playlistBatchFixture(2500);
await expectLater(
preparePlaylistBatch(
tracks: [...fixture, _InvalidTrack()],
existingKeys: const {},
addedAt: now,
useNative: true,
temporaryDirectory: root,
),
throwsA(anything),
);
expect(await root.list().toList(), isEmpty);
});
}
@@ -0,0 +1,61 @@
import 'dart:convert';
import 'dart:isolate';
import 'package:spotiflac_android/models/library_collections.dart';
import 'package:spotiflac_android/models/track.dart';
Future<List<Track>> playlistBatchFixture(int count) => Isolate.run(
() => List.generate(
count,
(index) => Track(
id: '$index',
source: 'example',
name: 'Track $index — 音楽 🎵',
artistName: 'Artist ${index % 100}',
albumName: 'Album ${index ~/ 12}',
albumArtist: 'Album artist ${index % 100}',
duration: 180,
coverUrl: 'https://example.test/cover/${index ~/ 12}.jpg',
trackNumber: index % 12 + 1,
discNumber: 1,
genre: 'Rock',
comment: List.filled(12, 'Metadata $index').join(' '),
),
growable: false,
),
);
// Production preparation before the batch-write optimization. Keep the same
// identity precedence, duplicate counting, ordering and serialization here.
({
List<CollectionTrackEntry> entries,
List<Map<String, String>> rows,
int duplicates,
})
inlinePlaylistBatch(
List<Track> tracks,
Set<String> existingKeys,
DateTime now,
) {
final knownKeys = {...existingKeys};
final entries = <CollectionTrackEntry>[];
var duplicates = 0;
for (final track in tracks) {
final key = trackCollectionKey(track);
if (!knownKeys.add(key)) {
duplicates++;
continue;
}
entries.add(CollectionTrackEntry(key: key, track: track, addedAt: now));
}
final rows = entries
.map(
(entry) => <String, String>{
'track_key': entry.key,
'track_json': jsonEncode(entry.track.toJson()),
'added_at': entry.addedAt.toIso8601String(),
},
)
.toList(growable: false);
return (entries: entries, rows: rows, duplicates: duplicates);
}