fix(orb): catch stream-read errors in readOrbIngestBody (#8330)

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<string | null> {
   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 {

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<Uint8Array>({
+      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<Uint8Array>({
+      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) =>