fix(runtime/ops): unwatch shared RecommendedWatcher on FsWatcher close (#34467)

## Summary

Closing a `Deno.FsWatcher` left its underlying notify watch behind.
Since #26200
the runtime shares one `RecommendedWatcher` across all
`Deno.watchFs(...)` calls
in an OpState, but the close path never told that watcher to unwatch
anything.
On Windows each `Watcher::watch(path, ...)` call registers a separate
`ReadDirectoryChangesW` operation, so opening and closing watchers
repeatedly
on the same path leaks watches whose callbacks all keep firing — the
next
`Deno.watchFs(...)` for that path then sees every event N times.

This PR reference-counts per `(path, recursive_mode)` in the shared
`WatcherInner`: `watch` is only called when the count transitions 0→1,
and
`unwatch` when it transitions 1→0. Each `FsEventsResource` carries an
`Arc<WatcherInner>` plus a unique `id` and the list of `(path, mode)`
pairs it
registered, so its `Drop` impl can both remove its `WatchSender` from
the
shared dispatch list and decrement the per-path refcounts.

The duplicate-event symptom only reproduces on Windows because Linux's
inotify
backend deduplicates `inotify_add_watch` on the same path and macOS's
FSEvents
does the same — the existing leaked-`WatchSender` and leaked-watch state
were
silent there, but they're still incorrect.

Fixes #27742.

Closes denoland/orchid#261

## Test plan

- [x] Added `watchFsCloseAndRecreateNoDuplicates` to
`tests/unit/fs_events_test.ts`
mirroring the reproducer in #27742 (open/close 3 watchers, then a 4th
observes a single fs change). Asserts each `(kind, path)` is reported
at most twice (a real `create` plus an optional `modify`). On the buggy
      build this would be N copies on Windows.
- [x] `cargo build --bin deno`
- [x] `cargo clippy -p deno_runtime`
- [x] All 12 tests in `tests/unit/fs_events_test.ts` pass with the
patched
      binary on Linux.
- [ ] CI to validate Windows.

Co-authored-by: divybot <divybot@users.noreply.github.com>
Co-authored-by: Divy Srivastava <me@littledivy.com>

Author: divybot

runtime/ops/fs_events.rs

@@ -2,11 +2,14 @@
 
 use std::borrow::Cow;
 use std::cell::RefCell;
+use std::collections::HashMap;
 use std::convert::From;
 use std::path::Path;
 use std::path::PathBuf;
 use std::rc::Rc;
 use std::sync::Arc;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering;
 
 use deno_core::AsyncRefCell;
 use deno_core::CancelFuture;
@@ -38,6 +41,13 @@ deno_core::extension!(
 struct FsEventsResource {
   receiver: AsyncRefCell<mpsc::Receiver<Result<FsEvent, NotifyError>>>,
   cancel: CancelHandle,
+  /// Shared backend used to clean up our watch on drop.
+  inner: Arc<WatcherInner>,
+  /// Identifies our `WatchSender` entry in [`WatcherInner::senders`].
+  id: u64,
+  /// The (path, recursive_mode) pairs this resource registered. Tracked so the
+  /// shared watcher unwatches them when the last interested resource closes.
+  watched: Vec<(PathBuf, RecursiveMode)>,
 }
 
 impl Resource for FsEventsResource {
@@ -50,6 +60,36 @@ impl Resource for FsEventsResource {
   }
 }
 
+impl Drop for FsEventsResource {
+  fn drop(&mut self) {
+    // Remove this resource's sender from the shared dispatch list so the
+    // watcher callback stops trying to deliver events to a dead channel.
+    self.inner.senders.lock().retain(|ws| ws.id != self.id);
+
+    // Reference-count the underlying watches: only unwatch when no other
+    // resource still depends on this path. Without this, calling
+    // `Deno.watchFs(path)` repeatedly leaks watches in the shared
+    // `RecommendedWatcher` — on Windows each leaked watch registers a
+    // separate `ReadDirectoryChangesW` request, so the next watcher created
+    // for the same path receives every event N times.
+    let mut watched_paths = self.inner.watched_paths.lock();
+    let mut watcher = self.inner.watcher.lock();
+    for (path, mode) in &self.watched {
+      let key = (path.clone(), *mode);
+      let Some(count) = watched_paths.get_mut(&key) else {
+        continue;
+      };
+      *count = count.saturating_sub(1);
+      if *count == 0 {
+        watched_paths.remove(&key);
+        // Best-effort: ignore errors (e.g. the watcher already lost track of
+        // the path because the path was deleted).
+        let _ = watcher.unwatch(path);
+      }
+    }
+  }
+}
+
 /// Represents a file system event.
 ///
 /// We do not use the event directly from the notify crate. We flatten
@@ -94,6 +134,9 @@ impl From<NotifyEvent> for FsEvent {
 }
 
 struct WatchSender {
+  /// Unique identifier so the matching entry can be removed when the
+  /// owning [`FsEventsResource`] is dropped.
+  id: u64,
   /// Original paths as provided by the caller.
   paths: Vec<PathBuf>,
   /// Pre-canonicalized versions of `paths`, computed once at watch time
@@ -103,9 +146,24 @@ struct WatchSender {
   sender: mpsc::Sender<Result<FsEvent, NotifyError>>,
 }
 
-struct WatcherState {
+struct WatcherInner {
+  /// Per-watcher dispatch list. The shared notify callback iterates this
+  /// list on every event. Wrapped in `Arc<Mutex<...>>` so the watcher
+  /// callback (which lives on the notify backend thread) can hold a
+  /// reference without keeping the rest of [`WatcherInner`] alive.
   senders: Arc<Mutex<Vec<WatchSender>>>,
-  watcher: RecommendedWatcher,
+  /// The shared `RecommendedWatcher` instance backing every
+  /// `Deno.watchFs(...)` call in this OpState.
+  watcher: Mutex<RecommendedWatcher>,
+  /// Reference count per (path, recursive_mode) registered with `watcher`.
+  /// When a count drops to zero, the path is unwatched.
+  watched_paths: Mutex<HashMap<(PathBuf, RecursiveMode), usize>>,
+  /// Monotonic counter for assigning `WatchSender::id`.
+  next_id: AtomicU64,
+}
+
+struct WatcherState {
+  inner: Arc<WatcherInner>,
 }
 
 #[allow(
@@ -215,32 +273,27 @@ pub enum FsEventsError {
 }
 
 fn make_watch_sender(
+  id: u64,
   paths: Vec<PathBuf>,
   sender: mpsc::Sender<Result<FsEvent, NotifyError>>,
 ) -> WatchSender {
   let canonical_paths = paths.iter().map(|p| canonicalize_path(p)).collect();
   WatchSender {
+    id,
     paths,
     canonical_paths,
     sender,
   }
 }
 
-fn start_watcher(
+fn ensure_watcher(
   state: &mut OpState,
-  paths: Vec<PathBuf>,
-  sender: mpsc::Sender<Result<FsEvent, NotifyError>>,
-) -> Result<(), FsEventsError> {
-  if let Some(watcher) = state.try_borrow_mut::<WatcherState>() {
-    watcher
-      .senders
-      .lock()
-      .push(make_watch_sender(paths, sender));
-    return Ok(());
+) -> Result<Arc<WatcherInner>, FsEventsError> {
+  if let Some(ws) = state.try_borrow::<WatcherState>() {
+    return Ok(ws.inner.clone());
   }
 
-  let senders = Arc::new(Mutex::new(vec![make_watch_sender(paths, sender)]));
-
+  let senders: Arc<Mutex<Vec<WatchSender>>> = Arc::new(Mutex::new(Vec::new()));
   let sender_clone = senders.clone();
   let watcher: RecommendedWatcher = Watcher::new(
     move |res: Result<NotifyEvent, NotifyError>| {
@@ -283,9 +336,18 @@ fn start_watcher(
   )
   .map_err(|e| FsEventsError::Notify(JsNotifyError(e)))?;
 
-  state.put::<WatcherState>(WatcherState { watcher, senders });
+  let inner = Arc::new(WatcherInner {
+    senders,
+    watcher: Mutex::new(watcher),
+    watched_paths: Mutex::new(HashMap::new()),
+    next_id: AtomicU64::new(0),
+  });
+
+  state.put::<WatcherState>(WatcherState {
+    inner: inner.clone(),
+  });
 
-  Ok(())
+  Ok(inner)
 }
 
 /// Make `path` absolute and collapse `.` / `..` segments so that paths
@@ -334,28 +396,88 @@ fn op_fs_events_open(
 
   let (sender, receiver) = mpsc::channel::<Result<FsEvent, NotifyError>>(16);
 
-  start_watcher(state, resolved_paths.clone(), sender)?;
+  let inner = ensure_watcher(state)?;
+
+  let id = inner.next_id.fetch_add(1, Ordering::Relaxed);
+
+  inner.senders.lock().push(make_watch_sender(
+    id,
+    resolved_paths.clone(),
+    sender,
+  ));
 
   let recursive_mode = if recursive {
     RecursiveMode::Recursive
   } else {
     RecursiveMode::NonRecursive
   };
-  for path in &resolved_paths {
-    let watcher = state.borrow_mut::<WatcherState>();
-    watcher
-      .watcher
-      .watch(path, recursive_mode)
-      .map_err(|e| FsEventsError::Notify(JsNotifyError(e)))?;
+
+  // Register each path with the shared watcher exactly once per
+  // (path, mode) pair. Subsequent resources requesting the same
+  // (path, mode) bump the refcount but skip the `watch` syscall.
+  // This is the core of the duplicate-event fix on Windows (see
+  // denoland/deno#27742): otherwise repeated `watch` calls register
+  // duplicate ReadDirectoryChangesW operations whose callbacks all
+  // fire on every change.
+  let mut watched = Vec::with_capacity(resolved_paths.len());
+  {
+    let mut watched_paths = inner.watched_paths.lock();
+    let mut watcher = inner.watcher.lock();
+    for path in &resolved_paths {
+      let key = (path.clone(), recursive_mode);
+      let count = watched_paths.entry(key.clone()).or_insert(0);
+      if *count == 0
+        && let Err(e) = watcher.watch(path, recursive_mode)
+      {
+        // Roll back any partial state we accumulated for this call so
+        // a failed open doesn't leave dangling refcounts/senders.
+        watched_paths.remove(&key);
+        drop(watcher);
+        drop(watched_paths);
+        rollback_partial_open(&inner, id, &watched);
+        return Err(FsEventsError::Notify(JsNotifyError(e)));
+      }
+      *count += 1;
+      watched.push((path.clone(), recursive_mode));
+    }
   }
+
   let resource = FsEventsResource {
     receiver: AsyncRefCell::new(receiver),
     cancel: Default::default(),
+    inner,
+    id,
+    watched,
   };
   let rid = state.resource_table.add(resource);
   Ok(rid)
 }
 
+/// Undo the `senders` push and any `watch` calls we performed before
+/// hitting an error in `op_fs_events_open`. Mirrors the cleanup that
+/// would run via [`FsEventsResource`]'s `Drop`, but is needed because
+/// the resource itself was never constructed.
+fn rollback_partial_open(
+  inner: &Arc<WatcherInner>,
+  id: u64,
+  watched: &[(PathBuf, RecursiveMode)],
+) {
+  inner.senders.lock().retain(|ws| ws.id != id);
+
+  let mut watched_paths = inner.watched_paths.lock();
+  let mut watcher = inner.watcher.lock();
+  for (path, mode) in watched {
+    let key = (path.clone(), *mode);
+    if let Some(count) = watched_paths.get_mut(&key) {
+      *count = count.saturating_sub(1);
+      if *count == 0 {
+        watched_paths.remove(&key);
+        let _ = watcher.unwatch(path);
+      }
+    }
+  }
+}
+
 #[op2]
 async fn op_fs_events_poll(
   state: Rc<RefCell<OpState>>,

tests/unit/fs_events_test.ts

@@ -258,6 +258,78 @@ Deno.test(
   },
 );
 
+// Regression test for https://github.com/denoland/deno/issues/27742
+// Closing a `Deno.FsWatcher` and creating a new one for the same path must
+// not produce duplicate events on the new watcher. Before the fix, the
+// shared `RecommendedWatcher` never had paths unwatched on close, so on
+// Windows each closed-then-recreated watcher left behind another
+// `ReadDirectoryChangesW` registration, causing N-fold duplicate events.
+Deno.test(
+  { permissions: { read: true, write: true } },
+  async function watchFsCloseAndRecreateNoDuplicates() {
+    const testDir = await makeTempDir();
+
+    async function openClose() {
+      const w = Deno.watchFs(testDir);
+      const closer = setTimeout(() => w.close(), 100);
+      for await (const _ of w) {
+        // drain
+      }
+      clearTimeout(closer);
+    }
+
+    // Open and close three watchers in sequence.
+    await openClose();
+    await openClose();
+    await openClose();
+
+    // Now open a fourth watcher and trigger a single fs event. We must
+    // only see events for that single change reach us once per real event.
+    using watcher = Deno.watchFs(testDir);
+
+    const target = testDir + "/probe.txt";
+    const writePromise = (async () => {
+      // Give the watcher a moment to settle before producing the event.
+      await delay(100);
+      Deno.writeFileSync(target, new Uint8Array([1, 2, 3]));
+    })();
+
+    // Collect events for a brief window after the write. Any event we
+    // observe must match a real change to probe.txt; we then count how
+    // many times each (kind, path) tuple appears. Filesystems legitimately
+    // emit at most a couple of events for a single write (`create` plus
+    // an optional `modify`), so seeing the same tuple 3+ times means the
+    // bug is still present.
+    const seen = new Map<string, number>();
+    const collectPromise = (async () => {
+      const start = Date.now();
+      for await (const event of watcher) {
+        if (event.paths.some((p) => p.endsWith("probe.txt"))) {
+          const key = `${event.kind}:${event.paths.join(",")}`;
+          seen.set(key, (seen.get(key) ?? 0) + 1);
+        }
+        if (Date.now() - start > 500) break;
+      }
+    })();
+
+    await writePromise;
+    // Wait long enough to receive any duplicates the buggy implementation
+    // would emit.
+    await delay(600);
+    watcher.close();
+    await collectPromise;
+
+    for (const [key, count] of seen) {
+      assert(
+        count <= 2,
+        `event "${key}" was emitted ${count} times; expected at most 2 ` +
+          `(create + optional modify). This indicates a leaked watch from ` +
+          `a previously-closed Deno.FsWatcher.`,
+      );
+    }
+  },
+);
+
 // On macOS, FSEvents does not reliably emit remove events for individually
 // watched files. The previous implementation masked this by forwarding
 // unrelated events for any non-existent file to all watchers (the bug