feat(datastore): Dispatch outboxStatus, subscriptionsEstablished, syncQueriesStarted events

[ruiguoamz]

Issue #, if available:

The rest of 6 events will be implemented in other PRs.

• Network Status
• modelSynced
• syncQueriesReady
• ready
• outboxMutationEnqueued
• outboxMutationProcessed

Description of changes:

Implemented three of nine hub events

• outboxStatus
• syncQueriesStarted
• subscriptionEstablished

outboxStatus

{"isEmpty": true/false}

public struct {
    let isEmpty: Bool
}

subscriptionsEstablished

N/A

When all subscriptions have been established, 3 per model, for every model, then emit a single subscriptionsEstablished event

syncQueriesStarted

{"models":["Task","Note"]}

struct SyncQueriesStartedEvent {
    let models: [String]
}

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

[lawmicha]

review for syncQueriesReady

[lawmicha]

review subscriptionsEstablished

[lawmicha]

partial review for outboxStatus

[lawmicha]

continued review for outboxStatus

[lawmicha]

can update title to be more explicit on the hub events being added?
[DataStore] Triggering hub events -> feat(datastore): Dispatch outboxStatus, subscriptionsEstablished, syncQueriesStarted events

[lawmicha]

One approach i'm thinking is in OutgoingMutationQueue, there is a state that is starting and a responder method that is start(api:mutationEventPublisher:). you can split this into two responder methods

when in .starting state, respond to it with checkStatus() responder method. In checkStatus, you can have the logic

func checkStatus() {
   // use storageAdapter and query for MutationEvents
   // dispatchOutBoxStatusEvent(isEmpty: events.isEmpty)
   stateMachine.notify(action: .dispatchedStatus)
}

Then (.starting, .dispatchedStatus) can resolve to the state .subscribing and triggers subscribe(api:mutationEvent)

/// note here that the old `start()` was replaced with this new `subscribe()` responder method, but content the same
private func subscribe(api: APICategoryGraphQLBehavior,
                       mutationEventPublisher: MutationEventPublisher) {
        log.verbose(#function)
        self.api = api
        operationQueue.isSuspended = false

        // State machine notification to ".receivedSubscription" will be handled in `receive(subscription:)`
        mutationEventPublisher.publisher.subscribe(self)
    }

This approach goes straight to introducing a new state and responder method, however you can try it out by putting checkStatus() as a call inside the existing start() logic, and you may find that we don't gain that much from introducing a new state in the state machine. the important part here is to get the actual number of mutation events to be processed at DataStore start, is it from storage adapter? is there a way do some sort of peek operation on the mutationEventPublisher? Once you have the actual number of mutation events to be processed, then you can emit the outbox status event with an accurate isEmpty field

[lawmicha]

can you add integration test that covers recieving the new events you are emitting in this PR? for outboxStatus event, since it's emitted one or more times, you can just make sure the expectation is fulfilled without failing on overfulfillment. the other two (subscriptionsEstblished and synQueriesStarted) it should probably be just a single count each

[lawmicha]

Are there any unit tests that you can improve or add?

• For example, InitialSyncOrchestratorTests.testInvokesCompletionCallback can probably be improved by adding a hub listener for the event that you are emitted and asserting that the model names match the expected syncable models

• OutgoingMutationQueueTests

• OutgoingMutationQueueMockStateTest

• RemoteSyncEngineTests, RemoteSyncEngineTests.testRemoteSyncEngineHappyPath, etc

[lawmicha]

looks good to me, question about naming for outboxStatus. Should we stick with JS/Android or if we should stick to something more common in iOS like outgoingMutationStatus?

[palpatim]

looks good to me, question about naming for outboxStatus. Should we stick with JS/Android or if we should stick to something more common in iOS like outgoingMutationStatus?

@lawmicha We need to keep the names the same for platform consistency. We can revisit naming, event dispatch rules, payloads, etc in the future.

[palpatim]

Clarify wording: "Dispatched when: 1) the DataStore starts; 2) each time a local mutation is enqueued into the outbox; 3) each time a local mutation is finished processing."

[ruiguoamz]

Updated

[palpatim]

"Dispatched when DataStore has finished establishing its subscriptions to all syncable models"

[ruiguoamz]

Updated

[palpatim]

"Dispatched when DataStore starts establishing its subscriptions to all syncable models"

[ruiguoamz]

For syncQueriesStarted, it is not Dispatched when DataStore starts establishing its subscriptions to all syncable models but is Dispatched when DataStore starts to sync Queries

[palpatim]

Unnecessary import

[ruiguoamz]

Removed

[palpatim]

"true if there are no events in the outbox at the time the event was dispatched

[palpatim]

Unnecessary import

[ruiguoamz]

Removed

[palpatim]

"A list of all model names for which DataStore has started establishing subscriptions"

[ruiguoamz]

Updated

[palpatim]

s/quering/querying/

[ruiguoamz]

updated

[palpatim]

How do you know the outbox is empty here? Shouldn't you be querying the mutation events from storage, or checking the operationCount of the outbox's operation queue?

[ruiguoamz]

Right, it makes sense.
Incorrectly, I thought because subscription.request(.max(1)) ensure the queue to have only one mutationEvent at a time so I can assert queue is empty when the event is finish processing. I miss the fact that events are stored inside the mutationEventDatabase.
Updated.

[lawmicha]

As i understand it, we were emitting outboxStatus events to this logic (this is probably wrong then)

• when the remote sync engine starts, check if there are mutation events from storage to determine isEmpty
• when a mutationEvent is being enqueued, we get this from the mutationEventPublisher, one mutationEvent at a time. hardcode isEmpty to false since we know we are processing one
• when the mutationEvent is done being processed, we emit a following outboxStatus event with isEmpty = true

example output
outBoxStatus - isEmpty = false - on start, check mutationEvent from storage, say there 5 mutation events
outboxStatus - isEmpty = false - queue mutationEvent 1
outboxStatus - isEmpty = true - finish processing mutationEvent 1
outboxStatus - isEmpty = false - queue mutationEvent 2
outboxStatus - isEmpty = true - finish processing mutationEvent 2
outboxStatus - isEmpty = false - queue mutationEvent 3
outboxStatus - isEmpty = true - finish processing mutationEvent 3
outboxStatus - isEmpty = false - queue mutationEvent 4
outboxStatus - isEmpty = true - finish processing mutationEvent 4
outboxStatus - isEmpty = false - queue mutationEvent 5
outboxStatus - isEmpty = true - finish processing mutationEvent 5

[lawmicha]

or is the expected behavior just a ping of the outbox status? for example
outBoxStatus - isEmpty = false - on start, check mutationEvent from storage, say there 5 mutation events
outboxStatus - isEmpty = false - queue mutationEvent 1
outboxStatus - isEmpty = false - finish processing mutationEvent 1
outboxStatus - isEmpty = false - queue mutationEvent 2
outboxStatus - isEmpty = false - finish processing mutationEvent 2
outboxStatus - isEmpty = false - queue mutationEvent 3
outboxStatus - isEmpty = false - finish processing mutationEvent 3
outboxStatus - isEmpty = false - queue mutationEvent 4
outboxStatus - isEmpty = false - finish processing mutationEvent 4
outboxStatus - isEmpty = false - queue mutationEvent 5
outboxStatus - isEmpty = true - finish processing mutationEvent 5

[manueliglesias]

@lawmicha the second behavior is the expected one.

For reference, this is the JS implementation:

https://github.com/aws-amplify/amplify-js/blob/a7feacea4ed506340d250249d0b15286fe3ef5fa/packages/datastore/src/sync/index.ts#L358-L365

It checks the queue