fix(DataStore): Reconcile mutation responses from conflict handler pa…

…th (#3370)
aws-amplify · Nov 20, 2023 · c72184e · c72184e
1 parent 7b507f3
commit c72184e
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 8 deletions.
diff --git a/...es/AWSDataStorePlugin/Sync/MutationSync/OutgoingMutationQueue/OutgoingMutationQueue.swift b/...es/AWSDataStorePlugin/Sync/MutationSync/OutgoingMutationQueue/OutgoingMutationQueue.swift
@@ -282,7 +282,8 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
             api: api,
             storageAdapter: storageAdapter,
             graphQLResponseError: graphQLResponseError,
-            apiError: apiError
+            apiError: apiError,
+            reconciliationQueue: reconciliationQueue
         ) { [weak self] result in
             guard let self = self else {
                 return

diff --git a/...ugin/Sync/MutationSync/OutgoingMutationQueue/ProcessMutationErrorFromCloudOperation.swift b/...ugin/Sync/MutationSync/OutgoingMutationQueue/ProcessMutationErrorFromCloudOperation.swift
@@ -27,20 +27,23 @@ class ProcessMutationErrorFromCloudOperation: AsynchronousOperation {
     private let completion: (Result<MutationEvent?, Error>) -> Void
     private var mutationOperation: AtomicValue<GraphQLOperation<MutationSync<AnyModel>>?>
     private weak var api: APICategoryGraphQLBehaviorExtended?
-
+    private weak var reconciliationQueue: IncomingEventReconciliationQueue?
+
     init(dataStoreConfiguration: DataStoreConfiguration,
          mutationEvent: MutationEvent,
          api: APICategoryGraphQLBehaviorExtended,
          storageAdapter: StorageEngineAdapter,
          graphQLResponseError: GraphQLResponseError<MutationSync<AnyModel>>? = nil,
          apiError: APIError? = nil,
+         reconciliationQueue: IncomingEventReconciliationQueue? = nil,
          completion: @escaping (Result<MutationEvent?, Error>) -> Void) {
         self.dataStoreConfiguration = dataStoreConfiguration
         self.mutationEvent = mutationEvent
         self.api = api
         self.storageAdapter = storageAdapter
         self.graphQLResponseError = graphQLResponseError
         self.apiError = apiError
+        self.reconciliationQueue = reconciliationQueue
         self.completion = completion
         self.mutationOperation = AtomicValue(initialValue: nil)
 
@@ -311,12 +314,27 @@ class ProcessMutationErrorFromCloudOperation: AsynchronousOperation {
         if case .failure(let error) = cloudResult {
             dataStoreConfiguration.errorHandler(error)
         }
-
-        if case .success(let response) = cloudResult,
-            case .failure(let error) = response {
-            dataStoreConfiguration.errorHandler(error)
+
+        if case let .success(graphQLResponse) = cloudResult {
+            if case .failure(let error) = graphQLResponse {
+                dataStoreConfiguration.errorHandler(error)
+            } else if case let .success(graphQLResult) = graphQLResponse {
+                guard let reconciliationQueue = reconciliationQueue else {
+                    let dataStoreError = DataStoreError.configuration(
+                        "reconciliationQueue is unexpectedly nil",
+                        """
+                        The reference to reconciliationQueue has been released while an ongoing mutation was being processed.
+                        \(AmplifyErrorMessages.reportBugToAWS())
+                        """
+                    )
+                    finish(result: .failure(dataStoreError))
+                    return
+                }
+
+                reconciliationQueue.offer([graphQLResult], modelName: mutationEvent.modelName)
+            }
         }
-
+        
         finish(result: .success(nil))
     }
 

diff --git a/...DataStorePluginTests/Sync/MutationQueue/ProcessMutationErrorFromCloudOperationTests.swift b/...DataStorePluginTests/Sync/MutationQueue/ProcessMutationErrorFromCloudOperationTests.swift
@@ -24,6 +24,7 @@ class ProcessMutationErrorFromCloudOperationTests: XCTestCase {
     var storageAdapter: StorageEngineAdapter!
     var localPost = Post(title: "localTitle", content: "localContent", createdAt: .now())
     let queue = OperationQueue()
+    let reconciliationQueue = MockReconciliationQueue()
 
     override func setUp() async throws {
         await tryOrFail {
@@ -573,12 +574,13 @@ class ProcessMutationErrorFromCloudOperationTests: XCTestCase {
             expectConflicthandlerCalled.fulfill()
             resolve(.retryLocal)
         })
-
+        
         let operation = ProcessMutationErrorFromCloudOperation(dataStoreConfiguration: configuration,
                                                                mutationEvent: mutationEvent,
                                                                api: mockAPIPlugin,
                                                                storageAdapter: storageAdapter,
                                                                graphQLResponseError: graphQLResponseError,
+                                                               reconciliationQueue: reconciliationQueue,
                                                                completion: completion)
 
         queue.addOperation(operation)
@@ -656,6 +658,7 @@ class ProcessMutationErrorFromCloudOperationTests: XCTestCase {
                                                                api: mockAPIPlugin,
                                                                storageAdapter: storageAdapter,
                                                                graphQLResponseError: graphQLResponseError,
+                                                               reconciliationQueue: reconciliationQueue,
                                                                completion: completion)
 
         queue.addOperation(operation)
@@ -950,6 +953,7 @@ class ProcessMutationErrorFromCloudOperationTests: XCTestCase {
                                                                api: mockAPIPlugin,
                                                                storageAdapter: storageAdapter,
                                                                graphQLResponseError: graphQLResponseError,
+                                                               reconciliationQueue: reconciliationQueue,
                                                                completion: completion)
 
         queue.addOperation(operation)
@@ -1029,6 +1033,7 @@ class ProcessMutationErrorFromCloudOperationTests: XCTestCase {
                                                                api: mockAPIPlugin,
                                                                storageAdapter: storageAdapter,
                                                                graphQLResponseError: graphQLResponseError,
+                                                               reconciliationQueue: reconciliationQueue,
                                                                completion: completion)
         queue.addOperation(operation)