Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 86 additions & 41 deletions src/runtime/node/path_watcher.rs
Original file line number Diff line number Diff line change
Expand Up @@ -203,13 +203,35 @@ pub struct PathWatcher {
/// `manager.mutex` on all platforms — every emit path (inotify/kqueue reader
/// threads and the Darwin FSEvents callback) holds it while iterating, so
/// attach/detach can never race with dispatch.
handlers: ArrayHashMap<*mut c_void, ChangeEvent>,
handlers: ArrayHashMap<*mut c_void, Handler>,

/// Per-platform per-watch state (inotify wds, kqueue fds, or the FSEventsWatcher).
#[cfg(not(windows))]
platform: PlatformWatch,
}

/// One `fs.watch()` call attached to a [`PathWatcher`].
struct Handler {
/// Basename of the path this call watched, before symlinks are resolved.
#[cfg(any(target_os = "linux", target_os = "android"))]
name: Box<[u8]>,
change: ChangeEvent,
}

impl Handler {
/// `watched_path` is `watch()`'s `path`: absolute, symlinks unresolved.
fn new(watched_path: &ZStr) -> Self {
// Not Linux: libuv's kqueue and FSEvents backends report the resolved name, the basename of `PathWatcher.path`.
#[cfg(not(any(target_os = "linux", target_os = "android")))]
let _ = watched_path;
Self {
#[cfg(any(target_os = "linux", target_os = "android"))]
name: Box::from(path::basename(watched_path.as_bytes())),
change: ChangeEvent::default(),
}
}
}

/// Per-handler duplicate suppression.
///
/// Suppresses only exact duplicates: same path hash *and* same event type
Expand Down Expand Up @@ -265,8 +287,8 @@ impl PathWatcher {
fn emit(&self, event_type: WatchEventKind, rel_path: &[u8], is_file: bool) {
let timestamp = bun_core::time::milli_timestamp();
let h = hash(rel_path);
for (&ctx, ev) in self.handlers.iter() {
if ev.should_emit(h, timestamp, event_type) {
for (&ctx, handler) in self.handlers.iter() {
if handler.change.should_emit(h, timestamp, event_type) {
(FSWatcher::ON_PATH_UPDATE)(
Some(ctx),
event_type.to_event(rel_path.into()),
Expand All @@ -276,15 +298,37 @@ impl PathWatcher {
}
}

/// Like [`emit`](Self::emit), but without per-handler duplicate suppression.
/// [`emit`](Self::emit) for an event about the watched path itself. Each handler gets its own name, as in libuv: https://github.com/libuv/libuv/blob/v1.52.1/src/unix/linux.c#L2625
#[cfg(any(target_os = "linux", target_os = "android"))]
fn emit_self(&self, event_type: WatchEventKind, is_file: bool) {
let timestamp = bun_core::time::milli_timestamp();
for (&ctx, handler) in self.handlers.iter() {
if handler
.change
.should_emit(hash(&handler.name), timestamp, event_type)
Comment thread
robobun marked this conversation as resolved.
Resolved
{
(FSWatcher::ON_PATH_UPDATE)(
Some(ctx),
event_type.to_event(handler.name.clone()),
is_file,
);
}
}
}

/// Like [`emit_self`](Self::emit_self), but without per-handler duplicate suppression.
/// The `IN_IGNORED` retiring a deleted inode's wd lands in the same
/// millisecond as its `IN_DELETE_SELF`, with the same path and type, so
/// `should_emit` would fold the two into one; node (libuv) delivers both.
/// Caller holds `manager.mutex`.
#[cfg(any(target_os = "linux", target_os = "android"))]
fn emit_unsuppressed(&self, event_type: WatchEventKind, rel_path: &[u8], is_file: bool) {
for &ctx in self.handlers.keys() {
(FSWatcher::ON_PATH_UPDATE)(Some(ctx), event_type.to_event(rel_path.into()), is_file);
fn emit_self_unsuppressed(&self, event_type: WatchEventKind, is_file: bool) {
for (&ctx, handler) in self.handlers.iter() {
(FSWatcher::ON_PATH_UPDATE)(
Some(ctx),
event_type.to_event(handler.name.clone()),
is_file,
);
}
}

Expand Down Expand Up @@ -473,14 +517,16 @@ pub(crate) fn watch(

let mut key_buf = path::path_buffer_pool::get();
let key = PathWatcherManager::make_key(key_buf.as_mut_slice(), resolved.as_bytes(), recursive);
// Per handler: libuv keeps one path per wd, so two node watchers on one inode share the first one's name. Here each reports its own.
let handler = Handler::new(path);

manager.mutex.lock();

// SAFETY: holding manager.mutex; exclusive access to manager.watchers,
// scoped to this lookup.
if let Some(&existing) = unsafe { (*manager.watchers.get()).get(key) } {
// SAFETY: existing is a live PathWatcher under manager.mutex.
unsafe { handle_oom((*existing).handlers.put(ctx, ChangeEvent::default())) };
unsafe { handle_oom((*existing).handlers.put(ctx, handler)) };
Comment thread
robobun marked this conversation as resolved.
Resolved
manager.mutex.unlock();
return Ok(existing);
}
Expand All @@ -501,7 +547,7 @@ pub(crate) fn watch(
platform: PlatformWatch::default(),
});
// SAFETY: watcher just allocated; we hold the only reference.
unsafe { handle_oom((*watcher).handlers.put(ctx, ChangeEvent::default())) };
unsafe { handle_oom((*watcher).handlers.put(ctx, handler)) };
// SAFETY: holding manager.mutex; exclusive access to manager.watchers.
unsafe { handle_oom((*manager.watchers.get()).put(key, watcher)) };

Expand Down Expand Up @@ -978,14 +1024,10 @@ impl Linux {
if let Some(owners) = wd_map.get_mut(&wd) {
for o in owners.drain(..) {
// SAFETY: o.watcher live under manager.mutex; shared
// access only — `emit_unsuppressed` takes `&self`.
// access only — `emit_self_unsuppressed` takes `&self`.
let w = unsafe { &*o.watcher };
if o.subpath.as_bytes().is_empty() && (w.is_file || !w.recursive) {
w.emit_unsuppressed(
WatchEventKind::Rename,
path::basename(w.path.as_bytes()),
w.is_file,
);
w.emit_self_unsuppressed(WatchEventKind::Rename, w.is_file);
let _ = handle_oom(touched.get_or_put(o.watcher));
}
// SAFETY: exclusive scoped access to this watcher's wd
Expand Down Expand Up @@ -1061,12 +1103,11 @@ impl Linux {
&*std::ptr::from_ref::<[u8]>(o.subpath.as_bytes()),
)
};
// SAFETY: owner_watcher live under manager.mutex. Copy the
// scalars and launder the path bytes via a raw ptr so `rel`
// (which may borrow them) is decoupled from the scoped `&mut`
// temporaries `add_one` takes below. `path` is a `ZBox`; its
// heap bytes are a separate allocation, so this mirrors the
// `owner_subpath` raw-ptr laundering above.
// SAFETY: owner_watcher live under manager.mutex. Copy the scalars
// and launder the path bytes via a raw ptr so they are decoupled
// from the scoped `&mut` temporaries `add_one` takes below. `path`
// is a `ZBox`; its heap bytes are a separate allocation, so this
// mirrors the `owner_subpath` raw-ptr laundering above.
let (watcher_is_file, watcher_recursive, watcher_path): (bool, bool, &[u8]) = unsafe {
(
(*owner_watcher).is_file,
Expand All @@ -1075,48 +1116,52 @@ impl Linux {
)
};

// Build the path relative to this owner's root.
let rel: &[u8] = if watcher_is_file {
path::basename(watcher_path)
// Build the path relative to this owner's root; `None` is the watched path itself.
let rel: Option<&[u8]> = if watcher_is_file {
None
} else if owner_subpath.is_empty() {
if name.is_empty() && !watcher_recursive {
// A nameless event on the root wd is about the watched
// directory itself (IN_DELETE_SELF, IN_MOVE_SELF,
// IN_ATTRIB); libuv reports basename(watched path),
// same as for a file. node's recursive watcher uses
// root-relative paths instead, so those keep "".
path::basename(watcher_path)
None
} else {
name
Some(name)
}
} else if name.is_empty() {
owner_subpath
Some(owner_subpath)
} else {
join_z_buf_spill::<platform::Posix>(
path_buf.as_mut_slice(),
&mut rel_spill,
&[owner_subpath, name],
Some(
join_z_buf_spill::<platform::Posix>(
path_buf.as_mut_slice(),
&mut rel_spill,
&[owner_subpath, name],
)
.as_bytes(),
)
.as_bytes()
};

// SAFETY: owner_watcher live under manager.mutex; `emit` takes `&self`.
let is_file_event = !is_dir_child
&& !((ev.mask & (IN::DELETE_SELF | IN::MOVE_SELF) != 0)
&& !watcher_is_file);
// SAFETY: owner_watcher live under manager.mutex; `emit` and
// `emit_self` take `&self`.
unsafe {
(*owner_watcher).emit(
event_type,
rel,
!is_dir_child
&& !((ev.mask & (IN::DELETE_SELF | IN::MOVE_SELF) != 0)
&& !watcher_is_file),
);
match rel {
Some(rel) => (*owner_watcher).emit(event_type, rel, is_file_event),
None => (*owner_watcher).emit_self(event_type, is_file_event),
}
}
let _ = handle_oom(touched.get_or_put(owner_watcher));

// Recursive: a new directory appeared under this owner's tree —
// start watching it so future events inside it are delivered.
// This is what makes `{recursive: true}` track structure changes
// after the initial crawl (#15939/#15085).
if watcher_recursive
if let Some(rel) = rel
&& watcher_recursive
&& is_dir_child
&& (ev.mask & (IN::CREATE | IN::MOVED_TO) != 0)
&& !name.is_empty()
Expand Down
72 changes: 72 additions & 0 deletions test/js/node/watch/fixtures/symlink-filename-fixture.mjs

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

67 changes: 67 additions & 0 deletions test/js/node/watch/fs.watch.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import {
isLinux,
isMacOS,
isWindows,
nodeExe,
tempDir,
tempDirWithFiles,
} from "harness";
Expand Down Expand Up @@ -620,6 +621,72 @@ describe("fs.watch", () => {
]);
});

// The "watched path" of a self-event is the path fs.watch was given, so a
// symlink is reported under its own name and not under the name of what it
// points to. kqueue (macOS files, FreeBSD) asks the kernel for the open
// file's path and reports the resolved name instead, so Linux only.
describe.skipIf(!isLinux)("a watched symlink is reported under its own name", () => {
async function runFixture(exe: string) {
using dir = tempDir("fs-watch-symlink-name", {});
await using proc = Bun.spawn({
cmd: [exe, path.join(import.meta.dir, "fixtures", "symlink-filename-fixture.mjs"), String(dir)],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
return { events: exitCode === 0 ? JSON.parse(stdout) : stdout, stderr, exitCode };
}

const removed = (name: string) => [
["change", name],
["rename", name],
["rename", name],
];
const oneWatcher = {
"file changed": { link: [["change", "link.txt"]] },
"file removed": { link: removed("link.txt") },
"directory removed": {
link: [
["rename", "link-dir"],
["rename", "link-dir"],
],
},
// An entry of the directory keeps its own name.
"directory entry created": { link: [["rename", "child.txt"]] },
};

test.concurrent("in bun, also when the file it points to is watched too", async () => {
expect(await runFixture(bunExe())).toEqual({
events: {
...oneWatcher,
"file watched first, file removed": { target: removed("target.txt"), link: removed("link.txt") },
"symlink watched first, file removed": { link: removed("link.txt"), target: removed("target.txt") },
},
stderr: "",
exitCode: 0,
});
});

// Deliberate difference: libuv keeps one path per inotify watch, so two
// node watchers on one inode both report the name the first one was given.
// https://github.com/libuv/libuv/blob/v1.52.1/src/unix/linux.c#L2706-L2714
test.concurrent.skipIf(!nodeExe())(
"in node, where two watchers on one file share the first one's name",
async () => {
expect(await runFixture(nodeExe()!)).toEqual({
events: {
...oneWatcher,
"file watched first, file removed": { target: removed("target.txt"), link: removed("target.txt") },
"symlink watched first, file removed": { link: removed("link.txt"), target: removed("link.txt") },
},
stderr: "",
exitCode: 0,
});
},
);
});

// Past fs.inotify.max_queued_events the kernel drops events and queues one
// IN_Q_OVERFLOW; Bun reports it as ('change', null) on every watcher sharing
// the inotify fd, the same shape node uses for overflow on Windows.
Expand Down
Loading