fix(datastore-v1): update pending mutation events version from mutation response

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/MutationSync/OutgoingMutationQueue/OutgoingMutationQueue.swift

@@ -262,7 +262,13 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
                 return
             }
             reconciliationQueue.offer([mutationSync], modelName: mutationEvent.modelName)
-            completeProcessingEvent(mutationEvent, mutationSync: mutationSync)
+            MutationEvent.reconcilePendingMutationEventsVersion(
+                sent: mutationEvent,
+                received: mutationSync,
+                storageAdapter: storageAdapter
+            ) { _ in
+                self.completeProcessingEvent(mutationEvent, mutationSync: mutationSync)
+            }
         } else {
             completeProcessingEvent(mutationEvent)
         }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/MutationSync/OutgoingMutationQueue/SyncMutationToCloudOperation.swift

@@ -102,6 +102,33 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
         }
     }
 
+    /// Always retrieve and use the largest version when available. The source of the version comes
+    /// from either the MutationEvent itself, which represents the queue request, or the persisted version
+    /// from the metadata table.
+    ///
+    /// **Version in the Mutation Event**. If there are mulitple mutation events pending, each outgoing
+    /// mutation processing will result in synchronously updating the pending mutation's version
+    /// before enqueuing the mutation response for reconciliation.
+    ///
+    /// **Version persisted in the metadata table**: Reconciliation will persist the latest version in the
+    /// metadata table. In cases of quick consecutive updates, the MutationEvent's version could
+    /// be greater than the persisted since the MutationEvent is updated from the original thread that
+    /// processed the outgoing mutation.
+    private func getLatestVersion(_ mutationEvent: MutationEvent) -> Int? {
+        let latestSyncedMetadataVersion = getLatestSyncMetadata()?.version
+        let mutationEventVersion = mutationEvent.version
+        switch (latestSyncedMetadataVersion, mutationEventVersion) {
+        case let (.some(syncedVersion), .some(version)):
+            return max(syncedVersion, version)
+        case let (.some(syncedVersion), .none):
+            return syncedVersion
+        case let (.none, .some(version)):
+            return version
+        case (.none, .none):
+            return nil
+        }
+    }
+
     /// Creates a GraphQLRequest based on given `mutationType`
     /// - Parameters:
     ///   - mutationType: mutation type
@@ -112,7 +139,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
         authType: AWSAuthorizationType? = nil
     ) -> GraphQLRequest<MutationSync<AnyModel>>? {
         var request: GraphQLRequest<MutationSync<AnyModel>>
-        let latestSyncMetadata = getLatestSyncMetadata()
+        let version = getLatestVersion(mutationEvent)
         do {
             var graphQLFilter: GraphQLFilter?
             if let graphQLFilterJSON = mutationEvent.graphQLFilterJSON {
@@ -131,7 +158,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
                 request = GraphQLRequest<MutationSyncResult>.deleteMutation(of: model,
                                                                             modelSchema: modelSchema,
                                                                             where: graphQLFilter,
-                                                                            version: latestSyncMetadata?.version)
+                                                                            version: version)
             case .update:
                 let model = try mutationEvent.decodeModel()
                 guard let modelSchema = ModelRegistry.modelSchema(from: mutationEvent.modelName) else {
@@ -143,7 +170,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
                 request = GraphQLRequest<MutationSyncResult>.updateMutation(of: model,
                                                                             modelSchema: modelSchema,
                                                                             where: graphQLFilter,
-                                                                            version: latestSyncMetadata?.version)
+                                                                            version: version)
             case .create:
                 let model = try mutationEvent.decodeModel()
                 guard let modelSchema = ModelRegistry.modelSchema(from: mutationEvent.modelName) else {
@@ -154,7 +181,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
                 }
                 request = GraphQLRequest<MutationSyncResult>.createMutation(of: model,
                                                                             modelSchema: modelSchema,
-                                                                            version: latestSyncMetadata?.version)
+                                                                            version: version)
             }
         } catch {
             let apiError = APIError.unknown("Couldn't decode model", "", error)

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/Support/MutationEvent+Extensions.swift

@@ -0,0 +1,103 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Amplify
+import Dispatch
+import AWSPluginsCore
+
+extension MutationEvent {
+    // Consecutive operations that modify a model results in a sequence of pending mutation events that
+    // have the current version of the model. The first mutation event has the correct version of the model,
+    // while the subsequent events will have lower versions if the first mutation event is successfully synced
+    // to the cloud. By reconciling the pending mutation events after syncing the first mutation event,
+    // we attempt to update the pending version to the latest version from the response.
+    // The before and after conditions for consecutive update scenarios are as below:
+    //  - Save, then immediately update
+    //      Queue Before - [(version: nil, inprocess: true, type: .create),
+    //                      (version: nil, inprocess: false, type: .update)]
+    //      Response - [version: 1, type: .create]
+    //      Queue After - [(version: 1, inprocess: false, type: .update)]
+    //  - Save, then immediately delete
+    //      Queue Before - [(version: nil, inprocess: true, type: .create),
+    //                      (version: nil, inprocess: false, type: .delete)]
+    //      Response - [version: 1, type: .create]
+    //      Queue After - [(version: 1, inprocess: false, type: .delete)]
+    //  - Save, sync, then immediately update and delete
+    //      Queue Before (After save, sync)
+    //          - [(version: 1, inprocess: true, type: .update), (version: 1, inprocess: false, type: .delete)]
+    //      Response - [version: 2, type: .update]
+    //      Queue After - [(version: 2, inprocess: false, type: .delete)]
+    //
+    // For a given model `id`, checks the version of the head of pending mutation event queue
+    // against the API response version in `mutationSync` and saves it in the mutation event table if
+    // the response version is a newer one
+    static func reconcilePendingMutationEventsVersion(sent mutationEvent: MutationEvent,
+                                                      received mutationSync: MutationSync<AnyModel>,
+                                                      storageAdapter: StorageEngineAdapter,
+                                                      completion: @escaping DataStoreCallback<Void>) {
+        MutationEvent.pendingMutationEvents(
+            forMutationEvent: mutationEvent,
+            storageAdapter: storageAdapter) { queryResult in
+            switch queryResult {
+            case .failure(let dataStoreError):
+                completion(.failure(dataStoreError))
+            case .success(let localMutationEvents):
+                guard let existingEvent = localMutationEvents.first else {
+                    completion(.success(()))
+                    return
+                }
+
+                guard let reconciledEvent = reconcile(pendingMutationEvent: existingEvent,
+                                                      with: mutationEvent,
+                                                      responseMutationSync: mutationSync) else {
+                    completion(.success(()))
+                    return
+                }
+
+                storageAdapter.save(reconciledEvent, condition: nil) { result in
+                    switch result {
+                    case .failure(let dataStoreError):
+                        completion(.failure(dataStoreError))
+                    case .success:
+                        completion(.success(()))
+                    }
+                }
+            }
+        }
+    }
+
+    static func reconcile(pendingMutationEvent: MutationEvent,
+                          with requestMutationEvent: MutationEvent,
+                          responseMutationSync: MutationSync<AnyModel>) -> MutationEvent? {
+        // return if version of the pending mutation event is not nil and
+        // is >= version contained in the response
+        if pendingMutationEvent.version != nil &&
+            pendingMutationEvent.version! >= responseMutationSync.syncMetadata.version {
+            return nil
+        }
+
+        do {
+            let responseModel = responseMutationSync.model.instance
+            let requestModel = try requestMutationEvent.decodeModel()
+
+            // check if the data sent in the request is the same as the response
+            // if it is, update the pending mutation event version to the response version
+            guard let modelSchema = ModelRegistry.modelSchema(from: requestMutationEvent.modelName),
+                  modelSchema.compare(responseModel, requestModel) else {
+                return nil
+            }
+
+            var pendingMutationEvent = pendingMutationEvent
+            pendingMutationEvent.version = responseMutationSync.syncMetadata.version
+            return pendingMutationEvent
+        } catch {
+            Amplify.log.verbose("Error decoding models: \(error)")
+            return nil
+        }
+    }
+
+}