Merge pull request #430 from twonlyapp/hotfix

Fix: Recover broken sessions more reliably

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 0.5.4
+
+- Fix: Recover broken sessions more reliably
+
 ## 0.5.3
 
 - Hotfix: Fixed stuck Signal sessions

lib/src/callbacks/logging.callbacks.dart

@@ -13,9 +13,9 @@ class LoggingCallbacks {
         dartLogSink.stream.listen(
           (log) {
             if (log.contains('INFO ')) {
-              Log.info(log.split('INFO ')[1]);
+              Log.rust(log.split('INFO ')[1]);
             } else if (log.contains('DEBUG ')) {
-              Log.info(log.split('DEBUG ')[1]);
+              Log.rust(log.split('DEBUG ')[1]);
             } else if (kDebugMode && !Platform.environment.containsKey('FLUTTER_TEST')) {
               // ignore: avoid_print
               print(log);

lib/src/services/api/api.service.dart

@@ -249,6 +249,14 @@ class ApiService {
   bool get isConnected => _channel != null && _channel!.closeCode == null;
 
   Future<void> _onDone() async {
+    final channel = _channel;
+    if (channel != null) {
+      // Not closed by us. The server's close code tells why it dropped the
+      // connection, e.g. 1013 (outbound queue full) or 4003 (ack timeouts).
+      Log.info(
+        'websocket closed with code ${channel.closeCode}: ${channel.closeReason}',
+      );
+    }
     _reconnectionDelay = 3;
     await onClosed();
   }

lib/src/services/api/mediafiles/upload.api.dart

@@ -52,17 +52,15 @@ Future<void> reuploadMediaFiles() async {
       final contacts = <int, Contact>{};
 
       for (final receipt in receipts) {
-        if (receipt.retryCount > 1 && receipt.lastRetry != null) {
+        if (!isRetransmissionDue(
-          final twentyFourHoursAgo = DateTime.now().subtract(
+          receipt,
-            const Duration(hours: 6),
+          minimumPause: const Duration(hours: 6),
-          );
+        )) {
-          if (receipt.lastRetry!.isAfter(twentyFourHoursAgo)) {
           Log.info(
-              'Ignoring ${receipt.receiptId} as it was retried in the last 6h',
+            'Ignoring ${receipt.receiptId} as its next retry is not due yet',
           );
           continue;
         }
-        }
 
         var messageId = receipt.messageId;
         if (receipt.messageId == null) {

lib/src/services/api/messages.api.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
+import 'dart:math';
 
 import 'package:clock/clock.dart';
 import 'package:drift/drift.dart';
@@ -36,7 +37,11 @@ Future<void> _resetStuckOutgoingSignalSessions() async {
       );
 
   for (final contactId in contactIds) {
-    if (!shouldAttemptResync(contactId)) continue;
+    final contact = await twonlyDB.contactsDao.getContactById(contactId);
+    if (contact == null ||
+        !shouldAttemptResync(contactId, contact.signalVersion)) {
+      continue;
+    }
 
     Log.warn(
       'Detected an out-of-sync outgoing Signal session with $contactId: '
@@ -53,7 +58,7 @@ Future<void> _resetStuckOutgoingSignalSessions() async {
     }
 
     if (!reset) {
-      recordResyncAttempt(contactId, success: false);
+      recordResyncAttempt(contactId, contact.signalVersion, success: false);
       continue;
     }
 
@@ -64,11 +69,35 @@ Future<void> _resetStuckOutgoingSignalSessions() async {
   }
 }
 
+/// Whether a message that the server already accepted may be sent again.
+///
+/// Every valid message from a contact marks everything they have not
+/// confirmed for retry. When their confirmation got lost, they stay silent
+/// until their duplicate cooldown ends, ten days on 0.5.2, so sending again on
+/// every message from them only floods their device. After the first retry the
+/// pause doubles with every attempt, from one hour up to a day, but never
+/// drops below [minimumPause].
+bool isRetransmissionDue(
+  Receipt receipt, {
+  Duration minimumPause = Duration.zero,
+}) {
+  final lastRetry = receipt.lastRetry;
+  if (receipt.retryCount <= 1 || lastRetry == null) return true;
+  var pause = Duration(hours: min(1 << min(receipt.retryCount - 2, 5), 24));
+  if (pause < minimumPause) pause = minimumPause;
+  return !clock.now().isBefore(lastRetry.add(pause));
+}
+
 Future<void> retransmitAllMessages() async {
   return lockRetransmission.protect(() async {
     await _resetStuckOutgoingSignalSessions();
 
-    final receipts = await twonlyDB.receiptsDao.getReceiptsForRetransmission();
+    final receipts = (await twonlyDB.receiptsDao.getReceiptsForRetransmission())
+        .where(
+          (receipt) =>
+              receipt.ackByServerAt == null || isRetransmissionDue(receipt),
+        )
+        .toList();
 
     if (receipts.isEmpty) return;
 
@@ -102,6 +131,13 @@ Future<void> retransmitAllMessages() async {
   });
 }
 
+Future<SignalEncryptResult?> _encryptFor(
+  Contact contact,
+  Uint8List plaintext,
+) => contact.signalVersion == SignalVersion.v2
+    ? signalEncryptMessageV2(contact.userId, plaintext)
+    : signalEncryptMessage(contact.userId, plaintext);
+
 final Map<String, Mutex> _tryToSendLocks = {};
 
 // When the ackByServerAt is set this value is written in the receipted
@@ -162,6 +198,11 @@ Future<(Uint8List, Uint8List?)?> _tryToSendCompleteMessageInternal({
       await twonlyDB.receiptsDao.deleteReceipt(receipt.receiptId);
       return null;
     }
+    if (contact.signalVersion == SignalVersion.v1) {
+      // Pending receipts make this an active session too, including messages
+      // created before active-session migration was introduced.
+      scheduleSignalSessionUpgrade(contact.userId);
+    }
 
     if (!onlyReturnEncryptedData &&
         receipt.ackByServerAt != null &&
@@ -201,23 +242,41 @@ Future<(Uint8List, Uint8List?)?> _tryToSendCompleteMessageInternal({
 
     if (message.type == pb.Message_Type.CIPHERTEXT ||
         message.type == pb.Message_Type.CIPHERTEXT_V2) {
-      final useV2 = contact.signalVersion == SignalVersion.v2;
+      final plaintext = Uint8List.fromList(message.encryptedContent);
-      final encryptResult = useV2
+      var target = contact;
-          ? await signalEncryptMessageV2(
+      var encryptResult = await _encryptFor(target, plaintext);
+
+      if (encryptResult == null &&
+          shouldAttemptResync(receipt.contactId, target.signalVersion)) {
+        // Usually there is no session to encrypt with, e.g. for a contact
+        // marked V2 that the Rust store holds no session for. Building one
+        // from the contact's current bundle can also move them to V2.
+        final resynced = await handleSessionResync(receipt.contactId);
+        if (resynced) {
+          target =
+              await twonlyDB.contactsDao.getContactById(receipt.contactId) ??
+              target;
+          encryptResult = await _encryptFor(target, plaintext);
+        }
+        recordResyncAttempt(
           receipt.contactId,
-              Uint8List.fromList(message.encryptedContent),
+          target.signalVersion,
-            )
+          success: false,
-          : await signalEncryptMessage(
-              receipt.contactId,
-              Uint8List.fromList(message.encryptedContent),
         );
+      }
 
       if (encryptResult == null) {
         Log.error(
           '[${receipt.receiptId}] Could not encrypt the message '
-          '(${useV2 ? 'V2' : 'V1'}) for user ${receipt.contactId}. '
+          '(${target.signalVersion == SignalVersion.v2 ? 'V2' : 'V1'}) '
-          'Aborting and trying again.',
+          'for user ${receipt.contactId}.',
         );
+        if (!onlyReturnEncryptedData) {
+          // The receipt stays, so the next retransmission tries again.
+          return null;
+        }
+        // The media upload goes out without this contact. Its receipt would
+        // otherwise be marked as uploaded together with the others.
         if (receipt.messageId != null) {
           await twonlyDB.messagesDao.handleMessageAckByServer(
             receipt.contactId,

lib/src/services/api/server_messages.api.dart

@@ -9,6 +9,7 @@ import 'package:mutex/mutex.dart';
 import 'package:twonly/globals.dart';
 import 'package:twonly/locator.dart';
 import 'package:twonly/src/database/daos/contacts.dao.dart';
+import 'package:twonly/src/database/tables/contacts.table.dart';
 import 'package:twonly/src/database/twonly.db.dart' hide Message;
 import 'package:twonly/src/model/protobuf/api/websocket/client_to_server.pb.dart'
     as client;
@@ -39,23 +40,6 @@ import 'package:twonly/src/services/signal/session.signal.dart';
 import 'package:twonly/src/utils/log.dart';
 import 'package:twonly/src/utils/misc.dart';
 
-const _signalUpgradeMailboxIdleDelay = Duration(milliseconds: 500);
-Timer? _signalUpgradeMailboxIdleTimer;
-
-void _scheduleSignalUpgradeAfterMailboxIdle() {
-  _signalUpgradeMailboxIdleTimer?.cancel();
-  _signalUpgradeMailboxIdleTimer = Timer(_signalUpgradeMailboxIdleDelay, () {
-    _signalUpgradeMailboxIdleTimer = null;
-    unawaited(upgradeSignalSessionsToV2());
-  });
-}
-
-void _upgradeSignalSessionsAfterMailboxDrain() {
-  _signalUpgradeMailboxIdleTimer?.cancel();
-  _signalUpgradeMailboxIdleTimer = null;
-  unawaited(upgradeSignalSessionsToV2());
-}
-
 Future<void> handleServerMessage(server.ServerToClient msg) async {
   Log.info('Processing a message from the server.');
   var receivedIncomingMessages = false;
@@ -78,7 +62,7 @@ Future<void> handleServerMessage(server.ServerToClient msg) async {
       Log.info(
         'Got ${msg.v0.newMessages.newMessages.length} messages from the server.',
       );
-      final brokenSessionsInCurrentBatch = <int>{};
+      final brokenSessionsInCurrentBatch = <(int, SignalVersion)>{};
       for (final newMessage in msg.v0.newMessages.newMessages) {
         try {
           await handleClient2ClientMessage(
@@ -116,12 +100,12 @@ Future<void> handleServerMessage(server.ServerToClient msg) async {
   Log.info('All messages from the server processed.');
 
   if (mailboxDrained) {
-    _upgradeSignalSessionsAfterMailboxDrain();
+    flushActiveSignalSessionUpgrades();
   } else if (receivedIncomingMessages) {
     // Released servers do not send a mailbox-drained marker. They wait for
-    // this ACK before sending the next batch, so a short, resettable idle
+    // this ACK before sending the next batch, so a resettable idle window
-    // window approximates the queue boundary without blocking delivery.
+    // approximates the queue boundary without blocking delivery.
-    _scheduleSignalUpgradeAfterMailboxIdle();
+    scheduleActiveSignalSessionUpgrades();
   }
 }
 
@@ -131,13 +115,17 @@ final Map<String, Mutex> _messageLocks = {};
 
 Future<void> handleClient2ClientMessage(
   NewMessage newMessage, {
-  Set<int>? brokenSessionsInCurrentBatch,
+  BrokenSignalSessions? brokenSessionsInCurrentBatch,
 }) async {
   final body = Uint8List.fromList(newMessage.body);
   final message = Message.fromBuffer(body);
   final receiptId = message.receiptId;
+  // Replies such as delivery receipts and decryption errors reuse the
+  // receipt id of the message they answer, so only the sender and the id
+  // together identify an incoming message.
+  final lockKey = '${newMessage.fromUserId}:$receiptId';
 
-  final mutex = _messageLocks.putIfAbsent(receiptId, Mutex.new);
+  final mutex = _messageLocks.putIfAbsent(lockKey, Mutex.new);
   if (mutex.isLocked) {
     Log.info(
       '[$receiptId] Skipping — already being processed by another handler',
@@ -152,7 +140,7 @@ Future<void> handleClient2ClientMessage(
         brokenSessionsInCurrentBatch: brokenSessionsInCurrentBatch,
       );
     } finally {
-      _messageLocks.remove(receiptId);
+      _messageLocks.remove(lockKey);
     }
   });
 }
@@ -160,12 +148,25 @@ Future<void> handleClient2ClientMessage(
 Future<void> _handleClient2ClientMessage(
   NewMessage newMessage,
   Message message, {
-  Set<int>? brokenSessionsInCurrentBatch,
+  BrokenSignalSessions? brokenSessionsInCurrentBatch,
 }) async {
   final fromUserId = newMessage.fromUserId.toInt();
   final receiptId = message.receiptId;
 
-  if (brokenSessionsInCurrentBatch?.contains(fromUserId) == true) {
+  final signalVersion = switch (message.type) {
+    Message_Type.CIPHERTEXT_V2 => SignalVersion.v2,
+    Message_Type.CIPHERTEXT || Message_Type.PREKEY_BUNDLE => SignalVersion.v1,
+    // Receipts and decryption errors are not encrypted and are never skipped.
+    _ => null,
+  };
+  if (signalVersion == SignalVersion.v1) {
+    // Only sessions participating in current traffic are migration targets.
+    // Each message resets the idle window so a server batch finishes first.
+    queueSignalSessionUpgrade(fromUserId);
+  }
+  if (signalVersion != null &&
+      brokenSessionsInCurrentBatch?.contains((fromUserId, signalVersion)) ==
+          true) {
     // This happens when a session goes out of sync (e.g. wrong message order).
     // We skip the remaining messages in the batch because each failed decryption
     // attempt is extremely slow (~1.2s) and would otherwise freeze the app.
@@ -181,13 +182,20 @@ Future<void> _handleClient2ClientMessage(
   }
 
   if (await twonlyDB.receiptsDao.isDuplicated(receiptId)) {
-    if (message.type == Message_Type.SENDER_DELIVERY_RECEIPT) {
+    if (signalVersion == null) {
+      // Only encrypted messages are confirmed. The others, delivery receipts
+      // and decryption errors, answer one of our messages: the first copy
+      // was handled, and answering a reply starts a loop.
       Log.info(
-        '[$receiptId] Delivery receipt is a duplicate. Skipping receipt response.',
+        '[$receiptId] Unencrypted message is a duplicate. Skipping receipt response.',
       );
       return;
     }
-    const duplicateReceiptCooldown = Duration(days: 10);
+    // A sender only retries a message it never saw confirmed, so our
+    // delivery receipt got lost and has to be sent again. The cooldown only
+    // keeps a batch the server delivers twice from being answered twice.
+    // 0.5.2 waited ten days here while the sender kept retrying.
+    const duplicateReceiptCooldown = Duration(hours: 1);
     final shouldResend = await twonlyDB.receiptsDao.claimDuplicateReceiptResend(
       receiptId,
       duplicateReceiptCooldown,
@@ -223,6 +231,12 @@ Future<void> _handleClient2ClientMessage(
 
   Log.info('[$receiptId] Started processing message');
 
+  // A message we could not process is answered with an error, and the sender
+  // sends it again. Remembering its receipt id would turn the next copy into
+  // a duplicate, which is answered with a delivery receipt for content we
+  // never read, once the error reply got lost.
+  var processed = true;
+
   switch (message.type) {
     case Message_Type.SENDER_DELIVERY_RECEIPT:
       Log.info('[$receiptId] Got delivery receipt!');
@@ -329,6 +343,7 @@ Future<void> _handleClient2ClientMessage(
         }
 
         response ??= Message(type: Message_Type.SENDER_DELIVERY_RECEIPT);
+        processed = response.type != Message_Type.PLAINTEXT_CONTENT;
 
         String? targetReceiptId;
         try {
@@ -359,7 +374,7 @@ Future<void> _handleClient2ClientMessage(
   }
 
   try {
-    await twonlyDB.receiptsDao.gotReceipt(receiptId);
+    if (processed) await twonlyDB.receiptsDao.gotReceipt(receiptId);
     Log.info('[$receiptId] Finished processing');
   } catch (e) {
     Log.warn('[$receiptId] Error marking message as received: $e');
@@ -375,7 +390,7 @@ Future<(EncryptedContent?, PlaintextContent?)> handleEncryptedMessageRaw(
   Uint8List encryptedContentRaw,
   Message_Type messageType,
   String receiptId, {
-  Set<int>? brokenSessionsInCurrentBatch,
+  BrokenSignalSessions? brokenSessionsInCurrentBatch,
 }) async {
   Log.info('[$receiptId] calling signalDecryptMessage');
   EncryptedContent? encryptedContent;

lib/src/services/signal/encryption.signal.dart

@@ -1,9 +1,13 @@
 import 'dart:typed_data';
 
+import 'package:drift/drift.dart' as drift;
 import 'package:libsignal_protocol_dart/libsignal_protocol_dart.dart';
 // ignore: implementation_imports
 import 'package:libsignal_protocol_dart/src/invalid_message_exception.dart';
 import 'package:twonly/core/bridge/wrapper/signal.dart';
+import 'package:twonly/locator.dart';
+import 'package:twonly/src/database/tables/contacts.table.dart';
+import 'package:twonly/src/database/twonly.db.dart';
 import 'package:twonly/src/model/protobuf/client/generated/messages.pb.dart'
     as pb;
 import 'package:twonly/src/services/api/messages.api.dart';
@@ -72,11 +76,15 @@ Future<SignalEncryptResult?> _signalEncryptMessageV1(
   }
 }
 
+/// Contacts whose session failed in the current batch of server messages, per
+/// Signal version: a broken V1 session says nothing about the V2 session.
+typedef BrokenSignalSessions = Set<(int, SignalVersion)>;
+
 Future<(pb.EncryptedContent?, pb.PlaintextContent_DecryptionErrorMessage_Type?)>
 signalDecryptMessageV2(
   int fromUserId,
   Uint8List encryptedContentRaw, {
-  Set<int>? brokenSessionsInCurrentBatch,
+  BrokenSignalSessions? brokenSessionsInCurrentBatch,
 }) async {
   Log.info('Acquiring lockingSignalProtocol for $fromUserId (V2)');
   final (
@@ -91,9 +99,20 @@ signalDecryptMessageV2(
         deviceId: 1,
         ciphertext: encryptedContentRaw,
       );
-      recordResyncAttempt(fromUserId, success: true);
+      recordResyncAttempt(fromUserId, SignalVersion.v2, success: true);
       return (pb.EncryptedContent.fromBuffer(plaintext), null, false);
     } catch (e) {
+      if (_isDuplicatedMessageError(e)) {
+        // Same as the DuplicateMessageException in V1: the message was
+        // decrypted before but not processed, so the session is intact and
+        // only the message has to be requested again.
+        Log.info('Message (V2) from $fromUserId was already decrypted: $e');
+        return (
+          null,
+          pb.PlaintextContent_DecryptionErrorMessage_Type.UNKNOWN,
+          false,
+        );
+      }
       Log.error('Could not decrypt message (V2) from $fromUserId: $e');
       return (
         null,
@@ -105,11 +124,15 @@ signalDecryptMessageV2(
 
   Log.info('Released lockingSignalProtocol for $fromUserId (V2)');
 
+  if (decryptedContent != null) {
+    await _useSignalV2For(fromUserId);
+  }
+
   if (needsResync) {
-    brokenSessionsInCurrentBatch?.add(fromUserId);
+    brokenSessionsInCurrentBatch?.add((fromUserId, SignalVersion.v2));
-    if (shouldAttemptResync(fromUserId)) {
+    if (shouldAttemptResync(fromUserId, SignalVersion.v2)) {
       if (await handleSessionResync(fromUserId)) {
-        recordResyncAttempt(fromUserId, success: false);
+        recordResyncAttempt(fromUserId, SignalVersion.v2, success: false);
         await sendCipherText(
           fromUserId,
           pb.EncryptedContent(
@@ -130,7 +153,7 @@ signalDecryptMessageV1(
   int fromUserId,
   Uint8List encryptedContentRaw,
   int type, {
-  Set<int>? brokenSessionsInCurrentBatch,
+  BrokenSignalSessions? brokenSessionsInCurrentBatch,
 }) async {
   Log.info('Acquiring lockingSignalProtocol for $fromUserId (V1)');
   final (
@@ -143,10 +166,9 @@ signalDecryptMessageV1(
       // Yield execution to the event loop to prevent UI freezing during bulk decryption
       await Future.delayed(Duration.zero);
 
-      final session = SessionCipher.fromStore(
+      final signalStore = (await getSignalStore())!;
-        (await getSignalStore())!,
+      final address = getSignalAddress(fromUserId);
-        getSignalAddress(fromUserId),
+      final session = SessionCipher.fromStore(signalStore, address);
-      );
 
       Uint8List plaintext;
 
@@ -156,9 +178,17 @@ signalDecryptMessageV1(
             PreKeySignalMessage(encryptedContentRaw),
           );
         case CiphertextMessage.whisperType:
-          plaintext = await session.decryptFromSignal(
+          final message = SignalMessage.fromSerialized(encryptedContentRaw);
-            SignalMessage.fromSerialized(encryptedContentRaw),
+          if (isBeyondReceivingWindow(
+            await signalStore.loadSession(address),
+            message,
+          )) {
+            throw InvalidMessageException(
+              'No valid sessions. Counter ${message.getCounter()} is beyond '
+              'the receiving window of every session state.',
             );
+          }
+          plaintext = await session.decryptFromSignal(message);
         default:
           Log.error('Unknown Message Decryption Type: $type');
           return (
@@ -168,7 +198,7 @@ signalDecryptMessageV1(
           );
       }
 
-      recordResyncAttempt(fromUserId, success: true);
+      recordResyncAttempt(fromUserId, SignalVersion.v1, success: true);
       return (pb.EncryptedContent.fromBuffer(plaintext), null, false);
     } on InvalidKeyIdException catch (e) {
       Log.warn(e);
@@ -209,12 +239,30 @@ signalDecryptMessageV1(
   // Handle session resync OUTSIDE the lock to avoid holding it during
   // network round-trips (which can block for up to 60 seconds)
   if (needsResync) {
-    brokenSessionsInCurrentBatch?.add(fromUserId);
+    brokenSessionsInCurrentBatch?.add((fromUserId, SignalVersion.v1));
-    if (shouldAttemptResync(fromUserId)) {
+    if (shouldAttemptResync(fromUserId, SignalVersion.v1)) {
-      if (await handleSessionResync(fromUserId)) {
+      return (decryptedContent, await _recoverFromBrokenV1Session(fromUserId));
+    }
+  }
+
+  return (decryptedContent, errorType);
+}
+
+/// Repairs the connection after a V1 message could not be decrypted and
+/// returns the decryption error to report back to the sender.
+Future<pb.PlaintextContent_DecryptionErrorMessage_Type>
+_recoverFromBrokenV1Session(int fromUserId) async {
+  var contact = await twonlyDB.contactsDao.getContactById(fromUserId);
+  if (contact?.signalVersion == SignalVersion.v2) {
+    // Our V2 session is not affected, rebuilding it would not help.
+    recordResyncAttempt(fromUserId, SignalVersion.v1, success: false);
+  } else {
+    if (!await handleSessionResync(fromUserId)) {
+      return pb.PlaintextContent_DecryptionErrorMessage_Type.UNKNOWN;
+    }
     // This flag prevents from resyncing the session the client received
     // multiple new messages from the server he could not decrypt
-        recordResyncAttempt(fromUserId, success: false);
+    recordResyncAttempt(fromUserId, SignalVersion.v1, success: false);
 
     // This message contains a new PreKeyBundle establishing a new signal
     // session
@@ -226,9 +274,63 @@ signalDecryptMessageV1(
         ),
       ),
     );
+    contact = await twonlyDB.contactsDao.getContactById(fromUserId);
+    if (contact?.signalVersion != SignalVersion.v2) {
+      return pb.PlaintextContent_DecryptionErrorMessage_Type.UNKNOWN;
     }
   }
+  // We answer this contact over V2, so a SESSION_OUT_OF_SYNC travels over V2
+  // as well and leaves the sender's V1 session broken: versions before 0.5.3
+  // keep sending V1 until they process our bundle. PREKEY_UNKNOWN makes the
+  // sender (any version) rebuild its session from our current bundle, which
+  // is V2 once we publish one, and resend the message.
+  return pb.PlaintextContent_DecryptionErrorMessage_Type.PREKEY_UNKNOWN;
 }
 
-  return (decryptedContent, errorType);
+/// Receiving over V2 proves that the Rust store holds a working session with
+/// the sender. A contact still marked V1 is switched, so replies no longer go
+/// over a V1 session that the sender may have repaired over V2 only.
+Future<void> _useSignalV2For(int contactId) async {
+  final contact = await twonlyDB.contactsDao.getContactById(contactId);
+  if (contact == null || contact.signalVersion == SignalVersion.v2) return;
+  Log.info('Got a V2 message from $contactId, switching the contact to V2.');
+  await twonlyDB.contactsDao.updateContact(
+    contactId,
+    const ContactsCompanion(signalVersion: drift.Value(SignalVersion.v2)),
+  );
+}
+
+/// Rust reports libsignal's DuplicatedMessage as
+/// `TwonlyError::DuplicatedSignalMessage`, which reaches Dart as text only.
+bool _isDuplicatedMessageError(Object error) =>
+    error.toString().contains('Duplicated Signal message');
+
+/// libsignal_protocol_dart refuses to skip more message keys than this on a
+/// single receiving chain (see SessionCipher._getOrCreateMessageKeys).
+const _maxSkippedMessageKeys = 2000;
+
+/// Whether libsignal_protocol_dart is certain to reject [message] because its
+/// counter lies more than [_maxSkippedMessageKeys] ahead in every session state
+/// that could otherwise decrypt it.
+///
+/// libsignal only reaches that verdict after deriving a new receiving chain,
+/// two ECDH operations and a key generation, for each archived state. With a
+/// few dozen states that blocks the isolate for seconds per message, while the
+/// chain indices alone give the same answer.
+bool isBeyondReceivingWindow(SessionRecord record, SignalMessage message) {
+  final ratchetKey = message.getSenderRatchetKey();
+  final counter = message.getCounter();
+  var rejectedStates = 0;
+  for (final state in [record.sessionState, ...record.previousSessionStates]) {
+    // libsignal rejects these states before it looks at the counter.
+    if (!state.hasSenderChain() ||
+        state.getSessionVersion() != message.getMessageVersion()) {
+      continue;
+    }
+    // Without a chain for this ratchet key libsignal starts one at index 0.
+    final index = state.getReceiverChainKey(ratchetKey)?.index ?? 0;
+    if (counter - index <= _maxSkippedMessageKeys) return false;
+    rejectedStates++;
+  }
+  return rejectedStates > 0;
 }

lib/src/services/signal/protocol_state.signal.dart

@@ -1,29 +1,39 @@
 import 'dart:math';
 import 'package:mutex/mutex.dart';
+import 'package:twonly/src/database/tables/contacts.table.dart';
 
 /// Unified lock for all Signal protocol operations (encryption, decryption, session management).
 final lockingSignalProtocol = Mutex();
 
-/// Tracking users who have already been resynced in the current session.
+/// Tracks recovery attempts independently for the legacy and PQ sessions of a
-final Map<int, ({int failureCount, DateTime lastAttempt})> _resyncAttempts = {};
+/// contact. A stale V1 message must not prevent V2 recovery (or vice versa).
+final Map<(int, SignalVersion), ({int failureCount, DateTime lastAttempt})>
+_resyncAttempts = {};
 
 const int maxResyncAttempts = 3;
 
-bool shouldAttemptResync(int userId) {
+bool shouldAttemptResync(int userId, SignalVersion signalVersion) {
-  final attempt = _resyncAttempts[userId];
+  final attempt = _resyncAttempts[(userId, signalVersion)];
   if (attempt == null) return true;
   if (attempt.failureCount >= maxResyncAttempts) return false;
 
-  final cooldown = Duration(minutes: 5 * pow(5, attempt.failureCount - 1).toInt());
+  final cooldown = Duration(
+    minutes: 5 * pow(5, attempt.failureCount - 1).toInt(),
+  );
   return DateTime.now().difference(attempt.lastAttempt) > cooldown;
 }
 
-void recordResyncAttempt(int userId, {required bool success}) {
+void recordResyncAttempt(
+  int userId,
+  SignalVersion signalVersion, {
+  required bool success,
+}) {
+  final key = (userId, signalVersion);
   if (success) {
-    _resyncAttempts.remove(userId);
+    _resyncAttempts.remove(key);
   } else {
-    final current = _resyncAttempts[userId];
+    final current = _resyncAttempts[key];
-    _resyncAttempts[userId] = (
+    _resyncAttempts[key] = (
       failureCount: (current?.failureCount ?? 0) + 1,
       lastAttempt: DateTime.now(),
     );

lib/src/services/signal/session.signal.dart

@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'dart:typed_data';
 
 import 'package:drift/drift.dart' as drift;
@@ -14,7 +15,24 @@ import 'package:twonly/src/services/signal/protocol_state.signal.dart';
 import 'package:twonly/src/services/signal/utils.signal.dart';
 import 'package:twonly/src/utils/log.dart';
 
-final Mutex _upgradeSignalSessionsLock = Mutex();
+class _SignalUpgradeState {
+  final lock = Mutex();
+  final activeV1Contacts = <int>{};
+  Timer? idleTimer;
+}
+
+/// Production isolates have one database, while integration tests host
+/// multiple clients in zones in the same isolate. Keeping the queue per
+/// database prevents one client's activity from being migrated in another
+/// client's Signal store.
+final Map<TwonlyDB, _SignalUpgradeState> _signalUpgradeStates = {};
+
+_SignalUpgradeState get _signalUpgradeState => _signalUpgradeStates.putIfAbsent(
+  twonlyDB,
+  _SignalUpgradeState.new,
+);
+
+const _signalUpgradeIdleDelay = Duration(milliseconds: 500);
 
 Future<bool> processSignalUserData(Response_UserData userData) async {
   return lockingSignalProtocol.protect(() async {
@@ -22,40 +40,80 @@ Future<bool> processSignalUserData(Response_UserData userData) async {
   });
 }
 
-/// Upgrades every known v1 contact that currently publishes a PQXDH bundle.
+/// Adds a V1 session to the active migration set without starting migration.
+/// Incoming batches use this so migration cannot begin midway through a slow
+/// decryption pass.
+void queueSignalSessionUpgrade(int contactId) {
+  final state = _signalUpgradeState;
+  state.activeV1Contacts.add(contactId);
+}
+
+/// Starts or resets the idle window for queued active V1 sessions.
+void scheduleActiveSignalSessionUpgrades() {
+  final state = _signalUpgradeState;
+  state.idleTimer?.cancel();
+  state.idleTimer = Timer(_signalUpgradeIdleDelay, () {
+    state.idleTimer = null;
+    unawaited(upgradeActiveSignalSessionsToV2());
+  });
+}
+
+/// Marks an outgoing V1 session as active and schedules its migration.
+void scheduleSignalSessionUpgrade(int contactId) {
+  queueSignalSessionUpgrade(contactId);
+  scheduleActiveSignalSessionUpgrades();
+}
+
+/// Runs pending active-session upgrades immediately after the mailbox drains.
+void flushActiveSignalSessionUpgrades() {
+  final state = _signalUpgradeState;
+  state.idleTimer?.cancel();
+  state.idleTimer = null;
+  unawaited(upgradeActiveSignalSessionsToV2());
+}
+
+/// Upgrades active V1 contacts that currently publish a PQXDH bundle.
 ///
 /// The legacy session is deliberately retained: messages that were already in
 /// flight before the upgrade can still be decrypted, while newly queued
 /// messages use the contact's updated v2 session.
-Future<void> upgradeSignalSessionsToV2() async {
+Future<void> upgradeActiveSignalSessionsToV2() async {
-  if (_upgradeSignalSessionsLock.isLocked) return;
+  final state = _signalUpgradeState;
-
   try {
-    await _upgradeSignalSessionsLock.protect(() async {
+    await state.lock.protect(() async {
-      final contacts = await twonlyDB.contactsDao.getAllContacts();
+      // Activity may arrive during an upgrade. Drain until every contact
-      final legacyContacts = contacts.where(
+      // queued by that activity has been considered.
-        (contact) =>
+      while (state.activeV1Contacts.isNotEmpty) {
-            contact.signalVersion == SignalVersion.v1 &&
+        final contactIds = Set<int>.of(state.activeV1Contacts);
-            !contact.accountDeleted,
+        state.activeV1Contacts.removeAll(contactIds);
+
+        for (final contactId in contactIds) {
+          try {
+            final contact = await twonlyDB.contactsDao.getContactById(
+              contactId,
             );
+            if (contact == null ||
+                contact.signalVersion != SignalVersion.v1 ||
+                contact.accountDeleted) {
+              continue;
+            }
 
-      for (final contact in legacyContacts) {
+            final userData = await apiService.getUserById(contactId);
-        try {
-          final userData = await apiService.getUserById(contact.userId);
             if (userData == null || !userData.hasPqcBundle()) continue;
 
             if (await processSignalUserData(userData)) {
-            Log.info('Upgraded Signal session with ${contact.userId} to v2.');
+              Log.info('Upgraded active Signal session with $contactId to v2.');
             }
           } catch (error) {
-          // Migration is best-effort. A contact that is temporarily
+            // Further traffic queues the contact again, so temporary failures
-          // unavailable remains on v1 and is retried after a later message
+            // cannot leave an active session on V1 indefinitely.
-          // queue drain.
             Log.warn(
-            'Could not upgrade Signal session with ${contact.userId}: $error',
+              'Could not upgrade active Signal session with $contactId: '
+              '$error',
             );
           }
         }
+      }
     });
   } catch (error) {
     // This function is started without awaiting it from the message loop.
@@ -76,7 +134,13 @@ Future<bool> resetSignalSession(int contactId) async {
       await signalStore?.deleteAllSessions(contactId.toString());
     }
     final reset = await _processSignalUserData(userData);
-    if (reset) recordResyncAttempt(contactId, success: true);
+    if (reset) {
+      recordResyncAttempt(
+        contactId,
+        userData.hasPqcBundle() ? SignalVersion.v2 : SignalVersion.v1,
+        success: true,
+      );
+    }
     return reset;
   });
 }

lib/src/utils/log.dart

@@ -22,7 +22,7 @@ class Log {
             record.level >= Level.WARNING) {
           // ignore: avoid_print
           print(
-            '${record.level.name} [${AppState.isInBackgroundTask ? 'b' : 'f'}] [twonly] ${record.loggerName} > ${record.message}',
+            '${record.level.name} [${_isolateTag(record)}] [twonly] ${record.loggerName} > ${record.message}',
           );
         }
       }
@@ -77,6 +77,20 @@ class Log {
     final message = filterLogMessage('$messageInput');
     Logger(_getCallerSourceCodeFilename()).fine(message, error, stackTrace);
   }
+
+  /// Logs a line forwarded from Rust.
+  static void rust(String message) {
+    Logger(_rustLoggerName).fine(filterLogMessage(message));
+  }
+}
+
+const _rustLoggerName = 'rust';
+
+/// Rust forwards its logs to whichever isolate registered its log sink last,
+/// so the receiving isolate says nothing about who caused a Rust line.
+String _isolateTag(LogRecord record) {
+  if (record.loggerName == _rustLoggerName) return 'r';
+  return AppState.isInBackgroundTask ? 'b' : 'f';
 }
 
 Future<String> loadLogFile() async {
@@ -115,7 +129,7 @@ Future<void> _writeLogToFile(LogRecord record) async {
   final logFile = File('${AppEnvironment.supportDir}/app.log');
 
   final logMessage =
-      '${clock.now()} ${record.level.name} [${AppState.isInBackgroundTask ? 'b' : 'f'}] [twonly] ${record.loggerName} > ${record.message}\n';
+      '${clock.now()} ${record.level.name} [${_isolateTag(record)}] [twonly] ${record.loggerName} > ${record.message}\n';
 
   return _protectFileAccess(() async {
     if (!logFile.existsSync()) {

lib/src/visual/views/memories/components/memory_thumbnail.comp.dart

@@ -94,7 +94,9 @@ class _MemoriesThumbnailCompState extends State<MemoriesThumbnailComp> {
       if (hasStored) {
         unawaited(
           media.createThumbnail().then((_) {
-            if (mounted) {
+            // Only a written thumbnail is worth resolving again. A video whose
+            // thumbnail cannot be created would otherwise retry forever.
+            if (mounted && media.thumbnailPath.existsSync()) {
               _resolveImage();
             }
           }),

lib/src/visual/views/settings/help/diagnostics.view.dart

@@ -333,7 +333,8 @@ class _LogEntry {
         .trim()
         .replaceAll('[twonly] ', '')
         .replaceAll('[f] ', '')
-        .replaceAll('[b] ', '');
+        .replaceAll('[b] ', '')
+        .replaceAll('[r] ', '');
 
     final fileNameS = msg.split(' > ');
     final fileName = fileNameS[0];

rust/src/error.rs

@@ -70,6 +70,12 @@ pub enum TwonlyError {
 
     #[error("Signal protocol error: {0}")]
     Signal(String),
+
+    /// libsignal refused a message whose counter was already used, so the
+    /// message was decrypted before and the session is intact. Dart matches
+    /// this text to tell it apart from a broken session.
+    #[error("Duplicated Signal message: {0}")]
+    DuplicatedSignalMessage(String),
 }
 
 impl From<String> for TwonlyError {

rust/src/signal/engine.rs

@@ -3,7 +3,7 @@ use libsignal_protocol::{
     message_encrypt, process_prekey_bundle, CiphertextMessageType, DeviceId, GenericSignedPreKey,
     IdentityKey, IdentityKeyPair, IdentityKeyStore, KyberPreKeyId, KyberPreKeyStore, PreKeyBundle,
     PreKeyId, PreKeySignalMessage, PreKeyStore, ProtocolAddress, PublicKey, SessionStore,
-    SignalMessage, SignedPreKeyId, SignedPreKeyStore, Timestamp,
+    SignalMessage, SignalProtocolError, SignedPreKeyId, SignedPreKeyStore, Timestamp,
 };
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
@@ -549,7 +549,7 @@ impl RustSignalEngine {
                 )
                 .assert_send()
                 .await
-                .map_err(|e| TwonlyError::Signal(e.to_string()))?
+                .map_err(decrypt_error)?
             }
             CiphertextMessageType::PreKey => {
                 let message = PreKeySignalMessage::try_from(ciphertext)
@@ -567,7 +567,7 @@ impl RustSignalEngine {
                 )
                 .assert_send()
                 .await
-                .map_err(|e| TwonlyError::Signal(e.to_string()))?
+                .map_err(decrypt_error)?
             }
             _ => {
                 return Err(TwonlyError::Signal("Invalid message type".to_string()));
@@ -578,6 +578,15 @@ impl RustSignalEngine {
     }
 }
 
+fn decrypt_error(error: SignalProtocolError) -> TwonlyError {
+    match error {
+        SignalProtocolError::DuplicatedMessage(..) => {
+            TwonlyError::DuplicatedSignalMessage(error.to_string())
+        }
+        error => TwonlyError::Signal(error.to_string()),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -706,4 +715,71 @@ mod tests {
             .await
             .is_ok());
     }
+
+    #[tokio::test]
+    async fn decrypting_a_message_twice_reports_a_duplicate() {
+        let (alice_engine, _alice_dir) = create_test_engine("alice").await;
+        let (bob_engine, _bob_dir) = create_test_engine("bob").await;
+
+        alice_engine
+            .process_prekey_bundle(
+                "bob".to_string(),
+                1,
+                bob_engine.generate_bundle().await.unwrap(),
+            )
+            .await
+            .unwrap();
+
+        let prekey_message = alice_engine
+            .encrypt_message("bob".to_string(), 1, b"first".to_vec())
+            .await
+            .unwrap();
+        assert_eq!(
+            prekey_message.last(),
+            Some(&(CiphertextMessageType::PreKey as u8))
+        );
+        bob_engine
+            .decrypt_message("alice".to_string(), 1, prekey_message.clone())
+            .await
+            .unwrap();
+        let error = bob_engine
+            .decrypt_message("alice".to_string(), 1, prekey_message)
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(error, TwonlyError::DuplicatedSignalMessage(_)),
+            "{error}"
+        );
+
+        // After Bob's reply Alice no longer needs a PreKey message.
+        let reply = bob_engine
+            .encrypt_message("alice".to_string(), 1, b"reply".to_vec())
+            .await
+            .unwrap();
+        alice_engine
+            .decrypt_message("bob".to_string(), 1, reply)
+            .await
+            .unwrap();
+        let whisper_message = alice_engine
+            .encrypt_message("bob".to_string(), 1, b"second".to_vec())
+            .await
+            .unwrap();
+        assert_eq!(
+            whisper_message.last(),
+            Some(&(CiphertextMessageType::Whisper as u8))
+        );
+        bob_engine
+            .decrypt_message("alice".to_string(), 1, whisper_message.clone())
+            .await
+            .unwrap();
+        let error = bob_engine
+            .decrypt_message("alice".to_string(), 1, whisper_message)
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(error, TwonlyError::DuplicatedSignalMessage(_)),
+            "{error}"
+        );
+        assert!(error.to_string().starts_with("Duplicated Signal message"));
+    }
 }

test/services/message_retransmission_test.dart

@@ -0,0 +1,318 @@
+import 'dart:async';
+import 'dart:io';
+
+import 'package:clock/clock.dart';
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:fixnum/fixnum.dart';
+import 'package:flutter/services.dart';
+import 'package:flutter_rust_bridge/flutter_rust_bridge_for_generated.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:hashlib/random.dart';
+import 'package:logging/logging.dart';
+import 'package:twonly/core/bridge.dart' as bridge;
+import 'package:twonly/core/frb_generated.dart';
+import 'package:twonly/globals.dart';
+import 'package:twonly/locator.dart';
+import 'package:twonly/src/callbacks/callbacks.dart';
+import 'package:twonly/src/database/tables/contacts.table.dart';
+import 'package:twonly/src/database/twonly.db.dart';
+import 'package:twonly/src/model/protobuf/api/websocket/server_to_client.pb.dart'
+    show NewMessage;
+import 'package:twonly/src/model/protobuf/client/generated/messages.pb.dart'
+    as pb;
+import 'package:twonly/src/services/api/api.service.dart';
+import 'package:twonly/src/services/api/messages.api.dart';
+import 'package:twonly/src/services/api/server_messages.api.dart';
+import 'package:twonly/src/services/signal/protocol_state.signal.dart';
+import 'package:twonly/src/services/signal/utils.signal.dart';
+import 'package:twonly/src/services/user.service.dart';
+import 'package:workmanager/workmanager.dart';
+
+import '../mocks/platform_channels.dart';
+import '../mocks/test_client.dart';
+import '../mocks/workmanager.dart';
+
+/// Runs [action] as [client] and returns what it logged meanwhile.
+Future<List<String>> logsOf(
+  TestClient client,
+  Future<void> Function() action,
+) async {
+  final lines = <String>[];
+  final subscription = Logger.root.onRecord.listen((record) {
+    if (record.zone?[#userId] == client.env.userId) lines.add(record.message);
+  });
+  try {
+    await client.run(action);
+    return lines;
+  } finally {
+    await subscription.cancel();
+  }
+}
+
+/// A text from [sender] to [receiver] as the server would deliver it. It is
+/// encrypted, but never sent.
+Future<NewMessage> textFrom(
+  TestClient sender,
+  TestClient receiver,
+  String text,
+) => sender.run(() async {
+  final (body, _) = (await sendCipherText(
+    receiver.realUserId,
+    pb.EncryptedContent(
+      groupId: sender.defaultGroup!.groupId,
+      textMessage: pb.EncryptedContent_TextMessage()
+        ..senderMessageId = uuid.v4()
+        ..text = text,
+    ),
+    onlyReturnEncryptedData: true,
+  ))!;
+  return NewMessage(body: body, fromUserId: Int64(sender.realUserId));
+});
+
+NewMessage withReceiptId(NewMessage message, String receiptId) => NewMessage(
+  body: (pb.Message.fromBuffer(
+    message.body,
+  )..receiptId = receiptId).writeToBuffer(),
+  fromUserId: message.fromUserId,
+);
+
+Future<T> later<T>(Duration elapsed, Future<T> Function() action) =>
+    withClock(Clock.fixed(DateTime.now().add(elapsed)), action);
+
+void main() {
+  if (!Platform.isMacOS) {
+    return;
+  }
+
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  late Directory tempDir;
+
+  setUpAll(() async {
+    Logger.root.level = Level.ALL;
+    driftRuntimeOptions.dontWarnAboutMultipleDatabases = true;
+    WorkmanagerPlatform.instance = MockWorkmanagerPlatform();
+    tempDir = Directory.systemTemp.createTempSync('twonly_duplicates_test_');
+    AppEnvironment.initTesting(
+      customCacheDir: tempDir.path,
+      customSupportDir: tempDir.path,
+    );
+
+    final dylibPath =
+        '${Directory.current.path}/rust/target/debug/librust_lib_twonly.dylib';
+    if (File(dylibPath).existsSync()) {
+      await RustLib.init(externalLibrary: ExternalLibrary.open(dylibPath));
+    } else {
+      await RustLib.init();
+    }
+    await initFlutterCallbacksForRust();
+
+    await bridge.initializeTwonlyFlutter(
+      config: bridge.InitConfig(
+        databaseDir: tempDir.path,
+        dataDir: tempDir.path,
+      ),
+    );
+
+    if (locator.isRegistered<TwonlyDB>()) await locator.unregister<TwonlyDB>();
+    if (locator.isRegistered<UserService>()) {
+      await locator.unregister<UserService>();
+    }
+    if (locator.isRegistered<ApiService>()) {
+      await locator.unregister<ApiService>();
+    }
+
+    locator
+      ..registerFactory<TwonlyDB>(() {
+        final db = Zone.current[#twonlyDB] as TwonlyDB?;
+        if (db != null) return db;
+        throw StateError('No TwonlyDB in active Zone.');
+      })
+      ..registerFactory<UserService>(() {
+        final us = Zone.current[#userService] as UserService?;
+        if (us != null) return us;
+        throw StateError('No UserService in active Zone.');
+      })
+      ..registerFactory<ApiService>(() {
+        final api = Zone.current[#apiService] as ApiService?;
+        if (api != null) return api;
+        throw StateError('No ApiService in active Zone.');
+      });
+  });
+
+  tearDownAll(() async {
+    if (tempDir.existsSync()) {
+      try {
+        tempDir.deleteSync(recursive: true);
+      } catch (_) {}
+    }
+  });
+
+  // Alice sends, Bob receives.
+  late TestClient alice;
+  late TestClient bob;
+
+  setUp(() async {
+    setupPlatformChannelMocks();
+    HttpOverrides.global = RealHttpOverrides();
+    TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+        .setMockMethodCallHandler(
+          const MethodChannel('dev.fluttercommunity.plus/package_info'),
+          (call) async {
+            return {
+              'appName': 'twonly',
+              'packageName': 'eu.twonly.app',
+              'version': '1.0.0',
+              'buildNumber': '100',
+            };
+          },
+        );
+    await Workmanager().initialize(() {});
+
+    bob = TestClient(7003);
+    alice = TestClient(8003);
+    await bob.init(disablePqc: true);
+    await alice.init(disablePqc: true);
+    await alice.initContact(bob);
+    await bob.initContact(alice);
+  });
+
+  tearDown(() async {
+    await alice.run(() async => alice.api.close(null));
+    await bob.run(() async => bob.api.close(null));
+    await alice.env.db.close();
+    await bob.env.db.close();
+  });
+
+  group('Duplicate messages', () {
+    setUp(() async {
+      // Acknowledged sessions, so Alice sends plain Whisper messages.
+      await alice.sendText(bob, 'hello');
+      await bob.expectMessage((m) => m.content == 'hello');
+      await bob.sendText(alice, 'hi');
+      await alice.expectMessage((m) => m.content == 'hi');
+    });
+
+    test('a duplicate is confirmed again once the cooldown ended', () async {
+      final message = await textFrom(alice, bob, 'once');
+      await bob.run(() => handleClient2ClientMessage(message));
+      await bob.expectMessage((m) => m.content == 'once');
+
+      final duringCooldown = await logsOf(
+        bob,
+        () => handleClient2ClientMessage(message),
+      );
+      expect(
+        duringCooldown,
+        contains(contains('Skipping delivery receipt during cooldown')),
+      );
+
+      final afterCooldown = await logsOf(
+        bob,
+        () => later(
+          const Duration(hours: 2),
+          () => handleClient2ClientMessage(message),
+        ),
+      );
+      expect(
+        afterCooldown,
+        contains(contains('Sending delivery receipt again')),
+      );
+    });
+
+    test(
+      'a message that could not be decrypted is read when it comes again',
+      () async {
+        final first = await textFrom(alice, bob, 'first');
+        await bob.run(() => handleClient2ClientMessage(first));
+        await bob.expectMessage((m) => m.content == 'first');
+
+        // The same ciphertext under a new receipt id cannot be decrypted
+        // twice, so Bob answers it with a decryption error.
+        final receiptId = uuid.v4();
+        final logs = await logsOf(
+          bob,
+          () => handleClient2ClientMessage(withReceiptId(first, receiptId)),
+        );
+        expect(logs, contains(contains('Received message with old counter')));
+        expect(
+          await bob.run(() => twonlyDB.receiptsDao.isDuplicated(receiptId)),
+          isFalse,
+        );
+
+        // Alice resends it, and this time it can be decrypted.
+        final resent = withReceiptId(
+          await textFrom(alice, bob, 'resent'),
+          receiptId,
+        );
+        await bob.run(() => handleClient2ClientMessage(resent));
+        await bob.expectMessage((m) => m.content == 'resent');
+      },
+    );
+
+    test('a duplicate reply is never answered', () async {
+      final decryptionError = NewMessage(
+        body: pb.Message(
+          receiptId: uuid.v4(),
+          type: pb.Message_Type.PLAINTEXT_CONTENT,
+          plaintextContent: pb.PlaintextContent(
+            decryptionErrorMessage: pb.PlaintextContent_DecryptionErrorMessage(
+              type: pb.PlaintextContent_DecryptionErrorMessage_Type.UNKNOWN,
+            ),
+          ),
+        ).writeToBuffer(),
+        fromUserId: Int64(alice.realUserId),
+      );
+      await bob.run(() => handleClient2ClientMessage(decryptionError));
+
+      final logs = await logsOf(
+        bob,
+        () => later(
+          const Duration(days: 11),
+          () => handleClient2ClientMessage(decryptionError),
+        ),
+      );
+      expect(logs, contains(contains('Unencrypted message is a duplicate')));
+      expect(logs, isNot(contains(contains('Sending delivery receipt again'))));
+    });
+  });
+
+  group('Lost sessions', () {
+    setUp(
+      () => alice.run(() async {
+        await (await getSignalStore())!.deleteAllSessions(
+          bob.realUserId.toString(),
+        );
+      }),
+    );
+
+    test('a message without a session builds one and arrives', () async {
+      await alice.sendText(bob, 'after the session was lost');
+      await bob.expectMessage((m) => m.content == 'after the session was lost');
+    });
+
+    test('a message that cannot be encrypted waits for a retry', () async {
+      // The resync limiter refuses to rebuild the session for now.
+      for (var i = 0; i < maxResyncAttempts; i++) {
+        recordResyncAttempt(
+          bob.realUserId,
+          SignalVersion.v1,
+          success: false,
+        );
+      }
+      final message = await alice.sendText(bob, 'waiting');
+      final receipts = await alice.run(
+        () => twonlyDB.receiptsDao.getReceiptsByContactAndMessageId(
+          bob.realUserId,
+          message.messageId,
+        ),
+      );
+      expect(receipts, hasLength(1));
+      expect(receipts.single.ackByServerAt, isNull);
+
+      recordResyncAttempt(bob.realUserId, SignalVersion.v1, success: true);
+      await alice.run(retransmitAllMessages);
+      await bob.expectMessage((m) => m.content == 'waiting');
+    });
+  });
+}

test/services/retransmission_backoff_test.dart

@@ -0,0 +1,66 @@
+import 'package:clock/clock.dart';
+import 'package:flutter/foundation.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:twonly/src/database/twonly.db.dart';
+import 'package:twonly/src/services/api/messages.api.dart';
+
+void main() {
+  final lastRetry = DateTime(2026, 9, 24, 12);
+
+  Receipt receipt(int retryCount) => Receipt(
+    receiptId: 'receipt',
+    contactId: 1,
+    message: Uint8List(0),
+    contactWillSendsReceipt: true,
+    willBeRetriedByMediaUpload: false,
+    ackByServerAt: lastRetry,
+    retryCount: retryCount,
+    lastRetry: lastRetry,
+    createdAt: lastRetry,
+  );
+
+  bool dueAfter(
+    Duration elapsed,
+    Receipt receipt, {
+    Duration minimumPause = Duration.zero,
+  }) => withClock(
+    Clock.fixed(lastRetry.add(elapsed)),
+    () => isRetransmissionDue(receipt, minimumPause: minimumPause),
+  );
+
+  test('the first retry is not held back', () {
+    expect(dueAfter(Duration.zero, receipt(0)), isTrue);
+    expect(dueAfter(Duration.zero, receipt(1)), isTrue);
+  });
+
+  test('the pause doubles from one hour with every attempt', () {
+    for (final (retryCount, hours) in [(2, 1), (3, 2), (4, 4), (6, 16)]) {
+      final pause = Duration(hours: hours);
+      final r = receipt(retryCount);
+      expect(
+        dueAfter(pause - const Duration(minutes: 1), r),
+        isFalse,
+        reason: 'retry $retryCount',
+      );
+      expect(dueAfter(pause, r), isTrue, reason: 'retry $retryCount');
+    }
+  });
+
+  test('the pause stops growing at one day', () {
+    final stuck = receipt(65);
+    expect(dueAfter(const Duration(hours: 23), stuck), isFalse);
+    expect(dueAfter(const Duration(hours: 24), stuck), isTrue);
+  });
+
+  test('the pause never drops below the minimum', () {
+    const sixHours = Duration(hours: 6);
+    final r = receipt(2);
+    expect(
+      dueAfter(const Duration(hours: 5), r, minimumPause: sixHours),
+      isFalse,
+    );
+    expect(dueAfter(sixHours, r, minimumPause: sixHours), isTrue);
+    // The first retry is not held back by the minimum either.
+    expect(dueAfter(Duration.zero, receipt(1), minimumPause: sixHours), isTrue);
+  });
+}

test/signal/broken_session_recovery_test.dart

@@ -0,0 +1,403 @@
+import 'dart:async';
+import 'dart:io';
+
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:fixnum/fixnum.dart';
+import 'package:flutter/services.dart';
+import 'package:flutter_rust_bridge/flutter_rust_bridge_for_generated.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:hashlib/random.dart';
+import 'package:libsignal_protocol_dart/libsignal_protocol_dart.dart'
+    show CiphertextMessage;
+import 'package:logging/logging.dart';
+import 'package:twonly/core/bridge.dart' as bridge;
+import 'package:twonly/core/frb_generated.dart';
+import 'package:twonly/globals.dart';
+import 'package:twonly/locator.dart';
+import 'package:twonly/src/callbacks/callbacks.dart';
+import 'package:twonly/src/database/tables/contacts.table.dart';
+import 'package:twonly/src/database/twonly.db.dart';
+import 'package:twonly/src/model/protobuf/api/websocket/server_to_client.pb.dart'
+    show NewMessage;
+import 'package:twonly/src/model/protobuf/client/generated/messages.pb.dart'
+    as pb;
+import 'package:twonly/src/services/api/api.service.dart';
+import 'package:twonly/src/services/api/server_messages.api.dart';
+import 'package:twonly/src/services/signal/encryption.signal.dart';
+import 'package:twonly/src/services/signal/identity.signal.dart';
+import 'package:twonly/src/services/signal/session.signal.dart';
+import 'package:twonly/src/services/user.service.dart';
+import 'package:workmanager/workmanager.dart';
+
+import '../mocks/platform_channels.dart';
+import '../mocks/test_client.dart';
+import '../mocks/workmanager.dart';
+
+// Every client in this process shares one Rust Signal engine, whose identity
+// is the one imported last. A contact can therefore only move from V1 to V2
+// by processing the bundle of the client that was initialized last.
+
+Future<void> uploadPqcKeys(TestClient client) => client.run(() async {
+  await UserService.update((user) {
+    user.signalLastPqcPreKeysUploaded = null;
+  });
+  await SignalIdentityService.onAuthenticated();
+});
+
+Future<SignalVersion?> versionOf(TestClient client, TestClient contact) =>
+    client.run(
+      () async => (await client.env.db.contactsDao.getContactById(
+        contact.realUserId,
+      ))?.signalVersion,
+    );
+
+/// Runs [action] as [client] and returns what it logged meanwhile.
+Future<(T, List<String>)> logsOf<T>(
+  TestClient client,
+  Future<T> Function() action,
+) async {
+  final lines = <String>[];
+  final subscription = Logger.root.onRecord.listen((record) {
+    if (record.zone?[#userId] == client.env.userId) lines.add(record.message);
+  });
+  try {
+    return (await client.run(action), lines);
+  } finally {
+    await subscription.cancel();
+  }
+}
+
+Uint8List textContent(String text) => pb.EncryptedContent(
+  textMessage: pb.EncryptedContent_TextMessage()..text = text,
+).writeToBuffer();
+
+/// [sender] encrypts messages over V1 that never reach [receiver], until its
+/// sending chain is further ahead than [receiver] can follow.
+Future<void> breakV1Chain(TestClient sender, TestClient receiver) =>
+    sender.run(() async {
+      for (var i = 0; i < 2002; i++) {
+        await signalEncryptMessage(receiver.realUserId, textContent('lost $i'));
+      }
+    });
+
+void main() {
+  if (!Platform.isMacOS) {
+    return;
+  }
+
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  late Directory tempDir;
+
+  setUpAll(() async {
+    Logger.root.level = Level.ALL;
+    driftRuntimeOptions.dontWarnAboutMultipleDatabases = true;
+    WorkmanagerPlatform.instance = MockWorkmanagerPlatform();
+    tempDir = Directory.systemTemp.createTempSync('twonly_recovery_test_');
+    AppEnvironment.initTesting(
+      customCacheDir: tempDir.path,
+      customSupportDir: tempDir.path,
+    );
+
+    final dylibPath =
+        '${Directory.current.path}/rust/target/debug/librust_lib_twonly.dylib';
+    if (File(dylibPath).existsSync()) {
+      await RustLib.init(externalLibrary: ExternalLibrary.open(dylibPath));
+    } else {
+      await RustLib.init();
+    }
+    await initFlutterCallbacksForRust();
+
+    await bridge.initializeTwonlyFlutter(
+      config: bridge.InitConfig(
+        databaseDir: tempDir.path,
+        dataDir: tempDir.path,
+      ),
+    );
+
+    if (locator.isRegistered<TwonlyDB>()) await locator.unregister<TwonlyDB>();
+    if (locator.isRegistered<UserService>()) {
+      await locator.unregister<UserService>();
+    }
+    if (locator.isRegistered<ApiService>()) {
+      await locator.unregister<ApiService>();
+    }
+
+    locator
+      ..registerFactory<TwonlyDB>(() {
+        final db = Zone.current[#twonlyDB] as TwonlyDB?;
+        if (db != null) return db;
+        throw StateError('No TwonlyDB in active Zone.');
+      })
+      ..registerFactory<UserService>(() {
+        final us = Zone.current[#userService] as UserService?;
+        if (us != null) return us;
+        throw StateError('No UserService in active Zone.');
+      })
+      ..registerFactory<ApiService>(() {
+        final api = Zone.current[#apiService] as ApiService?;
+        if (api != null) return api;
+        throw StateError('No ApiService in active Zone.');
+      });
+  });
+
+  tearDownAll(() async {
+    if (tempDir.existsSync()) {
+      try {
+        tempDir.deleteSync(recursive: true);
+      } catch (_) {}
+    }
+  });
+
+  group('Signal session recovery', () {
+    // Alice sends, Bob receives. Bob is always initialized first, so the
+    // shared Rust identity is Alice's.
+    late TestClient alice;
+    late TestClient bob;
+
+    setUp(() async {
+      setupPlatformChannelMocks();
+      HttpOverrides.global = RealHttpOverrides();
+      TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+          .setMockMethodCallHandler(
+            const MethodChannel('dev.fluttercommunity.plus/package_info'),
+            (call) async {
+              return {
+                'appName': 'twonly',
+                'packageName': 'eu.twonly.app',
+                'version': '1.0.0',
+                'buildNumber': '100',
+              };
+            },
+          );
+      await Workmanager().initialize(() {});
+
+      bob = TestClient(7002);
+      alice = TestClient(8001);
+      await bob.init(disablePqc: true);
+      await alice.init(disablePqc: true);
+    });
+
+    tearDown(() async {
+      await alice.run(() async => alice.api.close(null));
+      await bob.run(() async => bob.api.close(null));
+      await alice.env.db.close();
+      await bob.env.db.close();
+    });
+
+    /// Both sides talk V1 and Alice's session is acknowledged, so she sends
+    /// plain Whisper messages. Alice then publishes a V2 bundle.
+    Future<void> talkOverV1ThenPublishV2({bool letBobSettle = false}) async {
+      await alice.initContact(bob);
+      await bob.initContact(alice);
+      await alice.sendText(bob, 'hello over V1');
+      await bob.expectMessage((m) => m.content == 'hello over V1');
+      await bob.sendText(alice, 'reply over V1');
+      await alice.expectMessage((m) => m.content == 'reply over V1');
+      if (letBobSettle) {
+        // Bob upgrades V1 contacts on his own shortly after messages arrive.
+        // Let that happen while Alice has no V2 bundle to upgrade to.
+        await Future<void>.delayed(const Duration(seconds: 2));
+      }
+      await uploadPqcKeys(alice);
+    }
+
+    test('a replayed V2 message leaves the session alone', () async {
+      await uploadPqcKeys(alice);
+      await uploadPqcKeys(bob);
+      await alice.initContact(bob);
+      await bob.initContact(alice);
+      expect(await versionOf(bob, alice), SignalVersion.v2);
+
+      final ciphertext = (await alice.run(
+        () => signalEncryptMessageV2(bob.realUserId, textContent('once')),
+      ))!.ciphertext;
+      final (first, _) = await bob.run(
+        () => signalDecryptMessageV2(alice.realUserId, ciphertext),
+      );
+      expect(first!.textMessage.text, 'once');
+
+      final ((replayed, errorType), logs) = await logsOf(
+        bob,
+        () => signalDecryptMessageV2(alice.realUserId, ciphertext),
+      );
+
+      expect(replayed, isNull);
+      expect(
+        errorType,
+        pb.PlaintextContent_DecryptionErrorMessage_Type.UNKNOWN,
+      );
+      expect(logs, isNot(contains(contains('re-sync the session'))));
+      expect(logs, isNot(contains(contains('Could not decrypt message'))));
+    });
+
+    test('a V2 message moves a V1 contact to V2', () async {
+      await uploadPqcKeys(bob);
+      await alice.initContact(bob);
+      await bob.initContact(alice);
+      expect(await versionOf(alice, bob), SignalVersion.v2);
+      expect(await versionOf(bob, alice), SignalVersion.v1);
+
+      final ciphertext = (await alice.run(
+        () => signalEncryptMessageV2(bob.realUserId, textContent('over V2')),
+      ))!.ciphertext;
+      final (content, _) = await bob.run(
+        () => signalDecryptMessageV2(alice.realUserId, ciphertext),
+      );
+
+      expect(content!.textMessage.text, 'over V2');
+      expect(await versionOf(bob, alice), SignalVersion.v2);
+    });
+
+    test(
+      'a dead V1 chain asks the sender to rebuild, not us to reset V2',
+      () async {
+        await talkOverV1ThenPublishV2();
+        await bob.run(() async {
+          final userData = await bob.api.getUserById(alice.realUserId);
+          expect(await processSignalUserData(userData!), isTrue);
+        });
+        expect(await versionOf(bob, alice), SignalVersion.v2);
+
+        await breakV1Chain(alice, bob);
+        final encrypted = (await alice.run(
+          () => signalEncryptMessage(bob.realUserId, textContent('stuck')),
+        ))!;
+        expect(encrypted.type, pb.Message_Type.CIPHERTEXT);
+
+        final ((content, errorType), logs) = await logsOf(
+          bob,
+          () => signalDecryptMessageV1(
+            alice.realUserId,
+            encrypted.ciphertext,
+            CiphertextMessage.whisperType,
+          ),
+        );
+
+        expect(content, isNull);
+        expect(
+          errorType,
+          pb.PlaintextContent_DecryptionErrorMessage_Type.PREKEY_UNKNOWN,
+        );
+        expect(logs, contains(contains('beyond the receiving window')));
+        expect(logs, isNot(contains(contains('re-sync the session'))));
+      },
+      timeout: const Timeout(Duration(minutes: 2)),
+    );
+
+    test(
+      'a V1 resync that moves us to V2 asks the sender to rebuild as well',
+      () async {
+        await talkOverV1ThenPublishV2(letBobSettle: true);
+        expect(await versionOf(bob, alice), SignalVersion.v1);
+
+        await breakV1Chain(alice, bob);
+        final encrypted = (await alice.run(
+          () => signalEncryptMessage(bob.realUserId, textContent('stuck')),
+        ))!;
+
+        final ((content, errorType), logs) = await logsOf(
+          bob,
+          () => signalDecryptMessageV1(
+            alice.realUserId,
+            encrypted.ciphertext,
+            CiphertextMessage.whisperType,
+          ),
+        );
+
+        expect(content, isNull);
+        expect(
+          errorType,
+          pb.PlaintextContent_DecryptionErrorMessage_Type.PREKEY_UNKNOWN,
+        );
+        expect(logs, contains(contains('re-sync the session')));
+        expect(await versionOf(bob, alice), SignalVersion.v2);
+      },
+      timeout: const Timeout(Duration(minutes: 2)),
+    );
+
+    test(
+      'a broken V1 session does not hold back V2 messages of the same batch',
+      () async {
+        await talkOverV1ThenPublishV2();
+        await bob.run(() async {
+          final userData = await bob.api.getUserById(alice.realUserId);
+          expect(await processSignalUserData(userData!), isTrue);
+        });
+        // Bob's first V2 message gives Alice a V2 session to answer with.
+        await bob.sendText(alice, 'hello over V2');
+        await alice.expectMessage((m) => m.content == 'hello over V2');
+        await breakV1Chain(alice, bob);
+
+        Future<NewMessage> fromAlice(SignalEncryptResult encrypted) async =>
+            NewMessage(
+              body: pb.Message(
+                receiptId: uuid.v4(),
+                type: encrypted.type,
+                encryptedContent: encrypted.ciphertext,
+              ).writeToBuffer(),
+              fromUserId: Int64(alice.realUserId),
+            );
+        Uint8List text(String text) => pb.EncryptedContent(
+          groupId: alice.defaultGroup!.groupId,
+          textMessage: pb.EncryptedContent_TextMessage()
+            ..senderMessageId = uuid.v4()
+            ..text = text,
+        ).writeToBuffer();
+
+        final overDeadV1Chain = await alice.run(
+          () async => fromAlice(
+            (await signalEncryptMessage(
+              bob.realUserId,
+              text('over the dead V1 chain'),
+            ))!,
+          ),
+        );
+        final overV2 = await alice.run(
+          () async => fromAlice(
+            (await signalEncryptMessageV2(
+              bob.realUserId,
+              text('over V2 in the same batch'),
+            ))!,
+          ),
+        );
+
+        final brokenSessions = <(int, SignalVersion)>{};
+        await bob.run(() async {
+          for (final message in [overDeadV1Chain, overV2]) {
+            await handleClient2ClientMessage(
+              message,
+              brokenSessionsInCurrentBatch: brokenSessions,
+            );
+          }
+        });
+
+        expect(brokenSessions, {(alice.realUserId, SignalVersion.v1)});
+        await bob.expectMessage(
+          (m) => m.content == 'over V2 in the same batch',
+        );
+      },
+      timeout: const Timeout(Duration(minutes: 2)),
+    );
+
+    test(
+      'a message sent over a dead V1 chain still arrives',
+      () async {
+        await talkOverV1ThenPublishV2();
+        await bob.run(() async {
+          final userData = await bob.api.getUserById(alice.realUserId);
+          expect(await processSignalUserData(userData!), isTrue);
+        });
+        await breakV1Chain(alice, bob);
+
+        await alice.sendText(bob, 'sent over the dead chain');
+
+        final received = await bob.expectMessage(
+          (m) => m.content == 'sent over the dead chain',
+        );
+        expect(received.senderId, alice.realUserId);
+      },
+      timeout: const Timeout(Duration(minutes: 2)),
+    );
+  });
+}

test/signal/pqc_migration_test.dart

@@ -10,6 +10,7 @@ import 'package:twonly/core/bridge.dart' as bridge;
 import 'package:twonly/core/bridge/wrapper/key_manager.dart';
 import 'package:twonly/core/bridge/wrapper/signal.dart';
 import 'package:twonly/core/frb_generated.dart';
+import 'package:twonly/core/signal/engine.dart';
 import 'package:twonly/globals.dart';
 import 'package:twonly/locator.dart';
 import 'package:twonly/src/callbacks/callbacks.dart';
@@ -19,11 +20,45 @@ import 'package:twonly/src/model/protobuf/api/websocket/server_to_client.pb.dart
     as api_pb;
 import 'package:twonly/src/model/protobuf/client/generated/messages.pb.dart'
     as msg_pb;
+import 'package:twonly/src/services/api/api.service.dart';
 import 'package:twonly/src/services/signal/encryption.signal.dart';
 import 'package:twonly/src/services/signal/identity.signal.dart';
 import 'package:twonly/src/services/signal/session.signal.dart';
 import 'package:twonly/src/utils/log.dart';
 
+class _MigrationApiService extends ApiService {
+  final Map<int, api_pb.Response_UserData> users = {};
+  final List<int> requestedUserIds = [];
+
+  @override
+  Future<api_pb.Response_UserData?> getUserById(int userId) async {
+    requestedUserIds.add(userId);
+    return users[userId];
+  }
+}
+
+api_pb.Response_UserData _userDataFromBundle(
+  int userId,
+  FrbPreKeyBundle bundle,
+) => api_pb.Response_UserData(
+  userId: Int64(userId),
+  username: utf8.encode('contact_$userId'),
+  registrationId: Int64(bundle.registrationId),
+  publicIdentityKey: bundle.identityKey,
+  pqcBundle: api_pb.Response_PqcBundle(
+    prekey: api_pb.Response_PqcPreKey(
+      eccPreKeyId: bundle.preKeyId != null ? Int64(bundle.preKeyId!) : null,
+      eccPreKey: bundle.preKeyPublic,
+      kyberPreKeyId: Int64(bundle.kyberPreKeyId),
+      kyberPreKey: bundle.kyberPreKeyPublic,
+      kyberPreKeySignature: bundle.kyberPreKeySignature,
+    ),
+    eccSignedPrekeyId: Int64(bundle.signedPreKeyId),
+    eccSignedPrekey: bundle.signedPreKeyPublic,
+    eccSignedPrekeySignature: bundle.signedPreKeySignature,
+  ),
+);
+
 void main() {
   if (!Platform.isMacOS) {
     return;
@@ -32,6 +67,7 @@ void main() {
   TestWidgetsFlutterBinding.ensureInitialized();
   late Directory tempDir;
   late TwonlyDB db;
+  late _MigrationApiService api;
 
   setUpAll(() async {
     Log.init();
@@ -59,10 +95,16 @@ void main() {
     );
 
     db = TwonlyDB(NativeDatabase.memory());
-    locator.registerFactory<TwonlyDB>(() => db);
+    api = _MigrationApiService();
+    locator
+      ..registerFactory<TwonlyDB>(() => db)
+      ..registerSingleton<ApiService>(api);
   });
 
   setUp(() async {
+    api
+      ..users.clear()
+      ..requestedUserIds.clear();
     await createIfNotExistsSignalIdentity();
     await RustKeyManager.setUserId(userId: 1);
   });
@@ -179,4 +221,39 @@ void main() {
     expect(encryptedResult!.type, msg_pb.Message_Type.CIPHERTEXT_V2);
     expect(encryptedResult.ciphertext, isNotEmpty);
   });
+
+  test('only active V1 sessions are upgraded', () async {
+    const activeContactIds = [701, 702];
+    const inactiveContactId = 703;
+
+    for (final contactId in [...activeContactIds, inactiveContactId]) {
+      await db.contactsDao.insertContact(
+        ContactsCompanion.insert(
+          userId: Value(contactId),
+          username: 'contact_$contactId',
+          accepted: const Value(true),
+          signalVersion: const Value(SignalVersion.v1),
+        ),
+      );
+      api.users[contactId] = _userDataFromBundle(
+        contactId,
+        await RustSignal.generateBundle(),
+      );
+    }
+
+    activeContactIds.forEach(scheduleSignalSessionUpgrade);
+    await upgradeActiveSignalSessionsToV2();
+
+    for (final contactId in activeContactIds) {
+      expect(
+        (await db.contactsDao.getContactById(contactId))!.signalVersion,
+        SignalVersion.v2,
+      );
+    }
+    expect(
+      (await db.contactsDao.getContactById(inactiveContactId))!.signalVersion,
+      SignalVersion.v1,
+    );
+    expect(api.requestedUserIds, unorderedEquals(activeContactIds));
+  });
 }

test/signal/protocol_state_test.dart

@@ -0,0 +1,19 @@
+import 'package:flutter_test/flutter_test.dart';
+import 'package:twonly/src/database/tables/contacts.table.dart';
+import 'package:twonly/src/services/signal/protocol_state.signal.dart';
+
+void main() {
+  test('V1 recovery backoff does not disable V2 recovery', () {
+    const contactId = 912345;
+
+    for (var i = 0; i < maxResyncAttempts; i++) {
+      recordResyncAttempt(contactId, SignalVersion.v1, success: false);
+    }
+
+    expect(shouldAttemptResync(contactId, SignalVersion.v1), isFalse);
+    expect(shouldAttemptResync(contactId, SignalVersion.v2), isTrue);
+
+    recordResyncAttempt(contactId, SignalVersion.v1, success: true);
+    recordResyncAttempt(contactId, SignalVersion.v2, success: true);
+  });
+}

test/signal/receiving_window_test.dart

@@ -0,0 +1,158 @@
+import 'dart:convert';
+import 'dart:typed_data';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:libsignal_protocol_dart/libsignal_protocol_dart.dart';
+import 'package:libsignal_protocol_dart/src/invalid_message_exception.dart';
+import 'package:twonly/src/services/signal/encryption.signal.dart';
+
+class _Party {
+  _Party(this.name)
+    : store = InMemorySignalProtocolStore(
+        generateIdentityKeyPair(),
+        generateRegistrationId(false),
+      );
+
+  final String name;
+  final InMemorySignalProtocolStore store;
+  var _nextKeyId = 1;
+
+  SignalProtocolAddress get address => SignalProtocolAddress(name, 1);
+
+  SessionCipher cipherFor(_Party other) =>
+      SessionCipher.fromStore(store, other.address);
+
+  Future<PreKeyBundle> newBundle() async {
+    final keyId = _nextKeyId++;
+    final preKey = generatePreKeys(keyId, 1).single;
+    await store.storePreKey(preKey.id, preKey);
+    final identity = await store.getIdentityKeyPair();
+    final signedPreKey = generateSignedPreKey(identity, keyId);
+    await store.storeSignedPreKey(signedPreKey.id, signedPreKey);
+    return PreKeyBundle(
+      await store.getLocalRegistrationId(),
+      1,
+      preKey.id,
+      preKey.getKeyPair().publicKey,
+      signedPreKey.id,
+      signedPreKey.getKeyPair().publicKey,
+      signedPreKey.signature,
+      identity.getPublicKey(),
+    );
+  }
+
+  /// Starts a new session with [other] and has [other] accept it.
+  Future<void> startSessionWith(_Party other) async {
+    await SessionBuilder.fromSignalStore(
+      store,
+      other.address,
+    ).processPreKeyBundle(await other.newBundle());
+    final first = await cipherFor(other).encrypt(_text('first'));
+    await other.cipherFor(this).decrypt(PreKeySignalMessage(first.serialize()));
+  }
+}
+
+Uint8List _text(String text) => Uint8List.fromList(utf8.encode(text));
+
+/// Alice encrypts [count] messages that never reach Bob.
+Future<void> _loseMessages(SessionCipher aliceCipher, int count) async {
+  for (var i = 0; i < count; i++) {
+    await aliceCipher.encrypt(_text('lost $i'));
+  }
+}
+
+Future<SignalMessage> _whisper(SessionCipher cipher, String text) async =>
+    SignalMessage.fromSerialized(
+      (await cipher.encrypt(_text(text))).serialize(),
+    );
+
+void main() {
+  late _Party alice;
+  late _Party bob;
+  late SessionCipher aliceCipher;
+  late SessionCipher bobCipher;
+
+  setUp(() async {
+    alice = _Party('alice');
+    bob = _Party('bob');
+    await alice.startSessionWith(bob);
+    aliceCipher = alice.cipherFor(bob);
+    bobCipher = bob.cipherFor(alice);
+    // Bob's reply makes Alice ratchet to a sending chain Bob has not seen.
+    final reply = await bobCipher.encrypt(_text('reply'));
+    await aliceCipher.decryptFromSignal(
+      SignalMessage.fromSerialized(reply.serialize()),
+    );
+  });
+
+  test('accepts a message on a working chain', () async {
+    final message = await _whisper(aliceCipher, 'hello');
+
+    final record = await bob.store.loadSession(alice.address);
+    expect(isBeyondReceivingWindow(record, message), isFalse);
+    expect(utf8.decode(await bobCipher.decryptFromSignal(message)), 'hello');
+  });
+
+  test('agrees with libsignal at the edge of the receiving window', () async {
+    await _loseMessages(aliceCipher, 2000);
+    final atEdge = await _whisper(aliceCipher, 'counter 2000');
+    final beyond = await _whisper(aliceCipher, 'counter 2001');
+    expect(atEdge.getCounter(), 2000);
+
+    final record = await bob.store.loadSession(alice.address);
+    expect(isBeyondReceivingWindow(record, atEdge), isFalse);
+    expect(isBeyondReceivingWindow(record, beyond), isTrue);
+
+    await expectLater(
+      bobCipher.decryptFromSignal(beyond),
+      throwsA(isA<InvalidMessageException>()),
+    );
+    expect(
+      utf8.decode(await bobCipher.decryptFromSignal(atEdge)),
+      'counter 2000',
+    );
+  });
+
+  test('checks archived session states as well', () async {
+    await _loseMessages(aliceCipher, 2000);
+    final delivered = await _whisper(aliceCipher, 'counter 2000');
+    final heldBack = await _whisper(aliceCipher, 'counter 2001');
+    await _loseMessages(aliceCipher, 2000);
+    final farAhead = await _whisper(aliceCipher, 'counter 4002');
+    expect(farAhead.getCounter(), 4002);
+
+    // Bob's chain for these messages now continues at 2001.
+    await bobCipher.decryptFromSignal(delivered);
+    // A new session from Alice turns that chain's state into an archived one.
+    await alice.startSessionWith(bob);
+
+    final record = await bob.store.loadSession(alice.address);
+    expect(record.previousSessionStates, hasLength(1));
+
+    // Too far ahead for a new chain of the current state, but next in line
+    // for the archived one.
+    expect(isBeyondReceivingWindow(record, heldBack), isFalse);
+    // Too far ahead for both.
+    expect(isBeyondReceivingWindow(record, farAhead), isTrue);
+
+    await expectLater(
+      bobCipher.decryptFromSignal(farAhead),
+      throwsA(isA<InvalidMessageException>()),
+    );
+    expect(
+      utf8.decode(await bobCipher.decryptFromSignal(heldBack)),
+      'counter 2001',
+    );
+  });
+
+  test('leaves a missing session to libsignal', () async {
+    final message = await _whisper(aliceCipher, 'hello');
+
+    // Without a session there is no state to judge, libsignal reports the
+    // missing session itself.
+    final record = await bob.store.loadSession(
+      const SignalProtocolAddress('carol', 1),
+    );
+    expect(isBeyondReceivingWindow(record, message), isFalse);
+  });
+}

test/utils/log_test.dart

@@ -0,0 +1,55 @@
+import 'dart:io';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:twonly/globals.dart';
+import 'package:twonly/src/utils/log.dart';
+
+void main() {
+  late Directory dir;
+
+  setUpAll(() {
+    dir = Directory.systemTemp.createTempSync('twonly_log_test_');
+    AppEnvironment.initTesting(
+      customCacheDir: dir.path,
+      customSupportDir: dir.path,
+    );
+    Log.init();
+  });
+
+  tearDownAll(() => dir.deleteSync(recursive: true));
+
+  Future<List<String>> linesContaining(List<String> needles) async {
+    for (var i = 0; i < 100; i++) {
+      final lines = (await loadLogFile()).split('\n');
+      final found = [
+        for (final needle in needles)
+          lines.where((line) => line.contains(needle)).firstOrNull,
+      ];
+      if (!found.contains(null)) return found.cast<String>();
+      await Future<void>.delayed(const Duration(milliseconds: 10));
+    }
+    fail('The log file never contained all of $needles');
+  }
+
+  test(
+    'tags lines forwarded from Rust instead of the receiving isolate',
+    () async {
+      Log.rust('libsignal_protocol::session_management: forwarded line');
+      Log.info('line from Dart');
+
+      final [rustLine, dartLine] = await linesContaining([
+        'forwarded line',
+        'line from Dart',
+      ]);
+
+      expect(
+        rustLine,
+        endsWith(
+          'FINE [r] [twonly] rust > '
+          'libsignal_protocol::session_management: forwarded line',
+        ),
+      );
+      expect(dartLine, contains('FINE [f] [twonly] '));
+    },
+  );
+}

test/visual/memories/memory_thumbnail_test.dart

@@ -0,0 +1,81 @@
+import 'dart:io';
+
+import 'package:flutter/material.dart';
+import 'package:flutter/services.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:twonly/globals.dart';
+import 'package:twonly/src/database/tables/mediafiles.table.dart';
+import 'package:twonly/src/database/twonly.db.dart';
+import 'package:twonly/src/model/memory_item.model.dart';
+import 'package:twonly/src/services/mediafiles/mediafile.service.dart';
+import 'package:twonly/src/visual/views/memories/components/memory_thumbnail.comp.dart';
+
+void main() {
+  late Directory supportDir;
+  var thumbnailRequests = 0;
+
+  setUp(() {
+    supportDir = Directory.systemTemp.createTempSync('twonly_thumbnail_test_');
+    AppEnvironment.initTesting(
+      customCacheDir: supportDir.path,
+      customSupportDir: supportDir.path,
+    );
+    thumbnailRequests = 0;
+    // Stands in for a video the platform cannot create a thumbnail for.
+    TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+        .setMockMethodCallHandler(const MethodChannel('pro_video_editor'), (
+          call,
+        ) async {
+          if (call.method != 'getThumbnails') return null;
+          thumbnailRequests++;
+          await Future<void>.delayed(const Duration(milliseconds: 50));
+          return <Uint8List>[];
+        });
+  });
+
+  tearDown(() {
+    TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+        .setMockMethodCallHandler(
+          const MethodChannel('pro_video_editor'),
+          null,
+        );
+    supportDir.deleteSync(recursive: true);
+  });
+
+  testWidgets('does not retry a video thumbnail that cannot be created', (
+    tester,
+  ) async {
+    final media = MediaFileService(
+      MediaFile(
+        mediaId: 'video-without-thumbnail',
+        type: MediaType.video,
+        cloudState: CloudState.none,
+        requiresAuthentication: false,
+        stored: true,
+        isDraftMedia: false,
+        isFavorite: false,
+        hasCropAnalyzed: false,
+        hasThumbnail: false,
+        createdAt: DateTime(2026, 9, 24),
+      ),
+    );
+    media.storedPath.writeAsBytesSync([1, 2, 3]);
+
+    await tester.pumpWidget(
+      MaterialApp(
+        home: SizedBox(
+          width: 120,
+          height: 200,
+          child: MemoriesThumbnailComp(
+            galleryItem: MemoryItem(mediaService: media, messages: []),
+            onTap: () {},
+          ),
+        ),
+      ),
+    );
+    await tester.pump(const Duration(seconds: 2));
+
+    expect(thumbnailRequests, 1);
+    expect(media.thumbnailPath.existsSync(), isFalse);
+  });
+}