fix(db): re-key decoded ledger off volatile counter onto rec_ts (unrecoverable 1 Hz data loss)

lib/compute/derivation_engine.dart

@@ -1836,13 +1836,16 @@ class DerivationEngine {
             rangePages: rangePages,
             rangeRows: rangeRows,
           );
-          final firstCounter = (decodedRows.first['counter'] as num?)?.toInt();
-          final lastCounter = (decodedRows.last['counter'] as num?)?.toInt();
-          final rrRows = firstCounter == null || lastCounter == null
+          // The page is ordered rec_ts ASC, so first = min second, last = max.
+          // decoded_rr shares the rec_ts key, so this pulls exactly the page's
+          // beats — no counter span (which broke across the strap's reboot reset).
+          final firstRecTs = (decodedRows.first['rec_ts'] as num?)?.toInt();
+          final lastRecTs = (decodedRows.last['rec_ts'] as num?)?.toInt();
+          final rrRows = firstRecTs == null || lastRecTs == null
               ? const <Map<String, dynamic>>[]
-              : await LocalDb.decodedRrByCounterRange(
-                  fromCounter: firstCounter,
-                  toCounter: lastCounter,
+              : await LocalDb.decodedRrByRecTsRange(
+                  fromRecTs: firstRecTs,
+                  toRecTs: lastRecTs,
                 );
           worker.send({'type': 'page', 'frames': decodedRows, 'rr': rrRows});
           final last = decodedRows.last;

lib/compute/derive_prepare.dart

@@ -431,11 +431,14 @@ class _PrepareAccumulator {
     List<Map<String, dynamic>> rrRows,
   ) {
     if (frames.isEmpty) return;
-    final rrByCounter = <int, List<Map<String, dynamic>>>{};
+    // Associate beats to frames by rec_ts (their shared key). The strap's counter
+    // resets on reboot, so grouping by counter mis-joined two seconds that reused
+    // one counter within a page.
+    final rrByRecTs = <int, List<Map<String, dynamic>>>{};
     for (final row in rrRows) {
-      final counter = _num(row['counter'])?.toInt();
-      if (counter == null) continue;
-      rrByCounter.putIfAbsent(counter, () => <Map<String, dynamic>>[]).add(row);
+      final recTs = _num(row['rec_ts'])?.toInt();
+      if (recTs == null) continue;
+      rrByRecTs.putIfAbsent(recTs, () => <Map<String, dynamic>>[]).add(row);
     }
     for (final row in frames) {
       final recTs = _num(row['rec_ts'])?.toInt();
@@ -454,9 +457,7 @@ class _PrepareAccumulator {
       // tsSec is what lets `Substrate.fromJson` tell "absent" (empty ⇒
       // zero-filled) from "present but zero".
       skinContact.add(_num(row['skin_contact'])?.toInt() ?? 0);
-      final counter = _num(row['counter'])?.toInt();
-      if (counter == null) continue;
-      final beats = rrByCounter[counter];
+      final beats = rrByRecTs[recTs];
       if (beats == null) continue;
       for (final beat in beats) {
         final rr = _num(beat['rr_ms'])?.toDouble();

test/db_integrity_test.dart

@@ -1,12 +1,10 @@
 // DB integrity regressions, run against the REAL LocalDb over sqflite_ffi:
 //
-//  1. decoded_rr ORPHAN GUARD — a post-reboot rec_ts collision (two counters,
-//     one second) must not strand the evicted counter's RR beats under a
-//     counter with no decoded_onehz row (the counter-joined prune can never
-//     select those → permanent leak, and the loser's extra beat indexes would
-//     survive the winner's UNIQUE(rr_ts_ms, beat_index) REPLACE).
-//  2. prune ORPHAN SWEEP — pre-existing orphans (written by pre-guard builds)
-//     are cleaned by pruneRawBeforeRecTs once their window is pruned.
+//  1. decoded_rr REC_TS KEY — a post-reboot rec_ts collision (two counters, one
+//     second) keeps exactly the winner's beats; the write path DELETEs the
+//     second's beats before reinserting, so a shrinking beat count strands none.
+//  2. prune BY REC_TS — pruneDecodedBeforeRecTs deletes decoded_rr by its rec_ts
+//     key (no counter subquery, no orphan sweep) and keeps recent beats.
 //  3. importFromDbFile FINALIZED protection — a foreign export never overwrites
 //     a locally-finalized (day_id, algo_version) day_result row; non-finalized
 //     rows keep the merge-REPLACE behavior.
@@ -62,45 +60,44 @@ void main() {
     expect(health['ok'], isTrue, reason: '$health');
   });
 
-  test('rec_ts collision leaves no orphaned decoded_rr beats', () async {
+  test('rec_ts collision leaves exactly the winner\'s beats', () async {
     const ts = 1780000000;
-    // Pre-reboot record: high counter, THREE beats.
+    // Two records for the SAME second, different counters (a reboot straddle).
+    // THREE beats then TWO. REPLACE on the rec_ts key keeps the newest row; the
+    // write path DELETEs the second's beats before reinserting, so no stale
+    // high-index beat survives.
     await LocalDb.insertRecord(_raw(ts, 100), _sample(ts, 100, [800, 810, 820]));
-    // Post-reboot record for the SAME second: counter reset low, TWO beats.
-    // REPLACE on UNIQUE(rec_ts) evicts counter 100's decoded_onehz row; without
-    // the guard its beats (incl. beat_index 2, which the winner's two-beat
-    // REPLACE never touches) would be orphaned forever.
     await LocalDb.insertRecord(_raw(ts, 5), _sample(ts, 5, [900, 910]));
 
     final db = await LocalDb.instance;
     final onehz = await db.query('decoded_onehz', where: 'rec_ts = ?', whereArgs: [ts]);
     expect(onehz.length, 1);
     expect(onehz.first['counter'], 5); // newest-wins
 
-    // No RR beats survive under the evicted counter…
-    final loserBeats =
-        await db.query('decoded_rr', where: 'counter = ?', whereArgs: [100]);
-    expect(loserBeats, isEmpty);
-    // …and globally: zero orphans (every beat's counter owns a decoded row).
+    // Only the winner's two beats remain, keyed by rec_ts.
+    final beats = await db.query('decoded_rr',
+        where: 'rec_ts = ?', whereArgs: [ts], orderBy: 'beat_index ASC');
+    expect([for (final b in beats) b['rr_ms']], [900, 910]);
+
+    // Globally: zero orphans (every beat's rec_ts owns a decoded row).
     final orphans = await db.rawQuery(
       'SELECT COUNT(*) c FROM decoded_rr '
-      'WHERE counter NOT IN (SELECT counter FROM decoded_onehz)',
+      'WHERE rec_ts NOT IN (SELECT rec_ts FROM decoded_onehz)',
     );
     expect(orphans.first['c'], 0);
 
-    // The RR read path (decodedRrByCounterRange, joined to frames by counter in
-    // derive_prepare.addDecodedPage) sees ONLY the winner's beats.
-    final rr = await LocalDb.decodedRrByCounterRange(fromCounter: 0, toCounter: 1 << 30);
+    // The RR read path sees ONLY the winner's beats.
+    final rr = await LocalDb.decodedRrByRecTsRange(fromRecTs: ts, toRecTs: ts);
     expect([for (final r in rr) r['rr_ms']], [900, 910]);
-    expect({for (final r in rr) r['counter']}, {5});
   });
 
-  test('prune sweeps pre-existing decoded_rr orphans', () async {
+  test('prune deletes decoded_rr by rec_ts, keeps recent beats', () async {
     const oldTs = 1700000000; // strictly before the cutoff below
     final db = await LocalDb.instance;
-    // Simulate a pre-guard leak: an RR beat whose counter has no decoded row.
+    // An old beat (its owning row absent — e.g. a leftover) is still deleted by
+    // the plain rec_ts-range prune; no counter subquery, no orphan sweep needed.
     await db.insert('decoded_rr', {
-      'counter': 999999,
+      'rec_ts': oldTs,
       'beat_index': 0,
       'rr_ts_ms': oldTs * 1000,
       'rr_ms': 850,
@@ -111,9 +108,9 @@ void main() {
 
     final deleted = await LocalDb.pruneDecodedBeforeRecTs(oldTs + 1000);
 
-    final orphan = await db.query('decoded_rr', where: 'counter = ?', whereArgs: [999999]);
-    expect(orphan, isEmpty, reason: 'orphan sweep must clean the leaked beat');
-    final kept = await db.query('decoded_rr', where: 'counter = ?', whereArgs: [7]);
+    final old = await db.query('decoded_rr', where: 'rec_ts = ?', whereArgs: [oldTs]);
+    expect(old, isEmpty, reason: 'the pruned window\'s beats must be deleted');
+    final kept = await db.query('decoded_rr', where: 'rec_ts = ?', whereArgs: [keepTs]);
     expect(kept.length, 1);
     // used to always come back 0 even when rows genuinely got pruned - none
     // of the txn.delete() counts were ever added up.

test/db_migration_ladder_test.dart

@@ -46,6 +46,29 @@ const _v6RawDdl = '''
   )
 ''';
 
+/// The origin/main (pre-v33) COUNTER-keyed decoded store, exactly as a user who
+/// installed at v19..31 has it — and by then raw_records is already DROPPED, so
+/// the v33 rekey is the ONLY copy of their 1 Hz data (no raw-backfill safety
+/// net). This is the highest-risk path the re-key touches.
+const _counterKeyedDecodedDdl = [
+  '''
+  CREATE TABLE decoded_onehz (
+    counter INTEGER PRIMARY KEY, rec_ts INTEGER NOT NULL,
+    hr INTEGER NOT NULL, ax REAL NOT NULL, ay REAL NOT NULL, az REAL NOT NULL,
+    spo2_red_raw INTEGER NOT NULL, spo2_ir_raw INTEGER NOT NULL,
+    skin_temp_raw INTEGER NOT NULL)
+''',
+  'CREATE UNIQUE INDEX idx_decoded_onehz_rec_ts_unique ON decoded_onehz(rec_ts)',
+  '''
+  CREATE TABLE decoded_rr (
+    counter INTEGER NOT NULL, beat_index INTEGER NOT NULL,
+    rr_ts_ms INTEGER NOT NULL, rr_ms INTEGER NOT NULL,
+    PRIMARY KEY (counter, beat_index))
+''',
+  'CREATE UNIQUE INDEX idx_decoded_rr_ts_beat_unique '
+      'ON decoded_rr(rr_ts_ms, beat_index)',
+];
+
 /// The v5-era derived tables, so step 9's derived_day → day_result copy is real.
 const _v5DerivedDdl = [
   '''
@@ -417,4 +440,85 @@ void main() {
       expect(health['ok'], isTrue, reason: '$health');
     },
   );
+
+  test(
+    'v33 re-key converts a v31 COUNTER-keyed decoded store to rec_ts LOSSLESSLY '
+    '— raw_records is already dropped, so the rekey is the only copy',
+    () async {
+      const name = 'migrate_from_v31_counterkeyed_test.db';
+      created.add(name);
+      await _seedOldDb(
+        name,
+        31,
+        [..._counterKeyedDecodedDdl, ..._v5DerivedDdl],
+        seedRows: (db) async {
+          // Three distinct seconds, distinct counters (valid old-schema data).
+          for (final r in const [
+            [100, 1785000000, 60],
+            [101, 1785000001, 61],
+            [102, 1785000002, 62],
+          ]) {
+            await db.insert('decoded_onehz', {
+              'counter': r[0], 'rec_ts': r[1], 'hr': r[2],
+              'ax': 0.0, 'ay': 0.0, 'az': 0.0,
+              'spo2_red_raw': 0, 'spo2_ir_raw': 0, 'skin_temp_raw': 0,
+            });
+          }
+          // Beats under counter 100 (two) and 102 (one).
+          await db.insert('decoded_rr', {
+            'counter': 100, 'beat_index': 0,
+            'rr_ts_ms': 1785000000 * 1000, 'rr_ms': 800,
+          });
+          await db.insert('decoded_rr', {
+            'counter': 100, 'beat_index': 1,
+            'rr_ts_ms': 1785000000 * 1000, 'rr_ms': 810,
+          });
+          await db.insert('decoded_rr', {
+            'counter': 102, 'beat_index': 0,
+            'rr_ts_ms': 1785000002 * 1000, 'rr_ms': 900,
+          });
+        },
+      );
+
+      expect(await _openThroughLocalDb(name), LocalDb.schemaVersion);
+      final db = await LocalDb.instance;
+
+      // Every second survives; counter is preserved as the forensic column.
+      final oh = await db.query('decoded_onehz', orderBy: 'rec_ts ASC');
+      expect([for (final r in oh) r['rec_ts']],
+          [1785000000, 1785000001, 1785000002]);
+      expect([for (final r in oh) r['counter']], [100, 101, 102]);
+      // PK is now rec_ts, not counter.
+      final ohInfo = await db.rawQuery('PRAGMA table_info(decoded_onehz)');
+      expect(ohInfo.firstWhere((c) => c['name'] == 'rec_ts')['pk'], 1);
+      expect(ohInfo.firstWhere((c) => c['name'] == 'counter')['pk'], 0);
+
+      // Beats re-home onto their real second; decoded_rr loses its counter col.
+      final rr =
+          await db.query('decoded_rr', orderBy: 'rec_ts ASC, beat_index ASC');
+      expect([for (final r in rr) r['rec_ts']],
+          [1785000000, 1785000000, 1785000002]);
+      expect([for (final r in rr) r['rr_ms']], [800, 810, 900]);
+      final rrInfo = await db.rawQuery('PRAGMA table_info(decoded_rr)');
+      expect(rrInfo.any((c) => c['name'] == 'counter'), isFalse);
+
+      // No stranded beats, no cross-stamped timestamps, no leaked temp tables.
+      expect(
+        (await db.rawQuery(
+          'SELECT COUNT(*) c FROM decoded_rr WHERE rr_ts_ms != rec_ts * 1000',
+        )).first['c'],
+        0,
+      );
+      expect(
+        await db.rawQuery(
+          "SELECT name FROM sqlite_master WHERE name LIKE '%\\_v33' ESCAPE '\\' "
+          "OR name LIKE '%\\_new' ESCAPE '\\'",
+        ),
+        isEmpty,
+      );
+
+      final health = await LocalDb.schemaHealth();
+      expect(health['ok'], isTrue, reason: '$health');
+    },
+  );
 }