diff --git a/lib/services/library_collections_hydration.dart b/lib/services/library_collections_hydration.dart index 0f9d69d8..97d1927b 100644 --- a/lib/services/library_collections_hydration.dart +++ b/lib/services/library_collections_hydration.dart @@ -449,10 +449,8 @@ Future _readNativeCollectionRows( var count = 0; _CollectionRows? previousKind; var batch = >[]; - try { - await for (final line in File( - input.path, - ).openRead().transform(utf8.decoder).transform(const LineSplitter())) { + final lines = const LineSplitter().startChunkedConversion( + _CollectionLineSink((line) { final decoded = jsonDecode(line) as Map; final kind = switch (decoded['kind']) { 'wishlist' || 'playlist_track' => _CollectionRows.wishlist, @@ -466,15 +464,26 @@ Future _readNativeCollectionRows( }; if (previousKind != null && (kind != previousKind || batch.length == 256)) { - accumulator.add(previousKind, batch); + accumulator.add(previousKind!, batch); batch = >[]; } previousKind = kind; batch.add(Map.from(decoded['row'] as Map)); count++; + }), + ); + try { + // Decode file chunks asynchronously, then split and consume their lines + // synchronously on this worker. A stream event per row adds avoidable + // scheduling overhead to large snapshots. + await for (final chunk in File( + input.path, + ).openRead().transform(utf8.decoder)) { + lines.add(chunk); } + lines.close(); if (previousKind != null && batch.isNotEmpty) { - accumulator.add(previousKind, batch); + accumulator.add(previousKind!, batch); } if (input.count != null && count != input.count) { throw const FormatException('Incomplete native collection snapshot'); @@ -484,6 +493,17 @@ Future _readNativeCollectionRows( } } +class _CollectionLineSink implements Sink { + const _CollectionLineSink(this.onLine); + final void Function(String) onLine; + + @override + void add(String line) => onLine(line); + + @override + void close() {} +} + enum _CollectionExportRows { wishlist, loved, artist, playlistTrack } typedef _CollectionExportBatch = ({ diff --git a/test/library_collections_hydration_test.dart b/test/library_collections_hydration_test.dart index c7cd4d5c..33d72b03 100644 --- a/test/library_collections_hydration_test.dart +++ b/test/library_collections_hydration_test.dart @@ -78,6 +78,164 @@ LibraryCollectionsSnapshot collectionFixture(int count) => ); void main() { + Future nativeBytes( + List bytes, + int count, + Directory directory, + ) async { + final file = File('${directory.path}/rows.ndjson'); + await file.writeAsBytes(bytes); + return hydrateLibraryCollectionsSnapshot( + LibraryCollectionsSnapshot( + wishlistRows: const [], + lovedRows: const [], + playlistRows: const [], + playlistTrackRows: const [], + favoriteArtistRows: const [], + nativeRowsPath: file.path, + nativeRowsDirectory: directory.path, + nativeRowCount: count, + ), + ); + } + + String nativeLine(String kind, Map row) => + jsonEncode({'kind': kind, 'row': row}); + + test( + 'native chunked reader retains interleaved group order across 256-row batches', + () async { + final directory = await Directory.systemTemp.createTemp( + 'collection-row-groups-', + ); + final wishlist = List.generate(513, trackRow); + final loved = List.generate(3, (index) => trackRow(900 + index)); + final snapshot = LibraryCollectionsSnapshot( + wishlistRows: wishlist, + lovedRows: loved, + favoriteArtistRows: const [], + playlistRows: const [], + playlistTrackRows: const [], + ); + final lines = [ + for (final row in wishlist.take(257)) nativeLine('wishlist', row), + nativeLine('loved', loved[0]), + for (final row in wishlist.skip(257)) nativeLine('wishlist', row), + for (final row in loved.skip(1)) nativeLine('loved', row), + ]; + // Include LF, CRLF and lone CR; the last row intentionally has no newline. + final text = StringBuffer(); + for (var index = 0; index < lines.length; index++) { + text.write(lines[index]); + if (index + 1 != lines.length) { + text.write(['\n', '\r\n', '\r'][index % 3]); + } + } + final actual = await nativeBytes( + utf8.encode(text.toString()), + lines.length, + directory, + ); + expect( + jsonEncode(actual.toJson()), + jsonEncode(decodeLibraryCollectionsSnapshot(snapshot).toJson()), + ); + expect( + actual.wishlist.map((entry) => entry.key), + wishlist.map((row) => row['track_key']), + ); + expect( + actual.loved.map((entry) => entry.key), + loved.map((row) => row['track_key']), + ); + expect(await directory.exists(), isFalse); + }, + ); + + test( + 'native UTF-8 and CRLF survive exact file chunk boundaries and unterminated tail', + () async { + Map commentRow(int id, String comment) => { + ...trackRow(id), + 'track_json': jsonEncode( + (jsonDecode(trackRow(id)['track_json'] as String) + as Map) + ..['comment'] = comment, + ), + }; + final initial = nativeLine('wishlist', commentRow(0, '🎵終わり')); + final emojiPrefix = utf8 + .encode(initial.substring(0, initial.indexOf('🎵'))) + .length; + final unicodeRow = commentRow(0, '${'x' * (65535 - emojiPrefix)}🎵終わり'); + final unicodeLine = nativeLine('wishlist', unicodeRow); + final unicodeBytes = utf8.encode(unicodeLine); + expect(unicodeBytes[65535], 0xf0); + expect(unicodeBytes[65536], 0x9f); + final unicodeDirectory = await Directory.systemTemp.createTemp( + 'collection-utf8-boundary-', + ); + final unicode = await nativeBytes(unicodeBytes, 1, unicodeDirectory); + expect( + unicode.wishlist.single.track.comment, + (jsonDecode(unicodeRow['track_json'] as String) as Map)['comment'], + ); + expect(await unicodeDirectory.exists(), isFalse); + + final unpadded = nativeLine('wishlist', commentRow(1, '')); + final crRow = commentRow(1, 'y' * (65535 - utf8.encode(unpadded).length)); + final crLine = nativeLine('wishlist', crRow); + final tail = trackRow(2); + final crBytes = utf8.encode('$crLine\r\n${nativeLine('wishlist', tail)}'); + expect(crBytes[65535], 13); + expect(crBytes[65536], 10); + final crDirectory = await Directory.systemTemp.createTemp( + 'collection-crlf-boundary-', + ); + final crlf = await nativeBytes(crBytes, 2, crDirectory); + expect(crlf.wishlist.map((entry) => entry.key), [ + 'example:1', + 'example:2', + ]); + expect( + crlf.wishlist.first.track.comment, + 'y' * (65535 - utf8.encode(unpadded).length), + ); + expect(await crDirectory.exists(), isFalse); + }, + ); + + test( + 'native malformed bytes/envelopes and row-count mismatch always remove staging', + () async { + final valid = nativeLine('wishlist', trackRow(1)); + final invalid = <({List bytes, int count})>[ + (bytes: [...utf8.encode('$valid\n'), 0xc3, 0x28], count: 2), + (bytes: utf8.encode('$valid\n{"kind":'), count: 2), + (bytes: utf8.encode('$valid\n\n'), count: 2), + (bytes: utf8.encode(nativeLine('unknown', {})), count: 1), + ( + bytes: utf8.encode( + jsonEncode({'kind': 'wishlist', 'row': []}), + ), + count: 1, + ), + (bytes: utf8.encode(valid), count: 0), + (bytes: utf8.encode('$valid\r\n'), count: 2), + ]; + for (final input in invalid) { + final directory = await Directory.systemTemp.createTemp( + 'collection-invalid-chunks-', + ); + await expectLater( + nativeBytes(input.bytes, input.count, directory), + throwsA(isA()), + ); + expect(await directory.exists(), isFalse); + } + }, + ); + test( 'native row file preserves snapshot parity and is always cleaned', () async { diff --git a/test/library_collections_rows_parser_benchmark_test.dart b/test/library_collections_rows_parser_benchmark_test.dart new file mode 100644 index 00000000..050b597b --- /dev/null +++ b/test/library_collections_rows_parser_benchmark_test.dart @@ -0,0 +1,130 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; + +// Parse-only experiment: both readers perform the same outer/inner JSON +// decoding and ordered checksum. It excludes SQL, native serialization, +// isolate startup/transfers, model hydration and membership construction. +class _DecodedRows implements Sink { + int count = 0; + int checksum = 0; + + @override + void add(String line) { + final envelope = jsonDecode(line) as Map; + final row = Map.from(envelope['row'] as Map); + final track = + jsonDecode(row['track_json'] as String) as Map; + final text = + '${envelope['kind']}|${row['track_key']}|${track['name']}|${track['comment']}'; + for (final unit in text.codeUnits) { + checksum = ((checksum * 31) + unit) & 0x7fffffff; + } + count++; + } + + @override + void close() {} +} + +Future<({int count, int checksum})> _parse(File file, bool chunked) async { + final rows = _DecodedRows(); + if (chunked) { + final lines = const LineSplitter().startChunkedConversion(rows); + await for (final chunk in file.openRead().transform(utf8.decoder)) { + lines.add(chunk); + } + lines.close(); + } else { + await for (final line + in file + .openRead() + .transform(utf8.decoder) + .transform(const LineSplitter())) { + rows.add(line); + } + } + return (count: rows.count, checksum: rows.checksum); +} + +void main() { + test('paired native collection rows parse-only benchmark', () async { + final directory = await Directory.systemTemp.createTemp( + 'collection-parser-benchmark-', + ); + try { + final file = File('${directory.path}/rows.ndjson'); + const count = 8000; + await file.writeAsString( + [ + for (var index = 0; index < count; index++) + jsonEncode({ + 'kind': index.isEven ? 'wishlist' : 'loved', + 'row': { + 'track_key': 'example:$index', + 'added_at': '2026-10-07T00:00:00.000Z', + 'track_json': jsonEncode({ + 'id': '$index', + 'name': 'Song $index 音楽 🎵', + 'artistName': 'Artist ${index % 100}', + 'albumName': 'Album ${index ~/ 12}', + 'duration': 180, + 'comment': '"quoted" \\ / \r\n ${'x' * 512}', + 'coverUrl': 'https://example.test/$index.jpg', + 'availability': {'tidal': true, 'qobuz': false}, + }), + }, + }), + ].join('\n'), + ); + final expected = await _parse(file, false); + expect(expected.count, count); + // Warm both codecs/file paths before alternating paired measurements. + for (var warm = 0; warm < 2; warm++) { + expect(await _parse(file, true), expected); + expect(await _parse(file, false), expected); + } + final old = []; + final chunked = []; + for (var sample = 0; sample < 7; sample++) { + for (final useChunked + in sample.isEven ? [false, true] : [true, false]) { + final watch = Stopwatch()..start(); + final actual = await _parse(file, useChunked); + watch.stop(); + expect(actual, expected); + (useChunked ? chunked : old).add(watch.elapsedMicroseconds); + } + } + int median(List samples) => + (List.of(samples)..sort())[samples.length ~/ 2]; + final report = { + 'scope': + 'parse only: file read, strict UTF-8 decode, outer/inner JSON decode, ordered checksum; excludes full collection loading/hydration', + 'platform': Platform.operatingSystem, + 'mode': 'flutter test host', + 'rows': count, + 'bytes': await file.length(), + 'checksum': expected.checksum, + 'async_line_stream_us': old, + 'chunked_line_sink_us': chunked, + 'async_line_stream_median_us': median(old), + 'chunked_line_sink_median_us': median(chunked), + 'parity': true, + }; + const output = String.fromEnvironment('COLLECTION_ROWS_PARSER_OUTPUT'); + if (output.isNotEmpty) { + final result = File(output); + await result.parent.create(recursive: true); + await result.writeAsString( + const JsonEncoder.withIndent(' ').convert(report), + ); + } + // ignore: avoid_print + print('COLLECTION_ROWS_PARSER_REPORT ${jsonEncode(report)}'); + } finally { + await directory.delete(recursive: true); + } + }, skip: !const bool.fromEnvironment('RUN_COLLECTION_ROWS_PARSER_BENCHMARK')); +}