From 0acc82103d1a68d356124d5b96c62f2d5f7d4e9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Mon, 21 Sep 2026 17:47:59 +0200 Subject: [PATCH 1/2] fix(runtime): a process.stdin resume() that races the stopping fd-0 reader no longer strands stdin without a reader (#10895) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The async iterator pauses its source after every delivered chunk and resumes it on the next pull. On process.stdin, pause() latches STDIN_DETACHED — the fd-0 reader thread exits when it sees it at the top of its loop — and resume() clears the latch and respawns the reader unless STDIN_READER_STARTED says one is still running. The reader's stop decision and its STARTED reset were two separate steps, so a resume() that landed between them found STARTED still true, spawned nothing, and the old reader then left: fd 0 had no reader while every liveness view still reported an open, flowing stdin, and the process idled forever with input unread. Make the reader's check-and-clear and the restart CAS atomic with respect to each other under one lifecycle lock (never held across read()). The detach exit now releases the reader slot itself and disarms the drop guard, which otherwise could clobber the flag of a reader respawned in between. Introduced by bb573926b5 (2026-09-04, unified fd-0 reader): before it, a piped stdin was read by readline's own reader, which never consulted the latch, so v0.5.1520 does not reproduce. Fixes #10895 --- .../perry-runtime/src/os_process_streams.rs | 184 ++++++++++++++++-- .../tests/issue_10895_stdin_pipe_stall.rs | 152 +++++++++++++++ .../test_issue_10895_stdin_pipe_stall.ts | 32 +++ 3 files changed, 357 insertions(+), 11 deletions(-) create mode 100644 crates/perry/tests/issue_10895_stdin_pipe_stall.rs create mode 100644 test-files/test_issue_10895_stdin_pipe_stall.ts diff --git a/crates/perry-runtime/src/os_process_streams.rs b/crates/perry-runtime/src/os_process_streams.rs index e6e585f8e5..5ea11f954c 100644 --- a/crates/perry-runtime/src/os_process_streams.rs +++ b/crates/perry-runtime/src/os_process_streams.rs @@ -349,25 +349,101 @@ pub extern "C" fn js_register_stdin_reader_consumer( } } -fn ensure_stdin_reader() { +/// Serializes the reader's decision to STOP with every request to (re)START it +/// (#10895). +/// +/// `pause()` sets `STDIN_DETACHED`; the reader notices at the top of its loop +/// and exits, and `resume()` clears the latch and calls `ensure_stdin_reader`, +/// which spawns a reader only when `STDIN_READER_STARTED` is false. Those two +/// flags used to be read and written independently, so this interleaving lost +/// the restart for good: +/// +/// reader: sees `STDIN_DETACHED == true`, decides to exit +/// main: `resume()` → `STDIN_DETACHED = false`; CAS(STARTED: false→true) +/// FAILS — the dying reader has not cleared STARTED yet +/// reader: clears STARTED and is gone +/// +/// fd 0 then has no reader while every liveness view still says stdin is open +/// and flowing, so the process idles forever with input unread. The async +/// iterator pauses/resumes the source once per delivered chunk, so a piped +/// `for await (const chunk of process.stdin)` rolled this dice hundreds of +/// times per megabyte. +/// +/// Holding this lock across the reader's check-and-clear and across the +/// restart CAS makes the two atomic with respect to each other: a restart +/// request either runs entirely before the stop decision (the reader then sees +/// the cleared latch and keeps going) or entirely after it (STARTED is already +/// false, so a fresh reader is spawned). It is never held across `read()`. +static STDIN_READER_LIFECYCLE: std::sync::Mutex<()> = std::sync::Mutex::new(()); + +fn stdin_reader_lifecycle() -> std::sync::MutexGuard<'static, ()> { + STDIN_READER_LIFECYCLE + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// The stop half of the lifecycle handshake, over an explicit slot flag so the +/// unit tests can replay interleavings without touching the process-global +/// one. `should_stop` is evaluated UNDER the lock, and a true answer releases +/// the slot in the same step. +fn reader_slot_claim_stop( + started: &std::sync::atomic::AtomicBool, + should_stop: impl FnOnce() -> bool, +) -> bool { + let _lifecycle = stdin_reader_lifecycle(); + if should_stop() { + started.store(false, std::sync::atomic::Ordering::Release); + true + } else { + false + } +} + +/// The restart half: true when the caller now owns the (single) reader slot +/// and must spawn the reader. +fn reader_slot_claim_start(started: &std::sync::atomic::AtomicBool) -> bool { use std::sync::atomic::Ordering; - // A previous reader may have exited (EOF, error, or explicit detach); its - // drop guard resets `STDIN_READER_STARTED` to false, - // so a later `resume()`/`on(...)` can spin up a fresh reader. - if STDIN_READER_STARTED + let _lifecycle = stdin_reader_lifecycle(); + started .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) .is_ok() - { +} + +/// The reader's top-of-loop stop check. Returns true when the reader must +/// exit; STARTED has then ALREADY been cleared, under the lifecycle lock, so +/// the caller must not clear it again (a late clear would clobber the `true` +/// of a reader respawned in between and let a third one start). +fn stdin_reader_claim_stop() -> bool { + reader_slot_claim_stop(&STDIN_READER_STARTED, stdin_reader_should_stop) +} + +fn stdin_reader_claim_start() -> bool { + reader_slot_claim_start(&STDIN_READER_STARTED) +} + +fn ensure_stdin_reader() { + // A previous reader may have exited (EOF, error, or explicit detach), which + // resets `STDIN_READER_STARTED` to false so a later `resume()`/`on(...)` + // can spin up a fresh reader. The claim is atomic with a live reader's + // decision to stop (#10895). + if stdin_reader_claim_start() { std::thread::spawn(|| { use std::io::Read; - // On exit, clear STARTED so the reader can be restarted later. - struct ReaderGuard; + // On an EOF / error / panic exit, clear STARTED so the reader can + // be restarted later. The detach exit clears it itself, inside + // `stdin_reader_claim_stop`, and disarms this guard. + struct ReaderGuard { + armed: bool, + } impl Drop for ReaderGuard { fn drop(&mut self) { - STDIN_READER_STARTED.store(false, std::sync::atomic::Ordering::Release); + if self.armed { + let _lifecycle = stdin_reader_lifecycle(); + STDIN_READER_STARTED.store(false, std::sync::atomic::Ordering::Release); + } } } - let _guard = ReaderGuard; + let mut guard = ReaderGuard { armed: true }; let stdin = std::io::stdin(); let mut handle = stdin.lock(); // Read in chunks, not one byte at a time. A paste or a fast-typed @@ -387,7 +463,11 @@ fn ensure_stdin_reader() { // #9676: `stdin_reader_should_stop`, NOT `stdin_is_detached` — // an `unref()`d stdin still delivers data in Node, and reading // the liveness view here is what killed the reader for good. - if stdin_reader_should_stop() { + // #10895: the check and the STARTED reset are one step under + // the lifecycle lock, so a concurrent `resume()` can never + // find STARTED still true for a reader that is already leaving. + if stdin_reader_claim_stop() { + guard.armed = false; break; } match handle.read(&mut buf) { @@ -1266,6 +1346,88 @@ mod empty_checkpoint_tests { } } +#[cfg(test)] +mod reader_lifecycle_tests { + use super::{reader_slot_claim_start, reader_slot_claim_stop}; + use std::sync::atomic::{AtomicBool, Ordering}; + + /// #10895: replays the interleaving that stranded fd 0 without a reader — + /// the reader decides to stop, and `resume()` asks for a restart BEFORE + /// the dying reader has finished leaving. The stop decision must already + /// have released the reader slot, or the restart's claim fails and nobody + /// ever reads stdin again. + /// + /// Runs on local flags: no fd-0 reader is spawned and no process-global + /// stdin state is touched, so it cannot disturb the liveness tests. + #[test] + fn a_restart_requested_while_the_reader_is_leaving_is_not_lost() { + // A reader is running and `pause()` has latched the detach. + let started = AtomicBool::new(true); + let detached = AtomicBool::new(true); + assert!( + reader_slot_claim_stop(&started, || detached.load(Ordering::Acquire)), + "a detached reader must decide to stop" + ); + // `resume()`: clear the latch, then ask for a reader. The old reader + // has not run another instruction since its stop decision. + detached.store(false, Ordering::Release); + assert!( + reader_slot_claim_start(&started), + "restart lost: the stopping reader still held the reader slot" + ); + // The respawned reader owns the slot; a second request is a no-op. + assert!(!reader_slot_claim_start(&started)); + } + + /// The other order: `resume()` clears the latch before the reader looks. + /// The reader keeps running and no second reader may be started on fd 0. + #[test] + fn a_resume_that_beats_the_stop_check_keeps_the_one_reader() { + let started = AtomicBool::new(true); + let detached = AtomicBool::new(true); + detached.store(false, Ordering::Release); + assert!(!reader_slot_claim_start(&started)); + assert!(!reader_slot_claim_stop(&started, || detached.load(Ordering::Acquire))); + assert!(started.load(Ordering::Acquire)); + } + + /// Hammer the handshake from two threads: a "reader" that stops whenever + /// it sees the latch and a "main" that pauses/resumes. After every + /// resume the slot must be owned — by the surviving reader or by the + /// restart — never stranded. + #[test] + fn pause_resume_storm_never_strands_the_slot() { + use std::sync::Arc; + let started = Arc::new(AtomicBool::new(true)); + let detached = Arc::new(AtomicBool::new(false)); + let done = Arc::new(AtomicBool::new(false)); + let reader = { + let (started, detached, done) = (started.clone(), detached.clone(), done.clone()); + std::thread::spawn(move || { + while !done.load(Ordering::Acquire) { + // A live reader polls the latch between reads; one that + // stopped waits to be "respawned" by main's claim. + if started.load(Ordering::Acquire) { + reader_slot_claim_stop(&started, || detached.load(Ordering::Acquire)); + } + std::hint::spin_loop(); + } + }) + }; + for _ in 0..200_000 { + detached.store(true, Ordering::Release); // pause() + detached.store(false, Ordering::Release); // resume(): clear … + reader_slot_claim_start(&started); // … then ensure a reader + assert!( + started.load(Ordering::Acquire), + "resume() returned with no reader owning fd 0" + ); + } + done.store(true, Ordering::Release); + reader.join().unwrap(); + } +} + fn pump_stdin_data_chunks() { let has_bytes = STDIN_BUFFER.lock().map(|b| !b.is_empty()).unwrap_or(false); if !has_bytes { diff --git a/crates/perry/tests/issue_10895_stdin_pipe_stall.rs b/crates/perry/tests/issue_10895_stdin_pipe_stall.rs new file mode 100644 index 0000000000..f3203aa0cd --- /dev/null +++ b/crates/perry/tests/issue_10895_stdin_pipe_stall.rs @@ -0,0 +1,152 @@ +//! Regression coverage for #10895: `for await (const chunk of process.stdin)` +//! on a pipe stalled forever part-way through the input. +//! +//! The async iterator pauses its source after every delivered chunk and +//! resumes it on the next pull. On `process.stdin`, `pause()` latches +//! `STDIN_DETACHED` (the fd-0 reader thread exits when it sees it) and +//! `resume()` clears the latch and respawns the reader unless one is still +//! registered. A `resume()` that landed while the old reader was on its way +//! out found it still registered, spawned nothing, and the old reader then +//! left: no reader on fd 0, every liveness view still reporting an open, +//! flowing stdin, the process idle forever with input unread. +//! +//! It is a race, so this test is statistical by nature: many small writes +//! (each pause/resume cycle is one roll) over several rounds. On an unpatched +//! build a single 8 MiB round fed in 256-byte writes stalls roughly every +//! second time on macOS and rarely on Linux; the deterministic witness for +//! the interleaving itself is `reader_lifecycle_tests` in +//! `perry-runtime/src/os_process_streams.rs`. + +#![cfg(unix)] + +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +const SOURCE: &str = include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../test-files/test_issue_10895_stdin_pipe_stall.ts" +)); + +const TOTAL_BYTES: usize = 8 * 1024 * 1024; +const WRITE_BYTES: usize = 256; +const ROUNDS: usize = 12; +const ROUND_DEADLINE: Duration = Duration::from_secs(60); + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +fn compile(dir: &Path) -> PathBuf { + let source = dir.join("stdin_pipe_stall.ts"); + let binary = dir.join("stdin_pipe_stall_bin"); + std::fs::write(&source, SOURCE).expect("write stdin fixture"); + let output = Command::new(perry_bin()) + .current_dir(dir) + .arg("compile") + .arg(&source) + .arg("-o") + .arg(&binary) + .output() + .expect("compile stdin fixture"); + assert!( + output.status.success(), + "fixture compile failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + binary +} + +/// One round: pipe `TOTAL_BYTES` in `WRITE_BYTES` writes, close stdin, and +/// require the child to report every byte before the deadline. +fn run_round(binary: &Path, round: usize) { + let mut child = Command::new(binary) + .env("PERRY_10895_DRIVE", "1") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn stdin fixture"); + let mut stdin = child.stdin.take().expect("fixture stdin"); + let writer = std::thread::spawn(move || { + let piece = [1u8; WRITE_BYTES]; + let mut left = TOTAL_BYTES; + while left > 0 { + let n = left.min(WRITE_BYTES); + // A stalled child stops draining the pipe; the kill below then + // breaks this write with EPIPE. Either way the thread ends. + if stdin.write_all(&piece[..n]).is_err() { + return left; + } + left -= n; + } + drop(stdin); + 0 + }); + + let started = Instant::now(); + let status = loop { + if let Some(status) = child.try_wait().expect("poll stdin fixture") { + break Some(status); + } + if started.elapsed() > ROUND_DEADLINE { + break None; + } + std::thread::sleep(Duration::from_millis(5)); + }; + let Some(status) = status else { + let _ = child.kill(); + let _ = child.wait(); + let unwritten = writer.join().unwrap_or(TOTAL_BYTES); + panic!( + "round {round}: piped stdin stalled — the child was still alive after {:?} with \ + {unwritten} of {TOTAL_BYTES} bytes not even accepted by the pipe (#10895)", + ROUND_DEADLINE + ); + }; + assert_eq!(writer.join().expect("writer thread"), 0); + let mut stdout = String::new(); + child + .stdout + .take() + .expect("fixture stdout") + .read_to_string(&mut stdout) + .expect("read fixture stdout"); + assert!( + status.success(), + "round {round}: exit {status:?}: {stdout:?}" + ); + assert_eq!( + stdout.trim_end(), + format!("RESULT:{TOTAL_BYTES}"), + "round {round}: not every piped byte reached the iterator" + ); +} + +#[test] +fn piped_stdin_async_iteration_reads_to_eof_every_time() { + let dir = tempfile::tempdir().expect("create fixture directory"); + let binary = compile(dir.path()); + for round in 0..ROUNDS { + run_round(&binary, round); + } +} + +/// Undriven, the fixture must leave stdin alone (the parity sweep runs it +/// with whatever stdin the caller has). +#[test] +fn undriven_fixture_does_not_wait_on_stdin() { + let dir = tempfile::tempdir().expect("create fixture directory"); + let binary = compile(dir.path()); + let output = Command::new(&binary) + .stdin(Stdio::piped()) + .output() + .expect("run undriven fixture"); + assert!(output.status.success()); + assert_eq!( + String::from_utf8_lossy(&output.stdout).trim_end(), + "RESULT:idle" + ); +} diff --git a/test-files/test_issue_10895_stdin_pipe_stall.ts b/test-files/test_issue_10895_stdin_pipe_stall.ts new file mode 100644 index 0000000000..5e1196ef91 --- /dev/null +++ b/test-files/test_issue_10895_stdin_pipe_stall.ts @@ -0,0 +1,32 @@ +// #10895: `for await (const chunk of process.stdin)` on a pipe stalled forever +// part-way through the input. The async iterator pauses the source after every +// delivered chunk and resumes it on the next pull; `pause()` makes the fd-0 +// reader thread exit and `resume()` respawns it, and a resume that landed while +// the old reader was still on its way out lost the respawn for good. +// +// Driven by `crates/perry/tests/issue_10895_stdin_pipe_stall.rs`, which pipes +// several MiB in small writes. Undriven (the parity sweep) it must not touch +// stdin at all: the sweep inherits whatever stdin the caller has, and a fixture +// that waits for EOF on a terminal never finishes. +const driven = process.env.PERRY_10895_DRIVE === "1"; + +async function main(): Promise { + if (!driven) { + console.log("RESULT:idle"); + return; + } + let total = 0; + for await (const data of process.stdin) { + const chunk: Uint8Array = data; + total += chunk.length; + // Touch the payload so a chunk that arrives with the right length but the + // wrong bytes is caught too (every byte the driver writes is 1). + if (chunk[0] !== 1 || chunk[chunk.length - 1] !== 1) { + console.log("RESULT:corrupt@" + total); + return; + } + } + console.log("RESULT:" + total); +} + +main(); From 735f9b1e62e3546b0c63f6280e1f937de7aba5ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Mon, 21 Sep 2026 19:03:32 +0200 Subject: [PATCH 2/2] changelog: #10913 fragment --- .../10913-stdin-reader-restart-race.md | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) create mode 100644 changelog.d/10913-stdin-reader-restart-race.md diff --git a/changelog.d/10913-stdin-reader-restart-race.md b/changelog.d/10913-stdin-reader-restart-race.md new file mode 100644 index 0000000000..78e777013b --- /dev/null +++ b/changelog.d/10913-stdin-reader-restart-race.md @@ -0,0 +1,23 @@ +Fixed `for await (const chunk of process.stdin)` on a pipe stalling forever +part-way through the input (#10895). + +The async iterator pauses its source after every delivered chunk and resumes +it on the next pull. On `process.stdin`, `pause()` latches `STDIN_DETACHED` — +the fd-0 reader thread exits when it sees the latch at the top of its loop — +and `resume()` clears the latch and respawns the reader unless +`STDIN_READER_STARTED` says one is still running. The reader's stop decision +and its `STARTED` reset were two separate steps, so a `resume()` that landed +between them found `STARTED` still true, spawned nothing, and the old reader +then left: fd 0 had no reader while every liveness view still reported an +open, flowing stdin, and the process idled forever with input unread. One roll +per delivered chunk: near-certain at 4 MiB through a 16 KiB macOS pipe, 1 in +350 at 16 MiB on Linux. Introduced by the unified fd-0 reader (2026-09-04); +the published 0.5.1520 predates it. + +The reader's check-and-clear and the restart CAS are now atomic with respect +to each other under one lifecycle lock (never held across `read()`); the +detach exit releases the reader slot itself and disarms the drop guard. + +Tests: `crates/perry/tests/issue_10895_stdin_pipe_stall.rs` (8 MiB in 256-byte +writes, 12 rounds, red on unpatched main) and `reader_lifecycle_tests` in +`os_process_streams.rs`, which replays the interleaving deterministically.