Loadtest E2Es and monitor memory for leak detection

.github/workflows/bench.yml

@@ -11,7 +11,6 @@ concurrency:
 
 env:
   NODE_NO_WARNINGS: 1
-  CI: true
 
 jobs:
   bench:
@@ -22,7 +21,7 @@ jobs:
           - 10
           - 100
           - 1000
-    name: Benchmark / ${{matrix.e2e_runner}} / ${{matrix.products_size}} items
+    name: ${{matrix.e2e_runner}} / ${{matrix.products_size}} items
     runs-on: ubuntu-latest
     steps:
       - name: Checkout
@@ -36,4 +35,3 @@ jobs:
         env:
           PRODUCTS_SIZE: ${{matrix.products_size}}
           E2E_GATEWAY_RUNNER: ${{matrix.e2e_runner}}
-          CI: true

.github/workflows/memtest.yml

@@ -0,0 +1,39 @@
+name: Memtest
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+env:
+  NODE_NO_WARNINGS: 1
+  K6_VERSION: v0.56.0
+
+jobs:
+  memtest:
+    strategy:
+      matrix:
+        e2e_runner: [node, bun]
+    name: ${{matrix.e2e_runner}}
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Install k6
+        run: |
+          mkdir -p "$HOME/.local/bin"
+          cd "$HOME/.local/bin"
+          curl https://github.com/grafana/k6/releases/download/${{ env.K6_VERSION }}/k6-${{ env.K6_VERSION }}-linux-amd64.tar.gz -L | tar xvz --strip-components 1
+          echo "$PWD" >> $GITHUB_PATH
+      - name: Set up env
+        uses: the-guild-org/shared-config/setup@v1
+        with:
+          node-version-file: .node-version
+      - name: Test
+        run: yarn test:mem
+        env:
+          E2E_GATEWAY_RUNNER: ${{matrix.e2e_runner}}

e2e/federation-example/federation-example.memtest.ts

@@ -0,0 +1,18 @@
+import { createExampleSetup, createTenv } from '@internal/e2e';
+import { memtest } from '@internal/perf/memtest';
+
+const cwd = __dirname;
+
+const { gateway } = createTenv(cwd);
+const { supergraph, query } = createExampleSetup(cwd);
+
+memtest(
+  {
+    cwd,
+    query,
+  },
+  async () =>
+    gateway({
+      supergraph: await supergraph(),
+    }),
+);

internal/e2e/src/tenv.ts

@@ -14,7 +14,7 @@ import {
   fakePromise,
   registerAbortSignalListener,
 } from '@graphql-tools/utils';
-import { Proc, ProcOptions, spawn, waitForPort } from '@internal/proc';
+import { Proc, ProcOptions, Server, spawn, waitForPort } from '@internal/proc';
 import {
   boolEnv,
   createOpt,
@@ -82,11 +82,6 @@ yarn build && E2E_GATEWAY_RUNNER=bun-docker yarn workspace @graphql-hive/gateway
   return runner as ServeRunner;
 })();
 
-export interface Server extends Proc {
-  port: number;
-  protocol: string;
-}
-
 export interface GatewayOptions extends ProcOptions {
   port?: number;
   /**
@@ -711,7 +706,12 @@ export function createTenv(cwd: string): Tenv {
         gatewayPort && createPortOpt(gatewayPort),
         ...args,
       );
-      const service: Service = { ...proc, name, port, protocol };
+      const service: Service = {
+        ...proc,
+        name,
+        port,
+        protocol,
+      };
       await Promise.race([
         waitForExit
           .then(() => {
@@ -870,6 +870,7 @@ export function createTenv(cwd: string): Tenv {
       await ctr.start();
 
       const container: Container = {
+        waitForExit: ctr.wait(),
         containerName,
         name,
         port: hostPort,

internal/examples/src/convert.ts

@@ -180,6 +180,8 @@ export async function convertE2EToExample(config: ConvertE2EToExampleConfig) {
       !path.basename(extraDirOrFile).includes('.e2e.') &&
       // not a bench
       !path.basename(extraDirOrFile).includes('.bench.') &&
+      // not a memtest
+      !path.basename(extraDirOrFile).includes('.memtest.') &&
       // not a dockerile
       !path.basename(extraDirOrFile).includes('Dockerfile') &&
       // not test snapshots

internal/perf/package.json

@@ -0,0 +1,14 @@
+{
+  "name": "@internal/perf",
+  "type": "module",
+  "private": true,
+  "main": "./src/index.ts",
+  "dependencies": {
+    "parse-duration": "^2.0.0",
+    "regression": "^2.0.1"
+  },
+  "devDependencies": {
+    "@types/k6": "^0.54.2",
+    "@types/regression": "^2"
+  }
+}

internal/perf/src/index.ts

@@ -0,0 +1 @@
+export * from './loadtest';

internal/perf/src/loadtest-script.ts

@@ -0,0 +1,22 @@
+import { check } from 'k6';
+import { test } from 'k6/execution';
+import http from 'k6/http';
+
+export default function () {
+  const url = __ENV['URL'];
+  if (!url) {
+    return test.abort('Environment variable "URL" not provided');
+  }
+
+  const query = __ENV['QUERY'];
+  if (!query) {
+    return test.abort('Environment variable "QUERY" not provided');
+  }
+
+  const res = http.post(url, { query });
+
+  check(res, {
+    'status is 200': (res) => res.status === 200,
+    'body contains data': (res) => !!res.body?.toString().includes('"data":{'),
+  });
+}

internal/perf/src/loadtest.ts

@@ -0,0 +1,99 @@
+import path from 'path';
+import { setTimeout } from 'timers/promises';
+import { ProcOptions, Server, spawn } from '@internal/proc';
+import { trimError } from '@internal/testing';
+
+export interface LoadtestOptions extends ProcOptions {
+  cwd: string;
+  /** @default 100 */
+  vus?: number;
+  /** Duration of the loadtest in milliseconds. */
+  duration: number;
+  /**
+   * The snapshotting window of the GraphQL server memory in milliseconds.
+   *
+   * @default 1_000
+   */
+  memorySnapshotWindow?: number;
+  /** The GraphQL server on which the loadtest is running. */
+  server: Server;
+  /**
+   * The GraphQL query to execute for the loadtest.
+   */
+  query: string;
+}
+
+export async function loadtest(opts: LoadtestOptions) {
+  const {
+    cwd,
+    vus = 100,
+    duration,
+    memorySnapshotWindow = 1_000,
+    server,
+    query,
+    ...procOptions
+  } = opts;
+
+  if (duration < 3_000) {
+    throw new Error(`Duration has to be at least 3s, got "${duration}"`);
+  }
+
+  const ctrl = new AbortController();
+  const signal = AbortSignal.any([
+    ctrl.signal,
+    AbortSignal.timeout(
+      duration +
+        // allow 1s for the k6 process to exit gracefully
+        1_000,
+    ),
+  ]);
+
+  const [, waitForExit] = await spawn(
+    {
+      cwd,
+      ...procOptions,
+      signal,
+    },
+    'k6',
+    'run',
+    `--vus=${vus}`,
+    `--duration=${duration}ms`,
+    `--env=URL=${server.protocol}://localhost:${server.port}/graphql`,
+    `--env=QUERY=${query}`,
+    path.join(__dirname, 'loadtest-script.ts'),
+  );
+  using _ = {
+    [Symbol.dispose]() {
+      ctrl.abort();
+    },
+  };
+
+  const memoryInMBSnapshots: number[] = [];
+
+  // we dont use a `setInterval` because the proc.getStats is async and we want stats ordered by time
+  (async () => {
+    while (!signal.aborted) {
+      await setTimeout(memorySnapshotWindow);
+      try {
+        const { mem } = await server.getStats();
+        memoryInMBSnapshots.push(mem);
+      } catch (err) {
+        if (!signal.aborted) {
+          throw err;
+        }
+        return; // couldve been aborted after timeout or while waiting for stats
+      }
+    }
+  })();
+
+  await Promise.race([
+    waitForExit,
+    server.waitForExit.then(() => {
+      throw new Error(
+        `Server exited before the loadtest finished\n${trimError(server.getStd('both'))}`,
+      );
+    }),
+  ]);
+
+  return { memoryInMBSnapshots };
+}

internal/perf/src/memtest.ts

@@ -0,0 +1,72 @@
+import { Server } from '@internal/proc';
+import regression from 'regression';
+import { it } from 'vitest';
+import { loadtest, LoadtestOptions } from './loadtest';
+
+export interface MemtestOptions
+  extends Omit<LoadtestOptions, 'duration' | 'server'> {
+  /**
+   * Duration of the loadtest in milliseconds.
+   *
+   * @default 60_000
+   */
+  duration?: number;
+  /**
+   * Linear regression line slope threshold of the memory snapshots.
+   * If the slope is greater than this value, the test will fail.
+   *
+   * @default 5
+   */
+  memoryThresholdInMB?: number;
+}
+
+export function memtest(opts: MemtestOptions, setup: () => Promise<Server>) {
+  const { memoryThresholdInMB = 5, duration = 60_000, ...loadtestOpts } = opts;
+  it(
+    'should not have a memory increase trend',
+    {
+      timeout: duration + 10_000, // allow 10s for the test teardown
+    },
+    async ({ expect }) => {
+      const server = await setup();
+
+      const { memoryInMBSnapshots } = await loadtest({
+        ...loadtestOpts,
+        duration,
+        server,
+      });
+
+      const slope = calculateRegressionSlope(memoryInMBSnapshots);
+
+      expect(
+        slope,
+        `Memory increase trend detected with slope of ${slope}MB (exceding threshold of ${memoryThresholdInMB}MB)`,
+      ).toBeLessThan(memoryThresholdInMB);
+    },
+  );
+}
+
+/**
+ * Detects a memory increase trend in an array of memory snapshots over time using linear regression.
+ *
+ * @param snapshots - An array of memory snapshots in MB.
+ *
+ * @returns The slope of the linear regression line.
+ */
+function calculateRegressionSlope(snapshots: number[]) {
+  if (snapshots.length < 2) {
+    throw new Error('Not enough snapshots to determine trend');
+  }
+
+  const data: [x: number, y: number][] = snapshots.map((memInMB, timestamp) => [
+    timestamp,
+    memInMB,
+  ]);
+  const result = regression.linear(data);
+  const slope = result.equation[0];
+  if (!slope) {
+    throw new Error('Regression slope is zero');
+  }
+
+  return slope;
+}

internal/proc/src/index.ts

@@ -9,6 +9,7 @@
 import terminate from 'terminate/promise';
 
 export interface Proc extends AsyncDisposable {
+  waitForExit: Promise<void>;
   getStd(o: 'out' | 'err' | 'both'): string;
   getStats(): Promise<{
     // Total CPU utilization (of all cores) as a percentage.
@@ -18,6 +19,11 @@
   }>;
 }
 
+export interface Server extends Proc {
+  port: number;
+  protocol: string;
+}
+
 export interface ProcOptions {
   /**
    * Pipe the logs from the spawned process to the current process, or to a file
@@ -79,6 +85,7 @@
   let stderr = '';
   let stdboth = '';
   const proc: Proc = {
+    waitForExit,
     getStd(o) {
       switch (o) {
         case 'out':
@@ -145,7 +152,7 @@
     // process ended _and_ the stdio streams have been closed
     if (code) {
       exitDeferred.reject(
         new Error(
           `Exit code ${code} from ${cmd} ${args.join(' ')}\n${trimError(stdboth)}`,
         ),
       );

package.json

@@ -23,7 +23,8 @@
     "test": "vitest --project unit",
     "test:bun": "bun test --bail",
     "test:e2e": "vitest --project e2e",
-    "test:leaks": "cross-env \"LEAK_TEST=1\" jest --detectOpenHandles --detectLeaks"
+    "test:leaks": "cross-env \"LEAK_TEST=1\" jest --detectOpenHandles --detectLeaks",
+    "test:mem": "vitest --project memtest"
   },
   "devDependencies": {
     "@babel/core": "7.26.7",

tsconfig.json

@@ -14,6 +14,8 @@
       "@internal/testing": ["./internal/testing/src/index.ts"],
       "@internal/e2e": ["./internal/e2e/src/index.ts"],
       "@internal/proc": ["./internal/proc/src/index.ts"],
+      "@internal/perf": ["./internal/perf/src/index.ts"],
+      "@internal/perf/memtest": ["./internal/perf/src/memtest.ts"],
       "@internal/testing/to-be-similar-string": [
         "./internal/testing/src/to-be-similar-string.ts"
       ],