Merge pull request Expensify#85920 from callstack-internal/callstack-internal/szymonzalarski/fix-seqquential-queue-issues-after-revert

Author: mjasikowski

src/libs/API/index.ts

@@ -53,7 +53,10 @@ addMiddleware(SaveResponseInOnyx);
 // FraudMonitoring - Tags the request with the appropriate Fraud Protection event.
 addMiddleware(FraudMonitoring);
 
-let requestIndex = 0;
+// Use timestamp-based IDs to avoid collisions between browser tabs.
+// Each tab has its own JS context with its own counter, so a simple
+// incrementing number would collide across tabs.
+let requestIndex = Date.now();
 
 /**
  * Prepare the request to be sent. Bind data together with request metadata and apply optimistic Onyx data.
@@ -122,13 +125,13 @@ function prepareRequest<TCommand extends ApiCommand, TKey extends OnyxKey>(
 /**
  * Process a prepared request according to its type.
  */
-function processRequest<TKey extends OnyxKey>(request: OnyxRequest<TKey>, type: ApiRequestType): Promise<void | Response<TKey>> {
+async function processRequest<TKey extends OnyxKey>(request: OnyxRequest<TKey>, type: ApiRequestType): Promise<void | Response<TKey>> {
     Log.info('[API] Processing request', false, {command: request.command, type});
     // Write commands can be saved and retried, so push it to the SequentialQueue
     if (type === CONST.API_REQUEST_TYPE.WRITE) {
         Log.info('[API] Write command. Pushing to SequentialQueue', false, {command: request.command});
-        pushToSequentialQueue(request);
-        return Promise.resolve();
+        await pushToSequentialQueue(request);
+        return;
     }
 
     // Read requests are processed right away, but don't return the response to the caller
@@ -164,6 +167,7 @@ function write<TCommand extends WriteCommand, TKey extends OnyxKey>(
 ): Promise<void | Response<TKey>> {
     Log.info('[API] Called API write', false, {command, ...apiCommandParameters});
     const request = prepareRequest(command, CONST.API_REQUEST_TYPE.WRITE, apiCommandParameters, onyxData, conflictResolver);
+
     return processRequest(request, CONST.API_REQUEST_TYPE.WRITE);
 }
 

src/libs/Network/SequentialQueue.ts

@@ -6,6 +6,7 @@ import {
     endRequestAndRemoveFromQueue as endPersistedRequestAndRemoveFromQueue,
     getAll as getAllPersistedRequests,
     getCommands,
+    onCrossTabRequestsMerged as onPersistedRequestsCrossTabMerge,
     onInitialization as onPersistedRequestsInitialization,
     processNextRequest as processNextPersistedRequest,
     rollbackOngoingRequest as rollbackOngoingPersistedRequest,
@@ -436,7 +437,10 @@ onReconnection(flush);
 // Flush the queue when the persisted requests are initialized
 onPersistedRequestsInitialization(flush);
 
-function handleConflictActions<TKey extends OnyxKey>(conflictAction: ConflictData, newRequest: OnyxRequest<TKey>) {
+// Flush the queue when another tab enqueues new requests
+onPersistedRequestsCrossTabMerge(flush);
+
+async function handleConflictActions<TKey extends OnyxKey>(conflictAction: ConflictData, newRequest: OnyxRequest<TKey>): Promise<void> {
     Log.info('[SequentialQueue] handleConflictActions', false, {
         conflictType: conflictAction.type,
         newCommand: newRequest.command,
@@ -447,34 +451,34 @@ function handleConflictActions<TKey extends OnyxKey>(conflictAction: ConflictDat
         Log.info('[SequentialQueue] Conflict resolution: PUSH', false, {
             command: newRequest.command,
         });
-        savePersistedRequest(newRequest);
+        await savePersistedRequest(newRequest);
     } else if (conflictAction.type === 'replace') {
         Log.info('[SequentialQueue] Conflict resolution: REPLACE', false, {
             command: newRequest.command,
             replaceIndex: conflictAction.index,
             replacementRequest: conflictAction.request?.command ?? newRequest.command,
         });
-        updatePersistedRequest(conflictAction.index, conflictAction.request ?? (newRequest as AnyRequest));
+        await updatePersistedRequest(conflictAction.index, conflictAction.request ?? (newRequest as AnyRequest));
     } else if (conflictAction.type === 'delete') {
         Log.info('[SequentialQueue] Conflict resolution: DELETE', false, {
             command: newRequest.command,
             deleteIndices: conflictAction.indices,
             willPushNewRequest: conflictAction.pushNewRequest ?? false,
             hasNextAction: !!conflictAction.nextAction,
         });
-        deletePersistedRequestsByIndices(conflictAction.indices);
+        await deletePersistedRequestsByIndices(conflictAction.indices);
         if (conflictAction.pushNewRequest) {
             Log.info('[SequentialQueue] Pushing new request after delete', false, {
                 command: newRequest.command,
             });
-            savePersistedRequest(newRequest);
+            await savePersistedRequest(newRequest);
         }
         if (conflictAction.nextAction) {
             Log.info('[SequentialQueue] Processing next conflict action', false, {
                 command: newRequest.command,
                 nextActionType: conflictAction.nextAction.type,
             });
-            handleConflictActions(conflictAction.nextAction, newRequest);
+            await handleConflictActions(conflictAction.nextAction, newRequest);
         }
     } else {
         Log.info('[SequentialQueue] No action performed, request ignored', false, {
@@ -484,7 +488,7 @@ function handleConflictActions<TKey extends OnyxKey>(conflictAction: ConflictDat
     }
 }
 
-function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>) {
+function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>): Promise<void> {
     const currentRequests = getAllPersistedRequests();
     Log.info('[SequentialQueue] push() called', false, {
         command: newRequest.command,
@@ -494,6 +498,11 @@ function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>) {
         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.
+    let persistencePromise: Promise<void>;
+
     if (newRequest.checkAndFixConflictingRequest) {
         const requests = currentRequests;
         Log.info('[SequentialQueue] Checking for conflicts', false, {
@@ -510,13 +519,13 @@ function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>) {
         // don't try to serialize a function.
         // eslint-disable-next-line no-param-reassign
         delete newRequest.checkAndFixConflictingRequest;
-        handleConflictActions(conflictAction, newRequest);
+        persistencePromise = handleConflictActions(conflictAction, newRequest);
     } else {
         Log.info('[SequentialQueue] No conflict action. Adding request to Persisted Requests', false, {
             command: newRequest.command,
         });
         // Add request to Persisted Requests so that it can be retried if it fails
-        savePersistedRequest(newRequest);
+        persistencePromise = savePersistedRequest(newRequest);
     }
 
     // If we are offline we don't need to trigger the queue to empty as it will happen when we come back online
@@ -525,7 +534,7 @@ function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>) {
             command: newRequest.command,
             queueLength: getAllPersistedRequests().length,
         });
-        return;
+        return persistencePromise;
     }
 
     // If the queue is running this request will run once it has finished processing the current batch
@@ -539,13 +548,14 @@ function push<TKey extends OnyxKey>(newRequest: OnyxRequest<TKey>) {
             });
             flush(true);
         });
-        return;
+        return persistencePromise;
     }
 
     Log.info('[SequentialQueue] Queue is not running. Flushing the queue.', false, {
         command: newRequest.command,
     });
     flush(true);
+    return persistencePromise;
 }
 
 function getCurrentRequest(): Promise<void> {