fix(datastore): change OutgoingMutationQueue use TaskQueue for state …

…transitions (#3720) * fix(datastore): change OutgoingMutationQueue use TaskQueue for state transition * Update AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/MutationSync/OutgoingMutationQueue/OutgoingMutationQueue.swift Co-authored-by: Michael Law <[email protected]> --------- Co-authored-by: Michael Law <[email protected]>
aws-amplify · May 27, 2024 · e51aee8 · e51aee8
1 parent 63b4c1b
commit e51aee8
Showing 1 changed file with 6 additions and 5 deletions.
diff --git a/...es/AWSDataStorePlugin/Sync/MutationSync/OutgoingMutationQueue/OutgoingMutationQueue.swift b/...es/AWSDataStorePlugin/Sync/MutationSync/OutgoingMutationQueue/OutgoingMutationQueue.swift
@@ -27,10 +27,7 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
     private let operationQueue: OperationQueue
 
     /// A DispatchQueue for synchronizing state on the mutation queue
-    private let mutationDispatchQueue = DispatchQueue(
-        label: "com.amazonaws.OutgoingMutationQueue",
-        target: DispatchQueue.global()
-    )
+    private let mutationDispatchQueue = TaskQueue<Void>()
 
     private weak var api: APICategoryGraphQLBehaviorExtended?
     private weak var reconciliationQueue: IncomingEventReconciliationQueue?
@@ -55,7 +52,7 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
 
         let operationQueue = OperationQueue()
         operationQueue.name = "com.amazonaws.OutgoingMutationOperationQueue"
-        operationQueue.underlyingQueue = mutationDispatchQueue
+        operationQueue.qualityOfService = .default
         operationQueue.maxConcurrentOperationCount = 1
         operationQueue.isSuspended = true
 
@@ -139,6 +136,10 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
 
         queryMutationEventsFromStorage { [weak self] in
             guard let self = self else { return }
+            guard case .starting = self.stateMachine.state else {
+                self.log.debug("Unexpected state transition while performing `doStart()` during `.starting` state. Current state: \(self.stateMachine.state).")
+                return
+            }
 
             self.operationQueue.isSuspended = false
             // State machine notification to ".receivedSubscription" will be handled in `receive(subscription:)`