Fix worker event loop ref/unref + leak (#4114)

* make more tests pass

* worker changes

* fix some bugs

* remove this

* progress

* uh

* okay

* remove console log

* a

* comment assert for later

* mergable state

* remove test

* remove test

11 files changed, 78 insertions, 24 deletions

diff --git a/test/bun.lockb b/test/bun.lockb
index b717f5d1a..a4e2b2fdb 100755
--- a/test/bun.lockb
+++ b/test/bun.lockb
diff --git a/test/js/bun/util/main-worker-file.js b/test/js/bun/util/main-worker-file.js
index 80f83ba30..2d5c3aedd 100644
--- a/test/js/bun/util/main-worker-file.js
+++ b/test/js/bun/util/main-worker-file.js
@@ -11,5 +11,6 @@ if (isMainThread) {
   });
 
   await promise;
+
   worker.terminate();
 }
diff --git a/test/js/third_party/esbuild/esbuild-test.js b/test/js/third_party/esbuild/esbuild-test.js
index 09152246b..e0ce6f497 100644
--- a/test/js/third_party/esbuild/esbuild-test.js
+++ b/test/js/third_party/esbuild/esbuild-test.js
@@ -82,5 +82,3 @@ import { build, buildSync, transform, transformSync } from "esbuild";
     throw new Error("Test failed.");
   }
 }
-
-process.exit(0);
diff --git a/test/js/web/many-messages-event-loop.js b/test/js/web/many-messages-event-loop.js
deleted file mode 100644
index 2eaba2568..000000000
--- a/test/js/web/many-messages-event-loop.js
+++ /dev/null
@@ -1,11 +0,0 @@
-const worker = new Worker(new URL("worker-fixture-many-messages.js", import.meta.url).href);
-
-worker.postMessage("initial message");
-worker.addEventListener("message", ({ data }) => {
-  if (data.done) {
-    console.log("done");
-    worker.terminate();
-  } else {
-    worker.postMessage({ i: data.i + 1 });
-  }
-});
diff --git a/test/js/web/many-messages-event-loop.mjs b/test/js/web/many-messages-event-loop.mjs
new file mode 100644
index 000000000..deae5f791
--- /dev/null
+++ b/test/js/web/many-messages-event-loop.mjs
@@ -0,0 +1,11 @@
+const worker = new Worker(new URL(process.argv[2], import.meta.url));
+
+worker.postMessage("initial message");
+worker.addEventListener("message", function fn({ data }) {
+  if (data.done) {
+    console.log("done");
+    worker.removeEventListener("message", fn);
+  } else {
+    worker.postMessage({ i: data.i + 1 });
+  }
+});
diff --git a/test/js/web/worker-fixture-hang.js b/test/js/web/worker-fixture-hang.js
new file mode 100644
index 000000000..4db3f65f8
--- /dev/null
+++ b/test/js/web/worker-fixture-hang.js
@@ -0,0 +1,3 @@
+setTimeout(() => {
+  process.exit(2);
+}, 1000000);
diff --git a/test/js/web/worker-fixture-many-messages.js b/test/js/web/worker-fixture-many-messages.js
index 7a8f1d910..5bc4442a9 100644
--- a/test/js/web/worker-fixture-many-messages.js
+++ b/test/js/web/worker-fixture-many-messages.js
@@ -1,11 +1,11 @@
-addEventListener("message", e => {
-  const data = e.data;
+addEventListener("message", function fn({ data }) {
   // console.log("worker", data);
 
   if (data === "initial message") {
     postMessage({ i: 0 });
   } else if (data.i > 50) {
     postMessage({ done: true });
+    removeEventListener("message", fn);
   } else {
     postMessage({ i: data.i + 1 });
   }
diff --git a/test/js/web/worker-fixture-many-messages2.js b/test/js/web/worker-fixture-many-messages2.js
new file mode 100644
index 000000000..3c371b156
--- /dev/null
+++ b/test/js/web/worker-fixture-many-messages2.js
@@ -0,0 +1,12 @@
+onmessage = ({ data }) => {
+  // console.log("worker", data);
+
+  if (data === "initial message") {
+    postMessage({ i: 0 });
+  } else if (data.i > 50) {
+    postMessage({ done: true });
+    onmessage = null;
+  } else {
+    postMessage({ i: data.i + 1 });
+  }
+};
diff --git a/test/js/web/worker-fixture-process-exit.js b/test/js/web/worker-fixture-process-exit.js
index 0d93217d9..05746973b 100644
--- a/test/js/web/worker-fixture-process-exit.js
+++ b/test/js/web/worker-fixture-process-exit.js
@@ -1,3 +1,3 @@
 setTimeout(() => {
   process.exit(2);
-}, 100);
+}, 10);
diff --git a/test/js/web/worker-fixture-while-true.js b/test/js/web/worker-fixture-while-true.js
new file mode 100644
index 000000000..ca6233960
--- /dev/null
+++ b/test/js/web/worker-fixture-while-true.js
@@ -0,0 +1,4 @@
+let i = 0;
+while (true) {
+  postMessage({ i: i++ });
+}
diff --git a/test/js/web/worker.test.ts b/test/js/web/worker.test.ts
index 34d5f6f06..e1ab80487 100644
--- a/test/js/web/worker.test.ts
+++ b/test/js/web/worker.test.ts
@@ -98,9 +98,37 @@ test("sending 50 messages should just work", done => {
   });
 });
 
-test("worker by default will not close the event loop", done => {
+test("worker with event listeners doesnt close event loop", done => {
   const x = Bun.spawn({
-    cmd: [bunExe(), path.join(import.meta.dir, "many-messages-event-loop.js")],
+    cmd: [bunExe(), path.join(import.meta.dir, "many-messages-event-loop.mjs"), "worker-fixture-many-messages.js"],
+    env: bunEnv,
+    stdio: ["inherit", "pipe", "inherit"],
+  });
+
+  const timer = setTimeout(() => {
+    x.kill();
+    done(new Error("timeout"));
+  }, 1000);
+
+  x.exited.then(async code => {
+    clearTimeout(timer);
+    if (code !== 0) {
+      done(new Error("exited with non-zero code"));
+    } else {
+      const text = await new Response(x.stdout).text();
+      if (!text.includes("done")) {
+        console.log({ text });
+        done(new Error("event loop killed early"));
+      } else {
+        done();
+      }
+    }
+  });
+});
+
+test("worker with event listeners doesnt close event loop 2", done => {
+  const x = Bun.spawn({
+    cmd: [bunExe(), path.join(import.meta.dir, "many-messages-event-loop.mjs"), "worker-fixture-many-messages2.js"],
     env: bunEnv,
     stdio: ["inherit", "pipe", "inherit"],
   });
@@ -146,7 +174,6 @@ test("worker_threads with process.exit", done => {
   });
   worker.on("exit", event => {
     try {
-      console.log({ event });
       expect(event).toBe(2);
     } catch (e) {
       done(e);
@@ -155,19 +182,28 @@ test("worker_threads with process.exit", done => {
   });
 });
 
-test.skip("worker_threads with process.exit and terminate", async () => {
-  const worker = new wt.Worker(new URL("worker-fixture-process-exit.js", import.meta.url).href, {
+test("worker_threads terminate", async () => {
+  const worker = new wt.Worker(new URL("worker-fixture-hang.js", import.meta.url).href, {
     smol: true,
   });
   const code = await worker.terminate();
-  expect(code).toBe(2);
+  expect(code).toBe(0);
 });
 
-test.skip("worker_threads with process.exit (delay) and terminate", async () => {
+test("worker_threads with process.exit (delay) and terminate", async () => {
   const worker2 = new wt.Worker(new URL("worker-fixture-process-exit.js", import.meta.url).href, {
     smol: true,
   });
-  await Bun.sleep(100);
+  await Bun.sleep(200);
   const code2 = await worker2.terminate();
   expect(code2).toBe(2);
 });
+
+test.skip("terminating forcefully properly interrupts", async () => {
+  const worker2 = new wt.Worker(new URL("worker-fixture-while-true.js", import.meta.url).href, {});
+  await new Promise<void>(done => {
+    worker2.on("message", () => done());
+  });
+  const code2 = await worker2.terminate();
+  expect(code2).toBe(0);
+});