Abort stalled GraphQL invocations after a configurable timeout

.changeset/timeout-graphql-invocations.md

@@ -0,0 +1,5 @@
+---
+"executor": patch
+---
+
+Abort GraphQL tool calls that exceed the configured invocation timeout instead of waiting indefinitely for an upstream response.

packages/plugins/graphql/src/sdk/errors.ts

@@ -19,6 +19,8 @@ export class GraphqlExtractionError extends Schema.TaggedErrorClass<GraphqlExtra
 export class GraphqlInvocationError extends Data.TaggedError("GraphqlInvocationError")<{
   readonly message: string;
   readonly statusCode: Option.Option<number>;
+  readonly reason?: "invocation_timeout";
+  readonly timeoutMs?: number;
   readonly cause?: unknown;
 }> {}
 

packages/plugins/graphql/src/sdk/index.ts

@@ -1,6 +1,11 @@
 export { introspect, parseIntrospectionJson } from "./introspect";
 export { extract, type ExtractionOutput } from "./extract";
-export { invoke, invokeWithLayer } from "./invoke";
+export {
+  GRAPHQL_INVOCATION_TIMEOUT_MS,
+  invoke,
+  invokeWithLayer,
+  type GraphqlInvokeOptions,
+} from "./invoke";
 export {
   describeGraphqlAuthMethods,
   graphqlPlugin,

packages/plugins/graphql/src/sdk/invocation-timeout.test.ts

@@ -0,0 +1,107 @@
+import { describe, expect, it } from "@effect/vitest";
+import { Deferred, Effect, Option } from "effect";
+import { FetchHttpClient } from "effect/unstable/http";
+import { createServer } from "node:http";
+import type { AddressInfo, Socket } from "node:net";
+import { introspectionFromSchema } from "graphql";
+
+import {
+  AuthTemplateSlug,
+  ConnectionName,
+  IntegrationSlug,
+  ToolAddress,
+  createExecutor,
+} from "@executor-js/sdk";
+import { makeTestConfig, memoryCredentialsPlugin } from "@executor-js/sdk/testing";
+
+import { makeGreetingGraphqlSchema } from "../testing";
+import { graphqlPlugin } from "./plugin";
+
+const INVOCATION_TIMEOUT_MS = 100;
+
+const startHangingResponseServer = (closed: Deferred.Deferred<void>) =>
+  Effect.acquireRelease(
+    Effect.callback<{ endpoint: string; close: () => void }>((resume) => {
+      const sockets = new Set<Socket>();
+      const server = createServer((_request, response) => {
+        response.writeHead(200, { "content-type": "application/json" });
+        response.write('{"data":');
+      });
+      server.on("connection", (socket) => {
+        sockets.add(socket);
+        socket.on("close", () => {
+          sockets.delete(socket);
+          Effect.runFork(Deferred.succeed(closed, undefined));
+        });
+      });
+      server.listen(0, "127.0.0.1", () => {
+        const port = (server.address() as AddressInfo).port;
+        resume(
+          Effect.succeed({
+            endpoint: `http://127.0.0.1:${port}/graphql`,
+            close: () => {
+              for (const socket of sockets) socket.destroy();
+              server.close();
+            },
+          }),
+        );
+      });
+    }),
+    (server) => Effect.sync(() => server.close()),
+  );
+
+const introspectionJson = JSON.stringify({
+  data: introspectionFromSchema(makeGreetingGraphqlSchema({ includeMutation: false })),
+});
+
+describe("GraphQL invocation timeout", () => {
+  it.live("aborts a hanging response body and returns an actionable tool failure", () =>
+    Effect.gen(function* () {
+      const closed = yield* Deferred.make<void>();
+      const server = yield* startHangingResponseServer(closed);
+      const executor = yield* createExecutor(
+        makeTestConfig({
+          plugins: [
+            memoryCredentialsPlugin(),
+            graphqlPlugin({
+              httpClientLayer: FetchHttpClient.layer,
+              invokeOptions: { timeoutMs: INVOCATION_TIMEOUT_MS },
+            }),
+          ] as const,
+        }),
+      );
+
+      yield* executor.graphql.addIntegration({
+        endpoint: server.endpoint,
+        slug: "invocation_timeout",
+        introspectionJson,
+      });
+      yield* executor.connections.create({
+        owner: "org",
+        name: ConnectionName.make("main"),
+        integration: IntegrationSlug.make("invocation_timeout"),
+        template: AuthTemplateSlug.make("none"),
+        value: "unused",
+      });
+
+      const startedAt = Date.now();
+      const resultOption = yield* executor
+        .execute(ToolAddress.make("tools.invocation_timeout.org.main.query.hello"), {})
+        .pipe(Effect.timeoutOption(1_000));
+      const elapsedMs = Date.now() - startedAt;
+      const socketClosed = yield* Deferred.await(closed).pipe(Effect.timeoutOption(1_000));
+      const result = Option.getOrNull(resultOption);
+
+      expect(elapsedMs).toBeGreaterThanOrEqual(INVOCATION_TIMEOUT_MS - 25);
+      expect(elapsedMs).toBeLessThan(1_000);
+      expect(Option.isSome(socketClosed)).toBe(true);
+      expect(result).toMatchObject({
+        ok: false,
+        error: {
+          code: "graphql_request_timeout",
+          message: expect.stringContaining("GraphQL upstream did not complete within 100ms"),
+        },
+      });
+    }),
+  );
+});

packages/plugins/graphql/src/sdk/invoke.ts

@@ -21,6 +21,20 @@ export const endpointForTelemetry = (endpoint: string): string => {
   return url.toString();
 };
 
+// Below Cloudflare's approximate 125-second subrequest limit while preserving
+// slow upstream requests that Executor's HTTP transports support.
+export const GRAPHQL_INVOCATION_TIMEOUT_MS = 110_000;
+
+export interface GraphqlInvokeOptions {
+  readonly timeoutMs?: number;
+}
+
+const formatTimeout = (timeoutMs: number): string =>
+  timeoutMs % 1_000 === 0 ? `${timeoutMs / 1_000}s` : `${timeoutMs}ms`;
+
+const invocationTimeoutMessage = (timeoutMs: number): string =>
+  `GraphQL upstream did not complete within ${formatTimeout(timeoutMs)}. The request was aborted. Retry the operation or verify that the endpoint is responsive.`;
+
 /** The operation string to send for a call. A caller-supplied `select` overrides
  *  the default scalar-leaf selection: it is spliced into the field's selection
  *  set (`field { <select> }`) so nested/list data can be requested per call. Falls
@@ -62,8 +76,10 @@ export const invoke = Effect.fn("GraphQL.invoke")(function* (
   endpoint: string,
   resolvedHeaders: Record<string, string>,
   resolvedQueryParams: Record<string, string> = {},
+  options: GraphqlInvokeOptions = {},
 ) {
   const client = yield* HttpClient.HttpClient;
+  const timeoutMs = options.timeoutMs ?? GRAPHQL_INVOCATION_TIMEOUT_MS;
   const requestEndpoint = endpointWithQueryParams(endpoint, resolvedQueryParams);
   const telemetryEndpoint = endpointForTelemetry(endpoint);
 
@@ -107,41 +123,56 @@ export const invoke = Effect.fn("GraphQL.invoke")(function* (
     request = HttpClientRequest.setHeader(request, name, value);
   }
 
-  const response = yield* client.execute(request).pipe(
-    Effect.mapError(
-      (err) =>
-        new GraphqlInvocationError({
-          message: "GraphQL request failed",
-          statusCode: Option.none(),
-          cause: err,
-        }),
-    ),
+  return yield* Effect.gen(function* () {
+    const response = yield* client.execute(request).pipe(
+      Effect.mapError(
+        (err) =>
+          new GraphqlInvocationError({
+            message: "GraphQL request failed",
+            statusCode: Option.none(),
+            cause: err,
+          }),
+      ),
+    );
+
+    const status = response.status;
+    const contentType = response.headers["content-type"] ?? null;
+
+    const body: unknown = isJsonContentType(contentType)
+      ? yield* response.json.pipe(Effect.catch(() => response.text))
+      : yield* response.text;
+
+    // GraphQL responses are always 200 with { data, errors }
+    const gqlBody = body as { data?: unknown; errors?: unknown[] } | null;
+    const hasErrors = Array.isArray(gqlBody?.errors) && gqlBody.errors.length > 0;
+
+    yield* Effect.annotateCurrentSpan({
+      "http.status_code": status,
+      "plugin.graphql.has_errors": hasErrors,
+      "plugin.graphql.error_count": hasErrors ? gqlBody!.errors!.length : 0,
+    });
+
+    return InvocationResult.make({
+      status,
+      data: gqlBody?.data ?? null,
+      errors: hasErrors ? gqlBody!.errors : null,
+      body,
+      headers: { ...response.headers },
+    });
+  }).pipe(
+    Effect.timeoutOrElse({
+      duration: timeoutMs,
+      orElse: () =>
+        Effect.fail(
+          new GraphqlInvocationError({
+            message: invocationTimeoutMessage(timeoutMs),
+            statusCode: Option.none(),
+            reason: "invocation_timeout",
+            timeoutMs,
+          }),
+        ),
+    }),
   );
-
-  const status = response.status;
-  const contentType = response.headers["content-type"] ?? null;
-
-  const body: unknown = isJsonContentType(contentType)
-    ? yield* response.json.pipe(Effect.catch(() => response.text))
-    : yield* response.text;
-
-  // GraphQL responses are always 200 with { data, errors }
-  const gqlBody = body as { data?: unknown; errors?: unknown[] } | null;
-  const hasErrors = Array.isArray(gqlBody?.errors) && gqlBody.errors.length > 0;
-
-  yield* Effect.annotateCurrentSpan({
-    "http.status_code": status,
-    "plugin.graphql.has_errors": hasErrors,
-    "plugin.graphql.error_count": hasErrors ? gqlBody!.errors!.length : 0,
-  });
-
-  return InvocationResult.make({
-    status,
-    data: gqlBody?.data ?? null,
-    errors: hasErrors ? gqlBody!.errors : null,
-    body,
-    headers: { ...response.headers },
-  });
 });
 
 // ---------------------------------------------------------------------------
@@ -155,8 +186,9 @@ export const invokeWithLayer = (
   resolvedHeaders: Record<string, string>,
   resolvedQueryParams: Record<string, string>,
   httpClientLayer: Layer.Layer<HttpClient.HttpClient>,
+  options: GraphqlInvokeOptions = {},
 ) =>
-  invoke(operation, args, endpoint, resolvedHeaders, resolvedQueryParams).pipe(
+  invoke(operation, args, endpoint, resolvedHeaders, resolvedQueryParams, options).pipe(
     Effect.provide(httpClientLayer),
     Effect.withSpan("plugin.graphql.invoke", {
       attributes: {

packages/plugins/graphql/src/sdk/plugin.ts

@@ -48,7 +48,7 @@ import {
   GraphqlIntrospectionError,
   GraphqlInvocationError,
 } from "./errors";
-import { effectiveOperationString, invokeWithLayer } from "./invoke";
+import { effectiveOperationString, invokeWithLayer, type GraphqlInvokeOptions } from "./invoke";
 import { validateOperationString } from "./validate-selection";
 import { graphqlPresets } from "./presets";
 import { makeDefaultGraphqlStore, type GraphqlStore, type StoredOperation } from "./store";
@@ -871,6 +871,7 @@ export type GraphqlPluginExtension = ReturnType<typeof makeGraphqlExtension>;
 
 export interface GraphqlPluginOptions {
   readonly httpClientLayer?: Layer.Layer<HttpClient.HttpClient>;
+  readonly invokeOptions?: GraphqlInvokeOptions;
 }
 
 export const graphqlPlugin = definePlugin((options?: GraphqlPluginOptions) => {
@@ -1133,6 +1134,7 @@ export const graphqlPlugin = definePlugin((options?: GraphqlPluginOptions) => {
           headers,
           queryParams,
           httpClientLayer,
+          options?.invokeOptions,
         );
 
         // An HTTP auth wall is classified BEFORE the GraphQL-errors branch: a
@@ -1198,9 +1200,19 @@ export const graphqlPlugin = definePlugin((options?: GraphqlPluginOptions) => {
         }
         return ToolResult.ok(result.data);
       }).pipe(
-        Effect.catchTag("GraphqlAuthRequiredError", (error) =>
-          Effect.succeed(graphqlAuthToolFailure(error)),
-        ),
+        Effect.catchTags({
+          GraphqlAuthRequiredError: (error) => Effect.succeed(graphqlAuthToolFailure(error)),
+          GraphqlInvocationError: (error) =>
+            error.reason === "invocation_timeout"
+              ? Effect.succeed(
+                  graphqlToolFailure(
+                    "graphql_request_timeout",
+                    error.message,
+                    error.timeoutMs === undefined ? undefined : { timeoutMs: error.timeoutMs },
+                  ),
+                )
+              : Effect.fail(error),
+        }),
       ),
 
     // Per-connection cleanup. Operation bindings are catalog-level (shared