From a37805682624bf8ef165b26505fa1cf3cc393039 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 3 Sep 2026 16:20:42 +0000 Subject: [PATCH 1/7] fix(stdlib): clock pending Escape with timeout (#9593) Hold incomplete ANSI escape prefixes against an explicit 500 ms deadline registered with the event pump. Unrelated loop turns no longer flush early, idle waits no longer flush late, and completing bytes cancel the one-shot.\n\nCover bare Escape under idle and interval-driven schedules plus split and complete arrow sequences on a real PTY. --- changelog.d/9593-readline-escape-timeout.md | 11 + crates/perry-runtime/src/event_pump.rs | 47 ++-- crates/perry-runtime/src/lib.rs | 26 ++ .../perry-stdlib/src/common/async_bridge.rs | 8 +- crates/perry-stdlib/src/readline/mod.rs | 11 + crates/perry-stdlib/src/readline/mod_tests.rs | 33 ++- crates/perry-stdlib/src/readline/pump.rs | 81 +++++- .../perry-stdlib/src/readline/test_support.rs | 1 + .../issue_9593_readline_escape_timeout.rs | 240 ++++++++++++++++++ 9 files changed, 419 insertions(+), 39 deletions(-) create mode 100644 changelog.d/9593-readline-escape-timeout.md create mode 100644 crates/perry/tests/issue_9593_readline_escape_timeout.rs diff --git a/changelog.d/9593-readline-escape-timeout.md b/changelog.d/9593-readline-escape-timeout.md new file mode 100644 index 0000000000..86d6972eaa --- /dev/null +++ b/changelog.d/9593-readline-escape-timeout.md @@ -0,0 +1,11 @@ +**fix(stdlib): clock bare Escape with readline's 500 ms timeout (#9593)** + +A raw-mode Escape keypress no longer flushes on an arbitrary event-loop turn. +The readline pump now holds a torn escape prefix until an explicit 500 ms +deadline, matching Node's default `escapeCodeTimeout`. Bytes that finish an +arrow or other ANSI sequence cancel the one-shot, while expiry delivers a bare +Escape. + +The deadline participates in the event pump's wait budget. Bare Escape therefore +arrives at the same time whether the loop is otherwise idle or an unrelated +timer is firing, and arrows split across reads remain a single keypress. diff --git a/crates/perry-runtime/src/event_pump.rs b/crates/perry-runtime/src/event_pump.rs index d090ce7c4c..9d008beb17 100644 --- a/crates/perry-runtime/src/event_pump.rs +++ b/crates/perry-runtime/src/event_pump.rs @@ -7,7 +7,7 @@ //! - a cross-thread event source (tokio worker, `std::thread::spawn`) //! calls `js_notify_main_thread` after pushing into a queue that the //! pump drains, or -//! - the next timer / interval deadline elapses, or +//! - the next timer, interval, or registered native deadline elapses, or //! - a 1-second safety cap elapses (heartbeat). //! //! Result: cross-thread async-op latency on the event loop drops from @@ -418,7 +418,7 @@ pub(crate) fn js_notify_promise_progress() { // * `perry_poll()` — drains microtasks + stdlib // * `perry_has_work()` — true while anything is pending (microtasks, // timers across all 3 queues, stdlib handles) -// * `perry_next_wake_ms()` — minimum across the 3 timer queues, or -1 +// * `perry_next_wake_ms()` — minimum timer/native deadline, or -1 // // Pair with `perry_set_wake_callback` for polling-free integration. // ============================================================================ @@ -482,21 +482,30 @@ pub extern "C" fn perry_has_work() -> i32 { 0 } -/// Returns the closest pending wake-up across all 3 timer queues, in -/// milliseconds from now. Returns -1.0 when no timers are scheduled — -/// the host can then sleep indefinitely (or until an OS event / a wake -/// callback fires). +/// Return every source that can put a deadline on the next event-loop park. +/// The fourth slot is an optional callback rather than a hard stdlib reference, +/// so runtime-only binaries retain their existing link surface. +#[inline] +fn next_wake_sources_ms() -> [f64; 4] { + [ + js_timer_next_deadline(), + js_callback_timer_next_deadline(), + js_interval_timer_next_deadline(), + crate::stdlib_pump::stdlib_next_wake_ms(), + ] +} + +/// Returns the closest pending wake-up across all timer queues and registered +/// stdlib one-shots, in milliseconds from now. Returns -1.0 when no deadline is +/// scheduled — the host can then sleep indefinitely (or until an OS event / a +/// wake callback fires). /// /// NaN is *not* returned — keeps the return shape printable and avoids /// surprising hosts that compare with `<`. #[no_mangle] pub extern "C" fn perry_next_wake_ms() -> f64 { let mut best: f64 = -1.0; - for d in [ - js_timer_next_deadline(), - js_callback_timer_next_deadline(), - js_interval_timer_next_deadline(), - ] { + for d in next_wake_sources_ms() { if d < 0.0 { continue; } @@ -526,11 +535,11 @@ pub extern "C" fn js_event_loop_host_driven() -> i32 { EVENT_LOOP_HOST_DRIVEN.load(Ordering::Relaxed) as i32 } -/// Block until the next scheduled timer fires, a notify arrives, or the -/// 1-second idle cap elapses — whichever is earliest. Returns immediately -/// if a notify arrived since the last call (the flag is cleared on -/// return). Replaces the old `js_sleep_ms` in the generated event loop -/// and `await` busy-wait. +/// Block until the next scheduled timer/native deadline fires, a notify +/// arrives, or the 1-second idle cap elapses — whichever is earliest. Returns +/// immediately if a notify arrived since the last call (the flag is cleared on +/// return). Replaces the old `js_sleep_ms` in the generated event loop and +/// `await` busy-wait. #[no_mangle] pub extern "C" fn js_wait_for_event() { // `PERRY_GC_CENSUS`: one relaxed atomic load; services a pending SIGUSR2 @@ -574,11 +583,7 @@ pub extern "C" fn js_wait_for_event() { } let mut budget_ms: u64 = IDLE_CAP_MS; - for d in [ - js_timer_next_deadline(), - js_callback_timer_next_deadline(), - js_interval_timer_next_deadline(), - ] { + for d in next_wake_sources_ms() { if d >= 0.0 { let d_ms = d as u64; if d_ms < budget_ms { diff --git a/crates/perry-runtime/src/lib.rs b/crates/perry-runtime/src/lib.rs index e4e3016db2..7cc3b28249 100644 --- a/crates/perry-runtime/src/lib.rs +++ b/crates/perry-runtime/src/lib.rs @@ -367,6 +367,11 @@ pub(crate) mod stdlib_pump { use std::sync::Mutex; static STDLIB_PUMP_FN: AtomicPtr<()> = AtomicPtr::new(null_mut()); + /// Optional deadline provider for stdlib-owned one-shot work. The callback + /// returns milliseconds until its next wake, or -1 when it has no deadline. + /// Kept as a function pointer for the same reason as `STDLIB_PUMP_FN`: the + /// runtime must not hard-link perry-stdlib into runtime-only binaries. + static STDLIB_NEXT_WAKE_FN: AtomicPtr<()> = AtomicPtr::new(null_mut()); // Runtime-internal reactor pumps (child_process, node-pty) register here // when their first live handle appears, mirroring `STDLIB_PUMP_FN`. The @@ -517,6 +522,27 @@ pub(crate) mod stdlib_pump { STDLIB_PUMP_FN.store(f as *mut (), Ordering::Release); } + /// Register the stdlib's nearest-deadline provider. This lets native + /// one-shots participate in `js_wait_for_event` without manufacturing a JS + /// timer callback or relying on the one-second idle heartbeat. + #[no_mangle] + pub extern "C" fn js_register_stdlib_next_wake(f: extern "C" fn() -> f64) { + STDLIB_NEXT_WAKE_FN.store(f as *mut (), Ordering::Release); + } + + /// Milliseconds until the closest registered stdlib deadline, or -1 when + /// perry-stdlib is absent or currently owns no timed work. + pub(crate) fn stdlib_next_wake_ms() -> f64 { + let f = STDLIB_NEXT_WAKE_FN.load(Ordering::Acquire); + if f.is_null() { + return -1.0; + } + // SAFETY: `js_register_stdlib_next_wake` only stores callbacks with + // this exact ABI and signature. + let func: extern "C" fn() -> f64 = unsafe { std::mem::transmute(f) }; + func() + } + /// Run the registered stdlib pump if available. Safe to call even if perry-stdlib /// is not linked (no-op in that case). #[no_mangle] diff --git a/crates/perry-stdlib/src/common/async_bridge.rs b/crates/perry-stdlib/src/common/async_bridge.rs index 793a640b0a..b855d16313 100644 --- a/crates/perry-stdlib/src/common/async_bridge.rs +++ b/crates/perry-stdlib/src/common/async_bridge.rs @@ -290,9 +290,9 @@ where /// whichever first. Because the producing task ran in this same tick, its /// completion is observed in-thread and `perry_poll` drains it on the next loop /// turn; there is no cross-thread wake to lose. `budget_ms` is the loop's -/// computed sleep budget (min of the next perry-timer deadline and the 1 s idle -/// cap); a zero budget is floored to 1 ms so native work still gets one poll -/// cycle under a hot timer. +/// computed sleep budget (min of the next Perry timer/native deadline and the +/// 1 s idle cap); a zero budget is floored to 1 ms so native work still gets +/// one poll cycle under a hot timer. extern "C" fn stdlib_wait_driver(budget_ms: u64) { run_one_tick(budget_ms); } @@ -490,6 +490,7 @@ pub fn ensure_pump_registered() { extern "C" { fn js_register_stdlib_pump(f: extern "C" fn() -> i32); fn js_register_stdlib_has_active(f: extern "C" fn() -> i32); + fn js_register_stdlib_next_wake(f: extern "C" fn() -> f64); fn js_stdlib_init_dispatch(); } ensure_gc_scanner_registered(); @@ -508,6 +509,7 @@ pub fn ensure_pump_registered() { unsafe { js_register_stdlib_pump(js_stdlib_process_pending); js_register_stdlib_has_active(js_stdlib_has_active_handles); + js_register_stdlib_next_wake(crate::readline::js_readline_next_wake_ms); // Wire up the runtime-level HANDLE_METHOD_DISPATCH so that // generic `jsObject.method(args)` calls on stdlib handle types // (net.Socket, Fastify, ioredis) fall back to the right FFI diff --git a/crates/perry-stdlib/src/readline/mod.rs b/crates/perry-stdlib/src/readline/mod.rs index 229f6fcbf0..22459d6329 100644 --- a/crates/perry-stdlib/src/readline/mod.rs +++ b/crates/perry-stdlib/src/readline/mod.rs @@ -32,6 +32,7 @@ use std::io::Read; use std::io::{self, Write}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Mutex; +use std::time::{Duration, Instant}; use perry_runtime::closure::{ get_valid_func_ptr, js_closure_alloc, js_closure_call0, js_closure_call1, js_closure_call2, @@ -122,6 +123,12 @@ static PENDING_DATA: Mutex>> = Mutex::new(Vec::new()); /// reader queues one byte per chunk, so `\x1b[A` arrives as three chunks /// and `pump::coalesce_escape_sequences` parks an incomplete prefix here. static PENDING_ESCAPE: Mutex> = Mutex::new(Vec::new()); +/// One-shot deadline for the escape prefix above. A wake from another timer or +/// producer must not make a held ESC look old enough to flush; only this clock +/// can do that (#9593). +static PENDING_ESCAPE_DEADLINE: Mutex> = Mutex::new(None); +/// Node's default `readline.escapeCodeTimeout`. +const ESCAPE_CODE_TIMEOUT: Duration = Duration::from_millis(500); /// Whether the one-shot `'readable'` EOF notification has been delivered. static READABLE_EOF_NOTIFIED: AtomicBool = AtomicBool::new(false); /// `true` when raw mode is enabled — the reader thread checks this @@ -1851,6 +1858,9 @@ pub extern "C" fn js_readline_stdin_destroy() -> f64 { if let Ok(mut p) = PENDING_ESCAPE.lock() { p.clear(); } + if let Ok(mut deadline) = PENDING_ESCAPE_DEADLINE.lock() { + *deadline = None; + } if let Ok(mut q) = PENDING_LINES.lock() { q.clear(); } @@ -1878,6 +1888,7 @@ pub extern "C" fn js_readline_stdin_destroy() -> f64 { // --------------------------------------------------------------------------- mod pump; +pub(crate) use pump::js_readline_next_wake_ms; pub use pump::{js_readline_has_active, js_readline_process_pending}; // --------------------------------------------------------------------------- diff --git a/crates/perry-stdlib/src/readline/mod_tests.rs b/crates/perry-stdlib/src/readline/mod_tests.rs index 5d3e0d5b78..fd7788c2fc 100644 --- a/crates/perry-stdlib/src/readline/mod_tests.rs +++ b/crates/perry-stdlib/src/readline/mod_tests.rs @@ -251,7 +251,7 @@ fn stdin_unref_and_destroy_release_active_state() { } #[test] -fn split_escape_sequence_reassembles_to_single_keypress() { +fn complete_escape_sequence_reassembles_to_single_keypress() { // The raw-mode reader queues one byte per chunk, so an arrow key // arrives as `\x1b`, `[`, `A` in three chunks. The pump must // reassemble them into ONE 'up' keypress, not escape + [ + A. @@ -268,10 +268,28 @@ fn split_escape_sequence_reassembles_to_single_keypress() { } #[test] -fn bare_escape_flushes_on_next_tick() { - // A lone ESC can't be distinguished from the start of a sequence - // within one tick — it's held, then flushed as a bare 'escape' - // keypress on the next tick if nothing followed. +fn split_escape_sequence_reassembles_before_timeout() { + // ESC can arrive in one read and the rest of an arrow in a later read. + // The completing bytes cancel the one-shot instead of observing a tick as + // an implicit timeout and tearing the arrow apart. + let _g = reset(); + let event = event_name("keypress"); + let cb = keypress_recorder_callback(); + let _ = js_readline_stdin_on(event, cb); + test_inject_chunk(b"\x1b"); + assert_eq!(js_readline_process_pending(), 0); + assert!(js_readline_next_wake_ms() > 0.0); + test_inject_chunk(b"["); + test_inject_chunk(b"C"); + assert_eq!(js_readline_process_pending(), 1); + KEYPRESS_NAMES.with(|names| assert_eq!(*names.borrow(), vec!["right".to_string()])); + assert_eq!(js_readline_next_wake_ms(), -1.0); +} + +#[test] +fn bare_escape_flushes_only_after_escape_code_timeout() { + // A lone ESC can't be distinguished from the start of a sequence. Pump + // turns before the deadline must leave it held; expiry emits one keypress. let _g = reset(); let event = event_name("keypress"); let cb = keypress_recorder_callback(); @@ -280,8 +298,13 @@ fn bare_escape_flushes_on_next_tick() { assert_eq!(js_readline_process_pending(), 0); // The held prefix keeps the loop alive so the flush tick runs. assert_eq!(js_readline_has_active(), 1); + let remaining = js_readline_next_wake_ms(); + assert!((1.0..=500.0).contains(&remaining)); + assert_eq!(js_readline_process_pending(), 0); + *PENDING_ESCAPE_DEADLINE.lock().unwrap() = Some(Instant::now() - Duration::from_millis(1)); assert_eq!(js_readline_process_pending(), 1); KEYPRESS_NAMES.with(|names| assert_eq!(*names.borrow(), vec!["escape".to_string()])); + assert_eq!(js_readline_next_wake_ms(), -1.0); } #[test] diff --git a/crates/perry-stdlib/src/readline/pump.rs b/crates/perry-stdlib/src/readline/pump.rs index b73f3ee6ac..c1aa4e84e3 100644 --- a/crates/perry-stdlib/src/readline/pump.rs +++ b/crates/perry-stdlib/src/readline/pump.rs @@ -130,24 +130,83 @@ fn escape_state(acc: &[u8]) -> EscState { } } +fn arm_escape_timeout() { + if let Ok(mut deadline) = PENDING_ESCAPE_DEADLINE.lock() { + *deadline = Some(Instant::now() + ESCAPE_CODE_TIMEOUT); + } +} + +fn cancel_escape_timeout() { + if let Ok(mut deadline) = PENDING_ESCAPE_DEADLINE.lock() { + *deadline = None; + } +} + +fn escape_timeout_expired() -> bool { + let Ok(mut deadline) = PENDING_ESCAPE_DEADLINE.lock() else { + return false; + }; + if deadline.is_some_and(|at| at <= Instant::now()) { + *deadline = None; + true + } else { + false + } +} + +/// Deadline provider registered with perry-runtime's event pump. Returning the +/// ceiling avoids truncating a sub-millisecond remainder to zero and spinning +/// before the timeout is actually due. +pub(crate) extern "C" fn js_readline_next_wake_ms() -> f64 { + if STDIN_DESTROYED.load(Ordering::Acquire) || STDIN_PAUSED.load(Ordering::Acquire) { + return -1.0; + } + let Ok(deadline) = PENDING_ESCAPE_DEADLINE.lock() else { + return -1.0; + }; + let Some(deadline) = *deadline else { + return -1.0; + }; + let now = Instant::now(); + if deadline <= now { + 0.0 + } else { + deadline.duration_since(now).as_millis().saturating_add(1) as f64 + } +} + /// Reassemble ANSI escape sequences that the raw-mode reader queues as /// individual 1-byte chunks (`\x1b`, `[`, `A` → one `\x1b[A` chunk) so a /// single arrow key fires a single `'keypress'`/`'data'` event, matching a /// terminal's one-write delivery. A sequence still incomplete at the end of -/// a drain batch is carried in [`PENDING_ESCAPE`] and finished by the next -/// tick's bytes; if that next tick brings no new bytes the held bytes flush -/// as-is, so a bare Escape keypress is delivered one tick later (a -/// tick-granularity stand-in for Node's `escapeCodeTimeout`). +/// a drain batch is carried in [`PENDING_ESCAPE`] and arms a one-shot +/// `escapeCodeTimeout`. Bytes that complete the sequence cancel that deadline; +/// expiry flushes the held bytes as-is. Event-loop ticks from unrelated work do +/// not advance this state (#9593). fn coalesce_escape_sequences(raw: Vec>) -> Vec> { let mut acc: Vec = PENDING_ESCAPE .lock() .map(|mut p| std::mem::take(&mut *p)) .unwrap_or_default(); if raw.is_empty() { - // No new bytes this tick: a held ESC prefix is a bare Escape (or a - // torn sequence from a very slow terminal) — deliver it byte-wise - // instead of holding it forever. - return acc.into_iter().map(|b| vec![b]).collect(); + if acc.is_empty() { + cancel_escape_timeout(); + return Vec::new(); + } + if escape_timeout_expired() { + return acc.into_iter().map(|b| vec![b]).collect(); + } + // A timer, I/O source, or stale notify can produce arbitrarily many + // pump turns before the deadline. Put the prefix back unchanged. + if let Ok(mut p) = PENDING_ESCAPE.lock() { + *p = acc; + } + return Vec::new(); + } + // Fresh bytes either complete/invalidate the held prefix or replace it + // with a new incomplete one below. In every case the old one-shot is done. + if !acc.is_empty() { + cancel_escape_timeout(); } let mut out: Vec> = Vec::with_capacity(raw.len()); for chunk in raw { @@ -176,6 +235,7 @@ fn coalesce_escape_sequences(raw: Vec>) -> Vec> { if !acc.is_empty() { if let Ok(mut p) = PENDING_ESCAPE.lock() { *p = acc; + arm_escape_timeout(); } } out @@ -196,6 +256,7 @@ pub extern "C" fn js_readline_process_pending() -> i32 { if let Ok(mut p) = PENDING_ESCAPE.lock() { p.clear(); } + cancel_escape_timeout(); Vec::new() } else if STDIN_PAUSED.load(Ordering::Acquire) { // Paused: leave the queue AND any held escape prefix untouched so @@ -434,8 +495,8 @@ pub extern "C" fn js_readline_has_active() -> i32 { let paused = STDIN_PAUSED.load(Ordering::Acquire); let refed = STDIN_REFED.load(Ordering::Acquire); let has_lines = PENDING_LINES.lock().map(|q| !q.is_empty()).unwrap_or(false); - // A held escape prefix counts as pending data: the loop must tick once - // more so the accumulator can flush it as a bare Escape keypress. + // A held escape prefix counts as pending data: the loop must stay alive + // until its explicit escapeCodeTimeout deadline can flush it. let has_data = PENDING_DATA.lock().map(|q| !q.is_empty()).unwrap_or(false) || PENDING_ESCAPE .lock() diff --git a/crates/perry-stdlib/src/readline/test_support.rs b/crates/perry-stdlib/src/readline/test_support.rs index 29c0c2d700..79676d6cea 100644 --- a/crates/perry-stdlib/src/readline/test_support.rs +++ b/crates/perry-stdlib/src/readline/test_support.rs @@ -76,6 +76,7 @@ pub(super) fn reset() -> MutexGuard<'static, ()> { PENDING_LINES.lock().unwrap().clear(); PENDING_DATA.lock().unwrap().clear(); PENDING_ESCAPE.lock().unwrap().clear(); + *PENDING_ESCAPE_DEADLINE.lock().unwrap() = None; EOF_REACHED.store(false, Ordering::Release); READABLE_EOF_NOTIFIED.store(false, Ordering::Release); STDIN_PAUSED.store(false, Ordering::Release); diff --git a/crates/perry/tests/issue_9593_readline_escape_timeout.rs b/crates/perry/tests/issue_9593_readline_escape_timeout.rs new file mode 100644 index 0000000000..b0ba79ae13 --- /dev/null +++ b/crates/perry/tests/issue_9593_readline_escape_timeout.rs @@ -0,0 +1,240 @@ +//! Regression test for #9593: a torn raw-mode ESC prefix must use a real +//! `escapeCodeTimeout`, not "the next event-loop tick" as its clock. +//! +//! The child runs on a fresh PTY so `setRawMode(true)` exercises the production +//! stdin reader. We cover the two loop schedules that exposed the defect: +//! with no other timer the old code delivered bare ESC at the 1 s idle cap, +//! while an unrelated 50 ms interval made it arrive after about 50 ms. Node's +//! default is 500 ms in both cases. + +#![cfg(unix)] + +use std::fs::File; +use std::io::{BufRead, BufReader, Write}; +use std::os::fd::{FromRawFd, RawFd}; +use std::os::unix::process::CommandExt; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError}; +use std::time::{Duration, Instant}; + +const SOURCE: &str = r#" +import { emitKeypressEvents } from "node:readline"; + +emitKeypressEvents(process.stdin); +process.stdin.setRawMode(true); + +if (process.env.PERRY_9593_INTERVAL === "1") { + setInterval(() => {}, 50); +} + +process.stdin.on("keypress", (_sequence, key) => { + console.log("KEY:" + key.name); +}); +console.log("READY"); +"#; + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +fn compile(dir: &Path) -> PathBuf { + let entry = dir.join("main.ts"); + let output = dir.join("main_bin"); + std::fs::write(&entry, SOURCE).expect("write PTY fixture"); + 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 open_pty() -> (File, File) { + let mut master: RawFd = -1; + let mut slave: RawFd = -1; + let rc = unsafe { + libc::openpty( + &mut master, + &mut slave, + std::ptr::null_mut(), + std::ptr::null(), + std::ptr::null(), + ) + }; + assert_eq!(rc, 0, "openpty failed: {}", std::io::Error::last_os_error()); + assert!(master >= 0 && slave >= 0); + // SAFETY: openpty returned two fresh, owned descriptors above. + unsafe { (File::from_raw_fd(master), File::from_raw_fd(slave)) } +} + +struct PtyChild { + child: Child, + input: File, + lines: Receiver<(String, Instant)>, +} + +impl PtyChild { + fn spawn(program: &Path, arm_interval: bool) -> Self { + let (master, slave) = open_pty(); + let child_stdin = slave.try_clone().expect("clone PTY slave for stdin"); + let child_stdout = slave.try_clone().expect("clone PTY slave for stdout"); + let mut command = Command::new(program); + command + .stdin(Stdio::from(child_stdin)) + .stdout(Stdio::from(child_stdout)) + .stderr(Stdio::null()) + .env("PERRY_9593_INTERVAL", if arm_interval { "1" } else { "0" }); + // Give the child its own session and make fd 0's PTY its controlling + // terminal. The stdio descriptors have already been installed when + // this hook runs. + unsafe { + command.pre_exec(|| { + if libc::setsid() == -1 { + return Err(std::io::Error::last_os_error()); + } + if libc::ioctl(0, libc::TIOCSCTTY as _, 0) == -1 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + }); + } + let child = command.spawn().expect("spawn PTY child"); + drop(slave); + + let input = master.try_clone().expect("clone PTY master for writes"); + let (tx, lines) = mpsc::channel(); + std::thread::spawn(move || { + for line in BufReader::new(master).lines() { + match line { + Ok(line) => { + if tx + .send((line.trim_end_matches('\r').to_string(), Instant::now())) + .is_err() + { + break; + } + } + // Linux returns EIO from a PTY master after the slave closes. + Err(_) => break, + } + } + }); + + let mut session = Self { + child, + input, + lines, + }; + let (ready, _) = session + .recv_line(Duration::from_secs(30)) + .expect("PTY child never printed READY"); + assert_eq!(ready, "READY", "unexpected first PTY line"); + session + } + + fn send(&mut self, bytes: &[u8]) { + self.input.write_all(bytes).expect("write PTY input"); + self.input.flush().expect("flush PTY input"); + } + + fn recv_line(&mut self, timeout: Duration) -> Result<(String, Instant), RecvTimeoutError> { + self.lines.recv_timeout(timeout) + } + + fn recv_key(&mut self, timeout: Duration) -> (String, Instant) { + let started = Instant::now(); + loop { + let remaining = timeout.saturating_sub(started.elapsed()); + let (line, arrived) = self + .recv_line(remaining) + .unwrap_or_else(|_| panic!("no keypress line within {timeout:?}")); + if let Some(name) = line.strip_prefix("KEY:") { + return (name.to_string(), arrived); + } + } + } + + fn assert_no_key_waiting(&mut self) { + while let Ok((line, _)) = self.lines.try_recv() { + assert!( + !line.starts_with("KEY:"), + "escape prefix fired before its completing bytes arrived: {line}" + ); + } + } +} + +impl Drop for PtyChild { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +fn bare_escape_latency(program: &Path, arm_interval: bool) -> Duration { + let mut child = PtyChild::spawn(program, arm_interval); + // Let the generated loop settle into its wait path before injecting input. + std::thread::sleep(Duration::from_millis(100)); + let sent = Instant::now(); + child.send(b"\x1b"); + let (name, arrived) = child.recv_key(Duration::from_secs(2)); + assert_eq!(name, "escape"); + arrived.duration_since(sent) +} + +#[test] +fn escape_timeout_is_clocked_and_split_arrows_still_coalesce() { + let dir = tempfile::tempdir().expect("create fixture directory"); + let program = compile(dir.path()); + + let idle_latency = bare_escape_latency(&program, false); + let timer_latency = bare_escape_latency(&program, true); + let expected_min = Duration::from_millis(250); + let expected_max = Duration::from_millis(900); + for (schedule, latency) in [("idle", idle_latency), ("50 ms interval", timer_latency)] { + assert!( + (expected_min..=expected_max).contains(&latency), + "bare ESC under {schedule} was delivered after {latency:?}; expected the Node-compatible \ + 500 ms escapeCodeTimeout, independent of the event-loop wait budget" + ); + } + assert!( + idle_latency.abs_diff(timer_latency) < Duration::from_millis(250), + "bare ESC changed with an unrelated timer: idle={idle_latency:?}, interval={timer_latency:?}" + ); + + let mut child = PtyChild::spawn(&program, false); + child.send(b"\x1b"); + std::thread::sleep(Duration::from_millis(200)); + child.assert_no_key_waiting(); + let completed = Instant::now(); + child.send(b"[C"); + let (name, arrived) = child.recv_key(Duration::from_secs(1)); + assert_eq!( + name, "right", + "split arrow was torn into separate keypresses" + ); + assert!( + arrived.duration_since(completed) < Duration::from_millis(250), + "split arrow was not delivered promptly after completion" + ); + + let sent = Instant::now(); + child.send(b"\x1b[C"); + let (name, arrived) = child.recv_key(Duration::from_secs(1)); + assert_eq!(name, "right"); + assert!( + arrived.duration_since(sent) < Duration::from_millis(250), + "complete arrow sequence should be delivered immediately" + ); +} From 73c5a9e40f3e903ef0ded1c9b8d114202b677562 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 3 Sep 2026 16:26:52 +0000 Subject: [PATCH 2/7] fix(child_process): cancel timeout threads after exit --- .../src/child_process/reactor.rs | 39 +++----- .../src/child_process/reactor/timeout.rs | 89 +++++++++++++++++++ .../test_gap_9592_child_timeout_threads.ts | 47 ++++++++++ 3 files changed, 148 insertions(+), 27 deletions(-) create mode 100644 crates/perry-runtime/src/child_process/reactor/timeout.rs create mode 100644 test-files/test_gap_9592_child_timeout_threads.ts diff --git a/crates/perry-runtime/src/child_process/reactor.rs b/crates/perry-runtime/src/child_process/reactor.rs index 25d024cc9b..c445c7deec 100644 --- a/crates/perry-runtime/src/child_process/reactor.rs +++ b/crates/perry-runtime/src/child_process/reactor.rs @@ -466,25 +466,6 @@ fn cp_spawn_reader(handle: u64, mut pipe: R, fd: usize }); } -/// Spawn the waiter thread that reaps `child` and reports its exit status. -fn cp_spawn_waiter(handle: u64, waiter: CpWaiter) { - std::thread::spawn(move || { - let (code, signal) = waiter(); - cp_push_event(CpEvent::Exited { - handle, - code, - signal, - }); - }); -} - -fn cp_spawn_timeout(handle: u64, timeout: Duration, signal: i32) { - std::thread::spawn(move || { - std::thread::sleep(timeout); - cp_push_event(CpEvent::Timeout { handle, signal }); - }); -} - /// IPC reader (#1933): read newline-delimited JSON from the parent socket and /// push each line for main-thread parse + `'message'` delivery. For /// `serialization: 'advanced'` (#2130) the framing is instead a 4-byte @@ -773,10 +754,11 @@ fn cp_register_live_child_parts( for (fd, _, pipe) in extra_pipes { cp_spawn_reader(handle, pipe, fd); } - cp_spawn_waiter(handle, waiter); - if let Some(timeout) = timeout { - cp_spawn_timeout(handle, timeout, kill_signal); - } + cp_spawn_waiter( + handle, + waiter, + timeout.map(|timeout| (timeout, kill_signal)), + ); #[cfg(any(unix, windows))] { if let Some(sock) = ipc { @@ -1390,10 +1372,11 @@ pub(super) fn cp_exec_async( } Err(_) => (Some(-1), None), }); - cp_spawn_waiter(handle, waiter); - if let Some(timeout) = timeout { - cp_spawn_timeout(handle, timeout, kill_signal); - } + cp_spawn_waiter( + handle, + waiter, + timeout.map(|timeout| (timeout, kill_signal)), + ); crate::event_pump::js_notify_main_thread(); cp } @@ -1961,9 +1944,11 @@ pub(super) fn cp_live_stdin_queue_callback(handle: u64, callback_bits: u64) -> b mod integration; mod kill; mod stdin_drain; +mod timeout; #[cfg(all(test, windows))] #[path = "reactor/windows_kill_tests.rs"] mod windows_kill_tests; pub(crate) use integration::*; pub(crate) use kill::*; use stdin_drain::*; +use timeout::*; diff --git a/crates/perry-runtime/src/child_process/reactor/timeout.rs b/crates/perry-runtime/src/child_process/reactor/timeout.rs new file mode 100644 index 0000000000..12f388f101 --- /dev/null +++ b/crates/perry-runtime/src/child_process/reactor/timeout.rs @@ -0,0 +1,89 @@ +//! Cancellable child-process timeout waiters. + +use super::*; +use std::sync::mpsc::{self, RecvTimeoutError}; + +/// Spawn the process waiter and, when configured, a timeout waiter. The +/// process waiter owns the sole cancellation sender, so a completed child +/// wakes its timeout waiter instead of leaving an OS thread asleep until the +/// original deadline. +pub(super) fn cp_spawn_waiter(handle: u64, waiter: CpWaiter, timeout: Option<(Duration, i32)>) { + let timeout_cancel = timeout.map(|(timeout, signal)| { + let (cancel, cancelled) = mpsc::channel(); + std::thread::spawn(move || { + if matches!( + cancelled.recv_timeout(timeout), + Err(RecvTimeoutError::Timeout) + ) { + cp_push_event(CpEvent::Timeout { handle, signal }); + } + }); + cancel + }); + + std::thread::spawn(move || { + let (code, signal) = waiter(); + if let Some(cancel) = timeout_cancel { + let _ = cancel.send(()); + } + cp_push_event(CpEvent::Exited { + handle, + code, + signal, + }); + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Barrier}; + use std::time::Instant; + + #[test] + fn completed_children_wake_fifty_long_timeout_waiters() { + const CHILDREN: usize = 50; + let ready = Arc::new(Barrier::new(CHILDREN + 1)); + let mut cancels = Vec::with_capacity(CHILDREN); + let mut completions = Vec::with_capacity(CHILDREN); + + for _ in 0..CHILDREN { + let (cancel, cancelled) = mpsc::channel(); + let (completed, completion) = mpsc::channel(); + let ready = Arc::clone(&ready); + std::thread::spawn(move || { + ready.wait(); + let timed_out = matches!( + cancelled.recv_timeout(Duration::from_secs(60)), + Err(RecvTimeoutError::Timeout) + ); + let _ = completed.send(timed_out); + }); + cancels.push(cancel); + completions.push(completion); + } + + ready.wait(); + let started = Instant::now(); + for cancel in cancels { + cancel.send(()).expect("timeout waiter should be alive"); + } + for completion in completions { + assert_eq!( + completion.recv_timeout(Duration::from_secs(1)), + Ok(false), + "timeout waiter did not stop after child completion" + ); + } + assert!(started.elapsed() < Duration::from_secs(1)); + } + + #[test] + fn elapsed_deadline_still_selects_the_timeout_arm() { + let (_cancel, cancelled) = mpsc::channel::<()>(); + assert!(matches!( + cancelled.recv_timeout(Duration::from_millis(10)), + Err(RecvTimeoutError::Timeout) + )); + } +} diff --git a/test-files/test_gap_9592_child_timeout_threads.ts b/test-files/test_gap_9592_child_timeout_threads.ts new file mode 100644 index 0000000000..37204e2362 --- /dev/null +++ b/test-files/test_gap_9592_child_timeout_threads.ts @@ -0,0 +1,47 @@ +// #9592 — a timed spawn must not retain its timeout OS thread after the child +// exits. The Linux thread census catches the old one-sleeper-per-timeout leak; +// the slow-child arm keeps the actual timeout behavior covered everywhere. +import { spawn } from "node:child_process"; +import { readdirSync } from "node:fs"; + +function close(child: any): Promise { + return new Promise((resolve) => child.on("close", () => resolve())); +} + +function threadCount(): number { + return process.platform === "linux" ? readdirSync("/proc/self/task").length : 0; +} + +const baseline = threadCount(); +const quickChildren: Promise[] = []; +for (let i = 0; i < 50; i++) { + quickChildren.push( + close(spawn("/bin/true", [], { stdio: "ignore", timeout: 60_000 })), + ); +} +await Promise.all(quickChildren); + +let timeoutThreadsReleased = process.platform !== "linux"; +const releaseDeadline = Date.now() + 1_000; +while (!timeoutThreadsReleased && Date.now() < releaseDeadline) { + timeoutThreadsReleased = threadCount() <= baseline + 5; + if (!timeoutThreadsReleased) { + await new Promise((resolve) => setTimeout(resolve, 20)); + } +} +console.log("timeout threads released:", timeoutThreadsReleased); + +const started = Date.now(); +const slow = spawn("/bin/sleep", ["30"], { + stdio: "ignore", + timeout: 100, +}); +await close(slow); +const elapsed = Date.now() - started; +console.log( + "slow child killed on time:", + slow.killed && + slow.signalCode === "SIGTERM" && + elapsed >= 50 && + elapsed < 5_000, +); From b7bbedac76abeeba051247c09451ad195d7993f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 3 Sep 2026 16:28:01 +0000 Subject: [PATCH 3/7] docs(changelog): record child timeout cleanup --- changelog.d/9640-child-timeout-threads.md | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 changelog.d/9640-child-timeout-threads.md diff --git a/changelog.d/9640-child-timeout-threads.md b/changelog.d/9640-child-timeout-threads.md new file mode 100644 index 0000000000..2b243e2875 --- /dev/null +++ b/changelog.d/9640-child-timeout-threads.md @@ -0,0 +1,3 @@ +Fixed timed `child_process.spawn`, `exec`, and `execFile` calls retaining +one sleeping OS thread until the full timeout after a child had already exited. +Timeout workers now wake on child completion while preserving deadline kills. From c2d41fe84808c4b476bf33399cc2ba1b444cb6bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 3 Sep 2026 16:37:29 +0000 Subject: [PATCH 4/7] test(child_process): document timeout census helpers --- test-files/test_gap_9592_child_timeout_threads.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/test-files/test_gap_9592_child_timeout_threads.ts b/test-files/test_gap_9592_child_timeout_threads.ts index 37204e2362..778e5fddaa 100644 --- a/test-files/test_gap_9592_child_timeout_threads.ts +++ b/test-files/test_gap_9592_child_timeout_threads.ts @@ -4,10 +4,12 @@ import { spawn } from "node:child_process"; import { readdirSync } from "node:fs"; +/** Resolve after the child process and its stdio handles have closed. */ function close(child: any): Promise { return new Promise((resolve) => child.on("close", () => resolve())); } +/** Count this process's live OS threads when Linux exposes the task census. */ function threadCount(): number { return process.platform === "linux" ? readdirSync("/proc/self/task").length : 0; } From 4b4dab6997987ed58b63e1f709b461dcbd653bdf 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 5/7] 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 22459d6329..8fa5ab39ea 100644 --- a/crates/perry-stdlib/src/readline/mod.rs +++ b/crates/perry-stdlib/src/readline/mod.rs @@ -134,9 +134,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. @@ -1417,6 +1423,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 @@ -1532,8 +1544,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) { @@ -1547,7 +1559,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. @@ -1851,6 +1866,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 fd7788c2fc..d653a62f02 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 c1aa4e84e3..c49349f639 100644 --- a/crates/perry-stdlib/src/readline/pump.rs +++ b/crates/perry-stdlib/src/readline/pump.rs @@ -424,14 +424,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'`. @@ -458,12 +462,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. @@ -516,12 +524,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 @@ -534,7 +545,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 79676d6cea..8e2cc0c22d 100644 --- a/crates/perry-stdlib/src/readline/test_support.rs +++ b/crates/perry-stdlib/src/readline/test_support.rs @@ -78,6 +78,7 @@ pub(super) fn reset() -> MutexGuard<'static, ()> { PENDING_ESCAPE.lock().unwrap().clear(); *PENDING_ESCAPE_DEADLINE.lock().unwrap() = None; 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:?}" + ); +} From 9e830153c7ce063e7587e9f90f3d0064f6f8e5a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 3 Sep 2026 20:06:10 +0200 Subject: [PATCH 6/7] gc: price the idle reclaim by swept bytes, not old-gen occupancy (#9589) The reducer scored a full productive only when `old_gen_in_use_bytes` dropped. That number is the sum of the live old blocks' bump offsets, and a non-moving sweep cannot lower it: dead objects go back to the old-gen free list and the block keeps its offset, so only a whole-block release moves it. The unit tests passed because their litter fills blocks that end up entirely dead; every real workload does not. Measured on the compiled claude-code TUI, in a session with 279 collections behind it: two reducer fulls freed 2.37 MB and 0.51 MB while occupancy stood at 93.6 MB with 50 MB already on the old-gen free list, both scored unproductive, and the activity requirement had doubled twice inside a minute of idle. The reducer was switching itself off on exactly the workload it was built for. Price the full by the freed-bytes count the cycle already reports. The occupancy delta stays in the counters and on the diag line as the whole-block-release signal it actually is, next to a new `reusable=` field: the free-list residue that only compaction can return to the OS, which a budgeted (non-moving) cycle cannot do. Claude-Session: https://claude.ai/code/session_011qE5TRRJFzqxN44K34AnG2 --- changelog.d/9589-idle-reclaim-productivity.md | 1 + crates/perry-runtime/src/gc/idle_reclaim.rs | 45 ++++++++++++--- .../src/gc/tests/idle_reclaim.rs | 55 +++++++++++++++++++ 3 files changed, 93 insertions(+), 8 deletions(-) create mode 100644 changelog.d/9589-idle-reclaim-productivity.md diff --git a/changelog.d/9589-idle-reclaim-productivity.md b/changelog.d/9589-idle-reclaim-productivity.md new file mode 100644 index 0000000000..fd973578b7 --- /dev/null +++ b/changelog.d/9589-idle-reclaim-productivity.md @@ -0,0 +1 @@ +fix(gc): the idle-time reclaim now prices a full by what its sweep freed instead of by the old-gen occupancy delta (#9589). `arena::old_gen_in_use_bytes` sums the live old blocks' bump offsets, and a non-moving sweep cannot lower it — dead objects go back to the old-gen free list and the block keeps its offset — so the occupancy delta read ~0 for cycles that freed megabytes and the reducer backed off on every real workload. Measured on the compiled claude-code TUI: two reducer fulls freed 2.37 MB and 0.51 MB with occupancy flat at 93.6 MB and 50 MB already on the free list, and the activity requirement had doubled twice within a minute of idle. The `[gc-idle-reclaim] done` line now also reports the bar it was priced against and `reusable=`, the old-gen free-list residue that only compaction can return to the OS. diff --git a/crates/perry-runtime/src/gc/idle_reclaim.rs b/crates/perry-runtime/src/gc/idle_reclaim.rs index 273fa1e39d..eb7a32c729 100644 --- a/crates/perry-runtime/src/gc/idle_reclaim.rs +++ b/crates/perry-runtime/src/gc/idle_reclaim.rs @@ -58,9 +58,10 @@ //! 3. **Rate.** At least [`IDLE_RECLAIM_MIN_INTERVAL_MS`] since the reducer's //! own last full started. //! -//! **Productivity backoff.** A full that lowered old-gen occupancy by less than -//! [`IDLE_RECLAIM_PRODUCTIVE_PCT`] percent of what it started with, or by less -//! than [`IDLE_RECLAIM_PRODUCTIVE_MIN_BYTES`], doubles the activity requirement +//! **Productivity backoff.** A full whose SWEEP freed less than +//! [`IDLE_RECLAIM_PRODUCTIVE_PCT`] percent of the old-gen occupancy it started +//! with, or less than [`IDLE_RECLAIM_PRODUCTIVE_MIN_BYTES`], doubles the +//! activity requirement //! up to `2^`[`IDLE_RECLAIM_MAX_BACKOFF_SHIFT`] collections; a productive full //! resets it to one. This is what keeps a RETAINING idle heap (the TUI's //! six-second young-collection sawtooth) from paying a whole-heap mark per @@ -68,6 +69,19 @@ //! collections, and any burst — which produces many collections in a row — //! re-arms it promptly. //! +//! The price is the cycle's own freed-bytes count and NOT the change in +//! `arena::old_gen_in_use_bytes`, which is the sum of the live old blocks' +//! bump offsets: a non-moving sweep hands dead objects to the old-gen free +//! list and the block keeps its offset, so only a whole-block release moves +//! that number. Pricing on it scored every full on a real workload +//! unproductive — measured on the compiled claude-code TUI, two reducer fulls +//! freed 2.37 MB and 0.51 MB while occupancy stood still at 93.6 MB with +//! 50 MB already on the old-gen free list, and the reducer backed off to one +//! attempt per four collections inside a minute of idle. Returning that free +//! list to the OS needs compaction, which a budgeted (non-moving) cycle +//! cannot do; the `reusable=` field on the `[gc-idle-reclaim] done` line +//! reports how much is waiting for it. +//! //! **Work cap.** While a cycle is open the hook never spends more than //! [`IDLE_RECLAIM_MAX_WORK_MS_PER_SECOND`] of any wall-clock second stepping //! it; past that it lets the loop park for the rest of its budget. A healthy @@ -97,9 +111,9 @@ pub const IDLE_RECLAIM_QUIET_MS: u64 = 3_000; /// Minimum spacing between two reducer-started fulls. pub const IDLE_RECLAIM_MIN_INTERVAL_MS: u64 = 10_000; -/// A reducer full counts as productive when it lowered old-gen occupancy by at -/// least this many bytes AND by at least [`IDLE_RECLAIM_PRODUCTIVE_PCT`] percent -/// of the occupancy it started with. +/// A reducer full counts as productive when its sweep freed at least this many +/// bytes AND at least [`IDLE_RECLAIM_PRODUCTIVE_PCT`] percent of the old-gen +/// occupancy it started with. pub const IDLE_RECLAIM_PRODUCTIVE_MIN_BYTES: usize = 4 * 1024 * 1024; /// See [`IDLE_RECLAIM_PRODUCTIVE_MIN_BYTES`]. @@ -365,7 +379,14 @@ pub(super) fn note_cycle_completed(freed_bytes: u64) { let reclaimed = before.saturating_sub(after); OLD_RECLAIMED_BYTES.fetch_add(reclaimed as u64, Ordering::Relaxed); let bar = IDLE_RECLAIM_PRODUCTIVE_MIN_BYTES.max(before / 100 * IDLE_RECLAIM_PRODUCTIVE_PCT); - let productive = reclaimed >= bar; + // Price the full by what its sweep freed. `old_gen_occupancy` sums the + // live blocks' bump offsets, which a non-moving sweep cannot lower — + // it returns objects to the old-gen free list and the block keeps its + // offset — so the occupancy delta reads ~0 for a cycle that freed + // megabytes, and every real full scored unproductive (module docs). + // The delta stays in the counters and the trace as the whole-block + // release signal it actually is. + let productive = freed_bytes >= bar as u64; if productive { PRODUCTIVE.fetch_add(1, Ordering::Relaxed); st.backoff_shift = 0; @@ -379,7 +400,8 @@ pub(super) fn note_cycle_completed(freed_bytes: u64) { st.last_seen_external = external; if gc_diag_enabled() { eprintln!( - "[gc-idle-reclaim] done old_in_use={before}->{after} reclaimed_old={reclaimed} freed={freed_bytes} productive={productive} backoff_shift={}", + "[gc-idle-reclaim] done old_in_use={before}->{after} reclaimed_old={reclaimed} freed={freed_bytes} bar={bar} reusable={} productive={productive} backoff_shift={}", + old_free_bytes(), st.backoff_shift ); } @@ -526,6 +548,13 @@ pub(super) mod test_support { STATE.with(|s| *s.borrow_mut() = IdleReclaimState::default()); } + /// Pin the old-gen occupancy the in-flight full started with, so a + /// completion can be priced against a known `before` without reproducing + /// the allocation pattern that would produce one. + pub(crate) fn set_old_in_use_at_start(bytes: usize) { + STATE.with(|s| s.borrow_mut().old_in_use_at_start = bytes); + } + pub(crate) fn thread_attempts() -> u64 { STATE.with(|s| s.borrow().attempts) } diff --git a/crates/perry-runtime/src/gc/tests/idle_reclaim.rs b/crates/perry-runtime/src/gc/tests/idle_reclaim.rs index 8133dcfdbe..4b37368bac 100644 --- a/crates/perry-runtime/src/gc/tests/idle_reclaim.rs +++ b/crates/perry-runtime/src/gc/tests/idle_reclaim.rs @@ -419,3 +419,58 @@ fn idle_reclaim_full_reaches_the_allocator_purge() { ); } } + +/// A full is priced by what its SWEEP freed, not by the old-gen occupancy +/// delta. `arena::old_gen_in_use_bytes` is the sum of the live blocks' bump +/// offsets: a non-moving sweep returns dead objects to the old-gen free list +/// and the block keeps its offset, so a cycle that freed megabytes can leave +/// it unchanged. That is not a corner case — it is what the compiled TUI does +/// (2.37 MB freed, occupancy flat at 93.6 MB, scored unproductive, backoff to +/// one attempt per four collections inside a minute of idle). +#[test] +fn a_full_is_productive_on_swept_bytes_when_occupancy_cannot_move() { + let _reducer = IdleReclaimTestGuard::new(0); + // Price against the live occupancy, so the completion below sees no delta + // at all — the production shape, reproduced without the fragmentation. + let before = crate::arena::old_gen_in_use_bytes(); + set_old_in_use_at_start(before); + let bar = IDLE_RECLAIM_PRODUCTIVE_MIN_BYTES.max(before / 100 * IDLE_RECLAIM_PRODUCTIVE_PCT); + let productive_before = idle_reclaim_productive(); + + super::super::idle_reclaim::note_cycle_completed(bar as u64); + + assert_eq!( + idle_reclaim_productive(), + productive_before + 1, + "a sweep that freed the bar is productive however the block offsets read" + ); + assert_eq!( + idle_reclaim_backoff_shift(), + 0, + "a productive full resets the activity requirement" + ); +} + +/// The mirror: freeing less than the bar still backs off, so the pricing +/// change cannot be satisfied by calling every full productive. +#[test] +fn a_full_that_freed_less_than_the_bar_still_backs_off() { + let _reducer = IdleReclaimTestGuard::new(0); + let before = crate::arena::old_gen_in_use_bytes(); + set_old_in_use_at_start(before); + let bar = IDLE_RECLAIM_PRODUCTIVE_MIN_BYTES.max(before / 100 * IDLE_RECLAIM_PRODUCTIVE_PCT); + let productive_before = idle_reclaim_productive(); + + super::super::idle_reclaim::note_cycle_completed(bar as u64 - 1); + + assert_eq!( + idle_reclaim_productive(), + productive_before, + "one byte short of the bar is not productive" + ); + assert_eq!( + idle_reclaim_backoff_shift(), + 1, + "an unproductive full doubles the activity requirement" + ); +} From cc92b17425eb8e7288827a3c8d91ec250e8b68c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 3 Sep 2026 20:50:12 +0200 Subject: [PATCH 7/7] =?UTF-8?q?fix(test):=20openpty's=20trailing=20params?= =?UTF-8?q?=20are=20*mut=20on=20macOS,=20*const=20on=20Linux=20=E2=80=94?= =?UTF-8?q?=20null=5Fmut()=20coerces=20on=20both=20(#9593=20fixture)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/perry/tests/issue_9593_readline_escape_timeout.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/perry/tests/issue_9593_readline_escape_timeout.rs b/crates/perry/tests/issue_9593_readline_escape_timeout.rs index b0ba79ae13..1660930579 100644 --- a/crates/perry/tests/issue_9593_readline_escape_timeout.rs +++ b/crates/perry/tests/issue_9593_readline_escape_timeout.rs @@ -63,12 +63,14 @@ fn open_pty() -> (File, File) { let mut master: RawFd = -1; let mut slave: RawFd = -1; let rc = unsafe { + // `null_mut()` for all three: macOS types the trailing termios/winsize + // params `*mut`, Linux `*const`, and `*mut` coerces to `*const`. libc::openpty( &mut master, &mut slave, std::ptr::null_mut(), - std::ptr::null(), - std::ptr::null(), + std::ptr::null_mut(), + std::ptr::null_mut(), ) }; assert_eq!(rc, 0, "openpty failed: {}", std::io::Error::last_os_error());