Bunch of streams fixes (#4251)

* Update WebKit

* Don't do async hooks things when async hooks are not enabled

* Smarter scheduling of event loop tasks with the http server

* less exciting approach

* Bump WebKit

* Another approach

* Fix body-stream tests

* Fixes #1886

* Fix UAF in fetch body streaming

* Missing from commit

* Fix leak

* Fix the other leak

* Fix test

* Fix crash

* missing duperef

* Make this code clearer

* Ignore empty chunks

* Fixes #3969

* Delete flaky test

* Update bun-linux-build.yml

* Fix memory issue

* fix result body, and .done status before the last callback, dont touch headers after sent once

* refactor HTTPClientResult

* less flasky corrupted test

* oops

* fix mutex invalid state

* fix onProgressUpdate deinit/unlock

* fix onProgressUpdate deinit/unlock

* oops

* remove verbose

* fix posible null use

* avoid http null

* metadata can still be used onReject after toResponse

* dont leak task.http

* fix flask tests

* less flask close tests

---------

Co-authored-by: Jarred Sumner <709451+Jarred-Sumner@users.noreply.github.com>
Co-authored-by: cirospaciari <ciro.spaciari@gmail.com>

5 files changed, 199 insertions, 247 deletions

diff --git a/test/js/bun/http/fetch-file-upload.test.ts b/test/js/bun/http/fetch-file-upload.test.ts
index b070fbd6e..197470b9d 100644
--- a/test/js/bun/http/fetch-file-upload.test.ts
+++ b/test/js/bun/http/fetch-file-upload.test.ts
@@ -34,6 +34,104 @@ test("uploads roundtrip", async () => {
   server.stop(true);
 });
 
+// https://github.com/oven-sh/bun/issues/3969
+test("formData uploads roundtrip, with a call to .body", async () => {
+  const file = Bun.file(import.meta.dir + "/fetch.js.txt");
+  const body = new FormData();
+  body.append("file", file, "fetch.js.txt");
+
+  const server = Bun.serve({
+    port: 0,
+    development: false,
+    async fetch(req) {
+      req.body;
+
+      return new Response(await req.formData());
+    },
+  });
+
+  // @ts-ignore
+  const reqBody = new Request(`http://${server.hostname}:${server.port}`, {
+    body,
+    method: "POST",
+  });
+  const res = await fetch(reqBody);
+  expect(res.status).toBe(200);
+
+  // but it does for Response
+  expect(res.headers.get("Content-Type")).toStartWith("multipart/form-data; boundary=");
+  res.body;
+  const resData = await res.formData();
+  expect(await (resData.get("file") as Blob).arrayBuffer()).toEqual(await file.arrayBuffer());
+
+  server.stop(true);
+});
+
+test("req.formData throws error when stream is in use", async () => {
+  const file = Bun.file(import.meta.dir + "/fetch.js.txt");
+  const body = new FormData();
+  body.append("file", file, "fetch.js.txt");
+  var pass = false;
+  const server = Bun.serve({
+    port: 0,
+    development: false,
+    error(fail) {
+      pass = true;
+      if (fail.toString().includes("is already used")) {
+        return new Response("pass");
+      }
+      return new Response("fail");
+    },
+    async fetch(req) {
+      var reader = req.body?.getReader();
+      await reader?.read();
+      await req.formData();
+      throw new Error("should not reach here");
+    },
+  });
+
+  // @ts-ignore
+  const reqBody = new Request(`http://${server.hostname}:${server.port}`, {
+    body,
+    method: "POST",
+  });
+  const res = await fetch(reqBody);
+  expect(res.status).toBe(200);
+
+  // but it does for Response
+  expect(await res.text()).toBe("pass");
+  server.stop(true);
+});
+
+test("formData uploads roundtrip, without a call to .body", async () => {
+  const file = Bun.file(import.meta.dir + "/fetch.js.txt");
+  const body = new FormData();
+  body.append("file", file, "fetch.js.txt");
+
+  const server = Bun.serve({
+    port: 0,
+    development: false,
+    async fetch(req) {
+      return new Response(await req.formData());
+    },
+  });
+
+  // @ts-ignore
+  const reqBody = new Request(`http://${server.hostname}:${server.port}`, {
+    body,
+    method: "POST",
+  });
+  const res = await fetch(reqBody);
+  expect(res.status).toBe(200);
+
+  // but it does for Response
+  expect(res.headers.get("Content-Type")).toStartWith("multipart/form-data; boundary=");
+  const resData = await res.formData();
+  expect(await (resData.get("file") as Blob).arrayBuffer()).toEqual(await file.arrayBuffer());
+
+  server.stop(true);
+});
+
 test("uploads roundtrip with sendfile()", async () => {
   var hugeTxt = "huge".repeat(1024 * 1024 * 32);
   const path = join(tmpdir(), "huge.txt");
diff --git a/test/js/bun/http/serve.test.ts b/test/js/bun/http/serve.test.ts
index bba35c085..4c55e779a 100644
--- a/test/js/bun/http/serve.test.ts
+++ b/test/js/bun/http/serve.test.ts
@@ -213,63 +213,97 @@ it("request.url should be based on the Host header", async () => {
 
 describe("streaming", () => {
   describe("error handler", () => {
-    it("throw on pull reports an error and close the connection", async () => {
-      var pass = false;
+    it("throw on pull renders headers, does not call error handler", async () => {
+      var pass = true;
       await runTest(
         {
           error(e) {
-            pass = true;
-            return new Response("PASS", { status: 555 });
+            pass = false;
+            return new Response("FAIL!", { status: 555 });
           },
           fetch(req) {
             return new Response(
               new ReadableStream({
                 pull(controller) {
-                  throw new Error("FAIL");
+                  throw new Error("Not a real error");
                 },
+                cancel(reason) {},
               }),
+              {
+                status: 402,
+                headers: {
+                  "I-AM": "A-TEAPOT",
+                },
+              },
             );
           },
         },
         async server => {
           const response = await fetch(`http://${server.hostname}:${server.port}`);
-          if (response.status > 0) {
-            expect(response.status).toBe(555);
-            expect(await response.text()).toBe("PASS");
-          }
+          expect(response.status).toBe(402);
+          expect(response.headers.get("I-AM")).toBe("A-TEAPOT");
+          expect(await response.text()).toBe("");
           expect(pass).toBe(true);
         },
       );
     });
 
-    it("throw on pull after writing should not call the error handler", async () => {
-      var pass = true;
-      await runTest(
-        {
-          error(e) {
-            pass = false;
-            return new Response("FAIL", { status: 555 });
-          },
-          fetch(req) {
-            return new Response(
-              new ReadableStream({
+    describe("throw on pull after writing should not call the error handler", () => {
+      async function execute(options: ResponseInit) {
+        var pass = true;
+        await runTest(
+          {
+            error(e) {
+              pass = false;
+              return new Response("FAIL", { status: 555 });
+            },
+            fetch(req) {
+              const stream = new ReadableStream({
                 async pull(controller) {
                   controller.enqueue("PASS");
                   controller.close();
-                  throw new Error("error");
+                  throw new Error("FAIL");
                 },
-              }),
-            );
+              });
+              return new Response(stream, options);
+            },
           },
-        },
-        async server => {
-          const response = await fetch(`http://${server.hostname}:${server.port}`);
-          // connection terminated
-          expect(response.status).toBe(200);
-          expect(await response.text()).toBe("PASS");
-          expect(pass).toBe(true);
-        },
-      );
+          async server => {
+            const response = await fetch(`http://${server.hostname}:${server.port}`);
+            // connection terminated
+            expect(await response.text()).toBe("");
+            expect(response.status).toBe(options.status ?? 200);
+            expect(pass).toBe(true);
+          },
+        );
+      }
+
+      it("with headers", async () => {
+        await execute({
+          headers: {
+            "X-A": "123",
+          },
+        });
+      });
+
+      it("with headers and status", async () => {
+        await execute({
+          status: 204,
+          headers: {
+            "X-A": "123",
+          },
+        });
+      });
+
+      it("with status", async () => {
+        await execute({
+          status: 204,
+        });
+      });
+
+      it("with empty object", async () => {
+        await execute({});
+      });
     });
   });
 
@@ -1004,19 +1038,26 @@ it("request body and signal life cycle", async () => {
     };
 
     const server = Bun.serve({
+      port: 0,
       async fetch(req) {
-        await queueMicrotask(() => Bun.gc(true));
         return new Response(await renderToReadableStream(app_jsx), headers);
       },
     });
 
     try {
       const requests = [];
-      for (let i = 0; i < 1000; i++) {
-        requests.push(fetch(`http://${server.hostname}:${server.port}`));
+      for (let j = 0; j < 10; j++) {
+        for (let i = 0; i < 250; i++) {
+          requests.push(fetch(`http://${server.hostname}:${server.port}`));
+        }
+
+        await Promise.all(requests);
+        requests.length = 0;
+        Bun.gc(true);
       }
-      await Promise.all(requests);
-    } catch {}
+    } catch (e) {
+      console.error(e);
+    }
     await Bun.sleep(10);
     expect(true).toBe(true);
     server.stop(true);
diff --git a/test/js/bun/stream/direct-readable-stream.test.tsx b/test/js/bun/stream/direct-readable-stream.test.tsx
index c06840947..1f090671b 100644
--- a/test/js/bun/stream/direct-readable-stream.test.tsx
+++ b/test/js/bun/stream/direct-readable-stream.test.tsx
@@ -229,12 +229,17 @@ describe("ReactDOM", () => {
               server = serve({
                 port: 0,
                 async fetch(req) {
-                  return new Response(await renderToReadableStream(reactElement));
+                  return new Response(await renderToReadableStream(reactElement), {
+                    headers: {
+                      "X-React": "1",
+                    },
+                  });
                 },
               });
               const response = await fetch("http://localhost:" + server.port + "/");
               const result = await response.text();
               expect(result.replaceAll("<!-- -->", "")).toBe(inputString);
+              expect(response.headers.get("X-React")).toBe("1");
             } finally {
               server?.stop(true);
             }
diff --git a/test/js/third_party/napi_create_external/napi-create-external.test.ts b/test/js/third_party/napi_create_external/napi-create-external.test.ts
deleted file mode 100644
index 47025e100..000000000
--- a/test/js/third_party/napi_create_external/napi-create-external.test.ts
+++ /dev/null
@@ -1,195 +0,0 @@
-// @ts-nocheck
-import { test, it, describe, expect } from "bun:test";
-import { withoutAggressiveGC } from "harness";
-import * as _ from "lodash";
-
-function rebase(str, inBase, outBase) {
-  const mapBase = (b: number) => (b === 2 ? 32 : b === 16 ? 8 : null);
-  const stride = mapBase(inBase);
-  const pad = mapBase(outBase);
-  if (!stride) throw new Error(`Bad inBase ${inBase}`);
-  if (!pad) throw new Error(`Bad outBase ${outBase}`);
-  if (str.length % stride) throw new Error(`Bad string length ${str.length}`);
-  const out = [];
-  for (let i = 0; i < str.length; i += stride)
-    out.push(
-      parseInt(str.slice(i, i + stride), inBase)
-        .toString(outBase)
-        .padStart(pad, "0"),
-    );
-  return out.join("");
-}
-
-function expectDeepEqual(a, b) {
-  expect(a).toEqual(b);
-}
-class HashMaker {
-  constructor(length) {
-    this.length = length;
-    this._dist = {};
-  }
-  length: number;
-  _dist: any;
-
-  binToHex(binHash) {
-    if (binHash.length !== this.length) throw new Error(`Hash length mismatch ${this.length} != ${binHash.length}`);
-    return rebase(binHash, 2, 16);
-  }
-
-  makeBits() {
-    const bits = [];
-    for (let i = 0; i < this.length; i++) bits.push(i);
-    return _.shuffle(bits);
-  }
-
-  makeRandom() {
-    const bits = [];
-    for (let i = 0; i < this.length; i++) bits.push(Math.random() < 0.5 ? 1 : 0);
-    return bits;
-  }
-
-  get keySet() {
-    return (this._set = this._set || new Set(this.data));
-  }
-
-  randomKey() {
-    while (true) {
-      const hash = this.binToHex(this.makeRandom().join(""));
-      if (!this.keySet.has(hash)) return hash;
-    }
-  }
-
-  get data() {
-    return (this._data =
-      this._data ||
-      (() => {
-        const bits = this.makeBits();
-        const base = this.makeRandom();
-        const data = [];
-        for (let stride = 0; bits.length; stride++) {
-          const flip = bits.splice(0, stride);
-          for (const bit of flip) base[bit] = 1 - base[bit];
-          data.push(this.binToHex(base.join("")));
-        }
-        return data;
-      })());
-  }
-
-  get random() {
-    const d = this.data;
-    return d[Math.floor(Math.random() * d.length)];
-  }
-
-  distance(a, b) {
-    const bitCount = n => {
-      n = n - ((n >> 1) & 0x55555555);
-      n = (n & 0x33333333) + ((n >> 2) & 0x33333333);
-      return (((n + (n >> 4)) & 0xf0f0f0f) * 0x1010101) >> 24;
-    };
-
-    if (a === b) return 0;
-    if (a > b) return this.distance(b, a);
-    const hash = a + "-" + b;
-    return (this._dist[hash] =
-      this._dist[hash] ||
-      (() => {
-        let dist = 0;
-        for (let i = 0; i < a.length; i += 8) {
-          const va = parseInt(a.slice(i, i + 8), 16);
-          const vb = parseInt(b.slice(i, i + 8), 16);
-          dist += bitCount(va ^ vb);
-        }
-        return dist;
-      })());
-  }
-
-  query(baseKey, maxDist) {
-    const out = [];
-    for (const key of this.data) {
-      const distance = this.distance(key, baseKey);
-      if (distance <= maxDist) out.push({ key, distance });
-    }
-    return out.sort((a, b) => a.distance - b.distance);
-  }
-}
-
-const treeClass = require("bktree-fast/native");
-
-withoutAggressiveGC(() => {
-  // this test is too slow
-  for (let keyLen = 64; keyLen <= 64; keyLen += 64) {
-    // for (let keyLen = 64; keyLen <= 512; keyLen += 64) {
-    const hm = new HashMaker(keyLen);
-    describe(`Key length: ${keyLen}`, () => {
-      it("should compute distance", () => {
-        const tree = new treeClass(keyLen);
-        for (const a of hm.data) for (const b of hm.data) expect(tree.distance(a, b)).toBe(hm.distance(a, b));
-      });
-
-      it("should know which keys it has", () => {
-        const tree = new treeClass(keyLen).add(hm.data);
-        expectDeepEqual(
-          hm.data.map(hash => tree.has(hash)),
-          hm.data.map(() => true),
-        );
-        // Not interested in the hash
-        for (const hash of hm.data) expect(tree.has(hm.randomKey())).toBe(false);
-      });
-
-      it("should know the tree size", () => {
-        const tree = new treeClass(keyLen, { foo: 1 });
-        expect(tree.size).toBe(0);
-        tree.add(hm.data);
-        expect(tree.size).toBe(hm.data.length);
-        tree.add(hm.data);
-        expect(tree.size).toBe(hm.data.length);
-      });
-
-      it("should walk the tree", () => {
-        const tree = new treeClass(keyLen).add(hm.data);
-        const got = [];
-        tree.walk((hash, depth) => got.push(hash));
-        expectDeepEqual(got.sort(), hm.data.slice(0).sort());
-      });
-
-      it("should query", () => {
-        ((treeClass, expectDeepEqual) => {
-          const tree = new treeClass(keyLen).add(hm.data);
-
-          for (let dist = 0; dist <= hm.length; dist++) {
-            for (const baseKey of [hm.random, hm.data[0]]) {
-              const baseKey = hm.random;
-              const got = [];
-              tree.query(baseKey, dist, (key, distance) => got.push({ key, distance }));
-              const want = hm.query(baseKey, dist);
-              expectDeepEqual(
-                got.sort((a, b) => a.distance - b.distance),
-                want,
-              );
-              expectDeepEqual(tree.find(baseKey, dist), want);
-            }
-          }
-        })(treeClass, expectDeepEqual);
-      });
-    });
-  }
-
-  describe("Misc functions", () => {
-    it("should pad keys", () => {
-      const tree = new treeClass(64);
-      expect(tree.padKey("1")).toBe("0000000000000001");
-      tree.add(["1", "2", "3"]);
-
-      const got = [];
-      tree.query("2", 3, (hash, distance) => got.push({ hash, distance }));
-      const res = got.sort((a, b) => a.distance - b.distance);
-      const want = [
-        { hash: "0000000000000002", distance: 0 },
-        { hash: "0000000000000003", distance: 1 },
-        { hash: "0000000000000001", distance: 2 },
-      ];
-
-      expectDeepEqual(res, want);
-    });
-  });
-});
diff --git a/test/js/web/fetch/fetch.stream.test.ts b/test/js/web/fetch/fetch.stream.test.ts
index efef6a161..dc082fd2f 100644
--- a/test/js/web/fetch/fetch.stream.test.ts
+++ b/test/js/web/fetch/fetch.stream.test.ts
@@ -412,8 +412,7 @@ describe("fetch() with streaming", () => {
     const fixture = matrix[i];
     for (let j = 0; j < matrix.length; j++) {
       const fixtureb = matrix[j];
-      const test = fixture.name == "empty" && fixtureb.name == "empty" ? it.todo : it;
-      test(`can handle fixture ${fixture.name} x ${fixtureb.name}`, async () => {
+      it(`can handle fixture ${fixture.name} x ${fixtureb.name}`, async () => {
         let server: Server | null = null;
         try {
           //@ts-ignore
@@ -917,12 +916,11 @@ describe("fetch() with streaming", () => {
             drain(socket) {},
           },
         });
-
-        const res = await fetch(`http://${server.hostname}:${server.port}`, {
-          signal: AbortSignal.timeout(1000),
-        });
-        gcTick(false);
         try {
+          const res = await fetch(`http://${server.hostname}:${server.port}`, {
+            signal: AbortSignal.timeout(1000),
+          });
+          gcTick(false);
           const reader = res.body?.getReader();
 
           let buffer = Buffer.alloc(0);
@@ -995,6 +993,7 @@ describe("fetch() with streaming", () => {
                 compressed[0] = 0; // corrupt data
                 cork = false;
                 for (var i = 0; i < 5; i++) {
+                  compressed[size * i] = 0; // corrupt data even more
                   await write(compressed.slice(size * i, size * (i + 1)));
                 }
                 socket.flush();
@@ -1003,10 +1002,10 @@ describe("fetch() with streaming", () => {
             },
           });
 
-          const res = await fetch(`http://${server.hostname}:${server.port}`, {});
-          gcTick(false);
-
           try {
+            const res = await fetch(`http://${server.hostname}:${server.port}`, {});
+            gcTick(false);
+
             const reader = res.body?.getReader();
 
             let buffer = Buffer.alloc(0);
@@ -1079,23 +1078,27 @@ describe("fetch() with streaming", () => {
               // 10 extra missing bytes that we will never sent in this case we will wait to close
               await write("Content-Length: " + compressed.byteLength + 10 + "\r\n");
               await write("\r\n");
+
+              resolveSocket(socket);
+
               const size = compressed.byteLength / 5;
               for (var i = 0; i < 5; i++) {
                 cork = false;
                 await write(compressed.slice(size * i, size * (i + 1)));
               }
               socket.flush();
-              resolveSocket(socket);
             },
             drain(socket) {},
           },
         });
 
-        const res = await fetch(`http://${server.hostname}:${server.port}`, {});
-        gcTick(false);
+        let socket: Socket | null = null;
 
-        let socket: Socket | null = await promise;
         try {
+          const res = await fetch(`http://${server.hostname}:${server.port}`, {});
+          socket = await promise;
+          gcTick(false);
+
           const reader = res.body?.getReader();
 
           let buffer = Buffer.alloc(0);