From 4a2743b63f685041a03f5dc2c1a1fac5d128e090 Mon Sep 17 00:00:00 2001 From: MikenRafter Date: Fri, 23 Oct 2026 21:09:45 -0600 Subject: [PATCH 1/2] perf(db): screen reads use a query-only WAL reader connection Every screen read shared the one writer connection, so a sync or derive holding a write transaction made the screens wait behind it. Under WAL a second connection can read the last committed state while a write is open. - LocalDb opens a lazy second connection to the same file after the writer has opened, migrated and repaired the schema: no migration callbacks, PRAGMA query_only=ON, WAL verified (journal_mode through rawQuery). If it cannot open, or the file is not in WAL mode, reads fall back to the writer. - The reader exists only while the app is in the foreground. The app lifecycle observer enables it on a visible cold start and on resume, and hidden/paused/detached drain in-flight read leases and close it. close(), wipe, restore merge and a writer reopen or rebuild retire it first. - Only an explicit list of read-only repository paths use it (day overview, latest bundle, cross-day artifact, available days, sleep windows, the RHR chart, the day timeline and sessions). Every other LocalDb caller, including compute, stays on the writer, and so do getToday's freshness refresh and frozenHeadline's orphan cleanup, which write. The sync commit path is unchanged: the per-commit synchronous=FULL bracket and commit-before-ACK stay on the writer. --- lib/app.dart | 20 +- lib/data/db.dart | 249 ++++++++++-- lib/data/local_repository_impl.dart | 44 ++- test/wal_reader_connection_test.dart | 541 +++++++++++++++++++++++++++ test/wal_reader_wiring_test.dart | 106 ++++++ 5 files changed, 906 insertions(+), 54 deletions(-) create mode 100644 test/wal_reader_connection_test.dart create mode 100644 test/wal_reader_wiring_test.dart diff --git a/lib/app.dart b/lib/app.dart index 99c7abf1e..0e5146a75 100644 --- a/lib/app.dart +++ b/lib/app.dart @@ -6,6 +6,7 @@ import 'package:provider/provider.dart'; import 'ai/briefing.dart' show BriefingPeriod; import 'coach/coach_config.dart'; +import 'data/db.dart'; import 'l10n/app_localizations.dart'; import 'notify/notification_service.dart'; import 'notify/tap_router.dart'; @@ -51,6 +52,11 @@ class _OpenStrapAppState extends State with WidgetsBindingObserver void initState() { super.initState(); WidgetsBinding.instance.addObserver(this); + final lifecycleState = WidgetsBinding.instance.lifecycleState; + unawaited(LocalDb.setForeground( + lifecycleState != AppLifecycleState.hidden && + lifecycleState != AppLifecycleState.paused && + lifecycleState != AppLifecycleState.detached)); // Hand the BYOK provider config to AppState (briefing generation + the // key-aware AI notification schedule live there). WidgetsBinding.instance.addPostFrameCallback((_) { @@ -79,6 +85,7 @@ class _OpenStrapAppState extends State with WidgetsBindingObserver void didChangeAppLifecycleState(AppLifecycleState state) { final app = context.read(); if (state == AppLifecycleState.resumed) { + unawaited(LocalDb.setForeground(true)); // The user may have flipped our notification switch either way in OS // Settings while we were backgrounded. Drop the cached authorization // decision so the next present/schedule re-reads reality — a denial used @@ -114,10 +121,15 @@ class _OpenStrapAppState extends State with WidgetsBindingObserver // a lock screen looking current. unawaited(WidgetService.refresh(app.repo)); if (app.isPaired) app.openSession(); - } else if (state == AppLifecycleState.paused) { - // Backgrounded: hand the band to the iOS restore path so it can wake-and-drain - // in the background (no-op on Android, where the foreground service holds it). - app.pauseForBackground(); + } else if (state == AppLifecycleState.hidden || + state == AppLifecycleState.paused || + state == AppLifecycleState.detached) { + unawaited(LocalDb.setForeground(false)); + if (state == AppLifecycleState.paused) { + // Backgrounded: hand the band to the iOS restore path so it can wake-and-drain + // in the background (no-op on Android, where the foreground service holds it). + app.pauseForBackground(); + } } } diff --git a/lib/data/db.dart b/lib/data/db.dart index ed954c51b..3d225010b 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -236,7 +236,19 @@ class LocalDb { // object)` — a sustained crash burst (seen in the wild on background event // ingest). Reopen whenever the cached handle isn't actually open. if (db != null && db.isOpen) return db; - return _db = await _openOrRebuild(); + return _writerOpening ??= _openWriter(); + } + + static Future? _writerOpening; + + static Future _openWriter() async { + try { + // A reader must never retain a handle to a closed/replaced writer's file. + await _retireReader(drainAll: false); + return _db = await _openOrRebuild(); + } finally { + _writerOpening = null; + } } /// Set when [_openOrRebuild] had to quarantine an unopenable database and @@ -400,7 +412,7 @@ class LocalDb { // lost. Drop them here; the card is the one surface whose whole job is // telling the truth about a data-loss event. salvaged = Map.of( - await _mergeFromDbFile( + await _mergeFromDbFileImpl( quarantine, only: _salvageTables, tolerant: true, @@ -419,13 +431,168 @@ class LocalDb { } } - static Future close() async { - if (_db != null) { - await _db!.close(); - _db = null; + static Future close() => _withReaderInvalidated(() async { + // A cold open may have started before close, even without a screen lease. + await _writerOpening; + final db = _db; + if (db != null) { + await db.close(); + if (identical(_db, db)) _db = null; + } + }); + + static Database? _reader; + static Database? _readerWriter; + static Future? _readerOpening; + static Database? _readerFailedFor; + static bool _foreground = false; + static int _readerEpoch = 0; + static Future? _screenBarrier; + static Future? _readerClosing; + static final _readLeases = >{}; + static final _readerLeases = >{}; + + /// Bare acquisition for diagnostics. Screen queries use [withReadExecutor] + /// so teardown can wait across all awaits in a read. + static Future readExecutor() async { + while (_screenBarrier != null) { + await _screenBarrier; + } + return _acquireReadExecutor(); + } + + static Future _acquireReadExecutor() async { + final writer = await (_writerOpening ?? instance); + // Includes migrations, repair and any rebuild salvage. Existing leases may + // fall back to this writer while a store change waits for them to finish. + if (!_foreground || _screenBarrier != null) return writer; + final current = _reader; + if (current != null && current.isOpen && identical(_readerWriter, writer)) { + return current; + } + if (identical(_readerFailedFor, writer)) return writer; + final epoch = _readerEpoch; + final opening = _readerOpening ??= _openReader(writer, epoch); + final opened = await opening; + if (identical(_readerOpening, opening)) _readerOpening = null; + return _foreground && epoch == _readerEpoch && + identical(_db, writer) && identical(opened, _reader) + ? opened ?? writer + : writer; + } + + static Future _openReader(Database writer, int epoch) async { + Database? candidate; + try { + candidate = await (debugReaderOpener?.call(writer.path) ?? + openDatabase(writer.path, singleInstance: false)); + await candidate.execute('PRAGMA query_only=ON'); + // Read, never change, journal mode. A rollback-journal connection cannot + // provide the concurrent committed reads this connection exists for. + final mode = await candidate.rawQuery('PRAGMA journal_mode'); + if (mode.isEmpty || '${mode.first.values.first}'.toLowerCase() != 'wal') { + throw StateError('reader requires WAL'); + } + if (!_foreground || epoch != _readerEpoch || + !identical(_db, writer) || !writer.isOpen) { + await candidate.close(); + return null; + } + _reader = candidate; + _readerWriter = writer; + return candidate; + } catch (_) { + if (candidate != null) { + try { await candidate.close(); } catch (_) {} + } + if (epoch == _readerEpoch && identical(_db, writer)) { + _readerFailedFor = writer; + } + return null; + } + } + + static Future withReadExecutor( + Future Function(DatabaseExecutor db) read, + ) async { + while (_screenBarrier != null) { + await _screenBarrier; + } + // Register BEFORE acquisition: close also drains a read awaiting its opener + // and keeps the writer alive if that opener must fall back after invalidation. + final lease = Completer(); + _readLeases.add(lease); + try { + var db = await _acquireReadExecutor(); + if (!db.isOpen || (!identical(db, _db) && !identical(db, _reader))) { + db = await instance; + } + if (identical(db, _reader)) _readerLeases.add(lease); + return await read(db); + } finally { + _readerLeases.remove(lease); + _readLeases.remove(lease); + lease.complete(); + } + } + + static Future setForeground(bool foreground) { + _foreground = foreground; + return foreground ? Future.value() : _retireReader(); + } + + static Future _retireReader({bool drainAll = true}) { + _readerEpoch++; + _readerFailedFor = null; + final reader = _reader; + final opening = _readerOpening; + final previousClose = _readerClosing; + _reader = null; + _readerWriter = null; + final readerLeases = _readerLeases.toList(); + final allLeases = drainAll ? _readLeases.toList() : readerLeases; + // Track handle retirement separately from acquisition leases: a cold + // writer open can itself be awaited by a lease that close is draining. + final closing = () async { + await previousClose; + await Future.wait(readerLeases.map((lease) => lease.future)); + await opening; + if (identical(_readerOpening, opening)) _readerOpening = null; + if (reader != null) await reader.close(); + }(); + _readerClosing = closing; + return () async { + try { + await Future.wait([ + closing, + ...allLeases.map((lease) => lease.future), + ]); + } finally { + if (identical(_readerClosing, closing)) _readerClosing = null; + } + }(); + } + + /// Block new screen leases while a reset/merge/close changes the store. + /// Existing acquisitions finish on the writer if their reader open goes stale. + static Future _withReaderInvalidated(Future Function() action) async { + final previous = _screenBarrier; + final done = Completer(); + _screenBarrier = done.future; + try { + if (previous != null) await previous; + await _retireReader(); + return await action(); + } finally { + if (identical(_screenBarrier, done.future)) _screenBarrier = null; + done.complete(); } } + /// Replaces only the second connection's open, never the writer's open. + @visibleForTesting + static Future Function(String path)? debugReaderOpener; + /// Runs a write against a guaranteed-open handle, reopening ONCE if the /// cached handle was closed under us mid-write. /// @@ -4613,8 +4780,8 @@ class LocalDb { /// The dates whose stored [key] is a wearable's estimate, or a row its /// column serves as unavailable (`metric_method.class`): not a measurement /// a plain trend line may draw. - static Future> metricNotOursDates(String key) async { - final db = await instance; + static Future> metricNotOursDates(String key, {DatabaseExecutor? executor}) async { + final db = executor ?? await instance; return { for (final r in await db.query('metric_method', columns: ['date'], @@ -4626,8 +4793,8 @@ class LocalDb { /// The algo version behind each day's `metric_series` scalars, oldest first. /// Days with no stamp are simply absent — never guessed at. - static Future>> metricSeriesVersions() async { - final db = await instance; + static Future>> metricSeriesVersions({DatabaseExecutor? executor}) async { + final db = executor ?? await instance; return db.query('metric_series_version', orderBy: 'date ASC'); } @@ -8028,8 +8195,9 @@ class LocalDb { int fromTs, int toTs, { int limit = 5000, + DatabaseExecutor? executor, }) async { - final db = await instance; + final db = executor ?? await instance; return db.query( 'events', where: 'ts >= ? AND ts < ?', @@ -8892,8 +9060,8 @@ class LocalDb { ); } - static Future?> dayResult(String dayId) async { - final db = await instance; + static Future?> dayResult(String dayId, {DatabaseExecutor? executor}) async { + final db = executor ?? await instance; final rows = await db.query( 'day_result', where: 'day_id = ? AND algo_version <= ?', @@ -8938,8 +9106,8 @@ class LocalDb { } /// The N most recent days (newest day_id first), each at its LATEST version. - static Future>> recentDayResults(int limit) async { - final db = await instance; + static Future>> recentDayResults(int limit, {DatabaseExecutor? executor}) async { + final db = executor ?? await instance; // For each day_id pick MAX(algo_version), then join back for the full row. final rows = await db.rawQuery( 'SELECT r.* FROM day_result r ' @@ -9003,8 +9171,8 @@ class LocalDb { /// round trip. Rows: `{day_id, window_json}`. [before] keeps only days /// strictly earlier than that day_id. static Future>> sleepWindowRows(int limit, - {String? before}) async { - final db = await instance; + {String? before, DatabaseExecutor? executor}) async { + final db = executor ?? await instance; return db.rawQuery( 'SELECT r.day_id AS day_id, r.window_json AS window_json ' 'FROM day_result r ' @@ -9079,8 +9247,8 @@ class LocalDb { /// derived row) and a skip-marker day both render EMPTY, so neither must /// bound navigation. Skips are excluded via the `skipped` column, which /// `_markDaySkipped` sets alongside the `{skipped:true}` payload. - static Future> availableDayIds() async { - final db = await instance; + static Future> availableDayIds({DatabaseExecutor? executor}) async { + final db = executor ?? await instance; // Latest version per day (matches [dayResult]), then drop skip-markers. final rows = await db.rawQuery( 'SELECT r.day_id FROM day_result r ' @@ -9710,7 +9878,9 @@ class LocalDb { /// run inside `openDatabase` on iOS's CPU watchdog — not somewhere to be on /// the tail of a destructive user action). Views are `type='view'` and are /// not matched; the sqlite/Android internal tables are skipped by name. - static Future wipeAll() async { + static Future wipeAll() => _withReaderInvalidated(_wipeAll); + + static Future _wipeAll() async { final db = await instance; final rows = await db.rawQuery( "SELECT name FROM sqlite_master WHERE type = 'table' " @@ -10056,6 +10226,13 @@ class LocalDb { String path, { List? only, bool tolerant = false, + }) => _withReaderInvalidated(() => + _mergeFromDbFileImpl(path, only: only, tolerant: tolerant)); + + static Future> _mergeFromDbFileImpl( + String path, { + List? only, + bool tolerant = false, }) async { final src = await openDatabase(path, readOnly: true); final db = await instance; @@ -11045,8 +11222,9 @@ class LocalDb { String key, { int? limit, bool measuredOnly = false, + DatabaseExecutor? executor, }) async { - final db = await instance; + final db = executor ?? await instance; return db.query( 'metric_series', where: 'key = ? AND value IS NOT NULL${_validSql(key)}' @@ -11097,8 +11275,8 @@ class LocalDb { return [for (final r in rows.reversed) (r['value'] as num).toDouble()]; } - static Future?> baseline(String key) async { - final db = await instance; + static Future?> baseline(String key, {DatabaseExecutor? executor}) async { + final db = executor ?? await instance; final rows = await db.query( 'baselines', where: 'key = ?', @@ -12432,15 +12610,16 @@ class LocalDb { static Future>> sessionsInRange( int fromTs, - int toTs, - ) async { - final db = await instance; + int toTs, { + DatabaseExecutor? executor, + }) async { + final db = executor ?? await instance; return withoutFlagOffScores(await db.query( 'sessions', where: 'start_ts >= ? AND start_ts <= ?', whereArgs: [fromTs, toTs], orderBy: 'start_ts DESC', - )); + ), executor: db); } /// What a session's sensor scored: served empty, not stored empty. @@ -12466,10 +12645,10 @@ class LocalDb { /// ponytail: one window scan per session; batch it into one join if a long /// session list gets slow. static Future>> withoutFlagOffScores( - List> rows) async { + List> rows, {DatabaseExecutor? executor}) async { final out = >[]; for (final r in rows) { - out.add(await _flagOffOnly(r) + out.add(await _flagOffOnly(r, executor: executor) ? {...r, for (final c in _sensorScoreColumns) c: null} : r); } @@ -12479,7 +12658,8 @@ class LocalDb { /// Whether finished session [r]'s window has worn seconds and every one of /// them came from a device whose flag is now off (see /// [withoutFlagOffScores]). - static Future _flagOffOnly(Map r) async { + static Future _flagOffOnly(Map r, + {DatabaseExecutor? executor}) async { final s = (r['start_ts'] as num?)?.toInt(); final e = (r['end_ts'] as num?)?.toInt(); if (s == null || e == null || r['status'] != 'done') return false; @@ -12488,13 +12668,13 @@ class LocalDb { // wearable's) and once its raw is pruned. // ponytail: a stamped window the band also measured serves empty too, // not band-only; rescore it from the band's rows if that matters. - if ((await sessionSensorsOf(r['id'])) + if ((await sessionSensorsOf(r['id'], executor: executor)) .any((s) => !sessionSensorSources.contains(s))) { return true; } final on = {...sessionSensorSources, ?sessionWearableSource}; final sources = [ - for (final x in await (await instance).rawQuery( + for (final x in await (executor ?? await instance).rawQuery( 'SELECT DISTINCT source FROM decoded_onehz ' 'WHERE rec_ts >= ? AND rec_ts <= ? AND hr > 0', [s, e])) @@ -12530,8 +12710,9 @@ class LocalDb { } /// Every workout sensor stamped on session [id] ([stampSessionSensor]). - static Future> sessionSensorsOf(Object? id) async => { - for (final r in await (await instance).query('session_sensor', + static Future> sessionSensorsOf(Object? id, + {DatabaseExecutor? executor}) async => { + for (final r in await (executor ?? await instance).query('session_sensor', columns: ['source'], where: 'session_id = ?', whereArgs: [id])) r['source'] as String, }; diff --git a/lib/data/local_repository_impl.dart b/lib/data/local_repository_impl.dart index a6b79a5a8..89b3eb29b 100644 --- a/lib/data/local_repository_impl.dart +++ b/lib/data/local_repository_impl.dart @@ -98,7 +98,8 @@ class LocalRepositoryImpl extends LocalRepository { /// Decode a day_result row's payload bundle (latest algo_version), or null. Future?> _bundle(String date) async { - final row = await LocalDb.dayResult(date); + final row = await LocalDb.withReadExecutor( + (db) => LocalDb.dayResult(date, executor: db)); if (row == null) return null; final b = _decode(row['payload_json']); return b == null ? null : withoutWearableEstimates(b); @@ -117,7 +118,8 @@ class LocalRepositoryImpl extends LocalRepository { /// today's night is not the overnight yet (#448), so a partial night can't /// be served under the prior night's label. Future?> _latestBundle({String? skipDay}) async { - final rows = await LocalDb.recentDayResults(14); + final rows = await LocalDb.withReadExecutor( + (db) => LocalDb.recentDayResults(14, executor: db)); Map? newest, withScalars; for (final row in rows) { if (skipDay != null && row['day_id']?.toString() == skipDay) continue; @@ -155,21 +157,22 @@ class LocalRepositoryImpl extends LocalRepository { /// The stored artifact, UNGATED — only `_crossDay` and `getInsights` may /// read this, and only so `getInsights` can report why it withheld it. - Future?> _crossDayArtifact() async { - final r = await LocalDb.baseline('crossday'); + Future?> _crossDayArtifact() => + LocalDb.withReadExecutor((db) async { + final r = await LocalDb.baseline('crossday', executor: db); final cd = _decode(r?['payload_json']); if (cd == null) return null; // A wearable's estimated TRIMP anywhere in the crossday input (a ring's, // say) moves the load and the strain target built on it: the load's EWMA // runs over the input's whole span (90 day rows), not 42 days. Labelled // Estimated where shown, not dropped (whether it counts is the owner's). - final notOurs = await LocalDb.metricNotOursDates('trimp'); + final notOurs = await LocalDb.metricNotOursDates('trimp', executor: db); if (notOurs.isEmpty) return cd; final now = DateTime.now(); var from = dayLabelOf(DateTime(now.year, now.month, now.day - 90)); // 90 rows can span more than 90 calendar days: the input's oldest wins. final input = - _decode((await LocalDb.baseline('crossday_input'))?['payload_json']); + _decode((await LocalDb.baseline('crossday_input', executor: db))?['payload_json']); final inDays = input?['days']; if (inDays is List && inDays.isNotEmpty) { final first = (inDays.first as Map?)?['date']; @@ -181,7 +184,7 @@ class LocalRepositoryImpl extends LocalRepository { for (final k in const ['load', 'strain_coach']) if (cd[k] is Map) k: {...cd[k] as Map, 'estimated': true}, }; - } + }); Future?> _freshness(String key) async { final row = await LocalDb.computeFreshness(key); @@ -996,7 +999,8 @@ class LocalRepositoryImpl extends LocalRepository { } @override - Future> availableDays() => LocalDb.availableDayIds(); + Future> availableDays() => LocalDb.withReadExecutor( + (db) => LocalDb.availableDayIds(executor: db)); @override Future> getDaySleep(String date) => _daySleep(date); @@ -1770,14 +1774,16 @@ class LocalRepositoryImpl extends LocalRepository { } // Workouts + device events for that calendar day. - final sess = await LocalDb.sessionsInRange(dayStart, dayEnd); + final sess = await LocalDb.withReadExecutor( + (db) => LocalDb.sessionsInRange(dayStart, dayEnd, executor: db)); // Bounded BY THE DAY, in SQL. This used to pull `unuploadedEvents(limit: // 2000)` — `ORDER BY ts ASC LIMIT 2000`, i.e. the OLDEST 2000 rows — and // then filter that page down to this day. Once `events` held more than 2000 // rows the page could no longer reach recent days at all, so their markers // silently vanished from the timeline (the same oldest-N-vs-trailing-N // shape as the metricSeries(limit:) outage). - final dayEvents = await LocalDb.eventsInRange(dayStart, dayEnd); + final dayEvents = await LocalDb.withReadExecutor( + (db) => LocalDb.eventsInRange(dayStart, dayEnd, executor: db)); final events = >[ for (final e in dayEvents) { @@ -1879,7 +1885,8 @@ class LocalRepositoryImpl extends LocalRepository { @override Future>> sleepWindows( {int days = 60, String? before}) async { - final rows = await LocalDb.sleepWindowRows(days, before: before); + final rows = await LocalDb.withReadExecutor( + (db) => LocalDb.sleepWindowRows(days, before: before, executor: db)); final out = >[]; for (final r in rows) { final date = r['day_id'] as String?; @@ -1932,7 +1939,8 @@ class LocalRepositoryImpl extends LocalRepository { now.subtract(const Duration(days: 31)).millisecondsSinceEpoch ~/ 1000; final toSec = to ?? nowSec; - final manualRows = await LocalDb.sessionsInRange(fromSec, toSec); + final manualRows = await LocalDb.withReadExecutor( + (db) => LocalDb.sessionsInRange(fromSec, toSec, executor: db)); final manual = [for (final r in manualRows) _workoutOf(r)]; // Manual/live sessions are ALL there is now. There used to be a second @@ -2027,9 +2035,11 @@ class LocalRepositoryImpl extends LocalRepository { final key = _trendKey(metric); // A wearable's estimate is not drawn as a measured point, as on the // native cards ([withoutWearableEstimates]). - final notOurs = await LocalDb.metricNotOursDates(key); + final notOurs = await LocalDb.withReadExecutor( + (db) => LocalDb.metricNotOursDates(key, executor: db)); final rows = [ - for (final r in await LocalDb.metricSeries(key)) + for (final r in await LocalDb.withReadExecutor( + (db) => LocalDb.metricSeries(key, executor: db))) if (!notOurs.contains(r['date'])) r, ]; // THE PIN WINS. getToday serves the frozen morning headline for readiness, @@ -2058,7 +2068,8 @@ class LocalRepositoryImpl extends LocalRepository { // ONE read of metric_series_version, two consumers. `_algoBreaks` used to // fetch it privately; `coverage_devices` rides on the same rows because it // is the same table, the same day key and the same write. - final versions = await LocalDb.metricSeriesVersions(); + final versions = await LocalDb.withReadExecutor( + (db) => LocalDb.metricSeriesVersions(executor: db)); // WHO CONTRIBUTED TO EACH DAY'S SCALARS (final-plan §4.4). Days with no // stamp, and days stamped before schema 50, are simply ABSENT — never @@ -2109,7 +2120,8 @@ class LocalRepositoryImpl extends LocalRepository { // derived key — nothing here reconstructs it, and a day with no // `worn_min` row is simply absent. 'wear': [ - for (final r in await LocalDb.metricSeries('worn_min')) + for (final r in await LocalDb.withReadExecutor( + (db) => LocalDb.metricSeries('worn_min', executor: db))) {'t': _dateToEpoch(r['date'] as String), 'v': r['value']}, ], // L13 — WHERE THE MATHS CHANGED. The dates at which the algo version diff --git a/test/wal_reader_connection_test.dart b/test/wal_reader_connection_test.dart new file mode 100644 index 000000000..e4eb125aa --- /dev/null +++ b/test/wal_reader_connection_test.dart @@ -0,0 +1,541 @@ +// E22 phase 1: RED contracts against upstream's writer-only implementation. +// Real files matter: separate :memory: handles do not share a WAL database. +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +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/ble/ble_engine.dart' show DrainController; +import 'package:openstrap_edge/compute/derivation_engine.dart' show kAlgoVersion; +import 'package:openstrap_edge/data/db.dart'; +import 'package:openstrap_edge/data/day_label.dart'; +import 'package:openstrap_edge/data/local_repository_impl.dart'; +import 'package:openstrap_edge/data/models.dart'; + +const _wait = Duration(seconds: 2); +const _day = '2026-01-10'; +typedef _Sql = ({int id, int? txn, String sql, SqliteSqlCommandType type}); + +Future _putDay(String day, {double readiness = 61}) => + LocalDb.putDayResult( + dayId: day, algoVersion: kAlgoVersion, + payloadJson: jsonEncode({'scalars': {'readiness': readiness, 'rhr': 55}}), + windowJson: '{}', readiness: readiness, + series: {'rhr': 55}, + ); + +RawRecord _raw(int counter) => RawRecord( + counter: counter, packetType: 0x2f, hex: '2f18aabbccdd', + capturedAt: 1750000000000 + counter, recTs: 1750000000 + counter); +Sample _sample(int counter) => + Sample(tsEpoch: 1750000000 + counter, counter: counter, hr: 60, + rrIntervalsMs: [900, 910]); + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + final sql = <_Sql>[]; + final opens = []; + final closes = []; + late Directory dir; + late DatabaseFactory? savedFactory; + late String savedPath; + late String savedName; + late LocalRepositoryImpl repo; + var serial = 0; + + setUpAll(() async { + sqfliteFfiInit(); + savedFactory = databaseFactoryOrNull; + savedPath = await databaseFactoryFfi.getDatabasesPath(); + savedName = LocalDb.dbName; + dir = await Directory.systemTemp.createTemp('e22_wal_reader_'); + await databaseFactoryFfi.setDatabasesPath(dir.path); + // ignore: experimental_member_use + databaseFactory = SqfliteDatabaseFactoryLogger(databaseFactoryFfi, + options: SqfliteLoggerOptions(log: (event) { + if (event is SqfliteLoggerSqlEvent) { + sql.add((id: event.databaseId, txn: event.transactionId, + sql: event.sql.toLowerCase(), type: event.type)); + } else if (event is SqfliteLoggerBatchEvent) { + for (final op in event.operations) { + sql.add((id: event.databaseId, txn: event.transactionId, + sql: op.sql.toLowerCase(), type: op.type)); + } + } else if (event is SqfliteLoggerDatabaseOpenEvent) { + opens.add(event); + } else if (event is SqfliteLoggerDatabaseCloseEvent) { + closes.add(event.databaseId); + } + }), + ); + }); + setUp(() async { + await LocalDb.close(); + LocalDb.debugReaderOpener = null; + LocalDb.lastRebuild = null; + await LocalDb.setForeground(false); + LocalDb.dbName = 'reader_${serial++}.db'; + sql.clear(); + opens.clear(); + closes.clear(); + repo = LocalRepositoryImpl(getProfileMap: () => const {}); + await LocalDb.setForeground(true); + }); + tearDown(() async { + LocalDb.debugReaderOpener = null; + LocalDb.debugBeforeOrphanPinClear = null; + await LocalDb.setForeground(false); + await LocalDb.close(); + // Includes a handle opened by a delayed test opener if publication failed. + for (final event in opens) { + if (event.db?.isOpen ?? false) await event.db!.close(); + } + }); + tearDownAll(() async { + await LocalDb.close(); + LocalDb.dbName = savedName; + databaseFactoryOrNull = savedFactory; + await databaseFactoryFfi.setDatabasesPath(savedPath); + await dir.delete(recursive: true); + }); + + int idOf(Database db) => opens.lastWhere((e) => identical(e.db, db)).databaseId!; + + Future writerId() async => idOf(await LocalDb.instance); + + Future readerQueries(String table, int writer) async { + final hits = sql.where((e) => e.type == SqliteSqlCommandType.query && + RegExp('\\bfrom\\s+$table\\b').hasMatch(e.sql)).toList(); + expect(hits, isNotEmpty, reason: 'fixture must exercise $table'); + expect(hits.map((e) => e.id), everyElement(isNot(writer)), + reason: '$table screen SELECT must use the second connection'); + expect(hits.map((e) => e.id), everyElement(idOf(await LocalDb.readExecutor())), + reason: "$table must use LocalDb's owned reader handle"); + } + + // Release the writer even when a screen read times out on upstream, then + // drain that read before teardown. A failing RED test must not wedge SQLite. + Future holdingWrite(Future Function() body, + {bool rollback = false}) async { + final entered = Completer(); + final release = Completer(); + final writer = await LocalDb.instance; + final holding = writer.transaction((tx) async { + await tx.insert('day_result', { + 'day_id': '2026-01-11', 'algo_version': kAlgoVersion, + 'payload_json': '{"scalars":{"readiness":88}}', + 'window_json': '{}', 'computed_at': 1, 'readiness': 88, + }); + entered.complete(); + await release.future; + if (rollback) throw StateError('intentional rollback'); + }); + final settled = holding.then((_) => null, + onError: (Object e, StackTrace s) => e); + try { + await entered.future.timeout(_wait); + await body(); + } finally { + release.complete(); + final outcome = await settled.timeout(_wait); + if (!rollback && outcome != null) throw outcome; + if (rollback) expect(outcome, isA()); + } + } + + test('reader is a lazy second handle to the same WAL file', () async { + final writer = await LocalDb.instance; + expect(opens, hasLength(1)); + await LocalDb.setForeground(true); + expect(opens, hasLength(1), reason: 'foreground only permits lazy opening'); + final reader = await LocalDb.readExecutor(); + expect(identical(reader, writer), isFalse, + reason: 'upstream serializes all reads on the writer'); + expect(opens, hasLength(2)); + expect(reader.path, writer.path); + final files = await reader.rawQuery('PRAGMA database_list'); + expect(files.firstWhere((r) => r['name'] == 'main')['file'], writer.path); + final option = opens.last.options!; + expect(option.singleInstance, isFalse); + expect(option.readOnly, isFalse, reason: 'WAL may need to create -shm'); + expect(option.version, isNull); + expect(option.onCreate, isNull); + expect(option.onUpgrade, isNull); + expect(option.onOpen, isNull, reason: 'repair belongs only to writer'); + final readerId = idOf(reader); + final wal = sql.where((e) => e.id == readerId && + e.sql.startsWith('pragma journal_mode')).toList(); + expect(wal, isNotEmpty, reason: 'verify WAL before publishing reader'); + expect(wal.map((e) => e.type), everyElement(SqliteSqlCommandType.query)); + expect((await reader.rawQuery('PRAGMA journal_mode')).first.values.first, 'wal'); + }); + + test('cold reader waits for writer open and same-version schema repair', () async { + final writer = await LocalDb.instance; + final path = writer.path; + await writer.execute('DROP TABLE metric_method'); + await LocalDb.close(); + opens.clear(); + final reader = await LocalDb.readExecutor(); + expect((await reader.rawQuery('PRAGMA user_version')).first.values.first, + LocalDb.schemaVersion); + expect(await reader.rawQuery("SELECT name FROM sqlite_master " + "WHERE type='table' AND name='metric_method'"), hasLength(1)); + expect(opens, hasLength(2), reason: 'writer repair must precede second open'); + expect(opens.first.path, path); + expect(opens.first.options!.onOpen, isNotNull); + expect(opens.last.options!.onOpen, isNull); + }); + + test('query_only rejects a write through the reader', () async { + final reader = await LocalDb.readExecutor(); + await expectLater(reader.insert('sync_cursor', + {'name': 'reader_must_not_write', 'value': 'x', 'updated_at': 1}), + throwsA(isA())); + expect((await reader.rawQuery('PRAGMA query_only')).first.values.first, 1); + expect(await LocalDb.getCursor('reader_must_not_write'), isNull); + }); + + test('screen read completes during held write and sees only committed rows', () async { + await _putDay(_day); + // Open the reader before taking the write lock; opening is tested above. + await LocalDb.readExecutor(); + Future>? pending; + try { + await holdingWrite(() async { + pending = repo.getDayOverview(_day); + expect((await pending!.timeout(_wait))['readiness'], 61); + pending = repo.getDayOverview('2026-01-11'); + expect(await pending!.timeout(_wait), isEmpty, + reason: 'in-flight insert is invisible to the reader'); + }); + } finally { + await pending; + } + }); + + test('a new routed read sees a completed writer commit', () async { + await _putDay(_day); + final writer = await writerId(); + await repo.getDayOverview(_day); // establish the first snapshot + await _putDay('2026-01-11', readiness: 88); + sql.clear(); + expect((await repo.getDayOverview('2026-01-11'))['readiness'], 88); + await readerQueries('day_result', writer); + }); + + test('rolled-back rows never become visible on the reader', () async { + await _putDay(_day); + await LocalDb.readExecutor(); + Future>? pending; + try { + await holdingWrite(() async { + pending = repo.getDayOverview('2026-01-11'); + expect(await pending!.timeout(_wait), isEmpty); + }, rollback: true); + } finally { + await pending; + } + expect(await repo.getDayOverview('2026-01-11'), isEmpty); + }); + + test('failed reader open falls back to writer and returns stored values', () async { + await _putDay(_day); + var attempted = 0; + LocalDb.debugReaderOpener = (path) async { + attempted++; + throw StateError('simulated reader open failure'); + }; + final writer = await writerId(); + sql.clear(); + expect((await repo.getDayOverview(_day))['readiness'], 61); + expect(attempted, greaterThan(0), reason: 'must exercise the failed open'); + expect(sql.where((e) => e.sql.contains('from day_result')).map((e) => e.id), + everyElement(writer)); + expect(opens.where((e) => e.db?.isOpen ?? false), hasLength(1)); + }); + + test('non-WAL reader is rejected and closed before fallback', () async { + await _putDay(_day); + final writer = await LocalDb.instance; + await writer.rawQuery('PRAGMA journal_mode=DELETE'); + Database? candidate; + LocalDb.debugReaderOpener = (path) async => candidate = + await databaseFactory.openDatabase(path, + options: OpenDatabaseOptions(singleInstance: false)); + expect((await repo.getDayOverview(_day))['readiness'], 61); + expect(candidate, isNotNull, reason: 'must attempt the second open'); + expect(candidate!.isOpen, isFalse, reason: 'failed candidate must not leak'); + expect(await LocalDb.readExecutor(), same(writer)); + }); + + test('background reads use writer; resume permits a lazy fresh reader', () async { + final writer = await LocalDb.instance; + await LocalDb.setForeground(false); + expect(await LocalDb.readExecutor(), same(writer)); + expect(opens, hasLength(1)); + await LocalDb.setForeground(true); + expect(opens, hasLength(1)); + final reader = await LocalDb.readExecutor(); + expect(identical(reader, writer), isFalse); + await LocalDb.setForeground(false); + expect(reader.isOpen, isFalse); + expect(writer.isOpen, isTrue); + expect(await LocalDb.readExecutor(), same(writer)); + await LocalDb.setForeground(true); + expect(opens, hasLength(2), reason: 'resume does not eagerly reopen'); + final reopened = await LocalDb.readExecutor(); + expect(identical(reopened, reader), isFalse); + expect(identical(reopened, writer), isFalse); + }); + + test('LocalDb.close closes reader before writer', () async { + final writer = await writerId(); + final reader = idOf(await LocalDb.readExecutor()); + closes.clear(); + await LocalDb.close(); + expect(closes, [reader, writer], reason: 'reader must not outlive its file owner'); + }); + + for (final background in [true, false]) { + test('${background ? 'background' : 'close'} drains an in-flight read lease', () async { + await _putDay(_day); + final entered = Completer(); + final release = Completer(); + final reading = LocalDb.withReadExecutor((db) async { + entered.complete(); + await release.future; + return db.query('day_result', where: 'day_id = ?', whereArgs: [_day]); + }).then((rows) => rows, onError: (Object e, StackTrace s) => e); + await entered.future.timeout(_wait); + var finished = false; + final closing = (background ? LocalDb.setForeground(false) : LocalDb.close()) + .then((_) => finished = true); + Object? result; + try { + // Only a failure bound. The lease/release completers define ordering. + await Future.delayed(const Duration(milliseconds: 100)); + expect(finished, isFalse, reason: 'close must wait for the active lease'); + } finally { + release.complete(); + result = await reading.timeout(_wait); + await closing.timeout(_wait); + } + expect(result, isA>>(), + reason: 'the leased query must never run on a closed handle'); + expect(opens.where((e) => e.db?.isOpen ?? false), + hasLength(background ? 1 : 0)); + }); + + test('${background ? 'background' : 'close'} invalidates a pending reader open', () async { + await _putDay(_day); + final entered = Completer(); + final release = Completer(); + Database? candidate; + LocalDb.debugReaderOpener = (path) async { + final opened = await databaseFactory.openDatabase(path, + options: OpenDatabaseOptions(singleInstance: false)); + candidate = opened; + entered.complete(); + await release.future; + return opened; + }; + final reading = LocalDb.withReadExecutor((db) => db.query('day_result')) + .then((rows) => rows, onError: (Object e, StackTrace s) => e); + Future? closing; + Object? result; + try { + await entered.future.timeout(_wait, + onTimeout: () => fail('reader opener was never called')); + closing = background ? LocalDb.setForeground(false) : LocalDb.close(); + await Future.delayed(Duration.zero); + } finally { + release.complete(); + result = await reading.timeout(_wait); + await closing?.timeout(_wait); + } + expect(result, isA>>(), + reason: 'pending screen read must finish safely before shutdown'); + expect(candidate!.isOpen, isFalse, reason: 'stale opening must not leak'); + expect(opens.where((e) => e.db?.isOpen ?? false), + hasLength(background ? 1 : 0)); + }); + } + + test('wipe invalidates the reader and the next read sees the empty store', () async { + await _putDay(_day); + final reader = await LocalDb.readExecutor(); + await LocalDb.wipeAll(); + expect(reader.isOpen, isFalse, reason: 'store reset invalidates reader'); + expect(await repo.getDayOverview(_day), isEmpty); + expect(identical(await LocalDb.readExecutor(), reader), isFalse); + }); + + test('restore merge invalidates the reader and exposes restored rows', () async { + await _putDay(_day); + final writer = await LocalDb.instance; + // This backup is a real current-schema file, not a hand-built partial schema. + await writer.rawQuery('PRAGMA wal_checkpoint(TRUNCATE)'); + final backup = p.join(dir.path, 'restore_$serial.db'); + await File(writer.path).copy(backup); + await LocalDb.wipeAll(); + final reader = await LocalDb.readExecutor(); + await LocalDb.importFromDbFile(backup); + expect(reader.isOpen, isFalse, reason: 'merge invalidates old screen handle'); + expect((await repo.getDayOverview(_day))['readiness'], 61); + }); + + test('writer reopen retires an old reader before opening the file again', + () async { + // _openWriter retires the reader before _openOrRebuild touches the file, + // which is what keeps a rebuild from quarantining a file a reader still + // holds. A rebuild cannot be forced here: closing the live reader + // checkpoints its WAL frames back into the main file first. + await _putDay(_day); + final writer = await LocalDb.instance; + final reader = await LocalDb.readExecutor(); + expect(identical(reader, writer), isFalse, reason: 'needs a live reader'); + await writer.close(); + final reopened = await LocalDb.instance; + expect(identical(reopened, writer), isFalse); + expect(reader.isOpen, isFalse, reason: 'a reopened writer cannot keep the old reader'); + expect((await repo.getDayOverview(_day))['readiness'], 61); + expect(identical(await LocalDb.readExecutor(), reader), isFalse); + }); + + // Explicit screen allow-list. Default LocalDb callers below remain writers. + final routes = <({String name, Future Function() read, List tables})>[ + (name: '_bundle/dayResult', read: () => repo.getDayOverview(_day), + tables: ['day_result']), + (name: '_latestBundle/recentDayResults', read: () => repo.getDayOverview(todayLabel()), + tables: ['day_result']), + (name: '_crossDayArtifact/baseline', read: () => repo.getInsights(), + tables: ['baselines', 'metric_method']), + (name: 'availableDays/availableDayIds', read: () => repo.availableDays(), + tables: ['day_result']), + (name: 'sleepWindows/sleepWindowRows', read: () => repo.sleepWindows(), + tables: ['day_result']), + (name: 'getChart/series+provenance', read: () => repo.getChart('rhr'), + tables: ['metric_series', 'metric_series_version', 'metric_method']), + (name: 'getDayTimeline/events+sessions', read: () => repo.getDayTimeline(_day), + tables: ['day_result', 'events', 'sessions', 'session_sensor', 'decoded_onehz']), + (name: 'getSessions/sessionsInRange', read: () => repo.getSessions(from: 0, to: 1 << 40), + tables: ['sessions', 'session_sensor', 'decoded_onehz']), + ]; + for (final route in routes) { + test('routes ${route.name} through reader', () async { + await _putDay(_day); + await LocalDb.putBaseline('crossday', '{}'); + final writer = await LocalDb.instance; + await writer.insert('sessions', {'id': 's', 'start_ts': localDayStartSec(_day)!, + 'end_ts': localDayStartSec(_day)! + 60, 'status': 'done', + 'type': 'run', 'created_at': 1}); + final id = idOf(writer); + sql.clear(); + await route.read(); + for (final table in route.tables) { + await readerQueries(table, id); + } + }); + } + + test('shared LocalDb helpers still use writer by default for compute', () async { + await _putDay(_day); + await LocalDb.readExecutor(); + final writer = await writerId(); + sql.clear(); + await LocalDb.dayResult(_day); + await LocalDb.recentDayResults(2); + await LocalDb.availableDayIds(); + await LocalDb.sleepWindowRows(2); + await LocalDb.baseline('crossday'); + await LocalDb.metricSeries('rhr'); + await LocalDb.metricSeriesVersions(); + await LocalDb.metricNotOursDates('rhr'); + await LocalDb.eventsInRange(0, 1 << 40); + await LocalDb.sessionsInRange(0, 1 << 40); + expect(sql, isNotEmpty); + expect(sql.map((e) => e.id), everyElement(writer)); + }); + + test('getToday freshness refresh and reread stay on writer', () async { + await LocalDb.readExecutor(); + final writer = await writerId(); + sql.clear(); + await repo.getToday(); // missing freshness triggers the real refresh + final freshness = sql.where((e) => e.sql.contains('compute_freshness')).toList(); + expect(freshness.any((e) => e.type == SqliteSqlCommandType.insert), isTrue); + expect(freshness.where((e) => e.type == SqliteSqlCommandType.query).length, + greaterThanOrEqualTo(2)); + expect(freshness.map((e) => e.id), everyElement(writer)); + final refreshed = await LocalDb.computeFreshness('today'); + expect(jsonDecode(refreshed!['payload_json'] as String)['today_day'], todayLabel()); + }); + + test('frozenHeadline orphan lookup and compare-delete stay on writer', () async { + await LocalDb.setCursor(LocalDb.kFrozenHeadlineCursor, + '{"day":"2026-01-01","value":71}'); + await LocalDb.readExecutor(); + final writer = await writerId(); + sql.clear(); + expect(await LocalDb.frozenHeadline(), isNull); + expect(sql.any((e) => e.type == SqliteSqlCommandType.delete && + e.sql.contains('sync_cursor')), isTrue); + expect(sql.map((e) => e.id), everyElement(writer)); + expect(await LocalDb.getCursor(LocalDb.kFrozenHeadlineCursor), isNull); + }); + + for (final failing in [false, true]) { + test('safe-trim ${failing ? 'rollback' : 'commit'} keeps FULL bracket on writer', () async { + await LocalDb.readExecutor(); + final writer = await writerId(); + final handle = await LocalDb.instance; + if (failing) { + await handle.execute('CREATE TRIGGER refuse_cursor BEFORE INSERT ON sync_cursor ' + "WHEN new.name = 'strap_trim' BEGIN SELECT RAISE(ABORT, 'refuse trim'); END"); + } + final d = DrainController(onRecord: (sample, raw) async {}, + onRecordsBatch: null, onArchive: null, log: (_) {}, + onCommit: (raws, samples, token, {archives, ecgRawPackets, deviceFamily}) => + LocalDb.commitSyncBatch(raws, samples, trimToken: token, + archives: archives, ecgRawPackets: ecgRawPackets, deviceFamily: deviceFamily)); + d.onHistoricalRecord(_raw(5001), _sample(5001), 24); + sql.clear(); + final durable = await d.commit(const [1, 2, 3, 4, 5, 6, 7, 8]); + expect(durable, !failing, reason: 'false must withhold the engine ACK'); + final bracket = sql.where((e) => e.sql.startsWith('pragma synchronous=')).toList(); + expect(bracket.map((e) => e.sql), ['pragma synchronous=full', 'pragma synchronous=normal']); + expect(bracket.map((e) => e.id), everyElement(writer)); + final boundary = sql.where((e) => e.sql.startsWith('begin') || + e.sql == 'commit' || e.sql == 'rollback').toList(); + expect(boundary, isNotEmpty); + expect(boundary.map((e) => e.id), everyElement(writer)); + expect(sql.indexOf(bracket.first), lessThan(sql.indexOf(boundary.first))); + expect(sql.indexOf(bracket.last), greaterThan(sql.indexOf(boundary.last))); + expect((await handle.rawQuery('PRAGMA synchronous')).first.values.first, 1); + expect(await LocalDb.getCursor('strap_trim'), failing ? null : '0102030405060708'); + final rows = await handle.query('decoded_onehz', where: 'counter = 5001'); + expect(rows, hasLength(failing ? 0 : 1)); + final beats = await handle.query('decoded_rr', + where: 'device_id = ? AND ts_ms = ?', whereArgs: ['', 1750005001000]); + expect(beats, hasLength(failing ? 0 : 2)); + expect(await LocalDb.getCursorInt('counter_hw'), failing ? null : 5001); + expect(d.bufferedRecords, failing ? 1 : 0); + if (!failing) { + final writes = sql.where((e) => e.type == SqliteSqlCommandType.insert && + (e.sql.contains('decoded_onehz') || e.sql.contains('decoded_rr') || + e.sql.contains('sync_cursor'))).toList(); + expect(writes.any((e) => e.sql.contains('decoded_onehz')), isTrue); + expect(writes.any((e) => e.sql.contains('decoded_rr')), isTrue); + expect(writes.any((e) => e.sql.contains('sync_cursor')), isTrue); + expect(writes.map((e) => e.id), everyElement(writer)); + expect(writes.first.txn, isNotNull); + expect(writes.map((e) => e.txn).toSet(), {writes.first.txn}, + reason: 'decoded seconds, beats and ACK cursor share the durable transaction'); + } + }); + } +} diff --git a/test/wal_reader_wiring_test.dart b/test/wal_reader_wiring_test.dart new file mode 100644 index 000000000..1ed3df617 --- /dev/null +++ b/test/wal_reader_wiring_test.dart @@ -0,0 +1,106 @@ +// Structural guards for the app observer and the private radio ACK boundary. +// Database lifecycle and durability behavior run against real files in +// wal_reader_connection_test.dart. These guards cover their production callers +// without booting AppState's platform services or introducing another observer. +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; + +import 'support/dart_source.dart'; + +String _body(String code, String method) { + final call = RegExp('(?:void|Future)\\s+$method\\s*\\(') + .firstMatch(code)!; + var cursor = code.indexOf('(', call.start); + var depth = 0; + do { + if (code[cursor] == '(') depth++; + if (code[cursor] == ')') depth--; + cursor++; + } while (depth > 0); + final start = code.indexOf('{', cursor); + cursor = start; + do { + if (code[cursor] == '{') depth++; + if (code[cursor] == '}') depth--; + cursor++; + } while (depth > 0); + return code.substring(start + 1, cursor - 1); +} + +// Find the if/else-if branch that handles a given lifecycle state. Grouped +// OR conditions are allowed, as in the fork. Comments/strings are already gone. +String? _stateBranch(String code, String state) { + for (final condition in RegExp(r'\bif\s*\(', multiLine: true).allMatches(code)) { + var cursor = code.indexOf('(', condition.start); + final start = cursor; + var depth = 0; + do { + if (code[cursor] == '(') depth++; + if (code[cursor] == ')') depth--; + cursor++; + } while (depth > 0); + if (!RegExp('state\\s*==\\s*AppLifecycleState\\.$state\\b') + .hasMatch(code.substring(start, cursor))) { + continue; + } + final block = code.indexOf('{', cursor); + cursor = block; + do { + if (code[cursor] == '{') depth++; + if (code[cursor] == '}') depth--; + cursor++; + } while (depth > 0); + return code.substring(block + 1, cursor - 1); + } + return null; +} + +void main() { + final app = stripCommentsAndStrings(File('lib/app.dart').readAsStringSync()); + final lifecycle = _body(app, 'didChangeAppLifecycleState'); + + test('visible cold start enables lazy reads through the existing app observer', () { + final init = _body(app, 'initState'); + expect(init, contains('WidgetsBinding.instance.addObserver(this)')); + expect(init, contains('LocalDb.setForeground('), + reason: 'cold foreground start need not emit a resumed transition'); + expect(init, contains('lifecycleState'), reason: 'cold background is excluded'); + for (final state in ['hidden', 'paused', 'detached']) { + expect(init, contains('AppLifecycleState.$state')); + } + }); + + test('resume enables lazy reads in the app lifecycle observer', () { + final branch = _stateBranch(lifecycle, 'resumed'); + expect(branch, isNotNull); + expect(branch, matches(RegExp(r'LocalDb\.setForeground\(\s*true\s*\)')), + reason: 'resume must allow the next screen read to reopen the reader'); + }); + + for (final state in ['hidden', 'paused', 'detached']) { + test('$state closes reader through the app lifecycle observer', () { + final branch = _stateBranch(lifecycle, state); + expect(branch, isNotNull, reason: '$state must reach LocalDb'); + expect(branch, matches(RegExp(r'LocalDb\.setForeground\(\s*false\s*\)')), + reason: '$state must drain reads and close the second connection'); + }); + } + + test('HISTORY_END awaits durable commit and gates the verbatim ACK token', () { + final ble = stripCommentsAndStrings(File('lib/ble/ble_engine.dart').readAsStringSync()); + final marker = _body(ble, '_handleSyncMarker'); + final commit = RegExp(r'final\s+durable\s*=\s*await\s+d\.commit\(m\.token\)') + .firstMatch(marker); + final ack = RegExp(r'buildHistoryResultOk\([^;]*m\.token!', dotAll: true) + .firstMatch(marker); + expect(commit, isNotNull); + expect(ack, isNotNull); + expect(ack!.start, greaterThan(commit!.end)); + final gate = marker.substring(commit.end, ack.start); + expect(gate, contains('commitDurable: durable')); + expect(gate, matches(RegExp( + r'if\s*\(verdict\s*!=\s*TrimAckVerdict\.send\)\s*\{[^}]*return\s*;', + dotAll: true)), reason: 'failed/partial commit must return before ACK'); + }); +} From 1c31e34a7d6339a4cf62008dbe281423401ff151 Mon Sep 17 00:00:00 2001 From: MikenRafter Date: Sat, 24 Oct 2026 04:35:41 -0600 Subject: [PATCH 2/2] fix(db): reader close failures stay local; chart coverage reads use the reader - A throwing reader close() in _retireReader is swallowed like the one in _openReader: Android may already have closed the read-only handle, and backgrounding, wipe, restore and close must still go through. - getChart's multi-device coverage query runs on the same read executor as the chart's other reads instead of queueing behind the writer. - The cold-start lifecycle mapping is a function with a behavioral test, and the getToday freshness test accepts the day label from either side of the call, so it cannot fail across local midnight. --- lib/app.dart | 14 +++-- lib/data/db.dart | 9 +++- lib/data/local_repository_impl.dart | 5 +- test/wal_reader_connection_test.dart | 76 +++++++++++++++++++++++++++- test/wal_reader_wiring_test.dart | 19 +++++-- 5 files changed, 109 insertions(+), 14 deletions(-) diff --git a/lib/app.dart b/lib/app.dart index 0e5146a75..5d9b00633 100644 --- a/lib/app.dart +++ b/lib/app.dart @@ -47,16 +47,22 @@ class OpenStrapApp extends StatefulWidget { State createState() => _OpenStrapAppState(); } +/// Whether screen reads may use the reader connection at this lifecycle state. +/// Null (no report yet) counts as visible; the observer corrects it on the +/// first report. +@visibleForTesting +bool screenReadsForeground(AppLifecycleState? state) => + state != AppLifecycleState.hidden && + state != AppLifecycleState.paused && + state != AppLifecycleState.detached; + class _OpenStrapAppState extends State with WidgetsBindingObserver { @override void initState() { super.initState(); WidgetsBinding.instance.addObserver(this); - final lifecycleState = WidgetsBinding.instance.lifecycleState; unawaited(LocalDb.setForeground( - lifecycleState != AppLifecycleState.hidden && - lifecycleState != AppLifecycleState.paused && - lifecycleState != AppLifecycleState.detached)); + screenReadsForeground(WidgetsBinding.instance.lifecycleState))); // Hand the BYOK provider config to AppState (briefing generation + the // key-aware AI notification schedule live there). WidgetsBinding.instance.addPostFrameCallback((_) { diff --git a/lib/data/db.dart b/lib/data/db.dart index 3d225010b..917319008 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -558,7 +558,11 @@ class LocalDb { await Future.wait(readerLeases.map((lease) => lease.future)); await opening; if (identical(_readerOpening, opening)) _readerOpening = null; - if (reader != null) await reader.close(); + // Read-only handle: Android may already have closed it natively. A failed + // close must not block a store change or surface from setForeground. + if (reader != null) { + try { await reader.close(); } catch (_) {} + } }(); _readerClosing = closing; return () async { @@ -8734,9 +8738,10 @@ class LocalDb { required List signals, required int fromSec, required int toSec, + DatabaseExecutor? executor, }) async { if (signals.isEmpty || toSec <= fromSec) return const {}; - final db = await instance; + final db = executor ?? await instance; final marks = List.filled(signals.length, '?').join(','); final rows = await db.rawQuery( 'SELECT DISTINCT device_id, start_ts, end_ts FROM device_coverage ' diff --git a/lib/data/local_repository_impl.dart b/lib/data/local_repository_impl.dart index 89b3eb29b..c0fa8a075 100644 --- a/lib/data/local_repository_impl.dart +++ b/lib/data/local_repository_impl.dart @@ -2138,11 +2138,12 @@ class LocalRepositoryImpl extends LocalRepository { // single-device install runs no additional statement, ever. 'coverage_recording': signals.isEmpty || distinctDevices.length < 2 ? const >{} - : await LocalDb.coverageDevicesByDay( + : await LocalDb.withReadExecutor((db) => LocalDb.coverageDevicesByDay( signals: signals.toList(), fromSec: oldestDaySec, toSec: _nowSec(), - ), + executor: db, + )), }; } diff --git a/test/wal_reader_connection_test.dart b/test/wal_reader_connection_test.dart index e4eb125aa..43886c047 100644 --- a/test/wal_reader_connection_test.dart +++ b/test/wal_reader_connection_test.dart @@ -258,6 +258,37 @@ void main() { expect(opens.where((e) => e.db?.isOpen ?? false), hasLength(1)); }); + test('failed reader open retries after background, not on every read', () async { + await _putDay(_day); + var attempted = 0; + LocalDb.debugReaderOpener = (path) async { + attempted++; + throw StateError('simulated reader open failure'); + }; + await repo.getDayOverview(_day); + await repo.getDayOverview(_day); + expect(attempted, 1, reason: 'a broken open is not retried per screen read'); + await LocalDb.setForeground(false); + await LocalDb.setForeground(true); + LocalDb.debugReaderOpener = null; + final writer = await LocalDb.instance; + final reader = await LocalDb.readExecutor(); + expect(identical(reader, writer), isFalse, reason: 'next foreground retries'); + }); + + test('a throwing reader close neither fails backgrounding nor blocks close', () async { + LocalDb.debugReaderOpener = (path) async => _CloseThrows( + await databaseFactory.openDatabase(path, + options: OpenDatabaseOptions(singleInstance: false))); + final writer = await LocalDb.instance; + final reader = await LocalDb.readExecutor(); + expect(reader, isA<_CloseThrows>()); + await LocalDb.setForeground(false); + await LocalDb.setForeground(true); + await LocalDb.close(); + expect(writer.isOpen, isFalse, reason: 'close must still reach the writer'); + }); + test('non-WAL reader is rejected and closed before fallback', () async { await _putDay(_day); final writer = await LocalDb.instance; @@ -442,6 +473,22 @@ void main() { }); } + test('routes getChart multi-device coverage through reader', () async { + await _putDay(_day); + final writer = await LocalDb.instance; + await writer.insert('metric_series_version', + {'date': _day, 'algo_version': kAlgoVersion, 'coverage_devices': 'a,b'}, + conflictAlgorithm: ConflictAlgorithm.replace); + final noon = localDayStartSec(_day)! + 12 * 3600; + await writer.insert('device_coverage', {'device_id': 'b', 'signal': 'hr', + 'start_ts': noon, 'end_ts': noon + 3600}); + final id = idOf(writer); + sql.clear(); + final chart = await repo.getChart('rhr', signals: {'hr'}); + expect(chart['coverage_recording'], isNotEmpty, reason: 'fixture must query coverage'); + await readerQueries('device_coverage', id); + }); + test('shared LocalDb helpers still use writer by default for compute', () async { await _putDay(_day); await LocalDb.readExecutor(); @@ -465,14 +512,16 @@ void main() { await LocalDb.readExecutor(); final writer = await writerId(); sql.clear(); + final before = todayLabel(); await repo.getToday(); // missing freshness triggers the real refresh + final after = todayLabel(); final freshness = sql.where((e) => e.sql.contains('compute_freshness')).toList(); expect(freshness.any((e) => e.type == SqliteSqlCommandType.insert), isTrue); expect(freshness.where((e) => e.type == SqliteSqlCommandType.query).length, greaterThanOrEqualTo(2)); expect(freshness.map((e) => e.id), everyElement(writer)); final refreshed = await LocalDb.computeFreshness('today'); - expect(jsonDecode(refreshed!['payload_json'] as String)['today_day'], todayLabel()); + expect(jsonDecode(refreshed!['payload_json'] as String)['today_day'], anyOf(before, after)); }); test('frozenHeadline orphan lookup and compare-delete stay on writer', () async { @@ -539,3 +588,28 @@ void main() { }); } } + +// A reader whose native handle is already gone: close() throws, as on Android. +class _CloseThrows implements Database { + _CloseThrows(this._db); + final Database _db; + @override + String get path => _db.path; + @override + bool get isOpen => _db.isOpen; + @override + Future execute(String sql, [List? arguments]) => + _db.execute(sql, arguments); + @override + Future>> rawQuery(String sql, + [List? arguments]) => + _db.rawQuery(sql, arguments); + @override + Future close() async { + await _db.close(); + throw StateError('simulated native close failure'); + } + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + diff --git a/test/wal_reader_wiring_test.dart b/test/wal_reader_wiring_test.dart index 1ed3df617..5fa067314 100644 --- a/test/wal_reader_wiring_test.dart +++ b/test/wal_reader_wiring_test.dart @@ -4,7 +4,9 @@ // without booting AppState's platform services or introducing another observer. import 'dart:io'; +import 'package:flutter/widgets.dart' show AppLifecycleState; import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/app.dart' show screenReadsForeground; import 'support/dart_source.dart'; @@ -63,12 +65,19 @@ void main() { test('visible cold start enables lazy reads through the existing app observer', () { final init = _body(app, 'initState'); expect(init, contains('WidgetsBinding.instance.addObserver(this)')); - expect(init, contains('LocalDb.setForeground('), + expect(init, matches(RegExp(r'LocalDb\.setForeground\(\s*screenReadsForeground\(\s*' + r'WidgetsBinding\.instance\.lifecycleState\s*\)\s*\)')), reason: 'cold foreground start need not emit a resumed transition'); - expect(init, contains('lifecycleState'), reason: 'cold background is excluded'); - for (final state in ['hidden', 'paused', 'detached']) { - expect(init, contains('AppLifecycleState.$state')); - } + }); + + test('cold-start lifecycle state maps to the reader foreground flag', () { + // Null: no lifecycle report yet; the observer corrects it on the first one. + expect(screenReadsForeground(null), isTrue); + expect(screenReadsForeground(AppLifecycleState.resumed), isTrue); + expect(screenReadsForeground(AppLifecycleState.inactive), isTrue); + expect(screenReadsForeground(AppLifecycleState.hidden), isFalse); + expect(screenReadsForeground(AppLifecycleState.paused), isFalse); + expect(screenReadsForeground(AppLifecycleState.detached), isFalse); }); test('resume enables lazy reads in the app lifecycle observer', () {