DataStore.stop or DataStore.clear may interrupt DataStore.start

[HuiSF]

Before opening, please confirm:

• I have searched for duplicate or closed issues and discussions.

Language and Async Model

Java

Amplify Categories

DataStore

Gradle script dependencies

// Put output below this line
dependencies {
    implementation "com.amplifyframework:aws-datastore:1.32.2"
    implementation "com.amplifyframework:aws-api-appsync:1.32.2"
}

Environment information

------------------------------------------------------------
Gradle 6.7
------------------------------------------------------------

Build time:   2020-10-14 16:13:12 UTC
Revision:     312ba9e0f4f8a02d01854d1ed743b79ed996dfd3

Kotlin:       1.3.72
Groovy:       2.5.12
Ant:          Apache Ant(TM) version 1.10.8 compiled on May 10 2020
JVM:          1.8.0_292 (AdoptOpenJDK 25.292-b10)
OS:           Mac OS X 10.16 x86_64

Please include any relevant guides or documentation you're referencing

No response

Describe the bug

Original amplify-flutter issue: aws-amplify/amplify-flutter#1479

Using the code example provided in this document.

    Amplify.DataStore.clear(
        {
            Amplify.DataStore.start(
                { Log.i("MyAmplifyApp", "DataStore started") },
                { Log.e("MyAmplifyApp", "Error starting DataStore", it) }
            )
        },
        { Log.e("MyAmplifyApp", "Error clearing DataStore", it) }
    )

the start process may be interrupted by the clear call due to below error

W/amplify:aws-api: Thread interrupted awaiting subscription acknowledgement.
    java.lang.InterruptedException
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedNanos(AbstractQueuedSynchronizer.java:1081)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(AbstractQueuedSynchronizer.java:1369)
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:278)
        at com.amplifyframework.api.aws.SubscriptionEndpoint$Subscription.awaitSubscriptionReady(SubscriptionEndpoint.java:381)
        at com.amplifyframework.api.aws.SubscriptionEndpoint.requestSubscription(SubscriptionEndpoint.java:183)

It looks like that the onComplete action provided to the stop and the clear APIs is not guaranteed to be executed after stop and clear is completed finished.

Reproduction steps (if applicable)

1. using code example listed above
2. execute the logic
3. the start may fail (not 100% failing, but mostly)

Code Snippet

// Put your code below this line.

Log output

// Put your logs below this line

I/amplify:aws-datastore: Orchestrator lock acquired.
I/amplify:aws-datastore: Orchestrator transitioning from SYNC_VIA_API to STOPPED
I/amplify:aws-datastore: Setting currentState to LOCAL_ONLY
I/amplify:aws-datastore: Stopping subscription processor.
I/amplify:aws-datastore: Stopped subscription processor.
I/amplify:aws-datastore: Stopping observation of local storage changes.
I/amplify:aws-datastore: Setting currentState to STOPPED
I/amplify:flutter:datastore: Successfully stopped datastore remote synchronization
I/amplify:aws-datastore: Orchestrator lock released.
I/amplify:aws-datastore: Orchestrator lock acquired.
I/amplify:aws-datastore: Orchestrator transitioning from STOPPED to SYNC_VIA_API
I/amplify:aws-datastore: Starting to observe local storage changes.
D/EGL_emulation: app_time_stats: avg=8013.48ms min=8.68ms max=23772.65ms count=3
I/amplify:aws-datastore: Now observing local storage. Local changes will be enqueued to mutation outbox.
I/amplify:aws-datastore: Setting currentState to LOCAL_ONLY
I/amplify:aws-datastore: Setting currentState to SYNC_VIA_API
I/amplify:flutter:datastore: Successfully started datastore remote synchronization
I/amplify:aws-datastore: Starting API synchronization mode.
I/amplify:aws-datastore: Orchestrator lock released.
W/_triaging_marc: Long monitor contention with owner pool-4-thread-3 (19253) at void com.amplifyframework.api.aws.SubscriptionEndpoint.releaseSubscription(java.lang.String)(SubscriptionEndpoint.java:291) waiters=1 in void com.amplifyframework.api.aws.SubscriptionEndpoint.releaseSubscription(java.lang.String) for 377ms
I/flutter: outboxStatus
I/amplify:aws-datastore: Starting processing subscription events.
I/amplify:aws-api: No more active subscriptions. Closing web socket.
W/amplify:aws-api: Thread interrupted awaiting subscription acknowledgement.
    java.lang.InterruptedException
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedNanos(AbstractQueuedSynchronizer.java:1081)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(AbstractQueuedSynchronizer.java:1369)
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:278)
        at com.amplifyframework.api.aws.SubscriptionEndpoint$Subscription.awaitSubscriptionReady(SubscriptionEndpoint.java:381)
        at com.amplifyframework.api.aws.SubscriptionEndpoint.requestSubscription(SubscriptionEndpoint.java:183)
        at com.amplifyframework.api.aws.SubscriptionOperation.lambda$start$2$SubscriptionOperation(SubscriptionOperation.java:77)
        at com.amplifyframework.api.aws.-$$Lambda$SubscriptionOperation$dV6kPdbE8Z12Qd0_rE-16cujgUk.run(Unknown Source:2)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:463)
        at java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1137)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:637)
        at java.lang.Thread.run(Thread.java:1012)
W/_triaging_marc: Long monitor contention with owner pool-4-thread-2 (19252) at void com.amplifyframework.api.aws.SubscriptionEndpoint.releaseSubscription(java.lang.String)(SubscriptionEndpoint.java:291) waiters=2 in void com.amplifyframework.api.aws.SubscriptionEndpoint.requestSubscription(com.amplifyframework.api.graphql.GraphQLRequest, com.amplifyframework.api.aws.AuthorizationType, com.amplifyframework.core.Consumer, com.amplifyframework.core.Consumer, com.amplifyframework.core.Consumer, com.amplifyframework.core.Action) for 1.951s
Disconnected from the target VM, address: 'localhost:65258', transport: 'socket'

amplifyconfiguration.json

No response

GraphQL Schema

// Put your schema below this line

Additional information and screenshots

No response

[wojciechzahradnikdeviniti]

Hi, any updates on this issue? I still have a problem with that.

[wojciechzahradnikdeviniti]

Any updates?

[mattcreaser]

I'll take a look at this issue.

[mattcreaser]

@wojciechzahradnikdeviniti can I ask what version of Amplify you're using? We've made a number of changes around this behaviour recently.

[wojciechzahradnikdeviniti]

@wojciechzahradnikdeviniti can I ask what version of Amplify you're using? We've made a number of changes around this behaviour recently.

@mattcreaser I used 1.3.0 - 1.3.3 flutter versions of the lib and this bug occurs there

[mattcreaser]

Flutter 1.3.3 uses Android 2.11.2.

It's possible that #2551 addressed this issue in 2.11.3. I'll see if I am able to reproduce it in either version.

[mattcreaser]

This is reproducible in 2.11.3.

The issue occurs because stop will cancel any active subscriptions. This is an asynchronous process that has multiple levels of fire-and-forget operations, so stop does not wait for the subscriptions to be cancelled before calling its callback and having start begin establishing new subscriptions. If none of the new subscriptions have been fully established before the last one of the "old" subscriptions complete, then we fall into this code that closes the websocket out from underneath the in-progress "new" subscriptions.

It appears from my testing that typically websocket.close() is called before we begin to establish new subscriptions, and the issue is that the graceful close of the websocket is not yet complete when the new subscriptions are starting up. Therefore webSocketListener.isDisconnectedState() returns false causing the new subscriptions to use the currently-shutting-down websocket.

A simple fix is to set the webSocketListener to null when closing the websocket. This will cause the next subscription to create a new websocket, even if the prior socket is still in the process of closing. I've verified that I can't reproduce the issue with this change. We can also add locking to the SubscriptionEndpoint to try to avoid race conditions around the precise timing of the subscription calls.

[mattcreaser]

This should be fixed in Amplify Android 2.12.0. Please note that the fix only addresses the use case of calling stop/clear and then start once:

Amplify.DataStore.stop(
        {
            Amplify.DataStore.start(
                { Log.i("MyAmplifyApp", "DataStore started") },
                { Log.e("MyAmplifyApp", "Error starting DataStore", it) }
            )
        },
        { Log.e("MyAmplifyApp", "Error clearing DataStore", it) }
    )

It does not address possible issues that may arise with calling stop/clear and start multiple times each in a short timeframe.

[github-actions]

⚠️COMMENT VISIBILITY WARNING⚠️

Comments on closed issues are hard for our team to see.
If you need more assistance, please either tag a team member or open a new issue that references this one.
If you wish to keep having a conversation with other community members under this issue feel free to do so.