From a9a3b370799436101986c920ad4ef644c48696e3 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 16 Sep 2026 05:02:55 +0000 Subject: [PATCH] watcher(windows): keep one ReadDirectoryChangesW outstanding WindowsWatcher::next issued a new read before every wait, also while the read of an earlier, timed out wait was still queued. Every watcher cycle left one more read queued on the directory handle. All reads share one buffer. A read that finds changes already recorded completes inside the call and fills the buffer at once. A read that completes later fills it only when its packet is dequeued. With two reads in flight, the dequeue of the older packet overwrote the records of the newer read. The watcher parsed the older records twice and never reported the newer changes. Issue a read only when none is outstanding. --- src/watcher/WindowsWatcher.rs | 23 ++++++-- test/cli/hot/watch-many-dirs.test.ts | 87 +++++++++++++++++++++++++++- 2 files changed, 104 insertions(+), 6 deletions(-) diff --git a/src/watcher/WindowsWatcher.rs b/src/watcher/WindowsWatcher.rs index e39adde6e990..a67522f3b211 100644 --- a/src/watcher/WindowsWatcher.rs +++ b/src/watcher/WindowsWatcher.rs @@ -22,6 +22,8 @@ pub struct WindowsWatcher { pub(crate) watcher: DirWatcher, pub(crate) buf: PathBuffer, pub(crate) base_idx: usize, + /// The packet of a read into `watcher.buf` has not been dequeued from `iocp` yet. + read_pending: bool, } impl Default for WindowsWatcher { @@ -35,6 +37,7 @@ impl Default for WindowsWatcher { }, buf: PathBuffer::ZEROED, base_idx: 0, + read_pending: false, } } } @@ -292,9 +295,19 @@ impl WindowsWatcher { Ok(()) } + /// One read at a time: the dequeue of an older packet overwrites a newer read's records. + fn arm(&mut self) -> bun_sys::Result<()> { + if self.read_pending { + return Ok(()); + } + self.watcher.prepare()?; + self.read_pending = true; + Ok(()) + } + /// wait until new events are available fn next(&mut self, timeout: Timeout) -> bun_sys::Result> { - if let Err(err) = self.watcher.prepare() { + if let Err(err) = self.arm() { bun_core::scoped_log!(watcher, "prepare() returned error"); return Err(err); } @@ -313,6 +326,10 @@ impl WindowsWatcher { timeout as w::DWORD, ) }; + if overlapped == &raw mut self.watcher.overlapped { + // The packet of the read is off the port, whether the read succeeded or not. + self.read_pending = false; + } if rc == 0 { let err = w::Win32Error::get(); // `WAIT_TIMEOUT` (258) — not yet a named const on `bun_sys::windows::Win32Error`. @@ -343,9 +360,7 @@ impl WindowsWatcher { watcher, "ReadDirectoryChangesW buffer overflow (nbytes==0); re-arming" ); - if let Err(err) = self.watcher.prepare() { - return Err(err); - } + self.arm()?; continue; } return Ok(Some(EventIterator { diff --git a/test/cli/hot/watch-many-dirs.test.ts b/test/cli/hot/watch-many-dirs.test.ts index 991ef6c3b72f..d7e89f752d83 100644 --- a/test/cli/hot/watch-many-dirs.test.ts +++ b/test/cli/hot/watch-many-dirs.test.ts @@ -1,7 +1,7 @@ import { spawn } from "bun"; import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, forEachLine, isASAN, isCI, isLinux, tempDir } from "harness"; -import { mkdirSync, readdirSync, readlinkSync, realpathSync, writeFileSync } from "node:fs"; +import { bunEnv, bunExe, forEachLine, isASAN, isCI, isLinux, isWindows, tempDir } from "harness"; +import { mkdirSync, readdirSync, readlinkSync, realpathSync, statSync, writeFileSync } from "node:fs"; import { join } from "node:path"; describe("--hot with many directories", () => { @@ -183,6 +183,89 @@ if (globalThis.reloaded++ >= ${maxCount}) process.exit(0); 60000, ); + // The Windows watcher used to issue a new ReadDirectoryChangesW before every + // wait, also while the read of a timed out wait was still queued, so every + // watcher cycle left one more read queued. The queued reads share one + // buffer, and the records of one could be overwritten before the watcher + // parsed them: that change was never reported. Windows charges each queued + // read to the nonpaged pool quota of the process, about 500 bytes. + test.skipIf(!isWindows)("keeps one directory read queued across watcher cycles", async () => { + await using dir = tempDir("hot-one-read", { + "proj/entry.js": `console.log("ready");`, + }); + const cwd = join(String(dir), "proj"); + // Outside `cwd`, so that the watcher does not report its own trace writes. + const trace = join(String(dir), "trace.jsonl"); + + await using proc = spawn({ + cmd: [bunExe(), "--hot", "entry.js"], + cwd, + env: { ...bunEnv, BUN_WATCHER_TRACE: trace }, + stdout: "pipe", + stderr: "inherit", + }); + // One PowerShell answers both samples: its start is most of this test's time. + await using powershell = spawn({ + cmd: [ + "powershell", + "-NoProfile", + "-NonInteractive", + "-Command", + `while ($null -ne [Console]::In.ReadLine()) { [Console]::Out.WriteLine((Get-Process -Id ${proc.pid}).NonpagedSystemMemorySize64) }`, + ], + stdin: "pipe", + stdout: "pipe", + stderr: "inherit", + }); + const samples = forEachLine(powershell.stdout); + const nonpagedPoolBytes = async () => { + powershell.stdin.write("\n"); + powershell.stdin.flush(); + const { value: line, done } = await samples.next(); + if (done) throw new Error(`powershell exited before it printed a sample (exit ${powershell.exitCode})`); + return Number(line); + }; + + const ready = async () => { + for await (const line of forEachLine(proc.stdout)) if (line === "ready") return; + throw new Error(`--hot exited before it printed "ready" (exit ${proc.exitCode})`); + }; + await ready(); + + // A write of a file that nothing imports is one watcher cycle: the watcher + // traces a batch for the directory and reloads nothing. + const other = join(cwd, "other.txt"); + const traceSize = () => statSync(trace, { throwIfNoEntry: false })?.size ?? 0; + let writes = 0; + const cycle = async (patienceMs: number) => { + const size = traceSize(); + writeFileSync(other, String(writes++)); + const deadline = performance.now() + patienceMs; + while (traceSize() === size) { + if (performance.now() > deadline) return false; + await new Promise(resolve => setImmediate(resolve)); + } + return true; + }; + + // A new watcher records nothing until its thread has issued the first + // read, so the first writes can go unseen. + let armed = false; + for (let attempt = 0; attempt < 20 && !armed; attempt++) armed = await cycle(100); + if (!armed) throw new Error(`the watcher traced no write of other.txt to ${trace}`); + + const before = await nonpagedPoolBytes(); + for (let i = 0; i < 200; i++) { + if (!(await cycle(2000))) throw new Error(`the watcher did not trace write ${writes} of other.txt`); + } + const after = await nonpagedPoolBytes(); + + // Guard against a vacuous pass: the counter must be live. + expect(before).toBeGreaterThan(0); + // One read queued per cycle is about 100 KiB. + expect(after - before).toBeLessThan(16 * 1024); + }); + // Editing dep.js raises an inotify event on lib/, which makes the reloader // evict dep's watchlist entry; the reload then re-adds it with a fresh heap // copy of the path. Eviction used to discard the evicted entry's copy