fix(data): Resolve orchestrator transition threading issues.

[tylerjroach]

• PR title and description conform to Pull Request guidelines.

Issue #, if available:
aws-amplify/amplify-flutter#3017

Description of changes:

This PR intends to fix the crash caused by the UnicastSubject "buffer" in the SubscriptionProcessor being subscribed to more than once. I believe there were 2 root causes to this.

• The reachabilityMonitor subscriber was directly calling startApiSync and stopApiSync rather than use the transitionTo methods. The transitionTo methods read the current state, and ignore calls if the current state equals the requested state. I believe the crash was due to startApiSync being called multiple times, and through a race condition, there was a scenario that could allow the UnicastSubject to receive more than 1 subscription.
• When stopApiSync is called, the completable that is started in startApiSync does not stop by simply clearing the disposable. As a result, a call to start/stop/start could still result in a single buffer being subscribed to more than once. A check to stop additional processing was added in the startApiSync completable, if the disposable was cleared.

Lastly, I refactored the currentState.set calls to only be made inside the transitionTo api methods and added a lock for thread safety.

How did you test these changes?
(Please add a line here how the changes were tested)

Documentation update required?

• No
• Yes (Please include a PR link for the documentation update)

General Checklist

• Added Unit Tests
• Added Integration Tests
• Security oriented best practices and standards are followed (e.g. using input sanitization, principle of least privilege, etc)

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[mattcreaser]

Issue #2458 is also referencing this same problem.

[mattcreaser]

🤔 There are definitely some good improvements here - it's easy to see how the prior direct invocation of startApiSync could cause a crash when network status changed more quickly than the total sync time - but I am skeptical that the !emitter.isDisposed() check is sufficient to completely fix the issue. It seems that, although the door is mostly closed, it is still theoretically possible for multiple calls to startDrainingMutationBuffer to interleave.

What about wrapping the contents of the completable in a lock, such that we can guarantee that no threads can interleave between subscriptionProcessor.startSubscriptions() and subscriptionProcess.startDrainingMutationBuffer()? We could also use the same lock in stopApiSync() to ensure that the stopping does not happen before the starting is complete. this part is a bad idea

Another option would be to use a single-threaded scheduler instead of Schedulers.io() to run the startApiSync completable. That has the downside of permanently reserving a single thread for this purpose however.

[codecov-commenter]

Codecov Report

Merging #2551 (b6515e8) into main (79b7b47) will increase coverage by 0.02%.
The diff coverage is 61.17%.

❗ Your organization is not using the GitHub App Integration. As a result you may experience degraded service beginning May 15th. Please install the Github App Integration for your organization. Read more.

@@            Coverage Diff             @@
##             main    #2551      +/-   ##
==========================================
+ Coverage   40.79%   40.82%   +0.02%     
==========================================
  Files         873      873              
  Lines       27597    27621      +24     
  Branches     3877     3878       +1     
==========================================
+ Hits        11258    11276      +18     
- Misses      15135    15139       +4     
- Partials     1204     1206       +2     

[ankpshah]

Issue #1010 might also be referencing same problem