fix(datastore): Continue initial sync if atleast one model syncs successfully and other models fail due to Unauthorized API error

AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/InitialSync/InitialSyncOrchestrator.swift

@@ -164,6 +164,7 @@ final class AWSInitialSyncOrchestrator: InitialSyncOrchestrator {
             allMessages.joined(separator: "\n"),
             underlyingError
         )
+
         return .failure(syncError)
     }
 
@@ -245,6 +246,22 @@ extension AWSInitialSyncOrchestrator {
            case .unauthorized = AppSyncErrorType(errorTypeValue) {
             return true
         }
+
+        // Check is API error is of unauthorized type
+        if case let .api(amplifyError, _) = datastoreError,
+            let apiError = amplifyError as? APIError {
+            if case .operationError(let errorDescription, _, _) = apiError,
+               errorDescription.range(of: "Unauthorized",
+                                      options: .caseInsensitive) != nil {
+                return true
+            }
+
+            if case .httpStatusError(let statusCode, _) = apiError,
+               (statusCode == 401 || statusCode == 403) {
+                return true
+            }
+        }
+
         return false
     }
 

AmplifyPlugins/DataStore/Tests/AWSDataStorePluginTests/Sync/InitialSync/InitialSyncOrchestratorTests.swift

@@ -6,6 +6,7 @@
 //
 
 import XCTest
+import Foundation
 
 @testable import Amplify
 @testable import AmplifyTestCommon
@@ -101,7 +102,7 @@ class InitialSyncOrchestratorTests: XCTestCase {
         Amplify.Hub.removeListener(hubListener)
         sink.cancel()
     }
-
+    
     /// - Given: An InitialSyncOrchestrator with a model dependency graph, API is expected to return an error for certain models
     /// - When:
     ///    - The orchestrator starts up
@@ -201,6 +202,112 @@ class InitialSyncOrchestratorTests: XCTestCase {
         sink.cancel()
     }
 
+    /// - Given: An InitialSyncOrchestrator with a model dependency graph
+    ///          API is expected to return an unauthorized error for certain models
+    ///          and success for remaining models
+    /// - When:
+    ///    - The orchestrator starts up
+    /// - Then:
+    ///    - It invokes completion callback with success
+    func testInvokesCompletionCallbackWhenUnauthorizedAPIError() async throws {
+        ModelRegistry.reset()
+        PostCommentModelRegistration().registerModels(registry: ModelRegistry.self)
+        let responder = QueryRequestListenerResponder<PaginatedList<AnyModel>> { request, listener in
+            if request.document.contains("SyncPosts") {
+                let urlResponse = HTTPURLResponse(url: URL(string: "https://aws.amazon.com")!,
+                                                 statusCode: 401,
+                                                 httpVersion: nil,
+                                                 headerFields: nil)!
+                let event: GraphQLOperation<PaginatedList<AnyModel>>.OperationResult =
+                    .failure(APIError.httpStatusError(401, urlResponse))
+                listener?(event)
+            } else if request.document.contains("SyncComments") {
+                let startedAt = Int(Date().timeIntervalSince1970)
+                let list = PaginatedList<AnyModel>(items: [], nextToken: nil, startedAt: startedAt)
+                let event: GraphQLOperation<PaginatedList<AnyModel>>.OperationResult = .success(.success(list))
+                listener?(event)
+            }
+
+            return nil
+        }
+
+        let apiPlugin = MockAPICategoryPlugin()
+        apiPlugin.responders[.queryRequestListener] = responder
+
+        let storageAdapter = MockSQLiteStorageEngineAdapter()
+        storageAdapter.returnOnQueryModelSyncMetadata(nil)
+
+        let reconciliationQueue = MockReconciliationQueue()
+
+        let orchestrator: AWSInitialSyncOrchestrator =
+        AWSInitialSyncOrchestrator(dataStoreConfiguration: .default,
+                                   authModeStrategy: AWSDefaultAuthModeStrategy(),
+                                   api: apiPlugin,
+                                   reconciliationQueue: reconciliationQueue,
+                                   storageAdapter: storageAdapter)
+
+        let syncCallbackReceived = expectation(description: "Sync callback received, sync operation is complete")
+        let syncQueriesStartedReceived = expectation(description: "syncQueriesStarted received")
+
+        let filter = HubFilters.forEventName(HubPayload.EventName.DataStore.syncQueriesStarted)
+        let hubListener = Amplify.Hub.listen(to: .dataStore, isIncluded: filter) { payload in
+            guard let syncQueriesStartedEvent = payload.data as? SyncQueriesStartedEvent else {
+                XCTFail("Failed to cast payload data as SyncQueriesStartedEvent")
+                return
+            }
+            XCTAssertEqual(syncQueriesStartedEvent.models.count, 2)
+            syncQueriesStartedReceived.fulfill()
+        }
+
+        guard try await HubListenerTestUtilities.waitForListener(with: hubListener, timeout: 5.0) else {
+            XCTFail("Listener not registered for hub")
+            return
+        }
+
+        let syncStartedReceived = expectation(description: "Sync started received, sync operation started")
+        syncStartedReceived.expectedFulfillmentCount = 2
+        let finishedReceived = expectation(description: "InitialSyncOperation finished paginating and offering")
+        finishedReceived.expectedFulfillmentCount = 2
+        let successCompletionReceived = expectation(description: "InitialSyncOrchestrator completed with success")
+        let sink = orchestrator
+            .publisher
+            .sink(receiveCompletion: { completion in
+                switch completion {
+                case .finished:
+                    successCompletionReceived.fulfill()
+                case .failure:
+                    XCTFail("Should have finished with success")
+                }
+            }, receiveValue: { value in
+                switch value {
+                case .started:
+                    syncStartedReceived.fulfill()
+                case .finished(let modelName, let error):
+                    if modelName == "Post" {
+                        guard case .api(let apiError, _) = error,
+                              case .httpStatusError(_, _) = apiError as? APIError else {
+                            XCTFail("Should be api error")
+                            return
+                        }
+                    } else if modelName == "Comment" {
+                        XCTAssertNil(error)
+                    }
+                    finishedReceived.fulfill()
+                default:
+                    break
+                }
+            })
+
+        orchestrator.sync { _ in
+            syncCallbackReceived.fulfill()
+        }
+
+        await waitForExpectations(timeout: 1)
+        XCTAssertEqual(orchestrator.syncOperationQueue.maxConcurrentOperationCount, 1)
+        Amplify.Hub.removeListener(hubListener)
+        sink.cancel()
+    }
+
     /// - Given: An InitialSyncOrchestrator with a model dependency graph containing no associations
     /// - When:
     ///    - The orchestrator starts up