windows: Fix server hangs under some circumstance (notify-rs#674)

* Add coresponding test

* fix

* simplify example

* fmt

* clippy

* Even simpler

* Better fix

* Refactor

* Fix

* Fix merge conflicts

* Fix

* add changelog

* fix ci

* lower case

* typo

Author: JunkuiZhang

CHANGELOG.md

@@ -4,6 +4,7 @@
 - FEATURE: added support for the [`flume`](https://docs.rs/flume) crate
 - FIX: kqueue-backend: do not double unwatch top-level directory when recursively unwatching [#683]
 - FIX: Return the crate error `PathNotFound` instead bubbling up the std::io error  [#685] **breaking**
+- FIX: fix server hangs when trashing folders on Windows [#674]
 
 ## debouncer-full 0.6.0 (unreleased)
 - FEATURE: allow `FileIdCache` trait implementations to choose ownership of the returned file-ids

Cargo.toml

@@ -46,5 +46,6 @@ rstest = "0.24.0"
 serde = { version = "1.0.89", features = ["derive"] }
 serde_json = "1.0.39"
 tempfile = "3.10.0"
+trash = "5.2.2"
 walkdir = "2.4.0"
 windows-sys = "0.59.0"

notify/Cargo.toml

@@ -55,3 +55,6 @@ serde_json.workspace = true
 tempfile.workspace = true
 nix.workspace = true
 insta.workspace = true
+
+[target.'cfg(target_os = "windows")'.dev-dependencies]
+trash.workspace = true

notify/src/lib.rs

@@ -459,4 +459,23 @@ mod tests {
 
         panic!("did not receive expected event");
     }
+
+    #[test]
+    #[cfg(target_os = "windows")]
+    fn test_windows_trash_dir() -> std::result::Result<(), Box<dyn std::error::Error>> {
+        let dir = tempdir()?;
+        let child_dir = dir.path().join("child");
+        fs::create_dir(&child_dir)?;
+
+        let mut watcher = recommended_watcher(|_| {
+            // Do something with the event
+        })?;
+        watcher.watch(&child_dir, RecursiveMode::NonRecursive)?;
+
+        trash::delete(&child_dir)?;
+
+        watcher.watch(dir.path(), RecursiveMode::NonRecursive)?;
+
+        Ok(())
+    }
 }

notify/src/windows.rs

@@ -20,7 +20,8 @@ use std::slice;
 use std::sync::{Arc, Mutex};
 use std::thread;
 use windows_sys::Win32::Foundation::{
-    CloseHandle, ERROR_OPERATION_ABORTED, HANDLE, INVALID_HANDLE_VALUE, WAIT_OBJECT_0,
+    CloseHandle, ERROR_ACCESS_DENIED, ERROR_OPERATION_ABORTED, ERROR_SUCCESS, HANDLE,
+    INVALID_HANDLE_VALUE, WAIT_OBJECT_0,
 };
 use windows_sys::Win32::Storage::FileSystem::{
     CreateFileW, ReadDirectoryChangesW, FILE_ACTION_ADDED, FILE_ACTION_MODIFIED,
@@ -51,6 +52,13 @@ struct ReadDirectoryRequest {
     buffer: [u8; BUF_SIZE as usize],
     handle: HANDLE,
     data: ReadData,
+    action_tx: Sender<Action>,
+}
+
+impl ReadDirectoryRequest {
+    fn unwatch(&self) {
+        let _ = self.action_tx.send(Action::Unwatch(self.data.dir.clone()));
+    }
 }
 
 enum Action {
@@ -72,6 +80,7 @@ struct WatchState {
 }
 
 struct ReadDirectoryChangesServer {
+    tx: Sender<Action>,
     rx: Receiver<Action>,
     event_handler: Arc<Mutex<dyn EventHandler>>,
     meta_tx: Sender<MetaEvent>,
@@ -92,17 +101,21 @@ impl ReadDirectoryChangesServer {
         let sem_temp = wakeup_sem as u64;
         let _ = thread::Builder::new()
             .name("notify-rs windows loop".to_string())
-            .spawn(move || {
-                let wakeup_sem = sem_temp as HANDLE;
-                let server = ReadDirectoryChangesServer {
-                    rx: action_rx,
-                    event_handler,
-                    meta_tx,
-                    cmd_tx,
-                    watches: HashMap::new(),
-                    wakeup_sem,
-                };
-                server.run();
+            .spawn({
+                let tx = action_tx.clone();
+                move || {
+                    let wakeup_sem = sem_temp as HANDLE;
+                    let server = ReadDirectoryChangesServer {
+                        tx,
+                        rx: action_rx,
+                        event_handler,
+                        meta_tx,
+                        cmd_tx,
+                        watches: HashMap::new(),
+                        wakeup_sem,
+                    };
+                    server.run();
+                }
             });
         action_tx
     }
@@ -206,7 +219,7 @@ impl ReadDirectoryChangesServer {
         };
         // every watcher gets its own semaphore to signal completion
         let semaphore = unsafe { CreateSemaphoreW(ptr::null_mut(), 0, 1, ptr::null_mut()) };
-        if semaphore == ptr::null_mut() || semaphore == INVALID_HANDLE_VALUE {
+        if semaphore.is_null() || semaphore == INVALID_HANDLE_VALUE {
             unsafe {
                 CloseHandle(handle);
             }
@@ -223,7 +236,7 @@ impl ReadDirectoryChangesServer {
             complete_sem: semaphore,
         };
         self.watches.insert(path.clone(), ws);
-        start_read(&rd, self.event_handler.clone(), handle);
+        start_read(&rd, self.event_handler.clone(), handle, self.tx.clone());
         Ok(path)
     }
 
@@ -254,12 +267,18 @@ fn stop_watch(ws: &WatchState, meta_tx: &Sender<MetaEvent>) {
     let _ = meta_tx.send(MetaEvent::SingleWatchComplete);
 }
 
-fn start_read(rd: &ReadData, event_handler: Arc<Mutex<dyn EventHandler>>, handle: HANDLE) {
+fn start_read(
+    rd: &ReadData,
+    event_handler: Arc<Mutex<dyn EventHandler>>,
+    handle: HANDLE,
+    action_tx: Sender<Action>,
+) {
     let request = Box::new(ReadDirectoryRequest {
         event_handler,
         handle,
         buffer: [0u8; BUF_SIZE as usize],
         data: rd.clone(),
+        action_tx,
     });
 
     let flags = FILE_NOTIFY_CHANGE_FILE_NAME
@@ -317,15 +336,44 @@ unsafe extern "system" fn handle_event(
     let overlapped: Box<OVERLAPPED> = Box::from_raw(overlapped);
     let request: Box<ReadDirectoryRequest> = Box::from_raw(overlapped.hEvent as *mut _);
 
-    if error_code == ERROR_OPERATION_ABORTED {
-        // received when dir is unwatched or watcher is shutdown; return and let overlapped/request
-        // get drop-cleaned
-        ReleaseSemaphore(request.data.complete_sem, 1, ptr::null_mut());
-        return;
+    match error_code {
+        ERROR_OPERATION_ABORTED => {
+            // received when dir is unwatched or watcher is shutdown; return and let overlapped/request get drop-cleaned
+            ReleaseSemaphore(request.data.complete_sem, 1, ptr::null_mut());
+            return;
+        }
+        ERROR_ACCESS_DENIED => {
+            // This could happen when the watched directory is deleted or trashed, first check if it's the case.
+            // If so, unwatch the directory and return, otherwise, continue to handle the event.
+            if !request.data.dir.exists() {
+                request.unwatch();
+                ReleaseSemaphore(request.data.complete_sem, 1, ptr::null_mut());
+                return;
+            }
+        }
+        ERROR_SUCCESS => {
+            // Success, continue to handle the event
+        }
+        _ => {
+            // Some unidentified error occurred, log and unwatch the directory, then return.
+            log::error!(
+                "unknown error in ReadDirectoryChangesW for directory {}: {}",
+                request.data.dir.display(),
+                error_code
+            );
+            request.unwatch();
+            ReleaseSemaphore(request.data.complete_sem, 1, ptr::null_mut());
+            return;
+        }
     }
 
     // Get the next request queued up as soon as possible
-    start_read(&request.data, request.event_handler.clone(), request.handle);
+    start_read(
+        &request.data,
+        request.event_handler.clone(),
+        request.handle,
+        request.action_tx,
+    );
 
     // The FILE_NOTIFY_INFORMATION struct has a variable length due to the variable length
     // string as its last member. Each struct contains an offset for getting the next entry in
@@ -431,7 +479,7 @@ impl ReadDirectoryChangesWatcher {
         let (cmd_tx, cmd_rx) = unbounded();
 
         let wakeup_sem = unsafe { CreateSemaphoreW(ptr::null_mut(), 0, 1, ptr::null_mut()) };
-        if wakeup_sem == ptr::null_mut() || wakeup_sem == INVALID_HANDLE_VALUE {
+        if wakeup_sem.is_null() || wakeup_sem == INVALID_HANDLE_VALUE {
             return Err(Error::generic("Failed to create wakeup semaphore."));
         }