diff --git a/lib/compute/derivation_engine.dart b/lib/compute/derivation_engine.dart index 2ba99a93f..89b359a57 100644 --- a/lib/compute/derivation_engine.dart +++ b/lib/compute/derivation_engine.dart @@ -3681,7 +3681,7 @@ class DerivationEngine { onDayDone?.call(dayId, completed, orderedDays.length); } - await runWithConcurrency(orderedDays, _deriveConcurrency, processDay); + await _runDayPool(orderedDays, processDay); // BEFORE the prune below: a day still owed a strain score is re-scored // while its raw exists (and the prune holds it if this fails). _diag['stage'] = 'quiet_repass'; @@ -3928,7 +3928,7 @@ class DerivationEngine { onDayDone?.call(dayId, completed, orderedDays.length); } - await runWithConcurrency(orderedDays, _deriveConcurrency, processDay); + await _runDayPool(orderedDays, processDay); done += await _quietLevelRepass(profile, dataNowSec); // A SELECTED re-analyze that happens to cover the whole raw history, with // every day resolved, is a full restage by any other name — it re-derived @@ -4060,9 +4060,15 @@ class DerivationEngine { } // Tonight's sleep is the NEXT day's main sleep; it is only known once // that day has derived, and this day keeps re-deriving until it - // finalizes, so it converges on the following pass. + // finalizes, so it converges on the following pass. A day that would + // finalize in the same pass its next day is still deriving is held back + // ([PreparedDerivationDay.tonightPending]), or which of the two finished + // first would decide its stored result for good. + final nextDayId = localDateLabel(dayEnd); + final nextSpan = + storedWindowSpan(await LocalDb.sleepWindowJsonFor(nextDayId)); final tonight = tonightSleep( - storedWindowSpan(await LocalDb.sleepWindowJsonFor(localDateLabel(dayEnd))), + nextSpan, dayStartSec: dayStart, dayEndSec: dayEnd, sleepOffsetSec: candidate.present ? candidate.sleepOffsetSec : 0, @@ -4073,6 +4079,11 @@ class DerivationEngine { sleepSub: sleepSub, tonightSleepOnsetSec: tonight?.startSec ?? 0, tonightSleepOffsetSec: tonight?.endSec ?? 0, + // No span, not just no row: a stored '{}' is no window either. Not + // `tonight == null`: a next night that starts after midnight has a + // window but no tonight, and holding on it would hold every such day. + tonightPending: (nextSpan == null && _passUnfinished.contains(nextDayId)) || + debugHoldTonight.contains(dayId), ownership: ownership, priority: priority, ); @@ -5247,7 +5258,7 @@ class DerivationEngine { )..sort((a, b) => b.compareTo(a)); // newest first, like the sweep if (days.isEmpty) return 0; _log('derive: re-scoring ${days.length} day(s) owed a quiet level'); - await runWithConcurrency(days, _deriveConcurrency, (dayId) async { + await _runDayPool(days, (dayId) async { try { await debugQuietRepassHook?.call(dayId); final prepared = await _prepareTargetDay(dayId); @@ -5394,7 +5405,7 @@ class DerivationEngine { onDayDone?.call(dayId, completed, orderedDays.length); } - await runWithConcurrency(orderedDays, _deriveConcurrency, processDay); + await _runDayPool(orderedDays, processDay); if (orderedDays.isNotEmpty) { await _refreshBaselines(); @@ -5603,6 +5614,33 @@ class DerivationEngine { /// real wall-clock speedup from the device's other cores — in the FOREGROUND. /// A headless background slot has no spare cores to soak up, so it takes one /// lane; [DerivePacing] owns that decision and explains it. + /// The days this pass's pool has not finished yet ([_runDayPool]). + final Set _passUnfinished = {}; + + /// Test seam: days prepared as [PreparedDerivationDay.tonightPending] + /// whatever the next day's state, so a hold can be exercised without + /// racing two workers. Empty in production. + @visibleForTesting + static Set debugHoldTonight = {}; + + /// [runWithConcurrency] over [days], tracking which are still unfinished so + /// a day can tell its next day is mid-derive ([_prepareTargetDay]). + Future _runDayPool( + List days, Future Function(String dayId) body) async { + _passUnfinished.addAll(days); + try { + await runWithConcurrency(days, _deriveConcurrency, (dayId) async { + try { + await body(dayId); + } finally { + _passUnfinished.remove(dayId); + } + }); + } finally { + _passUnfinished.removeAll(days); + } + } + int get _deriveConcurrency { try { return _pacing.concurrency(Platform.numberOfProcessors); @@ -6424,8 +6462,9 @@ class DerivationEngine { // forceFinalize. Locking an all-absent row is what made the destructive // re-analyze permanent; leaving it unlocked means a later pass (or a // restored/backfilled substrate) can still fill the day in. - final finalized = - !producedNothing && (forceFinalize || (ageFinalized && secondHalfOk)); + final finalized = !producedNothing && + (forceFinalize || + (ageFinalized && secondHalfOk && !day.tonightPending)); // A failed/timed-out second half yields a headline-only bundle, and // putDayResult replaces the row wholesale — so re-deriving an already @@ -6454,7 +6493,10 @@ class DerivationEngine { finalizedByAge: finalized, ); effectivePartial = outcome.partial; - effectiveFinalized = outcome.finalized; + // The tonight hold applies here too: carrying a finalized row + // forward would otherwise lock a day the hold kept open. + effectiveFinalized = + outcome.finalized && (forceFinalize || !day.tonightPending); _log('derive ${day.date}: second half failed — carried the previous ' "result's detail blocks forward (v$prevVersion -> v$kAlgoVersion, " 'partial=$effectivePartial)'); @@ -6693,7 +6735,9 @@ class DerivationEngine { ], ); } - return secondHalfOk; + // A held day is not settled either: it must derive again next pass to + // read tonight's window, so it may not be recorded as unchanged. + return secondHalfOk && !day.tonightPending; } /// Bank one day's (phone steps, counter ticks) pair for the (family, diff --git a/lib/compute/derive_prepare.dart b/lib/compute/derive_prepare.dart index 6e37321f6..316077e2b 100644 --- a/lib/compute/derive_prepare.dart +++ b/lib/compute/derive_prepare.dart @@ -38,6 +38,11 @@ class PreparedDerivationDay { final int tonightSleepOnsetSec; final int tonightSleepOffsetSec; + /// The next day is being derived in this same pass and has no stored + /// window yet, so tonight's sleep is unknown only for now. Such a day is not + /// finalized this pass: the next pass reads the window and locks it then. + final bool tonightPending; + /// M5: the resolved exclusive-owner spans, one entry per anchor signal /// (`hr1Hz`, `rrIntervals`), over the union window `_prepareTargetDay` /// resolved once. Empty (never populated) on the import path — a bulk @@ -73,6 +78,7 @@ class PreparedDerivationDay { this.sleepSource = 'auto', this.tonightSleepOnsetSec = 0, this.tonightSleepOffsetSec = 0, + this.tonightPending = false, this.ownership = const {}, this.priority = const {}, }) : napSub = napSub ?? daySub; @@ -92,6 +98,7 @@ class PreparedDerivationDay { 'nap_sub': napSub.toJson(), 'tonight_sleep_onset_sec': tonightSleepOnsetSec, 'tonight_sleep_offset_sec': tonightSleepOffsetSec, + 'tonight_pending': tonightPending, // ponytail: `ownership` is NOT round-tripped here. Grepped for real // callers of PreparedDerivationDay.toJson()/fromJson() outside this file // — none exist (only `candidate.toPreparedDay(...)` at @@ -120,6 +127,7 @@ class PreparedDerivationDay { (m['tonight_sleep_onset_sec'] as num?)?.toInt() ?? 0, tonightSleepOffsetSec: (m['tonight_sleep_offset_sec'] as num?)?.toInt() ?? 0, + tonightPending: m['tonight_pending'] == true, daySub: daySub, sleepSub: Substrate.fromJson( ((m['sleep_sub'] as Map?) ?? const {}).cast(), @@ -260,6 +268,7 @@ class SleepSessionCandidate { Substrate? napSub, int tonightSleepOnsetSec = 0, int tonightSleepOffsetSec = 0, + bool tonightPending = false, Map> ownership = const {}, Map> priority = const {}, }) => PreparedDerivationDay( @@ -286,6 +295,7 @@ class SleepSessionCandidate { sleepSub: sleepSub, tonightSleepOnsetSec: tonightSleepOnsetSec, tonightSleepOffsetSec: tonightSleepOffsetSec, + tonightPending: tonightPending, ownership: ownership, priority: priority, ); diff --git a/test/derive_tonight_pending_test.dart b/test/derive_tonight_pending_test.dart new file mode 100644 index 000000000..59b1ed092 --- /dev/null +++ b/test/derive_tonight_pending_test.dart @@ -0,0 +1,118 @@ +// A day old enough to finalize, whose next day has not stored its window yet +// in the same pass, is held: not finalized and not recorded as unchanged, so +// the next pass reads tonight's window and locks the day then. The hold is +// forced through the test seam rather than by racing two workers. + +import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/compute/derivation_engine.dart'; +import 'package:openstrap_edge/compute/profile.dart'; +import 'package:openstrap_edge/data/db.dart'; +import 'package:path/path.dart' as p; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:sqflite_common_ffi/sqflite_ffi.dart'; + +import 'support/synthetic_day.dart'; + +const _profile = Profile(ageYears: 32, weightKg: 72, heightCm: 176, sex: 'm'); + +// Three days of data: the day before the first is old enough to finalize on +// the first pass. +final _days = [ + for (var i = 0; i < 3; i++) SyntheticDay(DateTime(2026, 7, 13 + i)), +]; +const _held = '2026-07-12'; + +Future _seed(String name) async { + SharedPreferences.setMockInitialValues({}); + await LocalDb.close(); + LocalDb.dbName = name; + await databaseFactory.deleteDatabase( + p.join(await databaseFactory.getDatabasesPath(), name), + ); + var committed = 0; + for (final d in _days) { + for (final (raws, samples) in d.primaryBatches()) { + final keep = [ + for (var j = 0; j < raws.length; j++) + if (raws[j].recTs! > committed) j, + ]; + if (keep.isEmpty) continue; + await LocalDb.commitSyncBatch( + [for (final j in keep) raws[j]], + [for (final j in keep) samples[j]], + deviceFamily: 'gen4', + ); + committed = raws[keep.last].recTs!; + } + } +} + +void main() { + setUpAll(() { + sqfliteFfiInit(); + databaseFactory = databaseFactoryFfi; + }); + + tearDown(() async { + DerivationEngine.debugHoldTonight = {}; + DerivationEngine.debugSecondHalfHook = null; + await LocalDb.close(); + }); + + test( + 'a held day is neither finalized nor fingerprinted, then settles', + () async { + await _seed('derive_tonight_pending.db'); + + // Two held passes: the second derives from unchanged inputs, which is + // where a fingerprint would be recorded and the day skipped for good. + DerivationEngine.debugHoldTonight = {_held}; + for (var i = 0; i < 2; i++) { + await DerivationEngine().run(_profile, heavy: true); + expect(await LocalDb.dayResult(_held), isNotNull); + expect( + await LocalDb.finalizedDayIds(kAlgoVersion), + isNot(contains(_held)), + reason: 'held pass $i', + ); + expect( + await LocalDb.deriveFingerprint(_held), + isNull, + reason: 'held pass $i', + ); + } + + DerivationEngine.debugHoldTonight = {}; + await DerivationEngine().run(_profile, heavy: true); + expect(await LocalDb.finalizedDayIds(kAlgoVersion), contains(_held)); + }, + timeout: const Timeout(Duration(minutes: 5)), + ); + + test( + 'a failed second half does not carry a finalized row past the hold', + () async { + await _seed('derive_tonight_pending_recovery.db'); + // A cold first pass may legitimately hold the day (the next day's + // window can land after it in the same pass), so a second pass is what + // guarantees the finalized row this test starts from. + await DerivationEngine().run(_profile, heavy: true); + await DerivationEngine().run(_profile, heavy: true); + expect(await LocalDb.finalizedDayIds(kAlgoVersion), contains(_held)); + + // Re-derived (forced) while held, with its second half failing: the + // recovery path carries the complete finalized row forward, and must not + // carry the lock with it. + DerivationEngine.debugHoldTonight = {_held}; + DerivationEngine.debugSecondHalfHook = (day) async { + if (day == _held) throw StateError('second half down'); + }; + await DerivationEngine().runDays(_profile, {_held}, force: true); + expect( + await LocalDb.finalizedDayIds(kAlgoVersion), + isNot(contains(_held)), + ); + }, + timeout: const Timeout(Duration(minutes: 5)), + ); +} diff --git a/test/derive_unchanged_skip_test.dart b/test/derive_unchanged_skip_test.dart index 647b20922..89194293e 100644 --- a/test/derive_unchanged_skip_test.dart +++ b/test/derive_unchanged_skip_test.dart @@ -66,7 +66,10 @@ Future _snapshot() async { if (!e.key.endsWith('_at') && e.key != 'updated_ts') e.key: e.value is String ? (e.value as String).replaceAll( - RegExp(r'"(updated|built)[A-Za-z_]*":\s*\d+'), '') + // Numbers and dates: built_for_day is the wall-clock + // day, so two halves either side of midnight differ. + RegExp(r'"(updated|built)[A-Za-z_]*":\s*(\d+|"[^"]*")'), + '') : e.value, }, ]; @@ -88,6 +91,59 @@ Future _snapshot() async { }); } +/// Where two [_snapshot]s first part: the table, the row, and each column +/// that differs, values cut short. +String _firstDifference(String a, String b) { + String cut(Object? v) { + final t = '$v'; + return t.length <= 400 ? t : '${t.substring(0, 400)}… (${t.length} chars)'; + } + + final x = jsonDecode(a) as Map; + final y = jsonDecode(b) as Map; + for (final table in x.keys) { + final rx = x[table] as List, ry = y[table] as List; + if (rx.length != ry.length) { + return '$table: ${rx.length} rows with the skip, ${ry.length} without'; + } + for (var i = 0; i < rx.length; i++) { + final m = rx[i] as Map, n = ry[i] as Map; + final cols = {...m.keys, ...n.keys} + .where((k) => jsonEncode(m[k]) != jsonEncode(n[k])) + .toList(); + if (cols.isEmpty) continue; + final key = m['day_id'] ?? m['date'] ?? m['key'] ?? m['id'] ?? i; + return [ + '$table row $i ($key) differs in ${cols.join(', ')}:', + for (final c in cols) + if (m[c] is String && n[c] is String) + _stringDifference(c as String, m[c] as String, n[c] as String) + else ...[ + ' $c with the skip: ${cut(m[c])}', + ' $c without: ${cut(n[c])}', + ], + ].join('\n'); + } + } + return 'snapshots differ outside the compared tables'; +} + +String _stringDifference(String col, String a, String b) { + var i = 0; + while (i < a.length && i < b.length && a.codeUnitAt(i) == b.codeUnitAt(i)) { + i++; + } + String around(String t) { + final from = i < 200 ? 0 : i - 200; + final to = i + 200 > t.length ? t.length : i + 200; + return t.substring(from, to); + } + + return ' $col first differs at char $i\n' + ' with the skip: …${around(a)}…\n' + ' without: …${around(b)}…'; +} + void main() { setUpAll(() { sqfliteFfiInit(); @@ -121,7 +177,10 @@ void main() { // The skip actually happened: by the last pass nothing was derived. expect(log.last.derived, 0, reason: '$log'); expect(log.last.unchanged, greaterThan(0), reason: '$log'); - expect(withSkip, everyDay); + // Not expect(withSkip, everyDay): on a mismatch that prints both + // multi-megabyte snapshots, and the CI log was cut off right there, so a + // failure read as a silent crash. Name the first difference instead. + if (withSkip != everyDay) fail(_firstDifference(withSkip, everyDay)); }, timeout: const Timeout(Duration(minutes: 10))); test('an override, new data, a profile change and a version bump re-derive',