fix session out of sync performance issues

lib/src/database/daos/messages.dao.dart

@@ -516,6 +516,15 @@ class MessagesDao extends DatabaseAccessor<TwonlyDB> with _$MessagesDaoMixin {
         .getSingleOrNull();
   }
 
+  Stream<MessageAction?> watchLastMessageAction(String messageId) {
+    return (((select(messageActions)..where(
+              (t) => t.messageId.equals(messageId),
+            ))
+            ..orderBy([(t) => OrderingTerm.desc(t.actionAt)]))
+          ..limit(1))
+        .watchSingleOrNull();
+  }
+
   Future<void> deleteMessagesById(String messageId) {
     return (delete(messages)..where((t) => t.messageId.equals(messageId))).go();
   }

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

@@ -56,9 +56,13 @@ Future<void> handleServerMessage(server.ServerToClient msg) async {
       Log.info(
         'Got ${msg.v0.newMessages.newMessages.length} messages from the server.',
       );
+      final brokenSessionsInCurrentBatch = <int>{};
       for (final newMessage in msg.v0.newMessages.newMessages) {
         try {
-          await handleClient2ClientMessage(newMessage);
+          await handleClient2ClientMessage(
+            newMessage,
+            brokenSessionsInCurrentBatch: brokenSessionsInCurrentBatch,
+          );
         } catch (e) {
           Log.error(e);
         }
@@ -83,7 +87,10 @@ DateTime lastPushKeyRequest = clock.now().subtract(const Duration(hours: 1));
 
 final Map<String, Mutex> _messageLocks = {};
 
-Future<void> handleClient2ClientMessage(NewMessage newMessage) async {
+Future<void> handleClient2ClientMessage(
+  NewMessage newMessage, {
+  Set<int>? brokenSessionsInCurrentBatch,
+}) async {
   final body = Uint8List.fromList(newMessage.body);
   final message = Message.fromBuffer(body);
   final receiptId = message.receiptId;
@@ -97,7 +104,11 @@ Future<void> handleClient2ClientMessage(NewMessage newMessage) async {
   }
   await mutex.protect(() async {
     try {
-      await _handleClient2ClientMessage(newMessage, message);
+      await _handleClient2ClientMessage(
+        newMessage,
+        message,
+        brokenSessionsInCurrentBatch: brokenSessionsInCurrentBatch,
+      );
     } finally {
       _messageLocks.remove(receiptId);
     }
@@ -106,11 +117,27 @@ Future<void> handleClient2ClientMessage(NewMessage newMessage) async {
 
 Future<void> _handleClient2ClientMessage(
   NewMessage newMessage,
-  Message message,
-) async {
+  Message message, {
+  Set<int>? brokenSessionsInCurrentBatch,
+}) async {
   final fromUserId = newMessage.fromUserId.toInt();
   final receiptId = message.receiptId;
 
+  if (brokenSessionsInCurrentBatch?.contains(fromUserId) == 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.
+    // By returning early, we skip gotReceipt() and error responses.
+    // The server still deletes the batch since we ACK the entire batch later.
+    // The sender keeps the message unacknowledged. Once they process our SESSION_OUT_OF_SYNC
+    // error and establish a new session, their retry logic will automatically re-encrypt
+    // and re-send these messages with the new keys.
+    Log.info(
+      'Skipping message from $fromUserId - session known broken in this batch',
+    );
+    return;
+  }
+
   if (await twonlyDB.receiptsDao.isDuplicated(receiptId)) {
     return;
   }
@@ -200,6 +227,7 @@ Future<void> _handleClient2ClientMessage(
             encryptedContentRaw,
             message.type,
             receiptId,
+            brokenSessionsInCurrentBatch: brokenSessionsInCurrentBatch,
           );
           if (plainTextContent != null) {
             response = Message(
@@ -265,13 +293,15 @@ Future<(EncryptedContent?, PlaintextContent?)> handleEncryptedMessageRaw(
   int fromUserId,
   Uint8List encryptedContentRaw,
   Message_Type messageType,
-  String receiptId,
-) async {
+  String receiptId, {
+  Set<int>? brokenSessionsInCurrentBatch,
+}) async {
   Log.info('[$receiptId] calling signalDecryptMessage');
   var (encryptedContent, decryptionErrorType) = await signalDecryptMessage(
     fromUserId,
     encryptedContentRaw,
     messageType.value,
+    brokenSessionsInCurrentBatch: brokenSessionsInCurrentBatch,
   );
 
   if (encryptedContent == null) {

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

@@ -38,8 +38,9 @@ Future<(EncryptedContent?, PlaintextContent_DecryptionErrorMessage_Type?)>
 signalDecryptMessage(
   int fromUserId,
   Uint8List encryptedContentRaw,
-  int type,
-) async {
+  int type, {
+  Set<int>? brokenSessionsInCurrentBatch,
+}) async {
   // Hold the lock only for the cryptographic operation, not for network I/O
   Log.info('Acquiring lockingSignalProtocol for $fromUserId');
   final (
@@ -74,6 +75,7 @@ signalDecryptMessage(
           );
       }
 
+      recordResyncAttempt(fromUserId, success: true);
       return (EncryptedContent.fromBuffer(plaintext), null, false);
     } on InvalidKeyIdException catch (e) {
       Log.warn(e);
@@ -113,22 +115,25 @@ signalDecryptMessage(
 
   // Handle session resync OUTSIDE the lock to avoid holding it during
   // network round-trips (which can block for up to 60 seconds)
-  if (needsResync && !resyncedUsers.contains(fromUserId)) {
-    if (await handleSessionResync(fromUserId)) {
-      // This flag prevents from resyncing the session the client received
-      // multiple new messages from the server he could not decrypt
-      resyncedUsers.add(fromUserId);
+  if (needsResync) {
+    brokenSessionsInCurrentBatch?.add(fromUserId);
+    if (shouldAttemptResync(fromUserId)) {
+      if (await handleSessionResync(fromUserId)) {
+        // This flag prevents from resyncing the session the client received
+        // multiple new messages from the server he could not decrypt
+        recordResyncAttempt(fromUserId, success: false);
 
-      // This message contains a new PreKeyBundle establishing a new signal
-      // session
-      await sendCipherText(
-        fromUserId,
-        EncryptedContent(
-          errorMessages: EncryptedContent_ErrorMessages(
-            type: EncryptedContent_ErrorMessages_Type.SESSION_OUT_OF_SYNC,
+        // This message contains a new PreKeyBundle establishing a new signal
+        // session
+        await sendCipherText(
+          fromUserId,
+          EncryptedContent(
+            errorMessages: EncryptedContent_ErrorMessages(
+              type: EncryptedContent_ErrorMessages_Type.SESSION_OUT_OF_SYNC,
+            ),
           ),
-        ),
-      );
+        );
+      }
     }
   }
 

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

@@ -1,12 +1,36 @@
+import 'dart:math';
 import 'package:mutex/mutex.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.
-final resyncedUsers = <int>{};
+final Map<int, ({int failureCount, DateTime lastAttempt})> _resyncAttempts = {};
 
-/// Reset the resync tracking set.
-void resetResyncedUsers() {
-  resyncedUsers.clear();
+const int maxResyncAttempts = 3;
+
+bool shouldAttemptResync(int userId) {
+  final attempt = _resyncAttempts[userId];
+  if (attempt == null) return true;
+  if (attempt.failureCount >= maxResyncAttempts) return false;
+
+  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}) {
+  if (success) {
+    _resyncAttempts.remove(userId);
+  } else {
+    final current = _resyncAttempts[userId];
+    _resyncAttempts[userId] = (
+      failureCount: (current?.failureCount ?? 0) + 1,
+      lastAttempt: DateTime.now(),
+    );
+  }
+}
+
+/// Reset the resync tracking set (currently unused, backoff handles expiry naturally).
+void resetResyncedUsers() {
+  // No-op. We want the backoff state to persist across reconnects.
 }