Merge branch 'main' into dylan/fix-windows-watcher-watchlist

src/Watcher.zig

@@ -112,14 +112,29 @@ pub fn start(this: *Watcher) !void {
     this.thread = try std.Thread.spawn(.{}, threadMain, .{this});
 }
 
-pub fn deinit(this: *Watcher, close_descriptors: bool) void {
+const DeinitOpts = struct {
+    close_descriptors: bool,
+    join_thread: bool,
+};
+
+pub fn deinit(this: *Watcher, opts: DeinitOpts) void {
     if (this.watchloop_handle != null) {
-        this.mutex.lock();
-        defer this.mutex.unlock();
-        this.close_descriptors = close_descriptors;
-        this.running = false;
+        {
+            this.mutex.lock();
+            defer this.mutex.unlock();
+            this.close_descriptors = opts.close_descriptors;
+            this.running = false;
+
+            if (opts.join_thread) {
+                this.platform.shutdown();
+            }
+        }
+
+        if (opts.join_thread) {
+            this.thread.join();
+        }
     } else {
-        if (close_descriptors and this.running) {
+        if (opts.close_descriptors and this.running) {
             const fds = this.watchlist.items(.fd);
             for (fds) |fd| {
                 fd.close();
@@ -241,7 +256,9 @@ fn threadMain(this: *Watcher) !void {
                 this.onError(this.ctx, err);
             }
         },
-        .result => {},
+        .result => {
+            this.watchloop_handle = null;
+        },
     }
 
     // deinit and close descriptors if needed

src/bake/DevServer.zig

@@ -383,7 +383,7 @@ pub fn init(options: Options) bun.JSOOM!*DevServer {
     dev.bun_watcher = Watcher.init(DevServer, dev, fs, bun.default_allocator) catch |err|
         return global.throwError(err, "while initializing file watcher for development server");
 
-    errdefer dev.bun_watcher.deinit(false);
+    errdefer dev.bun_watcher.deinit(.{ .close_descriptors = false, .join_thread = false });
     dev.bun_watcher.start() catch |err|
         return global.throwError(err, "while initializing file watcher thread for development server");
 
@@ -604,7 +604,7 @@ pub fn deinit(dev: *DevServer) void {
         .memory_visualizer_timer = if (dev.memory_visualizer_timer.state == .ACTIVE)
             dev.vm.timer.remove(&dev.memory_visualizer_timer),
         .graph_safety_lock = dev.graph_safety_lock.lock(),
-        .bun_watcher = dev.bun_watcher.deinit(true),
+        .bun_watcher = dev.bun_watcher.deinit(.{ .close_descriptors = true, .join_thread = false }),
         .dump_dir = if (bun.FeatureFlags.bake_debugging_features) if (dev.dump_dir) |*dir| dir.close(),
         .log = dev.log.deinit(),
         .server_fetch_function_callback = dev.server_fetch_function_callback.deinit(),

src/bun.js/VirtualMachine.zig

@@ -852,6 +852,7 @@ extern fn Zig__GlobalObject__destructOnExit(*JSGlobalObject) void;
 
 pub fn globalExit(this: *VirtualMachine) noreturn {
     bun.assert(this.isShuttingDown());
+    this.bun_watcher.deinitAndJoin();
     // FIXME: we should be doing this, but we're not, but unfortunately doing it
     //        causes like 50+ tests to break
     // this.eventLoop().tick();

src/bun.js/hot_reloader.zig

@@ -11,6 +11,16 @@ pub const ImportWatcher = union(enum) {
         }
     }
 
+    pub fn deinitAndJoin(this: ImportWatcher) void {
+        switch (this) {
+            inline .hot, .watch => |w| w.deinit(.{
+                .close_descriptors = false,
+                .join_thread = true,
+            }),
+            .none => {},
+        }
+    }
+
     pub inline fn watchlist(this: ImportWatcher) Watcher.WatchList {
         return switch (this) {
             inline .hot, .watch => |w| w.watchlist,

src/bun.js/node/path_watcher.zig

@@ -667,7 +667,7 @@ pub const PathWatcherManager = struct {
             return;
         }
 
-        this.main_watcher.deinit(false);
+        this.main_watcher.deinit(.{ .close_descriptors = false, .join_thread = false });
 
         if (this.watcher_count > 0) {
             while (this.watchers.pop()) |watcher| {

src/watcher/INotifyWatcher.zig

@@ -225,6 +225,11 @@ pub fn stop(this: *INotifyWatcher) void {
     }
 }
 
+pub fn shutdown(this: *INotifyWatcher) void {
+    this.stop();
+    Futex.wake(&this.watch_count, std.math.maxInt(u32));
+}
+
 /// Repeatedly called by the main watcher until the watcher is terminated.
 pub fn watchLoopCycle(this: *bun.Watcher) bun.sys.Maybe(void) {
     defer Output.flush();

src/watcher/KEventWatcher.zig

@@ -21,6 +21,10 @@ pub fn stop(this: *KEventWatcher) void {
     }
 }
 
+pub fn shutdown(this: *KEventWatcher) void {
+    this.stop();
+}
+
 pub fn watchEventFromKEvent(kevent: KEvent) Watcher.Event {
     return .{
         .op = .{

src/watcher/WindowsWatcher.zig

@@ -197,6 +197,15 @@ pub fn stop(this: *WindowsWatcher) void {
     w.CloseHandle(this.iocp);
 }
 
+pub fn shutdown(this: *WindowsWatcher) void {
+    _ = w.kernel32.PostQueuedCompletionStatus(
+        this.iocp,
+        0,
+        0,
+        &this.watcher.overlapped,
+    );
+}
+
 pub fn watchLoopCycle(this: *bun.Watcher) bun.sys.Maybe(void) {
     const buf = &this.platform.buf;
     const base_idx = this.platform.base_idx;
@@ -205,7 +214,7 @@ pub fn watchLoopCycle(this: *bun.Watcher) bun.sys.Maybe(void) {
 
     // first wait has infinite timeout - we're waiting for the next event and don't want to spin
     var timeout = WindowsWatcher.Timeout.infinite;
-    while (true) {
+    while (this.running) {
         var iter = switch (this.platform.next(timeout)) {
             .err => |err| return .{ .err = err },
             .result => |iter| iter orelse break,
@@ -214,7 +223,11 @@ pub fn watchLoopCycle(this: *bun.Watcher) bun.sys.Maybe(void) {
         // NOTE: using a 1ms timeout would be ideal, but that actually makes the thread wait for at least 10ms more than it should
         // Instead we use a 0ms timeout, which may not do as much coalescing but is more responsive.
         timeout = WindowsWatcher.Timeout.none;
+
+        this.mutex.lock();
+        defer this.mutex.unlock();
         const item_paths = this.watchlist.items(.file_path);
+
         log("number of watched items: {d}", .{item_paths.len});
         while (iter.next()) |event| {
             const convert_res = bun.strings.copyUTF16IntoUTF8(buf[base_idx..], event.filename);
@@ -242,7 +255,7 @@ pub fn watchLoopCycle(this: *bun.Watcher) bun.sys.Maybe(void) {
                 // Check if we're about to exceed the watch_events array capacity
                 if (event_id >= this.watch_events.len) {
                     // Process current batch of events
-                    switch (processWatchEventBatch(this, event_id)) {
+                    switch (processWatchEventBatch(this, event_id, .dont_lock)) {
                         .err => |err| return .{ .err = err },
                         .result => {},
                     }
@@ -258,7 +271,7 @@ pub fn watchLoopCycle(this: *bun.Watcher) bun.sys.Maybe(void) {
 
     // Process any remaining events in the final batch
     if (event_id > 0) {
-        switch (processWatchEventBatch(this, event_id)) {
+        switch (processWatchEventBatch(this, event_id, .lock)) {
             .err => |err| return .{ .err = err },
             .result => {},
         }
@@ -267,7 +280,7 @@ pub fn watchLoopCycle(this: *bun.Watcher) bun.sys.Maybe(void) {
     return .success;
 }
 
-fn processWatchEventBatch(this: *bun.Watcher, event_count: usize) bun.sys.Maybe(void) {
+fn processWatchEventBatch(this: *bun.Watcher, event_count: usize, lock: enum { lock, dont_lock }) bun.sys.Maybe(void) {
     if (event_count == 0) {
         return .success;
     }
@@ -293,8 +306,12 @@ fn processWatchEventBatch(this: *bun.Watcher, event_count: usize) bun.sys.Maybe(
 
     log("calling onFileUpdate (all_events.len = {d})", .{all_events.len});
 
-    this.writeTraceEvents(all_events, this.changed_filepaths[0 .. last_event_index + 1]);
-    this.onFileUpdate(this.ctx, all_events, this.changed_filepaths[0 .. last_event_index + 1], this.watchlist);
+    {
+        if (lock == .lock) this.mutex.lock();
+        defer if (lock == .lock) this.mutex.unlock();
+        this.writeTraceEvents(all_events, this.changed_filepaths[0 .. last_event_index + 1]);
+        this.onFileUpdate(this.ctx, all_events, this.changed_filepaths[0 .. last_event_index + 1], this.watchlist);
+    }
 
     return .success;
 }

test/cli/hot/hot.test.ts

@@ -1,7 +1,7 @@
 import { spawn } from "bun";
 import { beforeEach, expect, it } from "bun:test";
 import { copyFileSync, cpSync, readFileSync, renameSync, rmSync, unlinkSync, writeFileSync } from "fs";
-import { bunEnv, bunExe, isDebug, tmpdirSync, waitForFileToExist } from "harness";
+import { bunEnv, bunExe, isDebug, tempDir, tmpdirSync, waitForFileToExist } from "harness";
 import { join } from "path";
 
 const timeout = isDebug ? Infinity : 10_000;
@@ -32,6 +32,60 @@ it("preload not found should exit with code 1 and not time out", async () => {
   expect(await new Response(runner.stderr).text()).toContain("preload not found");
 });
 
+it("does not crash under stress", () => {
+  // 1 second test, crashes about 50% of the time
+  using testDir = tempDir("watcher-stress-test", {
+    "index.js": `
+const TEST_DIR = "./crash-test";
+import { writeFile, mkdir } from "fs/promises";
+
+setTimeout(() => {
+  process.exit(0);
+}, 1000);
+
+// Create a deeply nested module structure
+for (let i = 0; i < 100; i++) {
+  for (let j = 0; j < 10; j++) {
+    const dir = \`\${TEST_DIR}/dir\${i}\`;
+    await mkdir(dir, { recursive: true });
+    await writeFile(\`\${dir}/module\${j}.ts\`, \`export default \${i * 10 + j};\`);
+  }
+}
+
+// Spawn multiple async tasks to stress the watcher
+for (let worker = 0; worker < 10; worker++) {
+  (async () => {
+    while (true) {
+      // Dynamically import (adds to watchlist via ModuleLoader.zig)
+      const i = Math.floor(Math.random() * 100);
+      const j = Math.floor(Math.random() * 10);
+      try {
+        await import(\`\${TEST_DIR}/dir\${i}/module\${j}.ts\`);
+      } catch {}
+
+      await writeFile(\`\${TEST_DIR}/dir\${i}/module\${j}.ts\`, \`export default \${Date.now()};\`);
+    }
+  })();
+}
+`,
+  });
+
+  const { stdout, stderr, exitCode } = Bun.spawnSync({
+    cmd: [bunExe(), "--hot", "index.js"],
+    cwd: testDir,
+    stdout: "pipe",
+    stderr: "pipe",
+    stdin: "ignore",
+    env: bunEnv,
+  });
+  const out = stdout.toString();
+  const err = stderr.toString().replace(/DEBUG: Reloading...\n/g, "");
+  console.log({ out, err });
+  expect(out).toBeEmpty();
+  expect(err).toBeEmpty();
+  expect(exitCode).toBe(0);
+});
+
 it(
   "should hot reload when file is overwritten",
   async () => {