diff --git a/lib/data/db.dart b/lib/data/db.dart index 2ca1326..5e4b6c5 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -1199,82 +1199,121 @@ class LocalDb { } final db = await instance; - await db.transaction((txn) async { - // Read the existing high-water THROUGH the txn — never via the global db - // handle, which would deadlock against this same open transaction. - var maxCounter = await _cursorIntVia(txn, 'counter_hw') ?? 0; - var maxRecTs = await _cursorIntVia(txn, 'rec_ts_hw') ?? 0; - // CHUNKED BATCH: sqflite serialises an ENTIRE batch's operations+args into - // ONE platform-channel message, and the native side builds a single - // ArrayList of every argument. A large backlog offload (raws in the - // hundreds-of-thousands) blew the native heap in SqlCommand.getSqlArguments - // → OutOfMemoryError (Crashlytics 0.9.13). Committing in bounded chunks - // flushes and frees each message's args. These commits all happen INSIDE - // the single `db.transaction` below, so the safe-trim invariant holds: the - // whole offload (raw_archive + samples + decoded_onehz + decoded_rr + - // cursor) is still one atomic transaction — every row is durable before the - // caller echoes the HISTORY_END trim token, or none is. - const chunkOps = 4000; - var batch = txn.batch(); - var ops = 0; - Future flushChunk() async { - if (ops == 0) return; - await batch.commit(noResult: true); - batch = txn.batch(); - ops = 0; - } + // POWER-LOSS DURABILITY WINDOW. This is the ACK-gating commit: once it + // returns, the caller writes the BLE batch-ACK and the band trims its flash. + // Under WAL + synchronous=NORMAL (the default this connection opens with) a + // commit is durable only at the next checkpoint — so a kernel panic / + // battery-yank AFTER the ACK but BEFORE the -wal is checkpointed loses these + // just-committed rows from the phone while they are already gone from the + // band. Raise durability to FULL (fsync AT commit) for THIS commit only, + // leaving every other path at NORMAL — they are all recomputable and FULL + // everywhere is brutally slow. `synchronous` is per-connection and CANNOT be + // changed mid-transaction, so it is set on the connection BEFORE + // db.transaction opens and reset AFTER it commits. The reset lives in a + // finally: a leaked FULL from a throwing commit would fsync every subsequent + // write on this connection forever. `PRAGMA synchronous=FULL/NORMAL` returns + // NO rows → execute() (not rawQuery), kept non-fatal like the open-time + // PRAGMAs so a PRAGMA throw can never fail a durable commit. Every ACK-gating + // commit — the foreground drain AND the headless iOS-restore recovery drain + // (background_sync.dart) — funnels through here, so this one choke point + // covers them all. `synchronous` is per-connection, so the bracket is only + // safe because these drains never OVERLAP on a shared connection: the offload + // processor is single-flight (ble_engine.dart) and BandOwnership makes the + // headless drain yield when the foreground owns the band. Do not add a second + // concurrent caller of commitSyncBatch on the main-isolate connection without + // reinstating that serialization — a mid-window reset would silently + // downgrade this commit back to NORMAL. + try { + await db.execute('PRAGMA synchronous=FULL'); + } catch (_) { + /* durability upgrade is best-effort — NORMAL still commits correctly */ + } + try { + await db.transaction((txn) async { + // Read the existing high-water THROUGH the txn — never via the global db + // handle, which would deadlock against this same open transaction. + var maxCounter = await _cursorIntVia(txn, 'counter_hw') ?? 0; + var maxRecTs = await _cursorIntVia(txn, 'rec_ts_hw') ?? 0; + // CHUNKED BATCH: sqflite serialises an ENTIRE batch's operations+args into + // ONE platform-channel message, and the native side builds a single + // ArrayList of every argument. A large backlog offload (raws in the + // hundreds-of-thousands) blew the native heap in SqlCommand.getSqlArguments + // → OutOfMemoryError (Crashlytics 0.9.13). Committing in bounded chunks + // flushes and frees each message's args. These commits all happen INSIDE + // the single `db.transaction` below, so the safe-trim invariant holds: the + // whole offload (raw_archive + samples + decoded_onehz + decoded_rr + + // cursor) is still one atomic transaction — every row is durable before the + // caller echoes the HISTORY_END trim token, or none is. + const chunkOps = 4000; + var batch = txn.batch(); + var ops = 0; + Future flushChunk() async { + if (ops == 0) return; + await batch.commit(noResult: true); + batch = txn.batch(); + ops = 0; + } - // SAFE-TRIM INVARIANT: archive the undecodable records in the SAME - // transaction as the raw records + trim cursor, so they are durably set - // aside BEFORE the caller writes the batch-ACK that lets the band trim. - if (archives != null) { - for (final a in archives) { - batch.insert('raw_archive', { - 'counter': a.counter, - 'hex': a.hex, - 'packet_type': a.packetType, - 'rec_ts': a.recTs, - 'captured_at': a.capturedAt, - 'reason': a.reason, - }, conflictAlgorithm: ConflictAlgorithm.ignore); - if (++ops >= chunkOps) await flushChunk(); + // SAFE-TRIM INVARIANT: archive the undecodable records in the SAME + // transaction as the raw records + trim cursor, so they are durably set + // aside BEFORE the caller writes the batch-ACK that lets the band trim. + if (archives != null) { + for (final a in archives) { + batch.insert('raw_archive', { + 'counter': a.counter, + 'hex': a.hex, + 'packet_type': a.packetType, + 'rec_ts': a.recTs, + 'captured_at': a.capturedAt, + 'reason': a.reason, + }, conflictAlgorithm: ConflictAlgorithm.ignore); + if (++ops >= chunkOps) await flushChunk(); + } } - } - for (var i = 0; i < raws.length; i++) { - final raw = raws[i]; - final recTs = _recTsFor(raw); - final sample = samples[i]; - if (sample != null) { - batch.insert('samples', { - 'counter': raw.counter, - ...sample.toDbMap(), - }, conflictAlgorithm: ConflictAlgorithm.ignore); - ops++; + for (var i = 0; i < raws.length; i++) { + final raw = raws[i]; + final recTs = _recTsFor(raw); + final sample = samples[i]; + if (sample != null) { + batch.insert('samples', { + 'counter': raw.counter, + ...sample.toDbMap(), + }, conflictAlgorithm: ConflictAlgorithm.ignore); + ops++; + } + ops += _queueDecodedOneHz(batch, raw, sample); + if (raw.counter > maxCounter) maxCounter = raw.counter; + if (recTs > maxRecTs) maxRecTs = recTs; + if (ops >= chunkOps) await flushChunk(); } - ops += _queueDecodedOneHz(batch, raw, sample); - if (raw.counter > maxCounter) maxCounter = raw.counter; - if (recTs > maxRecTs) maxRecTs = recTs; - if (ops >= chunkOps) await flushChunk(); - } - checkpoint( - 'decoded_archive_queued raws=${raws.length} ' - 'archives=${archives?.length ?? 0}', - ); - await flushChunk(); - checkpoint('decoded_archive_committed'); - await setCursor('counter_hw', '$maxCounter', txn: txn); - await setCursor('rec_ts_hw', '$maxRecTs', txn: txn); - if (trimToken != null) await setCursor('strap_trim', trimToken, txn: txn); - if (extraCursors != null) { - for (final e in extraCursors.entries) { - await setCursor(e.key, e.value, txn: txn); + checkpoint( + 'decoded_archive_queued raws=${raws.length} ' + 'archives=${archives?.length ?? 0}', + ); + await flushChunk(); + checkpoint('decoded_archive_committed'); + await setCursor('counter_hw', '$maxCounter', txn: txn); + await setCursor('rec_ts_hw', '$maxRecTs', txn: txn); + if (trimToken != null) await setCursor('strap_trim', trimToken, txn: txn); + if (extraCursors != null) { + for (final e in extraCursors.entries) { + await setCursor(e.key, e.value, txn: txn); + } } + checkpoint( + 'cursor_advanced counter_hw=$maxCounter rec_ts_hw=$maxRecTs ' + 'trim=${trimToken != null}', + ); + }); + } finally { + // ALWAYS restore NORMAL — even if the commit threw — so a leaked FULL does + // not fsync every subsequent write on this connection. Non-fatal. + try { + await db.execute('PRAGMA synchronous=NORMAL'); + } catch (_) { + /* non-fatal — see open-time PRAGMA discipline */ } - checkpoint( - 'cursor_advanced counter_hw=$maxCounter rec_ts_hw=$maxRecTs ' - 'trim=${trimToken != null}', - ); - }); + } await _writeCaptureFreshness(raws); } diff --git a/pubspec.lock b/pubspec.lock index 3301a54..566a91a 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -892,10 +892,10 @@ packages: dependency: transitive description: name: meta - sha256: "23f08335362185a5ea2ad3a4e597f1375e78bce8a040df5c600c8d3552ef2394" + sha256: "1741988757a65eb6b36abe716829688cf01910bbf91c34354ff7ec1c3de2b349" url: "https://pub.dev" source: hosted - version: "1.17.0" + version: "1.18.0" mgrs_dart: dependency: transitive description: @@ -1328,7 +1328,7 @@ packages: source: hosted version: "2.4.2+3" sqflite_common: - dependency: transitive + dependency: "direct dev" description: name: sqflite_common sha256: "1581ffbf7a0e333b380d6a30737d78516b826cb35beb7fb0bf8a3ea0c678b465" @@ -1411,26 +1411,26 @@ packages: dependency: "direct dev" description: name: test - sha256: "280d6d890011ca966ad08df7e8a4ddfab0fb3aa49f96ed6de56e3521347a9ae7" + sha256: "8d9ceddbab833f180fbefed08afa76d7c03513dfdba87ffcec2718b02bbcbf20" url: "https://pub.dev" source: hosted - version: "1.30.0" + version: "1.31.0" test_api: dependency: transitive description: name: test_api - sha256: "8161c84903fd860b26bfdefb7963b3f0b68fee7adea0f59ef805ecca346f0c7a" + sha256: "949a932224383300f01be9221c39180316445ecb8e7547f70a41a35bf421fb9e" url: "https://pub.dev" source: hosted - version: "0.7.10" + version: "0.7.11" test_core: dependency: transitive description: name: test_core - sha256: "0381bd1585d1a924763c308100f2138205252fb90c9d4eeaf28489ee65ccde51" + sha256: "1991d4cfe85d5043241acac92962c3977c8d2f2add1ee73130c7b286417d1d34" url: "https://pub.dev" source: hosted - version: "0.6.16" + version: "0.6.17" timezone: dependency: "direct main" description: diff --git a/pubspec.yaml b/pubspec.yaml index f334e21..fc4661b 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -260,6 +260,10 @@ dev_dependencies: # exportDaysDb can run without a platform plugin (transitive via # path_provider; declared directly since test/ now imports it). path_provider_platform_interface: ^2.1.0 + # ack_commit_sync_full_test wraps the ffi factory in SqfliteDatabaseFactoryLogger + # to spy the PRAGMA synchronous=FULL/NORMAL bracket around the ACK-gating commit + # (transitive via sqflite_common_ffi; declared directly since test/ now imports it). + sqflite_common: ^2.5.0 flutter_launcher_icons: android: "launcher_icon" diff --git a/test/ack_commit_sync_full_test.dart b/test/ack_commit_sync_full_test.dart new file mode 100644 index 0000000..64c2197 --- /dev/null +++ b/test/ack_commit_sync_full_test.dart @@ -0,0 +1,98 @@ +// POWER-LOSS DURABILITY of the ACK-gating commit. `commitSyncBatch` is the one +// commit the safe-trim invariant hangs on: it must be durable (fsynced) BEFORE +// the caller writes the BLE batch-ACK that lets the band trim its flash. The DB +// otherwise runs WAL + synchronous=NORMAL (durable only at a checkpoint), so +// this path raises synchronous=FULL for its single transaction and restores +// NORMAL afterward — leaving every other (recomputable) path fast. This test +// pins that bracket: FULL is set around the commit, NORMAL is restored after, +// AND the restore still happens when the commit THROWS (a leaked FULL would +// fsync every subsequent write on the connection forever). +// +// We spy the real SQL stream via SqfliteDatabaseFactoryLogger (synchronous is +// per-connection and invisible from a second connection, so the log is the only +// honest observation point) and also read `PRAGMA synchronous` on LocalDb's own +// connection — the same one commitSyncBatch uses — to confirm the resting value. + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +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/models.dart'; + +void main() { + // Every `synchronous=…` statement executed on any connection, in order. + final syncStmts = []; + + setUpAll(() async { + sqfliteFfiInit(); + // ignore: experimental_member_use — stable enough to spy SQL in a test. + databaseFactory = SqfliteDatabaseFactoryLogger( + databaseFactoryFfi, + options: SqfliteLoggerOptions( + log: (event) { + if (event is SqfliteLoggerSqlEvent) { + final sql = event.sql.toLowerCase(); + if (sql.contains('pragma synchronous=')) syncStmts.add(sql); + } + }, + ), + ); + LocalDb.dbName = 'openstrap_ack_sync_full_test.db'; + final dir = await databaseFactory.getDatabasesPath(); + await databaseFactory.deleteDatabase(p.join(dir, LocalDb.dbName)); + // Force the open now so its onConfigure PRAGMAs aren't counted in per-test + // windows — each test clears syncStmts against an already-open connection. + await LocalDb.instance; + }); + + tearDownAll(() async { + await LocalDb.close(); + final dir = await databaseFactory.getDatabasesPath(); + await databaseFactory.deleteDatabase(p.join(dir, LocalDb.dbName)); + }); + + Future restingSynchronous() async { + final db = await LocalDb.instance; + final rows = await db.rawQuery('PRAGMA synchronous'); + return rows.first.values.first as int; // FULL=2, NORMAL=1 + } + + RawRecord recAt(int counter) => RawRecord( + counter: counter, + packetType: 0x2F, + hex: '2f18aabbccdd', + capturedAt: 1750000000000 + counter, + recTs: 1750000000 + counter, + ); + + test('commitSyncBatch brackets synchronous=FULL and restores NORMAL', () async { + expect(await restingSynchronous(), 1, reason: 'connection opens at NORMAL'); + + syncStmts.clear(); + await LocalDb.commitSyncBatch( + [recAt(5001)], + [Sample(tsEpoch: 1750005001, counter: 5001, hr: 60)], + trimToken: 'deadbeef', + ); + + expect(syncStmts, ['pragma synchronous=full', 'pragma synchronous=normal'], + reason: 'FULL is set before the commit and NORMAL restored right after'); + expect(await restingSynchronous(), 1, reason: 'connection left at NORMAL'); + }); + + test('synchronous is restored to NORMAL even when the commit throws', () async { + syncStmts.clear(); + // raws non-empty but samples empty → samples[i] throws RangeError INSIDE the + // db.transaction, after FULL is set. The finally must still restore NORMAL. + await expectLater( + LocalDb.commitSyncBatch([recAt(6001)], const []), + throwsA(isA()), + ); + + expect(syncStmts, ['pragma synchronous=full', 'pragma synchronous=normal'], + reason: 'a thrown commit must not leak FULL'); + expect(await restingSynchronous(), 1, + reason: 'FULL did not leak past the throwing commit'); + }); +}