| @@ -153,6 +153,9 @@ const normalizeContentType = (ct: string | null | undefined): string => · 152 unmodified lines |
| 153 | 153 | |
| 154 | 154 | export const STREAM_MAX_BYTES = 1_000_000; |
| 155 | 155 | 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; |
| 156 | 159 | // Below Cloudflare's ~125s subrequest limit so cloud fails first with an |
| 157 | 160 | // actionable error, but above the slowest legitimate buffered upstreams |
| 158 | 161 | // observed in production (30d telemetry: successful calls cluster under |
| @@ -170,6 +173,7 @@ export interface StreamingResponseCaps { · 11 unmodified lines |
| 170 | 173 | |
| 171 | 174 | export interface InvokeOptions { |
| 172 | 175 | readonly responseHeadersTimeoutMs?: number; |
| 176 | + readonly responseBodyTimeoutMs?: number; |
| 173 | 177 | } |
| 174 | 178 | |
| 175 | 179 | const formatTimeout = (timeoutMs: number): string => |
| @@ -178,6 +182,9 @@ const formatTimeout = (timeoutMs: number): string => · 2 unmodified lines |
| 178 | 182 | const responseHeadersTimeoutMessage = (timeoutMs: number): string => |
| 179 | 183 | `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.`; |
| 180 | 184 | |
| 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 | + |
| 181 | 188 | const STREAMING_RESPONSE_CONTENT_TYPES = new Set([...NDJSON_MEDIA_TYPES, "text/event-stream"]); |
| 182 | 189 | |
| 183 | 190 | const isStreamingResponseContentType = (ct: string | null | undefined): boolean => |
| @@ -1069,6 +1076,7 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( · 885 unmodified lines |
| 1069 | 1076 | const request = yield* buildRequest(operation, args, resolvedHeaders, integrationQueryParams); |
| 1070 | 1077 | |
| 1071 | 1078 | const responseHeadersTimeoutMs = options.responseHeadersTimeoutMs ?? RESPONSE_HEADERS_TIMEOUT_MS; |
| 1079 | + const responseBodyTimeoutMs = options.responseBodyTimeoutMs ?? RESPONSE_BODY_TIMEOUT_MS; |
| 1072 | 1080 | const runFork = Effect.runForkWith(yield* Effect.context<never>()); |
| 1073 | 1081 | const responseExitOption = yield* Effect.callback< |
| 1074 | 1082 | Option.Option<Exit.Exit<HttpClientResponse.HttpClientResponse, OpenApiInvocationError>> |
| @@ -1136,6 +1144,51 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( · 61 unmodified lines |
| 1136 | 1144 | cause: err, |
| 1137 | 1145 | }), |
| 1138 | 1146 | ); |
| 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 | + }); |
| 1139 | 1192 | const responseBodyBinding = Option.getOrUndefined(operation.responseBody); |
| 1140 | 1193 | const fileHint = responseBodyBinding |
| 1141 | 1194 | ? Option.getOrUndefined(responseBodyBinding.fileHint) |
| @@ -1151,23 +1204,21 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( · 9 unmodified lines |
| 1151 | 1204 | if (streamingBody) { |
| 1152 | 1205 | Object.assign(responseHeaders, streamingBody.headers); |
| 1153 | 1206 | } |
| 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 | + ); |
| 1154 | 1216 | const responseBody: unknown = |
| 1155 | 1217 | status === 204 |
| 1156 | 1218 | ? 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)); |
| 1171 | 1222 | |
| 1172 | 1223 | const dataBody = |
| 1173 | 1224 | ok && fileHint?.kind === "byteField" |