fix(download): reset the socket when a stream is shorter than Content-Length (#617)

The download route sets Content-Length from a stat and then streams the
object; when the stat size exceeds the bytes the stream yields (#590
"cause 2"), the client hangs on keep-alive framing waiting for a tail
that never arrives. Both send paths now run through a backpressure-safe
byte-counting Transform that resets the socket on a shortfall, so the
download fails at once instead of hanging. Adds a real-socket regression
test at the generic download route, the coverage gap #590 named.

Refs #590

Co-authored-by: harshjainnn <170849281+harshjainnn@users.noreply.github.com>

apps/api/src/routes/files.ts

@@ -1,5 +1,6 @@
 import { randomUUID } from "node:crypto";
 import { extname } from "node:path";
+import { pipeline, type Readable, Transform } from "node:stream";
 import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
 import sharp from "sharp";
 import { readImageDimensions } from "../lib/exiftool.js";
@@ -22,6 +23,62 @@ function isPathTraversal(segment: string): boolean {
   );
 }
 
+/**
+ * Stream a stored object to the client while verifying that the bytes delivered
+ * match the declared Content-Length. When a storage backend's stat() size
+ * disagrees with the bytes its read stream yields (issue #590 "cause 2"), a
+ * stream shorter than the declared length leaves the browser hanging on
+ * keep-alive framing, waiting for a tail that never arrives. That is the
+ * "download starts but never finishes" symptom. Resetting the socket on a
+ * shortfall turns that silent hang into an immediate, logged failure.
+ *
+ * The byte count lives inside a Transform, not a "data" listener on the piped
+ * stream, so it never forces the source into flowing mode and always respects
+ * backpressure: a slow or paused client is never mistaken for a stalled stream.
+ */
+function guardedDownloadStream(
+  request: FastifyRequest,
+  reply: FastifyReply,
+  source: Readable,
+  key: string,
+  expectedBytes: number,
+): Readable {
+  let delivered = 0;
+  const counted = new Transform({
+    transform(chunk, _encoding, callback) {
+      delivered += chunk.length;
+      callback(null, chunk);
+    },
+  });
+  // pipeline() forwards a source read error into `counted` (so Fastify tears the
+  // response down instead of the client hanging on a half-sent body) and
+  // destroys both streams if the client disconnects, so no source handle leaks.
+  pipeline(source, counted, (err) => {
+    const code = (err as NodeJS.ErrnoException | null)?.code;
+    // A premature close is the normal shape of a client cancelling a download.
+    // Fastify already resets the response and logs a genuine source error; this
+    // adds the object key it omits, at warn so it stays out of error tracking.
+    if (err && code !== "ERR_STREAM_PREMATURE_CLOSE") {
+      request.log.warn({ key, err }, "Download source stream failed before completing");
+    }
+  });
+  counted.on("end", () => {
+    // A stream that ends short of the declared length is what leaves the client
+    // hanging on keep-alive framing; reset the socket so the download fails at
+    // once instead of waiting for a tail that never arrives. An over-count does
+    // not reach here (Node breaks the response as the extra bytes are written),
+    // and it fails the client on its own, so a shortfall is the only case here.
+    if (delivered < expectedBytes) {
+      request.log.error(
+        { key, declaredSize: expectedBytes, delivered },
+        "Download stream ended short of the declared Content-Length; resetting connection",
+      );
+      reply.raw.destroy();
+    }
+  });
+  return counted;
+}
+
 export async function fileRoutes(app: FastifyInstance): Promise<void> {
   // ── POST /api/v1/upload ────────────────────────────────────────
   app.post(
@@ -147,7 +204,15 @@ export async function fileRoutes(app: FastifyInstance): Promise<void> {
           )
           .header("Content-Range", `bytes ${start}-${clampedEnd}/${size}`)
           .header("Content-Length", String(clampedEnd - start + 1))
-          .send(await getObjectStream(key, { start, end: clampedEnd }));
+          .send(
+            guardedDownloadStream(
+              request,
+              reply,
+              await getObjectStream(key, { start, end: clampedEnd }),
+              key,
+              clampedEnd - start + 1,
+            ),
+          );
       }
 
       reply
@@ -157,7 +222,9 @@ export async function fileRoutes(app: FastifyInstance): Promise<void> {
           `attachment; filename="${encodeURIComponent(filename)}"; filename*=UTF-8''${encodeURIComponent(filename)}`,
         )
         .header("Content-Length", String(size));
-      return reply.send(await getObjectStream(key));
+      return reply.send(
+        guardedDownloadStream(request, reply, await getObjectStream(key), key, size),
+      );
     },
   );
 

tests/integration/platform/download-socket.test.ts

@@ -0,0 +1,123 @@
+import type { AddressInfo } from "node:net";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import * as objectStorage from "../../../apps/api/src/lib/object-storage.js";
+import { deletePrefix, putObject } from "../../../apps/api/src/lib/object-storage.js";
+import { buildTestApp, type TestApp } from "../test-server.js";
+
+// Every other download test exercises the route through app.inject, which
+// buffers the whole body and so can never observe a stream that stalls or
+// under-delivers over a real connection. That blind spot is why #590 shipped:
+// the bug only surfaces on a live socket. #604 closed the gap for the
+// image-to-pdf route; this covers the generic
+// GET /api/v1/download/:jobId/:filename route every tool shares, by serving
+// over a real TCP socket (app.listen) and reading with fetch.
+describe("download endpoint over a real socket (#590)", () => {
+  let testApp: TestApp;
+  let baseUrl: string;
+  const jobId = `dlsock-${process.pid}`;
+  // 1 MiB spans many fs.ReadStream chunks and several TCP writes, so the parity
+  // assertion exercises real multi-chunk streaming, not a one-shot body that
+  // would pass trivially.
+  const SIZE = 1024 * 1024;
+
+  beforeAll(async () => {
+    testApp = await buildTestApp();
+    await putObject(`outputs/${jobId}/result.bin`, Buffer.alloc(SIZE, 0x61));
+    // Bind a real port so the download travels over TCP instead of app.inject.
+    await testApp.app.listen({ port: 0, host: "127.0.0.1" });
+    const addr = testApp.app.server.address() as AddressInfo;
+    baseUrl = `http://127.0.0.1:${addr.port}`;
+  }, 30_000);
+
+  afterAll(async () => {
+    await deletePrefix(`outputs/${jobId}/`);
+    await testApp.cleanup();
+  }, 10_000);
+
+  // Force the route's declared Content-Length (from getObjectSize) to exceed the
+  // bytes the stream yields: #590 "cause 2", a stat/stream size disagreement.
+  function mockOverReportedSize() {
+    const realGetObjectSize = objectStorage.getObjectSize;
+    return vi.spyOn(objectStorage, "getObjectSize").mockImplementation(async (key: string) => {
+      const actual = await realGetObjectSize(key);
+      return key.includes("result.bin") ? actual + 4096 : actual;
+    });
+  }
+
+  // Drive a download that should be reset mid-stream and report how it settled.
+  // A 10s deadline is the trip wire: a regression that lets the route hang aborts
+  // via the deadline instead, which the callers assert against.
+  async function fetchUnderDeadline(init: { headers?: Record<string, string> } = {}) {
+    const controller = new AbortController();
+    const deadline = setTimeout(() => controller.abort(), 10_000);
+    let status = 0;
+    let threw = false;
+    try {
+      const res = await fetch(`${baseUrl}/api/v1/download/${jobId}/result.bin`, {
+        ...init,
+        signal: controller.signal,
+      });
+      status = res.status;
+      await res.arrayBuffer();
+    } catch {
+      threw = true;
+    } finally {
+      clearTimeout(deadline);
+    }
+    return { status, threw, aborted: controller.signal.aborted };
+  }
+
+  it("delivers exactly Content-Length bytes over TCP", async () => {
+    // A complete 1 MiB body proves the multi-chunk stream finishes and sets
+    // X-Accel-Buffering. Shortfall detection is covered by the reset tests
+    // below; a truncated body never reaches the parity check here, because the
+    // client's read hangs or rejects first. The deadline turns a success-path
+    // hang regression into a fast failure instead of a 30s timeout.
+    const controller = new AbortController();
+    const deadline = setTimeout(() => controller.abort(), 10_000);
+    try {
+      const res = await fetch(`${baseUrl}/api/v1/download/${jobId}/result.bin`, {
+        signal: controller.signal,
+      });
+      expect(res.status).toBe(200);
+      // #604: reverse proxies must not buffer the download.
+      expect(res.headers.get("x-accel-buffering")).toBe("no");
+      const declared = Number(res.headers.get("content-length"));
+      expect(declared).toBe(SIZE);
+      const body = Buffer.from(await res.arrayBuffer());
+      expect(body.length).toBe(declared);
+    } finally {
+      clearTimeout(deadline);
+    }
+  }, 30_000);
+
+  it("resets the connection instead of hanging when the stream is short (full body)", async () => {
+    const spy = mockOverReportedSize();
+    try {
+      const { status, threw, aborted } = await fetchUnderDeadline();
+      expect(status).toBe(200);
+      // The client observed the reset (a premature-close read error)...
+      expect(threw).toBe(true);
+      // ...and it was the guard that reset it, not our deadline firing on a hang.
+      expect(aborted).toBe(false);
+    } finally {
+      spy.mockRestore();
+    }
+  }, 30_000);
+
+  it("resets a ranged (206) download instead of hanging when the stream is short", async () => {
+    // The guard wraps the range branch too, with its own byte math
+    // (clampedEnd - start + 1); exercise that path over the socket.
+    const spy = mockOverReportedSize();
+    try {
+      const { status, threw, aborted } = await fetchUnderDeadline({
+        headers: { range: "bytes=0-" },
+      });
+      expect(status).toBe(206);
+      expect(threw).toBe(true);
+      expect(aborted).toBe(false);
+    } finally {
+      spy.mockRestore();
+    }
+  }, 30_000);
+});