fix(DataStore): Optimize mutation event propagation after model synced in ModelSyncEventEmitter (#1479)

* fix(DataStore): Optimize mutation event propagation after model synced in ModelSyncEventEmitter

* address PR comments

* address PR comments

Author: lawmicha

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Subscribe/DataStoreObserveQueryOperation.swift

@@ -145,7 +145,7 @@ public class AWSDataStoreObserveQueryOperation<M: Model>: AsynchronousOperation,
            let remoteSyncEngine = storageAdapter.syncEngine as? RemoteSyncEngine,
            let modelSyncedEventEmitter = remoteSyncEngine
             .syncEventEmitter?.modelSyncedEventEmitters[modelType.modelName] {
-            return modelSyncedEventEmitter.dispatchedModelSyncedEvent.get()
+            return modelSyncedEventEmitter.dispatchedModelSyncedEvent
         }
         return false
     }

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

@@ -28,6 +28,7 @@ enum IncomingModelSyncedEmitterEvent {
 final class ModelSyncedEventEmitter {
     private let queue = DispatchQueue(label: "com.amazonaws.ModelSyncedEventEmitterQueue",
                                       target: DispatchQueue.global())
+    private let dispatchedModelSyncedEventLock = NSLock()
 
     private var syncOrchestratorSink: AnyCancellable?
     private var reconciliationQueueSink: AnyCancellable?
@@ -48,7 +49,27 @@ final class ModelSyncedEventEmitter {
         initialSyncOperationFinished && reconciledReceived == recordsReceived
     }
 
-    var dispatchedModelSyncedEvent: AtomicValue<Bool>
+    /// Used internally within ModelSyncedEventEmitter instances, not thread-safe, is accessed serially under the
+    /// DispatchQueue. Exists to avoid using `dispatchedModelSyncedEvent` which requires acquiring a lock each time.
+    private var _dispatchedModelSyncedEvent: Bool
+
+    /// Used by other internal classes for checking state in a thread-safe way.
+    var dispatchedModelSyncedEvent: Bool {
+        get {
+            dispatchedModelSyncedEventLock.lock()
+            defer {
+                dispatchedModelSyncedEventLock.unlock()
+            }
+            return _dispatchedModelSyncedEvent
+        }
+        set {
+            dispatchedModelSyncedEventLock.lock()
+            defer {
+                dispatchedModelSyncedEventLock.unlock()
+            }
+            _dispatchedModelSyncedEvent = newValue
+        }
+    }
 
     init(modelSchema: ModelSchema,
          initialSyncOrchestrator: InitialSyncOrchestrator?,
@@ -57,7 +78,7 @@ final class ModelSyncedEventEmitter {
         self.recordsReceived = 0
         self.reconciledReceived = 0
         self.initialSyncOperationFinished = false
-        self.dispatchedModelSyncedEvent = AtomicValue(initialValue: false)
+        self._dispatchedModelSyncedEvent = false
         self.modelSyncedEventBuilder = ModelSyncedEvent.Builder()
 
         self.modelSyncedEventTopic = PassthroughSubject<IncomingModelSyncedEmitterEvent, Never>()
@@ -122,7 +143,7 @@ final class ModelSyncedEventEmitter {
     }
 
     private func onReceiveReconciliationEvent(value: IncomingEventReconciliationQueueEvent) {
-        guard !dispatchedModelSyncedEvent.get() else {
+        guard !_dispatchedModelSyncedEvent else {
             switch value {
             case .mutationEventApplied(let event):
                 modelSyncedEventTopic.send(.mutationEventApplied(event))
@@ -171,7 +192,7 @@ final class ModelSyncedEventEmitter {
         let modelSyncedEventPayload = HubPayload(eventName: HubPayload.EventName.DataStore.modelSynced,
                                                  data: modelSyncedEventBuilder.build())
         Amplify.Hub.dispatch(to: .dataStore, payload: modelSyncedEventPayload)
-        dispatchedModelSyncedEvent.set(true)
+        dispatchedModelSyncedEvent = true
         modelSyncedEventTopic.send(.modelSyncedEvent)
         syncOrchestratorSink?.cancel()
     }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginTests/Sync/InitialSync/ModelSyncedEventEmitterTests.swift

@@ -115,4 +115,79 @@ class ModelSyncedEventEmitterTests: XCTestCase {
         emitterSink?.cancel()
         listener.cancel()
     }
+
+    /// Send randomly 80% of the events from `initialSyncOrchestrator`, and 20% of the events from
+    /// `reconciliationQueue`. This replicates the scenario of concurrent events received by the emitter. Then finished
+    /// event from `initialSyncOrchestrator`. Then concurrent send all events from `reconciliationQueue`, one of these
+    /// events will resolve to dispatching the modelSyncedEvent, followed by the remaining events being sent back from
+    /// the emitter's topic. This exercises that concurrent events do not cause a data race in the emitter.
+    ///
+    /// - Given: Emitter is seeded with events from both the `initialSyncOrchestrator` and `reconcilliationQueue`,
+    ///     followed by a finished event from initialSyncOrchestrator
+    /// - When:
+    ///    - Concurrently receive more events from the `reconcilliationQueue`
+    /// - Then:
+    ///    - `modelSyncedEvent` received and emitter continues to send mutation events
+    ///
+    func testConcurrent() throws {
+        let modelSyncedReceived = expectation(description: "modelSynced received")
+        let mutationEventAppliedReceived = expectation(description: "mutationEventApplied received")
+        mutationEventAppliedReceived.assertForOverFulfill = false
+        let mutationEventDroppedReceived = expectation(description: "mutationEventDropped received")
+        mutationEventDroppedReceived.assertForOverFulfill = false
+        let anyPostMetadata = MutationSyncMetadata(id: "1",
+                                                   deleted: false,
+                                                   lastChangedAt: Int(Date().timeIntervalSince1970),
+                                                   version: 1)
+        let testPost = Post(id: "1", title: "post1", content: "content", createdAt: .now())
+        let anyPost = AnyModel(testPost)
+        let anyPostMutationSync = MutationSync<AnyModel>(model: anyPost, syncMetadata: anyPostMetadata)
+        let postMutationEvent = try MutationEvent(untypedModel: testPost, mutationType: .create)
+
+        let emitter = ModelSyncedEventEmitter(modelSchema: Post.schema,
+                                              initialSyncOrchestrator: initialSyncOrchestrator,
+                                              reconciliationQueue: reconciliationQueue)
+
+        initialSyncOrchestrator?.initialSyncOrchestratorTopic.send(.started(modelName: Post.modelName,
+                                                                            syncType: .fullSync))
+        DispatchQueue.concurrentPerform(iterations: 1_000) { _ in
+            let index = Int.random(in: 1 ... 10)
+            if index == 1 {
+                reconciliationQueue?.incomingEventSubject.send(.mutationEventApplied(postMutationEvent))
+            } else if index == 2 {
+                reconciliationQueue?.incomingEventSubject.send(.mutationEventDropped(modelName: Post.modelName))
+            } else {
+                initialSyncOrchestrator?.initialSyncOrchestratorTopic.send(.enqueued(anyPostMutationSync,
+                                                                                     modelName: Post.modelName))
+            }
+
+        }
+        initialSyncOrchestrator?.initialSyncOrchestratorTopic.send(.finished(modelName: Post.modelName))
+
+        var emitterSink: AnyCancellable?
+        emitterSink = emitter.publisher.sink { _ in
+            XCTFail("Should not have completed")
+        } receiveValue: { value in
+            switch value {
+            case .modelSyncedEvent:
+                modelSyncedReceived.fulfill()
+            case .mutationEventApplied:
+                mutationEventAppliedReceived.fulfill()
+            case .mutationEventDropped:
+                mutationEventDroppedReceived.fulfill()
+            }
+        }
+
+        DispatchQueue.concurrentPerform(iterations: 3_000) { _ in
+            let index = Int.random(in: 1 ... 2)
+            if index == 1 {
+                reconciliationQueue?.incomingEventSubject.send(.mutationEventApplied(postMutationEvent))
+            } else if index == 2 {
+                reconciliationQueue?.incomingEventSubject.send(.mutationEventDropped(modelName: Post.modelName))
+            }
+        }
+
+        waitForExpectations(timeout: 10)
+        emitterSink?.cancel()
+    }
 }

AmplifyPlugins/DataStore/Podfile.lock

@@ -23,18 +23,18 @@ PODS:
     - AWSPluginsCore (= 1.15.2)
   - AppSyncRealTimeClient (1.5.0):
     - Starscream (~> 3.1.1)
-  - AWSAuthCore (2.26.1):
-    - AWSCore (= 2.26.1)
-  - AWSCognitoIdentityProvider (2.26.1):
-    - AWSCognitoIdentityProviderASF (= 2.26.1)
-    - AWSCore (= 2.26.1)
-  - AWSCognitoIdentityProviderASF (2.26.1)
-  - AWSCore (2.26.1)
-  - AWSMobileClient (2.26.1):
-    - AWSAuthCore (= 2.26.1)
-    - AWSCognitoIdentityProvider (= 2.26.1)
-    - AWSCognitoIdentityProviderASF (= 2.26.1)
-    - AWSCore (= 2.26.1)
+  - AWSAuthCore (2.26.2):
+    - AWSCore (= 2.26.2)
+  - AWSCognitoIdentityProvider (2.26.2):
+    - AWSCognitoIdentityProviderASF (= 2.26.2)
+    - AWSCore (= 2.26.2)
+  - AWSCognitoIdentityProviderASF (2.26.2)
+  - AWSCore (2.26.2)
+  - AWSMobileClient (2.26.2):
+    - AWSAuthCore (= 2.26.2)
+    - AWSCognitoIdentityProvider (= 2.26.2)
+    - AWSCognitoIdentityProviderASF (= 2.26.2)
+    - AWSCore (= 2.26.2)
   - AWSPluginsCore (1.15.2):
     - Amplify (= 1.15.2)
     - AWSCore (~> 2.26.1)
@@ -103,11 +103,11 @@ SPEC CHECKSUMS:
   AmplifyPlugins: bf1558a4efb6ed94b80eff7a20fda020322409e3
   AmplifyTestCommon: 283bd8bf694569b1dda7f40f40a8ad1f97784eeb
   AppSyncRealTimeClient: 2b4482b1770a3e5cf64f9714a6d198550017b5a2
-  AWSAuthCore: 264faaa6af5990af2c77effe8f398a4b80e6c44e
-  AWSCognitoIdentityProvider: 5df455775b57eb4e61862acd88d93e56113950c4
-  AWSCognitoIdentityProviderASF: b2c180f69537d57ff485f7314fd266032ca3f898
-  AWSCore: 0f855e20ccb13e028932b737f0706023a9561399
-  AWSMobileClient: ba255030a481b8a123b21585548e761456efc64b
+  AWSAuthCore: c785a33af10a45aab6456a5fa7150f759dad836f
+  AWSCognitoIdentityProvider: 0bea89a822d7dd76c7dc62212561a1074dee0031
+  AWSCognitoIdentityProviderASF: 425d7f40fb6a520f8437a5f8b991215239ec620b
+  AWSCore: dd7f74ab3f354aad435f83afdbfac91bcb7d44c2
+  AWSMobileClient: fc20c58eaad166b3d2a569ef9ce691d6fed0c5dd
   AWSPluginsCore: ff09e8b86b7969327a1ba0b8c6946d83008cbf81
   CwlCatchException: 70a52ae44ea5d46db7bd385f801a94942420cd8c
   CwlPreconditionTesting: d33a4e4f285c0b885fddcae5dfedfbb34d4f3961