From a30f42f64f7608a691c8a0edf32f6e96684e9bf4 Mon Sep 17 00:00:00 2001 From: MikenRafter Date: Fri, 23 Oct 2026 16:13:26 -0600 Subject: [PATCH 1/5] perf(sync): one commit for the post-ACK ledger rows; capture freshness in the batch transaction Each completed HISTORY_END wrote capture freshness with a read and an autocommit write after the batch commit, and the two post-ACK ledger rows as two more autocommits while the band waited for its next marker. - Capture freshness is written inside the rows + cursor transaction, so it rolls back with them instead of recording a capture that never committed. - The two post-ACK ledger rows share one transaction. - One `[perf] sync chunk` log line per completed chunk (records, beats, commit_ms, ack_ms, post_ms), so a slow sync can be attributed from an issue report. The per-commit synchronous=FULL bracket is unchanged; a test pins that every chunk during an offload gets its own bracket. Holding FULL for a whole offload was measured and rejected: every write during the offload then pays FULL (200-chunk drain 7.6-8.4 s against 5.2 s). Commit-before-ACK is unchanged: rows and cursor commit durably before the verbatim HISTORY_END token is echoed, and a failed commit never ACKs. --- lib/ble/ble_engine.dart | 98 +++-- lib/data/db.dart | 42 +- test/sync_chunk_overhead_test.dart | 604 +++++++++++++++++++++++++++++ 3 files changed, 695 insertions(+), 49 deletions(-) create mode 100644 test/sync_chunk_overhead_test.dart diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart index 6267d06ed..118c1fe09 100644 --- a/lib/ble/ble_engine.dart +++ b/lib/ble/ble_engine.dart @@ -7433,7 +7433,10 @@ class BleEngine { // band never trimming and re-flooding the same chunk forever. On // persistent failure bounce the link — the committed data is safe, and // the next session's re-delivery is dedup-safe (rows REPLACE by rec_ts). - if (!await _writeAckVerified(ack, session, taskGen: taskGen)) { + final ackWatch = clock.stopwatch()..start(); + final ackWritten = await _writeAckVerified(ack, session, taskGen: taskGen); + final ackMs = ackWatch.elapsedMilliseconds; + if (!ackWritten) { // False for one of three reasons: the writes genuinely exhausted, the // session died, or the TASK ended under the retries. Only the first is // this continuation's to handle — a stale one must not touch the @@ -7523,43 +7526,54 @@ class BleEngine { // over — records were banked and the band may advance. _noDurableProgress.trimAcked(); d.noteBatchAcked(); // ACKed and KEEP listening - await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry( - status: 'acknowledged', - ackedAt: DateTime.now().millisecondsSinceEpoch, - metaPatch: { - 'last_batch_token': tokenHex, - 'last_batch_id': m.batchId, - 'last_batch_records': d.records, - 'last_ack_batches': d.batches, - 'strap_history_oldest_ts': _strapHistoryOldestTs, - 'strap_history_newest_ts': _strapHistoryNewestTs, - // Which WHOOP generation this batch came from — records/sessions - // vary hugely in richness by generation (and, for gen5, by whether - // the R22 deep-buffer opt-in was sent), so downstream diagnostics - // need this without reaching into the transport layer. - 'band_generation': state.generation, - }, - )); - // Same event, but a REAL per-chunk row keyed by the token — closes out - // whatever ack_failed history this token accumulated above. - await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry( - chunkId: 'batch:$tokenHex', - kind: 'historical_batch', - status: 'acked', - ackedAt: DateTime.now().millisecondsSinceEpoch, - metaPatch: { - 'batch_id': m.batchId, - // Cumulative for the connection, not this chunk (kept as it was); - // the chunk's own counts and record-time range follow. - 'records': d.records, - if (d.lastChunkStats case final c?) ...{ - 'chunk_rows': c.raws, - 'chunk_archived': c.archives, - 'chunk_ts_min': ?c.tsMin, - 'chunk_ts_max': ?c.tsMax, + // Both post-ACK diagnostic rows share one transaction and one commit. + final postWatch = clock.stopwatch()..start(); + await _bestEffortLedgerWrite( + () => LocalDb.syncLedgerTransaction((txn) async { + await LocalDb.upsertSyncLedgerEntry( + txn: txn, + status: 'acknowledged', + ackedAt: DateTime.now().millisecondsSinceEpoch, + metaPatch: { + 'last_batch_token': tokenHex, + 'last_batch_id': m.batchId, + 'last_batch_records': d.records, + 'last_ack_batches': d.batches, + 'strap_history_oldest_ts': _strapHistoryOldestTs, + 'strap_history_newest_ts': _strapHistoryNewestTs, + // Which WHOOP generation this batch came from — records/sessions + // vary hugely in richness by generation (and, for gen5, by whether + // the R22 deep-buffer opt-in was sent), so downstream diagnostics + // need this without reaching into the transport layer. + 'band_generation': state.generation, }, - }, - )); + ); + // Same event, but a REAL per-chunk row keyed by the token — closes out + // whatever ack_failed history this token accumulated above. + await LocalDb.upsertSyncLedgerEntry( + txn: txn, + chunkId: 'batch:$tokenHex', + kind: 'historical_batch', + status: 'acked', + ackedAt: DateTime.now().millisecondsSinceEpoch, + metaPatch: { + 'batch_id': m.batchId, + // Cumulative for the connection, not this chunk (kept as it was); + // the chunk's own counts and record-time range follow. + 'records': d.records, + if (d.lastChunkStats case final c?) ...{ + 'chunk_rows': c.raws, + 'chunk_archived': c.archives, + 'chunk_ts_min': ?c.tsMin, + 'chunk_ts_max': ?c.tsMax, + }, + }, + ); + })); + final stats = d.lastChunkStats; + _log('[perf] sync chunk records=${stats?.raws ?? 0} ' + 'beats=${stats?.beats ?? 0} commit_ms=${stats?.commitMs ?? 0} ' + 'ack_ms=$ackMs post_ms=${postWatch.elapsedMilliseconds}'); _noteStored(); // a banked batch → schedule a (debounced) derive } else if (m.sub == SyncMeta.historyComplete) { final d = _drain; @@ -9295,7 +9309,8 @@ class DrainController { /// What the last token commit banked: rows, archived frames and the range of /// record times. Written to that chunk's ledger row when it is ACKed. - ({int raws, int archives, int? tsMin, int? tsMax})? lastChunkStats; + ({int raws, int archives, int? tsMin, int? tsMax, int beats, int commitMs})? + lastChunkStats; int recordsThisOffload = 0; // since the last HISTORY_COMPLETE / rearm int batches = 0; // total this connection // Since the last [startFreshTask] — what [awaitComplete]'s report carries. @@ -9763,6 +9778,7 @@ class DrainController { _samples.clear(); _archives.clear(); _ecgRaw.clear(); + var commitMs = 0; try { // Defense in depth (constructor already rejects onRecordsBatch-only): // never report durable success for buffered content without onCommit. @@ -9778,15 +9794,19 @@ class DrainController { 'archives=${archives.length}, token=${tokenHex != null})', ); } + final commitWatch = clock.stopwatch()..start(); await commit(raws, samples, tokenHex, archives: archives, ecgRawPackets: ecgRaw); + commitMs = commitWatch.elapsedMilliseconds; } if (tokenHex != null) { // What this chunk actually banked, and the record times it carried // (#450: a chunk whose times sit far behind the stored edge is a band // clock that moved, which the per-chunk ledger row then shows). int? lo, hi; + var beats = 0; for (final x in samples) { + beats += x?.rrIntervalsMs.length ?? 0; final t = x?.tsEpoch; if (t == null || t <= 0) continue; if (lo == null || t < lo) lo = t; @@ -9797,6 +9817,8 @@ class DrainController { archives: archives.length, tsMin: lo, tsMax: hi, + beats: beats, + commitMs: commitMs, ); } return raws.length + archives.length + ecgRaw.length; diff --git a/lib/data/db.dart b/lib/data/db.dart index ed954c51b..30c82dfff 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -4399,6 +4399,8 @@ class LocalDb { 'cursor_advanced counter_hw=$maxCounter rec_ts_hw=$maxRecTs ' 'trim=${trimToken != null}', ); + // Freshness must roll back with the rows and cursor on failure. + await _writeCaptureFreshness(txn, raws); }); } finally { // ALWAYS restore NORMAL — even if the commit threw — so a leaked FULL does @@ -4409,7 +4411,6 @@ class LocalDb { /* non-fatal — see open-time PRAGMA discipline */ } } - await _writeCaptureFreshness(raws); } // ── DERIVED STORE (permanent, rich) ──────────────────────────────────────── @@ -6419,7 +6420,10 @@ class LocalDb { return fallbackSec; } - static Future _writeCaptureFreshness(List raws) async { + static Future _writeCaptureFreshness( + DatabaseExecutor txn, + List raws, + ) async { if (raws.isEmpty) return; var latest = 0; for (final raw in raws) { @@ -6427,9 +6431,14 @@ class LocalDb { if (recTs > latest) latest = recTs; } if (latest <= 0) return; - final prev = await computeFreshness('capture'); + final prevRows = await txn.query( + 'compute_freshness', + where: 'key = ?', + whereArgs: ['capture'], + limit: 1, + ); Map payload = const {}; - final rawJson = prev?['payload_json']; + final rawJson = prevRows.isEmpty ? null : prevRows.first['payload_json']; if (rawJson is String && rawJson.isNotEmpty) { try { final d = jsonDecode(rawJson); @@ -6443,7 +6452,7 @@ class LocalDb { 'latest_raw_rec_ts': latest, 'latest_raw_day': _localDayLabelFromEpoch(latest), }; - await putComputeFreshness('capture', jsonEncode(payload)); + await putComputeFreshness('capture', jsonEncode(payload), txn: txn); } // Canonical LOCAL day label — shared with the UI/coach via day_label.dart so @@ -8123,8 +8132,9 @@ class LocalDb { static Future?> syncLedgerEntry([ String chunkId = 'capture', + DatabaseExecutor? txn, ]) async { - final db = await instance; + final db = txn ?? await instance; final rows = await db.query( 'sync_ledger', where: 'chunk_id = ?', @@ -8143,9 +8153,10 @@ class LocalDb { int? ackedAt, String? lastError, Map? metaPatch, + DatabaseExecutor? txn, }) async { - final db = await instance; - final existing = await syncLedgerEntry(chunkId); + final db = txn ?? await instance; + final existing = await syncLedgerEntry(chunkId, txn); final now = DateTime.now().millisecondsSinceEpoch; final meta = {}; if (existing != null) { @@ -8174,6 +8185,14 @@ class LocalDb { }, conflictAlgorithm: ConflictAlgorithm.replace); } + /// Commit related diagnostic ledger rows together after the trim ACK. + static Future syncLedgerTransaction( + Future Function(Transaction txn) body, + ) async { + final db = await instance; + await db.transaction(body); + } + static Future>> syncLedger() async { final db = await instance; return db.query('sync_ledger', orderBy: 'created_at ASC'); @@ -11438,9 +11457,10 @@ class LocalDb { static Future putComputeFreshness( String key, - String payloadJson, - ) async { - final db = await instance; + String payloadJson, { + DatabaseExecutor? txn, + }) async { + final db = txn ?? await instance; await db.insert('compute_freshness', { 'key': key, 'payload_json': payloadJson, diff --git a/test/sync_chunk_overhead_test.dart b/test/sync_chunk_overhead_test.dart new file mode 100644 index 000000000..668cd9bef --- /dev/null +++ b/test/sync_chunk_overhead_test.dart @@ -0,0 +1,604 @@ +// Band-sync per-chunk overhead (design: edge.research/design-sync-throughput.md). +// +// Each ACK-gating commit sets synchronous=FULL, reads it back, and restores +// NORMAL in finally. An active offload leaves the connection at NORMAL between +// chunks. Capture freshness shares the rows + cursor transaction; the two +// post-ACK ledger upserts share one separate transaction. Each completed chunk +// logs its counts and commit, ACK and post-ACK timings. +// +// AGENTS.md invariant 1: rows + cursor commit BEFORE the verbatim token is +// echoed; a failed commit or incomplete burst produces no ACK. +// +// Layout: +// A. LocalDb.commitSyncBatch durability and transactional freshness (2-3). +// B. The engine's HISTORY_END path over a real database (2g, 4-6), including +// a FULL/read-back/commit/NORMAL bracket for every chunk during an offload. +// +// The sqflite logger sees every call LocalDb makes, extending the SQL spy in +// ack_commit_sync_full_test.dart. Calls made outside a transaction carry +// no transactionId; that is how "one transaction" is told from "N autocommits". + +import 'dart:async'; +import 'dart:typed_data'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/ble/ble_engine.dart'; +import 'package:openstrap_edge/data/db.dart'; +import 'package:openstrap_edge/data/models.dart'; +import 'package:openstrap_protocol/openstrap_protocol.dart'; +import 'package:path/path.dart' as p; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:sqflite_common/sqflite_logger.dart'; +import 'package:sqflite_common_ffi/sqflite_ffi.dart'; + +// ── call log ──────────────────────────────────────────────────────────────── + +/// One thing that happened, in order: either a sqflite call or a marker the +/// test dropped in ('ack', 'commit-start', 'commit-end'). +class _Ev { + final String? method; // null for a marker + final List sql; + final Object? txId; + final String mark; + _Ev.call(this.method, this.sql, this.txId) : mark = ''; + _Ev.mark(this.mark) : method = null, sql = const [], txId = null; + + bool touches(String table) { + final re = RegExp('\\b$table\\b'); + return sql.any(re.hasMatch); + } + + bool get isWrite => sql.any( + (s) => RegExp(r'^\s*(INSERT|UPDATE|DELETE|REPLACE)\b', caseSensitive: false) + .hasMatch(s), + ); + + bool get isSynchronousPragma => + sql.any((s) => RegExp(r'pragma\s+synchronous', caseSensitive: false).hasMatch(s)); + + bool sqlIs(String re) => + sql.any((s) => RegExp(re, caseSensitive: false).hasMatch(s)); +} + +final _log = <_Ev>[]; +var _counting = false; + +void _mark(String m) { + if (_counting) _log.add(_Ev.mark(m)); +} + +/// Every SQL string one sqflite call carries (a batch carries one per op). +Iterable _sqlOf(Object? args) sync* { + if (args is Map) { + final sql = args['sql']; + if (sql is String) yield sql; + for (final v in args.values) { + if (v is List || v is Map) yield* _sqlOf(v); + } + } else if (args is List) { + for (final v in args) { + yield* _sqlOf(v); + } + } +} + +/// How many separate commits the writes to [table] in [events] cost: every +/// autocommit write is one, and every transaction holding a write is one. +/// Reads do not count; a batch outside a transaction is one call. +int _commitsTouching(Iterable<_Ev> events, String table) { + var autocommit = 0; + final txns = {}; + for (final e in events) { + if (e.method == null || !e.touches(table) || !e.isWrite) continue; + if (e.txId == null) { + autocommit++; + } else { + txns.add(e.txId!); + } + } + return autocommit + txns.length; +} + +// ── fixtures ──────────────────────────────────────────────────────────────── + +const _t0 = 1780000000; + +String _hex(int counter) => '2f18${counter.toRadixString(16).padLeft(8, '0')}'; + +RawRecord _raw(int counter) => RawRecord( + counter: counter, + packetType: 0x2f, + hex: _hex(counter), + capturedAt: (_t0 + counter) * 1000, + recTs: _t0 + counter, + ); + +Sample _sample(int counter) => + Sample(tsEpoch: _t0 + counter, counter: counter, hr: 60); + +/// `n` consecutive records starting at counter `from`. +(List, List) _records(int from, int n) => ( + [for (var i = 0; i < n; i++) _raw(from + i)], + [for (var i = 0; i < n; i++) _sample(from + i)], + ); + +Future _commit(int from, int n, {String? token}) { + final (raws, samples) = _records(from, n); + return LocalDb.commitSyncBatch(raws, samples, trimToken: token); +} + +late String _dbDir; + +Future _freshLocal() async { + await LocalDb.close(); + await databaseFactory.deleteDatabase(p.join(_dbDir, LocalDb.dbName)); + SharedPreferences.setMockInitialValues({}); + return LocalDb.instance; +} + +/// `PRAGMA synchronous` as the connection reports it now: 2 FULL, 1 NORMAL. +Future _syncLevel() async { + final db = await LocalDb.instance; + final rows = await db.rawQuery('PRAGMA synchronous'); + final v = rows.isEmpty ? null : rows.first.values.firstOrNull; + return v is num ? v.toInt() : null; +} + +Future _count(String table, [String where = '1=1']) async { + final db = await LocalDb.instance; + final r = await db.rawQuery('SELECT COUNT(*) AS c FROM $table WHERE $where'); + return (r.first['c'] as num).toInt(); +} + +// ── engine rig (B) ────────────────────────────────────────────────────────── + +int _wallNow() => DateTime.now().millisecondsSinceEpoch ~/ 1000; + +Uint8List _historyStart() => + Uint8List.fromList([PacketType.metadata, 0x01, SyncMeta.historyStart]); + +Uint8List _historyEnd({required int expected, required int token}) { + final inner = Uint8List(24); + inner[0] = PacketType.metadata; + inner[1] = 0x02; + inner[2] = SyncMeta.historyEnd; + final v = ByteData.sublistView(inner); + v.setUint32(3, 1786000000, Endian.little); + v.setUint32(9, expected, Endian.little); + v.setUint32(13, token, Endian.little); + v.setUint32(17, 0x18, Endian.little); + return inner; +} + +Uint8List _gen5V18Inner({required int ts, required int counter}) { + final inner = Uint8List(kGen5V18InnerLen); + final v = ByteData.sublistView(inner); + inner[0] = PacketType.historicalData; + inner[1] = 18; + inner[2] = 0x80; + v.setUint32(3, counter, Endian.little); + v.setUint32(7, ts, Endian.little); + inner[14] = 64; + v.setFloat32(33, 0.5, Endian.little); + v.setFloat32(45, 1.0, Endian.little); + return inner; +} + +/// The same fake gen5 link history_task_safety_test.dart uses, but its commit +/// sink is the REAL `LocalDb.commitSyncBatch`, so the call log sees the +/// commit, the ACK write and the post-ACK ledger writes in one timeline. +class _Rig { + final logs = []; + final acks = >[]; // body of each success result, in order + late final BleEngine engine; + + _Rig() { + _rigs.add(this); + engine = BleEngine(onRecord: (_, _) async {}, onState: (_) {}, log: logs.add); + engine.debugInstallFakeLink( + band: BandProfile.gen5, + onCommit: (raws, samples, token, + {archives, ecgRawPackets, deviceFamily}) async { + _mark('commit-start'); + await LocalDb.commitSyncBatch( + raws, + samples, + trimToken: token, + archives: archives, + ecgRawPackets: ecgRawPackets, + deviceFamily: deviceFamily, + ); + _mark('commit-end'); + }, + onWrite: (f) async { + final parsed = parseFrame(f, profile: BandProfile.gen5); + if (parsed == null || !parsed.valid) return true; + final opcode = parsed.inner[2]; + final body = parsed.inner.sublist(3); + if (opcode == Cmd.getClock) { + final seq = parsed.inner[1]; + scheduleMicrotask(() => engine.debugAbsorbDecoded(Decoded( + 'cmd_response', + { + 'opcode': Cmd.getClock, + 'req_seq': seq, + 'cmd_status': 1, + 'clock_epoch': _wallNow(), + }, + ))); + } + if (opcode == Cmd.historicalDataResult && + body.isNotEmpty && + body[0] == 0x01) { + acks.add(body); + _mark('ack'); + } + return true; + }, + ); + } + + void rx(Uint8List inner) => + engine.debugReceiveFrame(Frame(inner, true, true), role: 'data'); + + /// One full burst: [n] records, then the marker for [token]. + void chunk(int n, int token, {required int ts, required int counter}) { + for (var i = 0; i < n; i++) { + rx(_gen5V18Inner(ts: ts + i, counter: counter + i)); + } + rx(_historyEnd(expected: n, token: token)); + } + + /// Wait for the real receive queue, then for the logs to stop growing. + /// A quiet log alone could mean an ffi transaction is still in flight. + Future settle() async { + var last = -1; + var stable = 0; + for (var i = 0; i < 200 && stable < 4; i++) { + await pumpEventQueue(); + await Future.delayed(const Duration(milliseconds: 25)); + final now = _log.length + logs.length; + final snapshot = engine.offloadSnapshot; + final idle = snapshot['queue_draining'] == false && + snapshot['queued_frames'] == 0; + stable = idle && now == last ? stable + 1 : 0; + last = now; + } + expect(stable, 4, reason: 'the real receive queue must finish before assertions'); + } +} + +final _rigs = <_Rig>[]; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + setUpAll(() async { + sqfliteFfiInit(); + // The logger is the only public hook that sees every sqflite call. + // ignore: experimental_member_use + databaseFactory = SqfliteDatabaseFactoryLogger( + databaseFactoryFfi, + options: SqfliteLoggerOptions( + type: SqfliteDatabaseFactoryLoggerType.invoke, + log: (e) { + if (_counting && e is SqfliteLoggerInvokeEvent) { + final args = e.arguments; + _log.add(_Ev.call( + e.method, + _sqlOf(args).toList(), + args is Map ? args['transactionId'] : null, + )); + } + }, + ), + ); + SharedPreferences.setMockInitialValues({}); + LocalDb.dbName = 'openstrap_sync_chunk_overhead_test.db'; + _dbDir = await databaseFactory.getDatabasesPath(); + }); + + tearDownAll(() async { + _counting = false; + await LocalDb.close(); + await databaseFactory.deleteDatabase(p.join(_dbDir, LocalDb.dbName)); + }); + + setUp(() async { + _counting = false; + _log.clear(); + await _freshLocal(); + }); + + tearDown(() async { + _counting = false; + for (final r in _rigs) { + await r.engine.disconnect(); + await r.settle(); + } + _rigs.clear(); + }); + + group('A. commitSyncBatch round trips and durability', () { + // 2. Durability is bracketed per commit, including failures. + test('2b. a throwing commit restores NORMAL (1)', () async { + final db = await LocalDb.instance; + await db.execute('CREATE TRIGGER t_cursor_boom BEFORE INSERT ON sync_cursor ' + "WHEN NEW.name = 'strap_trim' BEGIN SELECT RAISE(ABORT, 'boom'); END"); + _counting = true; + await expectLater(_commit(1, 50, token: 't'), throwsA(anything)); + _counting = false; + + final full = _log.indexWhere( + (e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*FULL')); + final verify = _log.indexWhere( + (e) => e.sqlIs(r'^\s*pragma\s+synchronous\s*;?\s*$')); + final begin = _log.indexWhere((e) => e.sqlIs(r'^\s*BEGIN')); + final rollback = _log.indexWhere((e) => e.sqlIs(r'^\s*ROLLBACK')); + final normal = _log.indexWhere( + (e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*NORMAL')); + expect(full, isNonNegative); + expect(verify, greaterThan(full)); + expect(begin, greaterThan(verify)); + expect(rollback, greaterThan(begin)); + expect(normal, greaterThan(rollback), + reason: 'finally restores NORMAL after a failed transaction'); + expect(await _syncLevel(), 1); + expect(await _count('samples'), 0); + }); + + test('2d. a commit runs at FULL and restores NORMAL after COMMIT', () async { + _counting = true; + await _commit(1, 50, token: 't'); + _counting = false; + + int idx(bool Function(_Ev) f) => _log.indexWhere(f); + final full = idx((e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*FULL')); + final verify = idx( + (e) => e.sqlIs(r'^\s*pragma\s+synchronous\s*;?\s*$')); + final begin = idx((e) => e.sqlIs(r'^\s*BEGIN')); + final commit = idx((e) => e.sqlIs(r'^\s*COMMIT')); + final normal = idx((e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*NORMAL')); + expect(full, isNonNegative); + expect(verify, greaterThan(full)); + expect(begin, greaterThan(verify), + reason: 'FULL is set and read back BEFORE the txn opens'); + expect(commit, greaterThan(begin)); + expect(normal, greaterThan(commit), reason: 'NORMAL restored after COMMIT'); + expect(await _syncLevel(), 1); + }); + + // 3. Freshness rides in the batch transaction. + test('3a. a failing freshness write rolls the whole batch back ' + '(freshness is inside the transaction)', () async { + final db = await LocalDb.instance; + await db.execute( + 'CREATE TRIGGER t_fresh_boom BEFORE INSERT ON compute_freshness ' + "WHEN NEW.\"key\" = 'capture' BEGIN SELECT RAISE(ABORT, 'boom'); END"); + + await expectLater(_commit(1, 50, token: 't'), throwsA(anything)); + + expect(await _count('samples'), 0, + reason: 'the rows must roll back with the freshness row; today the ' + 'batch is already committed when freshness fails'); + expect(await _count('sync_cursor', "name = 'counter_hw'"), 0, + reason: 'and so must the cursor'); + expect(await _count('compute_freshness', "\"key\" = 'capture'"), 0); + }); + + test('3b. a failing cursor write leaves neither rows nor freshness ' + '(guard: holds today, must stay)', () async { + final db = await LocalDb.instance; + await db.execute('CREATE TRIGGER t_cursor_boom BEFORE INSERT ON sync_cursor ' + "WHEN NEW.name = 'strap_trim' BEGIN SELECT RAISE(ABORT, 'boom'); END"); + + await expectLater(_commit(1, 50, token: 't'), throwsA(anything)); + + expect(await _count('samples'), 0); + expect(await _count('sync_cursor', "name = 'counter_hw'"), 0); + expect(await _count('compute_freshness', "\"key\" = 'capture'"), 0); + }); + + test('3c. the capture-freshness select and write happen inside the batch ' + 'transaction, not after it', () async { + _counting = true; + await _commit(1, 50, token: 't'); + _counting = false; + + final fresh = _log.where((e) => e.touches('compute_freshness')).toList(); + expect(fresh, isNotEmpty, reason: 'harness check: freshness is written'); + final outside = fresh.where((e) => e.txId == null).length; + expect(outside, 0, + reason: '$outside freshness call(s) ran outside the commit ' + 'transaction (today: a select and an insert, 2 extra round trips)'); + final rows = _log.singleWhere((e) => e.touches('samples') && e.isWrite); + final cursor = _log.where((e) => e.touches('sync_cursor') && e.isWrite); + expect(rows.txId, isNotNull); + expect(cursor, isNotEmpty); + expect([...fresh, ...cursor].every((e) => e.txId == rows.txId), isTrue, + reason: 'freshness, decoded rows and trim cursor share the SAME txn'); + }); + + test('3d. the freshness row still carries the batch max record time ' + '(guard: holds today, must stay)', () async { + await _commit(1, 50, token: 't'); + final row = await LocalDb.computeFreshness('capture'); + expect(row, isNotNull); + expect(row!['payload_json'] as String, + contains('"latest_raw_rec_ts":${_t0 + 50}')); + }); + }); + + group('B. HISTORY_END over a real database', () { + final ts = _wallNow() - 3600; + + test('2g. EVERY commit during an offload has its own FULL bracket and ' + 'leaves NORMAL between chunks', () async { + final r = _Rig(); + r.engine.debugClaimHistoryTask(); + expect(await _syncLevel(), 1, + reason: 'starting an offload must leave the connection at NORMAL'); + for (var i = 0; i < 3; i++) { + _log.clear(); + _counting = true; + r.rx(_historyStart()); + r.chunk(3, 0xB200 + i, ts: ts + i * 100, counter: 2000 + i * 100); + await r.settle(); + _counting = false; + + expect(r.acks, hasLength(i + 1)); + expect(r.engine.offloadActive, isTrue, + reason: 'the same offload remains active between chunks'); + expect(_log.where((e) => e.isSynchronousPragma), hasLength(3), + reason: 'chunk $i sets FULL, reads it back, then restores NORMAL'); + final full = _log.indexWhere( + (e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*FULL')); + final verify = _log.indexWhere( + (e) => e.sqlIs(r'^\s*pragma\s+synchronous\s*;?\s*$')); + final begin = _log.indexWhere((e) => e.sqlIs(r'^\s*BEGIN')); + final commit = _log.indexWhere((e) => e.sqlIs(r'^\s*COMMIT')); + final normal = _log.indexWhere( + (e) => e.sqlIs(r'pragma\s+synchronous\s*=\s*NORMAL')); + final ack = _log.indexWhere((e) => e.mark == 'ack'); + expect(full, isNonNegative); + expect(verify, greaterThan(full)); + expect(begin, greaterThan(verify)); + expect(commit, greaterThan(begin)); + expect(normal, greaterThan(commit)); + expect(ack, greaterThan(normal)); + expect(await _syncLevel(), 1, + reason: 'chunk $i must leave NORMAL for post-ACK writes and idle time'); + } + }); + + // 4. Ledger: one transaction per marker. + test('4. the two post-ACK ledger upserts cost one commit, and both rows ' + 'still land', () async { + final r = _Rig(); + r.engine.debugClaimHistoryTask(); + await LocalDb.instance; // opened before counting: opening is not the cost + _counting = true; + r.rx(_historyStart()); + r.chunk(3, 0xB400, ts: ts, counter: 4000); + await r.settle(); + _counting = false; + + expect(r.acks, hasLength(1), reason: 'harness check: the chunk was ACKed'); + final ackAt = _log.indexWhere((e) => e.mark == 'ack'); + expect(ackAt, isNonNegative); + final after = _log.sublist(ackAt + 1); + final commits = _commitsTouching(after, 'sync_ledger'); + expect(commits, 1, + reason: 'one marker costs one ledger commit after the ACK; ' + 'today it is $commits (two separate upserts)'); + + // Same content as before: the 'capture' row and the per-token row. + final ledger = await LocalDb.syncLedger(); + final capture = ledger.where((e) => e['chunk_id'] == 'capture').toList(); + expect(capture, hasLength(1)); + expect(capture.single['status'], 'acknowledged'); + expect(capture.single['meta_json'] as String, contains('last_batch_token')); + final tok = ledger.where((e) => (e['chunk_id'] as String).startsWith('batch:')); + expect(tok, hasLength(1)); + expect(tok.single['status'], 'acked'); + expect(tok.single['meta_json'] as String, contains('"chunk_rows":3')); + }); + + // 5. Perf line. + test('5. each HISTORY_END commit logs exactly one [perf] sync chunk line ' + 'with per-chunk counts and timings', () async { + final r = _Rig(); + r.engine.debugClaimHistoryTask(); + r.rx(_historyStart()); + r.chunk(3, 0xB500, ts: ts, counter: 5000); + await r.settle(); + r.rx(_historyStart()); + r.chunk(2, 0xB501, ts: ts + 100, counter: 5100); + await r.settle(); + + expect(r.acks, hasLength(2), reason: 'harness check: both chunks ACKed'); + final lines = r.logs.where((l) => l.contains('[perf] sync chunk')).toList(); + expect(lines, hasLength(2), + reason: 'one line per HISTORY_END, no more, no fewer; saw $lines'); + + final re = RegExp(r'\brecords=(\d+)\b.*\bbeats=(\d+)\b.*\bcommit_ms=(\d+)\b' + r'.*\back_ms=(\d+)\b.*\bpost_ms=(\d+)\b'); + for (final l in lines) { + expect(re.hasMatch(l), isTrue, + reason: 'line must carry records= beats= commit_ms= ack_ms= ' + 'post_ms= in that order: $l'); + } + // Per chunk, not cumulative for the connection. + expect(re.firstMatch(lines[0])!.group(1), '3'); + expect(re.firstMatch(lines[1])!.group(1), '2'); + }); + + // 6. Invariant guards (must pass today and stay green). + test('6a. the ACK is written only after the commit completed, and echoes ' + 'the verbatim token (guard)', () async { + final r = _Rig(); + r.engine.debugClaimHistoryTask(); + _counting = true; + r.rx(_historyStart()); + r.chunk(3, 0xB600, ts: ts, counter: 6000); + await r.settle(); + _counting = false; + + final commitEnd = _log.indexWhere((e) => e.mark == 'commit-end'); + final txnCommit = _log.indexWhere((e) => e.sqlIs(r'^\s*COMMIT')); + final ack = _log.indexWhere((e) => e.mark == 'ack'); + expect(commitEnd, isNonNegative); + expect(txnCommit, isNonNegative); + expect(ack, greaterThan(commitEnd), + reason: 'ACK must follow commitSyncBatch returning'); + expect(ack, greaterThan(txnCommit)); + expect(r.acks.single.take(9).toList(), [ + 0x01, + 0x00, 0xB6, 0x00, 0x00, // token u32 LE + 0x18, 0x00, 0x00, 0x00, // batch id u32 LE + ]); + }); + + test('6b. a commit that throws produces no ACK and no acked ledger row ' + '(guard)', () async { + final db = await LocalDb.instance; + await db.execute('CREATE TRIGGER t_cursor_boom BEFORE INSERT ON sync_cursor ' + "WHEN NEW.name = 'strap_trim' BEGIN SELECT RAISE(ABORT, 'boom'); END"); + final r = _Rig(); + r.engine.debugClaimHistoryTask(); + r.rx(_historyStart()); + r.chunk(3, 0xB601, ts: ts, counter: 6100); + await r.settle(); + + expect(r.acks, isEmpty, reason: 'no ACK for a chunk that rolled back'); + expect(await _count('samples'), 0, reason: 'harness check: it rolled back'); + final ledger = await LocalDb.syncLedger(); + expect( + ledger.where((e) => e['status'] == 'acked' || e['status'] == 'acknowledged'), + isEmpty); + expect(r.logs.where((l) => l.contains('[perf] sync chunk')), isEmpty, + reason: 'a failed commit is not a completed chunk'); + }); + + test('6c. an incomplete burst is neither committed nor ACKed (guard)', + () async { + final r = _Rig(); + r.engine.debugClaimHistoryTask(); + _counting = true; + r.rx(_historyStart()); + r.rx(_gen5V18Inner(ts: ts, counter: 6200)); + r.rx(_historyEnd(expected: 3, token: 0xB602)); + await r.settle(); + _counting = false; + + expect(r.acks, isEmpty, reason: 'a partial chunk must never trim flash'); + // Upstream's count gate stores what arrived without the trim token + // (_refuseHistoryEndOnShortCount), so only the cursor and ACK are pinned. + expect(await LocalDb.getCursor('strap_trim'), isNull); + final ledger = await LocalDb.syncLedger(); + expect(ledger.where( + (e) => e['status'] == 'acked' || e['status'] == 'acknowledged'), isEmpty); + }); + + }); +} From 71cda76e6f3bbd8d548416045c0fbf9e61068eb1 Mon Sep 17 00:00:00 2001 From: MikenRafter Date: Fri, 23 Oct 2026 16:31:18 -0600 Subject: [PATCH 2/5] perf(import): restore the big tables inside SQLite (ATTACH + INSERT ... SELECT) A restore read every row of the source through sqflite in 2,000-row pages and wrote it back one insert at a time: about 200 platform calls per 200k rows of decoded_onehz, each marshalling the page through the platform channel. The source is now attached to the destination connection and the eleven plain tables (decoded_onehz, decoded_rr, samples, raw_archive, events, band_events, band_battery, day_result, metric_series, metric_series_version, metric_method) are copied with ordered INSERT OR REPLACE ... SELECT in 25k-row chunks, one transaction per chunk, so a band-sync commit never waits long behind one and a failure keeps the chunks already committed. The legacy key derivation, retired fields, the skin-temp sentinel, finalized-row protection and the documented beatless-import gap behave as on the row path; sleep_nap, sessions and the substrate archive keep the row path. Temp tables are dropped and the source detached on success and on every failure; errors still propagate. The row-by-row merge is kept as a test oracle (debugMergeRowByRow), and the tests require identical table contents from both paths. One difference: decoded_rr's stale-beat cleanup now uses the highest beat per second over the whole source. When a later 2,000-row page carried only lower indices for a second, the row path deleted higher beats the source has; the bulk path keeps them (pinned by a test). --- lib/data/db.dart | 351 ++++++++- test/import_bulk_restore_test.dart | 921 +++++++++++++++++++++++ test/import_restore_perf_probe_test.dart | 113 +++ 3 files changed, 1373 insertions(+), 12 deletions(-) create mode 100644 test/import_bulk_restore_test.dart create mode 100644 test/import_restore_perf_probe_test.dart diff --git a/lib/data/db.dart b/lib/data/db.dart index 30c82dfff..2b3961b8e 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -10071,14 +10071,43 @@ class LocalDb { @visibleForTesting static List get salvageTablesForTest => _salvageTables; + /// Forces the existing per-row merge for the bulk-restore equivalence tests. + @visibleForTesting + static bool debugRowByRowMerge = false; + + @visibleForTesting + static Future> debugMergeRowByRow(String path) async { + final previous = debugRowByRowMerge; + debugRowByRowMerge = true; + try { + return await _mergeFromDbFile(path); + } finally { + debugRowByRowMerge = previous; + } + } + static Future> _mergeFromDbFile( String path, { List? only, bool tolerant = false, }) async { - final src = await openDatabase(path, readOnly: true); final db = await instance; + final src = await openDatabase(path, readOnly: true); const tables = _restoreTables; + const bulkTables = { + 'decoded_onehz', + 'decoded_rr', + 'samples', + 'raw_archive', + 'events', + 'band_events', + 'band_battery', + 'day_result', + 'metric_series', + 'metric_series_version', + 'metric_method', + }; + // Columns this app's schema actually has, per table — so a row from a NEWER // export carrying extra columns this build doesn't know about is filtered // down (dropped) instead of throwing "no such column". A column the source @@ -10108,18 +10137,306 @@ class LocalDb { // version stamp and this device's baselines are the same day's history and // get the same protection, or trends and the rolling baselines would be // rebuilt from the other export's numbers while day detail serves ours. - final finalizedDays = { - for (final r in await db.query( - 'day_result', - columns: ['day_id'], - where: 'finalized = 1', - distinct: true, - )) - '${r['day_id']}', - }; + final finalizedDays = {}; var archiveSkipped = 0; int? archiveEdgeSec; + var impAttached = false; + var failed = true; + + // Attempt every cleanup operation. Preserve the original restore error, + // but surface a cleanup failure when the copy otherwise succeeded. + Future cleanupRestore({ + List tempTables = const [], + bool closeSource = false, + required bool preserveError, + }) async { + Object? firstError; + StackTrace? firstStack; + Future attempt(Future Function() action) async { + try { + await action(); + } catch (error, stack) { + firstError ??= error; + firstStack ??= stack; + } + } + + for (final name in tempTables) { + await attempt(() => db.execute('DROP TABLE IF EXISTS temp.$name')); + } + if (closeSource) { + if (impAttached) { + await attempt(() => db.execute('DETACH DATABASE imp')); + } + await attempt(src.close); + } + final error = firstError; + if (!preserveError && error != null) { + Error.throwWithStackTrace(error, firstStack!); + } + } + + String quoteSql(String name) => '"${name.replaceAll('"', '""')}"'; + String sourceColumn(String name) => 's.${quoteSql(name)}'; + String numericSql(String value) => + "typeof($value) IN ('integer', 'real')"; + + Future mergeBulkTable(String t) async { + if (!impAttached || !bulkTables.contains(t)) return false; + final exists = await db.rawQuery( + "SELECT 1 FROM imp.sqlite_master WHERE type = 'table' AND name = ?", + [t], + ); + if (exists.isEmpty) return true; + + if (t == 'day_result') importedDays = {}; + final destInfo = await db.rawQuery('PRAGMA main.table_info(${quoteSql(t)})'); + if (destInfo.isEmpty) return false; + final sourceInfo = await db.rawQuery('PRAGMA imp.table_info(${quoteSql(t)})'); + final destColumns = [ + for (final column in destInfo) column['name'] as String, + ]; + final sourceColumns = { + for (final column in sourceInfo) column['name'] as String, + }; + final maxRowid = Sqflite.firstIntValue( + await db.rawQuery('SELECT MAX(rowid) FROM imp.${quoteSql(t)}'), + ); + if (maxRowid == null || maxRowid <= 0) { + counts[t] = 0; + return true; + } + + if (!sourceColumns.any(destColumns.contains)) return false; + + const deviceKeyed = {'samples', 'decoded_onehz', 'decoded_rr'}; + final isDeviceKeyed = deviceKeyed.contains(t); + final sourceTsMs = sourceColumns.contains('ts_ms') + ? sourceColumn('ts_ms') + : 'NULL'; + final sourceDeviceId = sourceColumns.contains('device_id') + ? sourceColumn('device_id') + : 'NULL'; + final sourceRecTs = sourceColumns.contains('rec_ts') + ? sourceColumn('rec_ts') + : 'NULL'; + final sourceRrTsMs = sourceColumns.contains('rr_ts_ms') + ? sourceColumn('rr_ts_ms') + : 'NULL'; + final recoveredRecTs = t == 'decoded_rr' + ? 'CASE WHEN $sourceRecTs IS NULL THEN ' + 'CASE WHEN ${numericSql(sourceRrTsMs)} ' + 'THEN CAST($sourceRrTsMs AS INTEGER) / 1000 ELSE NULL END ' + 'ELSE $sourceRecTs END' + : sourceRecTs; + final keyTime = t == 'samples' + ? (sourceColumns.contains('ts') ? sourceColumn('ts') : 'NULL') + : t == 'decoded_rr' + ? recoveredRecTs + : sourceRecTs; + final derivedTsMs = + 'CASE WHEN $sourceTsMs IS NULL THEN CAST($keyTime AS INTEGER) * 1000 ' + 'ELSE $sourceTsMs END'; + final derivedDeviceId = + 'CASE WHEN $sourceTsMs IS NULL THEN COALESCE($sourceDeviceId, ' + "'$kPrimaryDeviceId') ELSE $sourceDeviceId END"; + + final expressions = {}; + for (final column in destColumns) { + if (sourceColumns.contains(column)) { + expressions[column] = sourceColumn(column); + } + } + if (t == 'decoded_onehz') { + if (destColumns.contains('on_wrist')) expressions['on_wrist'] = 'NULL'; + if (destColumns.contains('hr_valid')) expressions['hr_valid'] = 'NULL'; + if (destColumns.contains('skin_temp_c') && + sourceColumns.contains('skin_temp_c')) { + final skinTemp = sourceColumn('skin_temp_c'); + expressions['skin_temp_c'] = + 'CASE WHEN ${numericSql(skinTemp)} AND $skinTemp <= -49.995 ' + 'THEN NULL ELSE $skinTemp END'; + } + } + if (t == 'decoded_rr' && destColumns.contains('rec_ts')) { + expressions['rec_ts'] = recoveredRecTs; + } + if (isDeviceKeyed && destColumns.contains('device_id')) { + expressions['device_id'] = derivedDeviceId; + } + if (isDeviceKeyed && destColumns.contains('ts_ms')) { + expressions['ts_ms'] = derivedTsMs; + } + final columns = [ + for (final column in destColumns) + if (expressions.containsKey(column)) column, + ]; + if (columns.isEmpty) return false; + + final filters = []; + if (t == 'decoded_rr') { + // The oracle drops a beat with no rec_ts and no usable rr_ts_ms even + // when ts_ms is present; letting it through fails NOT NULL for the + // whole chunk. + filters.add('NOT ($sourceRecTs IS NULL AND NOT (${numericSql(sourceRrTsMs)}))'); + } + if (isDeviceKeyed) { + filters.add('NOT ($sourceTsMs IS NULL AND NOT (${numericSql(keyTime)}))'); + } + if (t == 'day_result') { + // The oracle compares interpolated strings, before destination + // affinity can coerce a foreign TEXT version such as '0100' to 100. + filters.add('NOT EXISTS (SELECT 1 FROM temp._restore_protected_day_results p ' + 'WHERE CAST(p.day_id AS TEXT) = CAST(${sourceColumn('day_id')} AS TEXT) AND ' + 'CAST(p.algo_version AS TEXT) = CAST(${sourceColumn('algo_version')} AS TEXT))'); + } + if (t == 'metric_series' || + t == 'metric_series_version' || + t == 'metric_method') { + filters.add('NOT EXISTS (SELECT 1 FROM temp._restore_finalized_days p ' + 'WHERE CAST(p.day_id AS TEXT) = CAST(${sourceColumn('date')} AS TEXT))'); + } + final unbounded = filters.isEmpty ? '1' : filters.join(' AND '); + final where = ['s.rowid > ?', 's.rowid <= ?', ...filters].join(' AND '); + final selectSql = 'SELECT ${[ + for (final column in columns) + '${expressions[column]} AS ${quoteSql(column)}', + ].join(', ')} FROM imp.${quoteSql(t)} s WHERE $where'; + // Rowid order: when derived keys collide, the later source row wins, as + // on the per-row path. + final insertSql = 'INSERT OR REPLACE INTO main.${quoteSql(t)} ' + '(${columns.map(quoteSql).join(', ')}) $selectSql ORDER BY s.rowid'; + + // Small enough that a band-sync commit (which must land before the ACK) + // never waits long behind one, and a salvage loses little to one + // unreadable chunk. + const chunkRows = 25000; + var failed = true; + var copied = 0; + try { + if (t == 'decoded_rr') { + // The highest beat per second over the WHOLE source, not one chunk: + // REPLACE renumbers rowids, so one second's beats can straddle + // chunks out of order, and a per-chunk max would clear beats an + // earlier chunk just restored. + // Unlike the oracle's per-2k-page sweep, this also retains an earlier + // high beat when a later oracle page carries only lower indices. + await db.execute('DROP TABLE IF EXISTS temp._restore_rr_src_max'); + await db.execute('''CREATE TEMP TABLE _restore_rr_src_max + (d TEXT, k INTEGER, m INTEGER, PRIMARY KEY (d, k))'''); + // Paged like the copy below, so a band-sync commit can land between + // pages instead of waiting out a scan of the whole source. + for (var lo = 0; lo < maxRowid; lo += chunkRows) { + // REPLACE with a LEFT JOIN rather than an UPSERT: Android 8-10 + // ship a system SQLite older than ON CONFLICT DO UPDATE. + await db.execute('''INSERT OR REPLACE INTO temp._restore_rr_src_max + (d, k, m) + SELECT c.d, c.k, MAX(c.m, COALESCE(x.m, c.m)) + FROM (SELECT $derivedDeviceId AS d, $derivedTsMs AS k, + MAX(CAST(${sourceColumn('beat_index')} AS INTEGER)) AS m + FROM imp.${quoteSql(t)} s + WHERE s.rowid > ? AND s.rowid <= ? AND $unbounded + AND ${numericSql(sourceColumn('beat_index'))} + GROUP BY d, k) c + LEFT JOIN temp._restore_rr_src_max x ON x.d = c.d AND x.k = c.k''', + [lo, lo + chunkRows]); + } + } + var lower = 0; + while (lower < maxRowid) { + final upper = lower + chunkRows > maxRowid ? maxRowid : lower + chunkRows; + final first = lower == 0; + await db.transaction((txn) async { + if (t == 'day_result' && first) { + // Inside the first chunk's transaction, so a day the engine + // finalizes while the restore runs is still protected. + await txn.execute( + 'DROP TABLE IF EXISTS temp._restore_protected_day_results'); + await txn.execute( + 'CREATE TEMP TABLE _restore_protected_day_results AS ' + 'SELECT day_id, algo_version FROM main.day_result ' + 'WHERE finalized = 1'); + } + if (t == 'day_result') { + final days = await txn.rawQuery( + 'SELECT DISTINCT ${sourceColumn('day_id')} AS day_id ' + 'FROM imp.${quoteSql(t)} s WHERE $where', + [lower, upper], + ); + importedDays?.addAll( + days.map((row) => '${row['day_id']}'), + ); + } + await txn.execute(insertSql, [lower, upper]); + copied += Sqflite.firstIntValue( + await txn.rawQuery('SELECT changes()'), + ) ?? + 0; + if (t == 'decoded_rr') { + // Materialised and driven from the chunk's seconds, so the + // delete walks decoded_rr's (device_id, ts_ms, beat_index) key + // once per carried second. A correlated EXISTS over the CTE + // scanned every local beat against every carried second. + await txn.execute('DROP TABLE IF EXISTS temp._restore_rr_carried'); + await txn.execute('''CREATE TEMP TABLE _restore_rr_carried AS + SELECT DISTINCT x.d AS d, x.k AS k, x.m AS m + FROM (SELECT $derivedDeviceId AS d, $derivedTsMs AS k + FROM imp.${quoteSql(t)} s WHERE $where + AND ${numericSql(sourceColumn('beat_index'))}) c + JOIN temp._restore_rr_src_max x ON x.d = c.d AND x.k = c.k''', + [lower, upper]); + await txn.execute('''DELETE FROM main.decoded_rr WHERE rowid IN ( + SELECT r.rowid FROM temp._restore_rr_carried c + JOIN main.decoded_rr r + ON r.device_id = c.d AND r.ts_ms = c.k AND r.beat_index > c.m)'''); + await txn.execute('DROP TABLE temp._restore_rr_carried'); + } + }); + lower = upper; + } + failed = false; + } finally { + await cleanupRestore( + tempTables: [ + if (t == 'day_result') '_restore_protected_day_results', + if (t == 'decoded_rr') ...[ + '_restore_rr_carried', '_restore_rr_src_max', + ], + ], + preserveError: failed, + ); + } + counts[t] = copied; + return true; + } + try { + if (!debugRowByRowMerge) { + try { + await db.execute('ATTACH DATABASE ? AS imp', [path]); + impAttached = true; + } on DatabaseException { + // A damaged-file salvage can still try the per-row reader. A public + // restore must propagate its failure instead of claiming success. + if (!tolerant) rethrow; + } + } + await db.transaction((txn) async { + finalizedDays.addAll({ + for (final r in await txn.query( + 'day_result', + columns: ['day_id'], + where: 'finalized = 1', + distinct: true, + )) + '${r['day_id']}', + }); + if (impAttached) { + await txn.execute('DROP TABLE IF EXISTS temp._restore_finalized_days'); + await txn.execute('CREATE TEMP TABLE _restore_finalized_days AS ' + 'SELECT day_id FROM main.day_result WHERE finalized = 1'); + } + }); if ((only ?? tables).contains(SubstrateArchive.table)) { try { await SubstrateArchive.adoptPolicy(src, db); @@ -10146,6 +10463,7 @@ class LocalDb { } for (final t in (only ?? tables)) { try { + if (await mergeBulkTable(t)) continue; // PAGED SOURCE READ — never `SELECT *` a whole table. // // This used to be a single `src.query(t)`. sqflite serialises an entire @@ -10575,8 +10893,16 @@ class LocalDb { counts[t] = 0; } } + failed = false; } finally { - await src.close(); + await cleanupRestore( + tempTables: impAttached ? [ + '_restore_rr_carried', '_restore_rr_src_max', + '_restore_protected_day_results', '_restore_finalized_days', + ] : const [], + closeSource: true, + preserveError: failed, + ); } if (counts.values.any((count) => count > 0)) { await ActivityStore.suppressImportedAlerts(db); @@ -10624,7 +10950,8 @@ class LocalDb { } // Last, so it can never be mistaken for a table row count by anything that // walks this map in order. - if (importedDays != null) counts['_days'] = importedDays.length; + final days = importedDays; + if (days != null) counts['_days'] = days.length; // Archive buckets that could not be merged (a codec this build does not // know) — kept out of the row counts, never silently. if (archiveSkipped > 0) counts['_substrate_archive_skipped'] = archiveSkipped; diff --git a/test/import_bulk_restore_test.dart b/test/import_bulk_restore_test.dart new file mode 100644 index 000000000..f01873820 --- /dev/null +++ b/test/import_bulk_restore_test.dart @@ -0,0 +1,921 @@ +// ED-1 E2 RED: ported from cb81734f onto upstream's LocalDb restore owner. +// Upstream still reads 2,000 rows into Dart maps and batches the writes. +// The proposed bulk path copies inside SQLite with ATTACH + INSERT SELECT, +// in 25,000-row chunks. These tests hold that path to the existing merge: +// 1. the same final database, rule by rule (LocalDb.debugMergeRowByRow is +// the oracle); +// 2. a bounded number of sqflite calls per big table, not one per page; +// 3. chunks commit separately and a rerun converges; +// 4. the source is detached on success and on failure. + +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:sqflite_common/sqflite_logger.dart'; +import 'package:sqflite_common_ffi/sqflite_ffi.dart'; +import 'package:openstrap_edge/data/db.dart'; +import 'package:openstrap_edge/data/substrate_archive_codec.dart'; + +// Every sqflite call LocalDb makes, while [_counting] is on. +final _invokes = []; +var _counting = false; + +/// The SQL strings one sqflite call carries (a batch carries one per op). +Iterable _sqlOf(Object? args) sync* { + if (args is Map) { + final sql = args['sql']; + if (sql is String) yield sql; + for (final v in args.values) { + if (v is List || v is Map) yield* _sqlOf(v); + } + } else if (args is List) { + for (final v in args) { + yield* _sqlOf(v); + } + } +} + +Iterable _operations(Object? args) sync* { + if (args is Map) { + if (args['sql'] is String) yield args; + for (final v in args.values) { + if (v is List || v is Map) yield* _operations(v); + } + } else if (args is List) { + for (final v in args) { + yield* _operations(v); + } + } +} + +/// sqflite calls whose SQL names [table]. +int _callsTouching(String table) { + final re = RegExp('\\b$table\\b'); + return _invokes + .where((e) => _sqlOf(e.arguments).any(re.hasMatch)) + .length; +} + +const _t0 = 1790000000; // 2026-09-21, an arbitrary fixed second + +/// Stamped with DateTime.now(): differs between any two +/// runs, so the equivalence dump leaves it out. +const _wallClockColumns = { + 'compute_freshness': {'updated_at'}, + 'activity_review_meta': {'activated_at', 'alert_after_ts'}, + 'substrate_archive': {'updated_at'}, +}; + +Map archiveSecond(int ts, int hr) => { + 'device_id': '', 'ts_ms': ts * 1000, 'rec_ts': ts, 'counter': ts - _t0, + 'hr': hr, +}; + +Map archiveRow(List> rows) { + final table = (ArchiveTableBuilder()..addRows(rows)).build(); + final w = SubstrateArchiveCodec.encodeMergeVerify( + ArchiveBucket(table, ArchiveTable.empty)); + return { + 'device_id': '', 'utc_day': _t0 ~/ 86400, + 'codec': SubstrateArchiveCodec.codecId, + 'from_ts': w.fromTs, 'to_ts': w.toTs, + 'onehz_rows': w.onehzRows, 'rr_rows': w.rrRows, + 'raw_bytes': w.rawBytes, 'fingerprint': w.fingerprint, + 'blob': w.blob, 'created_at': 1, 'updated_at': 1, + }; +} + +void main() { + late Directory tmp; + late String dbDir; + + setUpAll(() async { + sqfliteFfiInit(); + // The logger is the only public hook that sees every sqflite call. + // ignore: experimental_member_use + databaseFactory = SqfliteDatabaseFactoryLogger( + databaseFactoryFfi, + options: SqfliteLoggerOptions( + type: SqfliteDatabaseFactoryLoggerType.invoke, + log: (e) { + if (_counting && e is SqfliteLoggerInvokeEvent) _invokes.add(e); + }, + ), + ); + SharedPreferences.setMockInitialValues({}); + LocalDb.dbName = 'openstrap_bulk_restore_test.db'; + dbDir = await databaseFactory.getDatabasesPath(); + tmp = await Directory.systemTemp.createTemp('openstrap_bulk_'); + }); + + tearDownAll(() async { + await LocalDb.close(); + await databaseFactory.deleteDatabase(p.join(dbDir, LocalDb.dbName)); + if (await tmp.exists()) await tmp.delete(recursive: true); + }); + + tearDown(() { + _counting = false; + _invokes.clear(); + expect(LocalDb.debugRowByRowMerge, isFalse); + }); + + /// A brand-new, empty local database. + Future freshLocal() async { + await LocalDb.close(); + await databaseFactory.deleteDatabase(p.join(dbDir, LocalDb.dbName)); + SharedPreferences.setMockInitialValues({}); + return LocalDb.instance; + } + + /// Every table, every row, as SQL literals (quote() keeps the storage + /// class: 1, 1.0 and '1' differ), sorted so row order does not matter. + Future>> dump( + {Set? only, Set except = const {}}) async { + final db = await LocalDb.instance; + final tables = [ + for (final r in await db.rawQuery( + "SELECT name FROM sqlite_master WHERE type = 'table' " + "AND name NOT LIKE 'sqlite_stat%' ORDER BY name")) + r['name'] as String, + ].where((t) => (only == null || only.contains(t)) && !except.contains(t)); + final out = >{}; + for (final t in tables) { + final cols = [ + for (final c in await db.rawQuery('PRAGMA table_info("$t")')) + if (!(_wallClockColumns[t]?.contains(c['name']) ?? false)) + c['name'] as String, + ]; + final rows = await db.rawQuery( + 'SELECT ${cols.map((c) => 'quote("$c")').join(" || '|' || ")} ' + 'AS r FROM "$t"'); + out[t] = [for (final r in rows) r['r'] as String]..sort(); + } + return out; + } + + // ── fixture: one source row per per-row rule ─────────────────────────────── + + /// The local state the import lands on: a finalized day, a non-finalized + /// day, and beats from two devices on one second. + Future seedLocal() async { + final db = await LocalDb.instance; + // Collision must REPLACE the 1 Hz value, including replacing it with NULL. + await db.insert('decoded_onehz', { + 'device_id': '', 'ts_ms': _t0 * 1000, 'rec_ts': _t0, + 'counter': 99, 'hr': 99, 'step_count': 123, + }); + await db.insert('day_result', { + 'day_id': '2026-09-01', 'algo_version': 100, + 'payload_json': '{"who":"local"}', 'computed_at': 1, 'finalized': 1, + }); + await db.insert('day_result', { + 'day_id': '2026-09-03', 'algo_version': 100, + 'payload_json': '{"who":"local3"}', 'computed_at': 1, 'finalized': 0, + }); + await db.insert('metric_series', + {'date': '2026-09-01', 'key': 'rhr', 'value': 50.0}); + await db.insert('metric_series_version', + {'date': '2026-09-01', 'algo_version': 100, 'source': 'band'}); + await db.insert('metric_method', { + 'date': '2026-09-01', 'key': 'rhr', + 'method': 'local', 'family': 'f', 'class': 'c', + }); + await db.insert('baselines', { + 'key': 'movement_floor', 'payload_json': '{"who":"local"}', + 'updated_at': 1, + }); + for (final dev in ['', 'band2']) { + for (var b = 0; b < 4; b++) { + await db.insert('decoded_rr', { + 'device_id': dev, 'ts_ms': _t0 * 1000, 'rec_ts': _t0, + 'beat_index': b, 'rr_ts_ms': _t0 * 1000 + b * 800, 'rr_ms': 700 + b, + }); + } + } + // A second the source never mentions: untouched. + for (var b = 0; b < 2; b++) { + await db.insert('decoded_rr', { + 'device_id': '', 'ts_ms': (_t0 + 20) * 1000, 'rec_ts': _t0 + 20, + 'beat_index': b, 'rr_ts_ms': (_t0 + 20) * 1000, 'rr_ms': 900 + b, + }); + } + // Upstream's documented restore gap: a 1 Hz second without source RR + // does NOT clear local beats. Keep that behavior in this performance PR. + await db.insert('decoded_rr', { + 'device_id': '', 'ts_ms': (_t0 + 1) * 1000, 'rec_ts': _t0 + 1, + 'beat_index': 0, 'rr_ts_ms': (_t0 + 1) * 1000, 'rr_ms': 888, + }); + await db.insert('raw_archive', { + 'device_id': '', 'hex': 'local-only', 'counter': 1, + 'packet_type': 32, 'rec_ts': _t0, 'captured_at': 1, + 'reason': 'undecodable_rec_v20', + }); + await db.insert('substrate_archive', archiveRow([archiveSecond(_t0, 65)])); + } + + /// A foreign export exercising every per-row rule. No PRIMARY KEYs and no + /// NOT NULLs, so a value of the wrong storage class can be planted. + Future buildRulesSource() async { + final path = p.join(tmp.path, 'rules_src.db'); + await databaseFactoryFfi.deleteDatabase(path); + final src = await databaseFactoryFfi.openDatabase(path); + // decoded_onehz: a column this build does not know (future_col), columns + // this build has that the source lacks (step_count etc.), legacy + // on_wrist/hr_valid values and the -50 °C sentinel. + await src.execute('''CREATE TABLE decoded_onehz ( + device_id TEXT, ts_ms INTEGER, rec_ts, counter INTEGER, hr INTEGER, + ax REAL, skin_temp_c REAL, on_wrist INTEGER, hr_valid INTEGER, + future_col TEXT)'''); + Future onehz(Object? dev, Object? tsMs, Object? recTs, int counter, + double? skin) => + src.insert('decoded_onehz', { + 'device_id': dev, 'ts_ms': tsMs, 'rec_ts': recTs, + 'counter': counter, 'hr': 60 + counter, 'ax': 0.5, + 'skin_temp_c': skin, 'on_wrist': 1, 'hr_valid': 1, + 'future_col': 'x$counter', + }); + await onehz(null, null, _t0, 1, -50.0); // sentinel -> NULL, key derived + await onehz(null, null, _t0 + 1, 2, -49.99); // just above: kept + await onehz('band2', (_t0 + 2) * 1000, _t0 + 2, 3, 33.5); // keyed: as is + await onehz(null, null, 'abc', 4, 33.0); // non-numeric rec_ts: dropped + await onehz(null, null, '${_t0 + 4}', 5, 33.0); // numeric TEXT: dropped + await onehz('', null, (_t0 + 6).toDouble(), 6, 33.0); // REAL rec_ts + await onehz(null, null, _t0 + 7, 7, -49.995); // boundary: NULL + + // decoded_rr, legacy: no rec_ts, no device_id, no ts_ms. + await src.execute('''CREATE TABLE decoded_rr ( + counter INTEGER, beat_index INTEGER, rr_ts_ms, rr_ms INTEGER)'''); + Future rr(int counter, int beat, Object? rrTsMs, int ms) => + src.insert('decoded_rr', { + 'counter': counter, 'beat_index': beat, + 'rr_ts_ms': rrTsMs, 'rr_ms': ms, + }); + await rr(1, 0, _t0 * 1000, 500); // over local primary beats 0..3 + await rr(1, 1, _t0 * 1000 + 600, 510); + await rr(9, 0, 'bad', 520); // non-numeric: dropped + await rr(11, 0, (_t0 + 10) * 1000, 530); + await rr(11, 1, (_t0 + 10) * 1000 + 700, 540); + await rr(11, 2, (_t0 + 10) * 1000 + 900, 550); + + // samples, legacy: no device_id / ts_ms. + await src.execute('CREATE TABLE samples (counter INTEGER, ts, hr INTEGER)'); + await src.insert('samples', {'counter': 1, 'ts': _t0, 'hr': 61}); + await src.insert('samples', {'counter': 2, 'ts': _t0 + 1, 'hr': 62}); + await src.insert('samples', {'counter': 3, 'ts': 'nope', 'hr': 63}); + + // day_result without window_json / skipped / partial (dest defaults). + await src.execute('''CREATE TABLE day_result ( + day_id TEXT, algo_version INTEGER, payload_json TEXT, + computed_at INTEGER, finalized INTEGER, rhr REAL, rmssd REAL, + readiness REAL)'''); + Future day(String d, int v, String who, int fin) => + src.insert('day_result', { + 'day_id': d, 'algo_version': v, 'payload_json': '{"who":"$who"}', + 'computed_at': 2, 'finalized': fin, 'rhr': 55.0, + }); + await day('2026-09-01', 100, 'src', 1); // locally finalized: protected + await day('2026-09-01', 101, 'src101', 1); // new version: written + await day('2026-09-02', 100, 'staleSrc2', 1); // duplicate: later row wins + await day('2026-09-02', 100, 'src2', 1); + await day('2026-09-03', 100, 'src3', 0); // local not finalized: replaced + + await src.execute( + 'CREATE TABLE metric_series (date TEXT, key TEXT, value REAL)'); + for (final r in [ + ['2026-09-01', 'rhr', 60.0], + ['2026-09-01', 'hrv', 40.0], + ['2026-09-02', 'rhr', 55.0], + ['2026-09-03', 'rhr', 57.0], + ]) { + await src.insert( + 'metric_series', {'date': r[0], 'key': r[1], 'value': r[2]}); + } + await src.execute('CREATE TABLE metric_series_version ' + '(date TEXT, algo_version INTEGER, source TEXT)'); + for (final d in ['2026-09-01', '2026-09-02']) { + await src.insert('metric_series_version', + {'date': d, 'algo_version': 101, 'source': 'src'}); + } + await src.execute('CREATE TABLE metric_method (date TEXT, key TEXT, ' + 'method TEXT, family TEXT, class TEXT)'); + for (final d in ['2026-09-01', '2026-09-02']) { + await src.insert('metric_method', { + 'date': d, 'key': 'rhr', 'method': 'src', 'family': 'f', 'class': 'c', + }); + } + await src.execute('CREATE TABLE baselines ' + '(key TEXT, payload_json TEXT, updated_at INTEGER)'); + for (final key in ['movement_floor', 'new_baseline']) { + await src.insert('baselines', { + 'key': key, 'payload_json': '{"who":"source"}', 'updated_at': 2, + }); + } + + // events: upstream's shape plus an unknown column. + await src.execute('''CREATE TABLE events (device_id TEXT, hex TEXT, + event_id INTEGER, ts INTEGER, captured_at INTEGER, extra TEXT)'''); + await src.insert('events', { + 'device_id': '', 'hex': 'aa01', 'event_id': 3, 'ts': _t0, + 'captured_at': _t0 * 1000, 'extra': 'zz', + }); + await src.execute('CREATE TABLE band_battery (device_id TEXT, ts INTEGER, ' + 'battery_pct REAL, source TEXT)'); + await src.insert('band_battery', { + 'device_id': '', 'ts': _t0, 'battery_pct': 70.0, 'source': 'history', + }); + await src.execute('CREATE TABLE band_events (device_id TEXT, hex TEXT, ' + 'event_id INTEGER, name TEXT, ts INTEGER, captured_at INTEGER)'); + await src.insert('band_events', { + 'device_id': '', 'hex': 'bb01', 'event_id': 3, 'name': 'test', + 'ts': _t0, 'captured_at': 2, + }); + await src.execute('CREATE TABLE raw_archive (device_id TEXT, hex TEXT, ' + 'counter INTEGER, packet_type INTEGER, rec_ts INTEGER, ' + 'captured_at INTEGER, reason TEXT)'); + for (final hex in ['src-a', 'src-b']) { + await src.insert('raw_archive', { + 'device_id': '', 'hex': hex, 'counter': 1, 'packet_type': 32, + 'rec_ts': _t0, 'captured_at': 1, 'reason': 'undecodable_rec_v20', + }); + } + // Use upstream's lossless archive, including a collision and a new second. + // A straight REPLACE of this bucket would lose the local value. + final db = await LocalDb.instance; + final ddl = (await db.rawQuery("SELECT sql FROM sqlite_master " + "WHERE type = 'table' AND name = 'substrate_archive'")).single; + await src.execute(ddl['sql'] as String); + await src.insert('substrate_archive', archiveRow([ + archiveSecond(_t0, 1), archiveSecond(_t0 + 1, 66), + ])); + await src.close(); + return path; + } + + group('equivalence law: bulk == per-row', () { + late String rulesSrc; + setUpAll(() async => rulesSrc = await buildRulesSource()); + + Future<(Map, Map>)> run( + Future> Function(String) merge) async { + await freshLocal(); + await seedLocal(); + final counts = await merge(rulesSrc); + return (counts, await dump()); + } + + test('every table ends byte-identical, and the counts agree', () async { + final (oracleCounts, oracle) = await run(LocalDb.debugMergeRowByRow); + final (bulkCounts, bulk) = await run(LocalDb.importFromDbFile); + for (final t in oracle.keys) { + expect(bulk[t], oracle[t], reason: 'table $t'); + } + expect(bulk.keys.toSet(), oracle.keys.toSet()); + expect(bulkCounts, oracleCounts); + }); + + group('each rule, by hand', () { + late Map counts; + late Database db; + setUpAll(() async { + db = await freshLocal(); + await seedLocal(); + counts = await LocalDb.importFromDbFile(rulesSrc); + }); + + test('decoded_onehz: retired columns NULL, sentinel NULL, keys derived, ' + 'non-numeric rec_ts dropped, unknown column dropped', () async { + final rows = await db.rawQuery( + 'SELECT device_id, ts_ms, rec_ts, counter, skin_temp_c, on_wrist, ' + 'hr_valid, step_count FROM decoded_onehz ORDER BY device_id, ts_ms'); + expect([for (final r in rows) r['counter']], [1, 2, 6, 7, 3]); + final by = {for (final r in rows) r['counter']: r}; + expect(by[1]!['device_id'], ''); + expect(by[1]!['ts_ms'], _t0 * 1000); + expect(by[1]!['skin_temp_c'], isNull); + expect(by[2]!['skin_temp_c'], -49.99); + expect(by[7]!['skin_temp_c'], isNull); + expect(by[3]!['device_id'], 'band2'); + expect(by[3]!['ts_ms'], (_t0 + 2) * 1000); + expect(by[6]!['ts_ms'], (_t0 + 6) * 1000); + for (final r in rows) { + expect(r['on_wrist'], isNull); + expect(r['hr_valid'], isNull); + expect(r['step_count'], isNull, reason: 'missing column: default'); + } + expect(counts['decoded_onehz'], 5); + }); + + test('decoded_rr: rec_ts from rr_ts_ms, bad row dropped, stale local ' + 'tail cleared for the writing device only', () async { + final rows = await db.rawQuery( + 'SELECT device_id, ts_ms, rec_ts, beat_index, rr_ms FROM decoded_rr ' + 'ORDER BY device_id, ts_ms, beat_index'); + expect([ + for (final r in rows) + '${r['device_id']}/${r['ts_ms']}/${r['beat_index']}/${r['rr_ms']}' + ], [ + '/${_t0 * 1000}/0/500', + '/${_t0 * 1000}/1/510', + '/${(_t0 + 1) * 1000}/0/888', + '/${(_t0 + 10) * 1000}/0/530', + '/${(_t0 + 10) * 1000}/1/540', + '/${(_t0 + 10) * 1000}/2/550', + '/${(_t0 + 20) * 1000}/0/900', + '/${(_t0 + 20) * 1000}/1/901', + 'band2/${_t0 * 1000}/0/700', + 'band2/${_t0 * 1000}/1/701', + 'band2/${_t0 * 1000}/2/702', + 'band2/${_t0 * 1000}/3/703', + ]); + expect(rows.firstWhere((r) => r['ts_ms'] == (_t0 + 10) * 1000)['rec_ts'], + _t0 + 10); + expect(counts['decoded_rr'], 5); + }); + + test('samples: legacy rows keyed to the primary band at ts*1000', () async { + final rows = await db.rawQuery( + 'SELECT device_id, ts_ms, ts FROM samples ORDER BY ts_ms'); + expect([for (final r in rows) '${r['device_id']}/${r['ts_ms']}'], + ['/${_t0 * 1000}', '/${(_t0 + 1) * 1000}']); + }); + + test('1 Hz without source beats retains local RR, the upstream restore ' + 'gap; the replacement 1 Hz value still wins', () async { + final rows = await db.query('decoded_onehz', + where: 'device_id = ? AND ts_ms = ?', + whereArgs: ['', _t0 * 1000]); + expect(rows.single['hr'], 61); + expect(rows.single['step_count'], isNull); + final beats = await db.query('decoded_rr', + where: 'device_id = ? AND ts_ms = ?', + whereArgs: ['', (_t0 + 1) * 1000]); + expect(beats.single['rr_ms'], 888); + }); + + test('day_result: a finalized local (day, version) is never overwritten; ' + '_days counts distinct days written', () async { + final rows = await db.rawQuery( + 'SELECT day_id, algo_version, payload_json, window_json, skipped, ' + 'partial FROM day_result ORDER BY day_id, algo_version'); + expect([ + for (final r in rows) + '${r['day_id']}/${r['algo_version']}/${r['payload_json']}' + ], [ + '2026-09-01/100/{"who":"local"}', + '2026-09-01/101/{"who":"src101"}', + '2026-09-02/100/{"who":"src2"}', + '2026-09-03/100/{"who":"src3"}', + ]); + final imported = rows.where((r) => r['payload_json'] != '{"who":"local"}'); + for (final r in imported) { + expect(r['window_json'], '{}', reason: 'dest default'); + expect(r['skipped'], 0); + expect(r['partial'], 0); + } + expect(counts['_days'], 3); + expect(counts['day_result'], 4); + }); + + test('metric_series / _version / metric_method skip a locally ' + 'finalized day', () async { + final series = await db.rawQuery( + 'SELECT date, key, value FROM metric_series ORDER BY date, key'); + expect([for (final r in series) '${r['date']}/${r['key']}/${r['value']}'], + [ + '2026-09-01/rhr/50.0', + '2026-09-02/rhr/55.0', + '2026-09-03/rhr/57.0', + ]); + final ver = await db.rawQuery( + 'SELECT date, algo_version FROM metric_series_version ORDER BY date'); + expect([for (final r in ver) '${r['date']}/${r['algo_version']}'], + ['2026-09-01/100', '2026-09-02/101']); + final method = await db.rawQuery( + 'SELECT date, method FROM metric_method ORDER BY date'); + expect([for (final r in method) '${r['date']}/${r['method']}'], + ['2026-09-01/local', '2026-09-02/src']); + }); + + test('a table the export lacks is skipped; an unknown column is dropped', + () async { + expect(counts.containsKey('external_hr'), isFalse); + final ev = await db.rawQuery('SELECT * FROM events'); + expect(ev, hasLength(1)); + expect(ev.single.containsKey('extra'), isFalse); + }); + + test('finalized history keeps local baselines and admits missing keys', + () async { + final rows = await db.query('baselines', orderBy: 'key'); + expect([for (final r in rows) '${r['key']}/${r['payload_json']}'], [ + 'movement_floor/{"who":"local"}', 'new_baseline/{"who":"source"}', + ]); + }); + + test('raw archive keeps every frame; substrate archive merges losslessly ' + 'with local winning collisions', () async { + final raw = await db.query('raw_archive', orderBy: 'hex'); + expect([for (final r in raw) r['hex']], ['local-only', 'src-a', 'src-b']); + expect(counts['raw_archive'], 2); + final row = (await db.query('substrate_archive')).single; + final bucket = SubstrateArchiveCodec.decode(row['blob'] as Uint8List)!; + expect(bucket.onehz.rowCount, 2); + expect([for (var i = 0; i < 2; i++) bucket.onehz.valueAt('hr', i)], + [65, 66]); + expect(counts['substrate_archive'], 1); + }); + }); + + test('a corrupt source rethrows (a restore is not tolerant)', () async { + final path = await buildCorruptSource(tmp.path); + await freshLocal(); + await expectLater( + LocalDb.importFromDbFile(path), throwsA(isA())); + }); + }); + + group('25k boundary equivalence', () { + late String boundarySrc; + const n = 25001; + setUpAll(() async { + boundarySrc = p.join(tmp.path, 'boundary_src.db'); + final src = await databaseFactoryFfi.openDatabase(boundarySrc); + // No PK: rows 1 and 25001 derive the same 1 Hz key. Give the last + // row an explicit, sparse rowid so range paging must cross a gap too. + await src.execute('CREATE TABLE decoded_onehz ' + '(rec_ts INTEGER, counter INTEGER, hr INTEGER)'); + await src.execute(''' + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n + WHERE i < $n) + INSERT INTO decoded_onehz(rowid, rec_ts, counter, hr) + SELECT CASE WHEN i = $n THEN 50001 ELSE i END, + CASE WHEN i IN (1, $n) THEN $_t0 ELSE $_t0 + i END, + i, CASE WHEN i = $n THEN 87 ELSE 60 END FROM n'''); + await src.execute('''CREATE TABLE decoded_rr ( + device_id TEXT, ts_ms INTEGER, rec_ts INTEGER, beat_index INTEGER, + rr_ts_ms INTEGER, rr_ms INTEGER)'''); + // Higher beat first at rowid 24999, lower beat last at rowid 25001. + // Both fall in the SAME upstream 2k page, but DIFFERENT bulk chunks. + // Thus the oracle keeps both, while a per-25k-chunk tail max loses 1. + await src.execute(''' + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n + WHERE i < $n) + INSERT INTO decoded_rr + SELECT '', + (CASE WHEN i IN (24999, 25001) THEN $_t0 ELSE $_t0 + i END) * 1000, + CASE WHEN i IN (24999, 25001) THEN $_t0 ELSE $_t0 + i END, + CASE WHEN i = 24999 THEN 1 ELSE 0 END, + ($_t0 + i) * 1000, 500 + i % 100 FROM n'''); + await src.close(); + }); + + test('later source row wins across chunks and rowid gaps; whole-source ' + 'beat maximum keeps an earlier higher index', () async { + Future<(Map, Map>)> run( + Future> Function(String) merge) async { + await freshLocal(); + await seedLocal(); + final counts = await merge(boundarySrc); + return (counts, await dump()); + } + final (refCounts, ref) = await run(LocalDb.debugMergeRowByRow); + final (counts, actual) = await run(LocalDb.importFromDbFile); + expect(counts, refCounts); + expect(actual, ref); + final db = await LocalDb.instance; + final onehz = await db.query('decoded_onehz', + where: 'device_id = ? AND ts_ms = ?', whereArgs: ['', _t0 * 1000]); + expect(onehz.single['hr'], 87, reason: 'last source row wins'); + final rr = await db.query('decoded_rr', + where: 'device_id = ? AND ts_ms = ?', whereArgs: ['', _t0 * 1000], + orderBy: 'beat_index'); + expect([for (final r in rr) r['beat_index']], [0, 1]); + expect([for (final r in rr) r['rr_ms']], [501, 599]); + expect(await db.query('decoded_rr', where: 'device_id = ?', + whereArgs: ['band2']), hasLength(4)); + }, timeout: const Timeout(Duration(minutes: 5))); + + test('whole-source RR max preparation is paged at 25k rows', () async { + await freshLocal(); + _invokes.clear(); + _counting = true; + try { + await LocalDb.importFromDbFile(boundarySrc); + } finally { + _counting = false; + } + final prep = [ + for (final e in _invokes) + for (final op in _operations(e.arguments)) + if (RegExp(r'INSERT[\s\S]*\btemp\.[\s\S]*\bMAX\s*\(' + r'[\s\S]*\bdecoded_rr\b', caseSensitive: false) + .hasMatch(op['sql'] as String)) op, + ]; + expect(prep, hasLength(2), + reason: 'whole-source beat maxima need two bounded scans for 25001 rows'); + for (var i = 0; i < prep.length; i++) { + final sql = prep[i]['sql'] as String; + expect(sql, matches(RegExp(r'rowid\s*>\s*\?', caseSensitive: false))); + expect(sql, matches(RegExp(r'rowid\s*<=\s*\?', caseSensitive: false))); + final args = prep[i]['arguments'] as List; + expect(args, hasLength(2)); + expect(args[0], i * 25000); + expect(args[1], inInclusiveRange(i == 0 ? 25000 : n, (i + 1) * 25000)); + } + }, timeout: const Timeout(Duration(minutes: 5))); + }); + + test("bulk keeps a second's higher beats that a later source page omits " + '(row-by-row drops them)', () async { + final path = p.join(tmp.path, 'rr_page_tail_src.db'); + final src = await databaseFactoryFfi.openDatabase(path); + // No PK: repeated lower indices must get later rowids while the + // earlier higher beats remain in the source's union. + await src.execute('''CREATE TABLE decoded_rr ( + device_id TEXT, ts_ms INTEGER, rec_ts INTEGER, beat_index INTEGER, + rr_ts_ms INTEGER, rr_ms INTEGER)'''); + Future beat(int b) => src.insert('decoded_rr', { + 'device_id': '', 'ts_ms': _t0 * 1000, 'rec_ts': _t0, + 'beat_index': b, 'rr_ts_ms': _t0 * 1000 + b * 700, 'rr_ms': 700 + b, + }); + for (var b = 0; b < 6; b++) { + await beat(b); + } + await src.execute(''' + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n + WHERE i < 2001) + INSERT INTO decoded_rr + SELECT '', ($_t0 + i) * 1000, $_t0 + i, 0, + ($_t0 + i) * 1000, 800 FROM n'''); + // Rowids 2008 and 2009 land on the oracle's second 2k page. + for (var b = 0; b < 2; b++) { + await beat(b); + } + final sourceBeats = await src.rawQuery( + 'SELECT DISTINCT beat_index, rr_ts_ms, rr_ms FROM decoded_rr ' + 'WHERE device_id = ? AND ts_ms = ? ORDER BY beat_index', + ['', _t0 * 1000]); + await src.close(); + expect([for (final r in sourceBeats) r['beat_index']], [0, 1, 2, 3, 4, 5]); + + Future>> restoredBeats(Database db) => + db.query('decoded_rr', columns: ['beat_index', 'rr_ts_ms', 'rr_ms'], + where: 'device_id = ? AND ts_ms = ?', + whereArgs: ['', _t0 * 1000], orderBy: 'beat_index'); + + final bulkDb = await freshLocal(); + await LocalDb.importFromDbFile(path); + final bulk = await restoredBeats(bulkDb); + expect([for (final r in bulk) r['beat_index']], [0, 1, 2, 3, 4, 5]); + expect(bulk, sourceBeats, reason: 'bulk retains the whole-source union'); + + final oracleDb = await freshLocal(); + await LocalDb.debugMergeRowByRow(path); + final oracle = await restoredBeats(oracleDb); + expect([for (final r in oracle) r['beat_index']], [0, 1], + reason: 'the later page maximum of 1 deletes earlier beats 2..5'); + expect(oracle, sourceBeats.take(2).toList()); + }); + + test('RR with a supplied key but no usable record time is dropped', () async { + final path = p.join(tmp.path, 'invalid_rr_src.db'); + final src = await databaseFactoryFfi.openDatabase(path); + await src.execute('CREATE TABLE decoded_rr (device_id TEXT, ts_ms INTEGER, ' + 'rec_ts, beat_index INTEGER, rr_ts_ms, rr_ms INTEGER)'); + await src.insert('decoded_rr', { + 'device_id': '', 'ts_ms': _t0 * 1000, 'rec_ts': null, + 'beat_index': 0, 'rr_ts_ms': 'bad', 'rr_ms': 700, + }); + await src.close(); + await freshLocal(); + final refCounts = await LocalDb.debugMergeRowByRow(path); + final ref = await dump(); + await freshLocal(); + final counts = await LocalDb.importFromDbFile(path); + expect(counts, refCounts); + expect(await dump(), ref); + expect(counts['decoded_rr'], 0); + expect(await (await LocalDb.instance).query('decoded_rr'), isEmpty); + }); + + test('plain tables use ATTACH and INSERT SELECT on the public restore path', + () async { + final src = await buildRulesSource(); + await freshLocal(); + _invokes.clear(); + _counting = true; + try { + await LocalDb.importFromDbFile(src); + } finally { + _counting = false; + } + final sql = [for (final e in _invokes) ..._sqlOf(e.arguments)]; + expect(sql.any((s) => RegExp(r'^ATTACH\b', caseSensitive: false) + .hasMatch(s.trimLeft())), isTrue, + reason: 'bulk restore must attach the source on the destination handle'); + for (final t in [ + 'decoded_onehz', 'decoded_rr', 'samples', 'raw_archive', 'events', + 'band_events', 'band_battery', 'day_result', 'metric_series', + 'metric_series_version', 'metric_method', + ]) { + expect(sql.any((s) => RegExp( + 'INSERT\\s+OR\\s+REPLACE\\s+INTO\\s+(?:main\\.)?["`]?' + '$t\\b[\\s\\S]*\\bSELECT\\b', caseSensitive: false).hasMatch(s)), + isTrue, reason: '$t must copy inside SQLite'); + } + }); + + // ── volume: sqflite calls per table, chunk commits ──────────────────────── + + group('200k decoded_onehz + 100k decoded_rr', () { + const nOneHz = 200000, nRr = 100000; + late String bigSrc; + + setUpAll(() async { + bigSrc = p.join(tmp.path, 'big_src.db'); + await databaseFactoryFfi.deleteDatabase(bigSrc); + final src = await databaseFactoryFfi.openDatabase(bigSrc); + await src.execute('''CREATE TABLE decoded_onehz ( + device_id TEXT NOT NULL DEFAULT '', ts_ms INTEGER NOT NULL DEFAULT 0, + rec_ts INTEGER, counter INTEGER NOT NULL, hr INTEGER, ax REAL, ay REAL, + az REAL, PRIMARY KEY (device_id, ts_ms))'''); + await src.execute('''CREATE TABLE decoded_rr ( + device_id TEXT NOT NULL DEFAULT '', ts_ms INTEGER NOT NULL DEFAULT 0, + rec_ts INTEGER NOT NULL, beat_index INTEGER NOT NULL, + rr_ts_ms INTEGER NOT NULL, rr_ms INTEGER NOT NULL, + PRIMARY KEY (device_id, ts_ms, beat_index))'''); + await src.execute(''' + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n + WHERE i < $nOneHz) + INSERT INTO decoded_onehz + SELECT '', ($_t0 + i) * 1000, $_t0 + i, i, 40 + i % 100, 0.0, 0.0, 1.0 + FROM n'''); + await src.execute(''' + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n + WHERE i < ${nRr ~/ 2}), + b(k) AS (VALUES (0), (1)) + INSERT INTO decoded_rr + SELECT '', ($_t0 + i) * 1000, $_t0 + i, k, + ($_t0 + i) * 1000 + k * 700, 700 + i % 50 + FROM n, b ORDER BY i, k'''); + await src.close(); + }); + + test('a bounded number of sqflite calls per table, not one per page', + () async { + await freshLocal(); + _invokes.clear(); + _counting = true; + final Map counts; + try { + counts = await LocalDb.importFromDbFile(bigSrc); + } finally { + _counting = false; + } + expect(counts['decoded_onehz'], nOneHz); + expect(counts['decoded_rr'], nRr); + final onehz = _callsTouching('decoded_onehz'); + final rr = _callsTouching('decoded_rr'); + // Allow two calls per 25k 1 Hz chunk, five per RR chunk (including + // the paged whole-source max build and tail sweep), plus preparation. + // Upstream uses about two calls per 2k rows and exceeds both bounds. + expect(onehz, lessThanOrEqualTo(2 * (nOneHz ~/ 25000) + 10), + reason: 'decoded_onehz: $onehz sqflite calls'); + expect(rr, lessThanOrEqualTo(5 * (nRr ~/ 25000) + 15), + reason: 'decoded_rr: $rr sqflite calls'); + }, timeout: const Timeout(Duration(minutes: 10))); + + test('a failure in a later chunk leaves earlier chunks committed, and a ' + 'rerun converges to the clean result', () async { + const data = {'decoded_onehz', 'decoded_rr'}; + await freshLocal(); + await LocalDb.debugMergeRowByRow(bigSrc); + final clean = await dump(only: data); + + final db = await freshLocal(); + // The LAST source row fails, whatever the chunk size. + await db.execute('CREATE TRIGGER inject_fail BEFORE INSERT ON ' + 'decoded_onehz WHEN NEW.counter = $nOneHz ' + "BEGIN SELECT RAISE(ABORT, 'injected'); END"); + await expectLater(LocalDb.importFromDbFile(bigSrc), throwsA(anything)); + final span = (await db.rawQuery( + 'SELECT COUNT(*) AS n, MAX(counter) AS hi FROM decoded_onehz')) + .single; + final landed = span['n'] as int; + final maxCounter = span['hi'] as int?; + expect(landed, greaterThan(0), reason: 'chunk 1 committed'); + expect(landed, lessThan(nOneHz), reason: 'the failing chunk rolled back'); + expect(maxCounter, landed, reason: 'a contiguous rowid prefix'); + + await db.execute('DROP TRIGGER inject_fail'); + await LocalDb.importFromDbFile(bigSrc); + expect(await dump(only: data), clean); + }, timeout: const Timeout(Duration(minutes: 15))); + + test('bulk commits 25000 rows before a failure in the next chunk', () async { + final db = await freshLocal(); + await db.execute('CREATE TRIGGER inject_fail BEFORE INSERT ON ' + 'decoded_onehz WHEN NEW.counter = 25001 ' + "BEGIN SELECT RAISE(ABORT, 'injected-25001'); END"); + try { + await expectLater(LocalDb.importFromDbFile(bigSrc), throwsA( + isA().having((e) => e.toString(), 'message', + contains('injected-25001')))); + final span = (await db.rawQuery('SELECT COUNT(*) AS n, ' + 'MAX(counter) AS hi FROM decoded_onehz')).single; + expect(span, {'n': 25000, 'hi': 25000}, + reason: 'exactly one 25k chunk must remain committed'); + } finally { + await db.execute('DROP TRIGGER inject_fail'); + } + }, timeout: const Timeout(Duration(minutes: 5))); + }); + + // ── ATTACH hygiene ───────────────────────────────────────────────────────── + + group('the source is detached', () { + Future expectClean() async { + final db = await LocalDb.instance; + expect([ + for (final r in await db.rawQuery('PRAGMA database_list')) + if (r['name'] != 'temp') r['name'] as String, + ], ['main']); + expect(await db.rawQuery("SELECT name FROM sqlite_temp_master " + "WHERE type = 'table'"), isEmpty, + reason: 'restore preparation and carried-beat tables must be dropped'); + // Exercise the same handle after cleanup. A leaked transaction or + // attachment must not prevent the next band-sync write. + await db.transaction((txn) => txn.rawQuery('SELECT 1')); + } + + test('after a successful restore', () async { + final src = await buildRulesSource(); + await freshLocal(); + await LocalDb.importFromDbFile(src); + await expectClean(); + }); + + test('after a restore that threw mid-copy', () async { + final src = await buildRulesSource(); + final db = await freshLocal(); + await db.execute('CREATE TRIGGER inject_fail BEFORE INSERT ON ' + "decoded_onehz BEGIN SELECT RAISE(ABORT, 'injected'); END"); + await expectLater(LocalDb.importFromDbFile(src), throwsA( + isA().having((e) => e.toString(), 'message', + contains('injected')))); + await db.execute('DROP TRIGGER inject_fail'); + await expectClean(); + await LocalDb.importFromDbFile(src); + await expectClean(); + }); + + for (final table in ['decoded_rr', 'day_result']) { + test('after $table preparation and a failed insert', () async { + final src = await buildRulesSource(); + final db = await freshLocal(); + await db.execute('CREATE TRIGGER inject_fail BEFORE INSERT ON ' + "$table BEGIN SELECT RAISE(ABORT, 'injected-$table'); END"); + await expectLater(LocalDb.importFromDbFile(src), throwsA( + isA().having((e) => e.toString(), 'message', + contains('injected-$table')))); + await db.execute('DROP TRIGGER inject_fail'); + await expectClean(); + await LocalDb.importFromDbFile(src); + await expectClean(); + }); + } + + test('after a corrupt source', () async { + final path = await buildCorruptSource(tmp.path); + await freshLocal(); + await expectLater(LocalDb.importFromDbFile(path), throwsA(anything)); + await expectClean(); + }); + }); +} + +/// An export whose header and schema read fine but whose decoded_onehz pages +/// are garbage: opening works, reading the table fails SQLITE_CORRUPT. +Future buildCorruptSource(String dir) async { + final path = p.join(dir, 'corrupt_src.db'); + await databaseFactoryFfi.deleteDatabase(path); + final src = await databaseFactoryFfi.openDatabase(path); + await src.execute('''CREATE TABLE decoded_onehz ( + device_id TEXT NOT NULL DEFAULT '', ts_ms INTEGER NOT NULL DEFAULT 0, + rec_ts INTEGER, counter INTEGER NOT NULL, hr INTEGER, pad TEXT, + PRIMARY KEY (device_id, ts_ms))'''); + await src.execute(''' + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n + WHERE i < 4000) + INSERT INTO decoded_onehz + SELECT '', ($_t0 + i) * 1000, $_t0 + i, i, 60, hex(randomblob(40)) FROM n'''); + await src.close(); + final bytes = File(path).readAsBytesSync(); + bytes.fillRange(4096 * 3, bytes.length, 0xA5); + File(path).writeAsBytesSync(bytes, flush: true); + return path; +} diff --git a/test/import_restore_perf_probe_test.dart b/test/import_restore_perf_probe_test.dart new file mode 100644 index 000000000..e130806e8 --- /dev/null +++ b/test/import_restore_perf_probe_test.dart @@ -0,0 +1,113 @@ +// Optional timing probe: restores a supplied export into an empty FFI DB, prints +// per-table wall ms, sqflite calls and rows. Desktop ffi has no platform +// channel, so it understates Android. +// +// IMPORT_PERF_PROBE=1 IMPORT_PERF_SRC=/path/export.db \ +// flutter test test/import_restore_perf_probe_test.dart +// The supplied file is only read; restore runs against a temporary copy. + +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:sqflite_common/sqflite_logger.dart'; +import 'package:sqflite_common_ffi/sqflite_ffi.dart'; +import 'package:openstrap_edge/data/db.dart'; + +Iterable _sqlOf(Object? args) sync* { + if (args is Map) { + final sql = args['sql']; + if (sql is String) yield sql; + for (final v in args.values) { + if (v is List || v is Map) yield* _sqlOf(v); + } + } else if (args is List) { + for (final v in args) { + yield* _sqlOf(v); + } + } +} + +void main() { + final srcPath = Platform.environment['IMPORT_PERF_SRC'] ?? ''; + + test('restore timing per table', () async { + final tables = LocalDb.restoreTablesForTest; + final patterns = {for (final t in tables) t: RegExp('\\b$t\\b')}; + final wallUs = {}; + final calls = {}; + final clock = Stopwatch(); + var lastUs = 0; + var on = false; + var current = '(other)'; + + sqfliteFfiInit(); + // The logger is the only public hook that sees every sqflite call. + // ignore: experimental_member_use + databaseFactory = SqfliteDatabaseFactoryLogger( + databaseFactoryFfi, + options: SqfliteLoggerOptions( + type: SqfliteDatabaseFactoryLoggerType.invoke, + log: (e) { + if (!on || e is! SqfliteLoggerInvokeEvent) return; + final now = clock.elapsedMicroseconds; + final sql = _sqlOf(e.arguments).join('\n'); + String? owner; + for (final t in tables) { + if (patterns[t]!.hasMatch(sql)) { + owner = t; + break; + } + } + // BEGIN/COMMIT and other unnamed calls belong to the table being + // copied: the Dart time before them is that table's row work. + owner ??= current; + current = owner; + wallUs[owner] = (wallUs[owner] ?? 0) + (now - lastUs); + calls[owner] = (calls[owner] ?? 0) + 1; + lastUs = now; + }, + ), + ); + SharedPreferences.setMockInitialValues({}); + final tmp = await Directory.systemTemp.createTemp('openstrap_perf_'); + LocalDb.dbName = 'openstrap_perf_probe.db'; + final dbDir = await databaseFactory.getDatabasesPath(); + await databaseFactory.deleteDatabase(p.join(dbDir, LocalDb.dbName)); + try { + final copy = p.join(tmp.path, 'export.db'); + await File(srcPath).copy(copy); + await LocalDb.instance; + + on = true; + clock.start(); + final counts = await LocalDb.importFromDbFile(copy); + clock.stop(); + on = false; + + final rows = []; + var totalRows = 0; + for (final t in [...tables, '(other)']) { + final n = counts[t] ?? 0; + if (n > 0) totalRows += n; + final ms = (wallUs[t] ?? 0) / 1000; + if (n == 0 && ms < 5) continue; + rows.add('${t.padRight(26)} ${ms.toStringAsFixed(0).padLeft(8)} ms ' + '${'${calls[t] ?? 0}'.padLeft(7)} calls ${'$n'.padLeft(9)} rows'); + } + // ignore: avoid_print + print('[import-perf] total ${clock.elapsedMilliseconds} ms, ' + '$totalRows rows, _days=${counts['_days']}\n${rows.join('\n')}'); + } finally { + on = false; + await LocalDb.close(); + await databaseFactory.deleteDatabase(p.join(dbDir, LocalDb.dbName)); + await tmp.delete(recursive: true); + } + }, + skip: Platform.environment['IMPORT_PERF_PROBE'] != '1' || srcPath.isEmpty + ? 'set IMPORT_PERF_PROBE=1 and IMPORT_PERF_SRC to measure a restore' + : false, + timeout: const Timeout(Duration(minutes: 30))); +} From 917839d324200544b2b52cd4ab3c1a984452f668 Mon Sep 17 00:00:00 2001 From: MikenRafter Date: Sat, 24 Oct 2026 00:13:14 -0600 Subject: [PATCH 3/5] fix(import): page bulk restore chunks by row count; group RR maxima by ordinal A pre-v47 export keys decoded_onehz by rec_ts INTEGER PRIMARY KEY, so its rowids sit near 1.8e9 and fixed 25k rowid ranges walked ~70k mostly empty transactions. Chunk ends now come from the rows that exist. GROUP BY d, k resolved to a source column named d or k before the select alias, so a source with such a column got wrong per-second beat maxima. Group by ordinal instead. --- lib/data/db.dart | 25 ++++++++-- test/import_bulk_restore_test.dart | 74 ++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 5 deletions(-) diff --git a/lib/data/db.dart b/lib/data/db.dart index 2b3961b8e..08fe72501 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -10314,6 +10314,20 @@ class LocalDb { var failed = true; var copied = 0; try { + // Chunk ends by row count, not fixed rowid ranges: a legacy + // `rec_ts INTEGER PRIMARY KEY` table has rowids near 1.8e9, which + // fixed 25k ranges walked as ~70k mostly empty transactions. + final ends = []; + for (var lo = 0;;) { + final hi = Sqflite.firstIntValue(await db.rawQuery( + 'SELECT MAX(r) FROM (SELECT rowid AS r FROM imp.${quoteSql(t)} ' + 'WHERE rowid > ? ORDER BY rowid LIMIT $chunkRows)', + [lo], + )); + if (hi == null) break; + ends.add(hi); + lo = hi; + } if (t == 'decoded_rr') { // The highest beat per second over the WHOLE source, not one chunk: // REPLACE renumbers rowids, so one second's beats can straddle @@ -10326,9 +10340,11 @@ class LocalDb { (d TEXT, k INTEGER, m INTEGER, PRIMARY KEY (d, k))'''); // Paged like the copy below, so a band-sync commit can land between // pages instead of waiting out a scan of the whole source. - for (var lo = 0; lo < maxRowid; lo += chunkRows) { + for (var i = 0; i < ends.length; i++) { // REPLACE with a LEFT JOIN rather than an UPSERT: Android 8-10 // ship a system SQLite older than ON CONFLICT DO UPDATE. + // GROUP BY ordinals, not d/k: a source column of the same name + // would win over the alias. await db.execute('''INSERT OR REPLACE INTO temp._restore_rr_src_max (d, k, m) SELECT c.d, c.k, MAX(c.m, COALESCE(x.m, c.m)) @@ -10337,14 +10353,13 @@ class LocalDb { FROM imp.${quoteSql(t)} s WHERE s.rowid > ? AND s.rowid <= ? AND $unbounded AND ${numericSql(sourceColumn('beat_index'))} - GROUP BY d, k) c + GROUP BY 1, 2) c LEFT JOIN temp._restore_rr_src_max x ON x.d = c.d AND x.k = c.k''', - [lo, lo + chunkRows]); + [i == 0 ? 0 : ends[i - 1], ends[i]]); } } var lower = 0; - while (lower < maxRowid) { - final upper = lower + chunkRows > maxRowid ? maxRowid : lower + chunkRows; + for (final upper in ends) { final first = lower == 0; await db.transaction((txn) async { if (t == 'day_result' && first) { diff --git a/test/import_bulk_restore_test.dart b/test/import_bulk_restore_test.dart index f01873820..da460c7c6 100644 --- a/test/import_bulk_restore_test.dart +++ b/test/import_bulk_restore_test.dart @@ -702,6 +702,80 @@ void main() { expect(await (await LocalDb.instance).query('decoded_rr'), isEmpty); }); + test('a source RR column named d or k does not change the beat maxima', + () async { + final path = p.join(tmp.path, 'rr_alias_column_src.db'); + await databaseFactoryFfi.deleteDatabase(path); + final src = await databaseFactoryFfi.openDatabase(path); + // GROUP BY prefers a source column over a result alias of the same name. + await src.execute('''CREATE TABLE decoded_rr ( + device_id TEXT, ts_ms INTEGER, rec_ts INTEGER, beat_index INTEGER, + rr_ts_ms INTEGER, rr_ms INTEGER, d TEXT, k INTEGER)'''); + for (var b = 0; b < 4; b++) { + await src.insert('decoded_rr', { + 'device_id': '', 'ts_ms': _t0 * 1000, 'rec_ts': _t0, + 'beat_index': b, 'rr_ts_ms': _t0 * 1000 + b * 700, 'rr_ms': 700 + b, + 'd': b == 3 ? 'a' : 'b', 'k': 3 - b, + }); + } + await src.close(); + await freshLocal(); + final refCounts = await LocalDb.debugMergeRowByRow(path); + final ref = await dump(); + final db = await freshLocal(); + final counts = await LocalDb.importFromDbFile(path); + expect(counts, refCounts); + expect(await dump(), ref); + expect([for (final r in await db.query('decoded_rr', orderBy: 'beat_index')) + r['beat_index']], [0, 1, 2, 3]); + }); + + test('sparse source rowids are paged by row, not by rowid range', () async { + final path = p.join(tmp.path, 'sparse_rowid_src.db'); + await databaseFactoryFfi.deleteDatabase(path); + final src = await databaseFactoryFfi.openDatabase(path); + // A legacy `rec_ts INTEGER PRIMARY KEY` table has rowids near 1.79e9. + // A smaller gap keeps the red run short: 40 empty 25k ranges. + await src.execute('CREATE TABLE decoded_onehz ' + '(rec_ts INTEGER PRIMARY KEY, counter INTEGER, hr INTEGER)'); + await src.execute('''CREATE TABLE decoded_rr ( + device_id TEXT, ts_ms INTEGER, rec_ts INTEGER, beat_index INTEGER, + rr_ts_ms INTEGER, rr_ms INTEGER)'''); + for (final rowid in [1, 1000001]) { + await src.insert('decoded_onehz', + {'rec_ts': rowid, 'counter': rowid, 'hr': 60}); + await src.insert('decoded_rr', { + 'rowid': rowid, 'device_id': '', 'ts_ms': (_t0 + rowid) * 1000, + 'rec_ts': _t0 + rowid, 'beat_index': 0, + 'rr_ts_ms': (_t0 + rowid) * 1000, 'rr_ms': 700, + }); + } + await src.close(); + await freshLocal(); + final refCounts = await LocalDb.debugMergeRowByRow(path); + final ref = await dump(); + await freshLocal(); + _invokes.clear(); + _counting = true; + final Map counts; + try { + counts = await LocalDb.importFromDbFile(path); + } finally { + _counting = false; + } + expect(counts, refCounts); + expect(await dump(), ref); + for (final t in ['decoded_onehz', 'decoded_rr']) { + final inserts = [ + for (final e in _invokes) + for (final sql in _sqlOf(e.arguments)) + if (sql.startsWith('INSERT OR REPLACE INTO main."$t"')) sql, + ]; + expect(inserts, hasLength(1), reason: '$t: 2 rows are one chunk'); + } + expect(_callsTouching('decoded_rr'), lessThanOrEqualTo(15)); + }); + test('plain tables use ATTACH and INSERT SELECT on the public restore path', () async { final src = await buildRulesSource(); From 2ad19fe6ab34b52844c03aa53ce7c0141be823bb Mon Sep 17 00:00:00 2001 From: abdulsaheel Date: Fri, 30 Oct 2026 14:18:30 +0530 Subject: [PATCH 4/5] leave raw_archive out of the bulk copy it has its own merge that keeps the row already on the device, and that runs before the bulk path is ever reached, so listing it here only made the attach test expect a copy that never happens. --- lib/data/db.dart | 3 ++- test/import_bulk_restore_test.dart | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/lib/data/db.dart b/lib/data/db.dart index 6525a8ecb..bd00c4b46 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -11274,7 +11274,8 @@ class LocalDb { 'decoded_onehz', 'decoded_rr', 'samples', - 'raw_archive', + // Not raw_archive: it has its own merge (_mergeRawArchive), which keeps + // the row already on this device instead of replacing it. 'events', 'band_events', 'band_battery', diff --git a/test/import_bulk_restore_test.dart b/test/import_bulk_restore_test.dart index da460c7c6..8a5d88bf9 100644 --- a/test/import_bulk_restore_test.dart +++ b/test/import_bulk_restore_test.dart @@ -792,7 +792,7 @@ void main() { .hasMatch(s.trimLeft())), isTrue, reason: 'bulk restore must attach the source on the destination handle'); for (final t in [ - 'decoded_onehz', 'decoded_rr', 'samples', 'raw_archive', 'events', + 'decoded_onehz', 'decoded_rr', 'samples', 'events', 'band_events', 'band_battery', 'day_result', 'metric_series', 'metric_series_version', 'metric_method', ]) { From 2de6204cf259ce7a29ba6cf1b774501be9a0e6be Mon Sep 17 00:00:00 2001 From: abdulsaheel Date: Sat, 31 Oct 2026 22:09:53 +0530 Subject: [PATCH 5/5] point the restore tests at the archive view's new name --- test/import_bulk_restore_test.dart | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/import_bulk_restore_test.dart b/test/import_bulk_restore_test.dart index 8a5d88bf9..989f7d478 100644 --- a/test/import_bulk_restore_test.dart +++ b/test/import_bulk_restore_test.dart @@ -209,7 +209,7 @@ void main() { 'device_id': '', 'ts_ms': (_t0 + 1) * 1000, 'rec_ts': _t0 + 1, 'beat_index': 0, 'rr_ts_ms': (_t0 + 1) * 1000, 'rr_ms': 888, }); - await db.insert('raw_archive', { + await db.insert(LocalDb.rawArchiveView, { 'device_id': '', 'hex': 'local-only', 'counter': 1, 'packet_type': 32, 'rec_ts': _t0, 'captured_at': 1, 'reason': 'undecodable_rec_v20', @@ -518,7 +518,7 @@ void main() { test('raw archive keeps every frame; substrate archive merges losslessly ' 'with local winning collisions', () async { - final raw = await db.query('raw_archive', orderBy: 'hex'); + final raw = await db.query(LocalDb.rawArchiveView, orderBy: 'hex'); expect([for (final r in raw) r['hex']], ['local-only', 'src-a', 'src-b']); expect(counts['raw_archive'], 2); final row = (await db.query('substrate_archive')).single;