From a8a119dd5eb2addc262584118a5128293f3f9f63 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 17 Jul 2026 21:56:17 -0700 Subject: [PATCH 1/3] test openapi stalled response body timeout --- .../src/sdk/response-body-timeout.test.ts | 144 ++++++++++++++++++ 1 file changed, 144 insertions(+) create mode 100644 packages/plugins/openapi/src/sdk/response-body-timeout.test.ts diff --git a/packages/plugins/openapi/src/sdk/response-body-timeout.test.ts b/packages/plugins/openapi/src/sdk/response-body-timeout.test.ts new file mode 100644 index 000000000..72a1145b9 --- /dev/null +++ b/packages/plugins/openapi/src/sdk/response-body-timeout.test.ts @@ -0,0 +1,144 @@ +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 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 }), 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(); + 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 = (fiber: Fiber.Fiber, deadlineMs: number) => + Effect.callback>>((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(); + + // No response-body timeout option exists yet. The fix should add an + // InvokeOptions seam so this regression can use a shorter deadline. + 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 }, + ); +}); From 45d58fa97ab1d4456715b0f42e69464c35f4398f Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 17 Jul 2026 22:02:18 -0700 Subject: [PATCH 2/3] fix openapi response body timeout --- packages/plugins/openapi/src/sdk/backing.ts | 11 ++- packages/plugins/openapi/src/sdk/errors.ts | 2 +- packages/plugins/openapi/src/sdk/index.ts | 1 + packages/plugins/openapi/src/sdk/invoke.ts | 79 +++++++++++++++---- .../src/sdk/response-body-timeout.test.ts | 11 ++- 5 files changed, 85 insertions(+), 19 deletions(-) diff --git a/packages/plugins/openapi/src/sdk/backing.ts b/packages/plugins/openapi/src/sdk/backing.ts index 4791cd040..4e6e5ba95 100644 --- a/packages/plugins/openapi/src/sdk/backing.ts +++ b/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), ), ); diff --git a/packages/plugins/openapi/src/sdk/errors.ts b/packages/plugins/openapi/src/sdk/errors.ts index 64d440e72..6a5fc4a8c 100644 --- a/packages/plugins/openapi/src/sdk/errors.ts +++ b/packages/plugins/openapi/src/sdk/errors.ts @@ -39,7 +39,7 @@ export class OpenApiSpecOverrideError extends Schema.TaggedErrorClass; - readonly reason?: "response_headers_timeout" | "unknown_arguments"; + readonly reason?: "response_headers_timeout" | "response_body_timeout" | "unknown_arguments"; readonly cause?: unknown; }> {} diff --git a/packages/plugins/openapi/src/sdk/index.ts b/packages/plugins/openapi/src/sdk/index.ts index d9c1a4c7d..10086fb11 100644 --- a/packages/plugins/openapi/src/sdk/index.ts +++ b/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"; diff --git a/packages/plugins/openapi/src/sdk/invoke.ts b/packages/plugins/openapi/src/sdk/invoke.ts index 4a107dd1a..4831eafa0 100644 --- a/packages/plugins/openapi/src/sdk/invoke.ts +++ b/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()); const responseExitOption = yield* Effect.callback< Option.Option> @@ -1136,6 +1144,51 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( cause: err, }), ); + const readResponseBody = (body: Effect.Effect) => + Effect.gen(function* () { + const bodyExitOption = yield* Effect.callback< + Option.Option> + >((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" diff --git a/packages/plugins/openapi/src/sdk/response-body-timeout.test.ts b/packages/plugins/openapi/src/sdk/response-body-timeout.test.ts index 72a1145b9..08441afd2 100644 --- a/packages/plugins/openapi/src/sdk/response-body-timeout.test.ts +++ b/packages/plugins/openapi/src/sdk/response-body-timeout.test.ts @@ -18,6 +18,7 @@ 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( @@ -31,7 +32,13 @@ const TimeoutApi = HttpApi.make("responseBodyTimeoutTest") .annotateMerge(OpenApi.annotations({ title: "ResponseBodyTimeoutTest", version: "1.0.0" })); const testPlugins = () => - [openApiPlugin({ httpClientLayer: FetchHttpClient.layer }), memoryCredentialsPlugin()] as const; + [ + openApiPlugin({ + httpClientLayer: FetchHttpClient.layer, + invokeOptions: { responseBodyTimeoutMs: RESPONSE_BODY_TIMEOUT_MS }, + }), + memoryCredentialsPlugin(), + ] as const; const buildExecutor = (baseUrl: string) => Effect.gen(function* () { @@ -116,8 +123,6 @@ describe("OpenAPI response body timeout", () => { const { executor, address } = yield* buildExecutor(server.baseUrl); const startedAt = Date.now(); - // No response-body timeout option exists yet. The fix should add an - // InvokeOptions seam so this regression can use a shorter deadline. const invocation = yield* executor.execute(address, {}).pipe(Effect.forkDetach); const result = yield* observeUntil(invocation, BODY_READ_DEADLINE_MS); const elapsedMs = Date.now() - startedAt; From 1b7dac2a8de3abf6f0ac1fdd672bc67c1fb05f4e Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 17 Jul 2026 22:58:22 -0700 Subject: [PATCH 3/3] Add changeset --- .changeset/openapi-response-body-timeout.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/openapi-response-body-timeout.md diff --git a/.changeset/openapi-response-body-timeout.md b/.changeset/openapi-response-body-timeout.md new file mode 100644 index 000000000..fa78648ae --- /dev/null +++ b/.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.