From 2cee00aff66c9b51de0505ca4bf6153ae0e34b4a Mon Sep 17 00:00:00 2001 From: otsmr Date: Thu, 9 Jul 2026 16:35:34 +0200 Subject: [PATCH] fix session out of sync performance issues --- lib/src/database/daos/messages.dao.dart | 9 ++++ lib/src/services/api/server_messages.api.dart | 44 ++++++++++++++++--- .../services/signal/encryption.signal.dart | 37 +++++++++------- .../signal/protocol_state.signal.dart | 32 ++++++++++++-- 4 files changed, 95 insertions(+), 27 deletions(-) diff --git a/lib/src/database/daos/messages.dao.dart b/lib/src/database/daos/messages.dao.dart index c35ac882..53d717b5 100644 --- a/lib/src/database/daos/messages.dao.dart +++ b/lib/src/database/daos/messages.dao.dart @@ -516,6 +516,15 @@ class MessagesDao extends DatabaseAccessor with _$MessagesDaoMixin { .getSingleOrNull(); } + Stream watchLastMessageAction(String messageId) { + return (((select(messageActions)..where( + (t) => t.messageId.equals(messageId), + )) + ..orderBy([(t) => OrderingTerm.desc(t.actionAt)])) + ..limit(1)) + .watchSingleOrNull(); + } + Future deleteMessagesById(String messageId) { return (delete(messages)..where((t) => t.messageId.equals(messageId))).go(); } diff --git a/lib/src/services/api/server_messages.api.dart b/lib/src/services/api/server_messages.api.dart index 33293b85..88bfd3e0 100644 --- a/lib/src/services/api/server_messages.api.dart +++ b/lib/src/services/api/server_messages.api.dart @@ -56,9 +56,13 @@ Future handleServerMessage(server.ServerToClient msg) async { Log.info( 'Got ${msg.v0.newMessages.newMessages.length} messages from the server.', ); + final brokenSessionsInCurrentBatch = {}; 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 _messageLocks = {}; -Future handleClient2ClientMessage(NewMessage newMessage) async { +Future handleClient2ClientMessage( + NewMessage newMessage, { + Set? brokenSessionsInCurrentBatch, +}) async { final body = Uint8List.fromList(newMessage.body); final message = Message.fromBuffer(body); final receiptId = message.receiptId; @@ -97,7 +104,11 @@ Future 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 handleClient2ClientMessage(NewMessage newMessage) async { Future _handleClient2ClientMessage( NewMessage newMessage, - Message message, -) async { + Message message, { + Set? 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 _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? brokenSessionsInCurrentBatch, +}) async { Log.info('[$receiptId] calling signalDecryptMessage'); var (encryptedContent, decryptionErrorType) = await signalDecryptMessage( fromUserId, encryptedContentRaw, messageType.value, + brokenSessionsInCurrentBatch: brokenSessionsInCurrentBatch, ); if (encryptedContent == null) { diff --git a/lib/src/services/signal/encryption.signal.dart b/lib/src/services/signal/encryption.signal.dart index f13c9960..5c29910e 100644 --- a/lib/src/services/signal/encryption.signal.dart +++ b/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? 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, + ), ), - ), - ); + ); + } } } diff --git a/lib/src/services/signal/protocol_state.signal.dart b/lib/src/services/signal/protocol_state.signal.dart index 9d797473..58f830e8 100644 --- a/lib/src/services/signal/protocol_state.signal.dart +++ b/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 = {}; +final Map _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. }