From 07ef2dc077947b2ece52f717e6999d681ac130fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 3 Sep 2026 11:19:46 +0200 Subject: [PATCH] fix(runtime): fs.watch uses OS change notifications, not a 25 ms tree re-walk (#9591) Every watcher was a 25 ms setInterval whose tick re-walked the whole watch target on the main thread (41 % of a core at 3k files; a 362k-file cwd is a wedged loop). notify (inotify / FSEvents / ReadDirectoryChangesW / kqueue) now delivers events through the event pump's producer protocol and a runtime pump slot; liveness moved from the ref'd timer to a runtime has-active slot; the walker survives only as an off-main-thread fallback paced to 5 % of a core. Claude-Session: https://claude.ai/code/session_01NjZgUzTJMGtr8fpruGYdNp --- Cargo.lock | 1 + changelog.d/9613-fs-watch-os-events.md | 63 ++ crates/perry-runtime/Cargo.toml | 9 + crates/perry-runtime/src/fs/dir_glob_watch.rs | 3 + .../src/fs/dir_glob_watch/watch.rs | 511 ++++++---- .../src/fs/dir_glob_watch/watch_backend.rs | 896 ++++++++++++++++++ .../src/fs/dir_glob_watch/watch_fsevents.rs | 403 ++++++++ crates/perry-runtime/src/lib.rs | 73 ++ .../issue_9591_fs_watch_native_events.rs | 245 +++++ test-files/test_gap_9591_fs_watch_events.ts | 117 +++ 10 files changed, 2114 insertions(+), 207 deletions(-) create mode 100644 changelog.d/9613-fs-watch-os-events.md create mode 100644 crates/perry-runtime/src/fs/dir_glob_watch/watch_backend.rs create mode 100644 crates/perry-runtime/src/fs/dir_glob_watch/watch_fsevents.rs create mode 100644 crates/perry/tests/issue_9591_fs_watch_native_events.rs create mode 100644 test-files/test_gap_9591_fs_watch_events.ts diff --git a/Cargo.lock b/Cargo.lock index df85e88428..3192d55a58 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6345,6 +6345,7 @@ dependencies = [ "libmimalloc-sys", "mach2", "mimalloc", + "notify", "perry-diagnostics", "perry-dispatch", "perry-parser", diff --git a/changelog.d/9613-fs-watch-os-events.md b/changelog.d/9613-fs-watch-os-events.md new file mode 100644 index 0000000000..5695a227f7 --- /dev/null +++ b/changelog.d/9613-fs-watch-os-events.md @@ -0,0 +1,63 @@ +**fix(runtime): `fs.watch` / `fsPromises.watch` use OS change notifications instead of re-walking the tree every 25 ms (#9591)** + +Every watcher was a 25 ms `setInterval` whose tick re-walked the WHOLE watch +target (`read_dir` + `symlink_metadata` per entry) and diffed two maps, on the +main thread: ~3.4 µs per file per tick, 40 ticks a second. Watching 3 000 +files for 5 s cost 2.03 s of CPU (41 % of a core) against node's 0.03 s. +claude-code watches its cwd; the field session's cwd held 362 295 files, which +extrapolates to ~1.2 s of walking per 25 ms schedule — a wedged event loop, +#9588's symptom exactly. + +The OS now reports changes. `notify` (inotify on Linux/Android, +`ReadDirectoryChangesW` on Windows, kqueue on the BSDs) or, on macOS, FSEvents +bound at runtime through `dlopen` (`watch_fsevents.rs`) runs on its own thread +and queues each event; the queue follows the event pump's producer protocol +(push, then `js_notify_main_thread()`), and a runtime pump slot drains it once +per loop turn and routes each event to the watchers it concerns — +`'rename'` for create / remove / move, `'change'` for data and metadata +writes, filenames relative to the watched root. Nothing walks anything on a +timer any more. Watching 3 000 files for 4 s now costs single-digit +milliseconds of CPU, and a change surfaces in milliseconds instead of at the +next tick. + +Instances mirror libuv's sharing: non-recursive watchers share one instance +per JS thread, refcounted by canonical root — `fs.inotify.max_user_instances` +defaults to 128 and a chokidar / `tsc --watch` style consumer opens one +`fs.watch` per directory, so one instance per watcher would have failed at the +129th. Recursive watchers get their own instance, because notify keys its +per-path bookkeeping by path and a recursive root sharing an instance with a +non-recursive watch of one of its subdirectories would clobber it. Liveness +moved from the ref'd interval timer to a new runtime has-active slot +(`register_runtime_has_active`, the counterpart of `register_runtime_pump`), so +`persistent: false`, `ref()`, `unref()` and `close()` release the loop as before. +Backend errors (an exhausted inotify watch table while a recursive watcher +adds a new subdirectory, for instance) now reach `'error'` listeners as +Node-shaped fs errors, or the uncaught-exception path when there is none. + +macOS does not use notify's FSEvents backend on purpose: `fsevent-sys` links +CoreServices through `#[link]` metadata that does not survive perry's custom +link step, and every `fs` importer retains the watcher (the module table pins +`js_fs_watch`), so every such binary would have needed `-framework +CoreServices` — an umbrella that drags CoreFoundation — on its link line, the +launch-time cost perry keeps off console binaries. The ten CoreFoundation / +CoreServices entry points are resolved with `dlopen` at the first `fs.watch` +call instead; a program that never watches never loads the framework, and the +link line is unchanged. Flag classification follows libuv (rename beats +change; latency 0.05 s so bursts coalesce as under node), delivery uses a +private dispatch queue (libdispatch is in libSystem), and the stream is +rebuilt when the path set changes, as libuv and notify do. + +The walker survives only as the fallback for when the OS watch cannot be +established (watch limit, unsupported target, or `PERRY_FS_WATCH_POLL=1` as a +diagnostic switch). It runs on its own thread — the walk never blocks the loop +— and paces itself to 5 % of one core: each walk's duration times 20, clamped +to [25 ms, 5007 ms] (the old cadence at the bottom, `fs.watchFile`'s default at +the top). A 3 000-file tree polls every ~200 ms under it; the 362 k-file tree +every 5 s, off the main thread, instead of 40 times a second on it. + +Verification: `crates/perry/tests/issue_9591_fs_watch_native_events.rs` is the +issue's bar — watch 3 000 files for a 4 s window, assert < 5 % of a core AND +that a new file is reported within a second (the unfixed walker burns ~1.6 s +in that window); the same for the forced poller at a 12.5 % budget. +`test-files/test_gap_9591_fs_watch_events.ts` pins the event contract against +node for the callback, single-file, recursive and promise-iterator forms. diff --git a/crates/perry-runtime/Cargo.toml b/crates/perry-runtime/Cargo.toml index d7822cf070..cfece76e13 100644 --- a/crates/perry-runtime/Cargo.toml +++ b/crates/perry-runtime/Cargo.toml @@ -449,3 +449,12 @@ sha2 = "0.11" # the generated `[u16; 128]` arrays land in the binary. Already vetted in the # workspace lock (allsorts / lopdf depend on it). encoding_rs = "0.8" + +# #9591: OS change notifications behind `fs.watch` / `fsPromises.watch` +# (inotify / ReadDirectoryChangesW / kqueue), replacing the timer-driven full +# tree walk. See `fs/dir_glob_watch/watch_backend.rs`. NOT on macOS: notify's +# FSEvents backend needs `-framework CoreServices` on every link line that +# retains the watcher (every `fs` importer), which perry keeps off console +# binaries for launch time; `watch_fsevents.rs` binds FSEvents via dlopen. +[target.'cfg(not(target_os = "macos"))'.dependencies] +notify.workspace = true diff --git a/crates/perry-runtime/src/fs/dir_glob_watch.rs b/crates/perry-runtime/src/fs/dir_glob_watch.rs index 4a48e9d365..b189df3aca 100644 --- a/crates/perry-runtime/src/fs/dir_glob_watch.rs +++ b/crates/perry-runtime/src/fs/dir_glob_watch.rs @@ -10,6 +10,9 @@ use super::*; mod glob; mod opendir; mod watch; +mod watch_backend; +#[cfg(target_os = "macos")] +mod watch_fsevents; // Re-export the opendir entry points consumed cross-module (callbacks.rs, // node_submodules/fs_promises.rs) plus the unmangled FFI sync symbol. diff --git a/crates/perry-runtime/src/fs/dir_glob_watch/watch.rs b/crates/perry-runtime/src/fs/dir_glob_watch/watch.rs index 3502a4755d..68e127ee70 100644 --- a/crates/perry-runtime/src/fs/dir_glob_watch/watch.rs +++ b/crates/perry-runtime/src/fs/dir_glob_watch/watch.rs @@ -10,19 +10,20 @@ use crate::fs::{encoded_string_ptr, fs_encoding_option}; use crate::string::js_string_from_bytes; use std::cell::RefCell; -use std::collections::{BTreeMap, HashMap, VecDeque}; +use std::collections::{HashMap, VecDeque}; use std::fs; #[cfg(unix)] use std::os::unix::fs::PermissionsExt; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Once; +use super::watch_backend::{self, Backend, RawEvent, Source, WatchError, WatchEvent}; + use crate::closure::{ js_closure_alloc, js_closure_get_capture_f64, js_closure_set_capture_f64, js_register_closure_arity, ClosureHeader, }; -const FS_WATCH_POLL_INTERVAL_MS: f64 = 25.0; const WATCH_FILE_DEFAULT_INTERVAL_MS: f64 = 5007.0; #[derive(Clone, Copy)] @@ -31,32 +32,17 @@ struct WatchListener { once: bool, } -#[derive(Clone, PartialEq, Eq)] -struct WatchEntry { - is_file: bool, - is_dir: bool, - is_symlink: bool, - len: u64, - mode: u32, - modified_ns: i128, - created_ns: i128, -} - -type WatchSnapshot = BTreeMap; - -#[derive(Clone)] -struct WatchEvent { - event_type: &'static str, - filename: String, -} - struct FsWatchState { path: String, recursive: bool, encoding: String, object_value: f64, - timer_id: i64, - snapshot: WatchSnapshot, + /// The OS watch (or the poll thread) behind this watcher; dropping the + /// state releases it. See `watch_backend` (#9591). + backend: Backend, + /// `persistent` at creation, then `ref()` / `unref()`. A ref'd watcher + /// keeps the event loop alive through `fs_watch_has_active`. + refed: bool, listeners: HashMap>, signal: f64, abort_listener: f64, @@ -94,10 +80,12 @@ struct PromiseWatchState { recursive: bool, encoding: String, object_value: f64, - timer_id: i64, + /// Started lazily by the first `next()` (Node starts the FSEvent handle + /// when the async generator body first runs); `None` until then and + /// again once closed. + backend: Option, persistent: bool, active: bool, - snapshot: WatchSnapshot, queue: VecDeque, pending: VecDeque<*mut crate::promise::Promise>, signal: f64, @@ -264,100 +252,6 @@ fn remove_abort_listener(signal: f64, listener: f64) { } } -fn metadata_time_ns(time: std::io::Result) -> i128 { - time.ok() - .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) - .map(|d| d.as_nanos() as i128) - .unwrap_or(0) -} - -fn watch_entry_from_metadata(meta: &fs::Metadata) -> WatchEntry { - let ft = meta.file_type(); - #[cfg(unix)] - let mode = meta.permissions().mode(); - #[cfg(not(unix))] - let mode = if meta.permissions().readonly() { - 0o444 - } else { - 0o666 - }; - WatchEntry { - is_file: ft.is_file(), - is_dir: ft.is_dir(), - is_symlink: ft.is_symlink(), - len: meta.len(), - mode, - modified_ns: metadata_time_ns(meta.modified()), - created_ns: metadata_time_ns(meta.created()), - } -} - -fn relative_path(root: &Path, path: &Path) -> String { - path.strip_prefix(root) - .unwrap_or(path) - .to_string_lossy() - .replace('\\', "/") -} - -fn walk_watch_dir(root: &Path, dir: &Path, recursive: bool, out: &mut WatchSnapshot) { - let Ok(entries) = fs::read_dir(dir) else { - return; - }; - let mut paths: Vec = entries.flatten().map(|entry| entry.path()).collect(); - paths.sort(); - for path in paths { - let Ok(meta) = fs::symlink_metadata(&path) else { - continue; - }; - let rel = relative_path(root, &path); - out.insert(rel, watch_entry_from_metadata(&meta)); - if recursive && meta.is_dir() { - walk_watch_dir(root, &path, true, out); - } - } -} - -fn snapshot_watch_target(path: &str, recursive: bool) -> std::io::Result { - let root = Path::new(path); - let meta = fs::symlink_metadata(root)?; - let mut snapshot = WatchSnapshot::new(); - if meta.is_dir() { - walk_watch_dir(root, root, recursive, &mut snapshot); - } else { - let name = root - .file_name() - .map(|name| name.to_string_lossy().into_owned()) - .unwrap_or_else(|| path.to_string()); - snapshot.insert(name, watch_entry_from_metadata(&meta)); - } - Ok(snapshot) -} - -fn diff_watch_snapshots(previous: &WatchSnapshot, current: &WatchSnapshot) -> Vec { - let mut events = Vec::new(); - let mut keys = BTreeMap::::new(); - for key in previous.keys() { - keys.insert(key.clone(), ()); - } - for key in current.keys() { - keys.insert(key.clone(), ()); - } - for key in keys.keys() { - match (previous.get(key), current.get(key)) { - (None, Some(_)) | (Some(_), None) => events.push(WatchEvent { - event_type: "rename", - filename: key.clone(), - }), - (Some(a), Some(b)) if a != b => events.push(WatchEvent { - event_type: "change", - filename: key.clone(), - }), - _ => {} - } - } - events -} - fn stat_snapshot(path: &str) -> Option { let meta = fs::metadata(path).ok()?; let ft = meta.file_type(); @@ -602,15 +496,25 @@ fn emit_watch_file_change( fn close_fs_watcher(id: usize) { let removed = FS_WATCHERS.with(|watchers| watchers.borrow_mut().remove(&id)); - let Some(mut state) = removed else { + let Some(state) = removed else { return; }; - crate::timer::clearInterval(state.timer_id); - crate::async_hooks::destroy(state.async_ids.async_id); - remove_abort_listener(state.signal, state.abort_listener); - let close_listeners = take_event_listeners(&mut state.listeners, "close"); + let FsWatchState { + backend, + object_value, + mut listeners, + signal, + abort_listener, + async_ids, + .. + } = state; + // Release the OS watch before any user code runs. + drop(backend); + crate::async_hooks::destroy(async_ids.async_id); + remove_abort_listener(signal, abort_listener); + let close_listeners = take_event_listeners(&mut listeners, "close"); for listener in close_listeners { - emit_listener0(state.object_value, listener.callback); + emit_listener0(object_value, listener.callback); } } @@ -633,9 +537,7 @@ fn close_promise_watcher_return(id: usize) -> Vec<*mut crate::promise::Promise> let Some(state) = removed else { return Vec::new(); }; - if state.timer_id != 0 { - crate::timer::clearInterval(state.timer_id); - } + drop(state.backend); remove_abort_listener(state.signal, state.abort_listener); state.pending.into_iter().collect() } @@ -646,11 +548,8 @@ fn abort_promise_watcher(id: usize, reason: f64) -> Vec<*mut crate::promise::Pro let Some(state) = watchers.get_mut(&id) else { return Vec::new(); }; - if state.timer_id != 0 { - crate::timer::clearInterval(state.timer_id); - } + state.backend = None; remove_abort_listener(state.signal, state.abort_listener); - state.timer_id = 0; state.active = false; state.signal = undefined_value(); state.abort_listener = undefined_value(); @@ -754,69 +653,273 @@ fn reject_promise(promise: *mut crate::promise::Promise, reason: f64) { ); } -extern "C" fn fs_watcher_poll_impl(closure: *const ClosureHeader) -> f64 { - let id = js_closure_get_capture_f64(closure, 0) as usize; - let deliveries = FS_WATCHERS.with(|watchers| { - let mut watchers = watchers.borrow_mut(); - let Some(state) = watchers.get_mut(&id) else { - return Vec::new(); +// ============================================================================ +// #9591 — delivery. Events arrive on the backend's queue (notify's thread or +// the poll thread); the runtime pump slot below drains them once per event- +// loop turn and routes each to the JS watchers it concerns. +// ============================================================================ + +/// One routed event, keyed by watcher id only. The JS values delivery needs +/// (the watcher object, its listeners, a pending promise) are read from the +/// state maps at delivery time, right before they are rooted — never cached +/// across another watcher's callback, where a copying collection would leave +/// a cached value stale. A watcher closed by an earlier callback in the same +/// batch simply drops its later events, as Node does after `close()`. +enum Delivery { + Event { id: usize, event: WatchEvent }, + Error { id: usize, error: WatchError }, +} + +fn ensure_pump_registered() { + static REGISTER: Once = Once::new(); + REGISTER.call_once(|| { + crate::stdlib_pump::register_runtime_pump(2, fs_watch_pump_extern); + crate::stdlib_pump::register_runtime_has_active(0, fs_watch_has_active_extern); + }); +} + +extern "C" fn fs_watch_pump_extern() { + pump_fs_watch_events(); +} + +extern "C" fn fs_watch_has_active_extern() -> i32 { + i32::from(fs_watch_has_active()) +} + +/// A ref'd `fs.watch` handle or a started, persistent `fsPromises.watch` +/// iterator keeps the event loop alive — the role the ref'd interval timer +/// played before #9591. +fn fs_watch_has_active() -> bool { + FS_WATCHERS.with(|watchers| watchers.borrow().values().any(|state| state.refed)) + || PROMISE_WATCHERS.with(|watchers| { + watchers + .borrow() + .values() + .any(|state| state.active && state.persistent && !state.closed) + }) +} + +/// Drain the backend queue: route every raw event to the watchers it +/// concerns (state maps borrowed, no JS runs), then deliver (borrows +/// released, JS runs). +fn pump_fs_watch_events() { + let raw = watch_backend::drain_queue(); + if raw.is_empty() { + return; + } + let mut deliveries = Vec::new(); + for item in raw { + route_raw_event(item, &mut deliveries); + } + for delivery in deliveries { + match delivery { + Delivery::Event { id, event } => deliver_event(id, event), + Delivery::Error { id, error } => deliver_error(id, error), + } + } +} + +fn route_raw_event(item: RawEvent, out: &mut Vec) { + match item { + RawEvent::Polled { id, event } => out.push(Delivery::Event { id, event }), + RawEvent::Native { + source, + path, + class, + } => { + for (id, filename) in native_targets(source, &path) { + out.push(Delivery::Event { + id, + event: WatchEvent { + event_type: class.node_name(), + filename, + }, + }); + } + } + RawEvent::Error { + source, + paths, + error, + } => { + for id in error_targets(source, &paths) { + out.push(Delivery::Error { + id, + error: error.clone(), + }); + } + } + } +} + +/// The watchers an OS event on `path` from `source` is delivered to, with the +/// filename each reports. An own instance has exactly one owner; a shared +/// instance's event is offered to every shared watcher whose root is the +/// path or its parent (`filename_for` applies the depth rule). +fn native_targets(source: Source, path: &Path) -> Vec<(usize, String)> { + let mut out = Vec::new(); + let mut consider = |id: usize, backend: &Backend, recursive: bool| { + let root = match (source, backend) { + (Source::Own(owner), Backend::Own { root, .. }) if owner == id => root, + (Source::Shared, Backend::Shared { root }) => root, + _ => return, }; - let current = snapshot_watch_target(&state.path, state.recursive).unwrap_or_default(); - let events = diff_watch_snapshots(&state.snapshot, ¤t); - state.snapshot = current; - events - .into_iter() - .map(|event| { - let callbacks = take_event_listeners(&mut state.listeners, "change"); - (state.object_value, callbacks, event, state.encoding.clone()) - }) - .collect() + if let Some(filename) = watch_backend::filename_for(root, path, recursive) { + out.push((id, filename)); + } + }; + FS_WATCHERS.with(|watchers| { + for (id, state) in watchers.borrow().iter() { + consider(*id, &state.backend, state.recursive); + } }); - for (object_value, callbacks, event, encoding) in deliveries { - emit_fs_watch_event(object_value, callbacks, &event, &encoding); + PROMISE_WATCHERS.with(|watchers| { + for (id, state) in watchers.borrow().iter() { + if let Some(backend) = &state.backend { + consider(*id, backend, state.recursive); + } + } + }); + out +} + +fn error_targets(source: Source, paths: &[PathBuf]) -> Vec { + if let Source::Own(id) = source { + return vec![id]; } - undefined_value() + let mut out = Vec::new(); + let mut consider = |id: usize, backend: &Backend| { + if let Backend::Shared { root } = backend { + if watch_backend::error_concerns_root(root, paths) { + out.push(id); + } + } + }; + FS_WATCHERS.with(|watchers| { + for (id, state) in watchers.borrow().iter() { + consider(*id, &state.backend); + } + }); + PROMISE_WATCHERS.with(|watchers| { + for (id, state) in watchers.borrow().iter() { + if let Some(backend) = &state.backend { + consider(*id, backend); + } + } + }); + out } -extern "C" fn promise_watcher_poll_impl(closure: *const ClosureHeader) -> f64 { - let id = js_closure_get_capture_f64(closure, 0) as usize; - let actions = PROMISE_WATCHERS.with(|watchers| { +/// Deliver one event to watcher `id`: a callback watcher's `'change'` +/// listeners, or a promise watcher's oldest pending `next()` (queued when +/// none is waiting). Ids are unique across both maps. +fn deliver_event(id: usize, event: WatchEvent) { + let fs_target = FS_WATCHERS.with(|watchers| { let mut watchers = watchers.borrow_mut(); - let Some(state) = watchers.get_mut(&id) else { - return Vec::new(); - }; + let state = watchers.get_mut(&id)?; + let callbacks = take_event_listeners(&mut state.listeners, "change"); + Some((state.object_value, callbacks, state.encoding.clone())) + }); + if let Some((object_value, callbacks, encoding)) = fs_target { + emit_fs_watch_event(object_value, callbacks, &event, &encoding); + return; + } + let mut event = Some(event); + let promise_target = PROMISE_WATCHERS.with(|watchers| { + let mut watchers = watchers.borrow_mut(); + let state = watchers.get_mut(&id)?; if state.closed { - return Vec::new(); + return None; } - let current = snapshot_watch_target(&state.path, state.recursive).unwrap_or_default(); - let events = diff_watch_snapshots(&state.snapshot, ¤t); - state.snapshot = current; - let mut actions = Vec::new(); - for event in events { - if let Some(promise) = state.pending.pop_front() { - actions.push((promise, event, state.encoding.clone())); - } else { - state.queue.push_back(event); + match state.pending.pop_front() { + Some(promise) => Some((promise, state.encoding.clone())), + None => { + state + .queue + .push_back(event.take().expect("event consumed once")); + None } } - actions }); - for (promise, event, encoding) in actions { + if let (Some((promise, encoding)), Some(event)) = (promise_target, event) { resolve_promise_with_event(promise, event, encoding); } - undefined_value() +} + +/// Deliver a backend error to watcher `id` as a Node-shaped fs error: the +/// callback watcher's `'error'` listeners (uncaught when there are none), or +/// the promise watcher's pending `next()` rejections, after which the +/// iterator is finished. +fn deliver_error(id: usize, error: WatchError) { + let fs_target = FS_WATCHERS.with(|watchers| { + let mut watchers = watchers.borrow_mut(); + let state = watchers.get_mut(&id)?; + let callbacks = take_event_listeners(&mut state.listeners, "error"); + Some((state.object_value, callbacks, state.path.clone())) + }); + if let Some((object_value, callbacks, path)) = fs_target { + emit_fs_watch_error(object_value, callbacks, &error.to_io_error(), &path); + return; + } + let promise_path = PROMISE_WATCHERS.with(|watchers| { + let watchers = watchers.borrow(); + let state = watchers.get(&id)?; + if state.closed { + return None; + } + Some(state.path.clone()) + }); + if let Some(path) = promise_path { + let scope = crate::gc::RuntimeHandleScope::new(); + let reason = unsafe { build_fs_error_value(&error.to_io_error(), "watch", &path) }; + let reason_handle = scope.root_nanbox_f64(reason); + let pending = abort_promise_watcher(id, reason_handle.get_nanbox_f64()); + for promise in pending { + reject_promise(promise, reason_handle.get_nanbox_f64()); + } + } +} + +/// `'error'` delivery. With no listener Node's EventEmitter throws the error +/// as an uncaught exception; do the same through the process funnel. +fn emit_fs_watch_error( + object_value: f64, + callbacks: Vec, + error: &std::io::Error, + path: &str, +) { + let raw_callbacks: Vec = callbacks.iter().map(|listener| listener.callback).collect(); + let scope = crate::gc::RuntimeHandleScope::new(); + let callback_handles = scope.root_nanbox_f64_slice(&raw_callbacks); + let object_handle = scope.root_nanbox_f64(object_value); + let err_value = unsafe { build_fs_error_value(error, "watch", path) }; + if callbacks.is_empty() { + crate::os::emit_process_uncaught_exception(err_value); + return; + } + let err_handle = scope.root_nanbox_f64(err_value); + let refreshed_callbacks = + crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&callback_handles); + let prev_this = scope.root_nanbox_f64(crate::object::js_implicit_this_get()); + for callback in refreshed_callbacks { + let cb = extract_closure_ptr(callback); + if cb.is_null() { + continue; + } + crate::object::js_implicit_this_set(object_handle.get_nanbox_f64()); + with_watcher_uncaught_trap(|| { + crate::closure::js_closure_call1(cb, err_handle.get_nanbox_f64()); + }); + crate::object::js_implicit_this_set(prev_this.get_nanbox_f64()); + } } fn start_promise_watcher(id: usize, state: &mut PromiseWatchState) { if state.active || state.closed { return; } - let timer_callback = poll_closure_value(promise_watcher_poll_impl as *const u8, id); - let timer_id = crate::timer::setInterval(timer_callback as i64, FS_WATCH_POLL_INTERVAL_MS); - if !state.persistent { - crate::timer::js_timer_unref(timer_id); - } - state.timer_id = timer_id; + ensure_pump_registered(); + state.backend = Some(Backend::start(id, &state.path, state.recursive)); state.active = true; } @@ -877,8 +980,8 @@ extern "C" fn fs_watcher_ref_impl(closure: *const ClosureHeader) -> f64 { let id = js_closure_get_capture_f64(closure, 0) as usize; let self_value = js_closure_get_capture_f64(closure, 1); FS_WATCHERS.with(|watchers| { - if let Some(state) = watchers.borrow().get(&id) { - crate::timer::js_timer_ref(state.timer_id); + if let Some(state) = watchers.borrow_mut().get_mut(&id) { + state.refed = true; } }); self_value @@ -888,8 +991,8 @@ extern "C" fn fs_watcher_unref_impl(closure: *const ClosureHeader) -> f64 { let id = js_closure_get_capture_f64(closure, 0) as usize; let self_value = js_closure_get_capture_f64(closure, 1); FS_WATCHERS.with(|watchers| { - if let Some(state) = watchers.borrow().get(&id) { - crate::timer::js_timer_unref(state.timer_id); + if let Some(state) = watchers.borrow_mut().get_mut(&id) { + state.refed = false; } }); self_value @@ -1168,8 +1271,6 @@ extern "C" fn promise_watcher_self_impl(closure: *const ClosureHeader) -> f64 { fn ensure_watch_method_arities() { static REGISTER: Once = Once::new(); REGISTER.call_once(|| { - js_register_closure_arity(fs_watcher_poll_impl as *const u8, 0); - js_register_closure_arity(promise_watcher_poll_impl as *const u8, 0); js_register_closure_arity(watch_file_poll_impl as *const u8, 0); js_register_closure_arity(fs_watcher_abort_impl as *const u8, 0); js_register_closure_arity(promise_watcher_abort_impl as *const u8, 0); @@ -1459,7 +1560,8 @@ fn normalized_watch_args(arg1: f64, arg2: f64) -> (f64, Option) { } } -/// `fs.watch(path[, options][, listener])` — polling-backed watcher. +/// `fs.watch(path[, options][, listener])` — OS-event-backed watcher (#9591); +/// see `watch_backend`. #[no_mangle] pub extern "C" fn js_fs_watch(path_value: f64, arg1: f64, arg2: f64) -> f64 { validate::validate_path("filename", path_value); @@ -1475,20 +1577,18 @@ pub extern "C" fn js_fs_watch(path_value: f64, arg1: f64, arg2: f64) -> f64 { Ok(signal) => signal, Err(err) => crate::exception::js_throw(err), }; - let snapshot = match snapshot_watch_target(&path, recursive) { - Ok(snapshot) => snapshot, - Err(err) => unsafe { + // Node throws ENOENT & co. at call time; one stat answers that without + // the pre-#9591 full walk of the target. + if let Err(err) = fs::symlink_metadata(&path) { + unsafe { crate::exception::js_throw(build_fs_error_value(&err, "watch", &path)); - }, - }; + } + } let id = next_watch_id(); let object_value = build_fs_watcher_object(id); let async_ids = crate::async_hooks::init_resource("FSEVENTWRAP", object_value, true); - let timer_callback = poll_closure_value(fs_watcher_poll_impl as *const u8, id); - let timer_id = crate::timer::setInterval(timer_callback as i64, FS_WATCH_POLL_INTERVAL_MS); - if !persistent { - crate::timer::js_timer_unref(timer_id); - } + ensure_pump_registered(); + let backend = Backend::start(id, &path, recursive); let abort_listener = signal .map(|signal| add_abort_listener(signal, id, fs_watcher_abort_impl)) .unwrap_or_else(undefined_value); @@ -1505,8 +1605,8 @@ pub extern "C" fn js_fs_watch(path_value: f64, arg1: f64, arg2: f64) -> f64 { recursive, encoding, object_value, - timer_id, - snapshot, + backend, + refed: persistent, listeners, signal: signal_value, abort_listener, @@ -1624,16 +1724,14 @@ pub extern "C" fn js_fs_promises_watch(path_value: f64, options_value: f64) -> f Ok(signal) => signal, Err(err) => crate::exception::js_throw(err), }; - // Snapshot the watch target at creation time. This serves two purposes: - // 1. It validates the path synchronously, matching Node's `watch()` which - // throws (ENOENT etc.) at call time rather than at first iteration. - // 2. It seeds an initial baseline for the state. - let initial_snapshot = match snapshot_watch_target(&path, recursive) { - Ok(snapshot) => snapshot, - Err(err) => unsafe { + // Validate the path synchronously, matching Node's `watch()` which throws + // (ENOENT etc.) at call time rather than at first iteration. The OS watch + // itself starts on the first `next()` (`start_promise_watcher`). + if let Err(err) = fs::symlink_metadata(&path) { + unsafe { crate::exception::js_throw(build_fs_error_value(&err, "watch", &path)); - }, - }; + } + } let id = next_watch_id(); let object_value = build_promise_watcher_object(id); let abort_listener = signal @@ -1654,10 +1752,9 @@ pub extern "C" fn js_fs_promises_watch(path_value: f64, options_value: f64) -> f recursive, encoding, object_value, - timer_id: 0, + backend: None, persistent, active: false, - snapshot: initial_snapshot, queue: VecDeque::new(), pending: VecDeque::new(), signal: signal_value, diff --git a/crates/perry-runtime/src/fs/dir_glob_watch/watch_backend.rs b/crates/perry-runtime/src/fs/dir_glob_watch/watch_backend.rs new file mode 100644 index 0000000000..37e6dfe741 --- /dev/null +++ b/crates/perry-runtime/src/fs/dir_glob_watch/watch_backend.rs @@ -0,0 +1,896 @@ +//! Change-detection backend for `fs.watch` / `fsPromises.watch` (#9591). +//! +//! Before #9591 every watcher was a 25 ms `setInterval` whose tick re-walked +//! the WHOLE watch target (`read_dir` + `symlink_metadata` per entry) and +//! diffed two `BTreeMap`s — ~3.4 µs per file per tick, 40 ticks a second, on +//! the main thread. A 3 000-file tree cost 41 % of a core; claude-code watches +//! its cwd, and a 362 k-file cwd extrapolates to ~1.2 s of walking per 25 ms +//! schedule, which is a wedged event loop (#9588's symptom exactly). +//! +//! Now the OS tells us. A [`NativeInstance`] is inotify on Linux/Android, +//! `ReadDirectoryChangesW` on Windows and kqueue on the BSDs (all through +//! `notify`), and FSEvents on macOS (`watch_fsevents`, bound at runtime via +//! `dlopen` so no binary grows a CoreServices load command). Each backend runs +//! on its own thread and hands events to [`EventQueue`], which follows the +//! event pump's producer protocol — push, then `js_notify_main_thread()` — so +//! the main loop wakes once per burst and delivers from a runtime pump slot +//! (`stdlib_pump::register_runtime_pump`). Nothing walks anything on a timer. +//! +//! Instances, mirroring libuv's sharing: +//! +//! * **Non-recursive watchers share one instance per JS thread**, refcounted +//! by canonical root. libuv shares one inotify fd per loop for the same +//! reason: `fs.inotify.max_user_instances` defaults to 128, and a +//! chokidar / `tsc --watch` style consumer opens one `fs.watch` per +//! directory — thousands of them. One instance per watcher would fail +//! with EMFILE at the 129th directory. +//! * **Recursive watchers get their own instance.** notify keys its per-path +//! bookkeeping (inotify's wd map, FSEvents' `recursive_info`) by path, so a +//! recursive root and a non-recursive watch of one of its subdirectories on +//! the SAME instance would overwrite each other's mode and, on unwatch, +//! remove each other's descriptors. Separate instances have separate maps. +//! +//! Every event carries its [`Source`], so the router only offers a shared +//! instance's events to shared watchers (root == path, or root == parent) and +//! an own instance's events to its one owner. One known imperfection: a +//! shared instance watching both `/a` and `/a/f` (a directory and a file +//! inside it) sees two inotify descriptors report the same write as two +//! events with the same path, and each watcher receives both. Node itself +//! routinely delivers two `'change'` events per write on Linux, so consumers +//! already debounce. +//! +//! The walker survives only as [`PollHandle`]: the fallback when the native +//! watch cannot be established (inotify watch limit, an unsupported target, +//! or `PERRY_FS_WATCH_POLL=1`). It runs on its own thread — the walk never +//! blocks the loop — and paces itself to at most 1/[`POLL_DUTY_DIVISOR`] of +//! one core: each walk's duration times [`POLL_DUTY_DIVISOR`], clamped to +//! [[`POLL_MIN_INTERVAL_MS`], [`POLL_MAX_INTERVAL_MS`]]. + +use std::cell::RefCell; +use std::collections::{BTreeMap, HashMap, VecDeque}; +use std::fs; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +#[cfg(target_os = "macos")] +use super::watch_fsevents::NativeInstance; +#[cfg(not(target_os = "macos"))] +use native::NativeInstance; + +/// Lower bound of the fallback poller's cadence — the pre-#9591 fixed rate, +/// kept for tiny trees where a walk is cheaper than the interval. +pub(super) const POLL_MIN_INTERVAL_MS: u64 = 25; +/// Upper bound of the fallback poller's cadence: `fs.watchFile`'s default +/// interval, Node's own number for "polling is all we have". +pub(super) const POLL_MAX_INTERVAL_MS: u64 = 5007; +/// The fallback poller sleeps `POLL_DUTY_DIVISOR` × the last walk's duration +/// between walks, so it never exceeds 1/20 = 5 % of one core. +pub(super) const POLL_DUTY_DIVISOR: u32 = 20; + +/// Set `PERRY_FS_WATCH_POLL=1` to skip the OS watcher and use the walker for +/// every watcher (a diagnostic escape hatch; also how the fallback is tested). +const FORCE_POLL_ENV: &str = "PERRY_FS_WATCH_POLL"; + +// ============================================================================ +// Snapshot types — the fallback poller's diff domain. +// ============================================================================ + +#[derive(Clone, PartialEq, Eq)] +pub(super) struct WatchEntry { + is_file: bool, + is_dir: bool, + is_symlink: bool, + len: u64, + mode: u32, + modified_ns: i128, + created_ns: i128, +} + +pub(super) type WatchSnapshot = BTreeMap; + +/// One `(eventType, filename)` pair as `fs.watch` reports it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct WatchEvent { + pub(super) event_type: &'static str, + pub(super) filename: String, +} + +fn metadata_time_ns(time: std::io::Result) -> i128 { + time.ok() + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_nanos() as i128) + .unwrap_or(0) +} + +fn watch_entry_from_metadata(meta: &fs::Metadata) -> WatchEntry { + let ft = meta.file_type(); + #[cfg(unix)] + let mode = meta.permissions().mode(); + #[cfg(not(unix))] + let mode = if meta.permissions().readonly() { + 0o444 + } else { + 0o666 + }; + WatchEntry { + is_file: ft.is_file(), + is_dir: ft.is_dir(), + is_symlink: ft.is_symlink(), + len: meta.len(), + mode, + modified_ns: metadata_time_ns(meta.modified()), + created_ns: metadata_time_ns(meta.created()), + } +} + +/// `path` relative to `root`, with forward slashes. +pub(super) fn relative_path(root: &Path, path: &Path) -> String { + path.strip_prefix(root) + .unwrap_or(path) + .to_string_lossy() + .replace('\\', "/") +} + +fn walk_watch_dir(root: &Path, dir: &Path, recursive: bool, out: &mut WatchSnapshot) { + let Ok(entries) = fs::read_dir(dir) else { + return; + }; + let mut paths: Vec = entries.flatten().map(|entry| entry.path()).collect(); + paths.sort(); + for path in paths { + let Ok(meta) = fs::symlink_metadata(&path) else { + continue; + }; + let rel = relative_path(root, &path); + out.insert(rel, watch_entry_from_metadata(&meta)); + if recursive && meta.is_dir() { + walk_watch_dir(root, &path, true, out); + } + } +} + +/// Full walk of `path` (one level, or the whole tree when `recursive`). +/// This is the operation #9591 took off the main thread's 25 ms timer. +pub(super) fn snapshot_watch_target(path: &str, recursive: bool) -> std::io::Result { + let root = Path::new(path); + let meta = fs::symlink_metadata(root)?; + let mut snapshot = WatchSnapshot::new(); + if meta.is_dir() { + walk_watch_dir(root, root, recursive, &mut snapshot); + } else { + let name = root + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_else(|| path.to_string()); + snapshot.insert(name, watch_entry_from_metadata(&meta)); + } + Ok(snapshot) +} + +pub(super) fn diff_watch_snapshots( + previous: &WatchSnapshot, + current: &WatchSnapshot, +) -> Vec { + let mut events = Vec::new(); + let mut keys = BTreeMap::::new(); + for key in previous.keys() { + keys.insert(key.clone(), ()); + } + for key in current.keys() { + keys.insert(key.clone(), ()); + } + for key in keys.keys() { + match (previous.get(key), current.get(key)) { + (None, Some(_)) | (Some(_), None) => events.push(WatchEvent { + event_type: "rename", + filename: key.clone(), + }), + (Some(a), Some(b)) if a != b => events.push(WatchEvent { + event_type: "change", + filename: key.clone(), + }), + _ => {} + } + } + events +} + +// ============================================================================ +// Raw events — what the producers (the OS backend's thread, the poll thread) +// queue. +// ============================================================================ + +/// Node's two `fs.watch` event names. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum EventClass { + Rename, + Change, +} + +impl EventClass { + pub(super) fn node_name(self) -> &'static str { + match self { + EventClass::Rename => "rename", + EventClass::Change => "change", + } + } +} + +/// Which native instance produced an event — the router only offers an event +/// to the watchers registered on that instance (see the module docs). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Source { + /// This thread's shared non-recursive instance. + Shared, + /// The private instance owned by watcher `id`. + Own(usize), +} + +/// A backend failure, in a form that crosses threads and clones freely +/// (`std::io::Error` does neither); rebuilt into an `io::Error` for the +/// Node-shaped error object at delivery. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct WatchError { + pub(super) errno: Option, + pub(super) message: String, +} + +impl WatchError { + pub(super) fn from_io(err: &std::io::Error) -> Self { + WatchError { + errno: err.raw_os_error(), + message: err.to_string(), + } + } + + pub(super) fn generic(message: impl Into) -> Self { + WatchError { + errno: None, + message: message.into(), + } + } + + pub(super) fn to_io_error(&self) -> std::io::Error { + match self.errno { + Some(code) => std::io::Error::from_raw_os_error(code), + None => std::io::Error::other(self.message.clone()), + } + } +} + +pub(super) enum RawEvent { + /// An OS observation of `path`, not yet attributed to a watcher. + Native { + source: Source, + path: PathBuf, + class: EventClass, + }, + /// The fallback poller's diff for watcher `id` — already relative. + Polled { id: usize, event: WatchEvent }, + /// A backend error; `paths` is what the backend attached (may be empty). + Error { + source: Source, + paths: Vec, + error: WatchError, + }, +} + +// ============================================================================ +// The per-thread event queue — producer side of the event-pump protocol. +// ============================================================================ + +pub(super) struct EventQueue { + items: Mutex>, + /// Mirror of `items.len()` so the per-tick pump can skip the mutex when + /// nothing is queued (one relaxed load per event-loop turn). + len: AtomicUsize, +} + +impl EventQueue { + fn new() -> Self { + EventQueue { + items: Mutex::new(VecDeque::new()), + len: AtomicUsize::new(0), + } + } + + /// Queue a batch and wake the consumer ONCE. Safe from any thread. + pub(super) fn push_all(&self, batch: impl IntoIterator) { + let mut items = self.items.lock().unwrap_or_else(|e| e.into_inner()); + let before = items.len(); + items.extend(batch); + let after = items.len(); + self.len.store(after, Ordering::Release); + drop(items); + if after != before { + crate::event_pump::js_notify_main_thread(); + } + } + + fn drain(&self) -> Vec { + if self.len.load(Ordering::Acquire) == 0 { + return Vec::new(); + } + let mut items = self.items.lock().unwrap_or_else(|e| e.into_inner()); + let out: Vec = items.drain(..).collect(); + self.len.store(0, Ordering::Release); + out + } +} + +thread_local! { + // One queue per JS thread: a watcher created on a `perry/thread` worker + // is drained by that worker's pump, never by the main thread's. + static QUEUE: Arc = Arc::new(EventQueue::new()); +} + +fn queue_handle() -> Arc { + QUEUE.with(Arc::clone) +} + +/// Take everything queued for this thread. Called by the pump each turn. +pub(super) fn drain_queue() -> Vec { + QUEUE.with(|queue| queue.drain()) +} + +// ============================================================================ +// Native instances — `notify` everywhere but macOS (see `watch_fsevents`). +// ============================================================================ + +#[cfg(not(target_os = "macos"))] +mod native { + use super::*; + use notify::event::ModifyKind; + use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; + + /// Map one notify event onto Node's vocabulary. + /// + /// Node subscribes to create / delete / move / modify / attribute changes + /// and nothing else — libuv's inotify mask has no `IN_ACCESS` / + /// `IN_CLOSE_WRITE`, so `Access` events are dropped rather than surfacing + /// as a third `'change'` per write. `Other` covers rescan / mount / + /// overflow notices that libuv also does not forward. + pub(super) fn classify(source: Source, event: Event) -> Vec { + let class = match event.kind { + EventKind::Create(_) + | EventKind::Remove(_) + | EventKind::Modify(ModifyKind::Name(_)) => EventClass::Rename, + EventKind::Modify(_) | EventKind::Any => EventClass::Change, + EventKind::Access(_) | EventKind::Other => return Vec::new(), + }; + event + .paths + .into_iter() + .map(|path| RawEvent::Native { + source, + path, + class, + }) + .collect() + } + + #[cfg(unix)] + const ENOENT_CODE: i32 = libc::ENOENT; + #[cfg(not(unix))] + const ENOENT_CODE: i32 = 2; + #[cfg(unix)] + const ENOSPC_CODE: i32 = libc::ENOSPC; + #[cfg(not(unix))] + const ENOSPC_CODE: i32 = 28; + + fn errno_error(code: i32) -> WatchError { + WatchError { + errno: Some(code), + message: std::io::Error::from_raw_os_error(code).to_string(), + } + } + + pub(super) fn watch_error_from_notify(error: notify::Error) -> (Vec, WatchError) { + let notify::Error { kind, paths } = error; + let error = match kind { + notify::ErrorKind::Io(err) => WatchError::from_io(&err), + notify::ErrorKind::PathNotFound => errno_error(ENOENT_CODE), + notify::ErrorKind::MaxFilesWatch => errno_error(ENOSPC_CODE), + notify::ErrorKind::WatchNotFound => WatchError::generic("watch not found"), + notify::ErrorKind::Generic(message) => WatchError::generic(message), + notify::ErrorKind::InvalidConfig(_) => { + WatchError::generic("invalid watcher configuration") + } + }; + (paths, error) + } + + // As visible as `Backend::Own`, which carries it (private_interfaces). + pub(in crate::fs::dir_glob_watch) struct NativeInstance(RecommendedWatcher); + + impl NativeInstance { + pub(super) fn new(source: Source, queue: Arc) -> Result { + let handler = move |result: notify::Result| match result { + Ok(event) => queue.push_all(classify(source, event)), + Err(error) => { + let (paths, error) = watch_error_from_notify(error); + queue.push_all(std::iter::once(RawEvent::Error { + source, + paths, + error, + })); + } + }; + // The interval only matters where `RecommendedWatcher` is notify's + // own `PollWatcher` (targets with no OS facility); inotify and + // ReadDirectoryChangesW ignore it. Match the fallback poller's ceiling. + RecommendedWatcher::new( + handler, + Config::default().with_poll_interval(Duration::from_millis(POLL_MAX_INTERVAL_MS)), + ) + .map(NativeInstance) + .map_err(|error| watch_error_from_notify(error).1) + } + + pub(super) fn watch(&mut self, root: &Path, recursive: bool) -> Result<(), WatchError> { + let mode = if recursive { + RecursiveMode::Recursive + } else { + RecursiveMode::NonRecursive + }; + self.0 + .watch(root, mode) + .map_err(|error| watch_error_from_notify(error).1) + } + + pub(super) fn unwatch(&mut self, root: &Path) { + let _ = self.0.unwatch(root); + } + } +} + +fn new_native_watcher(source: Source) -> Result { + NativeInstance::new(source, queue_handle()) +} + +/// This thread's shared instance for non-recursive watchers. +struct SharedInstance { + watcher: NativeInstance, + /// canonical root → number of JS watchers registered on it. The OS watch + /// is added on 0 → 1 and removed on 1 → 0 (libuv's per-path refcount). + refs: HashMap, +} + +thread_local! { + // Outer `None`: never tried. `Some(None)`: construction failed — every + // non-recursive watcher on this thread uses the poller from then on. + static SHARED: RefCell>> = const { RefCell::new(None) }; +} + +fn with_shared(f: impl FnOnce(&mut SharedInstance) -> R) -> Option { + SHARED + .try_with(|cell| { + let mut slot = cell.borrow_mut(); + if slot.is_none() { + *slot = + Some( + new_native_watcher(Source::Shared) + .ok() + .map(|watcher| SharedInstance { + watcher, + refs: HashMap::new(), + }), + ); + } + slot.as_mut().and_then(|inst| inst.as_mut()).map(f) + }) + .ok() + .flatten() +} + +fn force_poll() -> bool { + static FORCE: OnceLock = OnceLock::new(); + *FORCE.get_or_init(|| std::env::var_os(FORCE_POLL_ENV).is_some_and(|v| v == "1")) +} + +/// The change source behind one JS watcher. Dropping it releases the OS +/// resource (or stops the poll thread). +pub(super) enum Backend { + /// Registered on this thread's shared instance under `root`. + Shared { root: PathBuf }, + /// Owns a private instance (recursive watchers). + Own { + root: PathBuf, + _watcher: NativeInstance, + }, + /// The walker fallback; the handle is held for its `Drop`, which stops + /// the thread. + Poll { _handle: PollHandle }, +} + +impl Backend { + /// Start watching `path` for watcher `id`. Never fails: if the OS watch + /// cannot be established the walker takes over (the caller has already + /// validated that `path` exists, so ENOENT was thrown before this). + pub(super) fn start(id: usize, path: &str, recursive: bool) -> Backend { + if !force_poll() { + if let Ok(backend) = start_native(id, path, recursive) { + return backend; + } + } + // Resolve once, like the native path: a later `process.chdir` must + // not move a relative watch target under the poller. + let resolved = fs::canonicalize(path) + .map(|p| p.to_string_lossy().into_owned()) + .unwrap_or_else(|_| path.to_string()); + Backend::Poll { + _handle: PollHandle::spawn(id, resolved, recursive), + } + } +} + +fn start_native(id: usize, path: &str, recursive: bool) -> Result { + // Events arrive with the OS's idea of the path (FSEvents reports the + // resolved real path, inotify echoes what it was given); registering the + // canonical root makes both strip cleanly at routing time. + let root = fs::canonicalize(path).map_err(|err| WatchError::from_io(&err))?; + if recursive { + let mut watcher = new_native_watcher(Source::Own(id))?; + watcher.watch(&root, true)?; + return Ok(Backend::Own { + root, + _watcher: watcher, + }); + } + let registered = with_shared(|inst| -> Result<(), WatchError> { + let count = inst.refs.entry(root.clone()).or_insert(0); + if *count == 0 { + inst.watcher.watch(&root, false)?; + } + *count += 1; + Ok(()) + }); + match registered { + Some(Ok(())) => Ok(Backend::Shared { root }), + Some(Err(err)) => Err(err), + None => Err(WatchError::generic("shared watcher unavailable")), + } +} + +impl Drop for Backend { + fn drop(&mut self) { + if let Backend::Shared { root } = self { + let root = std::mem::take(root); + let _ = with_shared(|inst| { + let Some(count) = inst.refs.get_mut(&root) else { + return; + }; + *count = count.saturating_sub(1); + if *count == 0 { + inst.refs.remove(&root); + inst.watcher.unwatch(&root); + } + }); + } + } +} + +// ============================================================================ +// Routing helpers (pure). +// ============================================================================ + +/// The `filename` a watcher rooted at `root` reports for an event on `path`, +/// or `None` when `path` is outside the watcher's scope: a non-recursive +/// watcher sees the root itself and its direct children; a recursive one sees +/// everything beneath. An event on the root itself (the watched directory +/// deleted or renamed) reports the root's own name, as libuv does. +pub(super) fn filename_for(root: &Path, path: &Path, recursive: bool) -> Option { + let rel = path.strip_prefix(root).ok()?; + let mut components = rel.components(); + match (components.next(), components.next()) { + (None, _) => Some( + root.file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_default(), + ), + (Some(_), None) => Some(relative_path(root, path)), + (Some(_), Some(_)) if recursive => Some(relative_path(root, path)), + _ => None, + } +} + +/// Whether a shared-instance error attached to `paths` concerns the watcher +/// rooted at `root`. An error with no paths concerns every shared watcher. +pub(super) fn error_concerns_root(root: &Path, paths: &[PathBuf]) -> bool { + paths.is_empty() + || paths + .iter() + .any(|path| path == root || path.parent() == Some(root)) +} + +// ============================================================================ +// The fallback poller. +// ============================================================================ + +/// Sleep between walks so the poller's duty cycle stays at or below +/// 1/`POLL_DUTY_DIVISOR`, within [`POLL_MIN_INTERVAL_MS`, `POLL_MAX_INTERVAL_MS`]. +pub(super) fn poll_interval_for(walk: Duration) -> Duration { + walk.saturating_mul(POLL_DUTY_DIVISOR).clamp( + Duration::from_millis(POLL_MIN_INTERVAL_MS), + Duration::from_millis(POLL_MAX_INTERVAL_MS), + ) +} + +pub(super) struct PollHandle { + active: Arc, + thread: Option>, +} + +impl PollHandle { + fn spawn(id: usize, path: String, recursive: bool) -> PollHandle { + let active = Arc::new(AtomicBool::new(true)); + let queue = queue_handle(); + let flag = Arc::clone(&active); + let thread = std::thread::Builder::new() + .name(format!("perry-fs-watch-poll-{id}")) + .spawn(move || poll_loop(id, &path, recursive, &queue, &flag)) + .ok(); + PollHandle { active, thread } + } +} + +impl Drop for PollHandle { + fn drop(&mut self) { + self.active.store(false, Ordering::Release); + // Wake the sleeper so it exits now; never join — a walk may be in + // flight and the caller is the JS thread. + if let Some(thread) = self.thread.take() { + thread.thread().unpark(); + } + } +} + +fn sleep_unless_stopped(interval: Duration, active: &AtomicBool) -> bool { + let deadline = Instant::now() + interval; + loop { + if !active.load(Ordering::Acquire) { + return false; + } + let now = Instant::now(); + if now >= deadline { + return true; + } + std::thread::park_timeout(deadline - now); + } +} + +fn poll_loop(id: usize, path: &str, recursive: bool, queue: &EventQueue, active: &AtomicBool) { + let mut previous: Option = None; + let mut interval = Duration::from_millis(POLL_MIN_INTERVAL_MS); + loop { + // The first pass is the baseline and runs immediately. + if previous.is_some() && !sleep_unless_stopped(interval, active) { + return; + } + if !active.load(Ordering::Acquire) { + return; + } + let started = Instant::now(); + let current = snapshot_watch_target(path, recursive).unwrap_or_default(); + let walk = started.elapsed(); + if let Some(prev) = &previous { + let events = diff_watch_snapshots(prev, ¤t); + if !events.is_empty() { + queue.push_all( + events + .into_iter() + .map(|event| RawEvent::Polled { id, event }), + ); + } + } + previous = Some(current); + interval = poll_interval_for(walk); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn filename_for_scopes_by_depth() { + let root = Path::new("/w/root"); + assert_eq!( + filename_for(root, Path::new("/w/root/a.txt"), false).as_deref(), + Some("a.txt") + ); + assert_eq!( + filename_for(root, Path::new("/w/root/sub/b.txt"), false), + None, + "a non-recursive watcher must not see grandchildren" + ); + assert_eq!( + filename_for(root, Path::new("/w/root/sub/b.txt"), true).as_deref(), + Some("sub/b.txt") + ); + assert_eq!( + filename_for(root, Path::new("/w/root"), false).as_deref(), + Some("root"), + "an event on the watched entry itself reports its own name (DELETE_SELF)" + ); + assert_eq!( + filename_for(root, Path::new("/w/other/a.txt"), true), + None, + "paths outside the root are never delivered" + ); + assert_eq!( + filename_for(root, Path::new("/w/rooted/a.txt"), true), + None, + "prefix matching is by path component, not by string prefix" + ); + } + + #[test] + fn error_concerns_root_matches_self_parent_or_everything() { + let root = Path::new("/w/root"); + assert!(error_concerns_root(root, &[])); + assert!(error_concerns_root(root, &[PathBuf::from("/w/root")])); + assert!(error_concerns_root(root, &[PathBuf::from("/w/root/x")])); + assert!(!error_concerns_root( + root, + &[PathBuf::from("/w/elsewhere/x")] + )); + } + + #[test] + fn poll_interval_is_twenty_walks_wide_and_clamped() { + assert_eq!( + poll_interval_for(Duration::from_micros(200)), + Duration::from_millis(POLL_MIN_INTERVAL_MS), + "a 0.2 ms walk keeps the old 25 ms cadence" + ); + assert_eq!( + poll_interval_for(Duration::from_millis(10)), + Duration::from_millis(200), + "a 10 ms walk (3k files) polls every 200 ms — 5 % of a core, not 41 %" + ); + assert_eq!( + poll_interval_for(Duration::from_millis(1200)), + Duration::from_millis(POLL_MAX_INTERVAL_MS), + "the 362k-file walk is capped at watchFile's default interval" + ); + } + + #[test] + fn watch_errors_keep_their_errno() { + let error = WatchError::from_io(&std::io::Error::from_raw_os_error(28)); + assert_eq!(error.errno, Some(28)); + assert_eq!(error.to_io_error().raw_os_error(), Some(28)); + let generic = WatchError::generic("boom"); + assert_eq!(generic.errno, None); + assert_eq!(generic.to_io_error().to_string(), "boom"); + } + + #[test] + fn queue_batches_wake_once_and_drain_in_order() { + let queue = EventQueue::new(); + assert!(queue.drain().is_empty()); + queue.push_all((0..3).map(|i| RawEvent::Polled { + id: i, + event: WatchEvent { + event_type: "change", + filename: format!("f{i}"), + }, + })); + assert_eq!(queue.len.load(Ordering::Acquire), 3); + let drained = queue.drain(); + let ids: Vec = drained + .iter() + .map(|item| match item { + RawEvent::Polled { id, .. } => *id, + _ => unreachable!(), + }) + .collect(); + assert_eq!(ids, vec![0, 1, 2]); + assert_eq!(queue.len.load(Ordering::Acquire), 0); + assert!(queue.drain().is_empty()); + } +} + +#[cfg(all(test, not(target_os = "macos")))] +mod notify_tests { + use super::native::{classify, watch_error_from_notify}; + use super::*; + use notify::event::{CreateKind, DataChange, MetadataKind, ModifyKind, RemoveKind, RenameMode}; + use notify::{Event, EventKind}; + + fn classes(kind: EventKind) -> Vec> { + let event = Event::new(kind).add_path(PathBuf::from("/w/a")); + let raw = classify(Source::Shared, event); + if raw.is_empty() { + return vec![None]; + } + raw.into_iter() + .map(|item| match item { + RawEvent::Native { class, .. } => Some(class), + _ => panic!("classify only produces Native events"), + }) + .collect() + } + + #[test] + fn classify_maps_onto_nodes_two_event_names() { + assert_eq!( + classes(EventKind::Create(CreateKind::File)), + vec![Some(EventClass::Rename)] + ); + assert_eq!( + classes(EventKind::Remove(RemoveKind::Any)), + vec![Some(EventClass::Rename)] + ); + assert_eq!( + classes(EventKind::Modify(ModifyKind::Name(RenameMode::Any))), + vec![Some(EventClass::Rename)] + ); + assert_eq!( + classes(EventKind::Modify(ModifyKind::Data(DataChange::Content))), + vec![Some(EventClass::Change)] + ); + assert_eq!( + classes(EventKind::Modify(ModifyKind::Metadata(MetadataKind::Any))), + vec![Some(EventClass::Change)] + ); + assert_eq!(classes(EventKind::Any), vec![Some(EventClass::Change)]); + } + + #[test] + fn classify_drops_what_libuv_never_subscribes_to() { + use notify::event::{AccessKind, AccessMode}; + assert_eq!( + classes(EventKind::Access(AccessKind::Close(AccessMode::Write))), + vec![None], + "IN_CLOSE_WRITE is not in libuv's inotify mask — it must not become a third 'change'" + ); + assert_eq!(classes(EventKind::Other), vec![None]); + } + + #[test] + fn classify_emits_one_event_per_path() { + let event = Event::new(EventKind::Modify(ModifyKind::Name(RenameMode::Both))) + .add_path(PathBuf::from("/w/old")) + .add_path(PathBuf::from("/w/new")); + let raw = classify(Source::Own(7), event); + let paths: Vec = raw + .into_iter() + .map(|item| match item { + RawEvent::Native { + source, + path, + class, + } => { + assert_eq!(source, Source::Own(7)); + assert_eq!(class, EventClass::Rename); + path + } + _ => unreachable!(), + }) + .collect(); + assert_eq!( + paths, + vec![PathBuf::from("/w/old"), PathBuf::from("/w/new")] + ); + } + + #[test] + fn notify_errors_keep_their_errno() { + let (paths, error) = watch_error_from_notify( + notify::Error::io(std::io::Error::from_raw_os_error(28)) + .add_path(PathBuf::from("/w/root")), + ); + assert_eq!(paths, vec![PathBuf::from("/w/root")]); + assert_eq!(error.errno, Some(28)); + let (_, generic) = watch_error_from_notify(notify::Error::generic("boom")); + assert_eq!(generic.errno, None); + assert_eq!(generic.message, "boom"); + } +} diff --git a/crates/perry-runtime/src/fs/dir_glob_watch/watch_fsevents.rs b/crates/perry-runtime/src/fs/dir_glob_watch/watch_fsevents.rs new file mode 100644 index 0000000000..f75b5e23f7 --- /dev/null +++ b/crates/perry-runtime/src/fs/dir_glob_watch/watch_fsevents.rs @@ -0,0 +1,403 @@ +//! FSEvents backend for `fs.watch` on macOS, bound at RUNTIME through +//! `dlopen` (#9591). +//! +//! Why not `notify`'s FSEvents backend: `fsevent-sys` declares +//! `#[link(name = "CoreServices", kind = "framework")]`, and that metadata does +//! not survive perry's custom link step, so every binary whose link retains +//! the watcher — which is every binary that imports `fs`; the module table +//! pins `js_fs_watch` — would need `-framework CoreServices` on its link line. +//! An extra `LC_LOAD_DYLIB` costs launch time (`link/build_and_run.rs`: ~1.8 ms +//! for CoreFoundation + objc on a 3 ms hello world, and CoreServices is an +//! umbrella that drags CoreFoundation and more), which perry deliberately +//! keeps off runtime-only console binaries. Binding the ten symbols we need +//! at the first `fs.watch` call keeps that property: a program that never +//! watches never loads the framework. +//! +//! Semantics follow libuv's `uv__fsevents_event_cb`: one event per reported +//! path, `'rename'` when any of Created / Removed / Renamed / RootChanged is +//! set, otherwise `'change'` when any of the modification flags is set; +//! latency 0.05 s so a burst coalesces the way it does under node. Events +//! arrive on a private dispatch queue (libdispatch lives in libSystem: no +//! CFRunLoop thread, no further dylib) and each callback pushes onto the +//! per-thread [`EventQueue`] like every other cross-thread producer. +//! +//! One instance owns one stream over a set of paths. FSEvents streams are +//! immutable, so adding or removing a path rebuilds the stream (stop, +//! invalidate, release, create, start) — the same thing libuv and notify do. + +use std::ffi::{c_char, c_void, CStr, CString}; +use std::os::unix::ffi::OsStrExt; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, OnceLock}; + +use super::watch_backend::{EventClass, EventQueue, RawEvent, Source, WatchError}; + +type CFRef = *const c_void; +type StreamRef = *mut c_void; +type DispatchQueue = *mut c_void; + +const CORE_FOUNDATION: &[u8] = + b"/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation\0"; +const CORE_SERVICES: &[u8] = b"/System/Library/Frameworks/CoreServices.framework/CoreServices\0"; + +const K_CF_STRING_ENCODING_UTF8: u32 = 0x0800_0100; +const K_FS_EVENT_STREAM_EVENT_ID_SINCE_NOW: u64 = u64::MAX; +/// libuv's latency: events within this window coalesce per path. +const STREAM_LATENCY_SECONDS: f64 = 0.05; +const CREATE_FLAG_FILE_EVENTS: u32 = 0x0000_0010; + +const EVENT_FLAG_ROOT_CHANGED: u32 = 0x0000_0020; +const EVENT_FLAG_ITEM_CREATED: u32 = 0x0000_0100; +const EVENT_FLAG_ITEM_REMOVED: u32 = 0x0000_0200; +const EVENT_FLAG_ITEM_INODE_META_MOD: u32 = 0x0000_0400; +const EVENT_FLAG_ITEM_RENAMED: u32 = 0x0000_0800; +const EVENT_FLAG_ITEM_MODIFIED: u32 = 0x0000_1000; +const EVENT_FLAG_ITEM_FINDER_INFO_MOD: u32 = 0x0000_2000; +const EVENT_FLAG_ITEM_CHANGE_OWNER: u32 = 0x0000_4000; +const EVENT_FLAG_ITEM_XATTR_MOD: u32 = 0x0000_8000; + +const RENAME_FLAGS: u32 = EVENT_FLAG_ITEM_CREATED + | EVENT_FLAG_ITEM_REMOVED + | EVENT_FLAG_ITEM_RENAMED + | EVENT_FLAG_ROOT_CHANGED; +const CHANGE_FLAGS: u32 = EVENT_FLAG_ITEM_MODIFIED + | EVENT_FLAG_ITEM_INODE_META_MOD + | EVENT_FLAG_ITEM_FINDER_INFO_MOD + | EVENT_FLAG_ITEM_CHANGE_OWNER + | EVENT_FLAG_ITEM_XATTR_MOD; + +/// libuv's classification of one FSEvents flag word: rename wins, then +/// change, else the event carries no action (HistoryDone, MustScanSubDirs, +/// bare IsFile/IsDir markers) and is dropped. +pub(super) fn classify_flags(flags: u32) -> Option { + if flags & RENAME_FLAGS != 0 { + Some(EventClass::Rename) + } else if flags & CHANGE_FLAGS != 0 { + Some(EventClass::Change) + } else { + None + } +} + +#[repr(C)] +struct StreamContext { + version: isize, + info: *mut c_void, + retain: Option *const c_void>, + release: Option, + copy_description: Option CFRef>, +} + +type StreamCallback = + unsafe extern "C" fn(StreamRef, *mut c_void, usize, *mut c_void, *const u32, *const u64); + +/// The CoreFoundation / CoreServices entry points, resolved once. +struct Api { + cf_string_create_with_cstring: unsafe extern "C" fn(CFRef, *const c_char, u32) -> CFRef, + cf_array_create: unsafe extern "C" fn(CFRef, *const CFRef, isize, *const c_void) -> CFRef, + cf_release: unsafe extern "C" fn(CFRef), + cf_type_array_callbacks: *const c_void, + stream_create: unsafe extern "C" fn( + CFRef, + StreamCallback, + *const StreamContext, + CFRef, + u64, + f64, + u32, + ) -> StreamRef, + stream_set_dispatch_queue: unsafe extern "C" fn(StreamRef, DispatchQueue), + stream_start: unsafe extern "C" fn(StreamRef) -> u8, + stream_stop: unsafe extern "C" fn(StreamRef), + stream_invalidate: unsafe extern "C" fn(StreamRef), + stream_release: unsafe extern "C" fn(StreamRef), +} + +// SAFETY: function pointers and one immutable data address into a framework +// that stays mapped for the life of the process. +unsafe impl Send for Api {} +unsafe impl Sync for Api {} + +extern "C" { + // libdispatch is part of libSystem — always linked, no new image. + fn dispatch_queue_create(label: *const c_char, attr: *const c_void) -> DispatchQueue; + fn dispatch_release(object: *mut c_void); +} + +fn unavailable(what: &str) -> WatchError { + WatchError { + errno: None, + message: format!("FSEvents unavailable: {what}"), + } +} + +unsafe fn open_image(path: &[u8]) -> Result<*mut c_void, WatchError> { + let handle = libc::dlopen( + path.as_ptr() as *const c_char, + libc::RTLD_LAZY | libc::RTLD_LOCAL, + ); + if handle.is_null() { + let name = CStr::from_bytes_with_nul(path) + .map(|c| c.to_string_lossy().into_owned()) + .unwrap_or_default(); + return Err(unavailable(&format!("cannot load {name}"))); + } + Ok(handle) +} + +unsafe fn symbol(handle: *mut c_void, name: &[u8]) -> Result { + let ptr = libc::dlsym(handle, name.as_ptr() as *const c_char); + if ptr.is_null() { + let name = CStr::from_bytes_with_nul(name) + .map(|c| c.to_string_lossy().into_owned()) + .unwrap_or_default(); + return Err(unavailable(&format!("missing symbol {name}"))); + } + // SAFETY: T is a pointer-sized function pointer or `*const c_void`. + Ok(std::mem::transmute_copy::<*mut c_void, T>(&ptr)) +} + +unsafe fn load_api() -> Result { + let cf = open_image(CORE_FOUNDATION)?; + let cs = open_image(CORE_SERVICES)?; + Ok(Api { + cf_string_create_with_cstring: symbol(cf, b"CFStringCreateWithCString\0")?, + cf_array_create: symbol(cf, b"CFArrayCreate\0")?, + cf_release: symbol(cf, b"CFRelease\0")?, + cf_type_array_callbacks: symbol(cf, b"kCFTypeArrayCallBacks\0")?, + stream_create: symbol(cs, b"FSEventStreamCreate\0")?, + stream_set_dispatch_queue: symbol(cs, b"FSEventStreamSetDispatchQueue\0")?, + stream_start: symbol(cs, b"FSEventStreamStart\0")?, + stream_stop: symbol(cs, b"FSEventStreamStop\0")?, + stream_invalidate: symbol(cs, b"FSEventStreamInvalidate\0")?, + stream_release: symbol(cs, b"FSEventStreamRelease\0")?, + }) +} + +fn api() -> Result<&'static Api, WatchError> { + static API: OnceLock> = OnceLock::new(); + API.get_or_init(|| unsafe { load_api() }) + .as_ref() + .map_err(Clone::clone) +} + +/// What the stream callback needs: where to push, and which instance it is. +struct Sink { + source: Source, + queue: Arc, +} + +unsafe extern "C" fn release_sink(info: *const c_void) { + drop(Arc::from_raw(info as *const Sink)); +} + +unsafe extern "C" fn stream_callback( + _stream: StreamRef, + info: *mut c_void, + num_events: usize, + event_paths: *mut c_void, + event_flags: *const u32, + _event_ids: *const u64, +) { + let sink = &*(info as *const Sink); + let paths = event_paths as *const *const c_char; + let mut batch = Vec::with_capacity(num_events); + for i in 0..num_events { + let Some(class) = classify_flags(*event_flags.add(i)) else { + continue; + }; + let bytes = CStr::from_ptr(*paths.add(i)).to_bytes(); + // Directory events may carry a trailing slash; the router compares + // whole components against the canonical root. + let trimmed = if bytes.len() > 1 && bytes.ends_with(b"/") { + &bytes[..bytes.len() - 1] + } else { + bytes + }; + batch.push(RawEvent::Native { + source: sink.source, + path: PathBuf::from(std::ffi::OsStr::from_bytes(trimmed)), + class, + }); + } + sink.queue.push_all(batch); +} + +/// One FSEvents stream over a set of watched paths. +pub(super) struct NativeInstance { + api: &'static Api, + sink: Arc, + dispatch_queue: DispatchQueue, + paths: Vec, + stream: StreamRef, +} + +// SAFETY: the raw handles are only ever used from the owning thread; FSEvents +// delivers on its own dispatch queue through `Sink`, which is `Send + Sync`. +unsafe impl Send for NativeInstance {} + +impl NativeInstance { + pub(super) fn new(source: Source, queue: Arc) -> Result { + let api = api()?; + let label = b"perry.fs.watch\0"; + let dispatch_queue = + unsafe { dispatch_queue_create(label.as_ptr() as *const c_char, std::ptr::null()) }; + if dispatch_queue.is_null() { + return Err(unavailable("dispatch_queue_create failed")); + } + Ok(NativeInstance { + api, + sink: Arc::new(Sink { source, queue }), + dispatch_queue, + paths: Vec::new(), + stream: std::ptr::null_mut(), + }) + } + + /// FSEvents streams are always recursive; the router applies the + /// non-recursive depth rule, so `recursive` is not needed here. + pub(super) fn watch(&mut self, root: &Path, _recursive: bool) -> Result<(), WatchError> { + if !self.paths.iter().any(|p| p == root) { + self.paths.push(root.to_path_buf()); + } + self.rebuild() + } + + pub(super) fn unwatch(&mut self, root: &Path) { + self.paths.retain(|p| p != root); + let _ = self.rebuild(); + } + + fn teardown(&mut self) { + if self.stream.is_null() { + return; + } + unsafe { + (self.api.stream_stop)(self.stream); + (self.api.stream_invalidate)(self.stream); + (self.api.stream_release)(self.stream); + } + self.stream = std::ptr::null_mut(); + } + + fn rebuild(&mut self) -> Result<(), WatchError> { + self.teardown(); + if self.paths.is_empty() { + return Ok(()); + } + let api = self.api; + unsafe { + let mut cf_paths: Vec = Vec::with_capacity(self.paths.len()); + for path in &self.paths { + let c_path = CString::new(path.as_os_str().as_bytes()) + .map_err(|_| unavailable("path contains a NUL byte"))?; + let cf_path = (api.cf_string_create_with_cstring)( + std::ptr::null(), + c_path.as_ptr(), + K_CF_STRING_ENCODING_UTF8, + ); + if cf_path.is_null() { + for created in cf_paths { + (api.cf_release)(created); + } + return Err(unavailable("CFStringCreateWithCString failed")); + } + cf_paths.push(cf_path); + } + let array = (api.cf_array_create)( + std::ptr::null(), + cf_paths.as_ptr(), + cf_paths.len() as isize, + api.cf_type_array_callbacks, + ); + // The array retains its members. + for cf_path in cf_paths { + (api.cf_release)(cf_path); + } + if array.is_null() { + return Err(unavailable("CFArrayCreate failed")); + } + // One strong reference per stream, handed to FSEvents; `release_sink` + // returns it when the stream is released. + let info = Arc::into_raw(Arc::clone(&self.sink)) as *mut c_void; + let context = StreamContext { + version: 0, + info, + retain: None, + release: Some(release_sink), + copy_description: None, + }; + let stream = (api.stream_create)( + std::ptr::null(), + stream_callback, + &context, + array, + K_FS_EVENT_STREAM_EVENT_ID_SINCE_NOW, + STREAM_LATENCY_SECONDS, + CREATE_FLAG_FILE_EVENTS, + ); + (api.cf_release)(array); + if stream.is_null() { + drop(Arc::from_raw(info as *const Sink)); + return Err(unavailable("FSEventStreamCreate failed")); + } + (api.stream_set_dispatch_queue)(stream, self.dispatch_queue); + if (api.stream_start)(stream) == 0 { + (api.stream_invalidate)(stream); + (api.stream_release)(stream); + return Err(unavailable("FSEventStreamStart failed")); + } + self.stream = stream; + } + Ok(()) + } +} + +impl Drop for NativeInstance { + fn drop(&mut self) { + self.teardown(); + unsafe { + dispatch_release(self.dispatch_queue); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn flags_classify_like_libuv() { + assert_eq!( + classify_flags(EVENT_FLAG_ITEM_CREATED | EVENT_FLAG_ITEM_MODIFIED), + Some(EventClass::Rename), + "a create coalesced with its first write is one 'rename', as under node" + ); + assert_eq!( + classify_flags(EVENT_FLAG_ITEM_MODIFIED), + Some(EventClass::Change) + ); + assert_eq!( + classify_flags(EVENT_FLAG_ITEM_XATTR_MOD), + Some(EventClass::Change) + ); + assert_eq!( + classify_flags(EVENT_FLAG_ITEM_REMOVED), + Some(EventClass::Rename) + ); + assert_eq!( + classify_flags(EVENT_FLAG_ROOT_CHANGED), + Some(EventClass::Rename) + ); + assert_eq!(classify_flags(0x0001 /* MustScanSubDirs */), None); + assert_eq!(classify_flags(0x0001_0000 /* ItemIsFile alone */), None); + } + + #[test] + fn the_framework_binds_at_runtime() { + // No `-framework CoreServices` on any link line: the ten entry points + // resolve through dlopen/dlsym on the running system. + let api = api().expect("CoreServices + CoreFoundation resolve via dlopen"); + assert!(!api.cf_type_array_callbacks.is_null()); + } +} diff --git a/crates/perry-runtime/src/lib.rs b/crates/perry-runtime/src/lib.rs index 8f6f32931c..e4e3016db2 100644 --- a/crates/perry-runtime/src/lib.rs +++ b/crates/perry-runtime/src/lib.rs @@ -404,6 +404,39 @@ pub(crate) mod stdlib_pump { } } + // #9591: the has-active counterpart of `RUNTIME_PUMP_FNS` — armed slots + // for runtime-internal subsystems whose live handles must keep the loop + // alive, with the same reason for the indirection: a direct call from + // `js_stdlib_has_active_handles` into the fs.watch backend would pin the + // OS watcher (and its notify dependency) into every binary. + const RUNTIME_HAS_ACTIVE_SLOTS: usize = 4; + static RUNTIME_HAS_ACTIVE_FNS: [AtomicPtr<()>; RUNTIME_HAS_ACTIVE_SLOTS] = [ + AtomicPtr::new(null_mut()), + AtomicPtr::new(null_mut()), + AtomicPtr::new(null_mut()), + AtomicPtr::new(null_mut()), + ]; + + /// Arm runtime has-active slot `slot` with `f` (see `register_runtime_pump`). + pub(crate) fn register_runtime_has_active(slot: usize, f: extern "C" fn() -> i32) { + if slot >= RUNTIME_HAS_ACTIVE_SLOTS { + return; + } + RUNTIME_HAS_ACTIVE_FNS[slot].store(std::hint::black_box(f as *mut ()), Ordering::Release); + } + + fn run_runtime_has_active() -> bool { + RUNTIME_HAS_ACTIVE_FNS.iter().any(|slot| { + let p = slot.load(Ordering::Acquire); + if p.is_null() { + return false; + } + // SAFETY: only `extern "C" fn() -> i32` values are ever stored. + let func: extern "C" fn() -> i32 = unsafe { std::mem::transmute(p) }; + func() != 0 + }) + } + // #2532 — auxiliary pump / has-active registries. // // perry-stdlib owns the single `STDLIB_PUMP_FN` slot above and drains @@ -588,6 +621,11 @@ pub(crate) mod stdlib_pump { if crate::os::js_process_signal_has_active() != 0 { return 1; } + // #9591: a ref'd `fs.watch` handle / started `fsPromises.watch` + // iterator (armed slot, see `register_runtime_has_active`). + if run_runtime_has_active() { + return 1; + } // #2532 — a live `perry-ext-*` handle (e.g. a listening HTTP // server registered out-of-tree) keeps the loop alive even when // perry-stdlib reports none. @@ -648,6 +686,41 @@ pub(crate) mod stdlib_pump { ); } + static HAS_ACTIVE_FLAG: AtomicI32 = AtomicI32::new(0); + extern "C" fn flag_has_active() -> i32 { + HAS_ACTIVE_FLAG.load(AtomicOrdering::SeqCst) + } + + /// #9591: an armed runtime has-active slot keeps the loop alive + /// exactly while its callback reports live work — the fs.watch + /// backend's liveness reaches the generated event loop through + /// this slot, not through a timer. + #[test] + fn runtime_has_active_slot_gates_the_loop() { + crate::os::test_set_stdin_data_listener(None); + register_runtime_has_active(RUNTIME_HAS_ACTIVE_SLOTS - 1, flag_has_active); + HAS_ACTIVE_FLAG.store(0, AtomicOrdering::SeqCst); + assert_eq!( + js_stdlib_has_active_handles(), + 0, + "an armed slot reporting 0 must not pin the loop" + ); + HAS_ACTIVE_FLAG.store(1, AtomicOrdering::SeqCst); + assert_eq!( + js_stdlib_has_active_handles(), + 1, + "an armed slot reporting live work must keep the loop alive" + ); + HAS_ACTIVE_FLAG.store(0, AtomicOrdering::SeqCst); + assert_eq!( + js_stdlib_has_active_handles(), + 0, + "the loop is released again once the work is gone" + ); + // Out-of-range slots are ignored, never a panic. + register_runtime_has_active(RUNTIME_HAS_ACTIVE_SLOTS, flag_has_active); + } + #[test] fn aux_pump_registration_is_idempotent() { // Registering the same fn pointer repeatedly stores it once, diff --git a/crates/perry/tests/issue_9591_fs_watch_native_events.rs b/crates/perry/tests/issue_9591_fs_watch_native_events.rs new file mode 100644 index 0000000000..9de590d702 --- /dev/null +++ b/crates/perry/tests/issue_9591_fs_watch_native_events.rs @@ -0,0 +1,245 @@ +//! Regression test for #9591: `fs.watch` must not re-walk its target on a +//! timer. +//! +//! Before the fix every watcher was a 25 ms `setInterval` whose tick walked +//! the WHOLE watch target (`read_dir` + `symlink_metadata` per entry) and +//! diffed two maps, on the main thread — ~3.4 µs per file per tick, 40 ticks +//! a second. Measured over 5 s of watching 3 000 files: 2.03 s of CPU (41 % +//! of a core) against Node's 0.03 s. claude-code watches its cwd; a 362 k-file +//! cwd extrapolates to ~1.2 s of walking per 25 ms schedule, i.e. a wedged +//! event loop. +//! +//! The fix hands change detection to the OS (`notify`: inotify / FSEvents / +//! ReadDirectoryChangesW) and keeps the walker only as an off-main-thread +//! fallback paced to 5 % of a core. This test is the issue's verification +//! bar: watch 3 000 files for a window, assert the process burned less than +//! 5 % of a core doing it, AND that a change is still seen promptly — both +//! halves matter, since a poller with a long enough interval would pass the +//! CPU bound alone. The unfixed walker burns ~1.6 s in this window; the +//! native backend burns a few milliseconds. +//! +//! The second test forces the fallback (`PERRY_FS_WATCH_POLL=1`) and checks +//! that it, too, respects the budget: at 3 000 files a walk is ~10 ms, so the +//! adaptive interval sits near 200 ms and the duty cycle at 5 %. + +use std::io::{BufRead, BufReader}; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::mpsc::{self, Receiver}; +use std::time::{Duration, Instant}; + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +const FILE_COUNT: usize = 3_000; +const DIR_COUNT: usize = 30; +const WINDOW_MS: u64 = 4_000; + +/// Watches argv[2] recursively, reports every event, and after argv[3] ms +/// prints the CPU it consumed WHILE watching (startup excluded), then closes +/// the watcher so the loop is free to exit. +const SOURCE: &str = r#" +import fs from "node:fs"; +const dir = process.argv[2]; +const windowMs = Number(process.argv[3]); +const watcher = fs.watch(dir, { recursive: true }, (eventType: string, filename: any) => { + console.log("EVENT:" + eventType + ":" + String(filename).replace(/\\/g, "/")); +}); +const start = process.cpuUsage(); +console.log("READY"); +setTimeout(() => { + const used = process.cpuUsage(start); + console.log("CPU_MS:" + ((used.user + used.system) / 1000).toFixed(1)); + watcher.close(); + console.log("CLOSED"); +}, windowMs); +"#; + +fn compile(dir: &Path, source: &str) -> PathBuf { + let entry = dir.join("main.ts"); + let output = dir.join("main_bin"); + std::fs::write(&entry, source).expect("write entry"); + let compile = Command::new(perry_bin()) + .current_dir(dir) + .arg("compile") + .arg(&entry) + .arg("-o") + .arg(&output) + .output() + .expect("run perry compile"); + assert!( + compile.status.success(), + "perry compile failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&compile.stdout), + String::from_utf8_lossy(&compile.stderr) + ); + output +} + +fn build_tree(root: &Path) { + for d in 0..DIR_COUNT { + let dir = root.join(format!("sub{d:02}")); + std::fs::create_dir_all(&dir).expect("create subdir"); + for f in 0..(FILE_COUNT / DIR_COUNT) { + std::fs::write(dir.join(format!("f{f:03}.txt")), "x").expect("write file"); + } + } +} + +struct Fixture { + child: Child, + lines: Receiver<(String, Instant)>, +} + +fn spawn_fixture(bin: &Path, tree: &Path, force_poll: bool) -> Fixture { + let mut command = Command::new(bin); + command + .arg(tree) + .arg(WINDOW_MS.to_string()) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()); + if force_poll { + command.env("PERRY_FS_WATCH_POLL", "1"); + } else { + command.env_remove("PERRY_FS_WATCH_POLL"); + } + let mut child = command.spawn().expect("spawn compiled binary"); + let stdout = child.stdout.take().expect("piped stdout"); + let (tx, rx) = mpsc::channel::<(String, Instant)>(); + std::thread::spawn(move || { + for line in BufReader::new(stdout).lines() { + match line { + Ok(l) => { + if tx.send((l, Instant::now())).is_err() { + break; + } + } + Err(_) => break, + } + } + }); + Fixture { child, lines: rx } +} + +impl Fixture { + fn wait_line( + &self, + predicate: impl Fn(&str) -> bool, + timeout: Duration, + what: &str, + ) -> (String, Instant) { + let deadline = Instant::now() + timeout; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + let (line, at) = self + .lines + .recv_timeout(remaining) + .unwrap_or_else(|_| panic!("no {what} line within {timeout:?}")); + if predicate(&line) { + return (line, at); + } + } + } +} + +struct Outcome { + cpu_ms: f64, + detect_latency: Duration, +} + +fn run_watch_window(label: &str, force_poll: bool) -> Outcome { + let dir = std::env::temp_dir().join(format!("perry_9591_{label}_{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create temp dir"); + let tree = dir.join("tree"); + build_tree(&tree); + let bin = compile(&dir, SOURCE); + + let mut fixture = spawn_fixture(&bin, &tree, force_poll); + fixture.wait_line(|l| l == "READY", Duration::from_secs(30), "READY"); + + // Let the loop settle, then make one change and time its arrival. + std::thread::sleep(Duration::from_millis(500)); + let probe = tree.join("sub05").join("probe_new_file.txt"); + let touched = Instant::now(); + std::fs::write(&probe, "hello").expect("write probe file"); + let (_, seen_at) = fixture.wait_line( + |l| l.starts_with("EVENT:") && l.ends_with("sub05/probe_new_file.txt"), + Duration::from_secs(10), + "EVENT for the probe file", + ); + let detect_latency = seen_at.duration_since(touched); + + let (cpu_line, _) = fixture.wait_line( + |l| l.starts_with("CPU_MS:"), + Duration::from_millis(WINDOW_MS + 15_000), + "CPU_MS", + ); + let cpu_ms: f64 = cpu_line["CPU_MS:".len()..].parse().expect("parse CPU_MS"); + fixture.wait_line(|l| l == "CLOSED", Duration::from_secs(10), "CLOSED"); + + // With the watcher closed nothing keeps the loop alive: the process must + // exit on its own (the old interval timer was ref'd; the new liveness + // slot must release the loop the same way). + let exit_deadline = Instant::now() + Duration::from_secs(10); + let status = loop { + if let Some(status) = fixture.child.try_wait().expect("try_wait") { + break status; + } + assert!( + Instant::now() < exit_deadline, + "{label}: the process did not exit after watcher.close() — a closed \ + watcher still reports itself active to the event loop" + ); + std::thread::sleep(Duration::from_millis(20)); + }; + assert!(status.success(), "{label}: fixture exited with {status}"); + let _ = std::fs::remove_dir_all(&dir); + + eprintln!("{label}: cpu {cpu_ms:.1} ms over {WINDOW_MS} ms window, change seen after {detect_latency:?}"); + Outcome { + cpu_ms, + detect_latency, + } +} + +#[test] +fn native_watch_of_three_thousand_files_is_idle_and_prompt() { + let outcome = run_watch_window("native", false); + // 5 % of the window. The pre-fix walker lands around 1 600 ms here. + let budget_ms = WINDOW_MS as f64 * 0.05; + assert!( + outcome.cpu_ms < budget_ms, + "fs.watch burned {:.1} ms of CPU over a {WINDOW_MS} ms window watching {FILE_COUNT} files \ + (budget {budget_ms:.0} ms) — the watcher is walking the tree on a timer again", + outcome.cpu_ms + ); + assert!( + outcome.detect_latency < Duration::from_secs(1), + "a new file took {:?} to surface — OS events deliver in milliseconds", + outcome.detect_latency + ); +} + +#[test] +fn poll_fallback_paces_itself_to_the_budget() { + let outcome = run_watch_window("poll", true); + // The walker sleeps 20 × its own duration between passes, so its duty + // cycle is ≤ 5 % plus the granularity of one walk. Allow 12.5 % so a + // loaded CI box cannot flake it; the pre-fix 25 ms cadence is 41 %. + let budget_ms = WINDOW_MS as f64 * 0.125; + assert!( + outcome.cpu_ms < budget_ms, + "the poll fallback burned {:.1} ms of CPU over a {WINDOW_MS} ms window watching {FILE_COUNT} \ + files (budget {budget_ms:.0} ms) — its interval is not scaling with the walk", + outcome.cpu_ms + ); + // ~10 ms walk ⇒ ~200 ms interval; well inside 3 s even on a slow disk. + assert!( + outcome.detect_latency < Duration::from_secs(3), + "the poll fallback took {:?} to see a new file", + outcome.detect_latency + ); +} diff --git a/test-files/test_gap_9591_fs_watch_events.ts b/test-files/test_gap_9591_fs_watch_events.ts new file mode 100644 index 0000000000..78d6a7efe7 --- /dev/null +++ b/test-files/test_gap_9591_fs_watch_events.ts @@ -0,0 +1,117 @@ +// #9591: fs.watch / fsPromises.watch are driven by OS change notifications +// (inotify / FSEvents / ReadDirectoryChangesW), not by a 25 ms timer that +// re-walked the whole tree. This fixture pins the observable contract that +// every backend shares with node: a change inside the watched target surfaces +// as an event naming that entry, relative to the root, on the callback +// watcher, on a single-file watcher, and on the promise-based async iterator. +// +// Only what every platform agrees on is printed. Event TYPES differ between +// inotify and FSEvents even under node (a create is one 'rename' on Linux, +// possibly 'rename' + 'change' on macOS), so the first event naming the file +// each phase creates is what gets reported — never the type or the count. +import fs from "node:fs"; +import fsp from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; + +const root = fs.mkdtempSync(path.join(os.tmpdir(), "perry-9591-")); +const TIMEOUT_MS = 5000; + +// libuv arms an FSEvents stream asynchronously (on its CF run-loop thread), so +// under node on macOS a write issued in the same tick as fs.watch() can land +// before the stream exists and never be reported. Give every new watcher a +// moment to arm before the change it is meant to see; inotify needs none of +// this, and the cost is a few hundred milliseconds of wall time. +const settle = () => new Promise((resolve) => setTimeout(resolve, 250)); + +function firstEventNamed(watcher: any, expected: string): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error("timeout waiting for " + expected)), + TIMEOUT_MS, + ); + watcher.on("change", (_eventType: string, filename: any) => { + const name = String(filename).replace(/\\/g, "/"); + if (name === expected) { + clearTimeout(timer); + resolve(name); + } + }); + }); +} + +async function main() { + // 1. Non-recursive directory watch: a new direct child. + { + const watcher = fs.watch(root); + const seen = firstEventNamed(watcher, "a.txt"); + await settle(); + fs.writeFileSync(path.join(root, "a.txt"), "one"); + console.log("watch:", await seen); + watcher.close(); + } + + // 2. Recursive watch: a file two levels down reports its relative path. + { + fs.mkdirSync(path.join(root, "sub")); + const watcher = fs.watch(root, { recursive: true }); + const seen = firstEventNamed(watcher, "sub/b.txt"); + await settle(); + fs.writeFileSync(path.join(root, "sub", "b.txt"), "two"); + console.log("recursive:", await seen); + watcher.close(); + } + + // 3. A single-file watch reports the file's own name. + { + const file = path.join(root, "a.txt"); + const watcher = fs.watch(file); + const seen = firstEventNamed(watcher, "a.txt"); + await settle(); + fs.writeFileSync(file, "one-more"); + console.log("file:", await seen); + watcher.close(); + } + + // 4. fsPromises.watch: the async iterator yields { eventType, filename }. + // The OS watch starts with the first next(), so the write is scheduled + // for after iteration begins (and after the stream has armed, as above). + { + const ac = new AbortController(); + const guard = setTimeout(() => ac.abort(), TIMEOUT_MS); + setTimeout(() => fs.writeFileSync(path.join(root, "c.txt"), "three"), 250); + try { + for await (const event of fsp.watch(root, { signal: ac.signal })) { + const name = String(event.filename).replace(/\\/g, "/"); + if (name === "c.txt") { + console.log("promises:", name); + break; + } + } + } catch (err: any) { + console.log("promises: aborted", err && err.name); + } finally { + clearTimeout(guard); + } + } + + // 5. Close is idempotent and 'close' fires once. + { + const watcher = fs.watch(root); + let closes = 0; + watcher.on("close", () => { + closes++; + }); + watcher.close(); + watcher.close(); + await new Promise((r) => setTimeout(r, 20)); + console.log("close events:", closes); + } + + fs.rmSync(root, { recursive: true, force: true }); + console.log("done"); +} + +main().catch((err) => { + console.log("failed:", err && err.message); +});