feat: implement retry encryption key request mechanism for livekit calls

coder-with-a-bushido · coder-with-a-bushido · commit 8a528b9d0bd3 · 2026-01-05T22:43:49.000+05:30
diff --git a/lib/src/voip/backend/livekit_backend.dart b/lib/src/voip/backend/livekit_backend.dart
@@ -24,6 +24,12 @@ class LiveKitBackend extends CallBackend {
   /// participant:keyIndex:keyBin
   final Map<CallParticipant, Map<int, Uint8List>> _encryptionKeysMap = {};
 
+  /// Tracks pending encryption key request retries per participant
+  final Map<CallParticipant, Timer> _requestEncryptionKeyPending = {};
+
+  /// Retry interval for key requests
+  static const Duration _keyRequestRetryInterval = Duration(seconds: 2);
+
   final List<Future> _setNewKeyTimeouts = [];
 
   int _indexCounter = 0;
@@ -79,10 +85,7 @@ class LiveKitBackend extends CallBackend {
         '_makeNewSenderKey using previous key because last created at ${_lastNewKeyTime.toString()}',
       );
       // still a fairly new key, just send that
-      await _sendEncryptionKeysEvent(
-        groupCall,
-        _latestLocalKeyIndex,
-      );
+      await _sendEncryptionKeysEvent(groupCall, _latestLocalKeyIndex);
       return;
     }
 
@@ -222,20 +225,22 @@ class LiveKitBackend extends CallBackend {
       );
       // now wait for the key to propogate and then set it, hopefully users can
       // stil decrypt everything
-      final useKeyTimeout =
-          Future.delayed(groupCall.voip.timeouts!.useKeyDelay, () async {
-        Logs().i(
-          '[VOIP E2EE] delayed setting key changed event for ${participant.id} idx $encryptionKeyIndex key $encryptionKeyBin',
-        );
-        await groupCall.voip.delegate.keyProvider?.onSetEncryptionKey(
-          participant,
-          encryptionKeyBin,
-          encryptionKeyIndex,
-        );
-        if (participant.isLocal) {
-          _currentLocalKeyIndex = encryptionKeyIndex;
-        }
-      });
+      final useKeyTimeout = Future.delayed(
+        groupCall.voip.timeouts!.useKeyDelay,
+        () async {
+          Logs().i(
+            '[VOIP E2EE] delayed setting key changed event for ${participant.id} idx $encryptionKeyIndex key $encryptionKeyBin',
+          );
+          await groupCall.voip.delegate.keyProvider?.onSetEncryptionKey(
+            participant,
+            encryptionKeyBin,
+            encryptionKeyIndex,
+          );
+          if (participant.isLocal) {
+            _currentLocalKeyIndex = encryptionKeyIndex;
+          }
+        },
+      );
       _setNewKeyTimeouts.add(useKeyTimeout);
     } else {
       Logs().i(
@@ -271,17 +276,15 @@ class LiveKitBackend extends CallBackend {
         '[VOIP E2EE] _sendEncryptionKeysEvent Tried to send encryption keys event but no keys found!',
       );
       await _makeNewSenderKey(groupCall, false);
-      await _sendEncryptionKeysEvent(
-        groupCall,
-        keyIndex,
-        sendTo: sendTo,
-      );
+      await _sendEncryptionKeysEvent(groupCall, keyIndex, sendTo: sendTo);
       return;
     }
 
     try {
       final keyContent = EncryptionKeysEventContent(
-        [EncryptionKeyEntry(keyIndex, base64Encode(myLatestKey))],
+        [
+          EncryptionKeyEntry(keyIndex, base64Encode(myLatestKey)),
+        ],
         groupCall.groupCallId,
       );
       final Map<String, Object> data = {
@@ -300,11 +303,7 @@ class LiveKitBackend extends CallBackend {
       );
     } catch (e, s) {
       Logs().e('[VOIP E2EE] Failed to send e2ee keys, retrying', e, s);
-      await _sendEncryptionKeysEvent(
-        groupCall,
-        keyIndex,
-        sendTo: sendTo,
-      );
+      await _sendEncryptionKeysEvent(groupCall, keyIndex, sendTo: sendTo);
     }
   }
 
@@ -375,6 +374,10 @@ class LiveKitBackend extends CallBackend {
     GroupCallSession groupCall,
     List<CallParticipant> remoteParticipants,
   ) async {
+    Logs().v(
+      '[VOIP E2EE] requesting stream encryption keys from ${remoteParticipants.map((e) => e.id)}',
+    );
+
     final Map<String, Object> data = {
       'conf_id': groupCall.groupCallId,
       'device_id': groupCall.client.deviceID!,
@@ -387,6 +390,29 @@ class LiveKitBackend extends CallBackend {
       data,
       EventTypes.GroupCallMemberEncryptionKeysRequest,
     );
+
+    // Set up retry timers for each participant
+    for (final rp in remoteParticipants) {
+      // Skip if a retry is already pending for this participant
+      if (_requestEncryptionKeyPending.containsKey(rp)) continue;
+
+      var retryCount = 0;
+      _requestEncryptionKeyPending[rp] = Timer.periodic(
+        _keyRequestRetryInterval,
+        (timer) {
+          retryCount++;
+          if (retryCount >= 5) {
+            Logs().w(
+              '[VOIP E2EE] Max retries (5) reached for ${rp.id}, giving up key request',
+            );
+            timer.cancel();
+            _requestEncryptionKeyPending.remove(rp);
+            return;
+          }
+          unawaited(requestEncrytionKey(groupCall, [rp]));
+        },
+      );
+    }
   }
 
   @override
@@ -403,8 +429,14 @@ class LiveKitBackend extends CallBackend {
     final keyContent = EncryptionKeysEventContent.fromJson(content);
 
     final callId = keyContent.callId;
-    final p =
-        CallParticipant(groupCall.voip, userId: userId, deviceId: deviceId);
+    final p = CallParticipant(
+      groupCall.voip,
+      userId: userId,
+      deviceId: deviceId,
+    );
+
+    // Cancel any pending retry for this participant since we received keys
+    _requestEncryptionKeyPending.remove(p)?.cancel();
 
     if (keyContent.keys.isEmpty) {
       Logs().w(
@@ -471,11 +503,7 @@ class LiveKitBackend extends CallBackend {
           groupCall,
           _latestLocalKeyIndex,
           sendTo: [
-            CallParticipant(
-              groupCall.voip,
-              userId: userId,
-              deviceId: deviceId,
-            ),
+            CallParticipant(groupCall.voip, userId: userId, deviceId: deviceId),
           ],
         );
         return true;
@@ -525,16 +553,14 @@ class LiveKitBackend extends CallBackend {
     if (_memberLeaveEncKeyRotateDebounceTimer != null) {
       _memberLeaveEncKeyRotateDebounceTimer!.cancel();
     }
-    _memberLeaveEncKeyRotateDebounceTimer =
-        Timer(groupCall.voip.timeouts!.makeKeyOnLeaveDelay, () async {
-      // we skipJoinDebounce here because we want to make sure a new key is generated
-      // and that the join debounce does not block us from making a new key
-      await _makeNewSenderKey(
-        groupCall,
-        true,
-        skipJoinDebounce: true,
-      );
-    });
+    _memberLeaveEncKeyRotateDebounceTimer = Timer(
+      groupCall.voip.timeouts!.makeKeyOnLeaveDelay,
+      () async {
+        // we skipJoinDebounce here because we want to make sure a new key is generated
+        // and that the join debounce does not block us from making a new key
+        await _makeNewSenderKey(groupCall, true, skipJoinDebounce: true);
+      },
+    );
   }
 
   @override
@@ -545,6 +571,12 @@ class LiveKitBackend extends CallBackend {
     _currentLocalKeyIndex = 0;
     _latestLocalKeyIndex = 0;
     _memberLeaveEncKeyRotateDebounceTimer?.cancel();
+
+    // Clean up all pending encryption key request retries
+    for (final timer in _requestEncryptionKeyPending.values) {
+      timer.cancel();
+    }
+    _requestEncryptionKeyPending.clear();
   }
 
   @override
diff --git a/test/livekit_backend_test.dart b/test/livekit_backend_test.dart
@@ -0,0 +1,193 @@
+import 'dart:convert';
+
+import 'package:test/test.dart';
+
+import 'package:matrix/matrix.dart';
+import 'fake_client.dart';
+import 'webrtc_stub.dart';
+
+void main() {
+  late Client matrix;
+  late Room room;
+  late VoIP voip;
+  late LiveKitBackend backend;
+
+  group('LiveKitBackend encryption key request retry tests', () {
+    Logs().level = Level.info;
+
+    setUp(() async {
+      matrix = await getClient();
+      await matrix.abortSync();
+
+      voip = VoIP(matrix, MockWebRTCDelegate());
+      VoIP.customTxid = '1234';
+      final id = '!calls:example.com';
+      room = matrix.getRoomById(id)!;
+
+      backend = LiveKitBackend(
+        livekitServiceUrl: 'https://livekit.example.com',
+        livekitAlias: 'test_alias',
+      );
+
+      // Clear logs before each test to avoid interference
+      Logs().outputEvents.clear();
+    });
+
+    /// Helper to create a group call session for testing
+    Future<GroupCallSession> createGroupCall(String callId) async {
+      final membership = CallMembership(
+        userId: matrix.userID!,
+        callId: callId,
+        backend: backend,
+        deviceId: matrix.deviceID!,
+        expiresTs:
+            DateTime.now().add(Duration(hours: 1)).millisecondsSinceEpoch,
+        roomId: room.id,
+        membershipId: voip.currentSessionId,
+        voip: voip,
+        eventId: 'membership_event_$callId',
+      );
+
+      room.setState(
+        Event(
+          content: {
+            'memberships': [membership.toJson()],
+          },
+          type: EventTypes.GroupCallMember,
+          eventId: 'membership_event_$callId',
+          senderId: matrix.userID!,
+          originServerTs: DateTime.now(),
+          room: room,
+          stateKey: matrix.userID!,
+        ),
+      );
+
+      await voip.createGroupCallFromRoomStateEvent(membership);
+      final groupCall = voip.getGroupCallById(room.id, callId);
+      await groupCall!.enter();
+      return groupCall;
+    }
+
+    /// Helper to count key request log messages for a specific participant
+    int countKeyRequestsFor(String participantId) {
+      return Logs()
+          .outputEvents
+          .where(
+            (event) =>
+                event.title
+                    .contains('requesting stream encryption keys from') &&
+                event.title.contains(participantId),
+          )
+          .length;
+    }
+
+    test(
+      'retry mechanism automatically re-requests keys when initial request fails',
+      () async {
+        // This test verifies: without retry, a failed key request leaves the call
+        // in an unrecoverable state. With retry, requests are automatically retried.
+
+        final groupCall = await createGroupCall('test_retry_mechanism');
+
+        const remoteUserId = '@retry_test_user:example.com';
+        const remoteDeviceId = 'RETRY_TEST_DEVICE';
+        final remoteParticipant = CallParticipant(
+          voip,
+          userId: remoteUserId,
+          deviceId: remoteDeviceId,
+        );
+
+        Logs().outputEvents.clear();
+        Logs().level = Level.verbose;
+
+        // Step 1: Initial key request (simulates framecryptor detecting missingKey)
+        await backend.requestEncrytionKey(groupCall, [remoteParticipant]);
+        expect(countKeyRequestsFor(remoteUserId), 1);
+
+        // Step 2: Wait for retry timer (2 second interval)
+        // WITHOUT retry: count stays at 1 (STUCK!)
+        // WITH retry: count increases (RECOVERY!)
+        await Future.delayed(Duration(milliseconds: 2100));
+
+        expect(
+          countKeyRequestsFor(remoteUserId),
+          greaterThan(1),
+          reason: 'Retry mechanism should automatically re-request keys. '
+              'Without retry, the call would be stuck in an unrecoverable state.',
+        );
+
+        await backend.dispose(groupCall);
+      },
+    );
+
+    test(
+      'each participant has independent retry - receiving keys for one does not affect another',
+      () async {
+        final groupCall = await createGroupCall('test_independent_retries');
+
+        const user1 = '@independent_user1:example.com';
+        const device1 = 'DEVICE_1';
+        const user2 = '@independent_user2:example.com';
+        const device2 = 'DEVICE_2';
+
+        final participant1 =
+            CallParticipant(voip, userId: user1, deviceId: device1);
+        final participant2 =
+            CallParticipant(voip, userId: user2, deviceId: device2);
+
+        Logs().outputEvents.clear();
+        Logs().level = Level.verbose;
+
+        // Request keys from both participants
+        await backend.requestEncrytionKey(groupCall, [participant1]);
+        await backend.requestEncrytionKey(groupCall, [participant2]);
+
+        expect(countKeyRequestsFor(user1), 1);
+        expect(countKeyRequestsFor(user2), 1);
+
+        // Wait for retry
+        await Future.delayed(Duration(milliseconds: 2100));
+        expect(countKeyRequestsFor(user1), greaterThan(1));
+        expect(countKeyRequestsFor(user2), greaterThan(1));
+
+        // Receive keys ONLY from participant 1
+        await backend.onCallEncryption(
+          groupCall,
+          user1,
+          device1,
+          {
+            'keys': [
+              {
+                'key': base64Encode([1, 2, 3, 4, 5, 6, 7, 8]),
+                'index': 0,
+              },
+            ],
+            'call_id': 'test_independent_retries',
+          },
+        );
+
+        final countUser1AfterKeys = countKeyRequestsFor(user1);
+        final countUser2AfterKeys = countKeyRequestsFor(user2);
+
+        // Wait another retry interval
+        await Future.delayed(Duration(milliseconds: 2100));
+
+        // User 1's retry should have stopped (received keys)
+        expect(
+          countKeyRequestsFor(user1),
+          countUser1AfterKeys,
+          reason: 'User 1 retry should stop after receiving keys.',
+        );
+
+        // User 2's retry should continue (no keys received)
+        expect(
+          countKeyRequestsFor(user2),
+          greaterThan(countUser2AfterKeys),
+          reason: 'User 2 retry should continue since no keys were received.',
+        );
+
+        await backend.dispose(groupCall);
+      },
+    );
+  });
+}
