fix(stdlib): pause stdin when readline closes

changelog.d/9641-readline-close-pauses-stdin.md

@@ -0,0 +1,12 @@
+**fix(stdlib): pause process.stdin when closing a readline interface (#9594)**
+
+Closing a stdin-backed readline interface now pauses the shared
+`process.stdin` stream, matching Node. Previously `rl.close()` fired the
+interface's close callback but left Perry's background stdin reader flowing,
+so bytes written afterwards still reached `process.stdin` `data` listeners and
+could keep a CLI alive unexpectedly.
+
+Interface close is now tracked separately from physical stdin EOF. An explicit
+`process.stdin.resume()` can therefore restore delivery after close, a later
+real EOF still reaches stdin's `end` / `close` listeners, and constructing a
+new readline interface resumes stdin as Node's constructor does.

crates/perry-stdlib/src/readline/mod.rs

@@ -127,9 +127,15 @@ static READABLE_EOF_NOTIFIED: AtomicBool = AtomicBool::new(false);
 /// `true` when raw mode is enabled — the reader thread checks this
 /// between bytes to decide which queue to push to.
 static RAW_MODE: AtomicBool = AtomicBool::new(false);
-/// Set when stdin returns EOF or `rl.close()` is called. The has-active
-/// check reads this to decide whether to keep the event loop alive.
+/// Set only when stdin returns EOF. Closing a readline interface pauses its
+/// input stream but does not end that stream; `process.stdin.resume()` may
+/// consume more bytes afterwards.
 static EOF_REACHED: AtomicBool = AtomicBool::new(false);
+/// Whether the physical stdin EOF has been dispatched to `end` / `close`
+/// listeners. This is separate from `CLOSE_FIRED`: explicitly closing a
+/// readline interface fires the interface's `close` event without ending
+/// `process.stdin`.
+static STDIN_END_FIRED: AtomicBool = AtomicBool::new(false);
 /// Whether the background reader thread has been spawned. Atomic
 /// (compare_exchange) so we don't accidentally spawn twice if two
 /// init paths race on first call.
@@ -1410,6 +1416,12 @@ pub extern "C" fn js_readline_create_interface(opts: f64) -> i64 {
     try_register_pump();
     let handle = create_interface_from_options(opts);
     if !with_interface(handle, |state| state.uses_custom_stream).unwrap_or(false) {
+        // Node's Interface constructor calls input.resume(). This matters for
+        // a second interface created after the first one was closed, since
+        // close() pauses the shared stdin stream.
+        if !STDIN_DESTROYED.load(Ordering::Acquire) && !EOF_REACHED.load(Ordering::Acquire) {
+            STDIN_PAUSED.store(false, Ordering::Release);
+        }
         ensure_reader_started();
     }
     handle
@@ -1525,8 +1537,8 @@ pub extern "C" fn js_readline_on(
     undefined()
 }
 
-/// rl.close() — synchronously fire the close callback (matching Node's
-/// `Interface.close()` semantics) and mark the interface as EOF.
+/// rl.close() — synchronously pause the input and fire the close callback,
+/// matching Node's `Interface.close()` semantics.
 #[no_mangle]
 pub extern "C" fn js_readline_close(_handle: i64) -> f64 {
     match with_interface(_handle, |state| state.uses_custom_stream) {
@@ -1540,7 +1552,10 @@ pub extern "C" fn js_readline_close(_handle: i64) -> f64 {
         None if _handle != STDIN_READLINE_HANDLE => return undefined(),
         _ => {}
     }
-    EOF_REACHED.store(true, Ordering::Release);
+    // Node implements Interface.close() by calling Interface.pause(), which
+    // in turn pauses the input stream. Do not mark stdin as EOF: user code can
+    // explicitly resume the shared stream after the interface is gone.
+    STDIN_PAUSED.store(true, Ordering::Release);
     // Node stops emitting 'line' after close(). Without clearing these, the
     // pump would still deliver a queued late line to the 'line' handler and
     // `has_line_callbacks` would keep the event loop alive.
@@ -1844,6 +1859,7 @@ pub extern "C" fn js_readline_stdin_destroy() -> f64 {
     RAW_MODE.store(false, Ordering::Release);
     STDIN_DATA_FLOWING.store(false, Ordering::Release);
     EOF_REACHED.store(true, Ordering::Release);
+    STDIN_END_FIRED.store(true, Ordering::Release);
     let _ = termios_impl::disable();
     if let Ok(mut q) = PENDING_DATA.lock() {
         q.clear();

crates/perry-stdlib/src/readline/mod_tests.rs

@@ -11,8 +11,45 @@ fn close_without_callbacks_is_noop() {
     let h = js_readline_create_interface(0.0);
     assert_eq!(h, STDIN_READLINE_HANDLE);
     js_readline_close(h);
+    assert!(STDIN_PAUSED.load(Ordering::Acquire));
+    assert!(!EOF_REACHED.load(Ordering::Acquire));
+    test_inject_line("late");
+    assert_eq!(js_readline_process_pending(), 0);
+    assert_eq!(PENDING_LINES.lock().unwrap().len(), 1);
+    assert_eq!(js_readline_has_active(), 0);
     assert_eq!(js_readline_process_pending(), 0);
+}
+
+#[test]
+fn stdin_close_pauses_delivery_until_explicit_resume() {
+    let _g = reset();
+    let h = js_readline_create_interface(0.0);
+    let event = event_name("data");
+    let cb = data_counter_callback();
+    let _ = js_readline_stdin_on(event, cb);
+
+    js_readline_close(h);
+    test_inject_chunk(b"late");
     assert_eq!(js_readline_process_pending(), 0);
+    assert_eq!(PENDING_DATA.lock().unwrap().len(), 1);
+    assert!(!EOF_REACHED.load(Ordering::Acquire));
+
+    let _ = js_readline_stdin_resume();
+    assert_eq!(js_readline_process_pending(), 1);
+    DATA_COUNT.with(|count| assert_eq!(*count.borrow(), 1));
+}
+
+#[test]
+fn new_stdin_interface_resumes_after_previous_interface_closed() {
+    let _g = reset();
+    let first = js_readline_create_interface(0.0);
+    js_readline_close(first);
+    assert!(STDIN_PAUSED.load(Ordering::Acquire));
+
+    let second = js_readline_create_interface(0.0);
+    assert_eq!(second, STDIN_READLINE_HANDLE);
+    assert!(!STDIN_PAUSED.load(Ordering::Acquire));
+    assert!(!EOF_REACHED.load(Ordering::Acquire));
 }
 
 #[test]

crates/perry-stdlib/src/readline/pump.rs

@@ -363,14 +363,18 @@ pub extern "C" fn js_readline_process_pending() -> i32 {
         }
     }
 
-    // Fire close callback once on EOF.
+    // Physical EOF closes an active readline interface and independently
+    // emits process.stdin's end/close events. `rl.close()` may already have
+    // fired the first event without stdin actually ending, so the two
+    // one-shot states must not be conflated.
     if EOF_REACHED.load(Ordering::Acquire) {
-        let already = CLOSE_FIRED.with(|f| {
+        let readline_close_already = CLOSE_FIRED.with(|f| {
             let was = *f.borrow();
             *f.borrow_mut() = true;
             was
         });
-        if !already {
+        let stdin_end_already = STDIN_END_FIRED.swap(true, Ordering::AcqRel);
+        if !stdin_end_already {
             // #9490: flush the stream decoder first — a sequence left
             // incomplete at EOF is one final `'data'` chunk of U+FFFD, ahead
             // of `'end'`/`'close'`.
@@ -397,12 +401,16 @@ pub extern "C" fn js_readline_process_pending() -> i32 {
                     fired += 1;
                 }
             }
+        }
+        if !readline_close_already {
             let cb = CLOSE_CALLBACK.with(|c| c.borrow_mut().take());
             if let Some(cb_i64) = cb {
                 let closure = cb_i64 as *const ClosureHeader;
                 js_closure_call0(closure);
                 fired += 1;
             }
+        }
+        if !stdin_end_already {
             // Every `process.stdin.on("end" | "close", …)` listener, in
             // registration order. Node fires all of them; the previous
             // single-slot storage kept only the last one registered.
@@ -455,12 +463,15 @@ pub extern "C" fn js_readline_has_active() -> i32 {
             .unwrap_or(false);
     let has_line_callbacks = QUESTION_CALLBACK.with(|c| c.borrow().is_some())
         || LINE_CALLBACK.with(|c| c.borrow().is_some());
-    let has_close_cb = !CLOSE_FIRED.with(|f| *f.borrow())
-        && (CLOSE_CALLBACK.with(|c| c.borrow().is_some())
-            || STDIN_END_CALLBACKS
-                .lock()
-                .map(|v| !v.is_empty())
-                .unwrap_or(false));
+    let has_readline_close_cb =
+        !CLOSE_FIRED.with(|f| *f.borrow()) && CLOSE_CALLBACK.with(|c| c.borrow().is_some());
+    let has_stdin_end_cb = !STDIN_END_FIRED.load(Ordering::Acquire)
+        && STDIN_END_CALLBACKS
+            .lock()
+            .map(|v| !v.is_empty())
+            .unwrap_or(false);
+    let has_close_cb = has_readline_close_cb || has_stdin_end_cb;
+    let has_dispatchable_lines = has_lines && !paused;
     let has_dispatchable_data = has_data && has_stdin_callbacks && !paused;
     let reader_keeps_alive = started
         && !eof
@@ -473,7 +484,7 @@ pub extern "C" fn js_readline_has_active() -> i32 {
             || has_close_cb);
     if !destroyed
         && refed
-        && (has_lines || has_dispatchable_data || has_close_cb || reader_keeps_alive)
+        && (has_dispatchable_lines || has_dispatchable_data || has_close_cb || reader_keeps_alive)
     {
         1
     } else {

crates/perry-stdlib/src/readline/test_support.rs

@@ -77,6 +77,7 @@ pub(super) fn reset() -> MutexGuard<'static, ()> {
     PENDING_DATA.lock().unwrap().clear();
     PENDING_ESCAPE.lock().unwrap().clear();
     EOF_REACHED.store(false, Ordering::Release);
+    STDIN_END_FIRED.store(false, Ordering::Release);
     READABLE_EOF_NOTIFIED.store(false, Ordering::Release);
     STDIN_PAUSED.store(false, Ordering::Release);
     STDIN_REFED.store(true, Ordering::Release);

crates/perry/tests/issue_9594_readline_close_pauses_stdin.rs

@@ -0,0 +1,182 @@
+//! Regression test for #9594: closing a stdin-backed readline interface must
+//! pause the shared `process.stdin` stream.
+//!
+//! The readline interface and `process.stdin` share one native reader. Perry
+//! used to mark the interface closed without pausing that reader, so bytes
+//! written after `rl.close()` still reached a `process.stdin` `data` listener.
+//! Node pauses stdin on both `rl.close()` and `rl.pause()`. The stream remains
+//! usable: an explicit `process.stdin.resume()` after close enables delivery
+//! again.
+
+use std::io::{BufRead, BufReader, Write};
+use std::path::{Path, PathBuf};
+use std::process::{Child, Command, Stdio};
+use std::sync::mpsc::{self, Receiver};
+use std::time::{Duration, Instant};
+
+fn perry_bin() -> PathBuf {
+    PathBuf::from(env!("CARGO_BIN_EXE_perry"))
+}
+
+const SOURCE: &str = r#"
+import * as readline from "readline";
+
+const mode = process.argv[2];
+const rl = readline.createInterface({
+  input: process.stdin,
+  output: process.stdout,
+  terminal: false,
+});
+
+rl.question("", (_answer: string) => {
+  const reportLateData = (chunk: string) => {
+    console.log("LATE_DATA:" + JSON.stringify(String(chunk)));
+  };
+
+  if (mode === "close-no-listener") {
+    rl.close();
+  } else if (mode === "close-listener-before") {
+    process.stdin.on("data", reportLateData);
+    rl.close();
+  } else if (mode === "pause") {
+    process.stdin.on("data", reportLateData);
+    rl.pause();
+  } else if (mode === "close-listener-after") {
+    rl.close();
+    process.stdin.on("data", reportLateData);
+  } else if (mode === "close-resume") {
+    rl.close();
+    process.stdin.on("data", reportLateData);
+    process.stdin.on("end", () => console.log("STDIN_END"));
+    process.stdin.resume();
+  }
+
+  console.log("READY_FOR_LATE");
+  setTimeout(() => console.log("DONE"), 150);
+});
+
+console.log("READY_FOR_ANSWER");
+"#;
+
+fn compile(dir: &Path) -> PathBuf {
+    let entry = dir.join("main.ts");
+    let output = dir.join("main_bin");
+    std::fs::write(&entry, SOURCE).expect("write entry");
+    let compile = Command::new(perry_bin())
+        .current_dir(dir)
+        .arg("compile")
+        .arg(&entry)
+        .arg("-o")
+        .arg(&output)
+        .output()
+        .expect("run perry compile");
+    assert!(
+        compile.status.success(),
+        "perry compile failed\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&compile.stdout),
+        String::from_utf8_lossy(&compile.stderr)
+    );
+    output
+}
+
+fn recv_until(rx: &Receiver<String>, expected: &str, output: &mut Vec<String>) {
+    let deadline = Instant::now() + Duration::from_secs(10);
+    loop {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        let line = rx
+            .recv_timeout(remaining)
+            .unwrap_or_else(|_| panic!("child never printed {expected}; output: {output:?}"));
+        let matched = line == expected;
+        output.push(line);
+        if matched {
+            return;
+        }
+    }
+}
+
+fn wait_for_exit(child: &mut Child, mode: &str, output: &[String]) {
+    let deadline = Instant::now() + Duration::from_secs(10);
+    loop {
+        match child.try_wait().expect("poll compiled fixture") {
+            Some(status) => {
+                assert!(
+                    status.success(),
+                    "{mode} fixture exited with {status}; output: {output:?}"
+                );
+                return;
+            }
+            None if Instant::now() >= deadline => {
+                let _ = child.kill();
+                let _ = child.wait();
+                panic!("{mode} fixture did not exit after stdin closed; output: {output:?}");
+            }
+            None => std::thread::sleep(Duration::from_millis(10)),
+        }
+    }
+}
+
+fn run_arm(bin: &Path, mode: &str) -> Vec<String> {
+    let mut child = Command::new(bin)
+        .arg(mode)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::inherit())
+        .spawn()
+        .expect("spawn compiled fixture");
+    let mut stdin = child.stdin.take().expect("piped stdin");
+    let stdout = child.stdout.take().expect("piped stdout");
+    let (tx, rx) = mpsc::channel();
+    let reader = std::thread::spawn(move || {
+        for line in BufReader::new(stdout).lines() {
+            let Ok(line) = line else { break };
+            if tx.send(line).is_err() {
+                break;
+            }
+        }
+    });
+
+    let mut output = Vec::new();
+    recv_until(&rx, "READY_FOR_ANSWER", &mut output);
+    writeln!(stdin, "answer").expect("write answer");
+    stdin.flush().expect("flush answer");
+    recv_until(&rx, "READY_FOR_LATE", &mut output);
+    writeln!(stdin, "late").expect("write late input");
+    stdin.flush().expect("flush late input");
+    drop(stdin);
+    recv_until(&rx, "DONE", &mut output);
+    wait_for_exit(&mut child, mode, &output);
+    reader.join().expect("join stdout reader");
+    while let Ok(line) = rx.try_recv() {
+        output.push(line);
+    }
+    output
+}
+
+#[test]
+fn readline_close_matches_stdin_pause_without_permanently_muting_stdin() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let bin = compile(dir.path());
+
+    for mode in [
+        "close-no-listener",
+        "close-listener-before",
+        "pause",
+        "close-listener-after",
+    ] {
+        let output = run_arm(&bin, mode);
+        assert!(
+            output.iter().all(|line| !line.starts_with("LATE_DATA:")),
+            "{mode} delivered bytes after stdin was paused: {output:?}"
+        );
+    }
+
+    let resumed = run_arm(&bin, "close-resume");
+    assert!(
+        resumed.iter().any(|line| line == "LATE_DATA:\"late\\n\""),
+        "explicit resume after close did not restore stdin delivery: {resumed:?}"
+    );
+    assert!(
+        resumed.iter().any(|line| line == "STDIN_END"),
+        "closing readline suppressed process.stdin's later EOF: {resumed:?}"
+    );
+}