In case of errors on some items during sync, no items are saved

[gbitaudeau]

Describe the bug

If during initial sync, AppSync responds with some items on error, these items are null in items response array. The other items which are not in error are not saved in datastore. Here a redacted result for sync Model named Draw with an error on one item:

{
  "data": {
    "syncDraws": {
      "items": [
        {
          "id": "70083950-9b18-4e62-90c6-433ae933def3",
          "drawBatch": {
            "id": "e236042a-7c21-4b82-90b3-d8b73d2c6b99",
            "_deleted": true
          },
          "item": {
            "_deleted": true,
            "id": "fa986c78-ff21-46e2-82ff-8d0a218510bc"
          },
          "owner": "cb8c1cd8-f22c-4e30-8a7c-42fa3e2e16b1",
          "updatedAt": "2023-09-20T09:09:06.134Z"
        },
        null
  },
  "errors": [
    {
      "path": [
        "syncDraws",
        "items",
        17,
        "drawBatch"
      ],
      "data": null,
      "errorType": "Unauthorized",
      "errorInfo": null,
      "locations": [
        {
          "line": 10,
          "column": 7,
          "sourceName": null
        }
      ],
      "message": "Not Authorized to access drawBatch on type Draw"
    }
  ]
}

Steps To Reproduce

Steps to reproduce the behavior:
1. Consider a schema.graphql where 2 objects are linked together with `@hasMany` and `@belongsTo` and each object have the following auth policy: `@auth(rules: [{ allow: owner}])`.
2. Create on object of each type and linked then together.
3. Change the owner of the object with the `@hasMany` relationship.
4. The sync of the object type with `@belongsTo` is no longer working.

Expected behavior

DataStore's initial sync should reconcile partial GraphQL responses by saving the items in the local Database and emitting the errors to the errorHandler.

Amplify Framework Version

2.19.0

Amplify Categories

DataStore

Dependency manager

Swift PM

Swift version

5.9

CLI version

12.4.0

Xcode version

15.0

Relevant log output

No response

Is this a regression?

No

Regression additional context

No response

Platforms

iOS

OS Version

iOS 17.0.1

Device

iPhone

Specific to simulators

No response

Additional context

See discussion #3232 (comment)

[lawmicha]

Hi @gbitaudeau, do you have an redacted version of your schema for us to test with?

[gbitaudeau]

Sorry for the delay, I will make a short project with a simplified schema and add it here as soon as I can.

[thisisabhash]

Great, thanks. Let us know how it goes and if you have any questions.

[gbitaudeau]

Hi,

Here you can find a way to reproduce easily a case where sync failed and no items were saved in datastore.

Please note that to make as short as possible, I changed how to reproduce the problem because I encountered this case due to authorization issue (what I described initially: owner changed on some model linked to other), but it takes more step to reproduce as you need to manage auth, signup, signin...

So using this schema:

# This "input" configures a global authorization rule to enable public access to
# all models in this schema. Learn more about authorization rules here: https://docs.amplify.aws/cli/graphql/authorization-rules
input AMPLIFY { globalAuthRule: AuthRule = { allow: public } } # FOR TESTING ONLY!

type Blog @model {
  id: ID!
  name: String!
  posts: [Post] @hasMany
}

type Post @model {
  id: ID!
  title: String!
  blog: Blog! @belongsTo
  comments: [Comment] @hasMany
}

type Comment @model {
  id: ID!
  post: Post! @belongsTo
  content: String!
}

You execute the following code:

import Foundation
import Amplify
import AWSDataStorePlugin
import AWSAPIPlugin

let task = Task {
    do {
        try await AmplifyConf.instance.start()
        try await AmplifyConf.instance.createNewObjects()
        guard try await AmplifyConf.instance.getComments().count > 0 else {
            fatalError("Comments are empty")
        }
        try await AmplifyConf.instance.makeWrongCommentForeignKey()
        try await AmplifyConf.instance.clearAndStart()
        try await Task.sleep(seconds: 2) // wait sync query to finish
        guard try await AmplifyConf.instance.getComments().count > 0 else {
            fatalError("Comments are empty after clearAndStart") // <== we finish here
        }
    } catch {
        print ("Error: \(error)")
    }
}

public class AmplifyConf {
    public static let instance = AmplifyConf()
    private var listenAmplifyHubTask: Task<Void, Never>?

    private let blog = Blog(id: "THE BLOG", name: "blob")
    
    private init() {
    }
    
    func start() async throws {
//        Amplify.Logging.logLevel = .debug
        
        let datastoreConfig = DataStoreConfiguration.custom { error in
            var dumped = String()
            dump(error, to: &dumped)
            print("ERROR \(error)")
        }
        try Amplify.add(plugin: AWSDataStorePlugin(modelRegistration: AmplifyModels(), 
                                                   configuration: datastoreConfig))
        try Amplify.add(plugin: AWSAPIPlugin(modelRegistration: AmplifyModels()))
        try Amplify.configure()
        
        listenAmplifyHub()
        
        try await clearAndStart()
    }
    
    private func listenAmplifyHub() {
        self.listenAmplifyHubTask = Task {
            for await event in Amplify.Hub.publisher(for: .dataStore).values {
                
                var dumped = String()
                dump(event, to: &dumped)
                print("event \(dumped)")
                
//                print("[Amplify HubEvent](dataStore) \(event.eventName)")
                 
            }
        }
    }
    
    public func clearAndStart() async throws {
        try await Amplify.DataStore.clear()
        try await Amplify.DataStore.start()
    }
    
    public func createNewObjects() async throws {
        let dateStr = "\(Date())"
        let post = Post(title: "post-\(dateStr)", blog: blog)
        let comment1 = Comment(post: post, content: "Content1 \(dateStr)")
        let comment2 = Comment(post: post, content: "Content2 \(dateStr)")
        try await Amplify.DataStore.save(blog)
        try await Amplify.DataStore.save(post)
        try await Amplify.DataStore.save(comment1)
        try await Amplify.DataStore.save(comment2)
    }
    
    public func makeWrongCommentForeignKey() async throws {
        guard var comment = try await Amplify.DataStore.query(Comment.self, sort: .descending(Comment.CodingKeys.createdAt), paginate: .firstResult).first else {
            print ("No comment")
            return
        }
        let post = Post(title: "Not Saved", blog: blog)
        comment.setPost(post)
        let _ = try await Amplify.API.mutate(request: .updateMutation(of: comment, version: 1))
    }
    
    public func getComments() async throws -> [Comment] {
        return try await Amplify.DataStore.query(Comment.self)
    }
}

For me the import part is not the details about makeWrongCommentForeignKey but the fact that after a mutation which make some model objects invalids, no model objects are synchronized at all.

The problem come from how SyncComment query response is managed in Amplify Datastore. Here a sample response:

{
  "data": {
    "syncComments": {
      "items": [
        null,
        {
          "_deleted": null,
          "_lastChangedAt": 1697194380978,
          "_version": 1,
          "content": "Content 2023-10-13 10:53:00 +0000",
          "createdAt": "2023-10-13T10:53:00.976Z",
          "id": "30A91BAD-FFE6-4798-9B12-4110C20BE538",
          "post": {
            "id": "30D0339C-DBA0-4F41-A442-D33C12E4CFB1",
            "_deleted": null
          },
          "updatedAt": "2023-10-13T10:53:00.976Z",
          "postCommentsId": "30D0339C-DBA0-4F41-A442-D33C12E4CFB1"
        }
      ]
    }
  },
  "errors": [
    {
      "path": [
        "syncComments",
        "items",
        0,
        "post"
      ],
      "locations": null,
      "message": "Cannot return null for non-nullable type: 'Post' within parent 'Comment' (/syncComments/items[0]/post)"
    }
  ]
}

Hope you have enough information.

[thisisabhash]

Thank you, we will investigate this and post further updates.

[gbitaudeau]

Hi @thisisabhash ,

In our latest production release, we added more logs to Crashlytics to be able to measure whether or not this bug affects a large number of our users.. An analysis on our data and Crashlytics show us than more than 500 users have the issue.

All affected users have model objects which suffer from a foreign key that targets an object not present in DynamoDB.
They get the following error during initial sync :

▿ DataStoreError: failed to process graphqlResponseData
Caused by:
APIError: failed to process graphqlResponseData
Caused by:
DataStoreError: The key `__typename` was not found
Recovery suggestion: Check if the parsed JSON contains the expected `__typename`
  ▿ api: (2 elements)
    ▿ .0: APIError: failed to process graphqlResponseData
Caused by:
DataStoreError: The key `__typename` was not found
Recovery suggestion: Check if the parsed JSON contains the expected `__typename`
      ▿ operationError: (3 elements)
        - .0: \"failed to process graphqlResponseData\"
        - .1: \"\"
        ▿ .2: Optional(DataStoreError: The key `__typename` was not found
Recovery suggestion: Check if the parsed JSON contains the expected `__typename`)
          ▿ some: DataStoreError: The key `__typename` was not found
Recovery suggestion: Check if the parsed JSON contains the expected `__typename`
            ▿ decodingError: (2 elements)
              - .0: \"The key `__typename` was not found\"
              - .1: \"Check if the parsed JSON contains the expected `__typename`\"
    - .1: nil

This error occurs when sync response contain a content like :

{
  "data": {
    "syncComments": {
      "items": [
        null,
        {
          "_deleted": null,
          "_lastChangedAt": 1697194380978,
          "_version": 1,
          "content": "Content 2023-10-13 10:53:00 +0000",
          "createdAt": "2023-10-13T10:53:00.976Z",
          "id": "30A91BAD-FFE6-4798-9B12-4110C20BE538",
          "post": {
            "id": "30D0339C-DBA0-4F41-A442-D33C12E4CFB1",
            "_deleted": null
          },
          "updatedAt": "2023-10-13T10:53:00.976Z",
          "postCommentsId": "30D0339C-DBA0-4F41-A442-D33C12E4CFB1"
        }
      ]
    }
  },
  "errors": [
    {
      "path": [
        "syncComments",
        "items",
        0,
        "post"
      ],
      "locations": null,
      "message": "Cannot return null for non-nullable type: 'Post' within parent 'Comment' (/syncComments/items[0]/post)"
    }
  ]
}

We investigated to try to find the root cause of why some foreign keys reference missing objects and we were able to find at least 3 possible scenarios (but perhaps there are others ...):

1. Case of an expired Cognito Token in request authorization header which produce an HTTP error code 401
2. Case of a network error which sometimes produce an URLError with the code URLError.Code. secureConnectionFailed
3. Case of a Amplify.DataStore.stop() while the mutations is processed (we think it can happen if application switch from background to foreground and vice versa)

I will describe just after how we were able to reproduce these 3 cases on the schema given here #3259 (comment). I will used the model describe by this schema instead of our real model to simplify.

When one of these error occurs on a create mutation for a post, the post is not saved in server side but is present in local datastore. Restarting the datastore will not send the post mutation. Then if a comment which belongs to this post is created, it is saved in dynamoDB and then sync failed.

To be able to fix sync problem in our application, we took the decision to replace all relations (hasMany , belongsTo , ...) with their foreign keys in the schema.graphql used to generate our model objects. Then we made some logic to compare our own device database with the content of the datastore to create again missing objects in datastore. This is a heavy process and we hope to not lose some data.

Case 1: Expired Cognito Token

We use amazon cognito user pools and all objects have only one auth rule : @auth(rules: [{ allow: owner}]) .
To simulate this error:

• configure cognito token to be valid only during 5 min
• start datastore
• create a post and a comment using:

let post = Post(id:"post-auth", title: "post", blog: blog)
let comment1 = Comment(id:"comment1-auth", post: post, content: "Content-1")
                     
try await Amplify.DataStore.save(post)
try await Amplify.DataStore.save(comment1)

• use a https proxy to rewrite the authorization header on CreatePost mutation with an expired token (took from a previous request made more than 5min ago). Don't rewrite authorization token on CreateMessage mutation.
• start the sync engine again using:

try await Amplify.DataStore.stop()
try await Amplify.DataStore.start()

• the error occurs.

The server return an HTTP error code 401. If it can help, we observed that in method SyncMutationToCloudOperation.getRetryAdviceIfRetryable l261 on

case .httpStatusError(_, let httpURLResponse) where httpURLResponse.statusCode == 401:
            advice = shouldRetryWithDifferentAuthType()

shouldRetryWithDifferentAuthType return a RequestRetryAdvice with shouldRetry set to false.

Case 2: Network secureConnectionFailed error

Simulate this error is quite challenging because the request CreatePost mutation should be stop during the ssl steps and other requests should not be stopped. In our log, this error look like:

Task <C383C0CB-CA8E-4D4B-9218-405A6ABC57CC>.<117> finished with error [-1200] Error Domain=NSURLErrorDomain Code=-1200 "An SSL error has occurred and a secure connection to the server cannot be made." UserInfo={NSErrorFailingURLStringKey=https://c5i3idiurzanjmgvpytfeppfkq.appsync-api.eu-west-3.amazonaws.com/graphql, NSLocalizedRecoverySuggestion=Would you like to connect to the server anyway?, _kCFStreamErrorDomainKey=3, _NSURLErrorFailingURLSessionTaskErrorKey=LocalDataTask <C383C0CB-CA8E-4D4B-9218-405A6ABC57CC>.<117>, _NSURLErrorRelatedURLSessionTaskErrorKey=(
    "LocalDataTask <C383C0CB-CA8E-4D4B-9218-405A6ABC57CC>.<117>"
), NSLocalizedDescription=An SSL error has occurred and a secure connection to the server cannot be made., NSErrorFailingURLKey=https://c5i3idiurzanjmgvpytfeppfkq.appsync-api.eu-west-3.amazonaws.com/graphql, NSUnderlyingError=0x283dfd590 {Error Domain=kCFErrorDomainCFNetwork Code=-1200 "(null)" UserInfo={_kCFStreamPropertySSLClientCertificateState=0, _kCFNetworkCFStreamSSLErrorOriginalValue=-9816, _kCFStreamErrorDomainKey=3, _kCFStreamErrorCodeKey=-9816, _NSURLErrorNWPathKey=satisfied (Path is satisfied), viable, interface: en0[802.11], ipv4, dns, proxy, uses wifi}}, _kCFStreamErrorCodeKey=-9816}

Stopping the connection randomly during a request doesn't always produce this error (but we randomly succeeded to get it severals times). To help, we modify RequestRetryablePolicy.determineRetryRequestAdvice with this version who simulate an urlError code not in the list (as secureConnectionFailed):

private func determineRetryRequestAdvice(basedOn urlError: URLError,
                                             attemptNumber: Int) -> RequestRetryAdvice {
//        switch urlError.code {
//        case .notConnectedToInternet,
//             .dnsLookupFailed,
//             .cannotConnectToHost,
//             .cannotFindHost,
//             .timedOut,
//             .dataNotAllowed,
//             .cannotParseResponse,
//             .networkConnectionLost:
//            let waitMillis = retryDelayInMillseconds(for: attemptNumber)
//            return RequestRetryAdvice(shouldRetry: true, retryInterval: .milliseconds(waitMillis))
//        default:
//            break
//        }
        return RequestRetryAdvice(shouldRetry: false)
    }

• start datastore
• create a post and a comment using:

let post = Post(id:"post-ne", title: "post", blog: blog)
let comment1 = Comment(id:"comment1-ne", post: post, content: "Content-1")
                     
try await Amplify.DataStore.save(post)
try await Amplify.DataStore.save(comment1)

• use a https proxy to abort the CreatePost request
• start the sync engine again using:

try await Amplify.DataStore.stop()
try await Amplify.DataStore.start()

• the error occurs.

In case 1 and 2, the CreateMessage request finish with the following error:

"▿ DataStoreError: GraphQL service returned a successful response containing errors: [Amplify.GraphQLError(message: \"Cannot return null for non-nullable type: \\\'Post\\\' within parent \\\'Comment\\\' (/createComment/post)\", locations: nil, path: Optional([Amplify.JSONValue.string(\"createComment\"), ....

Case 3: DataStore.stop during mutation

Use the following code:

import Foundation
import Amplify
import AWSDataStorePlugin
import AWSAPIPlugin

let task = Task {
    await AmplifyConf.instance.testStartStopLostMutation()
}

public actor AmplifyConf {
    public static let instance = AmplifyConf()
    private var listenAmplifyHubTask: Task<Void, Never>?
    
    private let blog = Blog(id: "THE BLOG", name: "blob")
    
    private var outboxEmpty = true
    
    private init() {
    }
    
    func start() async throws {
        //Amplify.Logging.logLevel = .debug
        
        let datastoreConfig = DataStoreConfiguration.custom(errorHandler: { error in
            var dumped = String()
            dump(error, to: &dumped)
            print("ERROR \(error)")
        })
        
        try Amplify.add(plugin: AWSDataStorePlugin(modelRegistration: AmplifyModels(),
                                                   configuration: datastoreConfig))
        try Amplify.add(plugin: AWSAPIPlugin(modelRegistration: AmplifyModels()))
        try Amplify.configure()
        
        listenAmplifyHub()
        
//        try await Amplify.DataStore.clear()
        try await Amplify.DataStore.start()
    }
    
    func testStartStopLostMutation() async {
        do {
            try await start()
            print(">>> wait blog")
            let blog = try await subscribeToBlog()
            
            let j = Int16.random(in: 0...Int16.max)
            
            for i in 0...100 {
                print(">>> for n°\(i)")
                let post = Post(id:"post-\(j)-\(i)", title: "post-\(j)-\(i)", blog: blog)
                let comment1 = Comment(id:"comment1-\(j)-\(i)", post: post, content: "Content-1-\(j)-\(i)")
                
                self.outboxEmpty = false
                let task = Task {
                    Task.detached {
                        try await Amplify.DataStore.stop()
                        try await Amplify.DataStore.start()
                    }
                    Task.detached {
                        try await Amplify.DataStore.save(post)
                        try await Amplify.DataStore.save(comment1)
                    }
                }
                let _ = await task.result
                    
                while self.outboxEmpty == false {
                    try await Task.sleep(seconds: 0.01)
                }
            }
        } catch {
            print ("ERROR: \(error)")
        }
    }
    
    // You must hold a reference to your subscription.
    var blogSubscription: AmplifyAsyncThrowingSequence<DataStoreQuerySnapshot<Blog>>?

    // Subscribe
    func subscribeToBlog() async throws -> Blog {
        let blogSubscription = Amplify.DataStore.observeQuery(
            for: Blog.self,
            where: Blog.self.keys.id.eq("THE BLOG")
        )
        
        // hold onto your subscription
        self.blogSubscription = blogSubscription
        // observe new snapshots
        for try await blogSnapshot in blogSubscription {
            if let blog = blogSnapshot.items.first {
                self.blogSubscription = nil
                return blog
            }
        }
        
        fatalError("No blog")
    }

    
    private func listenAmplifyHub() {
        self.listenAmplifyHubTask = Task {
            for await event in Amplify.Hub.publisher(for: .dataStore).stream {
                switch event.eventName {
                case HubPayload.EventName.DataStore.outboxStatus:
                    if let outboxEvent = event.data as? OutboxStatusEvent {
                        self.outboxEmpty = outboxEvent.isEmpty
                        print(">>> outboxEmpty \(outboxEmpty)")
                    }
                default:
                    break
                }
            }
        }
    }
}

• add a breakpoint inside DataStoreConfiguration.custom(errorHandler
• launch it on a device
• when a comment is created but its post is missing, the breakpoint should pause the process

[gbitaudeau]

I forgot to mention that we use Amplify 2.23.0 for latest production release.

[thisisabhash]

Hello @gbitaudeau , Thank you for the detailed information on the ticket. We are investigating this issue with priority.

[5d]

Hi @gbitaudeau ,

I've implemented a solution in the 5d/partial-sync branch. Could you please test it out when you have time?

[gbitaudeau]

Thanks @5d , I will take a look when I can.

Just to resume, there are 3 parts in this problem:

1. Make the sync engine ready even if there are errors in the sync response (which is the purpose of your branch I think). In our case, as we change the model to remove relations (hasMany, belongsTo, ...), this is not our first priority.
2. Ensure that mutations that are not passed server-side are not lost. As long as this is the case, Datastore and DynamoDB will continue to diverge and the number of objects with wrong foreign key will increase. This is what we are waiting the most.
3. Find how we can send again missing objects on the server and cleanup object with an incorrect foreign key that are no longer needed on the server. We've already worked on this and were able to build logic which covers send again missing objects. We can't find a good cleaning solution at the moment. This is our lowest priority.