History

Bound non-streaming response body reads in the OpenAPI plugin (#1430)

* test openapi stalled response body timeout * fix openapi response body timeout * Add changeset

Rhys Sullivan committed 3 months ago

Commit a86cc4e1 parent 72073476 files · +231 −16Browse files

Unified · Split

6 files
  1. .changeset/openapi-response-body-timeout.md +5 −0
  2. packages/plugins/openapi/src/sdk/backing.ts +10 −1
  3. packages/plugins/openapi/src/sdk/errors.ts +1 −1
  4. packages/plugins/openapi/src/sdk/index.ts +1 −0
  5. packages/plugins/openapi/src/sdk/invoke.ts +65 −14
  6. packages/plugins/openapi/src/sdk/response-body-timeout.test.ts +149 −0
.changeset/openapi-response-body-timeout.md+5−0
@@ -0,0 +1,5 @@
1+---
2+"@executor-js/plugin-openapi": patch
3+---
4+
5+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+10−1
@@ -716,7 +716,16 @@ export const invokeOpenApiBackedTool = (input: { · 715 unmodified lines
716716 details: error.cause ?? error,
717717 }),
718718 })
719− : Effect.fail(error),
719+ : error.reason === "response_body_timeout"
720+ ? Effect.succeed({
721+ ok: false as const,
722+ failure: ToolResult.fail({
723+ code: "upstream_response_body_timeout",
724+ message: error.message,
725+ details: error.cause ?? error,
726+ }),
727+ })
728+ : Effect.fail(error),
720729 ),
721730 );
722731
packages/plugins/openapi/src/sdk/errors.ts+1−1
@@ -39,7 +39,7 @@ export class OpenApiSpecOverrideError extends Schema.TaggedErrorClass<OpenApiSpe · 38 unmodified lines
3939 export class OpenApiInvocationError extends Data.TaggedError("OpenApiInvocationError")<{
4040 readonly message: string;
4141 readonly statusCode: Option.Option<number>;
42− readonly reason?: "response_headers_timeout" | "unknown_arguments";
42+ readonly reason?: "response_headers_timeout" | "response_body_timeout" | "unknown_arguments";
4343 readonly cause?: unknown;
4444 }> {}
4545
packages/plugins/openapi/src/sdk/index.ts+1−0
@@ -24,6 +24,7 @@ export { · 23 unmodified lines
2424 invokeWithLayer,
2525 buildRequest,
2626 annotationsForOperation,
27+ RESPONSE_BODY_TIMEOUT_MS,
2728 RESPONSE_HEADERS_TIMEOUT_MS,
2829 type InvokeOptions,
2930 } from "./invoke";
packages/plugins/openapi/src/sdk/invoke.ts+65−14
@@ -153,6 +153,9 @@ const normalizeContentType = (ct: string | null | undefined): string => · 152 unmodified lines
153153
154154 export const STREAM_MAX_BYTES = 1_000_000;
155155 export const STREAM_MAX_MS = 10_000;
156+// Buffered bodies should normally finish soon after headers arrive. This is
157+// deliberately generous while still providing plain Node runtimes a backstop.
158+export const RESPONSE_BODY_TIMEOUT_MS = 60_000;
156159 // Below Cloudflare's ~125s subrequest limit so cloud fails first with an
157160 // actionable error, but above the slowest legitimate buffered upstreams
158161 // observed in production (30d telemetry: successful calls cluster under
@@ -170,6 +173,7 @@ export interface StreamingResponseCaps { · 11 unmodified lines
170173
171174 export interface InvokeOptions {
172175 readonly responseHeadersTimeoutMs?: number;
176+ readonly responseBodyTimeoutMs?: number;
173177 }
174178
175179 const formatTimeout = (timeoutMs: number): string =>
@@ -178,6 +182,9 @@ const formatTimeout = (timeoutMs: number): string => · 2 unmodified lines
178182 const responseHeadersTimeoutMessage = (timeoutMs: number): string =>
179183 `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.`;
180184
185+const responseBodyTimeoutMessage = (timeoutMs: number): string =>
186+ `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.`;
187+
181188 const STREAMING_RESPONSE_CONTENT_TYPES = new Set([...NDJSON_MEDIA_TYPES, "text/event-stream"]);
182189
183190 const isStreamingResponseContentType = (ct: string | null | undefined): boolean =>
@@ -1069,6 +1076,7 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( · 885 unmodified lines
10691076 const request = yield* buildRequest(operation, args, resolvedHeaders, integrationQueryParams);
10701077
10711078 const responseHeadersTimeoutMs = options.responseHeadersTimeoutMs ?? RESPONSE_HEADERS_TIMEOUT_MS;
1079+ const responseBodyTimeoutMs = options.responseBodyTimeoutMs ?? RESPONSE_BODY_TIMEOUT_MS;
10721080 const runFork = Effect.runForkWith(yield* Effect.context<never>());
10731081 const responseExitOption = yield* Effect.callback<
10741082 Option.Option<Exit.Exit<HttpClientResponse.HttpClientResponse, OpenApiInvocationError>>
@@ -1136,6 +1144,51 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( · 61 unmodified lines
11361144 cause: err,
11371145 }),
11381146 );
1147+ const readResponseBody = <A>(body: Effect.Effect<A, OpenApiInvocationError, never>) =>
1148+ Effect.gen(function* () {
1149+ const bodyExitOption = yield* Effect.callback<
1150+ Option.Option<Exit.Exit<A, OpenApiInvocationError>>
1151+ >((resume, signal) => {
1152+ let settled = false;
1153+ const bodyEffect = body.pipe(
1154+ Effect.exit,
1155+ Effect.tap((exit) =>
1156+ Effect.sync(() => {
1157+ if (settled) return;
1158+ settled = true;
1159+ clearTimeout(timer);
1160+ resume(Effect.succeed(Option.some(exit)));
1161+ }),
1162+ ),
1163+ );
1164+ const fiber = runFork(bodyEffect);
1165+ const interrupt = () => {
1166+ runFork(Fiber.interrupt(fiber));
1167+ };
1168+ const timer = setTimeout(() => {
1169+ if (settled) return;
1170+ settled = true;
1171+ interrupt();
1172+ resume(Effect.succeed(Option.none()));
1173+ }, responseBodyTimeoutMs);
1174+ signal.addEventListener("abort", interrupt, { once: true });
1175+ return Effect.sync(() => {
1176+ clearTimeout(timer);
1177+ signal.removeEventListener("abort", interrupt);
1178+ if (!settled) interrupt();
1179+ });
1180+ });
1181+ if (Option.isNone(bodyExitOption)) {
1182+ return yield* new OpenApiInvocationError({
1183+ message: responseBodyTimeoutMessage(responseBodyTimeoutMs),
1184+ statusCode: Option.some(status),
1185+ reason: "response_body_timeout",
1186+ });
1187+ }
1188+ const bodyExit = bodyExitOption.value;
1189+ if (Exit.isFailure(bodyExit)) return yield* Effect.failCause(bodyExit.cause);
1190+ return bodyExit.value;
1191+ });
11391192 const responseBodyBinding = Option.getOrUndefined(operation.responseBody);
11401193 const fileHint = responseBodyBinding
11411194 ? Option.getOrUndefined(responseBodyBinding.fileHint)
@@ -1151,23 +1204,21 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( · 9 unmodified lines
11511204 if (streamingBody) {
11521205 Object.assign(responseHeaders, streamingBody.headers);
11531206 }
1207+ const bufferedBody = Effect.suspend(() =>
1208+ ok && fileHint?.kind === "binaryResponse"
1209+ ? response.arrayBuffer.pipe(
1210+ Effect.map((bytes) => fileFromBinaryBytes(new Uint8Array(bytes), fileHint, contentType)),
1211+ )
1212+ : isJsonContentType(contentType)
1213+ ? response.json.pipe(Effect.catch(() => response.text))
1214+ : response.text,
1215+ );
11541216 const responseBody: unknown =
11551217 status === 204
11561218 ? null
1157− : ok && fileHint?.kind === "binaryResponse"
1158− ? fileFromBinaryBytes(
1159− new Uint8Array(yield* response.arrayBuffer.pipe(mapBodyError)),
1160− fileHint,
1161− contentType,
1162− )
1163− : streamingBody
1164− ? streamingBody.data
1165− : isJsonContentType(contentType)
1166− ? yield* response.json.pipe(
1167− Effect.catch(() => response.text),
1168− mapBodyError,
1169− )
1170− : yield* response.text.pipe(mapBodyError);
1219+ : streamingBody
1220+ ? streamingBody.data
1221+ : yield* readResponseBody(bufferedBody.pipe(mapBodyError));
11711222
11721223 const dataBody =
11731224 ok && fileHint?.kind === "byteField"
packages/plugins/openapi/src/sdk/response-body-timeout.test.ts+149−0
@@ -0,0 +1,149 @@
1+import { describe, expect, it } from "@effect/vitest";
2+import { Effect, Exit, Fiber, Option, Schema } from "effect";
3+import { FetchHttpClient } from "effect/unstable/http";
4+import { HttpApi, HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi";
5+import { createServer } from "node:http";
6+import type { AddressInfo, Socket } from "node:net";
7+
8+import {
9+ createExecutor,
10+ AuthTemplateSlug,
11+ ConnectionName,
12+ IntegrationSlug,
13+ ToolAddress,
14+} from "@executor-js/sdk";
15+import { makeTestConfig, memoryCredentialsPlugin } from "@executor-js/sdk/testing";
16+import { makeOpenApiHttpApiTestIntegrationConfig } from "../testing";
17+
18+import { openApiPlugin } from "./plugin";
19+
20+const BODY_READ_DEADLINE_MS = 15_000;
21+const RESPONSE_BODY_TIMEOUT_MS = 100;
22+const TOOL = "data.getData";
23+
24+const DataGroup = HttpApiGroup.make("data").add(
25+ HttpApiEndpoint.get("getData", "/data", {
26+ success: Schema.Unknown,
27+ }),
28+);
29+
30+const TimeoutApi = HttpApi.make("responseBodyTimeoutTest")
31+ .add(DataGroup)
32+ .annotateMerge(OpenApi.annotations({ title: "ResponseBodyTimeoutTest", version: "1.0.0" }));
33+
34+const testPlugins = () =>
35+ [
36+ openApiPlugin({
37+ httpClientLayer: FetchHttpClient.layer,
38+ invokeOptions: { responseBodyTimeoutMs: RESPONSE_BODY_TIMEOUT_MS },
39+ }),
40+ memoryCredentialsPlugin(),
41+ ] as const;
42+
43+const buildExecutor = (baseUrl: string) =>
44+ Effect.gen(function* () {
45+ const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() }));
46+ yield* executor.openapi.addSpec(
47+ makeOpenApiHttpApiTestIntegrationConfig(TimeoutApi, {
48+ slug: "body_timeout",
49+ baseUrl,
50+ }),
51+ );
52+ yield* executor.connections.create({
53+ owner: "org",
54+ name: ConnectionName.make("main"),
55+ integration: IntegrationSlug.make("body_timeout"),
56+ template: AuthTemplateSlug.make("apiKey"),
57+ value: "token",
58+ });
59+ return {
60+ executor,
61+ address: ToolAddress.make(`tools.body_timeout.org.main.${TOOL}`),
62+ };
63+ });
64+
65+const startStalledBodyServer = () =>
66+ Effect.acquireRelease(
67+ Effect.callback<{ baseUrl: string; close: () => void }>((resume) => {
68+ const sockets = new Set<Socket>();
69+ const server = createServer((req, res) => {
70+ if (req.url?.split("?")[0] !== "/data") {
71+ res.writeHead(404).end();
72+ return;
73+ }
74+ res.writeHead(200, { "content-type": "application/json" });
75+ res.write('{"data":');
76+ });
77+ server.on("connection", (socket) => {
78+ sockets.add(socket);
79+ socket.on("close", () => sockets.delete(socket));
80+ });
81+ server.listen(0, "127.0.0.1", () => {
82+ const port = (server.address() as AddressInfo).port;
83+ resume(
84+ Effect.succeed({
85+ baseUrl: `http://127.0.0.1:${port}`,
86+ close: () => {
87+ for (const socket of sockets) socket.destroy();
88+ server.close();
89+ },
90+ }),
91+ );
92+ });
93+ }),
94+ (server) => Effect.sync(() => server.close()),
95+ );
96+
97+const observeUntil = <A, E>(fiber: Fiber.Fiber<A, E>, deadlineMs: number) =>
98+ Effect.callback<Option.Option<Exit.Exit<A, E>>>((resume) => {
99+ let settled = false;
100+ const timer = setTimeout(() => {
101+ if (settled) return;
102+ settled = true;
103+ resume(Effect.succeed(Option.none()));
104+ }, deadlineMs);
105+ const removeObserver = fiber.addObserver((exit) => {
106+ if (settled) return;
107+ settled = true;
108+ clearTimeout(timer);
109+ resume(Effect.succeed(Option.some(exit)));
110+ });
111+ return Effect.sync(() => {
112+ clearTimeout(timer);
113+ removeObserver();
114+ });
115+ });
116+
117+describe("OpenAPI response body timeout", () => {
118+ it.effect(
119+ "aborts a stalled non-streaming JSON body and returns an actionable tool failure",
120+ () =>
121+ Effect.gen(function* () {
122+ const server = yield* startStalledBodyServer();
123+ const { executor, address } = yield* buildExecutor(server.baseUrl);
124+ const startedAt = Date.now();
125+
126+ const invocation = yield* executor.execute(address, {}).pipe(Effect.forkDetach);
127+ const result = yield* observeUntil(invocation, BODY_READ_DEADLINE_MS);
128+ const elapsedMs = Date.now() - startedAt;
129+
130+ expect(
131+ Option.isSome(result),
132+ `body read did not time out: hung until test deadline (${elapsedMs}ms)`,
133+ ).toBe(true);
134+ if (Option.isNone(result)) return;
135+
136+ expect(Exit.isSuccess(result.value)).toBe(true);
137+ if (Exit.isFailure(result.value)) return;
138+
139+ expect(result.value.value).toMatchObject({
140+ ok: false,
141+ error: {
142+ code: "upstream_response_body_timeout",
143+ message: expect.stringContaining("response body"),
144+ },
145+ });
146+ }),
147+ { timeout: 20_000 },
148+ );
149+});