Amplify-js: Race condition in DataStore when reachability is switching back and forth

[nubpro]

Describe the bug
I actually notice this bug on production where I would see duplicate subscriptions of the same model but had no idea why causes it. I believe it happens because of unstable network connection which triggered Reachability on and off repeatedly.

I poked around with some testing and find it pretty easy to reproduce.

To Reproduce
Steps to reproduce the behavior:

1. On Client A, spam reachability on and off.
2. On Client B, add new item.
3. On Client A, you would get 2 DataStore.observe(), one being an INSERT and UPDATE. This is because there's two active subscriptions on onCreatePost.

Here's a video: https://streamable.com/iankoh

Expected behavior
At any time, there should be only 1 active subscription per subscription (onCreatePost).

Code Snippet
schema.graphql

enum PostStatus {
    ACTIVE
    INACTIVE
}

type Post @model {
    id: ID!
    title: String!
    rating: Int!
    status: PostStatus!
}

App.js

import React, { useEffect } from 'react';
import './App.css';

import { Hub, Reachability } from '@aws-amplify/core';
import { DataStore, Predicates } from '@aws-amplify/datastore';
import { Post, PostStatus } from './models';
import { withAuthenticator, AmplifySignOut } from '@aws-amplify/ui-react';

window.LOG_LEVEL = 'DEBUG';

function onCreate() {
    DataStore.save(
        new Post({
            title: `New title ${Date.now()}`,
            rating: (function getRandomInt(min, max) {
                min = Math.ceil(min);
                max = Math.floor(max);
                return Math.floor(Math.random() * (max - min)) + min;
            })(1, 7),
            status: PostStatus.ACTIVE,
        })
    );
}

function onDeleteAll() {
    DataStore.delete(Post, Predicates.ALL);
}

async function onQuery() {
    const posts = await DataStore.query(Post, (c) => c.rating('gt', 4));

    console.log(posts);
}

function App() {
    useEffect(() => {
        const subscription = DataStore.observe(Post).subscribe((msg) => {
            console.warn('Post subscription:', msg.opType, msg.element);
        });
        return () => subscription.unsubscribe();
    }, []);

    useEffect(() => {
        // Create listener
        const listener = Hub.listen('datastore', async (hubData) => {
            const { event, data } = hubData.payload;

            console.log(event, data);
        });

        // Remove listener
        return () => {
            listener();
        };
    }, []);

    return (
        <div className="App">
            <header className="App-header">
                <div>
                    <input type="button" value="NEW" onClick={onCreate} />

                    <input
                        type="button"
                        value="Reachability off"
                        onClick={async () => {
                            Reachability._observerOverride({
                                online: false,
                            });
                        }}
                    />

                    <input
                        type="button"
                        value="Reachability on"
                        onClick={async () => {
                            Reachability._observerOverride({
                                online: true,
                            });
                        }}
                    />

                    <AmplifySignOut />
                </div>
            </header>
        </div>
    );
}

export default withAuthenticator(App);

Environment

  System:
    OS: Windows 10 10.0.17763
    CPU: (12) x64 AMD Ryzen 5 3600 6-Core Processor
    Memory: 4.79 GB / 15.95 GB
  Binaries:
    Node: 14.15.1 - C:\Program Files\nodejs\node.EXE
    Yarn: 1.22.4 - C:\Program Files (x86)\Yarn\bin\yarn.CMD
    npm: 6.14.8 - C:\Program Files\nodejs\npm.CMD
    Watchman: 20200509.222254.0 - C:\Users\Chai\AppData\Local\watchman\watchman.EXE
  Browsers:
    Chrome: 86.0.4240.198
    Edge: Spartan (44.17763.831.0)
    Internet Explorer: 11.0.17763.771
  npmPackages:
    @aws-amplify/ui-react: ^0.2.28 => 0.2.28
    @testing-library/jest-dom: ^5.11.4 => 5.11.6
    @testing-library/react: ^11.1.0 => 11.1.2
    @testing-library/user-event: ^12.1.10 => 12.2.2
    aws-amplify: ^3.3.8 => 3.3.8
    react: ^17.0.1 => 17.0.1
    react-dom: ^17.0.1 => 17.0.1
    react-scripts: 4.0.0 => 4.0.0
    web-vitals: ^0.2.4 => 0.2.4
  npmGlobalPackages:
    @angular/cli: 8.1.3
    @aws-amplify/cli: 4.32.1
    @ionic/angular: 4.10.0
    cordova-res: 0.8.0
    cordova: 9.0.0
    create-react-app: 3.4.1
    create-react-native-app: 3.4.0
    firebase-tools: 7.6.2
    ionic: 5.4.5
    my-project: 0.0.1
    native-run: 0.2.8
    nodemon: 1.19.4
    react-devtools: 4.6.0
    react-native-cli: 2.0.1
    rimraf: 3.0.2
    serve: 11.2.0
    serverless: 1.62.0
    typescript: 4.0.5
    webextension-toolbox: 3.0.0

Smartphone (please complete the following information):

• Browser: Chrome, New Edge

[nubpro]

Here's a reliable way of reproducing the bug. No clicking involved.

   useEffect(() => {
        let postSubscription;

        // Create listener
        const listener = Hub.listen('datastore', async (hubData) => {
            const { event, data } = hubData.payload;

            console.log(event, data);

            if (event == 'storageSubscribed') {
                postSubscription = DataStore.observe(Post).subscribe((msg) => {
                    console.warn('Post subscription:', msg.opType, msg.element);
                });
            }

            if (event == 'ready') {
                Reachability._observerOverride({
                    online: false,
                });

                await timeout(10);

                Reachability._observerOverride({
                    online: true,
                });

                await timeout(10);

                Reachability._observerOverride({
                    online: false,
                });

                await timeout(10);

                Reachability._observerOverride({
                    online: true,
                });
            }
        });

        // Remove listener
        return () => {
            listener();

            if (postSubscription) {
                postSubscription.unsubscribe();
            }
        };
    }, []);

    function timeout(ms) {
        return new Promise((resolve) => setTimeout(resolve, ms));
    }

[iartemiev]

Reproduced this on macOS as well. Also, confirmed that the linked PR fixes this issue

[nubpro]

Reproduced this on macOS as well. Also, confirmed that the linked PR fixes this issue

Thanks for the reviewing my PR!
Since this is now fixed, I will be closing this