Contain static response disconnects at the Node transport boundary

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@askrjs/node",
-  "version": "0.2.1",
+  "version": "0.2.2",
   "description": "Node http adapter for @askrjs/server",
   "keywords": [
     "askr",

src/serve.ts

@@ -144,6 +144,7 @@ export async function serve(
         }
       }
       if (!candidate || !file?.isFile()) {
+        if (response.destroyed || response.writableEnded) return;
         response
           .writeHead(404, {
             "content-type": "text/plain; charset=utf-8",
@@ -152,6 +153,7 @@ export async function serve(
           .end("Not Found");
         return;
       }
+      if (response.destroyed || response.writableEnded) return;
       response.writeHead(200, {
         "content-type": mimeTypes[extension] ?? "application/octet-stream",
         "content-length": Number(file.size),
@@ -162,7 +164,15 @@ export async function serve(
         "x-content-type-options": "nosniff",
       });
       if (method === "HEAD") response.end();
-      else pipeline(createReadStream(candidate), response, () => undefined);
+      else {
+        const source = createReadStream(candidate);
+        const stopSource = () => source.destroy();
+        response.once("close", stopSource);
+        pipeline(source, response, (error) => {
+          response.off("close", stopSource);
+          if (error) source.destroy();
+        });
+      }
       return;
     }
     applicationHandler(request, response);

tests/node.test.ts

@@ -1065,6 +1065,49 @@ describe("serve", () => {
     }
   });
 
+  it("should keep serving after clients disconnect before and during static asset streaming", async () => {
+    const root = await mkdtemp(join(tmpdir(), "askr-node-disconnect-assets-"));
+    await writeFile(join(root, "large-12345678.js"), new Uint8Array(8 * 1024 * 1024));
+    const served = await serve(
+      { fetch: async () => new Response("app-alive") },
+      { assets: { root }, signals: false },
+    );
+    const address = served.server.address();
+    if (!address || typeof address === "string") throw new Error("Expected TCP address");
+
+    try {
+      for (let index = 0; index < 10; index += 1) {
+        const socket = createConnection(address.port, "127.0.0.1");
+        await once(socket, "connect");
+        socket.write(
+          `GET /large-12345678.js HTTP/1.1\r\nHost: 127.0.0.1:${address.port}\r\nConnection: close\r\n\r\n`,
+        );
+        socket.destroy();
+      }
+
+      await new Promise<void>((resolve, reject) => {
+        const request = nodeRequest(`${served.url}/large-12345678.js`);
+        request.once("response", (response) => {
+          response.once("data", () => {
+            response.destroy();
+            resolve();
+          });
+          response.once("error", () => resolve());
+        });
+        request.once("error", reject);
+        request.end();
+      });
+
+      expect(await (await fetch(`${served.url}/page`)).text()).toBe("app-alive");
+      const asset = await fetch(`${served.url}/large-12345678.js`);
+      expect(asset.status).toBe(200);
+      expect((await asset.arrayBuffer()).byteLength).toBe(8 * 1024 * 1024);
+    } finally {
+      await served.close();
+      await rm(root, { recursive: true, force: true });
+    }
+  });
+
   it("should let excluded dotted routes bypass static serving under concurrency", async () => {
     const root = await mkdtemp(join(tmpdir(), "askr-node-excluded-assets-"));
     await writeFile(join(root, "app.js"), "asset");