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

src/libs/actions/OnyxUpdates.ts

@@ -24,6 +24,12 @@ let lastUpdateIDAppliedToClient: number | undefined = 0;
 // applied so queued WRITE responses don't look like gaps; reset if the flush fails so recovery can kick in.
 let lastUpdateIDPendingFlush = 0;
 
+// Highest update ID accepted into the Pusher apply chain but not yet persisted. applyPusherOnyxUpdates serializes
+// every event through pusherEventsPromise, so the watermark trails an update the client already holds by p50 209ms
+// and p90 1.7s in production. Gap detection treats these as applied, otherwise the next event in the chain reads as
+// a gap and pauses the queue to refetch data we are in the middle of applying.
+let lastUpdateIDPendingApply = 0;
+
 function getEffectiveLastUpdateID(): number {
     return Math.max(lastUpdateIDAppliedToClient ?? 0, lastUpdateIDPendingFlush);
 }
@@ -38,10 +44,11 @@ Onyx.connectWithoutView({
     callback: (val) => {
         lastUpdateIDAppliedToClient = val;
 
-        // The persisted watermark is only ever cleared by Onyx.clear (sign-out), so drop the pending marker
+        // The persisted watermark is only ever cleared by Onyx.clear (sign-out), so drop the pending markers
         // too — a stale value from the previous session would mask real gaps after signing back in.
         if (val === undefined) {
             lastUpdateIDPendingFlush = 0;
+            lastUpdateIDPendingApply = 0;
         }
     },
 });
@@ -209,6 +216,10 @@ function apply<TKey extends OnyxKey>({lastUpdateID, type, request, response, upd
                 return result;
             })
             .catch((error) => {
+                // The updates never landed, so stop counting them as applied — the next gap check then sees the
+                // missing range against the persisted watermark and triggers recovery.
+                lastUpdateIDPendingApply = 0;
+
                 if (shouldAdvanceLastUpdateID) {
                     Log.alert('[OnyxUpdateManagerError] Applying the updates failed, not advancing lastUpdateID so the client can recover on the next reconnect', {
                         type,
@@ -241,6 +252,9 @@ function apply<TKey extends OnyxKey>({lastUpdateID, type, request, response, upd
         return advanceLastUpdateIDAfterApply(applyPromise);
     }
     if (type === CONST.ONYX_UPDATE_TYPES.PUSHER && updates) {
+        if (shouldAdvanceLastUpdateID) {
+            lastUpdateIDPendingApply = Math.max(lastUpdateIDPendingApply, Number(lastUpdateID));
+        }
         return advanceLastUpdateIDAfterApply(applyPusherOnyxUpdates(updates, Number(lastUpdateID)));
     }
     if (type === CONST.ONYX_UPDATE_TYPES.AIRSHIP && updates) {
@@ -282,9 +296,10 @@ function doesClientNeedToBeUpdated({previousUpdateID, clientLastUpdateID}: DoesC
         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);
+    // Updates staged for the deferred WRITE flush, and Pusher updates still working through the apply chain, count
+    // as applied here. Otherwise the responses of queued WRITE requests, and the event chained on an update we are
+    // already applying, would look like gaps and needlessly pause the queue to refetch data the client already has.
+    const lastUpdateIDFromClient = Math.max(clientLastUpdateID ?? lastUpdateIDAppliedToClient ?? 0, lastUpdateIDPendingFlush, lastUpdateIDPendingApply);
 
     // 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) {

tests/actions/ApplyOnyxUpdatesReliablyTest.ts

@@ -0,0 +1,102 @@
+import applyOnyxUpdatesReliably from '@libs/actions/applyOnyxUpdatesReliably';
+import {isPaused as isSequentialQueuePaused, unpause as unpauseSequentialQueue} from '@libs/Network/SequentialQueue';
+import PusherUtils from '@libs/PusherUtils';
+
+import CONST from '@src/CONST';
+import ONYXKEYS from '@src/ONYXKEYS';
+import type {OnyxUpdatesFromServer} from '@src/types/onyx';
+
+import Onyx from 'react-native-onyx';
+
+import waitForBatchedUpdates from '../utils/waitForBatchedUpdates';
+
+const pusherUpdate = (previousUpdateID: number, lastUpdateID: number): OnyxUpdatesFromServer<never> => ({
+    type: CONST.ONYX_UPDATE_TYPES.PUSHER,
+    previousUpdateID,
+    lastUpdateID,
+    updates: [{eventType: 'onyxApiUpdate', data: []}],
+});
+
+describe('actions/applyOnyxUpdatesReliably', () => {
+    beforeAll(() => {
+        Onyx.init({keys: ONYXKEYS});
+    });
+
+    beforeEach(async () => {
+        unpauseSequentialQueue();
+        await Onyx.clear();
+        await Onyx.set(ONYXKEYS.ONYX_UPDATES_LAST_UPDATE_ID_APPLIED_TO_CLIENT, 10);
+        await waitForBatchedUpdates();
+    });
+
+    it('does not pause the queue for an event chained on an update that is still being applied', async () => {
+        // Given update 20 arrived over Pusher and its apply is held mid-flight, so the watermark is still at 10
+        let releaseApply: () => void = () => {};
+        const handlerSpy = jest.spyOn(PusherUtils, 'triggerMultiEventHandler').mockReturnValueOnce(
+            new Promise<void>((resolve) => {
+                releaseApply = resolve;
+            }),
+        );
+        const heldApply = applyOnyxUpdatesReliably(pusherUpdate(10, 20));
+        await waitForBatchedUpdates();
+
+        // When the next event arrives, chained on the update we are still applying
+        const chainedApply = applyOnyxUpdatesReliably(pusherUpdate(20, 30));
+        await waitForBatchedUpdates();
+
+        // Then the queue is not paused to refetch data the client already received
+        expect(isSequentialQueuePaused()).toBe(false);
+
+        releaseApply();
+        await heldApply;
+        await chainedApply;
+        handlerSpy.mockRestore();
+    });
+
+    it('does not let an update left mid-apply by the previous session mask a gap after signing back in', async () => {
+        // Given update 20 arrived over Pusher and its apply is held mid-flight
+        let releaseApply: () => void = () => {};
+        const handlerSpy = jest.spyOn(PusherUtils, 'triggerMultiEventHandler').mockReturnValueOnce(
+            new Promise<void>((resolve) => {
+                releaseApply = resolve;
+            }),
+        );
+        const heldApply = applyOnyxUpdatesReliably(pusherUpdate(10, 20));
+        await waitForBatchedUpdates();
+
+        // When the user signs out, which clears the persisted watermark
+        await Onyx.clear();
+        await waitForBatchedUpdates();
+
+        // And an event chained on update 20 arrives in the new session
+        const chainedApply = applyOnyxUpdatesReliably(pusherUpdate(20, 30));
+        await waitForBatchedUpdates();
+
+        // Then the queue is paused, because the new session never received update 20
+        expect(isSequentialQueuePaused()).toBe(true);
+
+        releaseApply();
+        await heldApply;
+        await chainedApply;
+        handlerSpy.mockRestore();
+    });
+
+    // A failed apply leaves the module-level Pusher chain rejected, which poisons every later Pusher apply, so keep
+    // this case last and add new ones above it.
+    it('pauses the queue when the update it was waiting on failed to apply', async () => {
+        // Given applying update 20 from Pusher failed
+        const handlerSpy = jest.spyOn(PusherUtils, 'triggerMultiEventHandler').mockRejectedValueOnce(new Error('storage write failed'));
+        await expect(applyOnyxUpdatesReliably(pusherUpdate(10, 20))).rejects.toThrow('storage write failed');
+        await waitForBatchedUpdates();
+
+        // When an event chained on that update arrives
+        const chainedApply = applyOnyxUpdatesReliably(pusherUpdate(20, 30));
+        chainedApply.catch(() => {});
+        await waitForBatchedUpdates();
+
+        // Then the queue is paused so the update that never landed can be refetched
+        expect(isSequentialQueuePaused()).toBe(true);
+
+        handlerSpy.mockRestore();
+    });
+});