feat(ext/node): add abort helpers, process & streams fix (#25262)

This commit adds:

- `addAbortListener` in `node:events`
- `aborted` in `node:util`
- `execPath` and `execvArgs` named export from `node:process`
- `getDefaultHighWaterMark` from `node:stream`

The `execPath` is very hacky - because module namespaces can not have
real getters, `execPath` is an object with a `toString()` method that on
call returns the actual `execPath`, and replaces the `execPath` binding
with the string. This is done so that we don't require the `execPath`
permission on startup.

ext/node/lib.rs

@@ -502,6 +502,7 @@ deno_core::extension!(deno_node,
     "internal/error_codes.ts",
     "internal/errors.ts",
     "internal/event_target.mjs",
+    "internal/events/abort_listener.mjs",
     "internal/fixed_queue.ts",
     "internal/fs/streams.mjs",
     "internal/fs/utils.mjs",

ext/node/polyfills/_events.mjs

@@ -47,6 +47,8 @@ import {
 import { spliceOne } from "ext:deno_node/_utils.ts";
 import { nextTick } from "ext:deno_node/_process/process.ts";
 
+export { addAbortListener } from "./internal/events/abort_listener.mjs";
+
 const kCapture = Symbol("kCapture");
 const kErrorMonitor = Symbol("events.errorMonitor");
 const kMaxEventTargetListeners = Symbol("events.maxEventTargetListeners");

ext/node/polyfills/events.ts

@@ -1,6 +1,7 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 // @deno-types="./_events.d.ts"
 export {
+  addAbortListener,
   captureRejectionSymbol,
   default,
   defaultMaxListeners,

ext/node/polyfills/internal/events/abort_listener.mjs

@@ -0,0 +1,44 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+// Copyright Joyent, Inc. and Node.js contributors. All rights reserved. MIT license.
+
+import { primordials } from "ext:deno_node/internal/test/binding.ts";
+const { queueMicrotask } = primordials;
+import { SymbolDispose } from "ext:deno_web/00_infra.js";
+import * as abortSignal from "ext:deno_web/03_abort_signal.js";
+import { validateAbortSignal, validateFunction } from "../validators.mjs";
+import { codes } from "../errors.ts";
+const { ERR_INVALID_ARG_TYPE } = codes;
+
+/**
+ * @param {AbortSignal} signal
+ * @param {EventListener} listener
+ * @returns {Disposable}
+ */
+function addAbortListener(signal, listener) {
+  if (signal === undefined) {
+    throw new ERR_INVALID_ARG_TYPE("signal", "AbortSignal", signal);
+  }
+  validateAbortSignal(signal, "signal");
+  validateFunction(listener, "listener");
+
+  let removeEventListener;
+  if (signal.aborted) {
+    queueMicrotask(() => listener());
+  } else {
+    signal[abortSignal.add](() => {
+      removeEventListener?.();
+      listener();
+    });
+    removeEventListener = () => {
+      signal[abortSignal.remove](listener);
+    };
+  }
+  return {
+    __proto__: null,
+    [SymbolDispose]() {
+      removeEventListener?.();
+    },
+  };
+}
+
+export { addAbortListener };

ext/node/polyfills/process.ts

@@ -82,6 +82,8 @@ export const argv: string[] = ["", ""];
 // And retains any value as long as it's nullish or number-ish.
 let ProcessExitCode: undefined | null | string | number;
 
+export const execArgv: string[] = [];
+
 /** https://nodejs.org/api/process.html#process_process_exit_code */
 export const exit = (code?: number | string) => {
   if (code || code === 0) {
@@ -337,7 +339,20 @@ function uncaughtExceptionHandler(err: any, origin: string) {
   process.emit("uncaughtException", err, origin);
 }
 
-let execPath: string | null = null;
+export let execPath: string = Object.freeze({
+  __proto__: String.prototype,
+  toString() {
+    execPath = Deno.execPath();
+    return execPath;
+  },
+  get length() {
+    return this.toString().length;
+  },
+  [Symbol.for("Deno.customInspect")](inspect, options) {
+    return inspect(this.toString(), options);
+  },
+  // deno-lint-ignore no-explicit-any
+}) as any as string;
 
 // The process class needs to be an ES5 class because it can be instantiated
 // in Node without the `new` keyword. It's not a true class in Node. Popular
@@ -425,7 +440,7 @@ Process.prototype.cwd = cwd;
 Process.prototype.env = env;
 
 /** https://nodejs.org/api/process.html#process_process_execargv */
-Process.prototype.execArgv = [];
+Process.prototype.execArgv = execArgv;
 
 /** https://nodejs.org/api/process.html#process_process_exit_code */
 Process.prototype.exit = exit;
@@ -704,11 +719,7 @@ Process.prototype._eval = undefined;
 
 Object.defineProperty(Process.prototype, "execPath", {
   get() {
-    if (execPath) {
-      return execPath;
-    }
-    execPath = Deno.execPath();
-    return execPath;
+    return String(execPath);
   },
   set(path: string) {
     execPath = path;

ext/node/polyfills/stream.ts

@@ -19,6 +19,9 @@ import {
   Transform,
   Writable,
 } from "ext:deno_node/_stream.mjs";
+import {
+  getDefaultHighWaterMark,
+} from "ext:deno_node/internal/streams/state.mjs";
 
 export {
   _isUint8Array,
@@ -26,6 +29,7 @@ export {
   addAbortSignal,
   Duplex,
   finished,
+  getDefaultHighWaterMark,
   PassThrough,
   pipeline,
   Readable,

ext/node/polyfills/util.ts

@@ -25,9 +25,13 @@ const {
   StringPrototypeIsWellFormed,
   StringPrototypePadStart,
   StringPrototypeToWellFormed,
+  PromiseResolve,
 } = primordials;
 
-import { promisify } from "ext:deno_node/internal/util.mjs";
+import {
+  createDeferredPromise,
+  promisify,
+} from "ext:deno_node/internal/util.mjs";
 import { callbackify } from "ext:deno_node/_util/_util_callbackify.js";
 import { debuglog } from "ext:deno_node/internal/util/debuglog.ts";
 import {
@@ -41,8 +45,13 @@ import types from "node:util/types";
 import { Buffer } from "node:buffer";
 import { isDeepStrictEqual } from "ext:deno_node/internal/util/comparisons.ts";
 import process from "node:process";
-import { validateString } from "ext:deno_node/internal/validators.mjs";
+import {
+  validateAbortSignal,
+  validateString,
+} from "ext:deno_node/internal/validators.mjs";
 import { parseArgs } from "ext:deno_node/internal/util/parse_args/parse_args.js";
+import * as abortSignal from "ext:deno_web/03_abort_signal.js";
+import { ERR_INVALID_ARG_TYPE } from "ext:deno_node/internal/errors.ts";
 
 export {
   callbackify,
@@ -288,6 +297,24 @@ export function deprecate(fn: any, msg: string, code?: any) {
   return deprecated;
 }
 
+// deno-lint-ignore require-await
+export async function aborted(
+  signal: AbortSignal,
+  // deno-lint-ignore no-explicit-any
+  _resource: any,
+): Promise<void> {
+  if (signal === undefined) {
+    throw new ERR_INVALID_ARG_TYPE("signal", "AbortSignal", signal);
+  }
+  validateAbortSignal(signal, "signal");
+  if (signal.aborted) {
+    return PromiseResolve();
+  }
+  const abortPromise = createDeferredPromise();
+  signal[abortSignal.add](abortPromise.resolve);
+  return abortPromise.promise;
+}
+
 export { getSystemErrorName, isDeepStrictEqual };
 
 export default {
@@ -311,6 +338,7 @@ export default {
   isBuffer,
   _extend,
   getSystemErrorName,
+  aborted,
   deprecate,
   callbackify,
   parseArgs,

tests/unit_node/events_test.ts

@@ -1,6 +1,7 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
-import events, { EventEmitter } from "node:events";
+// @ts-expect-error: @types/node is outdated
+import events, { addAbortListener, EventEmitter } from "node:events";
 
 EventEmitter.captureRejections = true;
 
@@ -34,3 +35,13 @@ Deno.test("eventemitter async resource", () => {
   // @ts-ignore: @types/node is outdated
   foo.emit("bar");
 });
+
+Deno.test("addAbortListener", async () => {
+  const { promise, resolve } = Promise.withResolvers<void>();
+  const abortController = new AbortController();
+  addAbortListener(abortController.signal, () => {
+    resolve();
+  });
+  abortController.abort();
+  await promise;
+});

tests/unit_node/process_test.ts

@@ -7,6 +7,8 @@ import process, {
   argv,
   argv0 as importedArgv0,
   env,
+  execArgv as importedExecArgv,
+  execPath as importedExecPath,
   geteuid,
   pid as importedPid,
   platform as importedPlatform,
@@ -1121,3 +1123,11 @@ Deno.test("process.listeners - include SIG* events", () => {
 Deno.test(function processVersionsOwnProperty() {
   assert(Object.prototype.hasOwnProperty.call(process, "versions"));
 });
+
+Deno.test(function importedExecArgvTest() {
+  assert(Array.isArray(importedExecArgv));
+});
+
+Deno.test(function importedExecPathTest() {
+  assertEquals(importedExecPath, Deno.execPath());
+});

tests/unit_node/stream_test.ts

@@ -1,8 +1,10 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
-import { assert } from "@std/assert";
+import { assert, assertEquals } from "@std/assert";
 import { fromFileUrl, relative } from "@std/path";
 import { pipeline } from "node:stream/promises";
+// @ts-expect-error: @types/node is outdated
+import { getDefaultHighWaterMark } from "node:stream";
 import { createReadStream, createWriteStream } from "node:fs";
 
 Deno.test("stream/promises pipeline", async () => {
@@ -23,3 +25,8 @@ Deno.test("stream/promises pipeline", async () => {
     // pass
   }
 });
+
+Deno.test("stream getDefaultHighWaterMark", () => {
+  assertEquals(getDefaultHighWaterMark(false), 16 * 1024);
+  assertEquals(getDefaultHighWaterMark(true), 16);
+});

tests/unit_node/util_test.ts

@@ -330,3 +330,21 @@ Deno.test("[util] debuglog() and debug()", () => {
   assertEquals(util.debuglog, util.debug);
   assertEquals(utilDefault.debuglog, utilDefault.debug);
 });
+
+Deno.test("[util] aborted()", async () => {
+  const abortController = new AbortController();
+  let done = false;
+  const promise = util.aborted(
+    // deno-lint-ignore no-explicit-any
+    abortController.signal as any,
+    abortController.signal,
+  );
+  promise.then(() => {
+    done = true;
+  });
+  await new Promise((r) => setTimeout(r, 100));
+  assertEquals(done, false);
+  abortController.abort();
+  await promise;
+  assertEquals(done, true);
+});