dev server: guard the watcher hand-off with a mutex so no batch is lost

src/runtime/bake/dev_server/watcher_atomics.rs

@@ -129,21 +129,38 @@ impl WatcherAtomics {
         // SAFETY: caller contract — `this` is live; every `(*this)` / `(*ev)`
         // below is a field access on the heap `WatcherAtomics` allocation.
         unsafe {
-            let mut available = [true; 3];
-            if let Some(i) = (*this).current_event {
-                available[i as usize] = false;
-            }
-            if let Some(i) = (*this).pending_event {
-                available[i as usize] = false;
+            // Take the queued event back, so that this batch joins it. A second
+            // queued event would replace it, and the dev server would never see
+            // the files of the first one.
+            let mut withdrawn: Option<u8> = None;
+            if let Some(pending) = (*this).pending_event.take() {
+                if (*this)
+                    .next_event
+                    .compare_exchange(
+                        pending,
+                        NextEvent::WAITING.0,
+                        Ordering::AcqRel,
+                        Ordering::Acquire,
+                    )
+                    .is_ok()
+                {
+                    withdrawn = Some(pending);
+                } else {
+                    // The dev server took it, which means it is done with `current_event`.
+                    (*this).current_event = Some(pending);
+                }
             }
 
-            let index = 'find: {
-                for (i, &is_available) in available.iter().enumerate() {
-                    if is_available {
-                        break 'find i;
+            let index = match withdrawn {
+                Some(pending) => pending as usize,
+                None => 'find: {
+                    for i in 0..(*this).events.len() {
+                        if (*this).current_event != Some(i as u8) {
+                            break 'find i;
+                        }
                     }
+                    unreachable!()
                 }
-                unreachable!()
             };
             let ev: *mut HotReloadEvent = &raw mut (*this).events[index];
 
@@ -261,23 +278,17 @@ impl WatcherAtomics {
                 }
 
                 NextEvent::WAITING => {
-                    if (*this).pending_event.is_some() {
-                        // `pending_event` is running, which means we're done with `current_event`.
-                        (*this).current_event = (*this).pending_event;
-                    } // else, no pending event yet, but not done with `current_event`.
+                    // The dev server runs an event. `watcher_acquire_event` left nothing queued.
                     (*this).pending_event = Some(ev_index);
                 }
 
                 _ => {
-                    // This is an index into the `events` array.
-                    let old_index: u8 = old_next.0;
+                    // Only this thread queues an event, and `watcher_acquire_event` withdraws it.
                     debug_assert!(
-                        (*this).pending_event == Some(old_index),
-                        "watcher_release_and_submit_event: expected `pending_event` to be {}; got {:?}",
-                        old_index,
-                        (*this).pending_event,
+                        false,
+                        "watcher_release_and_submit_event: event {} was still queued",
+                        old_next.0,
                     );
-                    // The old pending event hadn't been run yet, so we can replace it with `ev`.
                     (*this).pending_event = Some(ev_index);
                 }
             }

test/bake/dev/hot.test.ts

@@ -1,6 +1,6 @@
 // Hot tests ensure that the `import.meta.hot` interface is functional
 import { expect } from "bun:test";
-import { renameSync, unlinkSync, writeFileSync } from "node:fs";
+import { existsSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
 import { devTest, emptyHtmlFile } from "../bake-harness";
 
 devTest("import.meta.hot.accept basic", {
@@ -691,3 +691,78 @@ devTest("dev.write resolves only after the new module body has run", {
     expect(await c.js`globalThis.marker`).toBe("updated");
   },
 });
+
+devTest("every save that arrives while a bundle runs reaches the next bundle", {
+  files: {
+    "bunfig.toml": `
+      [serve.static]
+      plugins = ["./hold-plugin.ts"]
+    `,
+    // Keeps a bundle that loads held.ts open for as long as hold/on exists.
+    "hold-plugin.ts": `
+      import { existsSync, readFileSync, writeFileSync } from "node:fs";
+      import { join } from "node:path";
+
+      const on = join(import.meta.dir, "hold", "on");
+      export default {
+        name: "hold",
+        setup(build) {
+          build.onLoad({ filter: /held\\.ts$/ }, async args => {
+            if (existsSync(on)) {
+              writeFileSync(join(import.meta.dir, "hold", "entered"), "");
+              while (existsSync(on)) await Bun.sleep(5);
+            }
+            return { contents: readFileSync(args.path, "utf8"), loader: "ts" };
+          });
+        },
+      };
+    `,
+    "hold/.keep": "",
+    "index.html": emptyHtmlFile({ scripts: ["index.ts"] }),
+    "index.ts": `
+      import { held } from "./held";
+      import { a } from "./a";
+      import { b } from "./b";
+      import { c } from "./c";
+      console.log(held, a, b, c);
+    `,
+    "held.ts": `export const held = "held-0";`,
+    "a.ts": `export const a = "a-0";`,
+    "b.ts": `export const b = "b-0";`,
+    "c.ts": `export const c = "c-0";`,
+  },
+  async test(dev) {
+    const served = async () => {
+      const html = await (await dev.fetch("/")).text();
+      const src = html.match(/<script[^>]+src="([^"]+)"[^>]*data-bun-dev-server-script/)![1];
+      return await (await dev.fetch(src)).text();
+    };
+    const until = async (what: string, done: () => boolean | Promise<boolean>) => {
+      const deadline = Date.now() + 30_000;
+      while (!(await done())) {
+        if (Date.now() > deadline) throw new Error("Timed out waiting for " + what);
+        await Bun.sleep(10);
+      }
+    };
+    expect(await served()).toContain('"held-0"');
+
+    writeFileSync(dev.join("hold/on"), "");
+    writeFileSync(dev.join("held.ts"), `export const held = "held-1";`);
+    await until("the bundle of held.ts to start", () => existsSync(dev.join("hold/entered")));
+
+    // Each save has to reach the dev server as a watcher batch of its own. The
+    // watcher gives no signal, so space them by far more than the window in
+    // which it merges events.
+    const saved = ["a", "b", "c"];
+    for (const name of saved) {
+      writeFileSync(dev.join(name + ".ts"), `export const ${name} = "${name}-1";`);
+      await Bun.sleep(100);
+    }
+    unlinkSync(dev.join("hold/on"));
+
+    // One bundle follows the held one. The last save is in it.
+    let text = "";
+    await until("the bundle after the held one", async () => (text = await served()).includes('"c-1"'));
+    expect(saved.filter(name => !text.includes(`"${name}-1"`))).toEqual([]);
+  },
+});