diff --git a/lib/storage/drift/blob_store.dart b/lib/storage/drift/blob_store.dart index e6bc8686..b458362f 100644 --- a/lib/storage/drift/blob_store.dart +++ b/lib/storage/drift/blob_store.dart @@ -204,8 +204,23 @@ class BlobStore { return rows.map((r) => r.payload).toList(); } - /// How many messages are stored under [storeKey]. The migration's readback - /// gate compares this against the number of records decoded from the blob. + /// Every stored identity under [storeKey]. + /// + /// The migration's readback gate asks this, not [messageRowCount], because + /// the question that matters is whether any record from the blob is MISSING + /// from rows. A total count cannot answer that once rows may legitimately + /// hold messages the frozen blob does not (#952). + Future> messageMergeKeys(String storeKey) async { + final key = _db.messageRecords.mergeKey; + final rows = + await (_db.selectOnly(_db.messageRecords) + ..addColumns([key]) + ..where(_db.messageRecords.storeKey.equals(storeKey))) + .get(); + return rows.map((r) => r.read(key)!).toSet(); + } + + /// How many messages are stored under [storeKey]. Future messageRowCount(String storeKey) async { final count = _db.messageRecords.mergeKey.count(); final row = @@ -311,12 +326,21 @@ class BlobStore { /// source element; a shortfall against step 2 refuses the migration. /// 4. The rows are inserted in one transaction, so a failure part-way leaves /// no partial set behind. - /// 5. The stored row count is read back and compared to the number of - /// DISTINCT merge keys. Only an exact match marks the key migrated. + /// 5. The stored identities are read back and every DISTINCT merge key from + /// the blob must be among them. Rows BEYOND that set are fine. /// /// Steps 2 and 5 together close the loop from blob element to stored row, /// and either can fail on its own. /// + /// Step 5 compared totals until #952. That was wrong in one direction: once + /// a key is on rows its blob is frozen, so an earlier refused attempt or a + /// newer message leaves rows the blob can never contain, the counts can + /// never be brought level again, and the key is refused on every future + /// attempt forever. Four of the owner's keys were stuck that way, including + /// his two largest channels. "No blob record is missing" is the invariant + /// that was actually wanted; it still catches a short insert, which is what + /// the step is for. + /// /// **The blob is never deleted**, not even on a verified match. Rows become /// the read and write path and the blob is left untouched. Reclaiming that /// space is deliberately a separate, owner-gated step: the moment the delete @@ -393,19 +417,36 @@ class BlobStore { return false; } - final stored = await messageRowCount(storeKey); - if (stored != deduped.length) { + final storedKeys = await messageMergeKeys(storeKey); + final missing = deduped.keys + .where((k) => !storedKeys.contains(k)) + .toList(); + if (missing.isNotEmpty) { appLogger.error( - 'Row migration of $storeKey REFUSED: read back $stored rows but ' - 'expected ${deduped.length}. The blob is kept, nothing is deleted, ' - 'and this key keeps reading from the blob. Rows written so far are ' - 'left in place and are harmless because the key is not marked ' - 'migrated.', + 'Row migration of $storeKey REFUSED: ${missing.length} of the ' + '${deduped.length} records in the blob are not present as rows after ' + 'the insert (e.g. ${missing.take(3).join(', ')}). The blob is kept, ' + 'nothing is deleted, and this key keeps reading from the blob.', tag: 'Storage', ); return false; } + // Rows beyond the blob's contents are expected, not a fault: once a key + // is on rows the blob is frozen, so an earlier refused attempt or a + // newer message leaves rows the blob will never contain. Comparing + // TOTALS here is what trapped four of the owner's keys permanently + // (#952), because the two numbers could never be brought back level. + final extra = storedKeys.length - deduped.length; + if (extra > 0) { + appLogger.info( + 'Row migration of $storeKey: $extra row(s) beyond the ' + '${deduped.length} in the blob were already stored. Every blob ' + 'record is present, so the migration stands.', + tag: 'Storage', + ); + } + await _markMigrated(storeKey); appLogger.info( 'Row migration of $storeKey: ${deduped.length} messages now stored as ' diff --git a/test/storage/migration_readback_gate_test.dart b/test/storage/migration_readback_gate_test.dart new file mode 100644 index 00000000..93309954 --- /dev/null +++ b/test/storage/migration_readback_gate_test.dart @@ -0,0 +1,182 @@ +import 'package:drift/native.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:meshcore_open/storage/drift/blob_store.dart'; +import 'package:meshcore_open/storage/drift/offband_database.dart'; +import 'package:meshcore_open/storage/prefs_manager.dart'; +import 'package:shared_preferences/shared_preferences.dart'; + +/// #952: the migration's readback gate asserts a superset, not equality. +/// +/// Found on the owner's live database while working #943. Four of his store +/// keys held rows whose `merge_key` did not appear in their blob (1, 3, 7 and +/// 5 of them), so the old gate's comparison of TOTALS could never come level +/// again: nothing removes those rows and the frozen blob can never come to +/// contain them. Every attempt refused, forever, and his two largest channels +/// (6,313 and 5,648 messages) were among them, still re-encoding their whole +/// history on every save. +/// +/// The invariant that was actually wanted is "no record from the blob is +/// missing from rows". These tests pin that, and that it still catches the +/// short insert the step exists for. +void main() { + late OffbandDatabase db; + late BlobStore blobs; + + const key = 'channel_messages_aaaaaaaaaapsk_deadbeef'; + + /// A channel history blob. Only `messageId` matters here: both stores' merge + /// key is `id:` whenever it is non-empty. + String blobWith(List ids) => + '[${ids.map((id) => '{"messageId":"$id","text":"t-$id","timestamp":1}').join(',')}]'; + + List toRows(String json) { + final ids = RegExp( + r'"messageId":"([^"]+)"', + ).allMatches(json).map((m) => m.group(1)!).toList(); + return ids + .map( + (id) => MessageRowInput( + mergeKey: 'id:$id', + timestampMs: 1, + payload: '{"messageId":"$id"}', + ), + ) + .toList(); + } + + setUp(() async { + SharedPreferences.setMockInitialValues({}); + PrefsManager.reset(); + await PrefsManager.initialize(); + db = OffbandDatabase(NativeDatabase.memory()); + blobs = BlobStore(db); + BlobStore.overrideForTest(blobs); + }); + + tearDown(() async { + await db.close(); + BlobStore.clearTestOverride(); + }); + + test('a clean key migrates, so the fixture itself is sound', () async { + await blobs.write(key, blobWith(['a', 'b', 'c'])); + + expect(await blobs.ensureMessageRowsMigrated(key, toRows), isTrue); + expect(await blobs.isOnMessageRows(key), isTrue); + expect(await blobs.messageRowCount(key), 3); + }); + + group('rows the blob does not contain no longer block the migration', () { + test('one extra row still migrates', () async { + await blobs.write(key, blobWith(['a', 'b', 'c'])); + // The state the owner's four keys were in. Under the old gate this + // compared 4 stored rows against 3 blob records and refused. + await blobs.upsertMessageRows(key, const [ + MessageRowInput(mergeKey: 'id:newer', timestampMs: 2, payload: '{}'), + ]); + + expect(await blobs.ensureMessageRowsMigrated(key, toRows), isTrue); + expect(await blobs.isOnMessageRows(key), isTrue); + }); + + test('the extra row is kept, not discarded', () async { + await blobs.write(key, blobWith(['a', 'b', 'c'])); + await blobs.upsertMessageRows(key, const [ + MessageRowInput( + mergeKey: 'id:newer', + timestampMs: 2, + payload: '{"messageId":"newer"}', + ), + ]); + await blobs.ensureMessageRowsMigrated(key, toRows); + + // Accepting the migration must not lose the message that caused the + // mismatch: it is newer than the blob, not spurious. + expect(await blobs.messageMergeKeys(key), contains('id:newer')); + expect(await blobs.messageRowCount(key), 4); + }); + + test('his worst key in miniature: many records, a few extra', () async { + await blobs.write(key, blobWith(List.generate(60, (i) => 'm$i'))); + await blobs.upsertMessageRows(key, const [ + MessageRowInput(mergeKey: 'id:x1', timestampMs: 9, payload: '{}'), + MessageRowInput(mergeKey: 'id:x2', timestampMs: 9, payload: '{}'), + MessageRowInput(mergeKey: 'id:x3', timestampMs: 9, payload: '{}'), + MessageRowInput(mergeKey: 'id:x4', timestampMs: 9, payload: '{}'), + MessageRowInput(mergeKey: 'id:x5', timestampMs: 9, payload: '{}'), + ]); + + expect(await blobs.ensureMessageRowsMigrated(key, toRows), isTrue); + expect(await blobs.messageRowCount(key), 65); + }); + + test('a key stuck under the old gate recovers on next access', () async { + // The outcome that matters for his install: no healing migration is + // needed, the four keys migrate normally once the comparison is right. + await blobs.write(key, blobWith(['a', 'b', 'c'])); + await blobs.upsertMessageRows(key, const [ + MessageRowInput(mergeKey: 'id:newer', timestampMs: 2, payload: '{}'), + ]); + expect(await blobs.isOnMessageRows(key), isFalse); + + expect(await blobs.ensureMessageRowsMigrated(key, toRows), isTrue); + expect(await blobs.isOnMessageRows(key), isTrue); + }); + }); + + group('the gate can still fail', () { + test('a short insert is refused and the key stays on the blob', () async { + // Proving the step still does its job. A store whose insert silently + // drops a row is exactly what the readback exists to catch, and the + // superset check must not have traded that away. + final lossy = _DropsOneRow(db); + BlobStore.overrideForTest(lossy); + await lossy.write(key, blobWith(['a', 'b', 'c'])); + + expect( + await lossy.ensureMessageRowsMigrated(key, toRows), + isFalse, + reason: 'one of the three blob records never reached a row', + ); + expect(await lossy.isOnMessageRows(key), isFalse); + }); + + test('a lossy converter is refused before anything is written', () async { + // The other half of the loop (step 2), unchanged by #952: the blob's + // element count is measured independently of the converter. + await blobs.write(key, blobWith(['a', 'b', 'c'])); + + expect( + await blobs.ensureMessageRowsMigrated(key, (json) { + final all = toRows(json); + return all.take(all.length - 1).toList(); + }), + isFalse, + ); + expect(await blobs.messageRowCount(key), 0); + }); + }); + + test('no message is lost either way: the blob is retained', () async { + final history = blobWith(['a', 'b', 'c']); + await blobs.write(key, history); + await blobs.upsertMessageRows(key, const [ + MessageRowInput(mergeKey: 'id:newer', timestampMs: 2, payload: '{}'), + ]); + await blobs.ensureMessageRowsMigrated(key, toRows); + + expect(await blobs.read(key), history); + }); +} + +/// A [BlobStore] whose row insert silently drops the first row, standing in +/// for a short or partially failed write. +class _DropsOneRow extends BlobStore { + _DropsOneRow(super.db); + + @override + Future upsertMessageRows( + String storeKey, + Iterable rows, + ) => super.upsertMessageRows(storeKey, rows.skip(1)); +} diff --git a/test/storage/migration_refusal_trap_test.dart b/test/storage/migration_refusal_trap_test.dart deleted file mode 100644 index f08ec147..00000000 --- a/test/storage/migration_refusal_trap_test.dart +++ /dev/null @@ -1,140 +0,0 @@ -import 'package:drift/native.dart'; -import 'package:flutter_test/flutter_test.dart'; -import 'package:meshcore_open/storage/drift/blob_store.dart'; -import 'package:meshcore_open/storage/drift/offband_database.dart'; -import 'package:meshcore_open/storage/prefs_manager.dart'; -import 'package:shared_preferences/shared_preferences.dart'; - -/// Found while working #943, on the owner's live database. -/// -/// Four of his store keys hold rows but are NOT in the migrated marker, so -/// they still read and write through the blob. Each one has rows whose -/// `merge_key` does not appear in its blob (1, 3, 7 and 5 of them). The -/// readback gate in `ensureMessageRowsMigrated` compares the key's TOTAL row -/// count against the number of distinct records in the blob, so those extra -/// rows make the two numbers differ and the migration is refused. -/// -/// The gate's own log line calls the leftover rows "harmless because the key -/// is not marked migrated". These tests check that claim. It does not hold: -/// nothing ever removes the extra rows and the blob can never come to contain -/// them, so the comparison fails identically on every future attempt and the -/// key is stuck on the blob path for good. -void main() { - late OffbandDatabase db; - late BlobStore blobs; - - const key = 'channel_messages_aaaaaaaaaapsk_deadbeef'; - - /// Two records, as a channel history blob. Only `messageId` matters here: - /// both stores' merge key is `id:` whenever it is non-empty. - String blobWith(List ids) => - '[${ids.map((id) => '{"messageId":"$id","text":"t-$id","timestamp":1}').join(',')}]'; - - List toRows(String json) { - final ids = RegExp( - r'"messageId":"([^"]+)"', - ).allMatches(json).map((m) => m.group(1)!).toList(); - return ids - .map( - (id) => MessageRowInput( - mergeKey: 'id:$id', - timestampMs: 1, - payload: '{"messageId":"$id"}', - ), - ) - .toList(); - } - - setUp(() async { - SharedPreferences.setMockInitialValues({}); - PrefsManager.reset(); - await PrefsManager.initialize(); - db = OffbandDatabase(NativeDatabase.memory()); - blobs = BlobStore(db); - BlobStore.overrideForTest(blobs); - }); - - tearDown(() async { - await db.close(); - BlobStore.clearTestOverride(); - }); - - test('a clean key migrates, so the fixture itself is sound', () async { - await blobs.write(key, blobWith(['a', 'b', 'c'])); - - expect(await blobs.ensureMessageRowsMigrated(key, toRows), isTrue); - expect(await blobs.isOnMessageRows(key), isTrue); - expect(await blobs.messageRowCount(key), 3); - }); - - test('one row the blob does not contain is enough to refuse', () async { - await blobs.write(key, blobWith(['a', 'b', 'c'])); - // The state the owner's four keys are in: a row for a message that is not - // in the blob. On his biggest channel this is 5 rows against 6,308. - await blobs.upsertMessageRows(key, const [ - MessageRowInput(mergeKey: 'id:newer', timestampMs: 2, payload: '{}'), - ]); - - expect( - await blobs.ensureMessageRowsMigrated(key, toRows), - isFalse, - reason: 'the gate compares 4 stored rows against 3 blob records', - ); - expect(await blobs.isOnMessageRows(key), isFalse); - }); - - test('and it refuses again, every time, for good', () async { - await blobs.write(key, blobWith(['a', 'b', 'c'])); - await blobs.upsertMessageRows(key, const [ - MessageRowInput(mergeKey: 'id:newer', timestampMs: 2, payload: '{}'), - ]); - - for (var attempt = 1; attempt <= 3; attempt++) { - expect( - await blobs.ensureMessageRowsMigrated(key, toRows), - isFalse, - reason: 'attempt $attempt refused', - ); - } - - // Nothing cleaned up the extra row, which is why the next attempt fails - // the same way. This is the trap: the key cannot recover on its own. - expect(await blobs.messageRowCount(key), 4); - expect(await blobs.isOnMessageRows(key), isFalse); - }); - - test('a growing blob does not rescue it either', () async { - await blobs.write(key, blobWith(['a', 'b', 'c'])); - await blobs.upsertMessageRows(key, const [ - MessageRowInput(mergeKey: 'id:newer', timestampMs: 2, payload: '{}'), - ]); - expect(await blobs.ensureMessageRowsMigrated(key, toRows), isFalse); - - // The key stays on the blob path, so new traffic lands in the blob. That - // is the only thing about this key that changes over time, and it moves - // the two numbers further apart rather than together. - await blobs.write(key, blobWith(['a', 'b', 'c', 'd', 'e'])); - - expect( - await blobs.ensureMessageRowsMigrated(key, toRows), - isFalse, - reason: '6 rows against 5 blob records: still unequal', - ); - }); - - test( - 'no message is actually lost, which is why this is not data loss', - () async { - // Worth pinning: the blob remains complete and authoritative for a refused - // key, so the cost here is the save path, not history. - final history = blobWith(['a', 'b', 'c']); - await blobs.write(key, history); - await blobs.upsertMessageRows(key, const [ - MessageRowInput(mergeKey: 'id:newer', timestampMs: 2, payload: '{}'), - ]); - await blobs.ensureMessageRowsMigrated(key, toRows); - - expect(await blobs.read(key), history); - }, - ); -}