From 3d8505fb24e133eb452f0bdc3544f1f700fd5c01 Mon Sep 17 00:00:00 2001 From: Adam Horodyski Date: Fri, 28 Aug 2026 11:53:03 +0200 Subject: [PATCH] Count in-flight Pusher applies as applied when detecting gaps applyPusherOnyxUpdates serializes every event through pusherEventsPromise and the watermark only advances once that chain settles, so an event whose previousUpdateID names an update the client already received but is still applying reads as a gap. The client then pauses the write queue and fetches data it is in the middle of writing. Measured over one production hour (2026-08-26 09:00-10:00 UTC): of the gap detections where both the Pusher arrival and the apply are observed, 74.0% had the missing update delivered before the gap fired and applied after it, at arrival to detection p50 209ms. Track the highest update ID accepted into the Pusher apply chain and count it as applied while it is in flight, mirroring the existing pending-flush marker for deferred WRITEs. Reset it if the apply fails, so a real gap is still detected, and on sign-out, so a stale value cannot mask a gap in the next session. Co-Authored-By: Claude Opus 5 (1M context) --- src/libs/actions/OnyxUpdates.ts | 23 +++- tests/actions/ApplyOnyxUpdatesReliablyTest.ts | 102 ++++++++++++++++++ 2 files changed, 121 insertions(+), 4 deletions(-) create mode 100644 tests/actions/ApplyOnyxUpdatesReliablyTest.ts diff --git a/src/libs/actions/OnyxUpdates.ts b/src/libs/actions/OnyxUpdates.ts index 156754d0a350..a2948d58f74f 100644 --- a/src/libs/actions/OnyxUpdates.ts +++ b/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({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({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) { diff --git a/tests/actions/ApplyOnyxUpdatesReliablyTest.ts b/tests/actions/ApplyOnyxUpdatesReliablyTest.ts new file mode 100644 index 000000000000..fdd78127d0cb --- /dev/null +++ b/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 => ({ + 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((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((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(); + }); +});