fix: commit pending proposals recovery strategy - WPB-23159

WireDomain/Sources/WireDomain/Components/ClientSessionComponent.swift

@@ -772,6 +772,19 @@ public final class ClientSessionComponent {
         )
     }
 
+    public func createInitiateResetMLSConversationUseCase() -> some InitiateResetMLSConversationUseCaseProtocol {
+        InitiateResetMLSConversationUseCase(
+            api: mlsAPI,
+            mlsService: mlsService,
+            conversationLocalStore: conversationLocalStore,
+            conversationRepository: conversationRepository,
+            lockRepository: ResetMLSConversationLockRepository(
+                userID: selfUserID
+            ),
+            selfDomain: backendMetadata.domain
+        )
+    }
+
     // MARK: - Other
 
     public private(set) lazy var conversationProtobufMessageProcessor = ConversationProtobufMessageProcessor(

WireDomain/Sources/WireDomain/Repositories/Conversations/LocalStore/ConversationLocalStore.swift

@@ -116,6 +116,7 @@ public final class ConversationLocalStore: ConversationLocalStoreProtocol {
             conversation.mlsStatus = .ready
             conversation.epoch = epoch
             conversation.mlsGroupID = mlsGroupID
+            conversation.commitPendingProposalDate = nil
         }
     }
 
@@ -126,6 +127,7 @@ public final class ConversationLocalStore: ConversationLocalStoreProtocol {
         await context.perform {
             conversation.mlsStatus = .pendingJoinAfterReset
             conversation.mlsGroupID = newMLSGroupID
+            conversation.commitPendingProposalDate = nil
             conversation.epoch = 0
         }
     }

WireDomain/Sources/WireDomain/Repositories/Conversations/Protocols/ConversationRepositoryProtocol.swift

@@ -198,4 +198,9 @@ public protocol ConversationRepositoryProtocol: Sendable {
     func isSelfAnActiveMember(
         in groupID: WireDataModel.MLSGroupID
     ) async -> Bool
+
+    /// Reset the pendingProposalDate for the conversation
+    /// - Parameter groupID: mlsGroupID of the conversation
+    func clearPendingProposals(in groupID: WireDataModel.MLSGroupID) async
+
 }

WireDomain/Sources/WireDomain/Repositories/Conversations/Repository/ConversationRepository.swift

@@ -390,6 +390,13 @@ public final class ConversationRepository: ConversationRepositoryProtocol {
         return isSelfAnActiveMember
     }
 
+    public func clearPendingProposals(in groupID: WireDataModel.MLSGroupID) async {
+        await conversationsLocalStore.execute(identifier: groupID) { conversation, context in
+            conversation?.commitPendingProposalDate = nil
+            context.saveOrRollback()
+        }
+    }
+
     // MARK: - Private
 
     private func addSystemMessage(

WireDomain/Sources/WireDomain/WorkAgent/WorkAgent.swift

@@ -136,6 +136,9 @@ public actor WorkAgent {
         task = nil
     }
 
+    public func clearSchedulerQueue() async {
+        await scheduler.clearAllItems()
+    }
 }
 
 extension LogAttributes {

WireDomain/Sources/WireDomain/WorkAgent/WorkItem/CommitPendingProposalItem.swift

@@ -25,11 +25,16 @@ struct CommitPendingProposalItem: WorkItem, CustomStringConvertible {
     private let repository: ConversationRepositoryProtocol
     private let mlsService: MLSServiceInterface
 
+    let _internalID = UUID()
+
+    var id: String {
+        "commitPendingProposalItem_\(_internalID)_\(groupID)_\(conversationID)"
+    }
+
     var description: String {
-        "CommitPendingProposalItem: \(id), mlsGroupID: \(groupID), conversationID: \(conversationID)"
+        "CommitPendingProposalItem: \(_internalID), mlsGroupID: \(groupID), conversationID: \(conversationID)"
     }
 
-    let id = UUID()
     var priority: WorkItemPriority {
         .medium
     }
@@ -60,6 +65,12 @@ struct CommitPendingProposalItem: WorkItem, CustomStringConvertible {
             return
         }
 
+        guard try await mlsService.conversationExists(groupID: groupID) else {
+            logger.warn("mls group does not exist, clearing pending proposal", attributes: logAttributes)
+            await repository.clearPendingProposals(in: groupID)
+            return
+        }
+
         logger.info("committing pending proposals now...", attributes: logAttributes)
         try await mlsService.commitPendingProposals(in: groupID)
     }

WireDomain/Sources/WireDomain/WorkAgent/WorkItem/RepairFaultyMLSRemovalKeysWorkItem.swift

@@ -27,7 +27,10 @@ import WireLogging
 
 struct RepairFaultyMLSRemovalKeysWorkItem: WorkItem {
 
-    let id = UUID()
+    var id: String {
+        "repairFaultyMLSRemovalKeys_\(UUID().uuidString)"
+    }
+
     var priority: WorkItemPriority {
         .low
     }

WireDomain/Sources/WireDomain/WorkAgent/WorkItem/UpdateConversationItem.swift

@@ -24,7 +24,16 @@ import WireNetwork
 struct UpdateConversationItem: WorkItem {
     private let repository: ConversationRepositoryProtocol
 
-    let id = UUID()
+    let _internalID = UUID()
+
+    var id: String {
+        "updateConversationItem_\(_internalID.uuidString)_\(conversationID)"
+    }
+
+    var description: String {
+        "UpdateConversationItem: \(_internalID.uuidString), conversationID: \(conversationID)"
+    }
+
     var priority: WorkItemPriority {
         .medium
     }

WireDomain/Sources/WireDomain/WorkAgent/WorkItem/WorkItem.swift

@@ -27,7 +27,7 @@ protocol WorkItem: Sendable {
 
     /// A unique identifier for this item.
 
-    var id: UUID { get }
+    var id: String { get }
 
     /// The urgency or importance of this ticket.
 

WireDomain/Sources/WireDomain/WorkAgent/WorkItemScheduler/PriorityOrderWorkItemScheduler.swift

@@ -54,4 +54,10 @@ actor PriorityOrderWorkItemScheduler: WorkItemScheduler {
         }
     }
 
+    func clearAllItems() async {
+        blockerQueue.removeAll()
+        highQueue.removeAll()
+        mediumQueue.removeAll()
+        lowQueue.removeAll()
+    }
 }

WireDomain/Sources/WireDomain/WorkAgent/WorkItemScheduler/WorkItemScheduler.swift

@@ -38,4 +38,8 @@ protocol WorkItemScheduler: Sendable {
     /// - Returns: The next available item.
 
     func dequeueNextItem() async -> (any WorkItem)?
+
+    /// Clears all items from queues
+
+    func clearAllItems() async
 }

WireDomain/Tests/WireDomainTests/WorkAgent/PriorityOrderWorkSchedulerTests.swift

@@ -121,7 +121,7 @@ struct PriorityOrderWorkItemSchedulerTests {
 
 private struct MockWorkItem: WorkItem, Equatable {
 
-    let id = UUID()
+    let id = UUID().uuidString
     let priority: WorkItemPriority
     func start() async throws {}
 

WireDomain/Tests/WireDomainTests/WorkAgent/WorkAgentTests.swift

@@ -88,11 +88,28 @@ struct WorkAgentTests {
         #expect(await item.startCalls == 1)
     }
 
+    @Test("Clear items are removed")
+    func clearItems() async throws {
+        // Given
+        let item = MockWorkItem(priority: .medium)
+        await sut.submitItem(item)
+        await Task.yield()
+
+        #expect(await scheduler.enqueuedItems.count == 1)
+
+        // When
+        await sut.clearSchedulerQueue()
+
+        // Then
+        #expect(await scheduler.items.isEmpty)
+        #expect(await scheduler.enqueuedItems.isEmpty)
+    }
+
 }
 
 private actor MockWorkItem: WorkItem {
 
-    let id = UUID()
+    let id = UUID().uuidString
     let priority: WorkItemPriority
 
     var startCalls = 0
@@ -137,4 +154,8 @@ private actor MockScheduler: WorkItemScheduler {
         return item
     }
 
+    func clearAllItems() async {
+        enqueuedItems.removeAll()
+        items.removeAll()
+    }
 }

WireDomain/Tests/WireDomainTests/WorkAgent/WorkItem/CommitPendingProposalItemTests.swift

@@ -39,6 +39,7 @@ class CommitPendingProposalItemTests {
         self.mlsGroupID = .random()
         self.mlsService = .init()
         mlsService.commitPendingProposalsIn_MockMethod = { _ in }
+        mlsService.conversationExistsGroupID_MockValue = true
         repository.isSelfAnActiveMemberIn_MockValue = true
     }
 
@@ -76,6 +77,22 @@ class CommitPendingProposalItemTests {
         #expect(mlsService.commitPendingProposalsIn_Invocations.isEmpty)
     }
 
+    @Test("It does not call commitPendingProposal when mlsGroup does not exist")
+    func startDoesNotCommitWhenMLSGroupDoesNotExist() async throws {
+        // Given
+        repository.clearPendingProposalsIn_MockMethod = { _ in }
+        mlsService.conversationExistsGroupID_MockValue = false
+
+        sut = makeProposalItem()
+
+        // When
+        try await sut.start()
+
+        // Then
+        #expect(mlsService.commitPendingProposalsIn_Invocations.isEmpty)
+        #expect(repository.clearPendingProposalsIn_Invocations.count == 1)
+    }
+
     @Test("It logs properly")
     func loggingDescription() {
         // Given

WireFoundation/Sources/WireFoundation/Utilities/BackoffRetry/BackoffRetrier.swift

@@ -25,7 +25,7 @@ public actor BackoffRetrier {
         _ seconds: Double
     ) async throws -> Void
 
-    enum Failure: Error {
+    public enum Failure: Error {
         case exceededMaxAttempts(latestError: any Error)
     }
 
@@ -36,14 +36,16 @@ public actor BackoffRetrier {
 
     public init(
         policy: BackoffRetryPolicy = .init(),
+        monitoringNetwork: Bool = true,
         sleep: @escaping SleepFunction = { delay in
             try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
         }
     ) {
         self.policy = policy
         self.sleep = sleep
-
-        setupObservers()
+        if monitoringNetwork {
+            setupObservers()
+        }
     }
 
     deinit {