Enforce deadlines across the whole provider request lifecycle

The abort timer was cleared as soon as fetch resolved with headers, so a server that stalled the response body hung the request forever - and one hung run froze the autonomous queue behind the worker re-entrancy guard. JSON requests now carry a header deadline plus a longer body deadline on one abort controller, and event streams fail when no chunk arrives within an idle window, cancelling the reader. Timeouts surface as ProviderHttpError so the run records a clear failure and the heartbeat retries.
2026-06-09 21:18:51 -04:00
1 parent f15ed8cb6f
commit 2dd4f3976a
@@ -19,7 +19,15 @@ export interface ProviderEventStreamResponse {
  rateLimitState: ProviderObservedRateLimitState | null;
 }

-const PROVIDER_REQUEST_TIMEOUT_MS = 60_000;
+const HEADERS_TIMEOUT_MS = 60_000;
+const BODY_TIMEOUT_MS = 300_000;
+const STREAM_IDLE_TIMEOUT_MS = 60_000;
+
+export interface ProviderRequestTimeouts {
+  headersTimeoutMs?: number;
+  bodyTimeoutMs?: number;
+  idleTimeoutMs?: number;
+}

 export class ProviderHttpError extends Error {
  constructor(
@@ -35,7 +43,12 @@ export function providerRateLimitStateFromError(error: unknown): ProviderObserve
  return error instanceof ProviderHttpError ? error.rateLimitState : null;
 }

-export async function postJson(endpoint: string, headers: Record<string, string>, body: JsonRecord): Promise<ProviderJsonResponse> {
+export async function postJson(
+  endpoint: string,
+  headers: Record<string, string>,
+  body: JsonRecord,
+  timeouts: ProviderRequestTimeouts = {},
+): Promise<ProviderJsonResponse> {
  return requestJson(endpoint, {
    method: "POST",
    headers: {
@@ -43,7 +56,7 @@ export async function postJson(endpoint: string, headers: Record<string, string>
      ...headers,
    },
    body: JSON.stringify(body),
-  });
+  }, timeouts);
 }

 export async function getJson(endpoint: string, headers: Record<string, string>): Promise<ProviderJsonResponse> {
@@ -54,8 +67,10 @@ export async function postEventStream(
  endpoint: string,
  headers: Record<string, string>,
  body: JsonRecord,
+  timeouts: ProviderRequestTimeouts = {},
 ): Promise<ProviderEventStreamResponse> {
-  const response = await requestProvider(endpoint, {
+  const controller = new AbortController();
+  const response = await withDeadline(controller, timeouts.headersTimeoutMs ?? HEADERS_TIMEOUT_MS, () => fetch(endpoint, {
    method: "POST",
    headers: {
      accept: "text/event-stream",
@@ -63,22 +78,24 @@ export async function postEventStream(
      ...headers,
    },
    body: JSON.stringify(body),
-  });
+    signal: controller.signal,
+  }));
  const rateLimitState = parseProviderRateLimitHeaders(response.headers);
  if (!response.ok) {
-    const raw = await response.text();
+    const raw = await withDeadline(controller, timeouts.bodyTimeoutMs ?? BODY_TIMEOUT_MS, () => response.text());
    const data = parseProviderJson(raw, false);
    const message = readProviderError(data) ?? (raw.trim() || `Provider request failed with HTTP ${response.status}.`);
    throw new ProviderHttpError(message, rateLimitState);
  }
  if (!response.body) throw new ProviderHttpError("Provider returned an empty event stream.", rateLimitState);
-  return { messages: readServerSentEvents(response.body), rateLimitState };
+  return { messages: readServerSentEvents(response.body, timeouts.idleTimeoutMs ?? STREAM_IDLE_TIMEOUT_MS), rateLimitState };
 }

-async function requestJson(endpoint: string, init: RequestInit): Promise<ProviderJsonResponse> {
-  const response = await requestProvider(endpoint, init);
+async function requestJson(endpoint: string, init: RequestInit, timeouts: ProviderRequestTimeouts = {}): Promise<ProviderJsonResponse> {
+  const controller = new AbortController();
+  const response = await withDeadline(controller, timeouts.headersTimeoutMs ?? HEADERS_TIMEOUT_MS, () => fetch(endpoint, { ...init, signal: controller.signal }));
  const rateLimitState = parseProviderRateLimitHeaders(response.headers);
-  const raw = await response.text();
+  const raw = await withDeadline(controller, timeouts.bodyTimeoutMs ?? BODY_TIMEOUT_MS, () => response.text());
  const data = parseProviderJson(raw, response.ok);
  if (!response.ok) {
    const message = readProviderError(data) ?? (raw.trim() || `Provider request failed with HTTP ${response.status}.`);
@@ -87,24 +104,35 @@ async function requestJson(endpoint: string, init: RequestInit): Promise<Provide
  return { data, rateLimitState };
 }

-async function requestProvider(endpoint: string, init: RequestInit): Promise<Response> {
-  const controller = new AbortController();
-  const timeout = setTimeout(() => controller.abort(), PROVIDER_REQUEST_TIMEOUT_MS);
+/**
+ * Runs one phase of a provider request under a hard deadline. The abort
+ * controller covers the WHOLE request — a server that returns headers and
+ * then stalls the body must not hang the caller, so the body read gets its
+ * own (longer) deadline after the headers arrive.
+ */
+async function withDeadline<T>(controller: AbortController, timeoutMs: number, operation: () => Promise<T>): Promise<T> {
+  const timer = setTimeout(() => controller.abort(), timeoutMs);
  try {
-    return await fetch(endpoint, { ...init, signal: controller.signal });
+    return await operation();
+  } catch (error) {
+    if (controller.signal.aborted) throw new ProviderHttpError("Provider request timed out.", null);
+    throw error;
  } finally {
-    clearTimeout(timeout);
+    clearTimeout(timer);
  }
 }

-export async function* readServerSentEvents(body: ReadableStream<Uint8Array>): AsyncIterable<ProviderEventStreamMessage> {
+export async function* readServerSentEvents(
+  body: ReadableStream<Uint8Array>,
+  idleTimeoutMs: number = STREAM_IDLE_TIMEOUT_MS,
+): AsyncIterable<ProviderEventStreamMessage> {
  const decoder = new TextDecoder();
  const state: EventStreamParseState = { eventName: null, dataLines: [] };
  let lineBuffer = "";
  const reader = body.getReader();
  try {
    for (;;) {
-      const { value, done } = await reader.read();
+      const { value, done } = await readWithIdleDeadline(reader, idleTimeoutMs);
      if (done) break;
      lineBuffer += decoder.decode(value, { stream: true });
      const lines = lineBuffer.split("\n");
@@ -121,11 +149,30 @@ export async function* readServerSentEvents(body: ReadableStream<Uint8Array>): A
    }
    const message = flushServerSentEvent(state);
    if (message) yield message;
+  } catch (error) {
+    await reader.cancel().catch(() => undefined);
+    throw error;
  } finally {
    reader.releaseLock();
  }
 }

+/** A healthy event stream emits chunks continuously; a read that produces nothing within the idle window means the connection is dead. */
+async function readWithIdleDeadline(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  idleTimeoutMs: number,
+): Promise<ReadableStreamReadResult<Uint8Array>> {
+  let timer: NodeJS.Timeout | undefined;
+  const idle = new Promise<never>((_, reject) => {
+    timer = setTimeout(() => reject(new ProviderHttpError("Provider stream timed out.", null)), idleTimeoutMs);
+  });
+  try {
+    return await Promise.race([reader.read(), idle]);
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
 function parseServerSentEventLine(state: EventStreamParseState, rawLine: string): ProviderEventStreamMessage | null {
  const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine;
  if (line === "") return flushServerSentEvent(state);
@@ -0,0 +1,64 @@
+import { createServer, type Server } from "node:http";
+import { afterEach, describe, expect, test } from "vitest";
+import { postEventStream, postJson, ProviderHttpError } from "../../src/main/services/provider-http";
+
+let server: Server | null = null;
+
+function listen(handler: Parameters<typeof createServer>[1]): Promise<string> {
+  return new Promise((resolve) => {
+    server = createServer(handler);
+    server.listen(0, "127.0.0.1", () => {
+      const address = server!.address() as { port: number };
+      resolve(`http://127.0.0.1:${address.port}`);
+    });
+  });
+}
+
+afterEach(() => {
+  server?.closeAllConnections();
+  server?.close();
+  server = null;
+});
+
+describe("provider http deadlines", () => {
+  test("a body that stalls after headers fails instead of hanging", async () => {
+    const baseUrl = await listen((_request, response) => {
+      response.writeHead(200, { "content-type": "application/json" });
+      response.write('{"partial":');
+      // Never ends the body.
+    });
+    await expect(postJson(baseUrl, {}, {}, { bodyTimeoutMs: 100 }))
+      .rejects.toThrow(/timed out/);
+  });
+
+  test("a server that never sends headers fails at the header deadline", async () => {
+    const baseUrl = await listen(() => {
+      // Accept the request and never respond.
+    });
+    await expect(postJson(baseUrl, {}, {}, { headersTimeoutMs: 100 }))
+      .rejects.toThrow(/timed out/);
+  });
+
+  test("an event stream that goes idle fails instead of hanging", async () => {
+    const baseUrl = await listen((_request, response) => {
+      response.writeHead(200, { "content-type": "text/event-stream" });
+      response.write('data: {"first":true}\n\n');
+      // Never sends another event.
+    });
+    const stream = await postEventStream(baseUrl, {}, {}, { idleTimeoutMs: 100 });
+    const seen: unknown[] = [];
+    await expect((async () => {
+      for await (const message of stream.messages) seen.push(message.data);
+    })()).rejects.toThrow(ProviderHttpError);
+    expect(seen).toEqual([{ first: true }]);
+  });
+
+  test("a healthy response is returned unchanged", async () => {
+    const baseUrl = await listen((_request, response) => {
+      response.writeHead(200, { "content-type": "application/json" });
+      response.end('{"ok":true}');
+    });
+    const result = await postJson(baseUrl, {}, {});
+    expect(result.data).toEqual({ ok: true });
+  });
+});