fix(ext/node): support process.stdin.unref()

ext/io/12_io.js

@@ -9,6 +9,7 @@ import { op_set_raw } from "ext:core/ops";
 const {
  Uint8Array,
  ArrayPrototypePush,
+ Symbol,
  TypedArrayPrototypeSubarray,
  TypedArrayPrototypeSet,
  TypedArrayPrototypeGetByteLength,
@@ -181,9 +182,14 @@ const STDIN_RID = 0;
 const STDOUT_RID = 1;
 const STDERR_RID = 2;
 
+const REF = Symbol("REF");
+const UNREF = Symbol("UNREF");
+
 class Stdin {
  #rid = STDIN_RID;
+ #ref = true;
  #readable;
+ #opPromise;
 
  constructor() {
  }
@@ -197,8 +203,14 @@ class Stdin {
  return this.#rid;
  }
 
- read(p) {
- return read(this.#rid, p);
+ async read(p) {
+ if (p.length === 0) return 0;
+ this.#opPromise = core.read(this.#rid, p);
+ if (!this.#ref) {
+ core.unrefOpPromise(this.#opPromise);
+ }
+ const nread = await this.#opPromise;
+ return nread === 0 ? null : nread;
  }
 
  readSync(p) {
@@ -224,6 +236,20 @@ class Stdin {
  isTerminal() {
  return core.isTerminal(this.#rid);
  }
+
+ [REF]() {
+ this.#ref = true;
+ if (this.#opPromise) {
+ core.refOpPromise(this.#opPromise);
+ }
+ }
+
+ [UNREF]() {
+ this.#ref = false;
+ if (this.#opPromise) {
+ core.unrefOpPromise(this.#opPromise);
+ }
+ }
 }
 
 class Stdout {
@@ -318,6 +344,7 @@ export {
  readAll,
  readAllSync,
  readSync,
+ REF,
  SeekMode,
  Stderr,
  stderr,
@@ -327,6 +354,7 @@ export {
  Stdout,
  stdout,
  STDOUT_RID,
+ UNREF,
  write,
  writeSync,
 };

ext/node/polyfills/tty.js

@@ -9,7 +9,10 @@ const {
 } = core;
 
 import { ERR_INVALID_FD } from "ext:deno_node/internal/errors.ts";
-import { LibuvStreamWrap } from "ext:deno_node/internal_binding/stream_wrap.ts";
+import {
+ kStreamBaseField,
+ LibuvStreamWrap,
+} from "ext:deno_node/internal_binding/stream_wrap.ts";
 import { providerType } from "ext:deno_node/internal_binding/async_wrap.ts";
 import { Socket } from "node:net";
 import { setReadStream } from "ext:deno_node/_process/streams.mjs";
@@ -36,6 +39,14 @@ class TTY extends LibuvStreamWrap {
  constructor(handle) {
  super(providerType.TTYWRAP, handle);
  }
+
+ ref() {
+ this[kStreamBaseField][io.REF]();
+ }
+
+ unref() {
+ this[kStreamBaseField][io.UNREF]();
+ }
 }
 
 export class ReadStream extends Socket {

tests/integration/run_tests.rs

@@ -5359,3 +5359,32 @@ fn code_cache_npm_with_require_test() {
  assert!(!output.stderr().contains("Updating V8 code cache"));
  }
 }
+
+#[test]
+fn node_process_stdin_unref_with_pty() {
+ TestContext::default()
+ .new_command()
+ .args_vec(["run", "--quiet", "run/node_process_stdin_unref_with_pty.js"])
+ .with_pty(|mut console| {
+ console.expect("START\r\n");
+ console.write_line("foo");
+ console.expect("foo\r\n");
+ console.write_line("bar");
+ console.expect("bar\r\n");
+ console.write_line("baz");
+ console.expect("baz\r\n");
+ });
+
+ TestContext::default()
+ .new_command()
+ .args_vec([
+ "run",
+ "--quiet",
+ "run/node_process_stdin_unref_with_pty.js",
+ "--unref",
+ ])
+ .with_pty(|mut console| {
+ // if process.stdin.unref is called, the program immediately ends by skipping reading from stdin.
+ console.expect("START\r\nEND\r\n");
+ });
+}

tests/testdata/run/node_process_stdin_unref_with_pty.js

@@ -0,0 +1,14 @@
+import process from "node:process";
+import util from "node:util";
+
+console.log("START");
+globalThis.addEventListener("unload", () => console.log("END"));
+
+const args = util.parseArgs({ options: { unref: { type: "boolean" } } });
+
+// call stdin.unref if --unref is passed
+if (args.values.unref) {
+ process.stdin.unref();
+}
+
+process.stdin.pipe(process.stdout);