fix(voyage): close response body stream when batch output JSONL parsing throws

solodmd · solodmd · commit 0ee26fcd5c2c · 2026-07-02T09:55:09.000+08:00
The batch output file download path creates a readline interface over a Readable.fromWeb() response body stream. If JSON.parse throws on a malformed JSONL line, the for-await loop exits via exception but the readline interface and underlying Readable stream were never explicitly closed or destroyed, leaving the HTTP response body stream dangling. Extract the stream reading into , a testable helper that wraps the iteration in a try-finally so both reader.close() and inputStream.destroy() are always called, matching the pattern established in #98493 for the same class of leak.
diff --git a/extensions/voyage/embedding-batch.test.ts b/extensions/voyage/embedding-batch.test.ts
@@ -1,9 +1,14 @@
 // Voyage batch tests cover bounded status/error response reads.
 import { describe, expect, it } from "vitest";
-import type { VoyageEmbeddingClient } from "./embedding-provider.js";
 import { testing } from "./embedding-batch.js";
+import type { VoyageEmbeddingClient } from "./embedding-provider.js";
 
-const { fetchVoyageBatchStatus, readVoyageBatchError, VOYAGE_BATCH_RESPONSE_MAX_BYTES } = testing;
+const {
+  fetchVoyageBatchStatus,
+  readVoyageBatchError,
+  readBatchOutputContent,
+  VOYAGE_BATCH_RESPONSE_MAX_BYTES,
+} = testing;
 
 function buildClient(): VoyageEmbeddingClient {
   return {
@@ -215,3 +220,91 @@ describe("voyage batch bounded reads", () => {
     ).rejects.toThrow(/voyage batch status failed: 503 voyage upstream is down/);
   });
 });
+
+describe("readBatchOutputContent stream cleanup", () => {
+  it("returns without error on a null body", async () => {
+    const response = new Response(null, { status: 200 });
+    const remaining = new Set<string>();
+    const errors: string[] = [];
+    const byCustomId = new Map<string, number[]>();
+
+    await expect(
+      readBatchOutputContent(response, remaining, errors, byCustomId),
+    ).resolves.toBeUndefined();
+  });
+
+  it("applies every valid JSONL line and tracks remaining and errors", async () => {
+    const body = [
+      JSON.stringify({
+        custom_id: "req-0",
+        response: { status_code: 200, body: { data: [{ embedding: [0.1, 0.2] }] } },
+      }),
+      JSON.stringify({
+        custom_id: "req-1",
+        response: { status_code: 200, body: { data: [{ embedding: [0.3, 0.4] }] } },
+      }),
+      JSON.stringify({ custom_id: "req-2", error: { message: "voyage failed" } }),
+      "",
+    ].join("\n");
+    const response = new Response(body, {
+      status: 200,
+      headers: { "content-type": "application/x-ndjson" },
+    });
+    const remaining = new Set(["req-0", "req-1", "req-2", "req-3"]);
+    const errors: string[] = [];
+    const byCustomId = new Map<string, number[]>();
+
+    await readBatchOutputContent(response, remaining, errors, byCustomId);
+
+    // valid lines should be removed from remaining
+    expect(remaining.has("req-0")).toBe(false);
+    expect(remaining.has("req-1")).toBe(false);
+    // Error line is tracked; applyEmbeddingBatchOutputLine also removes
+    // error custom_ids from remaining (both consumed and errored lines)
+    expect(remaining.has("req-2")).toBe(false);
+    expect(remaining.has("req-3")).toBe(true);
+    // error line should be tracked in errors array
+    expect(errors).toHaveLength(1);
+    expect(errors[0]).toContain("req-2");
+  });
+
+  it.runIf(process.platform === "linux")(
+    "cancels the response stream when JSON.parse encounters malformed JSONL",
+    async () => {
+      let wasCanceled = false;
+      const encoder = new TextEncoder();
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          // First line: valid
+          controller.enqueue(
+            encoder.encode(
+              JSON.stringify({
+                custom_id: "req-0",
+                response: { status_code: 200, body: { data: [{ embedding: [0.1] }] } },
+              }) + "\n",
+            ),
+          );
+          // Second line: malformed JSON — this should trigger the stream cancel
+          controller.enqueue(encoder.encode("{not valid json}\n"));
+        },
+        cancel() {
+          wasCanceled = true;
+        },
+      });
+      const response = new Response(stream, {
+        status: 200,
+        headers: { "content-type": "application/x-ndjson" },
+      });
+      const remaining = new Set(["req-0"]);
+      const errors: string[] = [];
+      const byCustomId = new Map<string, number[]>();
+
+      await expect(
+        readBatchOutputContent(response, remaining, errors, byCustomId),
+      ).rejects.toThrow();
+
+      // The stream should have been destroyed/canceled, not left dangling
+      expect(wasCanceled).toBe(true);
+    },
+  );
+});
diff --git a/extensions/voyage/embedding-batch.ts b/extensions/voyage/embedding-batch.ts
@@ -317,20 +317,7 @@ export async function runVoyageEmbeddingBatches(
           if (!contentRes.body) {
             return;
           }
-          const reader = createInterface({
-            input: Readable.fromWeb(
-              contentRes.body as unknown as import("stream/web").ReadableStream,
-            ),
-            terminal: false,
-          });
-
-          for await (const rawLine of reader) {
-            if (!rawLine.trim()) {
-              continue;
-            }
-            const line = JSON.parse(rawLine) as VoyageBatchOutputLine;
-            applyEmbeddingBatchOutputLine({ line, remaining, errors, byCustomId });
-          }
+          await readBatchOutputContent(contentRes, remaining, errors, byCustomId);
         },
       });
 
@@ -346,8 +333,45 @@ export async function runVoyageEmbeddingBatches(
   });
 }
 
+/**
+ * Stream-read a Voyage batch output file response body line by line, applying
+ * each parsed output line through `applyEmbeddingBatchOutputLine`.
+ *
+ * The response body is an untrusted external stream. This helper wraps the
+ * readline iteration in a try-finally so the readline interface and underlying
+ * Node.js Readable are always closed / destroyed, even when JSON.parse or
+ * applyEmbeddingBatchOutputLine throws.
+ */
+async function readBatchOutputContent(
+  contentRes: Response,
+  remaining: Set<string>,
+  errors: string[],
+  byCustomId: Map<string, number[]>,
+): Promise<void> {
+  if (!contentRes.body) {
+    return;
+  }
+  const inputStream = Readable.fromWeb(
+    contentRes.body as unknown as import("stream/web").ReadableStream,
+  );
+  const reader = createInterface({ input: inputStream, terminal: false });
+  try {
+    for await (const rawLine of reader) {
+      if (!rawLine.trim()) {
+        continue;
+      }
+      const line = JSON.parse(rawLine) as VoyageBatchOutputLine;
+      applyEmbeddingBatchOutputLine({ line, remaining, errors, byCustomId });
+    }
+  } finally {
+    reader.close();
+    inputStream.destroy();
+  }
+}
+
 export const testing = {
   fetchVoyageBatchStatus,
   readVoyageBatchError,
+  readBatchOutputContent,
   VOYAGE_BATCH_RESPONSE_MAX_BYTES,
 } as const;
