Disposable improvements (#7401)

* Improve Disposable usage and use `@whatwg-node/disposablestack` as ponyfill

* Husky

* chore(dependencies): updated changesets for modified dependencies

* Small change

* Hmm

* chore(dependencies): updated changesets for modified dependencies

* Remove disposablestack patch

* Fix leak

* E2E fixes

* Fix

* Fix types

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

.changeset/@graphql-mesh_cache-redis-7401-dependencies.md

@@ -0,0 +1,5 @@
+---
+"@graphql-mesh/cache-redis": patch
+---
+dependencies updates:
+  - Added dependency [`@whatwg-node/disposablestack@^0.0.1` ↗︎](https://www.npmjs.com/package/@whatwg-node/disposablestack/v/0.0.1) (to `dependencies`)

.changeset/@graphql-mesh_fusion-runtime-7401-dependencies.md

@@ -0,0 +1,6 @@
+---
+"@graphql-mesh/fusion-runtime": patch
+---
+dependencies updates:
+  - Added dependency [`@whatwg-node/disposablestack@^0.0.1` ↗︎](https://www.npmjs.com/package/@whatwg-node/disposablestack/v/0.0.1) (to `dependencies`)
+  - Removed dependency [`disposablestack@^1.1.6` ↗︎](https://www.npmjs.com/package/disposablestack/v/1.1.6) (from `dependencies`)

.changeset/@graphql-mesh_plugin-hive-7401-dependencies.md

@@ -0,0 +1,5 @@
+---
+"@graphql-mesh/plugin-hive": patch
+---
+dependencies updates:
+  - Added dependency [`@graphql-mesh/utils@^0.99.4` ↗︎](https://www.npmjs.com/package/@graphql-mesh/utils/v/0.99.4) (to `peerDependencies`)

.changeset/@graphql-mesh_serve-runtime-7401-dependencies.md

@@ -0,0 +1,6 @@
+---
+"@graphql-mesh/serve-runtime": patch
+---
+dependencies updates:
+  - Added dependency [`@whatwg-node/disposablestack@^0.0.1` ↗︎](https://www.npmjs.com/package/@whatwg-node/disposablestack/v/0.0.1) (to `dependencies`)
+  - Removed dependency [`disposablestack@^1.1.6` ↗︎](https://www.npmjs.com/package/disposablestack/v/1.1.6) (from `dependencies`)

.changeset/@graphql-mesh_transport-http-callback-7401-dependencies.md

@@ -0,0 +1,5 @@
+---
+"@graphql-mesh/transport-http-callback": patch
+---
+dependencies updates:
+  - Added dependency [`@graphql-mesh/utils@^0.99.4` ↗︎](https://www.npmjs.com/package/@graphql-mesh/utils/v/0.99.4) (to `dependencies`)

.changeset/@graphql-mesh_transport-ws-7401-dependencies.md

@@ -0,0 +1,5 @@
+---
+"@graphql-mesh/transport-ws": patch
+---
+dependencies updates:
+  - Added dependency [`@graphql-mesh/utils@^0.99.4` ↗︎](https://www.npmjs.com/package/@graphql-mesh/utils/v/0.99.4) (to `dependencies`)

.changeset/@graphql-mesh_utils-7401-dependencies.md

@@ -0,0 +1,6 @@
+---
+"@graphql-mesh/utils": patch
+---
+dependencies updates:
+  - Added dependency [`@whatwg-node/disposablestack@^0.0.1` ↗︎](https://www.npmjs.com/package/@whatwg-node/disposablestack/v/0.0.1) (to `dependencies`)
+  - Removed dependency [`disposablestack@^1.1.6` ↗︎](https://www.npmjs.com/package/disposablestack/v/1.1.6) (from `dependencies`)

.husky/pre-commit

@@ -1,4 +1 @@
-#!/bin/sh
-. "$(dirname "$0")/_/husky.sh"
-
 yarn lint-staged

declarations.d.ts

@@ -18,9 +18,4 @@ declare module '@newrelic/test-utilities' {
   export const TestAgent: any;
 }
 
-declare module 'disposablestack/AsyncDisposableStack' {
-  declare var AsyncDisposableStackCtor: typeof AsyncDisposableStack;
-  export = AsyncDisposableStackCtor;
-}
-
 declare var __VERSION__: string;

e2e/federation-subscriptions-passthrough/federation-subscriptions-passthrough.test.ts

@@ -1,8 +1,6 @@
-import { createServer } from 'http';
-import type { AddressInfo } from 'net';
 import { setTimeout } from 'timers/promises';
 import { createClient, type Client } from 'graphql-sse';
-import { createTenv } from '@e2e/tenv';
+import { createTenv, getAvailablePort } from '@e2e/tenv';
 import { TOKEN } from './services/products/server';
 
 const { composeWithApollo, service, serve } = createTenv(__dirname);
@@ -142,10 +140,7 @@ it('should subscribe and resolve via http callbacks', async () => {
   ]);
 
   // Get a random available port
-  const dummyServer = createServer();
-  await new Promise<void>(resolve => dummyServer.listen(0, resolve));
-  const availablePort = (dummyServer.address() as AddressInfo).port;
-  await new Promise(resolve => dummyServer.close(resolve));
+  const availablePort = await getAvailablePort();
 
   const publicUrl = `http://0.0.0.0:${availablePort}`;
   await serve({

e2e/utils/leftoverStack.ts

@@ -1,12 +1,9 @@
-import AsyncDisposableStack from 'disposablestack/AsyncDisposableStack';
+import { AsyncDisposableStack } from '@whatwg-node/disposablestack';
 
-let leftoverStack = new AsyncDisposableStack();
+export let leftoverStack = new AsyncDisposableStack();
 
-export function getLeftoverStack() {
-  if (leftoverStack.disposed) {
+afterAll(() =>
+  leftoverStack.disposeAsync().finally(() => {
     leftoverStack = new AsyncDisposableStack();
-  }
-  return leftoverStack;
-}
-
-afterAll(() => leftoverStack.disposeAsync());
+  }),
+);

e2e/utils/tbench.ts

@@ -1,6 +1,6 @@
 import { setTimeout } from 'timers/promises';
 import { spawn, Thread, Worker } from 'threads';
-import { getLeftoverStack } from './leftoverStack';
+import { leftoverStack } from './leftoverStack';
 import { timeout as jestTimeout, type Server } from './tenv';
 import type { benchGraphQLServer } from './workers/benchGraphQLServer';
 
@@ -51,7 +51,7 @@ export async function createTbench(vusCount: number): Promise<Tbench> {
       .map(() => spawn<typeof benchGraphQLServer>(new Worker('./workers/benchGraphQLServer.js'))),
   );
   vus.forEach(worker => {
-    getLeftoverStack().defer(() => Thread.terminate(worker));
+    leftoverStack.defer(() => Thread.terminate(worker));
   });
   return {
     async sustain({ server, duration = jestTimeout - 10_000, parallelRequestsPerVU = 10, params }) {

e2e/utils/tenv.ts

@@ -13,8 +13,9 @@ import {
   RemoteGraphQLDataSource,
   type ServiceEndpointDefinition,
 } from '@apollo/gateway';
+import { DisposableSymbols } from '@whatwg-node/disposablestack';
 import { createArg, createPortArg, createServicePortArg } from './args';
-import { getLeftoverStack } from './leftoverStack';
+import { leftoverStack } from './leftoverStack';
 
 export const retries = 120,
   interval = 500,
@@ -210,7 +211,7 @@ export function createTenv(cwd: string): Tenv {
       },
       async tempfile(name) {
         const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), 'graphql-mesh_e2e_fs'));
-        getLeftoverStack().defer(() => fs.rm(tempDir, { recursive: true }));
+        leftoverStack.defer(() => fs.rm(tempDir, { recursive: true }));
         return path.join(tempDir, name);
       },
       write(filePath, content) {
@@ -340,7 +341,7 @@ export function createTenv(cwd: string): Tenv {
       let output = '';
       if (opts?.output) {
         const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), 'graphql-mesh_e2e_compose'));
-        getLeftoverStack().defer(() => fs.rm(tempDir, { recursive: true }));
+        leftoverStack.defer(() => fs.rm(tempDir, { recursive: true }));
         output = path.join(tempDir, `${Math.random().toString(32).slice(2)}.${opts.output}`);
       }
       const [proc, waitForExit] = await spawn(
@@ -501,7 +502,7 @@ export function createTenv(cwd: string): Tenv {
         getStats() {
           throw new Error('Cannot get stats of a container.');
         },
-        async [Symbol.asyncDispose]() {
+        async [DisposableSymbols.asyncDispose]() {
           if (ctrl.signal.aborted) {
             // noop if already disposed
             return;
@@ -510,7 +511,7 @@ export function createTenv(cwd: string): Tenv {
           await ctr.stop({ t: 0 });
         },
       };
-      getLeftoverStack().use(container);
+      leftoverStack.use(container);
 
       // verify that the container has started
       await setTimeout(interval);
@@ -540,13 +541,13 @@ export function createTenv(cwd: string): Tenv {
           }
 
           if (status === 'none') {
-            await container[Symbol.asyncDispose]();
+            await container[DisposableSymbols.asyncDispose]();
             throw new DockerError(
               'Container has "none" health status, but has a healthcheck',
               container,
             );
           } else if (status === 'unhealthy') {
-            await container[Symbol.asyncDispose]();
+            await container[DisposableSymbols.asyncDispose]();
             throw new DockerError('Container is unhealthy', container);
           } else if (status === 'healthy') {
             break;
@@ -652,9 +653,9 @@ function spawn(
         mem: parseFloat(mem) * 0.001, // KB to MB
       };
     },
-    [Symbol.asyncDispose]: () => (child.kill(), waitForExit),
+    [DisposableSymbols.asyncDispose]: () => (child.kill(), waitForExit),
   };
-  getLeftoverStack().use(proc);
+  leftoverStack.use(proc);
 
   child.stdout.on('data', x => {
     stdout += x.toString();

examples/federation-example/tests/polling.test.ts

@@ -1,7 +1,8 @@
-import { exec, execSync } from 'child_process';
-import { createReadStream, readFile, readFileSync, write, writeFileSync } from 'fs';
-import { createServer } from 'http';
+import { exec } from 'child_process';
+import { readFileSync } from 'fs';
 import { join } from 'path';
+import { createDisposableServer } from '../../../packages/testing/createDisposableServer';
+import { disposableExec } from '../../../packages/testing/disposableExec';
 
 jest.setTimeout(30000);
 async function findAvailableHostName() {
@@ -25,16 +26,12 @@ async function findAvailableHostName() {
   throw new Error('No available hostname found');
 }
 describe('Polling Test', () => {
-  let cleanupCallbacks: (() => void)[] = [];
-  afterAll(() => {
-    cleanupCallbacks.forEach(cb => cb());
-  });
   it('should pass', async () => {
     const cwd = join(__dirname, 'fixtures/polling');
     const supergraphSdlPath = join(cwd, 'supergraph.graphql');
     const supergraphSdl = readFileSync(supergraphSdlPath, 'utf-8');
     let changedSupergraph = false;
-    const supergraphSdlServer = createServer((req, res) => {
+    await using supergraphSdlServer = await createDisposableServer((req, res) => {
       res.statusCode = 200;
       res.setHeader('Content-Type', 'text/plain');
       console.log('Serving supergraph SDL');
@@ -44,8 +41,6 @@ describe('Polling Test', () => {
         res.end(supergraphSdl);
       }
     });
-    await new Promise<void>(resolve => supergraphSdlServer.listen(0, resolve));
-    cleanupCallbacks.push(() => supergraphSdlServer.close());
     const SUPERGRAPH_SOURCE = `http://localhost:${(supergraphSdlServer.address() as any).port}`;
     console.info('Supergraph SDL server is running on ' + SUPERGRAPH_SOURCE);
     const buildCmd = exec(`${join(__dirname, '../node_modules/.bin/mesh')} build`, {
@@ -63,14 +58,13 @@ describe('Polling Test', () => {
         }
       });
     });
-    const serveCmd = exec(`${join(__dirname, '../node_modules/.bin/mesh')} start`, {
+    using serveCmd = disposableExec(`${join(__dirname, '../node_modules/.bin/mesh')} start`, {
       cwd,
       env: {
         ...process.env,
         SUPERGRAPH_SOURCE,
       },
     });
-    cleanupCallbacks.push(() => serveCmd.kill());
     await new Promise<void>(resolve => {
       serveCmd.stderr?.on('data', function stderrListener(data: string) {
         console.log(data);

examples/json-schema-example/tests/artifacts.test.ts

@@ -1,31 +1,13 @@
-import { createServer } from 'http';
-import { AddressInfo } from 'net';
 import { join } from 'path';
 import { DEFAULT_CLI_PARAMS, serveMesh } from '@graphql-mesh/cli';
 import { fs } from '@graphql-mesh/cross-helpers';
 import { Logger } from '@graphql-mesh/types';
 import { fetch } from '@whatwg-node/fetch';
 import { TerminateHandler } from '../../../packages/legacy/utils/dist/typings/registerTerminateHandler';
+import { getAvailablePort } from '../../../packages/testing/getAvailablePort';
 
 const { readFile } = fs.promises;
 
-const getFreePort = () =>
-  new Promise<number>((resolve, reject) => {
-    const server = createServer();
-    server.once('error', reject);
-    server.listen(0, () => {
-      const port = (server.address() as AddressInfo)?.port;
-      server.closeAllConnections();
-      server.close(err => {
-        if (err) {
-          reject(err);
-        } else {
-          resolve(port);
-        }
-      });
-    });
-  });
-
 describe('Artifacts', () => {
   it('should execute queries', async () => {
     const { getBuiltMesh } = await import('../.mesh/index');
@@ -70,7 +52,7 @@ describe('Artifacts', () => {
         log: jest.fn(),
         child: jest.fn(() => mockLogger),
       };
-      const PORT = await getFreePort();
+      const PORT = await getAvailablePort();
       await serveMesh(
         {
           baseDir: join(__dirname, '..'),

packages/cache/redis/package.json

@@ -40,6 +40,7 @@
   },
   "dependencies": {
     "@graphql-mesh/string-interpolation": "0.5.5",
+    "@whatwg-node/disposablestack": "^0.0.1",
     "ioredis": "^5.3.2",
     "ioredis-mock": "^8.8.3"
   },

packages/cache/redis/src/index.ts

@@ -9,6 +9,7 @@ import type {
   MeshPubSub,
   YamlConfig,
 } from '@graphql-mesh/types';
+import { DisposableSymbols } from '@whatwg-node/disposablestack';
 
 function interpolateStrWithEnv(str: string): string {
   return stringInterpolator.parse(str, { env: process.env });
@@ -62,7 +63,7 @@ export default class RedisCache<V = string> implements KeyValueCache<V>, Disposa
     });
   }
 
-  [Symbol.dispose](): void {
+  [DisposableSymbols.dispose](): void {
     this.client.disconnect();
   }
 

packages/fusion/composition/tests/loaders.spec.ts

@@ -42,6 +42,7 @@ describe('Loaders', () => {
       plugins() {
         return [useCustomFetch(mockFetch)];
       },
+      logging: !!process.env.DEBUG,
     });
     const res = await runtime.fetch('/graphql', {
       method: 'POST',

packages/fusion/runtime/package.json

@@ -63,8 +63,8 @@
     "@graphql-tools/stitching-directives": "^3.0.2",
     "@graphql-tools/utils": "^10.2.3",
     "@graphql-tools/wrap": "^10.0.5",
+    "@whatwg-node/disposablestack": "^0.0.1",
     "change-case": "^4.1.2",
-    "disposablestack": "^1.1.6",
     "graphql-yoga": "^5.6.0",
     "tslib": "^2.4.0"
   },

packages/fusion/runtime/src/unifiedGraphManager.ts

@@ -1,4 +1,3 @@
-import AsyncDisposableStack from 'disposablestack/AsyncDisposableStack';
 import type { DocumentNode, GraphQLSchema } from 'graphql';
 import { buildASTSchema, buildSchema, isSchema } from 'graphql';
 import { getInContextSDK } from '@graphql-mesh/runtime';
@@ -9,6 +8,7 @@ import { mapMaybePromise } from '@graphql-mesh/utils';
 import type { SubschemaConfig } from '@graphql-tools/delegate';
 import type { IResolvers, MaybePromise, TypeSource } from '@graphql-tools/utils';
 import { isDocumentNode } from '@graphql-tools/utils';
+import { AsyncDisposableStack, DisposableSymbols } from '@whatwg-node/disposablestack';
 import { compareSubgraphNames, handleFederationSupergraph } from './federation.js';
 import {
   compareSchemas,
@@ -203,7 +203,7 @@ export class UnifiedGraphManager<TContext> {
     return this.getAndSetUnifiedGraph();
   }
 
-  [Symbol.asyncDispose]() {
+  [DisposableSymbols.asyncDispose]() {
     return this.disposableStack.disposeAsync();
   }
 }

packages/fusion/runtime/src/utils.ts

@@ -11,6 +11,7 @@ import type {
 } from '@graphql-mesh/transport-common';
 import type { Logger } from '@graphql-mesh/types';
 import {
+  isDisposable,
   iterateAsync,
   loggerForExecutionRequest,
   mapMaybePromise,
@@ -399,7 +400,3 @@ export function compareSchemas(
   }
   return aStr === bStr;
 }
-
-export function isDisposable(obj: any): obj is Disposable | AsyncDisposable {
-  return obj?.[Symbol.dispose] != null || obj?.[Symbol.asyncDispose] != null;
-}

packages/fusion/runtime/tests/polling.test.ts

@@ -4,6 +4,7 @@ import { getUnifiedGraphGracefully } from '@graphql-mesh/fusion-composition';
 import { createDefaultExecutor, type DisposableExecutor } from '@graphql-mesh/transport-common';
 import { normalizedExecutor } from '@graphql-tools/executor';
 import { isAsyncIterable } from '@graphql-tools/utils';
+import { DisposableSymbols } from '@whatwg-node/disposablestack';
 import { UnifiedGraphManager } from '../src/unifiedGraphManager';
 
 describe('Polling', () => {
@@ -45,7 +46,7 @@ describe('Polling', () => {
         return {
           getSubgraphExecutor() {
             const executor: DisposableExecutor = createDefaultExecutor(schema);
-            executor[Symbol.asyncDispose] = disposeFn;
+            executor[DisposableSymbols.asyncDispose] = disposeFn;
             return executor;
           },
         };
@@ -96,7 +97,7 @@ describe('Polling', () => {
     // Check if transport executor is disposed per schema change
     expect(disposeFn).toHaveBeenCalledTimes(2);
 
-    await manager[Symbol.asyncDispose]();
+    await manager[DisposableSymbols.asyncDispose]();
     // Check if transport executor is disposed on global shutdown
     expect(disposeFn).toHaveBeenCalledTimes(3);
   });