diff --git a/MODULE.bazel b/MODULE.bazel index e513a2ddce..06405ca6fb 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -408,6 +408,14 @@ crate.annotation_select( ], ) +# Backport notify-rs/notify@a9c51183: wake idle Windows watchers on events and control requests. +crate.annotation( + crate = "notify", + patch_args = ["-p1"], + patches = ["//patches:notify_windows_idle.patch"], + version = "8.2.0", +) + # Match the CodeModeHost response-compression benchmark's zstd level. crate.annotation( crate = "tonic", diff --git a/codex-rs/file-watcher/src/file_watcher_tests.rs b/codex-rs/file-watcher/src/file_watcher_tests.rs index ecf54cabd1..84caf8d082 100644 --- a/codex-rs/file-watcher/src/file_watcher_tests.rs +++ b/codex-rs/file-watcher/src/file_watcher_tests.rs @@ -296,6 +296,35 @@ fn recursive_registration_downgrades_to_non_recursive_after_drop() { drop(non_recursive); } +#[cfg(windows)] +#[tokio::test] +async fn native_watcher_reports_changes_after_unwatch_and_rewatch() { + let temp_dir = tempfile::tempdir().expect("temp dir"); + let watcher = Arc::new(FileWatcher::new().expect("watcher")); + let (subscriber, mut rx) = watcher.add_subscriber(); + + for directory in ["first", "second"] { + let root = temp_dir.path().join(directory); + std::fs::create_dir(&root).expect("create watched directory"); + let registration = subscriber.register_path(root.clone(), /*recursive*/ false); + assert!(registration.registration_succeeded()); + + let changed_path = root.join("changed.txt"); + std::fs::write(&changed_path, "new content").expect("write watched file"); + let event = timeout(Duration::from_secs(10), rx.recv()) + .await + .expect("native watch event timeout"); + assert_eq!( + event, + Some(FileWatcherEvent { + paths: vec![changed_path] + }) + ); + + drop(registration); + } +} + #[test] fn unregister_holds_state_lock_until_unwatch_finishes() { let temp_dir = tempfile::tempdir().expect("temp dir"); diff --git a/patches/BUILD.bazel b/patches/BUILD.bazel index 7d6983a658..6fad1f052b 100644 --- a/patches/BUILD.bazel +++ b/patches/BUILD.bazel @@ -22,6 +22,7 @@ exports_files([ "ring_windows_msvc_include_dirs.patch", "tikv-jemalloc-sys_make_env.patch", "tikv-jemalloc-sys_dependency_flags.patch", + "notify_windows_idle.patch", "tonic-zstd-level1.patch", "v8_array_sort_elements_kind.patch", "v8_source_portability.patch", diff --git a/patches/notify_windows_idle.patch b/patches/notify_windows_idle.patch new file mode 100644 index 0000000000..feef188e6a --- /dev/null +++ b/patches/notify_windows_idle.patch @@ -0,0 +1,107 @@ +--- a/src/windows.rs ++++ b/src/windows.rs +@@ -14,6 +14,7 @@ + use std::ffi::OsString; + use std::os::raw::c_void; + use std::os::windows::ffi::{OsStrExt, OsStringExt}; ++use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; + use std::path::{Path, PathBuf}; + use std::ptr; + use std::slice; +@@ -86,7 +87,7 @@ + meta_tx: Sender, + cmd_tx: Sender>, + watches: HashMap, +- wakeup_sem: HANDLE, ++ wakeup_sem: Arc, + } + + impl ReadDirectoryChangesServer { +@@ -94,17 +95,14 @@ + event_handler: Arc>, + meta_tx: Sender, + cmd_tx: Sender>, +- wakeup_sem: HANDLE, ++ wakeup_sem: Arc, + ) -> Sender { + let (action_tx, action_rx) = unbounded(); +- // it is, in fact, ok to send the semaphore across threads +- let sem_temp = wakeup_sem as u64; + let _ = thread::Builder::new() + .name("notify-rs windows loop".to_string()) + .spawn({ + let tx = action_tx.clone(); + move || { +- let wakeup_sem = sem_temp as HANDLE; + let server = ReadDirectoryChangesServer { + tx, + rx: action_rx, +@@ -148,20 +146,14 @@ + if stopped { + break; + } +- + unsafe { + // wait with alertable flag so that the completion routine fires +- let waitres = WaitForSingleObjectEx(self.wakeup_sem, 100, 1); ++ let waitres = WaitForSingleObjectEx(self.wakeup_sem.as_raw_handle(), INFINITE, 1); + if waitres == WAIT_OBJECT_0 { + let _ = self.meta_tx.send(MetaEvent::WatcherAwakened); + } + } + } +- +- // we have to clean this up, since the watcher may be long gone +- unsafe { +- CloseHandle(self.wakeup_sem); +- } + } + + fn add_watch(&mut self, path: PathBuf, is_recursive: bool) -> Result { +@@ -468,7 +460,7 @@ + pub struct ReadDirectoryChangesWatcher { + tx: Sender, + cmd_rx: Receiver>, +- wakeup_sem: HANDLE, ++ wakeup_sem: Arc, + } + + impl ReadDirectoryChangesWatcher { +@@ -483,8 +475,13 @@ + return Err(Error::generic("Failed to create wakeup semaphore.")); + } + +- let action_tx = +- ReadDirectoryChangesServer::start(event_handler, meta_tx, cmd_tx, wakeup_sem); ++ let wakeup_sem = Arc::new(unsafe { OwnedHandle::from_raw_handle(wakeup_sem) }); ++ let action_tx = ReadDirectoryChangesServer::start( ++ event_handler, ++ meta_tx, ++ cmd_tx, ++ Arc::clone(&wakeup_sem), ++ ); + + Ok(ReadDirectoryChangesWatcher { + tx: action_tx, +@@ -494,11 +491,9 @@ + } + + fn wakeup_server(&mut self) { +- // breaks the server out of its wait state. right now this is really just an optimization, +- // so that if you add a watch you don't block for 100ms in watch() while the +- // server sleeps. ++ // Signal after enqueueing the action so the server can sleep between changes. + unsafe { +- ReleaseSemaphore(self.wakeup_sem, 1, ptr::null_mut()); ++ ReleaseSemaphore(self.wakeup_sem.as_raw_handle(), 1, ptr::null_mut()); + } + } + +@@ -579,6 +574,7 @@ + fn configure(&mut self, config: Config) -> Result { + let (tx, rx) = bounded(1); + self.tx.send(Action::Configure(config, tx))?; ++ self.wakeup_server(); + rx.recv()? + } +