fix(core): make sure pausedMutations always execute in serial (#4906)

* fix: make sure pausedMutations always execute in serial

when resumePausedMutations is called multiple times, we are starting the second paused mutation even though the first one has not yet finished. This can happen when a focus event and an online event happen in short succession

* fix: always return the same Promise when resumePausedMutations is called

* fix: make sure every call to resumePausedMutations is evaluated

mutation.continue() needs to immediately resolve back if we didn't really continue so that consecutive calls to resumePausedMutations can be appended to the promise chain

* fix: types

Author: TkDodo

packages/query-core/src/mutation.ts

@@ -159,12 +159,8 @@ export class Mutation<
     }
   }
 
-  continue(): Promise<TData> {
-    if (this.retryer) {
-      this.retryer.continue()
-      return this.retryer.promise
-    }
-    return this.execute()
+  continue(): Promise<unknown> {
+    return this.retryer?.continue() ?? this.execute()
   }
 
   async execute(): Promise<TData> {

packages/query-core/src/mutationCache.ts

@@ -79,6 +79,7 @@ export class MutationCache extends Subscribable<MutationCacheListener> {
 
   private mutations: Mutation<any, any, any, any>[]
   private mutationId: number
+  private resuming: Promise<unknown> | undefined
 
   constructor(config?: MutationCacheConfig) {
     super()
@@ -152,14 +153,22 @@ export class MutationCache extends Subscribable<MutationCacheListener> {
     })
   }
 
-  resumePausedMutations(): Promise<void> {
-    const pausedMutations = this.mutations.filter((x) => x.state.isPaused)
-    return notifyManager.batch(() =>
-      pausedMutations.reduce(
-        (promise, mutation) =>
-          promise.then(() => mutation.continue().catch(noop)),
-        Promise.resolve(),
-      ),
-    )
+  resumePausedMutations(): Promise<unknown> {
+    this.resuming = (this.resuming ?? Promise.resolve())
+      .then(() => {
+        const pausedMutations = this.mutations.filter((x) => x.state.isPaused)
+        return notifyManager.batch(() =>
+          pausedMutations.reduce(
+            (promise, mutation) =>
+              promise.then(() => mutation.continue().catch(noop)),
+            Promise.resolve() as Promise<unknown>,
+          ),
+        )
+      })
+      .then(() => {
+        this.resuming = undefined
+      })
+
+    return this.resuming
   }
 }

packages/query-core/src/queryClient.ts

@@ -594,7 +594,7 @@ export class QueryClient {
       .catch(noop)
   }
 
-  resumePausedMutations(): Promise<void> {
+  resumePausedMutations(): Promise<unknown> {
     return this.mutationCache.resumePausedMutations()
   }
 

packages/query-core/src/retryer.ts

@@ -21,7 +21,7 @@ interface RetryerConfig<TData = unknown, TError = unknown> {
 export interface Retryer<TData = unknown> {
   promise: Promise<TData>
   cancel: (cancelOptions?: CancelOptions) => void
-  continue: () => void
+  continue: () => Promise<unknown>
   cancelRetry: () => void
   continueRetry: () => void
 }
@@ -69,7 +69,7 @@ export function createRetryer<TData = unknown, TError = unknown>(
   let isRetryCancelled = false
   let failureCount = 0
   let isResolved = false
-  let continueFn: ((value?: unknown) => void) | undefined
+  let continueFn: ((value?: unknown) => boolean) | undefined
   let promiseResolve: (data: TData) => void
   let promiseReject: (error: TError) => void
 
@@ -118,9 +118,11 @@ export function createRetryer<TData = unknown, TError = unknown>(
   const pause = () => {
     return new Promise((continueResolve) => {
       continueFn = (value) => {
-        if (isResolved || !shouldPause()) {
-          return continueResolve(value)
+        const canContinue = isResolved || !shouldPause()
+        if (canContinue) {
+          continueResolve(value)
         }
+        return canContinue
       }
       config.onPause?.()
     }).then(() => {
@@ -208,7 +210,8 @@ export function createRetryer<TData = unknown, TError = unknown>(
     promise,
     cancel,
     continue: () => {
-      continueFn?.()
+      const didContinue = continueFn?.()
+      return didContinue ? promise : Promise.resolve()
     },
     cancelRetry,
     continueRetry,

packages/query-core/src/tests/queryClient.test.tsx

@@ -8,8 +8,9 @@ import type {
   QueryFunction,
   QueryObserverOptions,
 } from '..'
-import { InfiniteQueryObserver, QueryObserver } from '..'
+import { InfiniteQueryObserver, MutationObserver, QueryObserver } from '..'
 import { focusManager, onlineManager } from '..'
+import { noop } from '../utils'
 
 describe('queryClient', () => {
   let queryClient: QueryClient
@@ -1468,6 +1469,81 @@ describe('queryClient', () => {
       onlineManager.setOnline(undefined)
     })
 
+    test('should resume paused mutations when coming online', async () => {
+      const consoleMock = jest.spyOn(console, 'error')
+      consoleMock.mockImplementation(() => undefined)
+      onlineManager.setOnline(false)
+
+      const observer1 = new MutationObserver(queryClient, {
+        mutationFn: async () => 1,
+      })
+
+      const observer2 = new MutationObserver(queryClient, {
+        mutationFn: async () => 2,
+      })
+      void observer1.mutate().catch(noop)
+      void observer2.mutate().catch(noop)
+
+      await waitFor(() => {
+        expect(observer1.getCurrentResult().isPaused).toBeTruthy()
+        expect(observer2.getCurrentResult().isPaused).toBeTruthy()
+      })
+
+      onlineManager.setOnline(true)
+
+      await waitFor(() => {
+        expect(observer1.getCurrentResult().status).toBe('success')
+        expect(observer1.getCurrentResult().status).toBe('success')
+      })
+
+      onlineManager.setOnline(undefined)
+    })
+
+    test('should resume paused mutations one after the other when invoked manually at the same time', async () => {
+      const consoleMock = jest.spyOn(console, 'error')
+      consoleMock.mockImplementation(() => undefined)
+      onlineManager.setOnline(false)
+
+      const orders: Array<string> = []
+
+      const observer1 = new MutationObserver(queryClient, {
+        mutationFn: async () => {
+          orders.push('1start')
+          await sleep(50)
+          orders.push('1end')
+          return 1
+        },
+      })
+
+      const observer2 = new MutationObserver(queryClient, {
+        mutationFn: async () => {
+          orders.push('2start')
+          await sleep(20)
+          orders.push('2end')
+          return 2
+        },
+      })
+      void observer1.mutate().catch(noop)
+      void observer2.mutate().catch(noop)
+
+      await waitFor(() => {
+        expect(observer1.getCurrentResult().isPaused).toBeTruthy()
+        expect(observer2.getCurrentResult().isPaused).toBeTruthy()
+      })
+
+      onlineManager.setOnline(undefined)
+      void queryClient.resumePausedMutations()
+      await sleep(5)
+      await queryClient.resumePausedMutations()
+
+      await waitFor(() => {
+        expect(observer1.getCurrentResult().status).toBe('success')
+        expect(observer2.getCurrentResult().status).toBe('success')
+      })
+
+      expect(orders).toEqual(['1start', '1end', '2start', '2end'])
+    })
+
     test('should notify queryCache and mutationCache after multiple mounts and single unmount', async () => {
       const testClient = createQueryClient()
       testClient.mount()