fix(alm): cancel forkApi.delay and forkApi.pause if listener is cancelled or completed

packages/toolkit/src/listenerMiddleware/exceptions.ts

@@ -6,15 +6,15 @@ const completed = 'completed'
 const cancelled = 'cancelled'
 
 /* TaskAbortError error codes */
-export const taskCancelled = `${task}-${cancelled}` as const
-export const taskCompleted = `${task}-${completed}` as const
+export const taskCancelled = `task-${cancelled}` as const
+export const taskCompleted = `task-${completed}` as const
 export const listenerCancelled = `${listener}-${cancelled}` as const
 export const listenerCompleted = `${listener}-${completed}` as const
 
 export class TaskAbortError implements SerializedError {
  name = 'TaskAbortError'
- message = ''
- constructor(public code = 'unknown') {
- this.message = `task cancelled (reason: ${code})`
+ message: string
+ constructor(public code: string | undefined) {
+ this.message = `${task} ${cancelled} (reason: ${code})`
  }
 }

packages/toolkit/src/listenerMiddleware/index.ts

@@ -21,9 +21,11 @@ import type {
  ForkedTask,
  TypedRemoveListener,
  TaskResult,
+ AbortSignalWithReason,
 } from './types'
 import {
  abortControllerWithReason,
+ addAbortSignalListener,
  assertFunction,
  catchRejection,
 } from './utils'
@@ -74,11 +76,18 @@ const INTERNAL_NIL_TOKEN = {} as const
 
 const alm = 'listenerMiddleware' as const
 
-const createFork = (parentAbortSignal: AbortSignal) => {
+const createFork = (parentAbortSignal: AbortSignalWithReason<unknown>) => {
+ const linkControllers = (controller: AbortController) =>
+ addAbortSignalListener(parentAbortSignal, () =>
+ abortControllerWithReason(controller, parentAbortSignal.reason)
+ )
+
  return <T>(taskExecutor: ForkedTaskExecutor<T>): ForkedTask<T> => {
  assertFunction(taskExecutor, 'taskExecutor')
  const childAbortController = new AbortController()
 
+ linkControllers(childAbortController)
+
  const result = runTask<T>(
  async (): Promise<T> => {
  validateActive(parentAbortSignal)

packages/toolkit/src/listenerMiddleware/task.ts

@@ -1,6 +1,6 @@
 import { TaskAbortError } from './exceptions'
 import type { AbortSignalWithReason, TaskResult } from './types'
-import { catchRejection } from './utils'
+import { addAbortSignalListener, catchRejection } from './utils'
 
 /**
  * Synchronously raises {@link TaskAbortError} if the task tied to the input `signal` has been cancelled.
@@ -29,7 +29,7 @@ export const promisifyAbortSignal = (
  if (signal.aborted) {
  notifyRejection()
  } else {
- signal.addEventListener('abort', notifyRejection, { once: true })
+ addAbortSignalListener(signal, notifyRejection)
  }
  })
  )

packages/toolkit/src/listenerMiddleware/tests/fork.test.ts

@@ -2,9 +2,17 @@ import type { EnhancedStore } from '@reduxjs/toolkit'
 import { configureStore, createSlice, createAction } from '@reduxjs/toolkit'
 
 import type { PayloadAction } from '@reduxjs/toolkit'
-import type { ForkedTaskExecutor, TaskResult } from '../types'
+import type {
+ AbortSignalWithReason,
+ ForkedTaskExecutor,
+ TaskResult,
+} from '../types'
 import { createListenerMiddleware, TaskAbortError } from '../index'
-import { listenerCancelled, taskCancelled } from '../exceptions'
+import {
+ listenerCancelled,
+ listenerCompleted,
+ taskCancelled,
+} from '../exceptions'
 
 function delay(ms: number) {
  return new Promise((resolve) => setTimeout(resolve, ms))
@@ -312,6 +320,58 @@ describe('fork', () => {
  })
  })
 
+ test('forkApi.delay rejects as soon as the parent listener is cancelled', async () => {
+ let deferredResult = deferred()
+
+ startListening({
+ actionCreator: increment,
+ effect: async (_, listenerApi) => {
+ listenerApi.cancelActiveListeners()
+ await listenerApi.fork(async (forkApi) => {
+ await forkApi
+ .delay(100)
+ .then(deferredResult.resolve, deferredResult.resolve)
+
+ return 4
+ }).result
+
+ deferredResult.resolve(new Error('unreachable'))
+ },
+ })
+
+ store.dispatch(increment())
+
+ await Promise.resolve()
+
+ store.dispatch(increment())
+ expect(await deferredResult).toEqual(
+ new TaskAbortError(listenerCancelled)
+ )
+ })
+
+ test('forkApi.signal listener is invoked as soon as the parent listener is cancelled or completed', async () => {
+ let deferredResult = deferred()
+
+ startListening({
+ actionCreator: increment,
+ async effect(_, listenerApi) {
+ const wronglyDoNotAwaitResultOfTask = listenerApi.fork(
+ async (forkApi) => {
+ forkApi.signal.addEventListener('abort', () => {
+ deferredResult.resolve(
+ (forkApi.signal as AbortSignalWithReason<unknown>).reason
+ )
+ })
+ }
+ )
+ },
+ })
+
+ store.dispatch(increment)
+
+ expect(await deferredResult).toBe(listenerCompleted)
+ })
+
  test('fork.delay does not trigger unhandledRejections for completed or cancelled tasks', async () => {
  let deferredCompletedEvt = deferred()
  let deferredCancelledEvt = deferred()
@@ -384,6 +444,34 @@ describe('fork', () => {
  })
  })
 
+ test('forkApi.pause rejects as soon as the parent listener is cancelled', async () => {
+ let deferredResult = deferred()
+
+ startListening({
+ actionCreator: increment,
+ effect: async (_, listenerApi) => {
+ listenerApi.cancelActiveListeners()
+ const forkedTask = listenerApi.fork(async (forkApi) => {
+ await forkApi
+ .pause(delay(100))
+ .then(deferredResult.resolve, deferredResult.resolve)
+
+ return 4
+ })
+
+ await forkedTask.result
+ deferredResult.resolve(new Error('unreachable'))
+ },
+ })
+
+ store.dispatch(increment())
+
+ await Promise.resolve()
+
+ store.dispatch(increment())
+ expect(await deferredResult).toEqual(new TaskAbortError(listenerCancelled))
+ })
+
  test('forkApi.pause rejects if listener is cancelled', async () => {
  const incrementByInListener = createAction<number>('incrementByInListener')
 

packages/toolkit/src/listenerMiddleware/types.ts

@@ -54,12 +54,12 @@ export type MatchFunction<T> = (v: any) => v is T
 export interface ForkedTaskAPI {
  /**
  * Returns a promise that resolves when `waitFor` resolves or
- * rejects if the task has been cancelled or completed.
+ * rejects if the task or the parent listener has been cancelled or is completed.
  */
  pause<W>(waitFor: Promise<W>): Promise<W>
  /**
  * Returns a promise resolves after `timeoutMs` or
- * rejects if the task has been cancelled or is completed.
+ * rejects if the task or the parent listener has been cancelled or is completed.
  * @param timeoutMs
  */
  delay(timeoutMs: number): Promise<void>

packages/toolkit/src/listenerMiddleware/utils.ts

@@ -23,6 +23,13 @@ export const catchRejection = <T>(
  return promise
 }
 
+export const addAbortSignalListener = (
+ abortSignal: AbortSignal,
+ callback: (evt: Event) => void
+) => {
+ abortSignal.addEventListener('abort', callback, { once: true })
+}
+
 /**
  * Calls `abortController.abort(reason)` and patches `signal.reason`.
  * if it is not supported.