Replication observation mode ignored when push handler is waiting for response from backend

[emilhdiaz]

Have you found a bug?

Yes, a race condition that is not possible to replicate with a unit test alone. Another user in Discord noted a similar issue and I have confirmed the issue by placing breakpoints in code to simulate the desired order of operations.

1. Client sends a push request to the server
2. Server handles the request, updates some document fields, and publishes updated document via WS message
3. Client receives WS message and adds it to the pull stream
4. Client receives response for the push request

The result of that:

• Documents supplied to the pull stream are not saved
• Checkpoint provided to the pull stream is saved
• Any updates made by the server while handling the request (i.e. setting updatedAt server side) are ignored by the local db
• Impossible to recover via checkpoint iteration as latest checkpoint is sent to server yielding no new documents from server

Root cause

It appears that documents emitted to the pull stream, while the push handler is still waiting for a response from the server, are ignored and never applied to the local db. I have to imagine this has to do with some sort of locking mechanism that the collection employs during the push operation. Unless I missed a critical piece of the documentation... this issue is a possible race condition in the normal replication workflow.

Do you have a feature request?

Would be happy to contribute a PR with a small nudge in the right direction (just need a general area of the code start from).

[emilhdiaz]

Further confirming my suspicion, if I await awaitInSync() before placing the WS events on the pull stream the updates are correctly sequenced AFTER the push call returns and system behaves as expected.

// > client.ts
import { Subject } from 'rxjs';

const myPullStream$ = new Subject();
const eventSource = new EventSource('http://localhost/pullStream');

eventSource.onmessage = async (event) => {
    await replicationState.awaitInSync();
    const eventData = JSON.parse(event.data);
    myPullStream$.next({
        documents: eventData.documents,
        checkpoint: eventData.checkpoint
    });
};

const replicationState = await replicateRxCollection({
    /* ... */
    pull: {
        /* ... */
        stream$: myPullStream$.asObservable()
    }
    /* ... */
});

[pubkey]

Thank you for reporting, I will try to build a test case for that.

[pubkey]

Could reproduce, working on a fix

[emilhdiaz]

Fantastic! Thank you @pubkey . Will test the new 15.6.0 version shortly.

[emilhdiaz]

@pubkey sorry for the delay, just wanted to circle back and confirm that the new version is working as expected. Thanks again!