Added retry/restart logic to RemoteSyncEngine without cancel downstream components

Author: wooj2

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

@@ -23,10 +23,12 @@ extension RemoteSyncEngine {
         case activatedCloudSubscriptions(APICategoryGraphQLBehavior, MutationEventPublisher)
         case activatedMutationQueue
         case notifiedSyncStarted
+        case cleanedUp(AmplifyError?)
+        case scheduleRestart(AmplifyError?)
 
         // Terminal actions
         case receivedCancel
-        case errored(AmplifyError)
+        case errored(AmplifyError?)
 
         var displayName: String {
             switch self {
@@ -46,6 +48,10 @@ extension RemoteSyncEngine {
                 return "activatedMutationQueue"
             case .notifiedSyncStarted:
                 return "notifiedSyncStarted"
+            case .cleanedUp:
+                return "cleanedUp"
+            case .scheduleRestart:
+                return "scheduleRestart"
             case .receivedCancel:
                 return "receivedCancel"
             case .errored:

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

@@ -13,14 +13,11 @@ import Foundation
 extension RemoteSyncEngine {
     @available(iOS 13.0, *)
     func onReceiveCompletion(receiveCompletion: Subscribers.Completion<DataStoreError>) {
-        if case .failure(let error) = receiveCompletion {
-            remoteSyncTopicPublisher.send(completion: .failure(error))
-        }
-        if case .finished = receiveCompletion {
-            let unexpectedFinishError = DataStoreError.unknown("ReconcilationQueue sent .finished message",
-                                                               AmplifyErrorMessages.shouldNotHappenReportBugToAWS(),
-                                                               nil)
-            remoteSyncTopicPublisher.send(completion: .failure(unexpectedFinishError))
+        switch receiveCompletion {
+        case .failure(let error):
+            stateMachine.notify(action: .errored(error))
+        case .finished:
+            stateMachine.notify(action: .errored(nil))
         }
     }
 

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

@@ -37,12 +37,22 @@ extension RemoteSyncEngine {
 
             case (.activateMutationQueue, .activatedMutationQueue):
                 return .notifySyncStarted
+
             case (.activateMutationQueue, .errored(let error)):
                 return .cleanup(error)
 
             case (.notifySyncStarted, .notifiedSyncStarted):
                 return .syncEngineActive
 
+            case (.syncEngineActive, .errored(let error)):
+                return .cleanup(error)
+
+            case (.cleanup, .cleanedUp(let error)):
+                return .scheduleRestart(error)
+
+            case (.scheduleRestart, .receivedStart):
+                return .pauseSubscriptions
+
             default:
                 log.warn("Unexpected state transition. In \(currentState.displayName), got \(action.displayName)")
                 log.verbose("Unexpected state transition. In \(currentState), got \(action)")

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

@@ -0,0 +1,64 @@
+//
+// Copyright 2018-2020 Amazon.com,
+// Inc. or its affiliates. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Amplify
+import Foundation
+
+@available(iOS 13.0, *)
+extension RemoteSyncEngine {
+
+    func resetCurrentAttemptNumber() {
+        currentAttemptNumber = 1
+    }
+
+    func scheduleRestart(error: AmplifyError?) {
+        let advice = getRetryAdviceIfRetryable(error: error)
+        if advice.shouldRetry {
+            scheduleRestart(advice: advice)
+        } else {
+            if let error = error {
+                remoteSyncTopicPublisher.send(completion: .failure(DataStoreError.api(error)))
+            } else {
+                remoteSyncTopicPublisher.send(completion: .finished)
+            }
+        }
+
+    }
+
+    private func getRetryAdviceIfRetryable(error: Error?) -> RequestRetryAdvice {
+        //TODO: Parse error from the receive completion to use as an input into getting retry advice.
+        //      For now, specifying not connected to internet to force a retry up to our maximum
+        let urlError = URLError(.notConnectedToInternet)
+        let advice = requestRetryablePolicy.retryRequestAdvice(urlError: urlError,
+                                                               httpURLResponse: nil,
+                                                               attemptNumber: currentAttemptNumber)
+        return advice
+    }
+
+    private func scheduleRestart(advice: RequestRetryAdvice) {
+        log.verbose("\(#function) scheduling retry for restarting remote sync engine")
+        resolveReachabilityPublisher()
+        mutationRetryNotifier = MutationRetryNotifier(advice: advice,
+                                                      networkReachabilityPublisher: networkReachabilityPublisher) {
+                                                        self.mutationRetryNotifier = nil
+                                                        self.stateMachine.notify(action: .receivedStart)
+        }
+        currentAttemptNumber += 1
+    }
+
+    private func resolveReachabilityPublisher() {
+        if networkReachabilityPublisher == nil {
+            if let reachability = api as? APICategoryReachabilityBehavior {
+                do {
+                    networkReachabilityPublisher = try reachability.reachabilityPublisher()
+                } catch {
+                    log.error("\(#function): Unable to listen on reachability: \(error)")
+                }
+            }
+        }
+    }
+}

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

@@ -24,7 +24,9 @@ extension RemoteSyncEngine {
 
         case syncEngineActive
 
-        case cleanup(AmplifyError)
+        case cleanup(AmplifyError?)
+        case scheduleRestart(AmplifyError?)
+
         var displayName: String {
             switch self {
             case .notStarted:
@@ -47,6 +49,8 @@ extension RemoteSyncEngine {
                 return "syncEngineActive"
             case .cleanup:
                 return "cleanup"
+            case .scheduleRestart:
+                return "scheduleRestart"
             }
         }
     }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/RemoteSyncEngine.swift

@@ -44,14 +44,21 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
     let stateMachine: StateMachine<State, Action>
     private var stateMachineSink: AnyCancellable?
 
+    var networkReachabilityPublisher: AnyPublisher<ReachabilityUpdate, Never>?
+    var mutationRetryNotifier: MutationRetryNotifier?
+    let requestRetryablePolicy: RequestRetryablePolicy
+    var currentAttemptNumber: Int
+
     /// Initializes the CloudSyncEngine with the specified storageAdapter as the provider for persistence of
     /// MutationEvents, sync metadata, and conflict resolution metadata. Immediately initializes the incoming mutation
     /// queue so it can begin accepting incoming mutations from DataStore.
     convenience init(storageAdapter: StorageEngineAdapter,
                      outgoingMutationQueue: OutgoingMutationQueueBehavior? = nil,
                      initialSyncOrchestratorFactory: InitialSyncOrchestratorFactory? = nil,
                      reconciliationQueueFactory: IncomingEventReconciliationQueueFactory? = nil,
-                     stateMachine: StateMachine<State, Action>? = nil) throws {
+                     stateMachine: StateMachine<State, Action>? = nil,
+                     networkReachabilityPublisher: AnyPublisher<ReachabilityUpdate, Never>? = nil,
+                     requestRetryablePolicy: RequestRetryablePolicy? = nil) throws {
         let mutationDatabaseAdapter = try AWSMutationDatabaseAdapter(storageAdapter: storageAdapter)
         let awsMutationEventPublisher = AWSMutationEventPublisher(eventSource: mutationDatabaseAdapter)
         let outgoingMutationQueue = outgoingMutationQueue ?? OutgoingMutationQueue()
@@ -61,14 +68,18 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
             AWSInitialSyncOrchestrator.init(api:reconciliationQueue:storageAdapter:)
         let stateMachine = stateMachine ?? StateMachine(initialState: .notStarted,
                                                         resolver: RemoteSyncEngine.Resolver.resolve(currentState:action:))
+        let requestRetryablePolicy = requestRetryablePolicy ?? RequestRetryablePolicy()
+
 
         self.init(storageAdapter: storageAdapter,
                   outgoingMutationQueue: outgoingMutationQueue,
                   mutationEventIngester: mutationDatabaseAdapter,
                   mutationEventPublisher: awsMutationEventPublisher,
                   initialSyncOrchestratorFactory: initialSyncOrchestratorFactory,
                   reconciliationQueueFactory: reconciliationQueueFactory,
-                  stateMachine: stateMachine)
+                  stateMachine: stateMachine,
+                  networkReachabilityPublisher: networkReachabilityPublisher,
+                  requestRetryablePolicy: requestRetryablePolicy)
     }
 
     init(storageAdapter: StorageEngineAdapter,
@@ -77,19 +88,25 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
          mutationEventPublisher: MutationEventPublisher,
          initialSyncOrchestratorFactory: @escaping InitialSyncOrchestratorFactory,
          reconciliationQueueFactory: @escaping IncomingEventReconciliationQueueFactory,
-         stateMachine: StateMachine<State, Action>) {
+         stateMachine: StateMachine<State, Action>,
+         networkReachabilityPublisher: AnyPublisher<ReachabilityUpdate, Never>?,
+         requestRetryablePolicy: RequestRetryablePolicy) {
         self.storageAdapter = storageAdapter
         self.mutationEventIngester = mutationEventIngester
         self.mutationEventPublisher = mutationEventPublisher
         self.outgoingMutationQueue = outgoingMutationQueue
         self.initialSyncOrchestratorFactory = initialSyncOrchestratorFactory
         self.reconciliationQueueFactory = reconciliationQueueFactory
         self.remoteSyncTopicPublisher = PassthroughSubject<RemoteSyncEngineEvent, DataStoreError>()
+        self.networkReachabilityPublisher = networkReachabilityPublisher
+        self.requestRetryablePolicy = requestRetryablePolicy
 
         self.syncQueue = OperationQueue()
         syncQueue.name = "com.amazonaws.Amplify.\(AWSDataStorePlugin.self).CloudSyncEngine"
         syncQueue.maxConcurrentOperationCount = 1
 
+        self.currentAttemptNumber = 1
+
         self.stateMachine = stateMachine
         self.stateMachineSink = self.stateMachine
             .$state
@@ -128,10 +145,11 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
         case .syncEngineActive:
             break
 
-        case .cleanup(let amplifyError):
-            //todo
-            print("error: \(amplifyError)")
+        case .cleanup(let error):
+            cleanup(error: error)
 
+        case .scheduleRestart(let error):
+            scheduleRestart(error: error)
         }
     }
 
@@ -217,7 +235,7 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
     private func activateCloudSubscriptions() {
         log.debug(#function)
         reconciliationQueue?.start()
-        
+
         //Notifying the publisher & state machine are handled in:
         // RemoteSyncEngine+IncomingEventReconciliationQueueEvent.swift
     }
@@ -231,10 +249,20 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
         stateMachine.notify(action: .activatedMutationQueue)
     }
 
+    private func cleanup(error: AmplifyError?) {
+        // TODO:
+        // handle clean up here
+        remoteSyncTopicPublisher.send(.cleanedUp)
+        stateMachine.notify(action: .cleanedUp(error))
+    }
+
     private func notifySyncStarted() {
+        resetCurrentAttemptNumber()
         Amplify.Hub.dispatch(to: .dataStore,
                              payload: HubPayload(eventName: HubPayload.EventName.DataStore.syncStarted))
+
         remoteSyncTopicPublisher.send(.syncStarted)
+        stateMachine.notify(action: .notifiedSyncStarted)
     }
 
     func reset(onComplete: () -> Void) {

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/RemoteSyncEngineBehavior.swift

@@ -17,6 +17,7 @@ enum RemoteSyncEngineEvent {
     case subscriptionsActivated
     case mutationQueueStarted
     case syncStarted
+    case cleanedUp
     case mutationEvent(MutationEvent)
 }
 

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginTests/Sync/LocalSubscriptionTests.swift

@@ -41,7 +41,9 @@ class LocalSubscriptionTests: XCTestCase {
                                              mutationEventPublisher: awsMutationEventPublisher,
                                              initialSyncOrchestratorFactory: NoOpInitialSyncOrchestrator.factory,
                                              reconciliationQueueFactory: MockAWSIncomingEventReconciliationQueue.factory,
-                                             stateMachine: stateMachine)
+                                             stateMachine: stateMachine,
+                                             networkReachabilityPublisher: nil,
+                                             requestRetryablePolicy: MockRequestRetryablePolicy())
 
             storageEngine = StorageEngine(storageAdapter: storageAdapter,
                                           syncEngine: syncEngine)