Merge pull request #3021 from reduxjs/pr/fix-signal-race-memleak

Author: markerikson

packages/toolkit/src/listenerMiddleware/index.ts

@@ -39,10 +39,10 @@ import {
 } from './exceptions'
 import {
   runTask,
-  promisifyAbortSignal,
   validateActive,
   createPause,
   createDelay,
+  raceWithSignal,
 } from './task'
 export { TaskAbortError } from './exceptions'
 export type {
@@ -138,9 +138,9 @@ const createTakePattern = <S>(
     // Placeholder unsubscribe function until the listener is added
     let unsubscribe: UnsubscribeListener = () => {}
 
-    const tuplePromise = new Promise<[AnyAction, S, S]>((resolve) => {
+    const tuplePromise = new Promise<[AnyAction, S, S]>((resolve, reject) => {
       // Inside the Promise, we synchronously add the listener.
-      unsubscribe = startListening({
+      let stopListening = startListening({
         predicate: predicate as any,
         effect: (action, listenerApi): void => {
           // One-shot listener that cleans up as soon as the predicate passes
@@ -153,10 +153,13 @@ const createTakePattern = <S>(
           ])
         },
       })
+      unsubscribe = () => {
+        stopListening()
+        reject()
+      }
     })
 
     const promises: (Promise<null> | Promise<[AnyAction, S, S]>)[] = [
-      promisifyAbortSignal(signal),
       tuplePromise,
     ]
 
@@ -167,7 +170,7 @@ const createTakePattern = <S>(
     }
 
     try {
-      const output = await Promise.race(promises)
+      const output = await raceWithSignal(signal, Promise.race(promises))
 
       validateActive(signal)
       return output

packages/toolkit/src/listenerMiddleware/task.ts

@@ -1,6 +1,6 @@
 import { TaskAbortError } from './exceptions'
 import type { AbortSignalWithReason, TaskResult } from './types'
-import { addAbortSignalListener, catchRejection } from './utils'
+import { addAbortSignalListener, catchRejection, noop } from './utils'
 
 /**
  * Synchronously raises {@link TaskAbortError} if the task tied to the input `signal` has been cancelled.
@@ -15,24 +15,29 @@ export const validateActive = (signal: AbortSignal): void => {
 }
 
 /**
- * Returns a promise that will reject {@link TaskAbortError} if the task is cancelled.
- * @param signal
- * @returns
+ * Generates a race between the promise(s) and the AbortSignal
+ * This avoids `Promise.race()`-related memory leaks:
+ * https://github.com/nodejs/node/issues/17469#issuecomment-349794909
  */
-export const promisifyAbortSignal = (
-  signal: AbortSignalWithReason<string>
-): Promise<never> => {
-  return catchRejection(
-    new Promise<never>((_, reject) => {
-      const notifyRejection = () => reject(new TaskAbortError(signal.reason))
+export function raceWithSignal<T>(
+  signal: AbortSignalWithReason<string>,
+  promise: Promise<T>
+): Promise<T> {
+  let cleanup = noop
+  return new Promise<T>((resolve, reject) => {
+    const notifyRejection = () => reject(new TaskAbortError(signal.reason))
+
+    if (signal.aborted) {
+      notifyRejection()
+      return
+    }
 
-      if (signal.aborted) {
-        notifyRejection()
-      } else {
-        addAbortSignalListener(signal, notifyRejection)
-      }
-    })
-  )
+    cleanup = addAbortSignalListener(signal, notifyRejection)
+    promise.finally(() => cleanup()).then(resolve, reject)
+  }).finally(() => {
+    // after this point, replace `cleanup` with a noop, so there is no reference to `signal` any more
+    cleanup = noop
+  })
 }
 
 /**
@@ -73,7 +78,7 @@ export const runTask = async <T>(
 export const createPause = <T>(signal: AbortSignal) => {
   return (promise: Promise<T>): Promise<T> => {
     return catchRejection(
-      Promise.race([promisifyAbortSignal(signal), promise]).then((output) => {
+      raceWithSignal(signal, promise).then((output) => {
         validateActive(signal)
         return output
       })

packages/toolkit/src/listenerMiddleware/utils.ts

@@ -28,6 +28,7 @@ export const addAbortSignalListener = (
   callback: (evt: Event) => void
 ) => {
   abortSignal.addEventListener('abort', callback, { once: true })
+  return () => abortSignal.removeEventListener('abort', callback)
 }
 
 /**