De-dupe ReconnectApp in the persisted requests queue

src/libs/API/index.ts

@@ -9,7 +9,7 @@ import * as Request from '@libs/Request';
 import * as PersistedRequests from '@userActions/PersistedRequests';
 import CONST from '@src/CONST';
 import type OnyxRequest from '@src/types/onyx/Request';
-import type {PaginatedRequest, PaginationConfig} from '@src/types/onyx/Request';
+import type {PaginatedRequest, PaginationConfig, RequestConflictResolver} from '@src/types/onyx/Request';
 import type Response from '@src/types/onyx/Response';
 import type {ApiCommand, ApiRequestCommandParameters, ApiRequestType, CommandOfType, ReadCommand, SideEffectRequestCommand, WriteCommand} from './types';
 
@@ -45,7 +45,13 @@ type OnyxData = {
 /**
  * Prepare the request to be sent. Bind data together with request metadata and apply optimistic Onyx data.
  */
-function prepareRequest<TCommand extends ApiCommand>(command: TCommand, type: ApiRequestType, params: ApiRequestCommandParameters[TCommand], onyxData: OnyxData = {}): OnyxRequest {
+function prepareRequest<TCommand extends ApiCommand>(
+    command: TCommand,
+    type: ApiRequestType,
+    params: ApiRequestCommandParameters[TCommand],
+    onyxData: OnyxData = {},
+    conflictResolver: RequestConflictResolver = {},
+): OnyxRequest {
     Log.info('[API] Preparing request', false, {command, type});
 
     const {optimisticData, ...onyxDataWithoutOptimisticData} = onyxData;
@@ -71,6 +77,7 @@ function prepareRequest<TCommand extends ApiCommand>(command: TCommand, type: Ap
         command,
         data,
         ...onyxDataWithoutOptimisticData,
+        ...conflictResolver,
     };
 
     if (isWriteRequest) {
@@ -116,9 +123,14 @@ function processRequest(request: OnyxRequest, type: ApiRequestType): Promise<voi
  * @param [onyxData.failureData] - Onyx instructions that will be passed to Onyx.update() when the response has jsonCode !== 200.
  * @param [onyxData.finallyData] - Onyx instructions that will be passed to Onyx.update() when the response has jsonCode === 200 or jsonCode !== 200.
  */
-function write<TCommand extends WriteCommand>(command: TCommand, apiCommandParameters: ApiRequestCommandParameters[TCommand], onyxData: OnyxData = {}): void {
+function write<TCommand extends WriteCommand>(
+    command: TCommand,
+    apiCommandParameters: ApiRequestCommandParameters[TCommand],
+    onyxData: OnyxData = {},
+    conflictResolver: RequestConflictResolver = {},
+): void {
     Log.info('[API] Called API write', false, {command, ...apiCommandParameters});
-    const request = prepareRequest(command, CONST.API_REQUEST_TYPE.WRITE, apiCommandParameters, onyxData);
+    const request = prepareRequest(command, CONST.API_REQUEST_TYPE.WRITE, apiCommandParameters, onyxData, conflictResolver);
     processRequest(request, CONST.API_REQUEST_TYPE.WRITE);
 }
 

src/libs/Network/SequentialQueue.ts

@@ -24,7 +24,8 @@ let isReadyPromise = new Promise((resolve) => {
 resolveIsReadyPromise?.();
 
 let isSequentialQueueRunning = false;
-let currentRequest: Promise<void> | null = null;
+let currentRequest: OnyxRequest | null = null;
+let currentRequestPromise: Promise<void> | null = null;
 let isQueuePaused = false;
 
 /**
@@ -81,8 +82,9 @@ function process(): Promise<void> {
 
     const requestToProcess = persistedRequests[0];
 
+    currentRequest = requestToProcess;
     // Set the current request to a promise awaiting its processing so that getCurrentRequest can be used to take some action after the current request has processed.
-    currentRequest = Request.processWithMiddleware(requestToProcess, true)
+    currentRequestPromise = Request.processWithMiddleware(requestToProcess, true)
         .then((response) => {
             // A response might indicate that the queue should be paused. This happens when a gap in onyx updates is detected between the client and the server and
             // that gap needs resolved before the queue can continue.
@@ -112,7 +114,7 @@ function process(): Promise<void> {
                 });
         });
 
-    return currentRequest;
+    return currentRequestPromise;
 }
 
 function flush() {
@@ -161,6 +163,8 @@ function flush() {
                     resolveIsReadyPromise?.();
                 }
                 currentRequest = null;
+                currentRequestPromise = null;
+
                 // The queue can be paused when we sync the data with backend so we should only update the Onyx data when the queue is empty
                 if (PersistedRequests.getAll().length === 0) {
                     flushOnyxUpdatesQueue();
@@ -196,9 +200,23 @@ function isPaused(): boolean {
 // Flush the queue when the connection resumes
 NetworkStore.onReconnection(flush);
 
-function push(request: OnyxRequest) {
-    // Add request to Persisted Requests so that it can be retried if it fails
-    PersistedRequests.save(request);
+function push(newRequest: OnyxRequest) {
+    // If a request is already being processed, ignore it when looking for potentially conflicting requests
+    const requests = PersistedRequests.getAll().filter((persistedRequest) => persistedRequest !== currentRequest);
+
+    const {checkAndFixConflictingRequest} = newRequest;
+    if (checkAndFixConflictingRequest) {
+        const {conflictAction} = checkAndFixConflictingRequest(requests);
+
+        if (conflictAction.type === 'push') {
+            PersistedRequests.save(newRequest);
+        } else if (conflictAction.type === 'replace') {
+            PersistedRequests.update(conflictAction.index, newRequest);
+        }
+    } else {
+        // Add request to Persisted Requests so that it can be retried if it fails
+        PersistedRequests.save(newRequest);
+    }
 
     // If we are offline we don't need to trigger the queue to empty as it will happen when we come back online
     if (NetworkStore.isOffline()) {
@@ -215,10 +233,10 @@ function push(request: OnyxRequest) {
 }
 
 function getCurrentRequest(): Promise<void> {
-    if (currentRequest === null) {
+    if (currentRequestPromise === null) {
         return Promise.resolve();
     }
-    return currentRequest;
+    return currentRequestPromise;
 }
 
 /**

src/libs/actions/App.ts

@@ -274,7 +274,25 @@ function reconnectApp(updateIDFrom: OnyxEntry<number> = 0) {
             params.updateIDFrom = updateIDFrom;
         }
 
-        API.write(WRITE_COMMANDS.RECONNECT_APP, params, getOnyxDataForOpenOrReconnect());
+        API.write(WRITE_COMMANDS.RECONNECT_APP, params, getOnyxDataForOpenOrReconnect(), {
+            checkAndFixConflictingRequest: (persistedRequests) => {
+                const index = persistedRequests.findIndex((request) => request.command === WRITE_COMMANDS.RECONNECT_APP);
+                if (index === -1) {
+                    return {
+                        conflictAction: {
+                            type: 'push',
+                        },
+                    };
+                }
+
+                return {
+                    conflictAction: {
+                        type: 'replace',
+                        index,
+                    },
+                };
+            },
+        });
     });
 }
 

src/libs/actions/PersistedRequests.ts

@@ -23,9 +23,10 @@ function getLength(): number {
 }
 
 function save(requestToPersist: Request) {
-    const requests = [...persistedRequests, requestToPersist];
-    persistedRequests = requests;
-    Onyx.set(ONYXKEYS.PERSISTED_REQUESTS, requests).then(() => {
+    // If the command is not in the keepLastInstance array, add the new request as usual
+    persistedRequests = [...persistedRequests, requestToPersist];
+
+    Onyx.set(ONYXKEYS.PERSISTED_REQUESTS, persistedRequests).then(() => {
         Log.info(`[SequentialQueue] '${requestToPersist.command}' command queued. Queue length is ${getLength()}`);
     });
 }

src/types/onyx/Request.ts

@@ -55,8 +55,64 @@ type RequestData = {
     shouldSkipWebProxy?: boolean;
 };
 
+/**
+ * Model of a conflict request that has to be updated, in the request queue.
+ */
+type ConflictRequestUpdate = {
+    /**
+     * The action to take in case of a conflict.
+     */
+    type: 'replace';
+
+    /**
+     * The index of the request in the queue to update.
+     */
+    index: number;
+};
+
+/**
+ * Model of a conflict request that has to be saved at the end the request queue.
+ */
+type ConflictRequestSave = {
+    /**
+     * The action to take in case of a conflict.
+     */
+    type: 'push';
+};
+
+/**
+ * Model of a conflict request that no need to be updated or saved, in the request queue.
+ */
+type ConflictRequestNoAction = {
+    /**
+     * The action to take in case of a conflict.
+     */
+    type: 'noAction';
+};
+
+/**
+ * An object that has the request and the action to take in case of a conflict.
+ */
+type ConflictActionData = {
+    /**
+     * The action to take in case of a conflict.
+     */
+    conflictAction: ConflictRequestUpdate | ConflictRequestSave | ConflictRequestNoAction;
+};
+
+/**
+ * An object that describes how a new write request can identify any queued requests that may conflict with or be undone by the new request,
+ * and how to resolve those conflicts.
+ */
+type RequestConflictResolver = {
+    /**
+     * A function that checks if a new request conflicts with any existing requests in the queue.
+     */
+    checkAndFixConflictingRequest?: (persistedRequest: Request[]) => ConflictActionData;
+};
+
 /** Model of requests sent to the API */
-type Request = RequestData & OnyxData;
+type Request = RequestData & OnyxData & RequestConflictResolver;
 
 /**
  * An object used to describe how a request can be paginated.
@@ -85,4 +141,4 @@ type PaginatedRequest = Request &
     };
 
 export default Request;
-export type {OnyxData, RequestType, PaginationConfig, PaginatedRequest};
+export type {OnyxData, RequestType, PaginationConfig, PaginatedRequest, RequestConflictResolver, ConflictActionData};

tests/unit/SequentialQueueTest.ts

@@ -0,0 +1,148 @@
+import Onyx from 'react-native-onyx';
+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 type {ConflictActionData} from '../../src/types/onyx/Request';
+import * as TestHelper from '../utils/TestHelper';
+import waitForBatchedUpdates from '../utils/waitForBatchedUpdates';
+
+const request: Request = {
+    command: 'ReconnectApp',
+    successData: [{key: 'userMetadata', onyxMethod: 'set', value: {accountID: 1234}}],
+    failureData: [{key: 'userMetadata', onyxMethod: 'set', value: {}}],
+};
+
+describe('SequentialQueue', () => {
+    beforeAll(() => {
+        Onyx.init({
+            keys: ONYXKEYS,
+        });
+    });
+    beforeEach(() => {
+        global.fetch = TestHelper.getGlobalFetchMock();
+        return Onyx.clear().then(waitForBatchedUpdates);
+    });
+
+    it('should push one request and persist one', () => {
+        SequentialQueue.push(request);
+        expect(PersistedRequests.getLength()).toBe(1);
+    });
+
+    it('should push two requests and persist two', () => {
+        SequentialQueue.push(request);
+        SequentialQueue.push(request);
+        expect(PersistedRequests.getLength()).toBe(2);
+    });
+
+    it('should push two requests with conflict resolution and replace', () => {
+        SequentialQueue.push(request);
+        const requestWithConflictResolution: Request = {
+            command: 'ReconnectApp',
+            data: {accountID: 56789},
+            checkAndFixConflictingRequest: (persistedRequests) => {
+                // should be one instance of ReconnectApp, get the index to replace it later
+                const index = persistedRequests.findIndex((r) => r.command === 'ReconnectApp');
+                if (index === -1) {
+                    return {conflictAction: {type: 'push'}};
+                }
+
+                return {
+                    conflictAction: {type: 'replace', index},
+                };
+            },
+        };
+        SequentialQueue.push(requestWithConflictResolution);
+        expect(PersistedRequests.getLength()).toBe(1);
+    });
+
+    it('should push two requests with conflict resolution and push', () => {
+        SequentialQueue.push(request);
+        const requestWithConflictResolution: Request = {
+            command: 'ReconnectApp',
+            data: {accountID: 56789},
+            checkAndFixConflictingRequest: () => {
+                return {conflictAction: {type: 'push'}};
+            },
+        };
+        SequentialQueue.push(requestWithConflictResolution);
+        expect(PersistedRequests.getLength()).toBe(2);
+    });
+
+    it('should push two requests with conflict resolution and noAction', () => {
+        SequentialQueue.push(request);
+        const requestWithConflictResolution: Request = {
+            command: 'ReconnectApp',
+            data: {accountID: 56789},
+            checkAndFixConflictingRequest: () => {
+                return {conflictAction: {type: 'noAction'}};
+            },
+        };
+        SequentialQueue.push(requestWithConflictResolution);
+        expect(PersistedRequests.getLength()).toBe(1);
+    });
+
+    it('should add a new request even if a similar one is ongoing', async () => {
+        // .push at the end flush the queue
+        SequentialQueue.push(request);
+
+        // wait for Onyx.connect execute the callback and start processing the queue
+        await Promise.resolve();
+
+        const requestWithConflictResolution: Request = {
+            command: 'ReconnectApp',
+            data: {accountID: 56789},
+            checkAndFixConflictingRequest: (persistedRequests) => {
+                // should be one instance of ReconnectApp, get the index to replace it later
+                const index = persistedRequests.findIndex((r) => r.command === 'ReconnectApp');
+                if (index === -1) {
+                    return {conflictAction: {type: 'push'}};
+                }
+
+                return {
+                    conflictAction: {type: 'replace', index},
+                };
+            },
+        };
+
+        SequentialQueue.push(requestWithConflictResolution);
+        expect(PersistedRequests.getLength()).toBe(2);
+    });
+
+    it('should replace request request in queue while a similar one is ongoing', async () => {
+        // .push at the end flush the queue
+        SequentialQueue.push(request);
+
+        // wait for Onyx.connect execute the callback and start processing the queue
+        await Promise.resolve();
+
+        const conflicyResolver = (persistedRequests: Request[]): ConflictActionData => {
+            // should be one instance of ReconnectApp, get the index to replace it later
+            const index = persistedRequests.findIndex((r) => r.command === 'ReconnectApp');
+            if (index === -1) {
+                return {conflictAction: {type: 'push'}};
+            }
+
+            return {
+                conflictAction: {type: 'replace', index},
+            };
+        };
+
+        const requestWithConflictResolution: Request = {
+            command: 'ReconnectApp',
+            data: {accountID: 56789},
+            checkAndFixConflictingRequest: conflicyResolver,
+        };
+
+        const requestWithConflictResolution2: Request = {
+            command: 'ReconnectApp',
+            data: {accountID: 56789},
+            checkAndFixConflictingRequest: conflicyResolver,
+        };
+
+        SequentialQueue.push(requestWithConflictResolution);
+        SequentialQueue.push(requestWithConflictResolution2);
+
+        expect(PersistedRequests.getLength()).toBe(2);
+    });
+});