Fix memory leaks on AbortSignal usage and improve memory consumption #596

.changeset/@graphql-hive_gateway-runtime-598-dependencies.md

@@ -0,0 +1,7 @@
+---
+'@graphql-hive/gateway-runtime': patch
+---
+
+dependencies updates: 
+
+- Removed dependency [`@graphql-hive/gateway-abort-signal-any@workspace:^` ↗︎](https://www.npmjs.com/package/@graphql-hive/gateway-abort-signal-any/v/workspace:^) (from `dependencies`)

.changeset/@graphql-mesh_transport-common-598-dependencies.md

@@ -0,0 +1,7 @@
+---
+'@graphql-mesh/transport-common': patch
+---
+
+dependencies updates: 
+
+- Removed dependency [`@graphql-hive/gateway-abort-signal-any@workspace:^` ↗︎](https://www.npmjs.com/package/@graphql-hive/gateway-abort-signal-any/v/workspace:^) (from `dependencies`)

.changeset/@graphql-tools_executor-http-598-dependencies.md

@@ -0,0 +1,7 @@
+---
+'@graphql-tools/executor-http': patch
+---
+
+dependencies updates: 
+
+- Removed dependency [`@graphql-hive/gateway-abort-signal-any@workspace:^` ↗︎](https://www.npmjs.com/package/@graphql-hive/gateway-abort-signal-any/v/workspace:^) (from `dependencies`)

.changeset/fair-queens-sniff.md

@@ -0,0 +1,7 @@
+---
+'@graphql-mesh/transport-http-callback': patch
+'@graphql-tools/executor-http': patch
+'@graphql-hive/gateway-runtime': patch
+---
+
+Use native AbortSignal, AbortController APIs instead of custom ones

packages/executors/http/package.json

@@ -39,7 +39,6 @@
     "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0"
   },
   "dependencies": {
-    "@graphql-hive/gateway-abort-signal-any": "workspace:^",
     "@graphql-tools/executor-common": "workspace:^",
     "@graphql-tools/utils": "^10.7.0",
     "@repeaterjs/repeater": "^3.0.4",

packages/executors/http/src/index.ts

@@ -1,4 +1,3 @@
-import { abortSignalAny } from '@graphql-hive/gateway-abort-signal-any';
 import {
   defaultPrintFn,
   SerializedExecutionRequest,
@@ -132,24 +131,6 @@ export interface HTTPExecutorOptions {
 
 export type HeadersConfig = Record<string, string>;
 
-// To prevent event listener warnings
-function createSignalWrapper(signal: AbortSignal): AbortSignal {
-  const listeners = new Set<EventListener>();
-  signal.onabort = (event) => {
-    for (const listener of listeners) {
-      listener(event);
-    }
-  };
-  return Object.assign(signal, {
-    addEventListener(_type: 'abort', listener: EventListener) {
-      listeners.add(listener);
-    },
-    removeEventListener(_type: 'abort', listener: EventListener) {
-      listeners.delete(listener);
-    },
-  });
-}
-
 export function buildHTTPExecutor(
   options?: Omit<HTTPExecutorOptions, 'fetch'> & {
     fetch: SyncFetchFn;
@@ -177,13 +158,12 @@ export function buildHTTPExecutor(
 ): DisposableExecutor<any, HTTPExecutorOptions> {
   const printFn = options?.print ?? defaultPrintFn;
   const disposeCtrl = new AbortController();
-  const sharedSignal = createSignalWrapper(disposeCtrl.signal);
   const baseExecutor = (
     request: ExecutionRequest<any, any, any, HTTPExecutorOptions>,
     excludeQuery?: boolean,
   ) => {
-    if (sharedSignal.aborted) {
-      return createResultForAbort(sharedSignal.reason);
+    if (disposeCtrl.signal.aborted) {
+      return createResultForAbort(disposeCtrl.signal.reason);
     }
     const fetchFn = request.extensions?.fetch ?? options?.fetch ?? defaultFetch;
     let method = request.extensions?.method || options?.method;
@@ -230,7 +210,7 @@ export function buildHTTPExecutor(
       request.extensions = restExtensions;
     }
 
-    const signals = [sharedSignal];
+    const signals = [disposeCtrl.signal];
     const signalFromRequest = request.signal || request.info?.signal;
     if (signalFromRequest) {
       if (signalFromRequest.aborted) {
@@ -242,7 +222,7 @@ export function buildHTTPExecutor(
       signals.push(AbortSignal.timeout(options.timeout));
     }
 
-    const signal = abortSignalAny(signals);
+    const signal = AbortSignal.any(signals);
 
     const upstreamErrorExtensions: UpstreamErrorExtensions = {
       request: {
@@ -489,8 +469,8 @@ export function buildHTTPExecutor(
       function retryAttempt():
         | PromiseLike<ExecutionResult<any>>
         | ExecutionResult<any> {
-        if (sharedSignal.aborted) {
-          return createResultForAbort(sharedSignal.reason);
+        if (disposeCtrl.signal.aborted) {
+          return createResultForAbort(disposeCtrl.signal.reason);
         }
         attempt++;
         if (attempt > options!.retry!) {

packages/runtime/package.json

@@ -54,7 +54,6 @@
     "@envelop/disable-introspection": "^6.0.0",
     "@envelop/generic-auth": "^8.0.0",
     "@graphql-hive/core": "^0.8.1",
-    "@graphql-hive/gateway-abort-signal-any": "workspace:^",
     "@graphql-mesh/cross-helpers": "^0.4.9",
     "@graphql-mesh/fusion-runtime": "workspace:^",
     "@graphql-mesh/hmac-upstream-signature": "workspace:^",

packages/runtime/src/plugins/useUpstreamCancel.ts

@@ -1,7 +1,3 @@
-import {
-  abortSignalAny,
-  isAbortSignalFromAny,
-} from '@graphql-hive/gateway-abort-signal-any';
 import { GraphQLResolveInfo } from '@graphql-tools/utils';
 import type { GatewayPlugin } from '../types';
 
@@ -21,14 +17,10 @@ export function useUpstreamCancel(): GatewayPlugin {
       if (signalInInfo) {
         signals.push(signalInInfo);
       }
-      if (isAbortSignalFromAny(options.signal)) {
-        options.signal.addSignals(signals);
-      } else {
-        if (options.signal) {
-          signals.push(options.signal);
-        }
-        options.signal = abortSignalAny(signals);
+      if (options.signal) {
+        signals.push(options.signal);
       }
+      options.signal = AbortSignal.any(signals);
     },
     onSubgraphExecute({ executionRequest }) {
       const signals: AbortSignal[] = [];
@@ -38,14 +30,10 @@ export function useUpstreamCancel(): GatewayPlugin {
       if (executionRequest.context?.request?.signal) {
         signals.push(executionRequest.context.request.signal);
       }
-      if (isAbortSignalFromAny(executionRequest.signal)) {
-        executionRequest.signal.addSignals(signals);
-      } else {
-        if (executionRequest.signal) {
-          signals.push(executionRequest.signal);
-        }
-        executionRequest.signal = abortSignalAny(signals);
+      if (executionRequest.signal) {
+        signals.push(executionRequest.signal);
       }
+      executionRequest.signal = AbortSignal.any(signals);
     },
   };
 }

packages/runtime/src/plugins/useUpstreamTimeout.ts

@@ -1,12 +1,11 @@
-import { abortSignalAny } from '@graphql-hive/gateway-abort-signal-any';
 import { subgraphNameByExecutionRequest } from '@graphql-mesh/fusion-runtime';
 import { UpstreamErrorExtensions } from '@graphql-mesh/transport-common';
 import { getHeadersObj } from '@graphql-mesh/utils';
 import {
+  createDeferred,
   createGraphQLError,
   ExecutionRequest,
   ExecutionResult,
-  getAbortPromise,
   isAsyncIterable,
   MaybeAsyncIterable,
   MaybePromise,
@@ -51,21 +50,28 @@ export function useUpstreamTimeout<TContext extends Record<string, any>>(
             timeoutSignalsByExecutionRequest.get(executionRequest);
           if (!timeoutSignal) {
             timeoutSignal = AbortSignal.timeout(timeout);
+            timeoutSignalsByExecutionRequest.set(
+              executionRequest,
+              timeoutSignal,
+            );
           }
-          timeoutSignalsByExecutionRequest.set(executionRequest, timeoutSignal);
-          const timeout$ = getAbortPromise(timeoutSignal);
-          let finalSignal: AbortSignal | undefined = timeoutSignal;
-          const signals = new Set<AbortSignal>();
-          signals.add(timeoutSignal);
+          const timeoutDeferred = createDeferred<void>();
+          function rejectDeferred() {
+            timeoutDeferred.reject(timeoutSignal?.reason);
+          }
+          timeoutSignal.addEventListener('abort', rejectDeferred, {
+            once: true,
+          });
+          const signals: AbortSignal[] = [];
+          signals.push(timeoutSignal);
           if (executionRequest.signal) {
-            signals.add(executionRequest.signal);
-            finalSignal = abortSignalAny(signals);
+            signals.push(executionRequest.signal);
           }
           return Promise.race([
-            timeout$,
+            timeoutDeferred.promise,
             executor({
               ...executionRequest,
-              signal: finalSignal,
+              signal: AbortSignal.any(signals),
             }),
           ])
             .then((result) => {
@@ -96,6 +102,8 @@ export function useUpstreamTimeout<TContext extends Record<string, any>>(
               throw e;
             })
             .finally(() => {
+              timeoutDeferred.resolve();
+              timeoutSignal.removeEventListener('abort', rejectDeferred);
               // Remove from the map after used so we don't see it again
               errorExtensionsByExecRequest.delete(executionRequest);
               timeoutSignalsByExecutionRequest.delete(executionRequest);
@@ -128,15 +136,15 @@ export function useUpstreamTimeout<TContext extends Record<string, any>>(
           } else {
             timeoutSignal = AbortSignal.timeout(timeout);
           }
-          const signals = new Set<AbortSignal>();
-          signals.add(timeoutSignal);
+          const signals: AbortSignal[] = [];
+          signals.push(timeoutSignal);
           if (options.signal) {
-            signals.add(options.signal);
-            setOptions({
-              ...options,
-              signal: abortSignalAny(signals),
-            });
+            signals.push(options.signal);
           }
+          setOptions({
+            ...options,
+            signal: AbortSignal.any(signals),
+          });
         }
       }
       if (executionRequest) {