diff --git a/src/orb/ingest.ts b/src/orb/ingest.ts index dd52d104b..a6f1a2ba0 100644 --- a/src/orb/ingest.ts +++ b/src/orb/ingest.ts @@ -24,7 +24,11 @@ function parseContentLength(header: string | null | undefined): number | null { } /** Read the request body with a hard byte ceiling so a hostile sender can't make us buffer unbounded - * input. Returns null when the body exceeds MAX_ORB_INGEST_BODY_BYTES (the caller answers 413). */ + * input. Returns null when the body exceeds MAX_ORB_INGEST_BODY_BYTES OR when the underlying stream + * itself errors (a dropped connection / network reset mid-read, mirrors readOrbRelayRegisterBody in + * ../orb/relay.ts) — both callers (/v1/orb/ingest, /v1/ams/ingest) already treat null identically to + * "reject this request", so a transient read failure degrades the same way an oversized payload does, + * instead of throwing UNCAUGHT out of this function as a bare framework 500. */ export async function readOrbIngestBody(request: Request, contentLengthHeader: string | null | undefined): Promise { const declared = parseContentLength(contentLengthHeader); if (declared !== null && declared > MAX_ORB_INGEST_BODY_BYTES) return null; @@ -35,17 +39,21 @@ export async function readOrbIngestBody(request: Request, contentLengthHeader: s const decoder = new TextDecoder(); let total = 0; let out = ""; - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - total += value.byteLength; - if (total > MAX_ORB_INGEST_BODY_BYTES) { - await reader.cancel(); - return null; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + total += value.byteLength; + if (total > MAX_ORB_INGEST_BODY_BYTES) { + await reader.cancel(); + return null; + } + out += decoder.decode(value, { stream: true }); } - out += decoder.decode(value, { stream: true }); + return out + decoder.decode(); + } catch { + return null; } - return out + decoder.decode(); } interface OrbIngestEvent { diff --git a/test/integration/orb-ingest.test.ts b/test/integration/orb-ingest.test.ts index 064323b3b..d8f29313f 100644 --- a/test/integration/orb-ingest.test.ts +++ b/test/integration/orb-ingest.test.ts @@ -250,6 +250,20 @@ describe("readOrbIngestBody()", () => { const req = new Request("http://collector", { method: "POST", body: stream, ...({ duplex: "half" } as object) }); expect(await readOrbIngestBody(req, null)).toBeNull(); }); + + it("REGRESSION (#8330): returns null instead of throwing when the underlying stream errors mid-read", async () => { + // Mirrors readOrbRelayRegisterBody's identical dropped-connection regression test (orb-relay.test.ts) — a + // first successful chunk (some bytes already arrived) followed by a stream error is the realistic shape of + // a mid-read network drop, not an error on the very first read. + const stream = new ReadableStream({ + pull(controller) { + controller.enqueue(new TextEncoder().encode('{"instance_id":')); + controller.error(new Error("simulated network reset")); + }, + }); + const req = new Request("http://collector", { method: "POST", body: stream, ...({ duplex: "half" } as object) }); + await expect(readOrbIngestBody(req, null)).resolves.toBeNull(); + }); }); describe("POST /v1/orb/ingest route", () => { @@ -281,6 +295,22 @@ describe("POST /v1/orb/ingest route", () => { expect(((await res.json()) as { error: string }).error).toBe("payload_too_large"); }); + it("REGRESSION (#8330): a dropped connection mid-upload returns the same clean 413, not a framework 500", async () => { + const stream = new ReadableStream({ + pull(controller) { + controller.enqueue(new TextEncoder().encode('{"instance_id":')); + controller.error(new Error("simulated network reset")); + }, + }); + const res = await app.request( + "/v1/orb/ingest", + { method: "POST", body: stream, ...({ duplex: "half" } as object) }, + createTestEnv(), + ); + expect(res.status).toBe(413); + expect(((await res.json()) as { error: string }).error).toBe("payload_too_large"); + }); + it("optional collector token (#1285): open when unset; enforced once ORB_INGEST_TOKEN is set", async () => { const body = JSON.stringify({ instance_id: "abc0", events: [{ repo_hash: "rhash", pr_hash: "phash", outcome: "merged" }] }); const post = (env: Env, authorization?: string) =>