fix(executor): do not use leaking registerAbortSignalListener, and handle listeners inside the execution context

Author: ardatan

.changeset/swift-geese-behave.md

@@ -0,0 +1,7 @@
+---
+'@graphql-tools/executor': patch
+'@graphql-tools/utils': patch
+---
+
+In executor, do not use leaking `registerAbortSignalListener`, and handle listeners inside the
+execution context

packages/executor/src/execution/__tests__/abort-signal.test.ts

@@ -34,7 +34,7 @@ describe('Abort Signal', () => {
             subscribe() {
               return new Repeater(async (push, stop) => {
                 let i = 0;
-                stop.then(() => {
+                stop.finally(() => {
                   stopped = true;
                 });
 
@@ -150,7 +150,7 @@ describe('Abort Signal', () => {
             didInvokeFirstFn = true;
             return true;
           },
-          async second() {
+          second() {
             didInvokeSecondFn = true;
             controller.abort();
             return true;
@@ -162,18 +162,21 @@ describe('Abort Signal', () => {
         },
       },
     });
-    const result$ = normalizedExecutor({
-      schema,
-      document: parse(/* GraphQL */ `
-        mutation {
-          first
-          second
-          third
-        }
-      `),
-      signal: controller.signal,
-    });
-    expect(result$).rejects.toBeInstanceOf(DOMException);
+    await expect(
+      Promise.resolve().then(() =>
+        normalizedExecutor({
+          schema,
+          document: parse(/* GraphQL */ `
+            mutation {
+              first
+              second
+              third
+            }
+          `),
+          signal: controller.signal,
+        }),
+      ),
+    ).rejects.toBeInstanceOf(DOMException);
     expect(didInvokeFirstFn).toBe(true);
     expect(didInvokeSecondFn).toBe(true);
     expect(didInvokeThirdFn).toBe(false);

packages/executor/src/execution/__tests__/stream-test.ts

@@ -576,8 +576,9 @@ describe('Execute: stream directive', () => {
             path: ['friendList', 2],
           },
         ],
-        hasNext: false,
+        hasNext: true,
       },
+      { hasNext: false },
     ]);
   });
 
@@ -645,10 +646,10 @@ describe('Execute: stream directive', () => {
               path: ['friendList', 2],
             },
           ],
-          hasNext: false,
+          hasNext: true,
         },
       },
-      { done: true, value: undefined },
+      { done: false, value: { hasNext: false } },
       { done: true, value: undefined },
     ]);
   });

packages/executor/src/execution/execute.ts

@@ -34,7 +34,6 @@ import {
   collectFields,
   createGraphQLError,
   fakePromise,
-  getAbortPromise,
   getArgumentValues,
   getDefinedRootType,
   GraphQLResolveInfo,
@@ -52,11 +51,10 @@ import {
   Path,
   pathToArray,
   promiseReduce,
-  registerAbortSignalListener,
 } from '@graphql-tools/utils';
 import { TypedDocumentNode } from '@graphql-typed-document-node/core';
 import { DisposableSymbols } from '@whatwg-node/disposablestack';
-import { handleMaybePromise } from '@whatwg-node/promise-helpers';
+import { createDeferredPromise, handleMaybePromise } from '@whatwg-node/promise-helpers';
 import { coerceError } from './coerceError.js';
 import { flattenAsyncIterable } from './flattenAsyncIterable.js';
 import { invariant } from './invariant.js';
@@ -127,6 +125,8 @@ export interface ExecutionContext<TVariables = any, TContext = any> {
   errors: Array<GraphQLError>;
   subsequentPayloads: Set<AsyncPayloadRecord>;
   signal?: AbortSignal;
+  onSignalAbort?(handler: () => void): void;
+  signalPromise?: Promise<never>;
 }
 
 export interface FormattedExecutionResult<
@@ -421,6 +421,8 @@ export function buildExecutionContext<TData = any, TVariables = any, TContext =
     signal,
   } = args;
 
+  signal?.throwIfAborted();
+
   // If the schema used for execution is invalid, throw an error.
   assertValidSchema(schema);
 
@@ -489,6 +491,31 @@ export function buildExecutionContext<TData = any, TVariables = any, TContext =
     return coercedVariableValues.errors;
   }
 
+  signal?.throwIfAborted();
+
+  let onSignalAbort: ExecutionContext['onSignalAbort'];
+  let signalPromise: ExecutionContext['signalPromise'];
+
+  if (signal) {
+    const listeners = new Set<() => void>();
+    const signalDeferred = createDeferredPromise<never>();
+    signalPromise = signalDeferred.promise;
+    const sharedListener = () => {
+      signalDeferred.reject(signal.reason);
+      signal.removeEventListener('abort', sharedListener);
+    };
+    signal.addEventListener('abort', sharedListener, { once: true });
+    signalPromise.catch(() => {
+      for (const listener of listeners) {
+        listener();
+      }
+      listeners.clear();
+    });
+    onSignalAbort = handler => {
+      listeners.add(handler);
+    };
+  }
+
   return {
     schema,
     fragments,
@@ -502,6 +529,8 @@ export function buildExecutionContext<TData = any, TVariables = any, TContext =
     subsequentPayloads: new Set(),
     errors: [],
     signal,
+    onSignalAbort,
+    signalPromise,
   };
 }
 
@@ -626,9 +655,9 @@ function executeFields(
       }
     }
   } catch (error) {
-    if (containsPromise) {
+    if (error !== exeContext.signal?.reason && containsPromise) {
       // Ensure that any promises returned by other fields are handled, as they may also reject.
-      return promiseForObject(results, exeContext.signal).finally(() => {
+      return promiseForObject(results, exeContext.signal, exeContext.signalPromise).finally(() => {
         throw error;
       });
     }
@@ -643,7 +672,7 @@ function executeFields(
   // Otherwise, results is a map from field name to the result of resolving that
   // field, which is possibly a promise. Return a promise that will return this
   // same map, but with any promises replaced with the values they resolved to.
-  return promiseForObject(results, exeContext.signal);
+  return promiseForObject(results, exeContext.signal, exeContext.signalPromise);
 }
 
 /**
@@ -673,6 +702,7 @@ function executeField(
 
   // Get the resolve function, regardless of if its result is normal or abrupt (error).
   try {
+    exeContext.signal?.throwIfAborted();
     // Build a JS object of arguments from the field.arguments AST, using the
     // variables scope to fulfill any variable references.
     // TODO: find a way to memoize, in case this field is within a List type.
@@ -967,8 +997,9 @@ async function completeAsyncIteratorValue(
   iterator: AsyncIterator<unknown>,
   asyncPayloadRecord?: AsyncPayloadRecord,
 ): Promise<ReadonlyArray<unknown>> {
-  if (exeContext.signal && iterator.return) {
-    registerAbortSignalListener(exeContext.signal, () => {
+  exeContext.signal?.throwIfAborted();
+  if (iterator.return) {
+    exeContext.onSignalAbort?.(() => {
       iterator.return?.();
     });
   }
@@ -1746,18 +1777,25 @@ function executeSubscription(exeContext: ExecutionContext): MaybePromise<AsyncIt
     const result = resolveFn(rootValue, args, contextValue, info);
 
     if (isPromise(result)) {
-      return result.then(assertEventStream).then(undefined, error => {
-        throw locatedError(error, fieldNodes, pathToArray(path));
-      });
+      return result
+        .then(result => assertEventStream(result, exeContext.signal, exeContext.onSignalAbort))
+        .then(undefined, error => {
+          throw locatedError(error, fieldNodes, pathToArray(path));
+        });
     }
 
-    return assertEventStream(result, exeContext.signal);
+    return assertEventStream(result, exeContext.signal, exeContext.onSignalAbort);
   } catch (error) {
     throw locatedError(error, fieldNodes, pathToArray(path));
   }
 }
 
-function assertEventStream(result: unknown, signal?: AbortSignal): AsyncIterable<unknown> {
+function assertEventStream(
+  result: unknown,
+  signal?: AbortSignal,
+  onSignalAbort?: (handler: () => void) => void,
+): AsyncIterable<unknown> {
+  signal?.throwIfAborted();
   if (result instanceof Error) {
     throw result;
   }
@@ -1768,13 +1806,13 @@ function assertEventStream(result: unknown, signal?: AbortSignal): AsyncIterable
       'Subscription field must return Async Iterable. ' + `Received: ${inspect(result)}.`,
     );
   }
-  if (signal) {
+  if (onSignalAbort) {
     return {
       [Symbol.asyncIterator]() {
         const asyncIterator = result[Symbol.asyncIterator]();
 
         if (asyncIterator.return) {
-          registerAbortSignalListener(signal, () => {
+          onSignalAbort?.(() => {
             asyncIterator.return?.();
           });
         }
@@ -2101,8 +2139,6 @@ function yieldSubsequentPayloads(
 ): AsyncGenerator<SubsequentIncrementalExecutionResult, void, void> {
   let isDone = false;
 
-  const abortPromise = exeContext.signal ? getAbortPromise(exeContext.signal) : undefined;
-
   async function next(): Promise<IteratorResult<SubsequentIncrementalExecutionResult, void>> {
     if (isDone) {
       return { value: undefined, done: true };
@@ -2112,8 +2148,8 @@ function yieldSubsequentPayloads(
       record => record.promise,
     );
 
-    if (abortPromise) {
-      await Promise.race([abortPromise, ...subSequentPayloadPromises]);
+    if (exeContext.signalPromise) {
+      await Promise.race([exeContext.signalPromise, ...subSequentPayloadPromises]);
     } else {
       await Promise.race(subSequentPayloadPromises);
     }

packages/executor/src/execution/promiseForObject.ts

@@ -1,4 +1,4 @@
-import { getAbortPromise } from '@graphql-tools/utils';
+import { isPromise } from '@graphql-tools/utils';
 
 type ResolvedObject<TData> = {
   [TKey in keyof TData]: TData[TKey] extends Promise<infer TValue> ? TValue : TData[TKey];
@@ -11,19 +11,30 @@ type ResolvedObject<TData> = {
  * This is akin to bluebird's `Promise.props`, but implemented only using
  * `Promise.all` so it will work with any implementation of ES6 promises.
  */
-export async function promiseForObject<TData>(
+export function promiseForObject<TData>(
   object: TData,
   signal?: AbortSignal,
+  signalPromise?: Promise<never>,
 ): Promise<ResolvedObject<TData>> {
+  signal?.throwIfAborted();
   const resolvedObject = Object.create(null);
-  const promises = Promise.all(
-    Object.entries(object as any).map(async ([key, value]) => {
-      resolvedObject[key] = await value;
-    }),
-  );
-  if (signal) {
-    const abortPromise = getAbortPromise(signal);
-    return Promise.race([abortPromise, promises]).then(() => resolvedObject);
+  const promises: Promise<void>[] = [];
+  for (const key in object) {
+    const value = object[key];
+    if (isPromise(value)) {
+      promises.push(
+        value.then(value => {
+          signal?.throwIfAborted();
+          resolvedObject[key] = value;
+        }),
+      );
+    } else {
+      resolvedObject[key] = value;
+    }
   }
-  return promises.then(() => resolvedObject);
+  const promiseAll = Promise.all(promises);
+  if (signalPromise) {
+    return Promise.race([signalPromise, promiseAll]).then(() => resolvedObject);
+  }
+  return promiseAll.then(() => resolvedObject);
 }

packages/utils/src/registerAbortSignalListener.ts

@@ -1,3 +1,4 @@
+import { fakeRejectPromise } from '@whatwg-node/promise-helpers';
 import { memoize1 } from './memoize.js';
 
 // AbortSignal handler cache to avoid the "possible EventEmitter memory leak detected"
@@ -32,8 +33,11 @@ export function registerAbortSignalListener(signal: AbortSignal, listener: () =>
 }
 
 export const getAbortPromise = memoize1(function getAbortPromise(signal: AbortSignal) {
+  // If the signal is already aborted, return a rejected promise
+  if (signal.aborted) {
+    return fakeRejectPromise(signal.reason);
+  }
   return new Promise<void>((_resolve, reject) => {
-    // If the signal is already aborted, return a rejected promise
     if (signal.aborted) {
       reject(signal.reason);
       return;