fix(server) handle ignored body and node:http destroy (#10268)

src/bun.js/api/server.zig

@@ -1846,10 +1846,15 @@ fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool, comp
 
             if (this.request_body) |body| {
                 ctxLog("finalizeWithoutDeinit: request_body != null", .{});
+                // Case 1:
                 // User called .blob(), .json(), text(), or .arrayBuffer() on the Request object
                 // but we received nothing or the connection was aborted
                 // the promise is pending
-                if (body.value == .Locked and body.value.Locked.hasPendingPromise()) {
+                // Case 2:
+                // User ignored the body and the connection was aborted or ended
+                // Case 3:
+                // Stream was not consumed and the connection was aborted or ended
+                if (body.value == .Locked) {
                     body.value.toErrorInstance(JSC.toTypeError(.ABORT_ERR, "Request aborted", .{}, this.server.globalThis), this.server.globalThis);
                 }
             }

src/bun.js/webcore/body.zig

@@ -828,6 +828,7 @@ pub const Body = struct {
         }
 
         pub fn toErrorInstance(this: *Value, error_instance: JSC.JSValue, global: *JSGlobalObject) void {
+            error_instance.ensureStillAlive();
             if (this.* == .Locked) {
                 var locked = this.Locked;
                 locked.deinit = true;
@@ -842,7 +843,7 @@ pub const Body = struct {
                 }
 
                 if (locked.readable.get()) |readable| {
-                    readable.done(global);
+                    readable.abort(global);
                     locked.readable.deinit();
                 }
                 // will be unprotected by body value deinit

src/bun.js/webcore/streams.zig

@@ -142,21 +142,37 @@ pub const ReadableStream = struct {
     }
 
     pub fn done(this: *const ReadableStream, globalThis: *JSGlobalObject) void {
+        JSC.markBinding(@src());
+        // done is called when we are done consuming the stream
+        // cancel actually mark the stream source as done
+        // this will resolve any pending promises to done: true
+        switch (this.ptr) {
+            .Blob => |source| {
+                source.parent().cancel();
+            },
+            .File => |source| {
+                source.parent().cancel();
+            },
+            .Bytes => |source| {
+                source.parent().cancel();
+            },
+            else => {},
+        }
         this.detachIfPossible(globalThis);
     }
 
     pub fn cancel(this: *const ReadableStream, globalThis: *JSGlobalObject) void {
         JSC.markBinding(@src());
-
+        // cancel the stream
         ReadableStream__cancel(this.value, globalThis);
-        this.detachIfPossible(globalThis);
+        // mark the stream source as done
+        this.done(globalThis);
     }
 
     pub fn abort(this: *const ReadableStream, globalThis: *JSGlobalObject) void {
         JSC.markBinding(@src());
-
-        ReadableStream__cancel(this.value, globalThis);
-        this.detachIfPossible(globalThis);
+        // for now we are just calling cancel should be fine
+        this.cancel(globalThis);
     }
 
     pub fn forceDetach(this: *const ReadableStream, globalObject: *JSGlobalObject) void {

src/js/node/http.ts

@@ -1089,7 +1089,6 @@ Object.defineProperty(OutgoingMessage.prototype, "finished", {
 
 function emitCloseNT(self) {
   if (!self._closed) {
-    self.destroyed = true;
     self._closed = true;
     self.emit("close");
   }
@@ -1432,6 +1431,15 @@ class ClientRequest extends OutgoingMessage {
     this.#bodyChunks.push(...chunks);
     callback();
   }
+  _destroy(err, callback) {
+    this.destroyed = true;
+    // If request is destroyed we abort the current response
+    this[kAbortController]?.abort?.();
+    if (err) {
+      this.emit("error", err);
+    }
+    callback();
+  }
 
   _final(callback) {
     this.#finished = true;
@@ -1519,7 +1527,7 @@ class ClientRequest extends OutgoingMessage {
 
   abort() {
     if (this.aborted) return;
-    this[kAbortController]!.abort();
+    this[kAbortController]?.abort?.();
     // TODO: Close stream if body streaming
   }
 

test/js/bun/http/serve-body-leak.test.ts

@@ -0,0 +1,134 @@
+import { join } from "path";
+import { it, expect, beforeAll, afterAll } from "bun:test";
+import { bunExe, bunEnv } from "harness";
+import type { Subprocess } from "bun";
+
+const ACCEPTABLE_MEMORY_LEAK = 2; //MB for acceptable memory leak variance
+const payload = "1".repeat(32 * 1024); // decent size payload to test memory leak
+
+let url: URL;
+let process: Subprocess<"ignore", "pipe", "inherit"> | null = null;
+beforeAll(async () => {
+  process = Bun.spawn([bunExe(), "--smol", join(import.meta.dirname, "body-leak-test-fixture.ts")], {
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "inherit",
+    stdin: "ignore",
+  });
+  const { value } = await process.stdout.getReader().read();
+  url = new URL(new TextDecoder().decode(value));
+
+  await warmup();
+});
+afterAll(() => {
+  process?.kill();
+});
+
+async function getMemoryUsage(): Promise<number> {
+  return (await fetch(`${url.origin}/report`).then(res => res.json())) as number;
+}
+
+async function warmup() {
+  const batch = new Array(100);
+  for (let i = 0; i < 100; i++) {
+    for (let j = 0; j < 100; j++) {
+      // warmup the server with streaming requests, because is the most memory intensive
+      batch[j] = fetch(`${url.origin}/streaming`, {
+        method: "POST",
+        body: payload,
+      });
+    }
+    await Promise.all(batch);
+  }
+  // clean up memory before first test
+  await getMemoryUsage();
+}
+
+async function callBuffering() {
+  const result = await fetch(`${url.origin}/buffering`, {
+    method: "POST",
+    body: payload,
+  }).then(res => res.text());
+  expect(result).toBe("Ok");
+}
+async function callStreaming() {
+  const result = await fetch(`${url.origin}/streaming`, {
+    method: "POST",
+    body: payload,
+  }).then(res => res.text());
+  expect(result).toBe("Ok");
+}
+async function callIncompleteStreaming() {
+  const result = await fetch(`${url.origin}/incomplete-streaming`, {
+    method: "POST",
+    body: payload,
+  }).then(res => res.text());
+  expect(result).toBe("Ok");
+}
+async function callStreamingEcho() {
+  const result = await fetch(`${url.origin}/streaming-echo`, {
+    method: "POST",
+    body: payload,
+  }).then(res => res.text());
+  expect(result).toBe(payload);
+}
+async function callIgnore() {
+  const result = await fetch(url, {
+    method: "POST",
+    body: payload,
+  }).then(res => res.text());
+  expect(result).toBe("Ok");
+}
+
+async function calculateMemoryLeak(fn: () => Promise<void>) {
+  const start_memory = await getMemoryUsage();
+  const memory_examples: Array<number> = [];
+  let peak_memory = start_memory;
+  const batch = new Array(100);
+  for (let i = 0; i < 100; i++) {
+    for (let j = 0; j < 100; j++) {
+      batch[j] = fn();
+    }
+    await Promise.all(batch);
+    // garbage collect and check memory usage every 1000 requests
+    if (i > 0 && i % 10 === 0) {
+      const report = await getMemoryUsage();
+      if (report > peak_memory) {
+        peak_memory = report;
+      }
+      memory_examples.push(report);
+    }
+  }
+
+  // wait for the last memory usage to be stable
+  const end_memory = await getMemoryUsage();
+  if (end_memory > peak_memory) {
+    peak_memory = end_memory;
+  }
+  // use first example as a reference if is a memory leak this should keep increasing and not be stable
+  const consumption = end_memory - memory_examples[0];
+  // memory leak in MB
+  const leak = Math.floor(consumption > 0 ? consumption / 1024 / 1024 : 0);
+  return { leak, start_memory, peak_memory, end_memory, memory_examples };
+}
+
+for (const test_info of [
+  ["#10265 should not leak memory when ignoring the body", callIgnore, false],
+  ["should not leak memory when buffering the body", callBuffering, false],
+  ["should not leak memory when streaming the body", callStreaming, false],
+  ["should not leak memory when streaming the body incompletely", callIncompleteStreaming, true],
+  ["should not leak memory when streaming the body and echoing it back", callStreamingEcho, true],
+]) {
+  const [testName, fn, skip] = test_info as [string, () => Promise<void>, boolean];
+  it.todoIf(skip)(
+    testName,
+    async () => {
+      const report = await calculateMemoryLeak(fn);
+      // peak memory is too high
+      expect(report.peak_memory > report.start_memory * 2).toBe(false);
+      // acceptable memory leak
+      expect(report.leak).toBeLessThanOrEqual(ACCEPTABLE_MEMORY_LEAK);
+    },
+    30_000,
+  );
+}

test/js/bun/http/serve.test.ts

@@ -1484,3 +1484,32 @@ describe("should error with invalid options", async () => {
     }).toThrow("Expected lowMemoryMode to be a boolean");
   });
 });
+it("should resolve pending promise if requested ended with pending read", async () => {
+  let error: Error;
+  function shouldError(e: Error) {
+    error = e;
+  }
+  let is_done = false;
+  function shouldMarkDone(result: { done: boolean; value: any }) {
+    is_done = result.done;
+  }
+  await runTest(
+    {
+      fetch(req) {
+        // @ts-ignore
+        req.body?.getReader().read().catch(shouldError).then(shouldMarkDone);
+        return new Response("OK");
+      },
+    },
+    async server => {
+      const response = await fetch(server.url.origin, {
+        method: "POST",
+        body: "1".repeat(64 * 1024),
+      });
+      const text = await response.text();
+      expect(text).toContain("OK");
+      expect(is_done).toBe(true);
+      expect(error).toBeUndefined();
+    },
+  );
+});

test/js/node/http/fixtures/log-events.mjs

@@ -1,7 +1,7 @@
 import * as http from "node:http";
 
 const options = {
-  hostname: "www.google.com",
+  hostname: "www.example.com",
   port: 80,
   path: "/",
   method: "GET",

test/js/node/http/node-http.test.ts

@@ -1904,3 +1904,43 @@ it.skipIf(!process.env.TEST_INFO_STRIPE)("should be able to connect to stripe",
   err = await new Response(stderr).text();
   expect(err).toContain(`error: No such charge: '${charge_id}'\n`);
 });
+
+it("destroy should end download", async () => {
+  // just simulate some file that will take forever to download
+  const payload = Buffer.from("X".repeat(16 * 1024));
+
+  const server = Bun.serve({
+    port: 0,
+    async fetch(req) {
+      let running = true;
+      req.signal.onabort = () => (running = false);
+      return new Response(async function* () {
+        while (running) {
+          yield payload;
+          await Bun.sleep(10);
+        }
+      });
+    },
+  });
+
+  try {
+    let chunks = 0;
+
+    const { promise, resolve } = Promise.withResolvers();
+    const req = request(server.url, res => {
+      res.on("data", () => {
+        process.nextTick(resolve);
+        chunks++;
+      });
+    });
+    req.end();
+    // wait for the first chunk
+    await promise;
+    // should stop the download
+    req.destroy();
+    await Bun.sleep(200);
+    expect(chunks).toBeLessThanOrEqual(3);
+  } finally {
+    server.stop(true);
+  }
+});