fix(alm): prevent zombie listeners caused by forked tasks

package.json

@@ -45,7 +45,7 @@
   "scripts": {
     "build": "yarn build:packages",
     "test": "yarn test:packages",
-    "build:examples": "yarn workspaces foreach --include '@reduxjs/*' --include '@examples-query-react/*' -vtp run build",
+    "build:examples": "yarn workspaces foreach --include '@reduxjs/*' --include '@examples-query-react/*' --include '@examples-action-listener/*' -vtp run build",
     "build:docs": "yarn workspace website run build",
     "build:packages": "yarn workspaces foreach --include '@reduxjs/*' --include '@rtk-query/*' --include '@rtk-incubator/*' --topological-dev run build",
     "test:packages": "yarn workspaces foreach --include '@reduxjs/*' --include '@rtk-query/*'  --include '@rtk-incubator/*' run test",

packages/action-listener-middleware/README.md

@@ -285,7 +285,7 @@ Both these methods are cancelation-aware, and will throw a `TaskAbortError` if t
 
 - `fork: (executor: (forkApi: ForkApi) => T | Promise<T>) => ForkedTask<T>`: Launches a "child task" that may be used to accomplish additional work. Accepts any sync or async function as its argument, and returns a `{result, cancel}` object that can be used to check the final status and return value of the child task, or cancel it while in-progress.
 
-Child tasks can be launched, and waited on to collect their return values. The provided `executor` function will be called with a `forkApi` object containing `{pause, delay, signal}`, allowing it to pause or check cancelation status. It can also make use of the `listenerApi` from the listener's scope.
+Child tasks can be launched, and waited on to collect their return values. The provided `executor` function will be called asynchronously with a `forkApi` object containing `{pause, delay, signal}`, allowing it to pause or check cancelation status. It can also make use of the `listenerApi` from the listener's scope.
 
 An example of this might be a listener that forks a child task containing an infinite loop that listens for events from a server. The parent then uses `listenerApi.condition()` to wait for a "stop" action, and cancels the child task.
 

packages/action-listener-middleware/src/index.ts

@@ -28,6 +28,7 @@ import type {
   ForkedTask,
   TypedRemoveListener,
   TypedStopListening,
+  TaskResult,
 } from './types'
 import {
   abortControllerWithReason,
@@ -85,9 +86,6 @@ const createFork = (parentAbortSignal: AbortSignal) => {
   return <T>(taskExecutor: ForkedTaskExecutor<T>): ForkedTask<T> => {
     assertFunction(taskExecutor, 'taskExecutor')
     const childAbortController = new AbortController()
-    const cancel = () => {
-      abortControllerWithReason(childAbortController, taskCancelled)
-    }
 
     const result = runTask<T>(
       async (): Promise<T> => {
@@ -98,16 +96,17 @@ const createFork = (parentAbortSignal: AbortSignal) => {
           delay: createDelay(childAbortController.signal),
           signal: childAbortController.signal,
         })) as T
-        validateActive(parentAbortSignal)
         validateActive(childAbortController.signal)
         return result
       },
       () => abortControllerWithReason(childAbortController, taskCompleted)
     )
 
     return {
-      result,
-      cancel,
+      result: createPause<TaskResult<T>>(parentAbortSignal)(result),
+      cancel() {
+        abortControllerWithReason(childAbortController, taskCancelled)
+      },
     }
   }
 }

packages/action-listener-middleware/src/task.ts

@@ -37,8 +37,9 @@ export const promisifyAbortSignal = (
 
 /**
  * Runs a task and returns promise that resolves to {@link TaskResult}.
- *
  * Second argument is an optional `cleanUp` function that always runs after task.
+ *
+ * **Note:** `runTask` runs the executor in the next microtask.
  * @returns
  */
 export const runTask = async <T>(

packages/action-listener-middleware/src/tests/fork.test.ts

@@ -1,5 +1,5 @@
 import type { EnhancedStore } from '@reduxjs/toolkit'
-import { configureStore, createSlice } from '@reduxjs/toolkit'
+import { configureStore, createSlice, createAction } from '@reduxjs/toolkit'
 
 import type { PayloadAction } from '@reduxjs/toolkit'
 import type { ForkedTaskExecutor, TaskResult } from '../types'
@@ -99,24 +99,23 @@ describe('fork', () => {
     expect(hasRunAsyncExecutor).toBe(true)
   })
 
-  it('runs forked tasks that are cancelled if parent listener is cancelled', async () => {
+  test('forkedTask.result rejects TaskAbortError if listener is cancelled', async () => {
     const deferredForkedTaskError = deferred()
 
     startListening({
       actionCreator: increment,
-      effect: async (_, listenerApi) => {
+      async effect(_, listenerApi) {
         listenerApi.cancelActiveListeners()
-        const result = await listenerApi.fork(async () => {
-          await delay(20)
-
-          throw new Error('unreachable code')
-        }).result
+        listenerApi
+          .fork(async () => {
+            await delay(10)
 
-        if (result.status !== 'ok') {
-          deferredForkedTaskError.resolve(result.error)
-        } else {
-          deferredForkedTaskError.reject(new Error('unreachable code'))
-        }
+            throw new Error('unreachable code')
+          })
+          .result.then(
+            deferredForkedTaskError.resolve,
+            deferredForkedTaskError.resolve
+          )
       },
     })
 
@@ -386,26 +385,25 @@ describe('fork', () => {
   })
 
   test('forkApi.pause rejects if listener is cancelled', async () => {
-    let deferredResult = deferred()
+    const incrementByInListener = createAction<number>('incrementByInListener')
+
     startListening({
-      actionCreator: increment,
-      effect: async (_, listenerApi) => {
+      actionCreator: incrementByInListener,
+      async effect({ payload: amountToIncrement }, listenerApi) {
         listenerApi.cancelActiveListeners()
-        const forkedTask = listenerApi.fork(async (forkApi) => {
-          await forkApi.pause(delay(30))
-
-          return 4
-        })
-        deferredResult.resolve(await forkedTask.result)
+        await listenerApi.fork(async (forkApi) => {
+          await forkApi.pause(delay(10))
+          listenerApi.dispatch(incrementByAmount(amountToIncrement))
+        }).result
+        listenerApi.dispatch(incrementByAmount(2 * amountToIncrement))
       },
     })
 
-    store.dispatch(increment())
-    store.dispatch(increment())
+    store.dispatch(incrementByInListener(10))
+    store.dispatch(incrementByInListener(100))
 
-    expect(await deferredResult).toEqual({
-      status: 'cancelled',
-      error: new TaskAbortError(listenerCancelled),
-    })
+    await delay(50)
+
+    expect(store.getState().value).toEqual(300)
   })
 })

packages/action-listener-middleware/src/tests/listenerMiddleware.test.ts

@@ -709,24 +709,27 @@ describe('createListenerMiddleware', () => {
     })
 
     test('clear() cancels all running forked tasks', async () => {
-      const fork1Test = deferred()
+      const store = configureStore({
+        reducer: counterSlice.reducer,
+        middleware: (gDM) => gDM().prepend(middleware),
+      })
 
       startListening({
         actionCreator: testAction1,
-        async effect(_, { fork }) {
-          const taskResult = await fork(() => {
-            return 3
-          }).result
-          fork1Test.resolve(taskResult)
+        async effect(_, { fork, dispatch }) {
+          await fork(() => dispatch(incrementByAmount(3))).result
+          dispatch(incrementByAmount(4))
         },
       })
 
+      expect(store.getState().value).toBe(0)
       store.dispatch(testAction1('a'))
 
       clearListeners()
-      store.dispatch(testAction1('b'))
 
-      expect(await fork1Test).toHaveProperty('status', 'cancelled')
+      await Promise.resolve() // Forked tasks run on the next microtask.
+
+      expect(store.getState().value).toBe(0)
     })
   })
 

packages/action-listener-middleware/src/types.ts

@@ -105,6 +105,19 @@ export type TaskResult<Value> =
   | TaskCancelled
 
 export interface ForkedTask<T> {
+  /**
+   * A promise that resolves when the task is either completed or cancelled or rejects
+   * if parent listener execution is cancelled or completed.
+   *
+   * ### Example
+   * ```ts
+   * const result = await fork(async (forkApi) => Promise.resolve(4)).result
+   *
+   * if(result.status === 'ok') {
+   *   console.log(result.value) // logs 4
+   * }}
+   * ```
+   */
   result: Promise<TaskResult<T>>
   /**
    * Cancel task if it is in progress or not yet started,