fix(datastore): ModelSyncedEvent dispatch consistency (#1823)

* fix(DataStore): ModelSyncedEvent dispatch consistency

* address PR comments

Author: lawmicha

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/InitialSync/InitialSyncOperation.swift

@@ -193,7 +193,6 @@ final class InitialSyncOperation: AsynchronousOperation {
                 self.query(lastSyncTime: lastSyncTime, nextToken: nextToken)
             }
         } else {
-            initialSyncOperationTopic.send(.finished(modelName: modelSchema.name))
             updateModelSyncMetadata(lastSyncTime: syncQueryResult.startedAt)
         }
     }
@@ -233,8 +232,10 @@ final class InitialSyncOperation: AsynchronousOperation {
     private func finish(result: AWSInitialSyncOrchestrator.SyncOperationResult) {
         switch result {
         case .failure(let error):
+            initialSyncOperationTopic.send(.finished(modelName: modelSchema.name, error: error))
             initialSyncOperationTopic.send(completion: .failure(error))
         case .success:
+            initialSyncOperationTopic.send(.finished(modelName: modelSchema.name))
             initialSyncOperationTopic.send(completion: .finished)
         }
         super.finish()

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/InitialSync/InitialSyncOperationEvent.swift

@@ -19,7 +19,7 @@ enum InitialSyncOperationEvent {
 
     /// Published when the sync operation has completed and all remote models have been enqueued for reconciliation.
     /// Used by `ModelSyncedEventEmitter` to determine when to send `ModelSyncedEvent`
-    case finished(modelName: ModelName)
+    case finished(modelName: ModelName, error: DataStoreError? = nil)
 }
 
 enum SyncType {

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/InitialSync/InitialSyncOrchestrator.swift

@@ -115,21 +115,9 @@ final class AWSInitialSyncOrchestrator: InitialSyncOrchestrator {
         initialSyncOperationSinks[modelSchema.name] = initialSyncForModel
             .publisher
             .receive(on: concurrencyQueue)
-            .sink(receiveCompletion: { result in
-                if case .failure(let dataStoreError) = result {
-                    let syncError = DataStoreError.sync(
-                        "An error occurred syncing \(modelSchema.name)",
-                        "",
-                        dataStoreError)
-                    self.syncErrors.append(syncError)
-
-                    if self.isUnauthorizedError(syncError) {
-                        self.initialSyncOrchestratorTopic.send(.finished(modelName: modelSchema.name))
-                    }
-                }
-                self.initialSyncOperationSinks.removeValue(forKey: modelSchema.name)
-                self.onReceiveCompletion()
-            }, receiveValue: onReceiveValue(_:))
+            .sink(receiveCompletion: { result in self.onReceiveCompletion(modelSchema: modelSchema,
+                                                                          result: result) },
+                  receiveValue: onReceiveValue(_:))
 
         syncOperationQueue.addOperation(initialSyncForModel)
     }
@@ -138,7 +126,17 @@ final class AWSInitialSyncOrchestrator: InitialSyncOrchestrator {
         initialSyncOrchestratorTopic.send(value)
     }
 
-    private func onReceiveCompletion() {
+    private func onReceiveCompletion(modelSchema: ModelSchema, result: Subscribers.Completion<DataStoreError>) {
+        if case .failure(let dataStoreError) = result {
+            let syncError = DataStoreError.sync(
+                "An error occurred syncing \(modelSchema.name)",
+                "",
+                dataStoreError)
+            self.syncErrors.append(syncError)
+        }
+
+        initialSyncOperationSinks.removeValue(forKey: modelSchema.name)
+
         guard initialSyncOperationSinks.isEmpty else {
             return
         }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/InitialSync/ModelSyncedEventEmitter.swift

@@ -12,7 +12,7 @@ import Foundation
 
 enum IncomingModelSyncedEmitterEvent {
     case mutationEventApplied(MutationEvent)
-    case mutationEventDropped(modelName: String)
+    case mutationEventDropped(modelName: String, error: DataStoreError? = nil)
     case modelSyncedEvent(ModelSyncedEvent)
 }
 
@@ -90,7 +90,7 @@ final class ModelSyncedEventEmitter {
             return modelSchema.name == modelName
         case .enqueued(_, let modelName):
             return modelSchema.name == modelName
-        case .finished(let modelName):
+        case .finished(let modelName, _):
             return modelSchema.name == modelName
         }
     }
@@ -101,7 +101,7 @@ final class ModelSyncedEventEmitter {
         switch value {
         case .mutationEventApplied(let event):
             return modelSchema.name == event.modelName
-        case .mutationEventDropped(let modelName):
+        case .mutationEventDropped(let modelName, _):
             return modelSchema.name == modelName
         case .initialized, .started, .paused:
             return false
@@ -128,8 +128,8 @@ final class ModelSyncedEventEmitter {
             switch value {
             case .mutationEventApplied(let event):
                 modelSyncedEventTopic.send(.mutationEventApplied(event))
-            case .mutationEventDropped(let modelName):
-                modelSyncedEventTopic.send(.mutationEventDropped(modelName: modelName))
+            case .mutationEventDropped(let modelName, let error):
+                modelSyncedEventTopic.send(.mutationEventDropped(modelName: modelName, error: error))
             case .initialized, .started, .paused:
                 return
             }
@@ -155,11 +155,9 @@ final class ModelSyncedEventEmitter {
             if shouldSendModelSyncedEvent {
                 sendModelSyncedEvent()
             }
-        case .mutationEventDropped(let modelName):
+        case .mutationEventDropped(let modelName, let error):
             reconciledReceived += 1
-
-            modelSyncedEventTopic.send(.mutationEventDropped(modelName: modelName))
-
+            modelSyncedEventTopic.send(.mutationEventDropped(modelName: modelName, error: error))
             if shouldSendModelSyncedEvent {
                 sendModelSyncedEvent()
             }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/InitialSync/SyncEventEmitter.swift

@@ -12,7 +12,7 @@ import Foundation
 
 enum IncomingSyncEventEmitterEvent {
     case mutationEventApplied(MutationEvent)
-    case mutationEventDropped(modelName: String)
+    case mutationEventDropped(modelName: String, error: DataStoreError? = nil)
     case modelSyncedEvent(ModelSyncedEvent)
     case syncQueriesReadyEvent
 }
@@ -71,8 +71,8 @@ final class SyncEventEmitter {
         switch value {
         case .mutationEventApplied(let mutationEvent):
             syncEventEmitterTopic.send(.mutationEventApplied(mutationEvent))
-        case .mutationEventDropped(let modelName):
-            syncEventEmitterTopic.send(.mutationEventDropped(modelName: modelName))
+        case .mutationEventDropped(let modelName, let error):
+            syncEventEmitterTopic.send(.mutationEventDropped(modelName: modelName, error: error))
         case .modelSyncedEvent(let modelSyncedEvent):
             modelSyncedReceived += 1
             syncEventEmitterTopic.send(.modelSyncedEvent(modelSyncedEvent))

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/AWSIncomingEventReconciliationQueue.swift

@@ -121,8 +121,8 @@ final class AWSIncomingEventReconciliationQueue: IncomingEventReconciliationQueu
         switch receiveValue {
         case .mutationEvent(let event):
             eventReconciliationQueueTopic.send(.mutationEventApplied(event))
-        case .mutationEventDropped(let modelName):
-            eventReconciliationQueueTopic.send(.mutationEventDropped(modelName: modelName))
+        case .mutationEventDropped(let modelName, let error):
+            eventReconciliationQueueTopic.send(.mutationEventDropped(modelName: modelName, error: error))
         case .connected(modelName: let modelName):
             connectionStatusSerialQueue.async {
                 self.reconciliationQueueConnectionStatus[modelName] = true

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/SubscriptionSync/IncomingEventReconciliationQueue.swift

@@ -14,7 +14,7 @@ enum IncomingEventReconciliationQueueEvent {
     case started
     case paused
     case mutationEventApplied(MutationEvent)
-    case mutationEventDropped(modelName: String)
+    case mutationEventDropped(modelName: String, error: DataStoreError? = nil)
 }
 
 /// A queue that reconciles all incoming events for a model: responses from locally-sourced mutations, and subscription

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

@@ -164,8 +164,8 @@ final class AWSModelReconciliationQueue: ModelReconciliationQueue {
                     return
                 }
                 switch value {
-                case .mutationEventDropped(let modelName):
-                    self.modelReconciliationQueueSubject.send(.mutationEventDropped(modelName: modelName))
+                case .mutationEventDropped(let modelName, let error):
+                    self.modelReconciliationQueueSubject.send(.mutationEventDropped(modelName: modelName, error: error))
                 case .mutationEvent(let event):
                     self.modelReconciliationQueueSubject.send(.mutationEvent(event))
                 }

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

@@ -20,7 +20,7 @@ enum ModelReconciliationQueueEvent {
     case connected(modelName: String)
     case disconnected(modelName: String, reason: ModelConnectionDisconnectedReason)
     case mutationEvent(MutationEvent)
-    case mutationEventDropped(modelName: String)
+    case mutationEventDropped(modelName: String, error: DataStoreError? = nil)
 }
 
 @available(iOS 13.0, *)

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

@@ -112,7 +112,9 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
         }
 
         guard let storageAdapter = storageAdapter else {
-            stateMachine.notify(action: .errored(DataStoreError.nilStorageAdapter()))
+            let error = DataStoreError.nilStorageAdapter()
+            notifyDropped(count: remoteModels.count, error: error)
+            stateMachine.notify(action: .errored(error))
             return
         }
 
@@ -168,7 +170,9 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                 return
             }
             guard let storageAdapter = self.storageAdapter else {
-                result = .failure(DataStoreError.nilStorageAdapter())
+                let error = DataStoreError.nilStorageAdapter()
+                self.notifyDropped(count: modelIds.count, error: error)
+                result = .failure(error)
                 return
             }
 
@@ -181,6 +185,7 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                                                 storageAdapter: storageAdapter) { queryResult in
                 switch queryResult {
                 case .failure(let dataStoreError):
+                    self.notifyDropped(count: modelIds.count, error: dataStoreError)
                     result = .failure(dataStoreError)
                 case .success(let mutationEvents):
                     result = .success(mutationEvents)
@@ -190,17 +195,13 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
     }
 
     func reconcile(_ remoteModels: [RemoteModel], pendingMutations: [MutationEvent]) -> [RemoteModel] {
-        guard let remoteModel = remoteModels.first else {
+        guard !remoteModels.isEmpty else {
             return []
         }
 
         let remoteModelsToApply = RemoteSyncReconciler.filter(remoteModels,
                                                               pendingMutations: pendingMutations)
-
-        for _ in 0 ..< (remoteModels.count - remoteModelsToApply.count) {
-            notifyDropped()
-        }
-
+        notifyDropped(count: remoteModels.count - remoteModelsToApply.count)
         return remoteModelsToApply
     }
 
@@ -217,7 +218,9 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                 return
             }
             guard let storageAdapter = self.storageAdapter else {
-                result = .failure(DataStoreError.nilStorageAdapter())
+                let error = DataStoreError.nilStorageAdapter()
+                self.notifyDropped(count: remoteModels.count, error: error)
+                result = .failure(error)
                 return
             }
 
@@ -232,24 +235,23 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                        modelName: self.modelSchema.name)
                 result = .success((remoteModels, localMetadatas))
             } catch {
-                result = .failure(DataStoreError(error: error))
+                let error = DataStoreError(error: error)
+                self.notifyDropped(count: remoteModels.count, error: error)
+                result = .failure(error)
                 return
             }
         }
     }
 
     func getDispositions(for remoteModels: [RemoteModel],
                          localMetadatas: [LocalMetadata]) -> [RemoteSyncReconciler.Disposition] {
-        guard let remoteModel = remoteModels.first else {
+        guard !remoteModels.isEmpty else {
             return []
         }
 
         let dispositions = RemoteSyncReconciler.getDispositions(remoteModels,
                                                                 localMetadatas: localMetadatas)
-        for _ in 0 ..< (remoteModels.count - dispositions.count) {
-            notifyDropped()
-        }
-
+        notifyDropped(count: remoteModels.count - dispositions.count)
         return dispositions
     }
 
@@ -268,7 +270,9 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                 return
             }
             guard let storageAdapter = self.storageAdapter else {
-                result = .failure(DataStoreError.nilStorageAdapter())
+                let error = DataStoreError.nilStorageAdapter()
+                self.notifyDropped(count: dispositions.count, error: error)
+                result = .failure(error)
                 return
             }
 
@@ -348,8 +352,8 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                                   condition: nil) { response in
                 switch response {
                 case .failure(let dataStoreError):
+                    self.notifyDropped(error: dataStoreError)
                     if storageAdapter.shouldIgnoreError(error: dataStoreError) {
-                        self.notifyDropped()
                         promise(.success(.dropped))
                     } else {
                         promise(.failure(dataStoreError))
@@ -367,8 +371,8 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
             storageAdapter.save(untypedModel: remoteModel.model.instance) { response in
                 switch response {
                 case .failure(let dataStoreError):
+                    self.notifyDropped(error: dataStoreError)
                     if storageAdapter.shouldIgnoreError(error: dataStoreError) {
-                        self.notifyDropped()
                         promise(.success(.dropped))
                     } else {
                         promise(.failure(dataStoreError))
@@ -379,6 +383,7 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
                         anyModel = try savedModel.eraseToAnyModel()
                     } catch {
                         let dataStoreError = DataStoreError(error: error)
+                        self.notifyDropped(error: dataStoreError)
                         promise(.failure(dataStoreError))
                         return
                     }
@@ -401,6 +406,7 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
             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):
                     let appliedModel = MutationSync(model: inProcessModel.model, syncMetadata: syncMetadata)
@@ -411,8 +417,10 @@ class ReconcileAndLocalSaveOperation: AsynchronousOperation {
         }
     }
 
-    private func notifyDropped() {
-        mutationEventPublisher.send(.mutationEventDropped(modelName: modelSchema.name))
+    private func notifyDropped(count: Int = 1, error: DataStoreError? = nil) {
+        for _ in 0 ..< count {
+            mutationEventPublisher.send(.mutationEventDropped(modelName: modelSchema.name, error: error))
+        }
     }
 
     private func notify(savedModel: AppliedModel,
@@ -456,5 +464,5 @@ extension ReconcileAndLocalSaveOperation: DefaultLogger { }
 
 enum ReconcileAndLocalSaveOperationEvent {
     case mutationEvent(MutationEvent)
-    case mutationEventDropped(modelName: String)
+    case mutationEventDropped(modelName: String, error: DataStoreError? = nil)
 }