fix(cli): close asset streams when clients disconnect (#5321)

* fix(cli): close asset streams when clients disconnect

* chore: prepare conflict resolution against current main

* fix(cli): retain asset disconnect cleanup with current main

packages/cli/src/utils/staticProjectServer.test.ts

@@ -1,8 +1,10 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
-import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync, type ReadStream } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
-import { Server } from "node:http";
+import fs from "node:fs";
+import { get, Server } from "node:http";
+import { syncBuiltinESMExports } from "node:module";
 import {
   inProcessAnswer,
   serveStaticProjectHtml,
@@ -214,6 +216,70 @@ describe("serveStaticProjectHtml range support", () => {
   });
 });
 
+describe("serveStaticProjectHtml asset stream lifetime", () => {
+  it.each([
+    { label: "full", range: undefined, proxy: false },
+    { label: "open-ended range", range: "bytes=1048576-", proxy: false },
+    { label: "bounded range", range: "bytes=0-16777215", proxy: false },
+    { label: "proxy", range: undefined, proxy: true },
+  ])("closes a $label asset when its client disconnects", async ({ range, proxy }) => {
+    dir = mkdtempSync(join(tmpdir(), "hf-static-disconnect-"));
+    const assetPath = join(dir, "large.mp4");
+    writeFileSync(assetPath, "x");
+    fs.truncateSync(assetPath, 32 * 1024 * 1024);
+    mocks.resolveProxy.mockResolvedValue(assetPath);
+    server = await serveStaticProjectHtml(dir, "<html></html>");
+
+    const createStream = fs.createReadStream;
+    const streams: ReadStream[] = [];
+    const spy = vi.spyOn(fs, "createReadStream").mockImplementation((...args) => {
+      const stream = createStream(...args);
+      streams.push(stream);
+      return stream;
+    });
+    syncBuiltinESMExports();
+
+    try {
+      const status = await new Promise<number | undefined>((resolve, reject) => {
+        const req = get(
+          `${server?.url}large.mp4${proxy ? "?hf-proxy=h264" : ""}`,
+          { headers: range ? { Range: range } : {} },
+          (response) => {
+            response.on("data", () => {
+              response.destroy();
+              req.destroy();
+              resolve(response.statusCode);
+            });
+            response.on("error", reject);
+          },
+        );
+        req.on("error", reject);
+      });
+
+      expect(status).toBe(range ? 206 : 200);
+      expect(streams).toHaveLength(1);
+      await expect.poll(() => streams[0]?.closed, { timeout: 1500, interval: 10 }).toBe(true);
+      expect(streams[0]?.destroyed).toBe(true);
+      expect((await fetch(server.url)).status).toBe(200);
+    } finally {
+      await Promise.all(
+        streams.map(
+          (stream) =>
+            new Promise<void>((resolve) => {
+              if (stream.closed) resolve();
+              else {
+                stream.once("close", resolve);
+                stream.destroy();
+              }
+            }),
+        ),
+      );
+      spy.mockRestore();
+      syncBuiltinESMExports();
+    }
+  });
+});
+
 describe("serveStaticProjectHtml asset roots", () => {
   const extraDirs: string[] = [];
   const mk = (): string => {

packages/cli/src/utils/staticProjectServer.ts

@@ -171,6 +171,7 @@ function serveFileWithRange(
   // just `[start, end]` and closes its own fd on end/error. writeHead is
   // deferred to `open` so a failed open can still answer 500.
   const stream = createReadStream(filePath, { start, end });
+  res.once("close", () => stream.destroy());
   stream.on("open", () => {
     res.writeHead(status, headers);
     stream.pipe(res);