Bound non-streaming response body reads in the OpenAPI plugin

.changeset/openapi-response-body-timeout.md

@@ -0,0 +1,5 @@
+---
+"@executor-js/plugin-openapi": patch
+---
+
+OpenAPI invocations now bound how long a buffered (non-streaming) response body may take to arrive. An upstream that returns headers quickly and then stalls the body previously hung the call indefinitely on runtimes without a platform subrequest limit; it now aborts after the response-body timeout (default 60s, configurable via `invokeOptions.responseBodyTimeoutMs`) with a distinct `upstream_response_body_timeout` failure.

packages/plugins/openapi/src/sdk/backing.ts

@@ -716,7 +716,16 @@ export const invokeOpenApiBackedTool = (input: {
                 details: error.cause ?? error,
               }),
             })
-          : Effect.fail(error),
+          : error.reason === "response_body_timeout"
+            ? Effect.succeed({
+                ok: false as const,
+                failure: ToolResult.fail({
+                  code: "upstream_response_body_timeout",
+                  message: error.message,
+                  details: error.cause ?? error,
+                }),
+              })
+            : Effect.fail(error),
       ),
     );
 

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

@@ -39,7 +39,7 @@ export class OpenApiSpecOverrideError extends Schema.TaggedErrorClass<OpenApiSpe
 export class OpenApiInvocationError extends Data.TaggedError("OpenApiInvocationError")<{
   readonly message: string;
   readonly statusCode: Option.Option<number>;
-  readonly reason?: "response_headers_timeout" | "unknown_arguments";
+  readonly reason?: "response_headers_timeout" | "response_body_timeout" | "unknown_arguments";
   readonly cause?: unknown;
 }> {}
 

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

@@ -24,6 +24,7 @@ export {
   invokeWithLayer,
   buildRequest,
   annotationsForOperation,
+  RESPONSE_BODY_TIMEOUT_MS,
   RESPONSE_HEADERS_TIMEOUT_MS,
   type InvokeOptions,
 } from "./invoke";

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

@@ -153,6 +153,9 @@ const normalizeContentType = (ct: string | null | undefined): string =>
 
 export const STREAM_MAX_BYTES = 1_000_000;
 export const STREAM_MAX_MS = 10_000;
+// Buffered bodies should normally finish soon after headers arrive. This is
+// deliberately generous while still providing plain Node runtimes a backstop.
+export const RESPONSE_BODY_TIMEOUT_MS = 60_000;
 // Below Cloudflare's ~125s subrequest limit so cloud fails first with an
 // actionable error, but above the slowest legitimate buffered upstreams
 // observed in production (30d telemetry: successful calls cluster under
@@ -170,6 +173,7 @@ export interface StreamingResponseCaps {
 
 export interface InvokeOptions {
   readonly responseHeadersTimeoutMs?: number;
+  readonly responseBodyTimeoutMs?: number;
 }
 
 const formatTimeout = (timeoutMs: number): string =>
@@ -178,6 +182,9 @@ const formatTimeout = (timeoutMs: number): string =>
 const responseHeadersTimeoutMessage = (timeoutMs: number): string =>
   `Upstream returned no response headers within ${formatTimeout(timeoutMs)}. The endpoint may be a live stream with no data to send (for example, runtime logs of an idle deployment); the request was aborted. Retry when the resource has activity, or use a non-streaming endpoint.`;
 
+const responseBodyTimeoutMessage = (timeoutMs: number): string =>
+  `Upstream response body did not finish within ${formatTimeout(timeoutMs)}. The request was aborted. Retry the operation; if it times out again, check the upstream service health or use an endpoint with a bounded response.`;
+
 const STREAMING_RESPONSE_CONTENT_TYPES = new Set([...NDJSON_MEDIA_TYPES, "text/event-stream"]);
 
 const isStreamingResponseContentType = (ct: string | null | undefined): boolean =>
@@ -1069,6 +1076,7 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* (
   const request = yield* buildRequest(operation, args, resolvedHeaders, integrationQueryParams);
 
   const responseHeadersTimeoutMs = options.responseHeadersTimeoutMs ?? RESPONSE_HEADERS_TIMEOUT_MS;
+  const responseBodyTimeoutMs = options.responseBodyTimeoutMs ?? RESPONSE_BODY_TIMEOUT_MS;
   const runFork = Effect.runForkWith(yield* Effect.context<never>());
   const responseExitOption = yield* Effect.callback<
     Option.Option<Exit.Exit<HttpClientResponse.HttpClientResponse, OpenApiInvocationError>>
@@ -1136,6 +1144,51 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* (
         cause: err,
       }),
   );
+  const readResponseBody = <A>(body: Effect.Effect<A, OpenApiInvocationError, never>) =>
+    Effect.gen(function* () {
+      const bodyExitOption = yield* Effect.callback<
+        Option.Option<Exit.Exit<A, OpenApiInvocationError>>
+      >((resume, signal) => {
+        let settled = false;
+        const bodyEffect = body.pipe(
+          Effect.exit,
+          Effect.tap((exit) =>
+            Effect.sync(() => {
+              if (settled) return;
+              settled = true;
+              clearTimeout(timer);
+              resume(Effect.succeed(Option.some(exit)));
+            }),
+          ),
+        );
+        const fiber = runFork(bodyEffect);
+        const interrupt = () => {
+          runFork(Fiber.interrupt(fiber));
+        };
+        const timer = setTimeout(() => {
+          if (settled) return;
+          settled = true;
+          interrupt();
+          resume(Effect.succeed(Option.none()));
+        }, responseBodyTimeoutMs);
+        signal.addEventListener("abort", interrupt, { once: true });
+        return Effect.sync(() => {
+          clearTimeout(timer);
+          signal.removeEventListener("abort", interrupt);
+          if (!settled) interrupt();
+        });
+      });
+      if (Option.isNone(bodyExitOption)) {
+        return yield* new OpenApiInvocationError({
+          message: responseBodyTimeoutMessage(responseBodyTimeoutMs),
+          statusCode: Option.some(status),
+          reason: "response_body_timeout",
+        });
+      }
+      const bodyExit = bodyExitOption.value;
+      if (Exit.isFailure(bodyExit)) return yield* Effect.failCause(bodyExit.cause);
+      return bodyExit.value;
+    });
   const responseBodyBinding = Option.getOrUndefined(operation.responseBody);
   const fileHint = responseBodyBinding
     ? Option.getOrUndefined(responseBodyBinding.fileHint)
@@ -1151,23 +1204,21 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* (
   if (streamingBody) {
     Object.assign(responseHeaders, streamingBody.headers);
   }
+  const bufferedBody = Effect.suspend(() =>
+    ok && fileHint?.kind === "binaryResponse"
+      ? response.arrayBuffer.pipe(
+          Effect.map((bytes) => fileFromBinaryBytes(new Uint8Array(bytes), fileHint, contentType)),
+        )
+      : isJsonContentType(contentType)
+        ? response.json.pipe(Effect.catch(() => response.text))
+        : response.text,
+  );
   const responseBody: unknown =
     status === 204
       ? null
-      : ok && fileHint?.kind === "binaryResponse"
-        ? fileFromBinaryBytes(
-            new Uint8Array(yield* response.arrayBuffer.pipe(mapBodyError)),
-            fileHint,
-            contentType,
-          )
-        : streamingBody
-          ? streamingBody.data
-          : isJsonContentType(contentType)
-            ? yield* response.json.pipe(
-                Effect.catch(() => response.text),
-                mapBodyError,
-              )
-            : yield* response.text.pipe(mapBodyError);
+      : streamingBody
+        ? streamingBody.data
+        : yield* readResponseBody(bufferedBody.pipe(mapBodyError));
 
   const dataBody =
     ok && fileHint?.kind === "byteField"

packages/plugins/openapi/src/sdk/response-body-timeout.test.ts

@@ -0,0 +1,149 @@
+import { describe, expect, it } from "@effect/vitest";
+import { Effect, Exit, Fiber, Option, Schema } from "effect";
+import { FetchHttpClient } from "effect/unstable/http";
+import { HttpApi, HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi";
+import { createServer } from "node:http";
+import type { AddressInfo, Socket } from "node:net";
+
+import {
+  createExecutor,
+  AuthTemplateSlug,
+  ConnectionName,
+  IntegrationSlug,
+  ToolAddress,
+} from "@executor-js/sdk";
+import { makeTestConfig, memoryCredentialsPlugin } from "@executor-js/sdk/testing";
+import { makeOpenApiHttpApiTestIntegrationConfig } from "../testing";
+
+import { openApiPlugin } from "./plugin";
+
+const BODY_READ_DEADLINE_MS = 15_000;
+const RESPONSE_BODY_TIMEOUT_MS = 100;
+const TOOL = "data.getData";
+
+const DataGroup = HttpApiGroup.make("data").add(
+  HttpApiEndpoint.get("getData", "/data", {
+    success: Schema.Unknown,
+  }),
+);
+
+const TimeoutApi = HttpApi.make("responseBodyTimeoutTest")
+  .add(DataGroup)
+  .annotateMerge(OpenApi.annotations({ title: "ResponseBodyTimeoutTest", version: "1.0.0" }));
+
+const testPlugins = () =>
+  [
+    openApiPlugin({
+      httpClientLayer: FetchHttpClient.layer,
+      invokeOptions: { responseBodyTimeoutMs: RESPONSE_BODY_TIMEOUT_MS },
+    }),
+    memoryCredentialsPlugin(),
+  ] as const;
+
+const buildExecutor = (baseUrl: string) =>
+  Effect.gen(function* () {
+    const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() }));
+    yield* executor.openapi.addSpec(
+      makeOpenApiHttpApiTestIntegrationConfig(TimeoutApi, {
+        slug: "body_timeout",
+        baseUrl,
+      }),
+    );
+    yield* executor.connections.create({
+      owner: "org",
+      name: ConnectionName.make("main"),
+      integration: IntegrationSlug.make("body_timeout"),
+      template: AuthTemplateSlug.make("apiKey"),
+      value: "token",
+    });
+    return {
+      executor,
+      address: ToolAddress.make(`tools.body_timeout.org.main.${TOOL}`),
+    };
+  });
+
+const startStalledBodyServer = () =>
+  Effect.acquireRelease(
+    Effect.callback<{ baseUrl: string; close: () => void }>((resume) => {
+      const sockets = new Set<Socket>();
+      const server = createServer((req, res) => {
+        if (req.url?.split("?")[0] !== "/data") {
+          res.writeHead(404).end();
+          return;
+        }
+        res.writeHead(200, { "content-type": "application/json" });
+        res.write('{"data":');
+      });
+      server.on("connection", (socket) => {
+        sockets.add(socket);
+        socket.on("close", () => sockets.delete(socket));
+      });
+      server.listen(0, "127.0.0.1", () => {
+        const port = (server.address() as AddressInfo).port;
+        resume(
+          Effect.succeed({
+            baseUrl: `http://127.0.0.1:${port}`,
+            close: () => {
+              for (const socket of sockets) socket.destroy();
+              server.close();
+            },
+          }),
+        );
+      });
+    }),
+    (server) => Effect.sync(() => server.close()),
+  );
+
+const observeUntil = <A, E>(fiber: Fiber.Fiber<A, E>, deadlineMs: number) =>
+  Effect.callback<Option.Option<Exit.Exit<A, E>>>((resume) => {
+    let settled = false;
+    const timer = setTimeout(() => {
+      if (settled) return;
+      settled = true;
+      resume(Effect.succeed(Option.none()));
+    }, deadlineMs);
+    const removeObserver = fiber.addObserver((exit) => {
+      if (settled) return;
+      settled = true;
+      clearTimeout(timer);
+      resume(Effect.succeed(Option.some(exit)));
+    });
+    return Effect.sync(() => {
+      clearTimeout(timer);
+      removeObserver();
+    });
+  });
+
+describe("OpenAPI response body timeout", () => {
+  it.effect(
+    "aborts a stalled non-streaming JSON body and returns an actionable tool failure",
+    () =>
+      Effect.gen(function* () {
+        const server = yield* startStalledBodyServer();
+        const { executor, address } = yield* buildExecutor(server.baseUrl);
+        const startedAt = Date.now();
+
+        const invocation = yield* executor.execute(address, {}).pipe(Effect.forkDetach);
+        const result = yield* observeUntil(invocation, BODY_READ_DEADLINE_MS);
+        const elapsedMs = Date.now() - startedAt;
+
+        expect(
+          Option.isSome(result),
+          `body read did not time out: hung until test deadline (${elapsedMs}ms)`,
+        ).toBe(true);
+        if (Option.isNone(result)) return;
+
+        expect(Exit.isSuccess(result.value)).toBe(true);
+        if (Exit.isFailure(result.value)) return;
+
+        expect(result.value.value).toMatchObject({
+          ok: false,
+          error: {
+            code: "upstream_response_body_timeout",
+            message: expect.stringContaining("response body"),
+          },
+        });
+      }),
+    { timeout: 20_000 },
+  );
+});