Avoid periodic wakeups in idle Windows file watchers (#60560)

## Why

The Windows `notify` backend wakes every 100 ms even when idle.

## What changed

- Patch `notify` 8.2.0 in Bazel to wait indefinitely for events or control requests.
- Wake the watcher when configuration requests are queued and share ownership of the wakeup semaphore with `Arc<OwnedHandle>`.

## Testing

Add a Windows regression test that verifies file changes are reported after unregistering one directory and registering another on the same watcher.

GitOrigin-RevId: b53b3c8bc514008554fdeffaef3f7bf8305a4803

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",

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");

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",

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<MetaEvent>,
+     cmd_tx: Sender<Result<PathBuf>>,
+     watches: HashMap<PathBuf, WatchState>,
+-    wakeup_sem: HANDLE,
++    wakeup_sem: Arc<OwnedHandle>,
+ }
+ 
+ impl ReadDirectoryChangesServer {
+@@ -94,17 +95,14 @@
+         event_handler: Arc<Mutex<dyn EventHandler>>,
+         meta_tx: Sender<MetaEvent>,
+         cmd_tx: Sender<Result<PathBuf>>,
+-        wakeup_sem: HANDLE,
++        wakeup_sem: Arc<OwnedHandle>,
+     ) -> Sender<Action> {
+         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<PathBuf> {
+@@ -468,7 +460,7 @@
+ pub struct ReadDirectoryChangesWatcher {
+     tx: Sender<Action>,
+     cmd_rx: Receiver<Result<PathBuf>>,
+-    wakeup_sem: HANDLE,
++    wakeup_sem: Arc<OwnedHandle>,
+ }
+ 
+ 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<bool> {
+         let (tx, rx) = bounded(1);
+         self.tx.send(Action::Configure(config, tx))?;
++        self.wakeup_server();
+         rx.recv()?
+     }
+