fix: skip stale groups in MLS conversation recovery instead of aborting batch [WPB-23438] (#3852)

* fix: skip stale groups in MLS conversation recovery instead of aborting batch

When isLocalGroupEpochStale fails with ConversationNotFound (stale
epoch group ID not in local CoreCrypto state), the recovery manager
would propagate the error and abort the entire batch via
foldToEitherWhileRight. This left needsToRecoverMLSGroups=true,
causing the recovery to retry on every sync state change — producing
repeated ConversationNotFound log noise.

Now ConversationNotFound is treated as non-fatal: the group is skipped
and recovery continues with the remaining conversations.

* style: fix detekt argument wrapping in MLS recovery

Author: mastaab

logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/RecoverMLSConversationsUseCase.kt

@@ -19,6 +19,7 @@
 package com.wire.kalium.logic.feature.conversation
 
 import com.wire.kalium.common.error.CoreFailure
+import com.wire.kalium.common.error.MLSFailure
 import com.wire.kalium.logic.data.client.ClientRepository
 import com.wire.kalium.logic.data.conversation.Conversation
 import com.wire.kalium.logic.data.conversation.Conversation.ProtocolInfo.MLSCapable.GroupState
@@ -86,7 +87,19 @@ internal class RecoverMLSConversationsUseCaseImpl(
                 mlsConversationRepository.isLocalGroupEpochStale(mlsContext, protocol.groupId, protocol.epoch)
             }
                 .fold({ checkEpochFailure ->
-                    Either.Left(checkEpochFailure)
+                    if (checkEpochFailure is MLSFailure.ConversationNotFound) {
+                        kaliumLogger.w(
+                            "Marking ${protocol.groupId.toLogString()} as PENDING_AFTER_RESET: " +
+                                "group not found in local MLS state, will retry join on next sync"
+                        )
+                        conversationRepository.updateConversationGroupState(
+                            protocol.groupId,
+                            GroupState.PENDING_AFTER_RESET
+                        )
+                        Either.Right(Unit)
+                    } else {
+                        Either.Left(checkEpochFailure)
+                    }
                 }, { isGroupOutOfSync ->
                     if (isGroupOutOfSync) {
                         joinExistingMLSConversationUseCase(transactionContext, conversation.id).onFailure { joinFailure ->

logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/RecoverMLSConversationsUseCaseTests.kt

@@ -18,7 +18,9 @@
 
 package com.wire.kalium.logic.feature.conversation
 
+import com.wire.kalium.common.error.MLSFailure
 import com.wire.kalium.common.error.StorageFailure
+import com.wire.kalium.logic.data.conversation.Conversation.ProtocolInfo.MLSCapable.GroupState
 import com.wire.kalium.logic.data.client.ClientRepository
 import com.wire.kalium.logic.data.conversation.Conversation
 import com.wire.kalium.logic.data.conversation.ConversationRepository
@@ -143,6 +145,72 @@ class RecoverMLSConversationsUseCaseTests {
         assertIs<RecoverMLSConversationsResult.Success>(actual)
     }
 
+    @Test
+    fun givenEpochCheckFailsWithConversationNotFound_ThenGroupIsMarkedPendingAndOtherGroupsProcessed() = runTest {
+        val conversations = listOf(Arrangement.MLS_CONVERSATION1, Arrangement.MLS_CONVERSATION2)
+        val (arrangement, recoverMLSConversationsUseCase) = Arrangement()
+            .withConversationsByGroupStateReturns(Either.Right(conversations))
+            .withJoinExistingMLSConversationUseCaseSuccessful()
+            .withIsMLSSupported(true)
+            .withHasRegisteredMLSClient(true)
+            .withUpdateConversationGroupStateSuccessful()
+            .withEpochCheckFailsWithConversationNotFoundFor(Arrangement.GROUP_ID1)
+            .withConversationIsOutOfSyncReturnsTrueFor(listOf(Arrangement.GROUP_ID2))
+            .arrange()
+
+        val actual = recoverMLSConversationsUseCase(arrangement.transactionContext)
+
+        // Both groups should have epoch check attempted
+        coVerify {
+            arrangement.mlsConversationRepository.isLocalGroupEpochStale(any(), any(), any())
+        }.wasInvoked(conversations.size)
+
+        // First group should be marked as PENDING_AFTER_RESET so it can be re-joined on next sync
+        coVerify {
+            arrangement.conversationRepository.updateConversationGroupState(
+                eq(Arrangement.GROUP_ID1), eq(GroupState.PENDING_AFTER_RESET)
+            )
+        }.wasInvoked(once)
+
+        // Only the second group should be joined (first was marked pending due to ConversationNotFound)
+        coVerify {
+            arrangement.joinExistingMLSConversationUseCase.invoke(any(), eq(Arrangement.MLS_CONVERSATION2.id), any())
+        }.wasInvoked(once)
+
+        // Overall result should be Success since ConversationNotFound is non-fatal
+        assertIs<RecoverMLSConversationsResult.Success>(actual)
+    }
+
+    @Test
+    fun givenAllGroupsFailWithConversationNotFound_ThenAllMarkedPendingAndResultIsSuccess() = runTest {
+        val conversations = listOf(Arrangement.MLS_CONVERSATION1, Arrangement.MLS_CONVERSATION2)
+        val (arrangement, recoverMLSConversationsUseCase) = Arrangement()
+            .withConversationsByGroupStateReturns(Either.Right(conversations))
+            .withJoinExistingMLSConversationUseCaseSuccessful()
+            .withIsMLSSupported(true)
+            .withHasRegisteredMLSClient(true)
+            .withUpdateConversationGroupStateSuccessful()
+            .withEpochCheckFailsWithConversationNotFoundFor(Arrangement.GROUP_ID1)
+            .withEpochCheckFailsWithConversationNotFoundFor(Arrangement.GROUP_ID2)
+            .arrange()
+
+        val actual = recoverMLSConversationsUseCase(arrangement.transactionContext)
+
+        // Both groups should be marked as PENDING_AFTER_RESET
+        coVerify {
+            arrangement.conversationRepository.updateConversationGroupState(
+                any(), eq(GroupState.PENDING_AFTER_RESET)
+            )
+        }.wasInvoked(conversations.size)
+
+        // No groups should be joined since all were marked pending
+        coVerify {
+            arrangement.joinExistingMLSConversationUseCase.invoke(any(), any(), any())
+        }.wasNotInvoked()
+
+        assertIs<RecoverMLSConversationsResult.Success>(actual)
+    }
+
     @Test
     fun whenFetchingListOfConversationsFails_ThenShouldReturnFailure() = runTest {
         val (arrangement, recoverMLSConversationsUseCase) = Arrangement()
@@ -216,6 +284,18 @@ class RecoverMLSConversationsUseCaseTests {
             }.returns(Either.Right(true))
         }
 
+        suspend fun withUpdateConversationGroupStateSuccessful() = apply {
+            coEvery {
+                conversationRepository.updateConversationGroupState(any(), any())
+            }.returns(Either.Right(Unit))
+        }
+
+        suspend fun withEpochCheckFailsWithConversationNotFoundFor(groupID: GroupID) = apply {
+            coEvery {
+                mlsConversationRepository.isLocalGroupEpochStale(any(), eq(groupID), any())
+            }.returns(Either.Left(MLSFailure.ConversationNotFound))
+        }
+
         suspend fun withJoinExistingMLSConversationUseCaseFailsFor(failedGroupId: ConversationId) = apply {
             coEvery {
                 joinExistingMLSConversationUseCase.invoke(any(), eq(failedGroupId), any())