From 65f114be7f50f9c4868f27eb03b574d8fba85140 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 3 Sep 2026 18:28:10 +0200 Subject: [PATCH] fix(stdlib): pause stdin when readline closes (#9594) --- .../9641-readline-close-pauses-stdin.md | 12 ++ crates/perry-stdlib/src/readline/mod.rs | 26 ++- crates/perry-stdlib/src/readline/mod_tests.rs | 37 ++++ crates/perry-stdlib/src/readline/pump.rs | 31 ++- .../perry-stdlib/src/readline/test_support.rs | 1 + .../issue_9594_readline_close_pauses_stdin.rs | 182 ++++++++++++++++++ 6 files changed, 274 insertions(+), 15 deletions(-) create mode 100644 changelog.d/9641-readline-close-pauses-stdin.md create mode 100644 crates/perry/tests/issue_9594_readline_close_pauses_stdin.rs diff --git a/changelog.d/9641-readline-close-pauses-stdin.md b/changelog.d/9641-readline-close-pauses-stdin.md new file mode 100644 index 0000000000..05df404e5a --- /dev/null +++ b/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. diff --git a/crates/perry-stdlib/src/readline/mod.rs b/crates/perry-stdlib/src/readline/mod.rs index 229f6fcbf0..9ecc0b23ac 100644 --- a/crates/perry-stdlib/src/readline/mod.rs +++ b/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(); diff --git a/crates/perry-stdlib/src/readline/mod_tests.rs b/crates/perry-stdlib/src/readline/mod_tests.rs index 5d3e0d5b78..91dca9faf3 100644 --- a/crates/perry-stdlib/src/readline/mod_tests.rs +++ b/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] diff --git a/crates/perry-stdlib/src/readline/pump.rs b/crates/perry-stdlib/src/readline/pump.rs index b73f3ee6ac..6be359cf3f 100644 --- a/crates/perry-stdlib/src/readline/pump.rs +++ b/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 { diff --git a/crates/perry-stdlib/src/readline/test_support.rs b/crates/perry-stdlib/src/readline/test_support.rs index 29c0c2d700..c5e8e2bed4 100644 --- a/crates/perry-stdlib/src/readline/test_support.rs +++ b/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); diff --git a/crates/perry/tests/issue_9594_readline_close_pauses_stdin.rs b/crates/perry/tests/issue_9594_readline_close_pauses_stdin.rs new file mode 100644 index 0000000000..10b4d9bc64 --- /dev/null +++ b/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, expected: &str, output: &mut Vec) { + 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 { + 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:?}" + ); +}