SequentialQueue: await Onyx persist before flush

adhorodyski · adhorodyski · commit 70d1ac9d44af · 2026-05-26T16:48:30.000+02:00
diff --git a/src/libs/Network/SequentialQueue.ts b/src/libs/Network/SequentialQueue.ts
@@ -504,7 +504,7 @@ async function handleConflictActions<TKey extends OnyxKey>(conflictAction: Confl
     }
 }
 
-function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>): Promise<void> {
+async function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>): Promise<void> {
     const currentRequests = getAllPersistedRequests();
     Log.info('[SequentialQueue] push() called', false, {
         command: newRequest.command,
@@ -514,9 +514,10 @@ function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>): Promise<void
         isSequentialQueueRunning,
     });
 
-    // Save the request to the persisted queue. The in-memory update inside save()
-    // happens synchronously, so flush() below will see the new request immediately.
-    // The returned promise resolves when disk persistence completes.
+    // Persist-before-fire: await disk write before any flush(). The in-memory update
+    // inside save()/handleConflictActions happens synchronously, but the network call
+    // must NOT race the Onyx.set() promise. If the process is killed after fire but
+    // before disk commit, the request is lost — no retry on next launch.
     let persistencePromise: Promise<void>;
 
     if (newRequest.checkAndFixConflictingRequest) {
@@ -544,13 +545,20 @@ function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>): Promise<void
         persistencePromise = savePersistedRequest(newRequest);
     }
 
+    await persistencePromise;
+    Log.info('[SequentialQueue] Request persisted, proceeding to flush', false, {
+        command: newRequest.command,
+        isOffline: isOfflineNetwork(),
+        isSequentialQueueRunning,
+    });
+
     // If we are offline we don't need to trigger the queue to empty as it will happen when we come back online
     if (isOfflineNetwork()) {
         Log.info('[SequentialQueue] Request persisted but not flushing — we are offline', false, {
             command: newRequest.command,
             queueLength: getAllPersistedRequests().length,
         });
-        return persistencePromise;
+        return;
     }
 
     // If the queue is running this request will run once it has finished processing the current batch
@@ -564,14 +572,13 @@ function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>): Promise<void
             });
             flush(true);
         });
-        return persistencePromise;
+        return;
     }
 
     Log.info('[SequentialQueue] Queue is not running. Flushing the queue.', false, {
         command: newRequest.command,
     });
     flush(true);
-    return persistencePromise;
 }
 
 function getCurrentRequest(): Promise<void> {
diff --git a/tests/unit/SequentialQueuePersistBeforeFireTest.ts b/tests/unit/SequentialQueuePersistBeforeFireTest.ts
@@ -0,0 +1,221 @@
+import Onyx from 'react-native-onyx';
+import type {OnyxKey} from 'react-native-onyx';
+import OnyxUtils from 'react-native-onyx/dist/OnyxUtils';
+import * as PersistedRequests from '@userActions/PersistedRequests';
+import ONYXKEYS from '@src/ONYXKEYS';
+import * as SequentialQueue from '../../src/libs/Network/SequentialQueue';
+import type Request from '../../src/types/onyx/Request';
+import * as TestHelper from '../utils/TestHelper';
+import waitForBatchedUpdates from '../utils/waitForBatchedUpdates';
+
+jest.mock('@libs/ActiveClientManager', () => ({
+    isClientTheLeader: jest.fn(() => true),
+    init: jest.fn(),
+    isReady: jest.fn(() => Promise.resolve()),
+}));
+
+type DeferredSet = {key: OnyxKey; value: unknown; resolve: () => void; flush: () => Promise<void>};
+
+/**
+ * Intercept `Onyx.set` for the given key and defer resolution so tests can observe
+ * the moment-of-fire vs. moment-of-disk-commit ordering. Returns the list of
+ * captured sets (mutated as more arrive) and a restore function.
+ */
+function deferOnyxSetFor(key: OnyxKey) {
+    const captured: DeferredSet[] = [];
+    const originalSet = Onyx.set.bind(Onyx);
+    const spy = jest.spyOn(Onyx, 'set').mockImplementation((targetKey, value) => {
+        if (targetKey !== key) {
+            return originalSet(targetKey, value);
+        }
+        return new Promise<void>((resolvePromise) => {
+            captured.push({
+                key: targetKey,
+                value,
+                resolve: resolvePromise,
+                flush: () => originalSet(targetKey, value as never).then(resolvePromise),
+            });
+        });
+    });
+    return {captured, restore: () => spy.mockRestore()};
+}
+
+const request: Request<'userMetadata'> = {
+    command: 'ReconnectApp',
+    successData: [{key: 'userMetadata', onyxMethod: 'set', value: {accountID: 1234}}],
+    failureData: [{key: 'userMetadata', onyxMethod: 'set', value: {}}],
+};
+
+beforeAll(() => {
+    Onyx.init({keys: ONYXKEYS});
+});
+
+let fetchMock: jest.Mock;
+beforeEach(async () => {
+    fetchMock = TestHelper.getGlobalFetchMock() as unknown as jest.Mock;
+    global.fetch = fetchMock as unknown as typeof fetch;
+    await Onyx.clear();
+    SequentialQueue.resetQueue();
+    await SequentialQueue.clearQueueFlushedData();
+    await waitForBatchedUpdates();
+});
+
+describe('SequentialQueue.push - persist-before-fire', () => {
+    it('online: does not call fetch until Onyx.set for PERSISTED_REQUESTS resolves', async () => {
+        const {captured, restore} = deferOnyxSetFor(ONYXKEYS.PERSISTED_REQUESTS);
+
+        try {
+            const pushPromise = SequentialQueue.push({...request});
+
+            // Pump microtasks. With persistence deferred, push() must not resolve and
+            // fetch must not be called.
+            await waitForBatchedUpdates();
+            await waitForBatchedUpdates();
+            expect(fetchMock).not.toHaveBeenCalled();
+            expect(captured.length).toBeGreaterThanOrEqual(1);
+
+            // Resolve the deferred persistence — only now is the request authoritative on disk.
+            await captured.at(0)?.flush();
+            await pushPromise;
+            await waitForBatchedUpdates();
+            await waitForBatchedUpdates();
+
+            expect(fetchMock).toHaveBeenCalled();
+        } finally {
+            restore();
+        }
+    });
+
+    it('at the await boundary, on-disk PERSISTED_REQUESTS already contains the new request', async () => {
+        await SequentialQueue.push({...request});
+
+        // Read straight from Onyx to confirm the persisted shape, not just in-memory state.
+        const onDisk = await OnyxUtils.get(ONYXKEYS.PERSISTED_REQUESTS);
+        expect(onDisk).toBeDefined();
+        expect(onDisk?.some((r) => r.command === request.command)).toBe(true);
+    });
+
+    it('offline: push() resolves after persistence and does not call fetch', async () => {
+        await Onyx.set(ONYXKEYS.NETWORK, {shouldForceOffline: true});
+        await waitForBatchedUpdates();
+
+        const {captured, restore} = deferOnyxSetFor(ONYXKEYS.PERSISTED_REQUESTS);
+        try {
+            let resolved = false;
+            const pushPromise = SequentialQueue.push({...request}).then(() => {
+                resolved = true;
+            });
+
+            await waitForBatchedUpdates();
+            expect(resolved).toBe(false);
+            expect(captured.length).toBeGreaterThanOrEqual(1);
+
+            await captured.at(0)?.flush();
+            await pushPromise;
+            await waitForBatchedUpdates();
+
+            expect(resolved).toBe(true);
+            expect(fetchMock).not.toHaveBeenCalled();
+            expect(PersistedRequests.getAll().some((r) => r.command === request.command)).toBe(true);
+        } finally {
+            restore();
+            await Onyx.set(ONYXKEYS.NETWORK, {shouldForceOffline: false});
+        }
+    });
+
+    it('conflict resolution (replace): final queue state is on disk before any fire', async () => {
+        // Pause so the seeded request sits in the queue instead of being consumed by process().
+        SequentialQueue.pause();
+
+        await SequentialQueue.push({...request});
+        await waitForBatchedUpdates();
+        fetchMock.mockClear();
+
+        const {captured, restore} = deferOnyxSetFor(ONYXKEYS.PERSISTED_REQUESTS);
+        try {
+            const replacement: Request<never> = {
+                command: 'ReconnectApp',
+                data: {accountID: 99999},
+                checkAndFixConflictingRequest: (persisted) => {
+                    const index = persisted.findIndex((r) => r.command === 'ReconnectApp');
+                    if (index === -1) {
+                        return {conflictAction: {type: 'push'}};
+                    }
+                    return {conflictAction: {type: 'replace', index}};
+                },
+            };
+
+            const pushPromise = SequentialQueue.push(replacement);
+            await waitForBatchedUpdates();
+
+            // No fetch fires while disk is pending (queue is also paused, but this guard
+            // catches a regression where push() would call flush() before the await).
+            expect(fetchMock).not.toHaveBeenCalled();
+            expect(captured.length).toBeGreaterThanOrEqual(1);
+
+            for (const capturedSet of captured) {
+                // Drain whatever conflict-resolution writes were issued (push/replace/delete chains
+                // may issue more than one) so persistence is fully on disk before flush proceeds.
+                await capturedSet.flush();
+            }
+            await pushPromise;
+            await waitForBatchedUpdates();
+
+            const onDisk = await OnyxUtils.get(ONYXKEYS.PERSISTED_REQUESTS);
+            expect(onDisk?.some((r) => r.data?.accountID === 99999)).toBe(true);
+        } finally {
+            restore();
+            SequentialQueue.unpause();
+        }
+    });
+
+    it('queue-already-running path: push() awaits persistence even when a flush is already in flight', async () => {
+        // First push starts the queue. We do not pause/intercept it — let it run.
+        await SequentialQueue.push({...request});
+        await waitForBatchedUpdates();
+        fetchMock.mockClear();
+
+        const {captured, restore} = deferOnyxSetFor(ONYXKEYS.PERSISTED_REQUESTS);
+        try {
+            let resolved = false;
+            const pushPromise = SequentialQueue.push({...request, data: {accountID: 4242}}).then(() => {
+                resolved = true;
+            });
+            await waitForBatchedUpdates();
+
+            // push() must not have resolved yet — persistence is pending.
+            expect(resolved).toBe(false);
+            expect(captured.length).toBeGreaterThanOrEqual(1);
+
+            for (const capturedSet of captured) {
+                await capturedSet.flush();
+            }
+            await pushPromise;
+            expect(resolved).toBe(true);
+        } finally {
+            restore();
+        }
+    });
+});
+
+describe('PersistedRequests.save - regression guard', () => {
+    it('save() resolves only after the underlying Onyx.set completes', async () => {
+        const {captured, restore} = deferOnyxSetFor(ONYXKEYS.PERSISTED_REQUESTS);
+        try {
+            let resolved = false;
+            const savePromise = PersistedRequests.save({...request, requestID: 7777}).then(() => {
+                resolved = true;
+            });
+
+            await waitForBatchedUpdates();
+            expect(resolved).toBe(false);
+            expect(captured.length).toBeGreaterThanOrEqual(1);
+
+            await captured.at(0)?.flush();
+            await savePromise;
+            expect(resolved).toBe(true);
+        } finally {
+            restore();
+        }
+    });
+});
diff --git a/tests/unit/SequentialQueueTest.ts b/tests/unit/SequentialQueueTest.ts
