refactor(api): extract the log-export stream builder to clear the new-code complexity ratchet (#13999)

changelog.d/fixes/13999-logs-export-mid-stream-error.md

@@ -0,0 +1 @@
+- **fix(api):** `GET /api/logs/export` now settles its HTTP response instead of hanging forever when the DB row source throws mid-stream — the row-iteration loop is wrapped in try/catch, the failure is logged, and the JSON document is closed out cleanly with additive `emitted`/`error` trailer fields so the client always gets a response (#13999).

src/app/api/logs/export/route.ts

@@ -2,6 +2,9 @@ import { countCallLogsSince, iterateCallLogsSince } from "@/lib/usage/callLogs";
 import { requireManagementAuth } from "@/lib/api/requireManagementAuth";
 import { countProxyLogsSince, iterateProxyLogsSince } from "@/lib/db/proxyLogs";
 import { sanitizeErrorMessage } from "@omniroute/open-sse/utils/error";
+import { logger } from "@/shared/utils/logger";
+
+const log = logger.child({ module: "logs-export" });
 
 /**
  * GET /api/logs/export — export logs as JSON (streamed)
@@ -32,6 +35,77 @@ import { sanitizeErrorMessage } from "@omniroute/open-sse/utils/error";
 const MAX_ROWS = 50_000;
 const DEFAULT_ROWS = 10_000;
 
+/**
+ * Build the streamed JSON body for a log export.
+ *
+ * Streams one row at a time — the row source (`rows`) is a cursor/generator
+ * bounded by SQL LIMIT, so peak memory is bounded by one hydrated row, not the
+ * full matching set (#13123). `capped`/`limit`/`totalAvailable` travel in the
+ * HEADER (not a trailer, as before) so a client consuming the stream
+ * incrementally learns about truncation before it has processed every row.
+ */
+function buildLogExportStream({
+  rows,
+  header,
+  logType,
+  hours,
+}: {
+  rows: AsyncIterable<unknown> | Iterable<unknown>;
+  header: Record<string, unknown>;
+  logType: string;
+  hours: number;
+}): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  return new ReadableStream({
+    async start(controller) {
+      // header ends with `}`, we strip it to append `,"logs":[...]}`
+      controller.enqueue(encoder.encode(JSON.stringify(header).slice(0, -1) + ',"logs":['));
+      let index = 0;
+      let streamError: unknown = null;
+      try {
+        for await (const row of rows) {
+          if (index > 0) controller.enqueue(encoder.encode(","));
+          controller.enqueue(encoder.encode(JSON.stringify(row)));
+          index++;
+        }
+      } catch (err) {
+        // #13999: the row source (a DB cursor/generator) can throw partway through
+        // iteration, after headers and some rows have already gone out over the wire.
+        // Letting the exception propagate out of an async `start()` auto-errors the
+        // underlying Web ReadableStream, but once that stream is bridged onto a real
+        // Node HTTP response (as any Node-based adapter does), a source error does not
+        // end or destroy the destination response — the client's fetch() never resolves
+        // and never rejects, and it hangs forever (proven by
+        // tests/unit/repro-13999-mid-stream-error.test.ts). Instead of erroring the
+        // stream, close the JSON document out cleanly with a trailing `error`/`emitted`
+        // marker so the HTTP response always completes, and log the failure server-side.
+        streamError = err;
+        log.error(
+          { err, emitted: index, type: logType, hours },
+          "logs export stream failed mid-iteration"
+        );
+      }
+      controller.enqueue(encoder.encode("]"));
+      controller.enqueue(encoder.encode(streamError ? errorTail(index, streamError) : "}\n"));
+      controller.close();
+    },
+  });
+}
+
+/**
+ * `,"emitted":N,"error":"..."}` — the sibling fields that close a truncated
+ * export out as well-formed JSON (#13999).
+ */
+function errorTail(emitted: number, streamError: unknown): string {
+  const tail = JSON.stringify({
+    emitted,
+    error: sanitizeErrorMessage(
+      streamError instanceof Error ? streamError.message : String(streamError)
+    ),
+  });
+  return "," + tail.slice(1, -1) + "}\n";
+}
+
 export async function GET(request: Request) {
   const authError = await requireManagementAuth(request);
   if (authError) return authError;
@@ -70,34 +144,16 @@ export async function GET(request: Request) {
     const count = Math.min(totalAvailable, limit);
     const filename = `omniroute-${tableName}-${hours}h-${new Date().toISOString().slice(0, 10)}.json`;
 
-    // Stream the JSON response one row at a time — the row source itself
-    // (`rows`) is a cursor/generator bounded by SQL LIMIT, so peak memory is
-    // bounded by one hydrated row, not the full matching set (#13123).
-    const encoder = new TextEncoder();
-    const stream = new ReadableStream({
-      async start(controller) {
-        // `capped`/`limit`/`totalAvailable` are written into the HEADER (not
-        // just a trailer at the end, as before) so a client consuming the
-        // stream incrementally learns about truncation before it has
-        // processed every row. Only present when the export is actually
-        // capped, matching the previous (trailer-only) contract shape.
-        const header = JSON.stringify({
-          count,
-          hours,
-          type: logType,
-          ...(capped ? { capped: true, limit, totalAvailable } : {}),
-        });
-        // header ends with `}`, we strip it to append `,"logs":[...]}`
-        controller.enqueue(encoder.encode(header.slice(0, -1) + ',"logs":['));
-        let index = 0;
-        for await (const row of rows) {
-          if (index > 0) controller.enqueue(encoder.encode(","));
-          controller.enqueue(encoder.encode(JSON.stringify(row)));
-          index++;
-        }
-        controller.enqueue(encoder.encode("]}\n"));
-        controller.close();
+    const stream = buildLogExportStream({
+      rows,
+      header: {
+        count,
+        hours,
+        type: logType,
+        ...(capped ? { capped: true, limit, totalAvailable } : {}),
       },
+      logType,
+      hours,
     });
 
     return new Response(stream, {

tests/unit/repro-13999-mid-stream-error.test.ts

@@ -0,0 +1,155 @@
+/**
+ * Regression test for #13999 item 1: GET /api/logs/export's ReadableStream
+ * had no try/catch around the row-iteration loop inside `start()`. When the
+ * row source threw mid-iteration, the stream had ALREADY sent `200 OK` +
+ * `Content-Disposition` + a valid-looking JSON header + some rows, and the
+ * route never called `controller.error()` or logged anything.
+ *
+ * IMPORTANT finding (from the original triage): calling `route.GET()`
+ * directly in-process and reading the resulting `Response` does NOT
+ * reproduce the reported symptom — Node's spec-compliant ReadableStream
+ * auto-errors the stream when an async `start()` rejects, so an in-memory
+ * `response.text()` correctly rejects. The silent hang only appeared once
+ * the Response body crossed a REAL HTTP transport (exactly what the
+ * dashboard's `fetch()` does): piping a Web ReadableStream into a Node
+ * `http.ServerResponse` via `Readable.fromWeb(...).pipe(res)` does NOT call
+ * `res.end()`/`res.destroy()` when the source errors — the connection was
+ * left open, so the client's `fetch()`/`res.text()` never resolved AND
+ * never rejected.
+ *
+ * The fix makes the route catch the mid-iteration error itself, log it, and
+ * close the JSON document out cleanly with a trailing `error`/`emitted`
+ * marker instead of letting the stream error — so the HTTP response always
+ * completes (no hang) and the client can detect the truncation from the
+ * response body itself.
+ *
+ * This test spins up a real (loopback, ephemeral-port) `http.Server` whose
+ * request handler calls the REAL `route.GET()` and bridges its Response the
+ * same way any Node-based HTTP adapter would, then makes a REAL `fetch()`
+ * against it — proving the response settles, not just asserting on the
+ * in-process Response object.
+ *
+ * Run: node --experimental-test-module-mocks --import tsx/esm --test tests/unit/repro-13999-mid-stream-error.test.ts
+ */
+import { test, mock } from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import http from "node:http";
+import { Readable } from "node:stream";
+import type { ReadableStream as NodeWebReadableStream } from "node:stream/web";
+
+import { useDecollidedMigrationsDir } from "./helpers/decollidedMigrationsDir.ts";
+
+useDecollidedMigrationsDir();
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-logs-export-13999-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+
+if (typeof mock.module !== "function") {
+  test("(skipped) requires --experimental-test-module-mocks", () => {
+    assert.ok(true);
+  });
+} else {
+  // Mock the DB-layer module the route imports from ("@/lib/usage/callLogs"),
+  // resolved here via the same relative path the sibling #13123 test uses
+  // directly (tests/unit/logs-export-streaming-13123.test.ts:23) — proving
+  // the specifier resolves to the same file the route sees.
+  mock.module("../../src/lib/usage/callLogs.ts", {
+    exports: {
+      countCallLogsSince: () => 3,
+      iterateCallLogsSince: async function* () {
+        yield { id: "call-0", note: "first row hydrated fine" };
+        yield { id: "call-1", note: "second row hydrated fine" };
+        // Simulate a DB error hydrating the 3rd row (getCallLogById throwing,
+        // a corrupt artifact read, a connection blip, etc.) — exactly the
+        // failure mode #13999 describes: it happens strictly AFTER headers
+        // and the first chunks have already gone out over the wire.
+        throw new Error("simulated DB error hydrating call-2 mid-stream");
+      },
+    },
+  });
+
+  const route = await import("../../src/app/api/logs/export/route.ts");
+
+  test("#13999: a mid-stream DB error must not hang the real HTTP response — it must settle with a truncation marker", async () => {
+    const server = http.createServer(async (req, res) => {
+      const response = await route.GET(
+        new Request(`http://localhost${req.url}`, {
+          headers: req.headers as Record<string, string>,
+        })
+      );
+      res.writeHead(response.status, Object.fromEntries(response.headers.entries()));
+      if (!response.body) {
+        res.end();
+        return;
+      }
+      const nodeStream = Readable.fromWeb(
+        response.body as unknown as NodeWebReadableStream<Uint8Array>
+      );
+      nodeStream.on("error", () => {});
+      nodeStream.pipe(res);
+    });
+
+    await new Promise<void>((resolve) => server.listen(0, resolve));
+    const address = server.address();
+    const port = typeof address === "object" && address ? address.port : 0;
+
+    try {
+      const abortController = new AbortController();
+      const readPromise = fetch(
+        `http://localhost:${port}/api/logs/export?hours=168&type=call-logs`,
+        {
+          signal: abortController.signal,
+        }
+      ).then((r) => r.text());
+
+      let settled = false;
+      let settledText: string | null = null;
+      readPromise.then(
+        (text) => {
+          settled = true;
+          settledText = text;
+        },
+        () => {
+          settled = true;
+        }
+      );
+
+      // Give the real request a generous window to either succeed (with a
+      // clear truncation signal) or fail cleanly. Fixed behavior settles
+      // almost immediately; this window only guards against a regression.
+      await new Promise((resolve) => setTimeout(resolve, 2000));
+
+      if (!settled) {
+        abortController.abort();
+      }
+
+      assert.equal(
+        settled,
+        true,
+        "#13999 regressed: the export HTTP response neither resolved nor rejected " +
+          "within 2s of a mid-stream DB error — it hung again."
+      );
+
+      assert.ok(settledText, "expected the response body to be readable text");
+      const parsed = JSON.parse(settledText as unknown as string);
+      assert.equal(
+        parsed.emitted,
+        2,
+        "expected exactly the 2 successfully-hydrated rows to be marked emitted"
+      );
+      assert.equal(
+        parsed.logs.length,
+        2,
+        "expected exactly the 2 successfully-hydrated rows in the body"
+      );
+      assert.ok(
+        typeof parsed.error === "string" && parsed.error.length > 0,
+        "expected a sanitized error message surfaced in the truncated JSON body"
+      );
+    } finally {
+      server.close();
+    }
+  });
+}