fix(fs): register watch callbacks as change listeners

CHANGELOG.md

@@ -297,3 +297,5 @@
 - Complete missing private `node:vm` cache bytecode in the decoded execution tree, avoiding recompilation of already cached function bodies at the same 256 MiB budget.
 
 - Release completed `node:worker_threads` Worker user listeners and captured caller state while preserving internal port hooks to avoid delayed owner-exit memory release on Windows ARM64.
+
+- Register the listener passed to `fs.watch()` on the returned watcher's `change` event, preserving listener order and removal. Ports [oven-sh/bun#43765](https://github.com/oven-sh/bun/pull/43765); thanks robobun.

docs/runtime/nodejs-compat.mdx

@@ -57,6 +57,8 @@ Relative and URL-rooted absolute ESM specifiers decode URL escapes, including im
 
 ### [`node:fs`](https://nodejs.org/api/fs.html)
 
+The listener passed to `fs.watch()` is a `change` listener on the returned `FSWatcher`, including for listener inspection, removal, and manually emitted events.
+
 On macOS, `fs.watch(file)` registers a kqueue watch before returning, including unlink and writes through an open descriptor. Pending file events coalesce until JavaScript delivery, with libuv's event precedence. Directory watches use FSEvents. A new watch on a replaced file observes the new inode.
 
 Recursive `fs.promises.readdir` with `withFileTypes: true` lists directory symlinks without descending into their targets. Callback and synchronous Dirent overloads, and all recursive string overloads, follow directory symlinks on Windows and POSIX.

src/js/internal/fs/watch.ts

@@ -131,7 +131,6 @@ function assertFSEventHandle(handle) {
 
 class FSWatcher extends EventEmitter {
   #watcher;
-  #listener;
   #ignoreMatcher;
   #silentClose = false;
   _handle;
@@ -151,12 +150,7 @@ class FSWatcher extends EventEmitter {
       options = { encoding: options };
     }
 
-    if (typeof listener !== "function") {
-      listener = () => {};
-    }
-
     this.#ignoreMatcher = createIgnoreMatcher(options?.ignore);
-    this.#listener = listener;
     try {
       this.#watcher = fs.watch(path, options || {}, this.#onEvent.bind(this));
     } catch (e: any) {
@@ -166,13 +160,20 @@ class FSWatcher extends EventEmitter {
       // Only ENOENT is suppressed, and the watcher is left unstarted rather than
       // reporting the failure. The option defaults to true, so an absent value
       // still throws; any other falsy value suppresses.
-      if (options?.throwIfNoEntry !== undefined && !options.throwIfNoEntry && e.code === "ENOENT") {
-        this._handle = new FSEvent(this);
-        return;
-      }
-      throw e;
+      if (options?.throwIfNoEntry === undefined || options.throwIfNoEntry || e.code !== "ENOENT") throw e;
     }
     this._handle = new FSEvent(this);
+
+    // https://github.com/nodejs/node/blob/v26.3.0/lib/fs.js#L2540-L2542
+    if (listener) {
+      try {
+        this.addListener("change", listener);
+      } catch (e) {
+        // node leaks the started handle here, which keeps the process alive
+        closeNativeWatcherWithoutEvent(this);
+        throw e;
+      }
+    }
   }
 
   #onEvent(eventType, filenameOrError, fromAbort = false) {
@@ -203,7 +204,6 @@ class FSWatcher extends EventEmitter {
         return;
       }
       this.emit("change", eventType, filenameOrError);
-      this.#listener(eventType, filenameOrError);
     }
   }
 
@@ -245,7 +245,6 @@ class FSWatcher extends EventEmitter {
     dispatchNativeEvent = function dispatchNativeEvent(watcher, eventType, filename) {
       if (filename != null && watcher.#ignoreMatcher?.(filename)) return;
       watcher.emit("change", eventType, filename);
-      watcher.#listener(eventType, filename);
     };
     closeNativeWatcher = function closeNativeWatcher(watcher) {
       watcher.#watcher?.close();

test/js/node/watch/fixtures/listener-oracle.mjs

@@ -0,0 +1,28 @@
+import assert from "node:assert/strict";
+import { watch } from "node:fs";
+
+for (const options of [undefined, {}, "utf8", null]) {
+  const calls = [];
+  let receiver;
+  function listener(...args) {
+    receiver = this;
+    calls.push(args);
+  }
+  const watcher = options === undefined ? watch(process.argv[2], listener) : watch(process.argv[2], options, listener);
+  try {
+    assert.equal(watcher.listenerCount("change"), 1);
+    assert.deepEqual(watcher.listeners("change"), [listener]);
+    watcher.emit("change", "rename", "synthetic.txt");
+    assert.equal(receiver, watcher);
+    assert.deepEqual(calls, [["rename", "synthetic.txt"]]);
+    watcher.removeListener("change", listener);
+    watcher.emit("change", "change", "removed.txt");
+    assert.equal(calls.length, 1);
+    watcher.on("change", listener);
+    watcher.removeAllListeners("change");
+    assert.equal(watcher.listenerCount("change"), 0);
+  } finally {
+    watcher.close();
+  }
+}
+console.log("listener oracle passed");

test/js/node/watch/fs.watch.test.ts

@@ -296,6 +296,181 @@ describe("fs.watch", () => {
     }
   });
 
+  // node registers the listener argument with watcher.addListener("change", listener):
+  // https://github.com/nodejs/node/blob/v26.3.0/lib/fs.js#L2540-L2542
+  describe("the listener argument", () => {
+    test("Node oracle: synthetic change events use the registered listener", async () => {
+      using dir = tempDir("fs-watch-listener-oracle", {});
+      await using proc = Bun.spawn({
+        cmd: [bunExe(), path.join(import.meta.dir, "fixtures", "listener-oracle.mjs"), String(dir)],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+      expect({ stdout, stderr, exitCode }).toEqual({ stdout: "listener oracle passed\n", stderr: "", exitCode: 0 });
+    });
+
+    type Listener = (eventType: string, filename: string | null) => void;
+    type WatcherWithHandle = FSWatcher & {
+      _handle: { onchange(status: number, eventType: string, filename: string): void };
+    };
+    const missing = path.join(testDir, "listener-argument-404");
+
+    test.each<[string, (listener: Listener) => FSWatcher]>([
+      ["(path, listener)", listener => fs.watch(testDir, listener)],
+      ["(path, options, listener)", listener => fs.watch(testDir, { persistent: false }, listener)],
+      ["(path, encoding, listener)", listener => fs.watch(testDir, "utf8", listener)],
+      ["(path, null, listener)", listener => fs.watch(testDir, null, listener)],
+      ["(path, listener, ignored)", listener => (fs.watch as any)(testDir, listener, () => {})],
+      [
+        "(missing path, { throwIfNoEntry: false }, listener)",
+        listener => fs.watch(missing, { throwIfNoEntry: false } as fs.WatchOptionsWithStringEncoding, listener),
+      ],
+    ])("is a 'change' listener of the returned watcher: fs.watch%s", (_, watch) => {
+      const listener = () => {};
+      const watcher = watch(listener);
+      try {
+        expect(watcher.listeners("change")).toEqual([listener]);
+        expect(watcher.rawListeners("change")).toEqual([listener]);
+        expect(watcher.eventNames()).toEqual(["change"]);
+      } finally {
+        watcher.close();
+      }
+    });
+
+    // node's guard is `if (listener)`: a falsy value that is not a function is ignored, not rejected.
+    test.each([undefined, null, 0, "", false])("%p registers no listener", listener => {
+      const watcher = fs.watch(testDir, {}, listener as any);
+      try {
+        expect(watcher.eventNames()).toEqual([]);
+      } finally {
+        watcher.close();
+      }
+    });
+
+    test.each([
+      ["str", "type string ('str')"],
+      [1, "type number (1)"],
+      [{}, "an instance of Object"],
+      [true, "type boolean (true)"],
+    ])("%p is rejected by addListener", (listener, received) => {
+      const rejected = expect.objectContaining({
+        name: "TypeError",
+        code: "ERR_INVALID_ARG_TYPE",
+        message: `The "listener" argument must be of type function. Received ${received}`,
+      });
+      expect(() => fs.watch(testDir, {}, listener as any).close()).toThrow(rejected);
+      // The suppressed ENOENT leaves the watch unstarted, and the listener is still registered.
+      expect(() => fs.watch(missing, { throwIfNoEntry: false } as fs.WatchOptions, listener as any).close()).toThrow(
+        rejected,
+      );
+      // The watch starts before the listener is registered, so its error wins.
+      expect(() => fs.watch(missing, {}, listener as any).close()).toThrow(expect.objectContaining({ code: "ENOENT" }));
+    });
+
+    // node leaves the started handle open when addListener throws, and a persistent one keeps
+    // the process alive. Bun closes it.
+    test("a rejected listener does not leave the started watch open", async () => {
+      await using proc = Bun.spawn({
+        cmd: [
+          bunExe(),
+          "-e",
+          `
+          const fs = require("fs");
+          const live = new AbortController();
+          const shapes = {
+            "default": {},
+            "recursive": { recursive: true },
+            "aborted signal": { signal: AbortSignal.abort() },
+            "live signal": { signal: live.signal },
+          };
+          for (const [name, options] of Object.entries(shapes)) {
+            try {
+              fs.watch(${JSON.stringify(testDir)}, options, "not a function").close();
+              console.log(name + ": no throw");
+            } catch (err) {
+              console.log(name + ": " + err.code);
+            }
+          }
+          live.abort();
+          setImmediate(() => Bun.gc(true));
+          `,
+        ],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+      expect({ stdout: stdout.trim().split(/\r?\n/), stderr, exitCode, signalCode: proc.signalCode }).toEqual({
+        stdout: [
+          "default: ERR_INVALID_ARG_TYPE",
+          "recursive: ERR_INVALID_ARG_TYPE",
+          "aborted signal: ERR_INVALID_ARG_TYPE",
+          "live signal: ERR_INVALID_ARG_TYPE",
+        ],
+        stderr: "",
+        exitCode: 0,
+        signalCode: null,
+      });
+    });
+
+    test("runs in registration order and removeAllListeners('change') removes it", async () => {
+      using dir = tempDir("fs-watch-listener-argument", {});
+      const root = String(dir);
+      const calls: string[] = [];
+      const watcher = fs.watch(root, () => void calls.push("listener argument"));
+      let files = 0;
+      const interval = repeat(() => fs.writeFileSync(path.join(root, `file-${files++}.txt`), "hello"));
+      try {
+        watcher.on("change", () => void calls.push("on('change')"));
+        await EventEmitter.once(watcher, "change");
+        expect(calls.slice(0, 2)).toEqual(["listener argument", "on('change')"]);
+
+        watcher.removeAllListeners("change");
+        calls.length = 0;
+        await EventEmitter.once(watcher, "change");
+        expect(calls).toEqual([]);
+      } finally {
+        clearInterval(interval);
+        watcher.close();
+      }
+    });
+
+    test("receives what _handle.onchange dispatches until it is removed", () => {
+      const calls: unknown[][] = [];
+      const listener = (...args: unknown[]) => void calls.push(["listener argument", ...args]);
+      const watcher = fs.watch(testDir, listener) as WatcherWithHandle;
+      try {
+        watcher.on("change", (...args) => void calls.push(["on('change')", ...args]));
+        watcher._handle.onchange(0, "rename", "a.txt");
+        watcher.off("change", listener);
+        watcher._handle.onchange(0, "change", "b.txt");
+        expect(calls).toEqual([
+          ["listener argument", "rename", "a.txt"],
+          ["on('change')", "rename", "a.txt"],
+          ["on('change')", "change", "b.txt"],
+        ]);
+      } finally {
+        watcher.close();
+      }
+    });
+
+    test("a throw from it stops the 'change' listeners added after it", () => {
+      const later = mock();
+      const watcher = fs.watch(testDir, () => {
+        throw new Error("from the listener argument");
+      }) as WatcherWithHandle;
+      try {
+        watcher.on("change", later);
+        expect(() => watcher._handle.onchange(0, "change", "a.txt")).toThrow("from the listener argument");
+        expect(later).not.toHaveBeenCalled();
+      } finally {
+        watcher.close();
+      }
+    });
+  });
+
   test("errors from watching a missing path keep path and filename properties", () => {
     const missing = path.join(testDir, "missing-subdir", "404.txt");
     try {