fix(#343): close the data-loss holes found by Gemini review

Adversarial review (standards#145) of the storage/migration surface found four
real defects, all data-integrity. All confirmed against the code and fixed; no
false positives.

BLOCKER - concurrent save race. saveChannelMessages / saveMessages are
read-modify-write with an await gap, so two saves to the same key raced and the
second clobbered the first, silently losing messages (e.g. a message and its
delivery ack arriving together). BlobStore now provides synchronized(key, ...),
a per-key operation chain; every RMW - merge-save, remove, and the load-path
legacy migration - runs through it. Different keys stay concurrent. New test
fires two concurrent saves and asserts the union survives.

BLOCKER - non-atomic DB relocation. The documents->support move copied and
deleted each file (.sqlite/-wal/-shm) in turn, so a failure after the main file
moved stranded the DB across two locations and corrupted it. Now copies all
files, verifies each by size, and only then deletes the sources; on any failure
it rolls back the destination and leaves the original intact.

MAJOR - load-path legacy migration raced save. The #194 index->PSK adoption did
a blind write to the PSK key that could clobber a save that landed first. It now
runs under the key lock and MERGES (union) instead of overwriting, so both the
adopted history and any fresh message survive.

MINOR - channel merge key lacked a sender. Two senders posting identical text at
the same timestamp without a messageId would collide and lose one. The key now
includes the sender, matching MessageStore.

flutter analyze clean, dart format clean, 509 tests pass. Full Gemini log in
docs/llm-consultations/.
integration/290-306-335
Strycher 21 hours ago
parent 2fefb6aa91
commit 3ff272ad8c

@ -73,43 +73,49 @@ class ChannelMessageStore {
// Deletion has its own path (removeChannelMessage) so this never // Deletion has its own path (removeChannelMessage) so this never
// resurrects a message the user deleted. // resurrects a message the user deleted.
final key = _storageKey(channelIndex); final key = _storageKey(channelIndex);
final byKey = <String, ChannelMessage>{}; final blobs = BlobStore.instance;
// Serialise the whole read-modify-write against other saves/deletes on this
final existing = await BlobStore.instance.readWithPrefsFallback(key); // key so two concurrent saves cannot clobber each other (Gemini review).
if (existing != null && existing.isNotEmpty) { await blobs.synchronized(key, () async {
try { final byKey = <String, ChannelMessage>{};
for (final e in jsonDecode(existing) as List<dynamic>) { final existing = await blobs.readWithPrefsFallback(key);
final m = _messageFromJson(e as Map<String, dynamic>); if (existing != null && existing.isNotEmpty) {
byKey[_mergeKey(m)] = m; try {
for (final e in jsonDecode(existing) as List<dynamic>) {
final m = _messageFromJson(e as Map<String, dynamic>);
byKey[_mergeKey(m)] = m;
}
} catch (e) {
// SAFELANE 6: never merge into an empty base and truncate silently.
appLogger.error(
'Failed to decode existing channel $channelIndex history before '
'merge; aborting save to avoid truncation: $e',
tag: 'Storage',
);
return;
} }
} catch (e) {
// SAFELANE 6: a decode failure here must be loud, not silently drop the
// persisted history by merging into an empty base.
appLogger.error(
'Failed to decode existing channel $channelIndex history before '
'merge; aborting save to avoid truncation: $e',
tag: 'Storage',
);
return;
} }
} for (final m in messages) {
for (final m in messages) { byKey[_mergeKey(m)] = m;
byKey[_mergeKey(m)] = m; }
} final merged = byKey.values.toList()
..sort((a, b) => a.timestamp.compareTo(b.timestamp));
final merged = byKey.values.toList() await blobs.write(key, jsonEncode(merged.map(_messageToJson).toList()));
..sort((a, b) => a.timestamp.compareTo(b.timestamp)); });
await BlobStore.instance.write(
key,
jsonEncode(merged.map(_messageToJson).toList()),
);
} }
/// Stable identity for merge/dedupe. messageId when present, else a composite /// Stable identity for merge/dedupe. messageId when present, else a composite
/// that distinguishes distinct messages that share no id. /// that distinguishes distinct messages that share no id.
String _mergeKey(ChannelMessage m) { String _mergeKey(ChannelMessage m) {
if (m.messageId.isNotEmpty) return 'id:${m.messageId}'; if (m.messageId.isNotEmpty) return 'id:${m.messageId}';
return 'x:${m.packetHash ?? ''}:${m.timestamp.millisecondsSinceEpoch}:${m.text}'; // Include the sender: two different senders can post identical text at the
// same timestamp without a messageId, and would otherwise collide and lose
// one (Gemini review, 2026-07-20).
final sender = m.senderKey == null
? ''
: m.senderKey!.map((b) => b.toRadixString(16)).join();
return 'x:$sender:${m.packetHash ?? ''}:'
'${m.timestamp.millisecondsSinceEpoch}:${m.text}';
} }
/// Removes a single message from the persisted history. The explicit delete /// Removes a single message from the persisted history. The explicit delete
@ -121,28 +127,27 @@ class ChannelMessageStore {
) async { ) async {
if (publicKeyHex.isEmpty) return; if (publicKeyHex.isEmpty) return;
final key = _storageKey(channelIndex); final key = _storageKey(channelIndex);
final existing = await BlobStore.instance.readWithPrefsFallback(key); final blobs = BlobStore.instance;
if (existing == null || existing.isEmpty) return; await blobs.synchronized(key, () async {
final existing = await blobs.readWithPrefsFallback(key);
final List<dynamic> raw; if (existing == null || existing.isEmpty) return;
try { final List<dynamic> raw;
raw = jsonDecode(existing) as List<dynamic>; try {
} catch (e) { raw = jsonDecode(existing) as List<dynamic>;
appLogger.error( } catch (e) {
'Failed to decode channel $channelIndex history for delete: $e', appLogger.error(
tag: 'Storage', 'Failed to decode channel $channelIndex history for delete: $e',
); tag: 'Storage',
return; );
} return;
final target = _mergeKey(message); }
final kept = raw final target = _mergeKey(message);
.map((e) => _messageFromJson(e as Map<String, dynamic>)) final kept = raw
.where((m) => _mergeKey(m) != target) .map((e) => _messageFromJson(e as Map<String, dynamic>))
.toList(); .where((m) => _mergeKey(m) != target)
await BlobStore.instance.write( .toList();
key, await blobs.write(key, jsonEncode(kept.map(_messageToJson).toList()));
jsonEncode(kept.map(_messageToJson).toList()), });
);
} }
/// Load messages for a specific channel /// Load messages for a specific channel
@ -177,9 +182,30 @@ class ChannelMessageStore {
appLogger.info( appLogger.info(
'Migrating channel messages $legacyKey -> $key (PSK-keyed, #194)', 'Migrating channel messages $legacyKey -> $key (PSK-keyed, #194)',
); );
await blobs.write(key, legacy); // Under the key lock, and MERGE rather than overwrite: a save may
// have landed on the PSK key between the read above and here, and a
// blind write would clobber it (Gemini review). Union keeps both the
// adopted legacy history and any freshly-saved message.
jsonString = await blobs.synchronized(key, () async {
final byKey = <String, ChannelMessage>{};
for (final srcJson in [await blobs.read(key), legacy]) {
if (srcJson == null || srcJson.isEmpty) continue;
try {
for (final e in jsonDecode(srcJson) as List<dynamic>) {
final m = _messageFromJson(e as Map<String, dynamic>);
byKey[_mergeKey(m)] = m;
}
} catch (_) {
// Skip an undecodable source rather than aborting the adoption.
}
}
final merged = byKey.values.toList()
..sort((a, b) => a.timestamp.compareTo(b.timestamp));
final encoded = jsonEncode(merged.map(_messageToJson).toList());
await blobs.write(key, encoded);
return encoded;
});
await blobs.deleteEverywhere(legacyKey); await blobs.deleteEverywhere(legacyKey);
jsonString = legacy;
break; break;
} }
} }

@ -1,3 +1,5 @@
import 'dart:async';
import 'package:flutter/foundation.dart'; import 'package:flutter/foundation.dart';
import '../../utils/app_logger.dart'; import '../../utils/app_logger.dart';
@ -46,6 +48,24 @@ class BlobStore {
static bool isBulkKey(String key) => migratedPrefixes.any(key.startsWith); static bool isBulkKey(String key) => migratedPrefixes.any(key.startsWith);
/// Per-key operation chain. Merge-on-save and the legacy-key migration are
/// read-modify-write sequences with an await gap; two of them racing on the
/// same key would let the second clobber the first and silently lose
/// messages (Gemini review, 2026-07-20). Every RMW on a key runs through
/// [synchronized], which serialises operations per key while leaving
/// different keys concurrent.
final Map<String, Future<void>> _keyChains = {};
/// Serialises [action] against other synchronized actions on the same [key].
Future<T> synchronized<T>(String key, Future<T> Function() action) {
final prior = _keyChains[key] ?? Future<void>.value();
final result = prior.then((_) => action());
// Next op waits for this one; swallow errors so one failure does not wedge
// the chain for the key.
_keyChains[key] = result.then((_) {}, onError: (_) {});
return result;
}
/// Reads a bulk key, falling back to SharedPreferences if drift does not /// Reads a bulk key, falling back to SharedPreferences if drift does not
/// have it. /// have it.
/// ///

@ -76,26 +76,59 @@ class OffbandDatabase extends _$OffbandDatabase {
final docs = await getApplicationDocumentsDirectory(); final docs = await getApplicationDocumentsDirectory();
final oldPath = p.join(docs.path, '$_dbName.sqlite'); final oldPath = p.join(docs.path, '$_dbName.sqlite');
if (File(oldPath).existsSync() && oldPath != newPath) { if (File(oldPath).existsSync() && oldPath != newPath) {
for (final suffix in ['', '-wal', '-shm']) { // Copy ALL files first, verify, and only then delete the sources.
final src = File('$oldPath$suffix'); // Deleting per-file mid-loop (Gemini review) could strand the main
if (src.existsSync()) { // .sqlite in one place and its -wal in another if a later copy
src.copySync('$newPath$suffix'); // failed, corrupting the DB. Copy-all-then-delete-all makes a
src.deleteSync(); // partial failure recoverable: the original stays intact and the
// half-written destination is cleaned up.
final suffixes = ['', '-wal', '-shm'];
final present = suffixes
.where((s) => File('$oldPath$s').existsSync())
.toList();
try {
for (final s in present) {
File('$oldPath$s').copySync('$newPath$s');
} }
// Verify every copy exists with a matching size before deleting.
for (final s in present) {
final src = File('$oldPath$s');
final dst = File('$newPath$s');
if (!dst.existsSync() || dst.lengthSync() != src.lengthSync()) {
throw StateError('copy verification failed for $newPath$s');
}
}
for (final s in present) {
File('$oldPath$s').deleteSync();
}
appLogger.info(
'Relocated drift DB out of the documents dir (OneDrive on '
'Windows) into the app-support dir: $oldPath -> $newPath',
tag: 'Storage',
);
} catch (e) {
// Roll back any partial destination so drift does not open a
// half-copied DB; the original in the documents dir is untouched.
for (final s in present) {
final dst = File('$newPath$s');
if (dst.existsSync()) {
try {
dst.deleteSync();
} catch (_) {}
}
}
rethrow;
} }
appLogger.info(
'Relocated drift DB out of the documents dir (OneDrive on '
'Windows) into the app-support dir: $oldPath -> $newPath',
tag: 'Storage',
);
} }
} catch (e) { } catch (e) {
// Relocation is best-effort: if it fails, drift opens a fresh DB at the // Relocation is best-effort. On failure the ORIGINAL DB is left intact
// correct location and the migration re-runs from SharedPreferences. // in the documents dir (the destination was rolled back), so no data is
// Loud, never silent (SAFELANE 6). // lost - drift opens a fresh DB at the support location and the user's
// history remains recoverable from the old path. Loud, never silent.
appLogger.error( appLogger.error(
'Failed to relocate the drift DB from the documents dir: $e. ' 'Failed to relocate the drift DB from the documents dir: $e. The '
'A fresh DB will be created at the app-support location.', 'original at the old path is intact; a fresh DB will open at the '
'app-support location. Recover the old DB manually if needed.',
tag: 'Storage', tag: 'Storage',
); );
} }

@ -29,34 +29,32 @@ class MessageStore {
// truncate the store. Upsert by identity; deletion is explicit // truncate the store. Upsert by identity; deletion is explicit
// (removeMessage). // (removeMessage).
final key = '$keyFor$contactKeyHex'; final key = '$keyFor$contactKeyHex';
final byKey = <String, Message>{}; final blobs = BlobStore.instance;
await blobs.synchronized(key, () async {
final existing = await BlobStore.instance.readWithPrefsFallback(key); final byKey = <String, Message>{};
if (existing != null && existing.isNotEmpty) { final existing = await blobs.readWithPrefsFallback(key);
try { if (existing != null && existing.isNotEmpty) {
for (final e in jsonDecode(existing) as List<dynamic>) { try {
final m = _messageFromJson(e as Map<String, dynamic>); for (final e in jsonDecode(existing) as List<dynamic>) {
byKey[_mergeKey(m)] = m; final m = _messageFromJson(e as Map<String, dynamic>);
byKey[_mergeKey(m)] = m;
}
} catch (e) {
appLogger.error(
'Failed to decode existing DM history before merge; aborting save '
'to avoid truncation: $e',
tag: 'Storage',
);
return;
} }
} catch (e) {
appLogger.error(
'Failed to decode existing DM history before merge; aborting save '
'to avoid truncation: $e',
tag: 'Storage',
);
return;
} }
} for (final m in messages) {
for (final m in messages) { byKey[_mergeKey(m)] = m;
byKey[_mergeKey(m)] = m; }
} final merged = byKey.values.toList()
..sort((a, b) => a.timestamp.compareTo(b.timestamp));
final merged = byKey.values.toList() await blobs.write(key, jsonEncode(merged.map(_messageToJson).toList()));
..sort((a, b) => a.timestamp.compareTo(b.timestamp)); });
await BlobStore.instance.write(
key,
jsonEncode(merged.map(_messageToJson).toList()),
);
} }
String _mergeKey(Message m) { String _mergeKey(Message m) {

@ -79,6 +79,25 @@ void main() {
expect(all.single.reactions['thumbsup'], 2); expect(all.single.reactions['thumbsup'], 2);
}); });
test(
'concurrent saves to one channel do not lose messages (Gemini)',
() async {
await store.saveChannelMessages(0, [msg(0)]);
// Fire two saves without awaiting between them: they race on the same key.
// Serialisation must make the result the union, not last-writer-wins.
final a = store.saveChannelMessages(0, [msg(1)]);
final b = store.saveChannelMessages(0, [msg(2)]);
await Future.wait([a, b]);
final all = await store.loadChannelMessages(0);
expect(
all.map((m) => m.text),
containsAll(['m0', 'm1', 'm2']),
reason: 'neither concurrent save may clobber the other',
);
expect(all, hasLength(3));
},
);
test('deleting a message does NOT resurrect it on the next save', () async { test('deleting a message does NOT resurrect it on the next save', () async {
await store.saveChannelMessages(0, [for (var i = 0; i < 5; i++) msg(i)]); await store.saveChannelMessages(0, [for (var i = 0; i < 5; i++) msg(i)]);

Loading…
Cancel
Save

Powered by TurnKey Linux.