Count in-flight Pusher applies as applied when detecting update gaps

src/libs/actions/OnyxUpdates.ts

@@ -22,10 +22,12 @@ let lastUpdateIDAppliedToClient: number | undefined = 0;
 
 // Highest update ID staged for the deferred WRITE flush but not yet persisted. Gap detection treats these as
 // applied so queued WRITE responses don't look like gaps; reset if the flush fails so recovery can kick in.
-let lastUpdateIDPendingFlush = 0;
+let lastUpdateIDPendingWriteFlush = 0;
+
+let lastUpdateIDPendingPusherApply = 0;
 
 function getEffectiveLastUpdateID(): number {
-    return Math.max(lastUpdateIDAppliedToClient ?? 0, lastUpdateIDPendingFlush);
+    return Math.max(lastUpdateIDAppliedToClient ?? 0, lastUpdateIDPendingWriteFlush);
 }
 
 function getPersistedLastUpdateID(): number {
@@ -41,7 +43,8 @@ Onyx.connectWithoutView({
         // The persisted watermark is only ever cleared by Onyx.clear (sign-out), so drop the pending marker
         // too — a stale value from the previous session would mask real gaps after signing back in.
         if (val === undefined) {
-            lastUpdateIDPendingFlush = 0;
+            lastUpdateIDPendingWriteFlush = 0;
+            lastUpdateIDPendingPusherApply = 0;
         }
     },
 });
@@ -203,12 +206,19 @@ function apply<TKey extends OnyxKey>({lastUpdateID, type, request, response, upd
                     Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, Number(lastUpdateID));
                 }
                 // The persisted watermark now covers the staged WRITE updates, so the pending marker is no longer needed
-                if (lastUpdateIDPendingFlush && lastUpdateIDPendingFlush <= Number(lastUpdateID)) {
-                    lastUpdateIDPendingFlush = 0;
+                if (lastUpdateIDPendingWriteFlush && lastUpdateIDPendingWriteFlush <= Number(lastUpdateID)) {
+                    lastUpdateIDPendingWriteFlush = 0;
+                }
+                if (lastUpdateIDPendingPusherApply && lastUpdateIDPendingPusherApply <= Number(lastUpdateID)) {
+                    lastUpdateIDPendingPusherApply = 0;
                 }
                 return result;
             })
             .catch((error) => {
+                // Intentionally cleared for any failed apply, including HTTPS and Airship: the marker is a flat max, so
+                // keeping it after an unrelated lower-ID failure would mask that gap. Errs toward a redundant refetch.
+                lastUpdateIDPendingPusherApply = 0;
+
                 if (shouldAdvanceLastUpdateID) {
                     Log.alert('[OnyxUpdateManagerError] Applying the updates failed, not advancing lastUpdateID so the client can recover on the next reconnect', {
                         type,
@@ -229,18 +239,22 @@ function apply<TKey extends OnyxKey>({lastUpdateID, type, request, response, upd
         // SequentialQueue only flushes after this promise settles, so awaiting the flush here would deadlock.
         if (request.data?.apiRequestType === CONST.API_REQUEST_TYPE.WRITE) {
             if (shouldAdvanceLastUpdateID) {
-                lastUpdateIDPendingFlush = Math.max(lastUpdateIDPendingFlush, Number(lastUpdateID));
+                lastUpdateIDPendingWriteFlush = Math.max(lastUpdateIDPendingWriteFlush, Number(lastUpdateID));
             }
             advanceLastUpdateIDAfterApply(applyPromise.then(() => getCurrentFlushPromise())).catch(() => {
                 // The staged updates never applied, so stop counting them as pending — the next gap check
                 // then sees the missing range against the persisted watermark and triggers recovery.
-                lastUpdateIDPendingFlush = 0;
+                lastUpdateIDPendingWriteFlush = 0;
             });
             return applyPromise;
         }
         return advanceLastUpdateIDAfterApply(applyPromise);
     }
     if (type === CONST.ONYX_UPDATE_TYPES.PUSHER && updates) {
+        if (shouldAdvanceLastUpdateID) {
+            lastUpdateIDPendingPusherApply = Math.max(lastUpdateIDPendingPusherApply, Number(lastUpdateID));
+        }
+
         return advanceLastUpdateIDAfterApply(applyPusherOnyxUpdates(updates, Number(lastUpdateID)));
     }
     if (type === CONST.ONYX_UPDATE_TYPES.AIRSHIP && updates) {
@@ -268,23 +282,32 @@ function saveUpdateInformation<TKey extends OnyxKey>(updateParams: OnyxUpdatesFr
 type DoesClientNeedToBeUpdatedParams = {
     clientLastUpdateID?: number;
     previousUpdateID?: number;
+    updateType?: AnyOnyxUpdatesFromServer['type'];
 };
 
+function isSerializedBehindPusherApply(updateType?: AnyOnyxUpdatesFromServer['type']): boolean {
+    return updateType === CONST.ONYX_UPDATE_TYPES.PUSHER;
+}
+
 /**
  * This function will receive the previousUpdateID from any request/pusher update that has it, compare to our current app state
  * and return if an update is needed
  * @param previousUpdateID The previousUpdateID contained in the response object
  * @param clientLastUpdateID an optional override for the lastUpdateIDAppliedToClient
+ * @param updateType the transport the update being checked arrived on
  */
-function doesClientNeedToBeUpdated({previousUpdateID, clientLastUpdateID}: DoesClientNeedToBeUpdatedParams): boolean {
+function doesClientNeedToBeUpdated({previousUpdateID, clientLastUpdateID, updateType}: DoesClientNeedToBeUpdatedParams): boolean {
     // If no previousUpdateID is sent, this is not a WRITE request so we don't need to update our current state
     if (!previousUpdateID) {
         return false;
     }
 
-    // Updates staged for the deferred WRITE flush count as applied here, otherwise the responses of queued
-    // WRITE requests would look like gaps until the flush runs and needlessly pause the queue to refetch.
-    const lastUpdateIDFromClient = Math.max(clientLastUpdateID ?? lastUpdateIDAppliedToClient ?? 0, lastUpdateIDPendingFlush);
+    // QueuedOnyxUpdates defers the Onyx write for WRITE requests, so their own responses arrive before the watermark moves.
+    const lastUpdateIDFromClient = Math.max(
+        clientLastUpdateID ?? lastUpdateIDAppliedToClient ?? 0,
+        lastUpdateIDPendingWriteFlush,
+        isSerializedBehindPusherApply(updateType) ? lastUpdateIDPendingPusherApply : 0,
+    );
 
     // If we don't have any value in lastUpdateIDFromClient, this is the first time we're receiving anything, so we need to do a last reconnectApp
     if (!lastUpdateIDFromClient) {

src/libs/actions/applyOnyxUpdatesReliably.ts

@@ -48,7 +48,7 @@ export default function applyOnyxUpdatesReliably<TKey extends OnyxKey>(
     }
 
     const previousUpdateID = Number(updates.previousUpdateID) ?? CONST.DEFAULT_NUMBER_ID;
-    if (!doesClientNeedToBeUpdated({previousUpdateID, clientLastUpdateID})) {
+    if (!doesClientNeedToBeUpdated({previousUpdateID, clientLastUpdateID, updateType: updates.type})) {
         return onyxApply(updates).then();
     }
 

tests/unit/OnyxUpdatesPusherChainFailureTest.ts

@@ -0,0 +1,75 @@
+import PusherUtils from '@libs/PusherUtils';
+
+import CONST from '@src/CONST';
+import {apply, doesClientNeedToBeUpdated} from '@src/libs/actions/OnyxUpdates';
+import ONYXKEYS from '@src/ONYXKEYS';
+import type {OnyxUpdatesFromServer} from '@src/types/onyx';
+
+import type {OnyxKey} from 'react-native-onyx';
+
+import Onyx from 'react-native-onyx';
+
+import getOnyxValue from '../utils/getOnyxValue';
+import waitForBatchedUpdates from '../utils/waitForBatchedUpdates';
+
+const pusherUpdate = (previousUpdateID: number, lastUpdateID: number): OnyxUpdatesFromServer<OnyxKey> => ({
+    type: CONST.ONYX_UPDATE_TYPES.PUSHER,
+    previousUpdateID,
+    lastUpdateID,
+    updates: [{eventType: 'onyxApiUpdate', data: []}],
+});
+
+// A rejected Pusher apply leaves the module-scoped pusherEventsPromise rejected for the rest of the module's life,
+// so this lives in its own file rather than poisoning the chain for the other tests.
+describe('OnyxUpdates, when a Pusher apply fails', () => {
+    beforeAll(() => {
+        Onyx.init({
+            keys: ONYXKEYS,
+        });
+    });
+
+    beforeEach(() => Onyx.clear().then(waitForBatchedUpdates));
+
+    it('relies on pusherEventsPromise staying rejected to stop a follower whose gap check the failed update had suppressed', async () => {
+        // Given the client is caught up to update 10 and update 20 from Pusher is held mid-apply
+        await Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);
+        await waitForBatchedUpdates();
+
+        let failHeldApply: (error: Error) => void = () => {};
+        let handlerCallCount = 0;
+        const handlerSpy = jest.spyOn(PusherUtils, 'triggerMultiEventHandler').mockImplementation(() => {
+            handlerCallCount += 1;
+            if (handlerCallCount > 1) {
+                return Promise.resolve();
+            }
+            return new Promise<void>((resolve, reject) => {
+                failHeldApply = reject;
+            });
+        });
+        const heldApply = apply(pusherUpdate(10, 20));
+        await waitForBatchedUpdates();
+
+        // When update 30 arrives chained on it, so the pending marker tells it there is no gap
+        expect(doesClientNeedToBeUpdated({previousUpdateID: 20, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(false);
+        const followerApply = apply(pusherUpdate(20, 30));
+        await waitForBatchedUpdates();
+
+        // And update 20 then fails to apply
+        failHeldApply(new Error('storage write failed'));
+        await expect(heldApply).rejects.toThrow('storage write failed');
+        await expect(followerApply).rejects.toThrow('storage write failed');
+        await waitForBatchedUpdates();
+
+        // Then update 30 is never written either. Nothing checks that update IDs are contiguous before advancing the
+        // watermark, so were it written the watermark would move to 30 and updates 11 to 20 would be lost with no gap
+        // left to trigger recovery. Serializing on pusherEventsPromise is the only thing preventing that.
+        expect(handlerCallCount).toBe(1);
+        expect(await getOnyxValue(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT)).toBe(10);
+
+        // And both updates are back in the gap, so recovery can refetch them
+        expect(doesClientNeedToBeUpdated({previousUpdateID: 20, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(true);
+        expect(doesClientNeedToBeUpdated({previousUpdateID: 30, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(true);
+
+        handlerSpy.mockRestore();
+    });
+});

tests/unit/OnyxUpdatesTest.ts

@@ -1,4 +1,5 @@
 import {SIDE_EFFECT_REQUEST_COMMANDS} from '@libs/API/types';
+import PusherUtils from '@libs/PusherUtils';
 
 import CONST from '@src/CONST';
 import * as OnyxUpdates from '@src/libs/actions/OnyxUpdates';
@@ -24,6 +25,12 @@ describe('OnyxUpdatesTest', () => {
 
     beforeEach(() => Onyx.clear().then(waitForBatchedUpdates));
 
+    let releaseHeldApply: (() => void) | undefined;
+    afterEach(() => {
+        releaseHeldApply?.();
+        releaseHeldApply = undefined;
+    });
+
     it('applies Airship Onyx updates correctly', () => {
         const reportID = NumberUtils.rand64();
         const reportActionID = NumberUtils.rand64();
@@ -249,7 +256,7 @@ describe('OnyxUpdatesTest', () => {
 
         // Then a following response chained on update 20 is not treated as a gap, even though the
         // persisted watermark is still at 10 — otherwise every queued WRITE would pause the queue
-        expect(OnyxUpdates.doesClientNeedToBeUpdated({previousUpdateID: 20})).toBe(false);
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({previousUpdateID: 20, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(false);
 
         // And once the flush applies the staged updates, the persisted watermark catches up
         await flushQueue();
@@ -302,19 +309,157 @@ describe('OnyxUpdatesTest', () => {
             },
         });
         await waitForBatchedUpdates();
-        expect(OnyxUpdates.doesClientNeedToBeUpdated({previousUpdateID: 20})).toBe(false);
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({previousUpdateID: 20, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(false);
 
         // When the user signs out, which clears Onyx storage
         await Onyx.clear();
         await waitForBatchedUpdates();
 
         // Then the pending watermark from the previous session no longer masks gaps in the new session
-        expect(OnyxUpdates.doesClientNeedToBeUpdated({clientLastUpdateID: 5, previousUpdateID: 15})).toBe(true);
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({clientLastUpdateID: 5, previousUpdateID: 15, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(true);
 
         // Drain the staged updates so they don't leak into other tests
         await flushQueue();
     });
 
+    const applyHeldPusherUpdate = (previousUpdateID: number, lastUpdateID: number) => {
+        let releaseApply: () => void = () => {};
+        const handlerSpy = jest.spyOn(PusherUtils, 'triggerMultiEventHandler').mockReturnValueOnce(
+            new Promise<void>((resolve) => {
+                releaseApply = resolve;
+                releaseHeldApply = resolve;
+            }),
+        );
+        const applyPromise = OnyxUpdates.apply({
+            type: CONST.ONYX_UPDATE_TYPES.PUSHER,
+            previousUpdateID,
+            lastUpdateID,
+            updates: [{eventType: 'onyxApiUpdate', data: []}],
+        });
+
+        return {
+            release: () => {
+                releaseApply();
+                handlerSpy.mockRestore();
+                return applyPromise;
+            },
+        };
+    };
+
+    it('does not report a gap for a Pusher update that is still applying', async () => {
+        // Given the client is caught up to update 10
+        await Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);
+        await waitForBatchedUpdates();
+
+        // When update 20 arrives over Pusher and its apply is held mid-flight
+        const heldApply = applyHeldPusherUpdate(10, 20);
+        await waitForBatchedUpdates();
+
+        // Then the next event, chained on update 20, is not treated as a gap even though the watermark is still at 10
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({previousUpdateID: 20, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(false);
+
+        await heldApply.release();
+    });
+
+    it('reports a gap for an HTTPS response chained on a Pusher update that is still applying', async () => {
+        // Given the client is caught up to update 10
+        await Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);
+        await waitForBatchedUpdates();
+
+        // When update 20 arrives over Pusher and its apply is held mid-flight
+        const heldApply = applyHeldPusherUpdate(10, 20);
+        await waitForBatchedUpdates();
+
+        // Then an HTTPS response chained on update 20 still reports the gap, because its apply runs on its own
+        // promise chain and would advance the watermark past the updates the held apply has not written yet
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({previousUpdateID: 20})).toBe(true);
+
+        await heldApply.release();
+    });
+
+    it('keeps a Pusher update that is still applying out of the catch-up fetch range', async () => {
+        // Given the client is caught up to update 10
+        await Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);
+        await waitForBatchedUpdates();
+
+        // When update 20 arrives over Pusher and its apply is held mid-flight
+        const heldApply = applyHeldPusherUpdate(10, 20);
+        await waitForBatchedUpdates();
+
+        // Then a genuinely later gap is still detected
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({previousUpdateID: 30, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(true);
+
+        // And it fetches from the persisted watermark, so a rejected apply cannot strand update 20
+        expect(OnyxUpdates.getEffectiveLastUpdateID()).toBe(10);
+
+        await heldApply.release();
+    });
+
+    it('clears the pending apply watermark on sign-out', async () => {
+        // Given the client is caught up to update 10 and update 20 from Pusher is held mid-apply
+        await Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);
+        await waitForBatchedUpdates();
+        const heldApply = applyHeldPusherUpdate(10, 20);
+        await waitForBatchedUpdates();
+
+        // When the user signs out, which clears Onyx storage
+        await Onyx.clear();
+        await waitForBatchedUpdates();
+
+        // Then the pending marker from the previous session no longer masks gaps in the new session
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({clientLastUpdateID: 5, previousUpdateID: 15, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(true);
+
+        // And the held apply settling afterwards does not reintroduce it
+        await heldApply.release();
+        await waitForBatchedUpdates();
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({clientLastUpdateID: 5, previousUpdateID: 15, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(true);
+    });
+
+    it('clears the Pusher pending apply marker when an unrelated apply fails below it, so a real gap is never masked', async () => {
+        // Given the client is caught up to update 10 and update 20 from Pusher is held mid-apply
+        await Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);
+        await waitForBatchedUpdates();
+        const heldApply = applyHeldPusherUpdate(10, 20);
+        await waitForBatchedUpdates();
+
+        // When an unrelated READ request's apply fails at update 15, below the held Pusher update
+        const updateSpy = jest.spyOn(Onyx, 'update').mockRejectedValueOnce(new Error('storage write failed'));
+        await expect(
+            OnyxUpdates.apply({
+                type: CONST.ONYX_UPDATE_TYPES.HTTPS,
+                previousUpdateID: 10,
+                lastUpdateID: 15,
+                request: {command: 'OpenReport', data: {apiRequestType: CONST.API_REQUEST_TYPE.READ}},
+                response: {
+                    jsonCode: 200,
+                    onyxData: [{onyxMethod: 'merge', key: `${ONYXKEYS.COLLECTION.REPORT}${NumberUtils.rand64()}`, value: {}}],
+                },
+            }),
+        ).rejects.toThrow('storage write failed');
+        await waitForBatchedUpdates();
+
+        // Then update 20 stops counting as applied, so the gap left by update 15 is detected instead of masked
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({previousUpdateID: 15, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(true);
+
+        updateSpy.mockRestore();
+        await heldApply.release();
+    });
+
+    it('stops counting a Pusher update as in flight once its apply has settled', async () => {
+        // Given the client is caught up to update 10
+        await Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);
+        await waitForBatchedUpdates();
+
+        // When update 20 arrives over Pusher and its apply finishes
+        const heldApply = applyHeldPusherUpdate(10, 20);
+        await waitForBatchedUpdates();
+        await heldApply.release();
+        await waitForBatchedUpdates();
+
+        // Then a caller that overrides the watermark with its own lower value is no longer told it is caught up
+        expect(OnyxUpdates.doesClientNeedToBeUpdated({clientLastUpdateID: 5, previousUpdateID: 15, updateType: CONST.ONYX_UPDATE_TYPES.PUSHER})).toBe(true);
+    });
+
     it('does not move the watermark backwards when a slower older update settles after a newer one', async () => {
         // Given the client is caught up to update 10
         await Onyx.merge(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);