fix(DataStore): ModelSyncedEvent before last MutationEvent (#1472)

* fix(DataStore): ModelSyncedEvent before last MutationEvent

* address PR comments

Author: lawmicha

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

@@ -10,11 +10,21 @@ import AWSPluginsCore
 import Combine
 import Foundation
 
-@available(iOS 13.0, *)
+enum IncomingModelSyncedEmitterEvent {
+    case mutationEventApplied(MutationEvent)
+    case mutationEventDropped(modelName: String)
+    case modelSyncedEvent
+}
+
 /// Listens to events published by both the `InitialSyncOrchestrator` and `IncomingEventReconciliationQueue`,
 /// and emits a `ModelSyncedEvent` when the initial sync is complete. This class expects
 /// `InitialSyncOrchestrator` and `IncomingEventReconciliationQueue` to have matching counts
-/// for the events they enqueue and process, respectively.
+/// for the events they enqueue and process, respectively. Always send back the reconciled event
+/// (`.mutationEventApplied`, `.mutationEventDropped`). The flow also provides a guaranteed sequence of events for the
+/// mutation event which causes the `ModelSyncedEvent` to be emitted afterwards by
+///     - Check if it `ModelSyncedEvent` should be emitted, if so, emit it.
+///     - Then send the mutation event which was used in the check above.
+@available(iOS 13.0, *)
 final class ModelSyncedEventEmitter {
     private let queue = DispatchQueue(label: "com.amazonaws.ModelSyncedEventEmitterQueue",
                                       target: DispatchQueue.global())
@@ -29,11 +39,15 @@ final class ModelSyncedEventEmitter {
 
     private var modelSyncedEventBuilder: ModelSyncedEvent.Builder
 
-    private var modelSyncedEventTopic: PassthroughSubject<Never, Never>
-    var publisher: AnyPublisher<Never, Never> {
+    private var modelSyncedEventTopic: PassthroughSubject<IncomingModelSyncedEmitterEvent, Never>
+    var publisher: AnyPublisher<IncomingModelSyncedEmitterEvent, Never> {
         return modelSyncedEventTopic.eraseToAnyPublisher()
     }
 
+    var shouldDispatchModelSyncedEvent: Bool {
+        initialSyncOperationFinished && reconciledReceived == recordsReceived
+    }
+
     var dispatchedModelSyncedEvent: AtomicValue<Bool>
 
     init(modelSchema: ModelSchema,
@@ -46,7 +60,7 @@ final class ModelSyncedEventEmitter {
         self.dispatchedModelSyncedEvent = AtomicValue(initialValue: false)
         self.modelSyncedEventBuilder = ModelSyncedEvent.Builder()
 
-        self.modelSyncedEventTopic = PassthroughSubject<Never, Never>()
+        self.modelSyncedEventTopic = PassthroughSubject<IncomingModelSyncedEmitterEvent, Never>()
 
         self.syncOrchestratorSink = initialSyncOrchestrator?
             .publisher
@@ -87,8 +101,8 @@ final class ModelSyncedEventEmitter {
             return modelSchema.name == event.modelName
         case .mutationEventDropped(let modelName):
             return modelSchema.name == modelName
-        default:
-            return true
+        case .initialized, .started, .paused:
+            return false
         }
     }
 
@@ -108,6 +122,18 @@ final class ModelSyncedEventEmitter {
     }
 
     private func onReceiveReconciliationEvent(value: IncomingEventReconciliationQueueEvent) {
+        guard !dispatchedModelSyncedEvent.get() else {
+            switch value {
+            case .mutationEventApplied(let event):
+                modelSyncedEventTopic.send(.mutationEventApplied(event))
+            case .mutationEventDropped(let modelName):
+                modelSyncedEventTopic.send(.mutationEventDropped(modelName: modelName))
+            case .initialized, .started, .paused:
+                return
+            }
+            return
+        }
+
         switch value {
         case .mutationEventApplied(let event):
             reconciledReceived += 1
@@ -121,14 +147,22 @@ final class ModelSyncedEventEmitter {
             default:
                 log.error("Unexpected mutationType received: \(event.mutationType)")
             }
-        case .mutationEventDropped:
+
+            if shouldDispatchModelSyncedEvent {
+                dispatchModelSyncedEvent()
+            }
+
+            modelSyncedEventTopic.send(.mutationEventApplied(event))
+        case .mutationEventDropped(let modelName):
             reconciledReceived += 1
-        default:
-            return
-        }
 
-        if initialSyncOperationFinished && reconciledReceived == recordsReceived {
-            dispatchModelSyncedEvent()
+            if shouldDispatchModelSyncedEvent {
+                dispatchModelSyncedEvent()
+            }
+
+            modelSyncedEventTopic.send(.mutationEventDropped(modelName: modelName))
+        case .initialized, .started, .paused:
+            return
         }
     }
 
@@ -138,13 +172,8 @@ final class ModelSyncedEventEmitter {
                                                  data: modelSyncedEventBuilder.build())
         Amplify.Hub.dispatch(to: .dataStore, payload: modelSyncedEventPayload)
         dispatchedModelSyncedEvent.set(true)
-        modelSyncedEventTopic.send(completion: .finished)
-        cancel()
-    }
-
-    func cancel() {
+        modelSyncedEventTopic.send(.modelSyncedEvent)
         syncOrchestratorSink?.cancel()
-        reconciliationQueueSink?.cancel()
     }
 }
 

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

@@ -9,18 +9,44 @@ import Amplify
 import AWSPluginsCore
 import Combine
 
+enum IncomingSyncEventEmitterEvent {
+    case mutationEventApplied(MutationEvent)
+    case mutationEventDropped(modelName: String)
+}
+
+/// SyncEventEmitter holds onto one ModelSyncedEventEmitter per model. It counts the number of `modelSyncedEvent` to
+/// emit the `syncQueriesReady` and sends back the reconciliation events (`.mutationEventApplied`,
+/// `.mutationEventDropped`) to its subscribers.
 @available(iOS 13.0, *)
 final class SyncEventEmitter {
+    private let queue = DispatchQueue(label: "com.amazonaws.SyncEventEmitter",
+                                      target: DispatchQueue.global())
+
     var modelSyncedEventEmitters: [String: ModelSyncedEventEmitter]
     var initialSyncCompleted: AnyCancellable?
 
+    private var syncableModels: Int
+    private var modelSyncedReceived: Int
+
+    private var syncEventEmitterTopic: PassthroughSubject<IncomingSyncEventEmitterEvent, Never>
+    var publisher: AnyPublisher<IncomingSyncEventEmitterEvent, Never> {
+        return syncEventEmitterTopic.eraseToAnyPublisher()
+    }
+
+    var shouldDispatchSyncQueriesReadyEvent: Bool {
+        syncableModels == modelSyncedReceived
+    }
+
     init(initialSyncOrchestrator: InitialSyncOrchestrator?,
          reconciliationQueue: IncomingEventReconciliationQueue?) {
         self.modelSyncedEventEmitters = [String: ModelSyncedEventEmitter]()
+        self.syncEventEmitterTopic = PassthroughSubject<IncomingSyncEventEmitterEvent, Never>()
+        self.modelSyncedReceived = 0
 
         let syncableModelSchemas = ModelRegistry.modelSchemas.filter { $0.isSyncable }
+        self.syncableModels = syncableModelSchemas.count
 
-        var publishers = [AnyPublisher<Never, Never>]()
+        var publishers = [AnyPublisher<IncomingModelSyncedEmitterEvent, Never>]()
         for syncableModelSchema in syncableModelSchemas {
             let modelSyncedEventEmitter = ModelSyncedEventEmitter(modelSchema: syncableModelSchema,
                                                                   initialSyncOrchestrator: initialSyncOrchestrator,
@@ -31,16 +57,31 @@ final class SyncEventEmitter {
 
         self.initialSyncCompleted = Publishers
             .MergeMany(publishers)
-            .sink(receiveCompletion: { [weak self] _ in
-                self?.dispatchSyncQueriesReady()
-            }, receiveValue: { _ in })
+            .receive(on: queue)
+            .sink(receiveCompletion: { _ in },
+                  receiveValue: { [weak self] value in
+                self?.onReceiveModelSyncedEmitterEvent(value: value)
+            })
+    }
+
+    private func onReceiveModelSyncedEmitterEvent(value: IncomingModelSyncedEmitterEvent) {
+        switch value {
+        case .mutationEventApplied(let mutationEvent):
+            syncEventEmitterTopic.send(.mutationEventApplied(mutationEvent))
+        case .mutationEventDropped(let modelName):
+            syncEventEmitterTopic.send(.mutationEventDropped(modelName: modelName))
+        case .modelSyncedEvent:
+            modelSyncedReceived += 1
+            if shouldDispatchSyncQueriesReadyEvent {
+                dispatchSyncQueriesReady()
+            }
+        }
     }
 
     private func dispatchSyncQueriesReady() {
         let syncQueriesReadyEventPayload = HubPayload(eventName: HubPayload.EventName.DataStore.syncQueriesReady)
         Amplify.Hub.dispatch(to: .dataStore, payload: syncQueriesReadyEventPayload)
     }
-
 }
 
 @available(iOS 13.0, *)

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/RemoteSyncEngine+IncomingEventReconciliationQueueEvent.swift

@@ -46,6 +46,13 @@ extension RemoteSyncEngine {
             }
         case .paused:
             remoteSyncTopicPublisher.send(.subscriptionsPaused)
+        case .mutationEventDropped, .mutationEventApplied:
+            break
+        }
+    }
+
+    func onReceive(receiveValue: IncomingSyncEventEmitterEvent) {
+        switch receiveValue {
         case .mutationEventApplied(let mutationEvent):
             remoteSyncTopicPublisher.send(.mutationEvent(mutationEvent))
         case .mutationEventDropped:

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/RemoteSyncEngine+Retryable.swift

@@ -22,7 +22,6 @@ extension RemoteSyncEngine {
             scheduleRestart(advice: advice)
         } else {
             remoteSyncTopicPublisher.send(completion: .failure(DataStoreError.api(error)))
-            cancelEmitters()
             if let completionBlock = finishedCompletionBlock {
                 completionBlock(.failure(causedBy: error))
                 finishedCompletionBlock = nil

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/RemoteSyncEngine.swift

@@ -29,14 +29,15 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
     private let initialSyncOrchestratorFactory: InitialSyncOrchestratorFactory
 
     var syncEventEmitter: SyncEventEmitter?
-    private var readyEventEmitter: ReadyEventEmitter?
+    var readyEventEmitter: ReadyEventEmitter?
 
     private let mutationEventIngester: MutationEventIngester
     let mutationEventPublisher: MutationEventPublisher
     private let outgoingMutationQueue: OutgoingMutationQueueBehavior
     private var outgoingMutationQueueSink: AnyCancellable?
 
     private var reconciliationQueueSink: AnyCancellable?
+    private var syncEventEmitterSink: AnyCancellable?
 
     let remoteSyncTopicPublisher: PassthroughSubject<RemoteSyncEngineEvent, DataStoreError>
     var publisher: AnyPublisher<RemoteSyncEngineEvent, DataStoreError> {
@@ -230,7 +231,7 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
 
     func terminate() {
         remoteSyncTopicPublisher.send(completion: .finished)
-        cancelEmitters()
+        cleanup()
         if let completionBlock = finishedCompletionBlock {
             completionBlock(.successfulVoid)
             finishedCompletionBlock = nil
@@ -305,6 +306,13 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
         syncEventEmitter = SyncEventEmitter(initialSyncOrchestrator: initialSyncOrchestrator,
                                             reconciliationQueue: reconciliationQueue)
 
+        syncEventEmitterSink = syncEventEmitter?
+            .publisher
+            .sink(
+                receiveCompletion: { _ in },
+                receiveValue: { [weak self] in self?.onReceive(receiveValue: $0) }
+            )
+
         readyEventEmitter = ReadyEventEmitter(
             remoteSyncEnginePublisher: publisher,
             completion: { }
@@ -358,17 +366,15 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
     }
 
     private func cleanup(error: AmplifyError) {
-        reconciliationQueue?.cancel()
-        reconciliationQueue = nil
+        cleanup()
         outgoingMutationQueue.stopSyncingToCloud {
             self.remoteSyncTopicPublisher.send(.cleanedUp)
             self.stateMachine.notify(action: .cleanedUp(error))
         }
     }
 
     private func cleanupForTermination() {
-        reconciliationQueue?.cancel()
-        reconciliationQueue = nil
+        cleanup()
         outgoingMutationQueue.stopSyncingToCloud {
             self.mutationEventPublisher.cancel()
             self.remoteSyncTopicPublisher.send(.cleanedUpForTermination)
@@ -394,8 +400,12 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
     }
 
     /// Must be invoked from workQueue (as during a `respond` call)
-    func cancelEmitters() {
+    func cleanup() {
+        reconciliationQueue?.cancel()
+        reconciliationQueue = nil
+        reconciliationQueueSink = nil
         syncEventEmitter = nil
+        syncEventEmitterSink = nil
         readyEventEmitter = nil
     }
 }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginIntegrationTests/DataStoreHubEventsTests.swift

@@ -154,12 +154,16 @@ class DataStoreHubEventTests: HubEventsIntegrationTestBase {
                 }
 
                 if modelSyncedEvents.count == 2 {
-                    XCTAssertEqual(modelSyncedEvents[0].modelName, expectedSyncedModelNames[0])
-                    XCTAssertTrue(modelSyncedEvents[0].isFullSync)
-                    XCTAssertFalse(modelSyncedEvents[0].isDeltaSync)
-                    XCTAssertEqual(modelSyncedEvents[1].modelName, expectedSyncedModelNames[1])
-                    XCTAssertTrue(modelSyncedEvents[1].isFullSync)
-                    XCTAssertFalse(modelSyncedEvents[1].isDeltaSync)
+                    guard let postModelSyncedEvent = modelSyncedEvents.first(where: { $0.modelName == "Post" }),
+                            let commentModelSyncedEvent = modelSyncedEvents.first(where: { $0.modelName == "Comment" }) else {
+                        XCTFail("Could not get modelSyncedEvent for Post and Comment")
+                        return
+                    }
+
+                    XCTAssertTrue(postModelSyncedEvent.isFullSync)
+                    XCTAssertFalse(postModelSyncedEvent.isDeltaSync)
+                    XCTAssertTrue(commentModelSyncedEvent.isFullSync)
+                    XCTAssertFalse(commentModelSyncedEvent.isDeltaSync)
                     modelSyncedReceived.fulfill()
                 }
             }