fix(datastore-v1): mutation event got ignored while executing parallel saving (#2782)

* fix(datastore-v1): mutation event got ignored while executing parallel saving

* refactor: add ifSome operator for optaional

Author: Di Wu

Amplify/Core/Support/Optional+Extension.swift

@@ -0,0 +1,22 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+
+import Foundation
+
+extension Optional {
+    ///
+    /// Performing side effect function when data is exist
+    /// - parameters:
+    ///     - then: a closure that takes wrapped data as a parameter
+    @_spi(OptionalExtension)
+    public func ifSome(_ then: (Wrapped) throws -> Void) rethrows {
+        if case .some(let wrapped) = self {
+            try then(wrapped)
+        }
+    }
+}

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

@@ -5,7 +5,7 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
-import Amplify
+@_spi(OptionalExtension) import Amplify
 import Combine
 import Foundation
 
@@ -208,24 +208,31 @@ extension AWSMutationDatabaseAdapter: MutationEventIngester {
 
     /// Saves the deconflicted mutationEvent, invokes `nextEventPromise` if it exists, and the save was successful,
     /// and finally invokes the completion promise from the future of the original invocation of `submit`
-    func save(mutationEvent: MutationEvent,
-              storageAdapter: StorageEngineAdapter,
-              completionPromise: @escaping Future<MutationEvent, DataStoreError>.Promise) {
-
+    func save(
+        mutationEvent: MutationEvent,
+        storageAdapter: StorageEngineAdapter,
+        completionPromise: @escaping Future<MutationEvent, DataStoreError>.Promise
+    ) {
         log.verbose("\(#function) mutationEvent: \(mutationEvent)")
+        let nextEventPromise = self.nextEventPromise.getAndSet(nil)
         var eventToPersist = mutationEvent
-        if nextEventPromise.get() != nil {
+        if nextEventPromise != nil {
             eventToPersist.inProcess = true
         }
+
         storageAdapter.save(eventToPersist, condition: nil) { result in
             switch result {
             case .failure(let dataStoreError):
                 self.log.verbose("\(#function): Error saving mutation event: \(dataStoreError)")
+                // restore the `nextEventPromise` value when failed to save mutation event
+                // as nextEventPromise is expecting to hanlde error of querying unprocessed mutaiton events
+                // not the failure of saving mutaiton event operation
+                nextEventPromise.ifSome(self.nextEventPromise.set(_:))
             case .success(let savedMutationEvent):
                 self.log.verbose("\(#function): saved \(savedMutationEvent)")
-                if let nextEventPromise = self.nextEventPromise.getAndSet(nil) {
+                nextEventPromise.ifSome {
                     self.log.verbose("\(#function): invoking nextEventPromise with \(savedMutationEvent)")
-                    nextEventPromise(.success(savedMutationEvent))
+                    $0(.success(savedMutationEvent))
                 }
             }
             self.log.verbose("\(#function): invoking completionPromise with \(result)")

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginIntegrationTests/DataStoreEndToEndTests.swift

@@ -679,6 +679,63 @@ class DataStoreEndToEndTests: SyncEngineIntegrationTestBase {
         sink.cancel()
     }
 
+    ///
+    /// - Given: DataStore with clean state
+    /// - When:
+    ///     - do some mutaions to ensure MutationEventPublisher works fine
+    ///     - wait 1 second for OutgoingMutationQueue stateMachine transform to waitingForEventToProcess state
+    ///     - do some mutations in parallel
+    /// - Then: verify no mutaiton loss
+    func testParallelMutations_whenWaitingForEventToProcess_noMutationLoss() throws {
+        setUp(withModels: TestModelRegistration())
+        try startAmplifyAndWaitForReady()
+        let parallelSize = 100
+        let initExpectation = expectation(description: "expect MutationEventPublisher works fine")
+        let parallelExpectation = expectation(description: "expect parallel processing no data loss")
+
+        let newPost = Post(title: UUID().uuidString, content: UUID().uuidString, createdAt: .now())
+
+        let titlePrefix = UUID().uuidString
+        let posts = (0 ..< parallelSize).map {
+            Post(title: "\(titlePrefix)-\($0)", content: UUID().uuidString, createdAt: .now())
+        }
+        var expectedResult = Set<String>()
+        let cancellable = Amplify.Hub.publisher(for: .dataStore)
+            .sink { payload in
+                let event = DataStoreHubEvent(payload: payload)
+                switch event {
+                case .outboxMutationProcessed(let mutationEvent):
+                    guard mutationEvent.modelName == Post.modelName,
+                          let post = mutationEvent.element.model as? Post
+                    else { break }
+
+                    if post.title == newPost.title {
+                        initExpectation.fulfill()
+                    }
+
+                    if post.title.hasPrefix(titlePrefix) {
+                        expectedResult.insert(post.title)
+                    }
+
+                    if expectedResult.count == parallelSize {
+                        parallelExpectation.fulfill()
+                    }
+                default: break
+                }
+            }
+        _ = Amplify.DataStore.save(newPost)
+        wait(for: [initExpectation], timeout: 5)
+        wait(for: 1)
+
+        for post in posts {
+            _ = Amplify.DataStore.save(post)
+        }
+
+        wait(for: [parallelExpectation], timeout: Double(parallelSize))
+        cancellable.cancel()
+        XCTAssertEqual(expectedResult, Set(posts.map(\.title)))
+    }
+
     // MARK: - Helpers
 
     func validateSavePost() throws {

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginIntegrationTests/TestSupport/SyncEngineIntegrationTestBase.swift

@@ -135,3 +135,16 @@ class SyncEngineIntegrationTestBase: DataStoreTestBase {
     }
 
 }
+
+extension XCTestCase {
+
+  func wait(for seconds: TimeInterval) {
+      let waitExpectation = expectation(description: "Waiting")
+
+      let when = DispatchTime.now() + seconds
+      DispatchQueue.main.asyncAfter(deadline: when) {
+          waitExpectation.fulfill()
+      }
+      wait(for: [waitExpectation], timeout: seconds + 0.5)
+  }
+}

AmplifyPlugins/DataStore/Podfile.lock

@@ -105,16 +105,16 @@ CHECKOUT OPTIONS:
     :tag: 2.1.0
 
 SPEC CHECKSUMS:
-  Amplify: d0c5ec982b3448d158980577fe73a3aa6d27d984
-  AmplifyPlugins: 20fce06296f5946bdc2d3aab93f3b81f813abb03
-  AmplifyTestCommon: c28a34e5f412f4d2acbb9f7b250b603fc8d9f4be
+  Amplify: c1341383bbaf4a2576802eaee7cb5a7ec15d7795
+  AmplifyPlugins: 807bdcbdad2eb83d742e5df413eacbd8a3cfad37
+  AmplifyTestCommon: b8097d1abdb02919414245d19a2f5c4382dac6af
   AppSyncRealTimeClient: ec19a24f635611b193eb98a2da573abcf98b793b
   AWSAuthCore: 88e77e867b210e5d09e35a484de19753d587aee3
   AWSCognitoIdentityProvider: 37ff510e8f64dc6a1240088ba92ad4d6f0cd841e
   AWSCognitoIdentityProviderASF: f2cd19990c4ae642ad73d09a4945018a994c9ff8
   AWSCore: 493e49f8118e04fa57d927ceb117ba24a9b5ca02
   AWSMobileClient: 36d9bb90118da3ba14a87f9999818cb314953c5c
-  AWSPluginsCore: a3033b4838dd83e62efc3c7eee660e897485441c
+  AWSPluginsCore: 50d53c4ac99f1d97f893cff748c32ac11668a7eb
   CwlCatchException: 86760545af2a490a23e964d76d7c77442dbce79b
   CwlCatchExceptionSupport: a004322095d7101b945442c86adc7cec0650f676
   CwlMachBadInstructionHandler: aa1fe9f2d08b29507c150d099434b2890247e7f8

AmplifyTests/CoreTests/Optional+ExtensionTests.swift

@@ -0,0 +1,62 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+@_spi(OptionalExtension) import Amplify
+import XCTest
+
+class OptionalExtensionTests: XCTestCase {
+
+    /// - Given:
+    ///     Optional of integer with none
+    /// - When:
+    ///     apply a function to increase a captured local integer variable with value 0
+    /// - Then:
+    ///     - the ifSome function on optional should not be applied
+    ///     - local integer variable value stays 0
+    func testIfSome_withNone_doNothing() {
+        var sideEffect = 0
+        let optional: Int? = .none
+        optional.ifSome { _ in sideEffect += 1 }
+        XCTAssertEqual(0, sideEffect)
+    }
+
+    /// - Given:
+    ///     Optional of integer with value 10
+    /// - When:
+    ///     apply a function to increase a captured local integer variable with value 0
+    /// - Then:
+    ///     - the ifSome function on optioanl should be applied
+    ///     - capture local integer value equals 1
+    func testIfSome_withValue_applyFunction() {
+        var sideEffect = 0
+        let optional: Int? = .some(10)
+        optional.ifSome { _ in sideEffect += 1 }
+        XCTAssertEqual(1, sideEffect)
+    }
+
+    /// - Given:
+    ///     Optional of integer with value 10
+    /// - When:
+    ///     apply a function that throw error
+    /// - Then:
+    ///     - the ifSome function on optioanl should be applied
+    ///     - the error is rethrowed
+    func testIfSome_withValue_applyFunctionRethrowError() {
+        let optional: Int? = .some(10)
+        let expectedError = TestRuntimeError()
+        XCTAssertThrowsError(try optional.ifSome {_ in
+            throw expectedError
+        }) { error in
+            XCTAssertEqual(expectedError, error as? TestRuntimeError)
+        }
+    }
+
+}
+
+fileprivate struct TestRuntimeError: Error, Equatable {
+    let id = UUID()
+}