fix(ext/node): FsWatcher ref and unref (#22987)

Fixes #22973

---------

Co-authored-by: Satya Rohith <me@satyarohith.com>

ext/node/polyfills/_fs/_fs_watch.ts

@@ -146,7 +146,7 @@ export function watch(
       }
       throw e;
     }
-  });
+  }, () => iterator);
 
   if (listener) {
     fsWatcher.on("change", listener.bind({ _handle: fsWatcher }));
@@ -252,6 +252,7 @@ class StatWatcher extends EventEmitter {
   #bigint: boolean;
   #refCount = 0;
   #abortController = new AbortController();
+
   constructor(bigint: boolean) {
     super();
     this.#bigint = bigint;
@@ -306,19 +307,22 @@ class StatWatcher extends EventEmitter {
     this.emit("stop");
   }
   ref() {
-    notImplemented("FSWatcher.ref() is not implemented");
+    notImplemented("StatWatcher.ref() is not implemented");
   }
   unref() {
-    notImplemented("FSWatcher.unref() is not implemented");
+    notImplemented("StatWatcher.unref() is not implemented");
   }
 }
 
 class FSWatcher extends EventEmitter {
   #closer: () => void;
   #closed = false;
-  constructor(closer: () => void) {
+  #watcher: () => Deno.FsWatcher;
+
+  constructor(closer: () => void, getter: () => Deno.FsWatcher) {
     super();
     this.#closer = closer;
+    this.#watcher = getter;
   }
   close() {
     if (this.#closed) {
@@ -329,10 +333,10 @@ class FSWatcher extends EventEmitter {
     this.#closer();
   }
   ref() {
-    notImplemented("FSWatcher.ref() is not implemented");
+    this.#watcher().ref();
   }
   unref() {
-    notImplemented("FSWatcher.unref() is not implemented");
+    this.#watcher().unref();
   }
 }
 

runtime/js/40_fs_events.js

@@ -17,6 +17,7 @@ import { SymbolDispose } from "ext:deno_web/00_infra.js";
 
 class FsWatcher {
   #rid = 0;
+  #promise;
 
   constructor(paths, options) {
     const { recursive } = options;
@@ -32,9 +33,18 @@ class FsWatcher {
     return this.#rid;
   }
 
+  unref() {
+    core.unrefOpPromise(this.#promise);
+  }
+
+  ref() {
+    core.refOpPromise(this.#promise);
+  }
+
   async next() {
     try {
-      const value = await op_fs_events_poll(this.#rid);
+      this.#promise = op_fs_events_poll(this.#rid);
+      const value = await this.#promise;
       return value ? { value, done: false } : { value: undefined, done: true };
     } catch (error) {
       if (ObjectPrototypeIsPrototypeOf(BadResourcePrototype, error)) {

tests/unit_node/_fs/_fs_watch_test.ts

@@ -38,3 +38,17 @@ Deno.test({
     unwatchFile(file);
   },
 });
+
+Deno.test({
+  name: "watch.unref() should work",
+  sanitizeOps: false,
+  sanitizeResources: false,
+  async fn() {
+    const file = Deno.makeTempFileSync();
+    const watcher = watch(file, () => {});
+    // Wait for the watcher to be initialized
+    await wait(10);
+    // @ts-ignore node types are outdated in deno.
+    watcher.unref();
+  },
+});