fix(datastore): sync pending mutation events with latest synced metadata

Amplify/Categories/DataStore/Model/Internal/Model+Codable.swift

@@ -76,6 +76,7 @@ extension Model where Self: Codable {
             resolvedEncoder = encoder
         } else {
             resolvedEncoder = JSONEncoder(dateEncodingStrategy: ModelDateFormatting.encodingStrategy)
+            resolvedEncoder.outputFormatting = .sortedKeys
         }
 
         let data = try resolvedEncoder.encode(self)

Amplify/Categories/DataStore/Model/Internal/Schema/ModelValueConverter.swift

@@ -44,7 +44,9 @@ extension ModelValueConverter {
     }
 
     static var jsonEncoder: JSONEncoder {
-        JSONEncoder(dateEncodingStrategy: ModelDateFormatting.encodingStrategy)
+        let encoder = JSONEncoder(dateEncodingStrategy: ModelDateFormatting.encodingStrategy)
+        encoder.outputFormatting = .sortedKeys
+        return encoder
     }
 
     /// - Warning: Although this has `public` access, it is intended for internal & codegen use and should not be used

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/MutationSync/AWSMutationDatabaseAdapter/AWSMutationDatabaseAdapter+MutationEventIngester.swift

@@ -34,22 +34,6 @@ extension AWSMutationDatabaseAdapter: MutationEventIngester {
     func resolveConflictsThenSave(mutationEvent: MutationEvent,
                                   storageAdapter: StorageEngineAdapter,
                                   completionPromise: @escaping Future<MutationEvent, DataStoreError>.Promise) {
-
-        // We don't want to query MutationSync<AnyModel> because a) we already have the model, and b) delete mutations
-        // are submitted *after* the delete has already been applied to the local data store, meaning there is no model
-        // to query.
-        var mutationEvent = mutationEvent
-        do {
-            // swiftlint:disable:next todo
-            // TODO: Refactor this so that it's clear that the storage engine is not responsible for setting the version
-            // perhaps as simple as renaming to `submit(unversionedMutationEvent:)` or similar
-            let syncMetadata = try storageAdapter.queryMutationSyncMetadata(for: mutationEvent.modelId,
-                                                                               modelName: mutationEvent.modelName)
-            mutationEvent.version = syncMetadata?.version
-        } catch {
-            completionPromise(.failure(DataStoreError(error: error)))
-        }
-
         MutationEvent.pendingMutationEvents(
             forMutationEvent: mutationEvent,
             storageAdapter: storageAdapter) { result in
@@ -201,8 +185,6 @@ extension AWSMutationDatabaseAdapter: MutationEventIngester {
         }
         resolvedEvent.mutationType = updatedMutationType
 
-        resolvedEvent.version = candidate.version
-
         return resolvedEvent
     }
 

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

@@ -207,6 +207,9 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
 
         let syncMutationToCloudOperation = SyncMutationToCloudOperation(
             mutationEvent: mutationEvent,
+            getLatestSyncMetadata: {
+                try? self.storageAdapter.queryMutationSyncMetadata(for: mutationEvent.modelId, modelName: mutationEvent.modelName)
+            },
             api: api,
             authModeStrategy: authModeStrategy
         ) { [weak self] result in
@@ -259,12 +262,7 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
                 return
             }
             reconciliationQueue.offer([mutationSync], modelName: mutationEvent.modelName)
-            MutationEvent.reconcilePendingMutationEventsVersion(
-                sent: mutationEvent,
-                received: mutationSync,
-                storageAdapter: storageAdapter) { _ in
-                self.completeProcessingEvent(mutationEvent, mutationSync: mutationSync)
-            }
+            completeProcessingEvent(mutationEvent, mutationSync: mutationSync)
         } else {
             completeProcessingEvent(mutationEvent)
         }

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

@@ -30,7 +30,7 @@ class ProcessMutationErrorFromCloudOperation: AsynchronousOperation {
     private var mutationOperation: AtomicValue<GraphQLOperation<MutationSync<AnyModel>>?>
     private weak var api: APICategoryGraphQLBehavior?
     private weak var reconciliationQueue: IncomingEventReconciliationQueue?
-    
+
     init(dataStoreConfiguration: DataStoreConfiguration,
          mutationEvent: MutationEvent,
          api: APICategoryGraphQLBehavior,
@@ -334,7 +334,7 @@ class ProcessMutationErrorFromCloudOperation: AsynchronousOperation {
                     finish(result: .failure(dataStoreError))
                     return
                 }
-                
+
                 reconciliationQueue.offer([graphQLResult], modelName: mutationEvent.modelName)
             }
         }

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

@@ -19,6 +19,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
     typealias MutationSyncCloudResult = GraphQLOperation<MutationSync<AnyModel>>.OperationResult
 
     private weak var api: APICategoryGraphQLBehavior?
+    private let getLatestSyncMetadata: () -> MutationSyncMetadata?
     private let mutationEvent: MutationEvent
     private let completion: GraphQLOperation<MutationSync<AnyModel>>.ResultListener
     private let requestRetryablePolicy: RequestRetryablePolicy
@@ -32,13 +33,15 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
     private var authTypesIterator: AWSAuthorizationTypeIterator?
 
     init(mutationEvent: MutationEvent,
+         getLatestSyncMetadata: @escaping () -> MutationSyncMetadata?,
          api: APICategoryGraphQLBehavior,
          authModeStrategy: AuthModeStrategy,
          networkReachabilityPublisher: AnyPublisher<ReachabilityUpdate, Never>? = nil,
          currentAttemptNumber: Int = 1,
          requestRetryablePolicy: RequestRetryablePolicy? = RequestRetryablePolicy(),
          completion: @escaping GraphQLOperation<MutationSync<AnyModel>>.ResultListener) {
         self.mutationEvent = mutationEvent
+        self.getLatestSyncMetadata = getLatestSyncMetadata
         self.api = api
         self.networkReachabilityPublisher = networkReachabilityPublisher
         self.completion = completion
@@ -109,7 +112,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
         authType: AWSAuthorizationType? = nil
     ) -> GraphQLRequest<MutationSync<AnyModel>>? {
         var request: GraphQLRequest<MutationSync<AnyModel>>
-
+        let version = getLatestSyncMetadata()?.version
         do {
             var graphQLFilter: GraphQLFilter?
             if let graphQLFilterJSON = mutationEvent.graphQLFilterJSON {
@@ -128,7 +131,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
                 request = GraphQLRequest<MutationSyncResult>.deleteMutation(of: model,
                                                                             modelSchema: modelSchema,
                                                                             where: graphQLFilter,
-                                                                            version: mutationEvent.version)
+                                                                            version: version)
             case .update:
                 let model = try mutationEvent.decodeModel()
                 guard let modelSchema = ModelRegistry.modelSchema(from: mutationEvent.modelName) else {
@@ -140,7 +143,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
                 request = GraphQLRequest<MutationSyncResult>.updateMutation(of: model,
                                                                             modelSchema: modelSchema,
                                                                             where: graphQLFilter,
-                                                                            version: mutationEvent.version)
+                                                                            version: version)
             case .create:
                 let model = try mutationEvent.decodeModel()
                 guard let modelSchema = ModelRegistry.modelSchema(from: mutationEvent.modelName) else {
@@ -151,7 +154,7 @@ class SyncMutationToCloudOperation: AsynchronousOperation {
                 }
                 request = GraphQLRequest<MutationSyncResult>.createMutation(of: model,
                                                                             modelSchema: modelSchema,
-                                                                            version: mutationEvent.version)
+                                                                            version: version)
             }
         } catch {
             let apiError = APIError.unknown("Couldn't decode model", "", error)

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/ReconcileAndLocalSave/ReconcileAndLocalSaveOperation.swift

@@ -106,6 +106,14 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
 
     // MARK: - Responder methods
 
+    /// The reconcile function incorporates incoming mutation events into the local database through the following steps:
+    /// 1. Retrieve the local metadata of the models.
+    /// 2. Generate dispositions based on incoming mutation events and local metadata.
+    /// 3. Categorize dispositions into:
+    ///   3.1 Apply metadata only for those with existing pending mutations.
+    ///     3.1.1 Notify the count of these incoming mutation events as dropped items.
+    ///   3.2 Apply incoming mutation and metadata for those without existing pending mutations.
+    /// 4. Notify the final result.
     func reconcile(remoteModels: [RemoteModel]) {
         guard !isCancelled else {
             log.info("\(#function) - cancelled, aborting")
@@ -126,16 +134,24 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
 
         do {
             try storageAdapter.transaction {
-                queryPendingMutations(forModels: remoteModels.map(\.model))
+                queryLocalMetadata(remoteModels)
                     .subscribe(on: workQueue)
-                    .flatMap { mutationEvents -> Future<([RemoteModel], [LocalMetadata]), DataStoreError> in
-                        let remoteModelsToApply = self.reconcile(remoteModels, pendingMutations: mutationEvents)
-                        return self.queryLocalMetadata(remoteModelsToApply)
+                    .map { remoteModels, localMetadatas in
+                        self.getDispositions(for: remoteModels, localMetadatas: localMetadatas)
                     }
-                    .flatMap { (remoteModelsToApply, localMetadatas) -> Future<Void, DataStoreError> in
-                        let dispositions = self.getDispositions(for: remoteModelsToApply,
-                                                                localMetadatas: localMetadatas)
-                        return self.applyRemoteModelsDispositions(dispositions)
+                    .flatMap { dispositions in
+                        self.queryPendingMutations(forModels: dispositions.map(\.remoteModel.model)).map { pendingMutations in
+                            (pendingMutations, dispositions)
+                        }
+                    }
+                    .map { pendingMutations, dispositions in
+                        self.separateDispositions(pendingMutations: pendingMutations, dispositions: dispositions)
+                    }
+                    .flatMap { dispositions, dispositionsOnlyApplyMetadata in
+                        self.waitAllPublisherFinishes(publishers: dispositionsOnlyApplyMetadata.map(self.saveMetadata(disposition:)))
+                            .flatMap { _ in
+                                self.applyRemoteModelsDispositions(dispositions)
+                            }
                     }
                     .sink(
                         receiveCompletion: {
@@ -195,6 +211,28 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
         }
     }
 
+    func separateDispositions(
+        pendingMutations: [MutationEvent],
+        dispositions: [RemoteSyncReconciler.Disposition]
+    ) -> ([RemoteSyncReconciler.Disposition], [RemoteSyncReconciler.Disposition]) {
+        if dispositions.isEmpty {
+            return ([], [])
+        }
+
+        let pendingMutationModelIds = Set(pendingMutations.map(\.modelId))
+
+        let dispositionsToApply = dispositions.filter {
+            !pendingMutationModelIds.contains($0.remoteModel.model.identifier)
+        }
+
+        let dispositionsOnlyApplyMetadata = dispositions.filter {
+            pendingMutationModelIds.contains($0.remoteModel.model.identifier)
+        }
+
+        notifyDropped(count: dispositionsOnlyApplyMetadata.count)
+        return (dispositionsToApply, dispositionsOnlyApplyMetadata)
+    }
+
     func reconcile(_ remoteModels: [RemoteModel], pendingMutations: [MutationEvent]) -> [RemoteModel] {
         guard !remoteModels.isEmpty else {
             return []
@@ -284,8 +322,7 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
             }
 
             let publishers = dispositions.map { disposition ->
-                Publishers.FlatMap<Future<Void, DataStoreError>,
-                                   Future<ReconcileAndLocalSaveOperation.ApplyRemoteModelResult, DataStoreError>> in
+                AnyPublisher<Void, DataStoreError> in
 
                 switch disposition {
                 case .create(let remoteModel):
@@ -296,7 +333,8 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                                               applyResult: applyResult,
                                               mutationType: .create)
                         }
-                    return publisher
+
+                    return publisher.eraseToAnyPublisher()
                 case .update(let remoteModel):
                     let publisher = self.save(storageAdapter: storageAdapter,
                                               remoteModel: remoteModel)
@@ -305,7 +343,7 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                                               applyResult: applyResult,
                                               mutationType: .update)
                         }
-                    return publisher
+                    return publisher.eraseToAnyPublisher()
                 case .delete(let remoteModel):
                     let publisher = self.delete(storageAdapter: storageAdapter,
                                                 remoteModel: remoteModel)
@@ -314,7 +352,7 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                                               applyResult: applyResult,
                                               mutationType: .delete)
                         }
-                    return publisher
+                    return publisher.eraseToAnyPublisher()
                 }
             }
 
@@ -367,8 +405,10 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
         }
     }
 
-    private func save(storageAdapter: StorageEngineAdapter,
-                      remoteModel: RemoteModel) -> Future<ApplyRemoteModelResult, DataStoreError> {
+    private func save(
+        storageAdapter: StorageEngineAdapter,
+        remoteModel: RemoteModel
+    ) -> Future<ApplyRemoteModelResult, DataStoreError> {
         Future<ApplyRemoteModelResult, DataStoreError> { promise in
             storageAdapter.save(untypedModel: remoteModel.model.instance) { response in
                 switch response {
@@ -396,29 +436,69 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
         }
     }
 
-    private func saveMetadata(storageAdapter: StorageEngineAdapter,
-                              applyResult: ApplyRemoteModelResult,
-                              mutationType: MutationEvent.MutationType) -> Future<Void, DataStoreError> {
-        Future<Void, DataStoreError> { promise in
-            guard case let .applied(inProcessModel) = applyResult else {
-                promise(.successfulVoid)
-                return
-            }
+    private func saveMetadata(
+        disposition: RemoteSyncReconciler.Disposition
+    ) -> AnyPublisher<Void, Never> {
+        guard let storageAdapter = self.storageAdapter else {
+            return Just(()).eraseToAnyPublisher()
+        }
 
-            storageAdapter.save(inProcessModel.syncMetadata, condition: nil) { result in
-                switch result {
-                case .failure(let dataStoreError):
-                    self.notifyDropped(error: dataStoreError)
-                    promise(.failure(dataStoreError))
-                case .success(let syncMetadata):
+        return saveMetadata(
+            storageAdapter: storageAdapter,
+            remoteModel: disposition.remoteModel,
+            mutationType: disposition.mutationType
+        )
+        .map { _ in () }
+        .catch { _ in Just(()) }
+        .eraseToAnyPublisher()
+    }
+
+    private func saveMetadata(
+        storageAdapter: StorageEngineAdapter,
+        remoteModel: RemoteModel,
+        mutationType: MutationEvent.MutationType
+    ) -> Future<MutationSyncMetadata, DataStoreError> {
+        Future { promise in
+            storageAdapter.save(
+                remoteModel.syncMetadata,
+                condition: nil) { result in
+                    promise(result)
+                }
+        }
+    }
+
+    private func saveMetadata(
+        storageAdapter: StorageEngineAdapter,
+        applyResult: ApplyRemoteModelResult,
+        mutationType: MutationEvent.MutationType
+    ) -> AnyPublisher<Void, DataStoreError> {
+        if case .applied(let inProcessModel) = applyResult {
+            return self.saveMetadata(storageAdapter: storageAdapter, remoteModel: inProcessModel, mutationType: mutationType)
+                .handleEvents(receiveOutput: { syncMetadata in
                     let appliedModel = MutationSync(model: inProcessModel.model, syncMetadata: syncMetadata)
                     self.notify(savedModel: appliedModel, mutationType: mutationType)
-                    promise(.successfulVoid)
-                }
-            }
+                }, receiveCompletion: { completion in
+                    if case .failure(let error) = completion {
+                        self.notifyDropped(error: error)
+                    }
+                })
+                .map { _ in () }
+                .eraseToAnyPublisher()
         }
+        return Just(()).setFailureType(to: DataStoreError.self).eraseToAnyPublisher()
     }
 
+    private func waitAllPublisherFinishes<T>(publishers: [AnyPublisher<T, Never>]) -> Future<Void, DataStoreError> {
+         Future { promise in
+             Publishers.MergeMany(publishers)
+                 .collect()
+                 .sink(receiveCompletion: { _ in
+                     promise(.successfulVoid)
+                 }, receiveValue: { _ in })
+                 .store(in: &self.cancellables)
+         }
+     }
+
     private func notifyDropped(count: Int = 1, error: DataStoreError? = nil) {
         for _ in 0 ..< count {
             mutationEventPublisher.send(.mutationEventDropped(modelName: modelSchema.name, error: error))

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/ReconcileAndLocalSave/RemoteSyncReconciler.swift

@@ -17,6 +17,21 @@ struct RemoteSyncReconciler {
         case create(RemoteModel)
         case update(RemoteModel)
         case delete(RemoteModel)
+
+        var remoteModel: RemoteModel {
+            switch self {
+            case .create(let model), .update(let model), .delete(let model):
+                return model
+            }
+        }
+
+        var mutationType: MutationEvent.MutationType {
+            switch self {
+            case .create: return .create
+            case .update: return .update
+            case .delete: return .delete
+            }
+        }
     }
 
     /// Filter the incoming `remoteModels` against the pending mutations.