fix(#952): the migration gate asserts a superset, not equal totals
Four of the owner's store keys could never migrate to rows. Each held rows whose merge_key is absent from its blob (1, 3, 7 and 5 of them), and the readback gate compared the key's TOTAL row count against the number of distinct records in the blob. Those two numbers can never be brought level again: nothing removes the extra rows, and the frozen blob can never come to contain them. Every attempt refused, for good. His two largest channels were among them, 6,313 and 5,648 messages, so they kept re-encoding 3.77 MB and 3.55 MB of history on every save. That is the cost #920 removed and they were not getting it. No data loss: the blob stays complete and authoritative for a refused key. The gate asserted the wrong invariant. What it exists to catch is a record from the blob failing to reach a row, which is a SUBSET question. Equality became wrong the moment rows could legitimately hold messages the frozen blob does not, which is true for every migrated key the instant a new message arrives. - messageMergeKeys(storeKey) reads back the stored identities, and the gate now requires every deduped blob key to be among them. Rows beyond that are logged and accepted rather than treated as corruption. - The refusal log names how many records are missing and samples three, which is more use than the two totals it printed before. - The four keys need no healing migration: they migrate normally at next access, pinned by "a key stuck under the old gate recovers on next access". The step can still fail, and both halves of the loop are proved: a BlobStore whose insert silently drops a row is refused and stays on the blob, and a lossy converter is still refused before anything is written. Renamed the test file, since it now guards the gate rather than describing the trap it was written to demonstrate. Task: meshcore-open-lkp Agent Mail: TealGlen Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>docs/962-reclamation-plan
diffwireauto
3 weeks ago
committed by
Strycher
parent
6abd5af8d9
commit
eef192ea7f
@ -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:<messageId>` whenever it is non-empty.
|
||||
String blobWith(List<String> ids) =>
|
||||
'[${ids.map((id) => '{"messageId":"$id","text":"t-$id","timestamp":1}').join(',')}]';
|
||||
|
||||
List<MessageRowInput> 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<void> upsertMessageRows(
|
||||
String storeKey,
|
||||
Iterable<MessageRowInput> rows,
|
||||
) => super.upsertMessageRows(storeKey, rows.skip(1));
|
||||
}
|
||||
@ -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:<messageId>` whenever it is non-empty.
|
||||
String blobWith(List<String> ids) =>
|
||||
'[${ids.map((id) => '{"messageId":"$id","text":"t-$id","timestamp":1}').join(',')}]';
|
||||
|
||||
List<MessageRowInput> 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);
|
||||
},
|
||||
);
|
||||
}
|
||||
Loading…
Reference in new issue