fix: abort federation composition (#6723)

Author: n1ru4l

packages/libraries/core/package.json

@@ -46,7 +46,7 @@
   },
   "dependencies": {
     "@graphql-tools/utils": "^10.0.0",
-    "@whatwg-node/fetch": "^0.10.1",
+    "@whatwg-node/fetch": "^0.10.5",
     "async-retry": "^1.3.3",
     "lodash.sortby": "^4.7.0",
     "tiny-lru": "^8.0.2"

packages/libraries/yoga/package.json

@@ -56,9 +56,9 @@
     "@graphql-yoga/plugin-disable-introspection": "2.7.0",
     "@graphql-yoga/plugin-graphql-sse": "3.7.0",
     "@graphql-yoga/plugin-response-cache": "3.9.0",
-    "@whatwg-node/fetch": "0.10.1",
+    "@whatwg-node/fetch": "0.10.5",
     "graphql-ws": "5.16.1",
-    "graphql-yoga": "5.10.8",
+    "graphql-yoga": "5.13.3",
     "nock": "14.0.0",
     "vitest": "3.0.5",
     "ws": "8.18.0"

packages/libraries/yoga/tests/yoga.spec.ts

@@ -882,7 +882,7 @@ describe('subscription usage reporting', () => {
             :
 
             event: next
-            data: {"errors":[{"message":"Unexpected error.","locations":[{"line":1,"column":1}]}]}
+            data: {"errors":[{"message":"Unexpected error.","locations":[{"line":1,"column":1}],"extensions":{"code":"INTERNAL_SERVER_ERROR"}}]}
 
             event: complete
             data:

packages/migrations/package.json

@@ -21,7 +21,7 @@
     "@types/bcryptjs": "2.4.6",
     "@types/node": "22.10.5",
     "@types/pg": "8.11.10",
-    "@whatwg-node/fetch": "0.10.1",
+    "@whatwg-node/fetch": "0.10.5",
     "bcryptjs": "2.4.3",
     "copyfiles": "2.4.1",
     "date-fns": "4.1.0",

packages/services/api/package.json

@@ -16,6 +16,7 @@
     "@bentocache/plugin-prometheus": "0.2.0",
     "@date-fns/utc": "2.1.0",
     "@graphql-hive/core": "workspace:*",
+    "@graphql-hive/signal": "1.0.0",
     "@graphql-inspector/core": "5.1.0-alpha-20231208113249-34700c8a",
     "@graphql-tools/merge": "9.0.24",
     "@hive/cdn-script": "workspace:*",
@@ -56,7 +57,7 @@
     "graphql-modules": "3.0.0",
     "graphql-parse-resolve-info": "4.13.0",
     "graphql-scalars": "1.24.0",
-    "graphql-yoga": "5.10.8",
+    "graphql-yoga": "5.13.3",
     "ioredis": "5.4.2",
     "ioredis-mock": "8.9.0",
     "lodash": "4.17.21",

packages/services/api/src/modules/schema/providers/orchestrators/federation.ts

@@ -1,4 +1,5 @@
 import { CONTEXT, Inject, Injectable, Scope } from 'graphql-modules';
+import { abortSignalAny } from '@graphql-hive/signal';
 import type { ContractsInputType, SchemaBuilderApi } from '@hive/schema';
 import { traceFn } from '@hive/service-common';
 import { createTRPCProxyClient, httpLink } from '@trpc/client';
@@ -19,6 +20,7 @@ export class FederationOrchestrator implements Orchestrator {
   type = ProjectType.FEDERATION;
   private logger: Logger;
   private schemaService;
+  private incomingRequestAbortSignal: AbortSignal;
 
   constructor(
     logger: Logger,
@@ -37,6 +39,7 @@ export class FederationOrchestrator implements Orchestrator {
         }),
       ],
     });
+    this.incomingRequestAbortSignal = context.request.signal;
   }
 
   private createConfig(config: Project['externalComposition']): ExternalCompositionConfig {
@@ -82,18 +85,47 @@ export class FederationOrchestrator implements Orchestrator {
       'Composing and Validating Federated Schemas (method=%s)',
       config.native ? 'native' : config.external.enabled ? 'external' : 'v1',
     );
-    const result = await this.schemaService.composeAndValidate.mutate({
-      type: 'federation',
-      schemas: schemas.map(s => ({
-        raw: s.raw,
-        source: s.source,
-        url: s.url ?? null,
-      })),
-      external: this.createConfig(config.external),
-      native: config.native,
-      contracts: config.contracts,
-    });
+    const timeoutAbortSignal = AbortSignal.timeout(30_000);
+
+    const onTimeout = () => {
+      this.logger.debug('Composition HTTP request aborted due to timeout of 30 seconds.');
+    };
+    timeoutAbortSignal.addEventListener('abort', onTimeout);
+
+    const onIncomingRequestAbort = () => {
+      this.logger.debug('Composition HTTP request aborted due to incoming request being canceled.');
+    };
+    this.incomingRequestAbortSignal.addEventListener('abort', onIncomingRequestAbort);
 
-    return result;
+    try {
+      const result = await this.schemaService.composeAndValidate.mutate(
+        {
+          type: 'federation',
+          schemas: schemas.map(s => ({
+            raw: s.raw,
+            source: s.source,
+            url: s.url ?? null,
+          })),
+          external: this.createConfig(config.external),
+          native: config.native,
+          contracts: config.contracts,
+        },
+        {
+          // We want to abort composition if the request that does the composition is aborted
+          // We also limit the maximum time allowed for composition requests to 30 seconds to avoid
+          //
+          // The reason for these is a potential dead-lock.
+          //
+          // Note: We are using `abortSignalAny` over `AbortSignal.any` because of leak issues.
+          // @source https://github.com/nodejs/node/issues/57584
+          signal: abortSignalAny([this.incomingRequestAbortSignal, timeoutAbortSignal]),
+        },
+      );
+
+      return result;
+    } finally {
+      timeoutAbortSignal.removeEventListener('abort', onTimeout);
+      this.incomingRequestAbortSignal.removeEventListener('abort', onIncomingRequestAbort);
+    }
   }
 }

packages/services/api/src/modules/shared/providers/mutex.ts

@@ -108,7 +108,7 @@ export class Mutex {
 
     // We try to acquire the lock until the retry counter is exceeded or the lock as been successfully acquired.
     do {
-      logger.debug('Acquiring lock (id=%s, attempt=%n)', id, attemptCounter + 1);
+      logger.debug('Acquiring lock (id=%s, attempt=%d)', id, attemptCounter + 1);
 
       lockToAcquire = await Promise.race([
         // we avoid using any of the acquire settings for auto-extension, retrying, etc.
@@ -171,7 +171,7 @@ export class Mutex {
     }
 
     let extendTimeout: NodeJS.Timeout | undefined;
-    // we have a global timeout of 90 seconds to avoid dead-licks
+    // we have a global timeout of 90 seconds to avoid dead-locks
     const globalTimeout = setTimeout(() => {
       logger.error('Global lock timeout exceeded (id=%s)', id);
       void cleanup();
@@ -188,12 +188,10 @@ export class Mutex {
       clearTimeout(globalTimeout);
 
       extendTimeout = undefined;
-      if (lock.expiration > new Date().getTime()) {
-        void lock.release().catch(err => {
-          logger.error('Error while releasing lock (id=%s)', id);
-          console.error(err);
-        });
-      }
+      void lock.release().catch(err => {
+        logger.error('Error while releasing lock (id=%s)', id);
+        console.error(err);
+      });
     }
 
     async function extendLock(isInitial = false) {

packages/services/broker-worker/package.json

@@ -10,7 +10,7 @@
   "devDependencies": {
     "@cloudflare/workers-types": "4.20241230.0",
     "@types/service-worker-mock": "2.0.4",
-    "@whatwg-node/server": "0.9.65",
+    "@whatwg-node/server": "0.10.3",
     "esbuild": "0.25.0",
     "itty-router": "4.2.2",
     "toucan-js": "4.1.0",

packages/services/cdn-worker/package.json

@@ -19,7 +19,7 @@
   "devDependencies": {
     "@cloudflare/workers-types": "4.20241230.0",
     "@types/service-worker-mock": "2.0.4",
-    "@whatwg-node/server": "0.9.65",
+    "@whatwg-node/server": "0.10.3",
     "bcryptjs": "2.4.3",
     "dotenv": "16.4.7",
     "esbuild": "0.25.0",

packages/services/demo/federation/package.json

@@ -8,7 +8,7 @@
   "dependencies": {
     "@apollo/subgraph": "2.9.3",
     "graphql": "16.9.0",
-    "graphql-yoga": "5.10.8"
+    "graphql-yoga": "5.13.3"
   },
   "devDependencies": {
     "wrangler": "3.107.2"