diff --git a/changelog.d/8531-child-process-writable-callbacks.md b/changelog.d/8531-child-process-writable-callbacks.md new file mode 100644 index 0000000000..f8b6ad3d18 --- /dev/null +++ b/changelog.d/8531-child-process-writable-callbacks.md @@ -0,0 +1,5 @@ +### Fixed + +- Child-process stdin writables now invoke completion callbacks for Node's + `write` and `end` overloads on a later event-loop turn, preventing OpenCode + LSP and formatter pipelines from stalling while they await a completed write. diff --git a/crates/perry-runtime/src/child_process/builder.rs b/crates/perry-runtime/src/child_process/builder.rs index 53dfc4d63f..ece4e23d4d 100644 --- a/crates/perry-runtime/src/child_process/builder.rs +++ b/crates/perry-runtime/src/child_process/builder.rs @@ -28,6 +28,11 @@ pub(crate) fn cp_cast2(f: extern "C" fn(*const ClosureHeader, f64, f64) -> f64) unsafe { std::mem::transmute(f) } } #[allow(clippy::missing_transmute_annotations)] +/// Erase a three-argument native method to the common child-process method ABI. +pub(crate) fn cp_cast3(f: extern "C" fn(*const ClosureHeader, f64, f64, f64) -> f64) -> CpFn { + unsafe { std::mem::transmute(f) } +} +#[allow(clippy::missing_transmute_annotations)] pub(crate) fn cp_cast4(f: extern "C" fn(*const ClosureHeader, f64, f64, f64, f64) -> f64) -> CpFn { unsafe { std::mem::transmute(f) } } @@ -46,8 +51,9 @@ pub(crate) fn cp_register_arities() { crate::closure::js_register_closure_length(cp_method_dispose as *const u8, 0); js_register_closure_arity(cp_method_read as *const u8, 1); js_register_closure_arity(cp_method_pipe as *const u8, 1); - js_register_closure_arity(cp_method_write2 as *const u8, 2); - js_register_closure_arity(cp_method_stdin_end as *const u8, 1); + js_register_closure_arity(cp_method_stdin_write as *const u8, 3); + js_register_closure_arity(cp_method_stdin_end as *const u8, 3); + js_register_closure_arity(cp_stream_callback_thunk as *const u8, 0); // #3316: `send(message, sendHandle, options, callback)` — dispatch with 4 // padded slots so the trailing callback is visible regardless of call-site // arity, and report `child.send.length === 4` like Node. @@ -157,8 +163,8 @@ pub(crate) fn cp_build_writable() -> f64 { ("removeListener", cp_cast2(cp_method_remove_listener)), ("off", cp_cast2(cp_method_remove_listener)), ("emit", cp_cast2(cp_method_emit)), - ("write", cp_cast2(cp_method_write2)), - ("end", cp_cast1(cp_method_stdin_end)), + ("write", cp_cast3(cp_method_stdin_write)), + ("end", cp_cast3(cp_method_stdin_end)), ("destroy", cp_cast0(cp_method_this0)), ("cork", cp_cast0(cp_method_this0)), ("uncork", cp_cast0(cp_method_this0)), diff --git a/crates/perry-runtime/src/child_process/emitter.rs b/crates/perry-runtime/src/child_process/emitter.rs index 37ee89e0fe..312ca96a42 100644 --- a/crates/perry-runtime/src/child_process/emitter.rs +++ b/crates/perry-runtime/src/child_process/emitter.rs @@ -272,20 +272,63 @@ pub(crate) extern "C" fn cp_pipe_end_thunk(closure: *const ClosureHeader) -> f64 } cp_undefined() } -/// `child.stdin.write(chunk[, encoding][, callback])` — #1934. The `this` is -/// the stdin Writable; route the bytes to the live child's stdin via the -/// reactor. Returns `true` (Node's `write` returns whether the buffer can take -/// more — `true` for our synchronous pipe write). -pub(crate) extern "C" fn cp_method_write2( +/// Return the last callable argument from the two optional trailing stream +/// slots (`write(chunk, cb)` / `write(chunk, encoding, cb)`, and the matching +/// `end` overloads). +fn cp_stream_callback(arg2: f64, arg3: f64) -> Option { + [arg3, arg2] + .into_iter() + .find(|value| !crate::fs::extract_closure_ptr(*value).is_null()) +} + +/// Defer a successful writable completion callback to the next event-loop +/// turn. The wrapper closure roots the callback until delivery and invokes it +/// with no arguments, so `callback(error)` observes `undefined` on success. +fn cp_defer_stream_callback(callback: f64) { + let scope = crate::gc::RuntimeHandleScope::new(); + let callback = scope.root_nanbox_f64(callback); + let deferred = + scope.root_raw_mut_ptr(js_closure_alloc(cp_stream_callback_thunk as *const u8, 1)); + deferred.with_mut_ptr(|deferred: *mut ClosureHeader| { + js_closure_set_capture_ptr(deferred, 0, callback.get_nanbox_f64().to_bits() as i64); + }); + deferred.with_mut_ptr(|deferred: *mut ClosureHeader| { + crate::timer::js_set_immediate_callback(deferred as i64); + }); +} + +/// Deliver a deferred child-stdin completion callback captured in slot zero. +pub(crate) extern "C" fn cp_stream_callback_thunk(closure: *const ClosureHeader) -> f64 { + let callback = f64::from_bits(js_closure_get_capture_ptr(closure, 0) as u64); + let args: [f64; 0] = []; + unsafe { + let _ = js_native_call_value(callback, args.as_ptr(), 0); + } + cp_undefined() +} + +/// `child.stdin.write(chunk[, encoding][, callback])` — #1934 / #8512. The +/// `this` is the stdin Writable; route the bytes to the live child's stdin and +/// complete the optional callback asynchronously. OpenCode's Effect +/// `NodeSink.fromWritable` waits for this callback before sending the next LSP +/// frame, so dropping it stalls long-running language servers. +/// +/// Returns `true`: writes are synchronously drained into the OS pipe, so there +/// is no buffered high-water mark that could require a later `drain` event. +pub(crate) extern "C" fn cp_method_stdin_write( closure: *const ClosureHeader, chunk: f64, - _enc: f64, + arg2: f64, + arg3: f64, ) -> f64 { let this = cp_this(closure); if let Some(handle) = cp_handle_of(this) { let bytes = cp_value_to_bytes(chunk); reactor::cp_live_stdin_write(handle, &bytes); } + if let Some(callback) = cp_stream_callback(arg2, arg3) { + cp_defer_stream_callback(callback); + } TAG_TRUE_F64 } @@ -423,9 +466,15 @@ extern "C" fn cp_disconnect_emit_thunk(closure: *const ClosureHeader) -> f64 { cp_undefined() } -/// `child.stdin.end([chunk])` — write the optional final chunk, then close the -/// pipe so the child sees EOF (#1934). The `this` is the stdin Writable. -pub(crate) extern "C" fn cp_method_stdin_end(closure: *const ClosureHeader, chunk: f64) -> f64 { +/// `child.stdin.end([chunk][, encoding][, callback])` — write the optional +/// final chunk, close the pipe so the child sees EOF, then complete the +/// optional callback asynchronously (#1934 / #8512). +pub(crate) extern "C" fn cp_method_stdin_end( + closure: *const ClosureHeader, + chunk: f64, + arg2: f64, + arg3: f64, +) -> f64 { let this = cp_this(closure); if let Some(handle) = cp_handle_of(this) { // Optional final data chunk. Skip `undefined`, the `0.0` arg-padding @@ -443,6 +492,11 @@ pub(crate) extern "C" fn cp_method_stdin_end(closure: *const ClosureHeader, chun reactor::cp_live_stdin_close(handle); } cp_set_field(this, b"writable", TAG_FALSE_F64); + if let Some(callback) = cp_stream_callback(arg2, arg3) + .or_else(|| (!crate::fs::extract_closure_ptr(chunk).is_null()).then_some(chunk)) + { + cp_defer_stream_callback(callback); + } this } diff --git a/crates/perry-runtime/src/child_process/mod.rs b/crates/perry-runtime/src/child_process/mod.rs index b21281f585..c66f5d7214 100644 --- a/crates/perry-runtime/src/child_process/mod.rs +++ b/crates/perry-runtime/src/child_process/mod.rs @@ -93,7 +93,8 @@ pub(crate) use emitter::{ cp_emit, cp_method_child_spawn, cp_method_disconnect, cp_method_dispose, cp_method_emit, cp_method_kill, cp_method_on, cp_method_pipe, cp_method_read, cp_method_remove_all_listeners, cp_method_remove_listener, cp_method_send, cp_method_set_encoding, cp_method_stdin_end, - cp_method_this0, cp_method_this1, cp_method_write2, cp_send_callback_thunk, js_fork_child, + cp_method_stdin_write, cp_method_this0, cp_method_this1, cp_send_callback_thunk, + cp_stream_callback_thunk, js_fork_child, }; // builder.rs — heap object construction + shape ids. diff --git a/crates/perry/tests/issue_8512_opencode_process_lifecycle.rs b/crates/perry/tests/issue_8512_opencode_process_lifecycle.rs new file mode 100644 index 0000000000..34025efc7c --- /dev/null +++ b/crates/perry/tests/issue_8512_opencode_process_lifecycle.rs @@ -0,0 +1,150 @@ +//! Regression coverage for OpenCode's child-process writable path (#8512). +//! +//! OpenCode adapts `child.stdin` through Effect's `NodeSink.fromWritable` for +//! LSP framing and process pipelines. That adapter waits for Node's optional +//! `write` / `end` completion callbacks. Perry used to perform the pipe write +//! but drop those callbacks, leaving the sink (and therefore the LSP or +//! formatter lifecycle) pending forever. + +#![cfg(unix)] + +use std::io::Read; +use std::path::PathBuf; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +const FIXTURE: &str = r#" +import { spawn } from "node:child_process"; + +async function completion(register: (done: () => void) => void): Promise { + let synchronous = true; + const asynchronous = await new Promise((resolve) => { + register(() => resolve(!synchronous)); + synchronous = false; + }); + return asynchronous; +} + +async function main() { + const childCwd = process.env.OPENCODE_CHILD_CWD as string; + const child = spawn("sh", ["-c", 'printf "%s\\n%s\\n" "$PWD" "$OPENCODE_LSP"; cat'], { + cwd: childCwd, + env: { PATH: process.env.PATH, OPENCODE_LSP: "ready" }, + stdio: ["pipe", "pipe", "pipe"], + }); + + let stdout = ""; + child.stdout!.on("data", (chunk: Buffer) => { + stdout += chunk.toString("utf8"); + }); + const closed = new Promise<{ code: number | null; signal: string | null }>((resolve) => { + child.once("close", (code: number | null, signal: string | null) => resolve({ code, signal })); + }); + + const shortWriteAsync = await completion((done) => child.stdin!.write("frame-one\n", done)); + const encodedWriteAsync = await completion((done) => child.stdin!.write("frame-two\n", "utf8", done)); + const endAsync = await completion((done) => child.stdin!.end(done)); + const result = await closed; + + console.log("WRITE_CB_ASYNC:" + shortWriteAsync); + console.log("ENCODED_WRITE_CB_ASYNC:" + encodedWriteAsync); + console.log("END_CB_ASYNC:" + endAsync); + console.log("EXIT:" + result.code + ":" + result.signal); + console.log("CWD:" + stdout.startsWith(childCwd + "\n")); + console.log("ENV:" + stdout.includes("\nready\n")); + console.log("FRAMES:" + stdout.includes("frame-one\nframe-two\n")); +} + +main(); +"#; + +#[test] +fn opencode_lsp_writable_callbacks_complete() { + let dir = tempfile::tempdir().expect("tempdir"); + let entry = dir.path().join("main.ts"); + let output = dir.path().join("main_bin"); + let child_cwd = dir.path().join("child-cwd"); + std::fs::create_dir(&child_cwd).expect("create distinct child cwd"); + let child_cwd = std::fs::canonicalize(child_cwd).expect("canonicalize child cwd"); + std::fs::write(&entry, FIXTURE).expect("write fixture"); + + let compile = Command::new(perry_bin()) + .current_dir(dir.path()) + .arg("compile") + .arg(&entry) + .arg("-o") + .arg(&output) + .output() + .expect("run perry compile"); + assert!( + compile.status.success(), + "perry compile failed:\nstdout: {}\nstderr: {}", + String::from_utf8_lossy(&compile.stdout), + String::from_utf8_lossy(&compile.stderr) + ); + + // A dropped completion callback leaves `cat` alive with stdin open, so use + // a hard timeout to turn that historical hang into a deterministic test + // failure. + let mut child = Command::new(&output) + .current_dir(dir.path()) + .env("OPENCODE_CHILD_CWD", &child_cwd) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("run compiled fixture"); + let mut stdout_pipe = child.stdout.take().expect("stdout pipe"); + let mut stderr_pipe = child.stderr.take().expect("stderr pipe"); + let stdout_reader = std::thread::spawn(move || { + let mut buf = String::new(); + let _ = stdout_pipe.read_to_string(&mut buf); + buf + }); + let stderr_reader = std::thread::spawn(move || { + let mut buf = String::new(); + let _ = stderr_pipe.read_to_string(&mut buf); + buf + }); + + let start = Instant::now(); + let status = loop { + if let Some(status) = child.try_wait().expect("poll compiled fixture") { + break status; + } + if start.elapsed() > Duration::from_secs(15) { + let _ = child.kill(); + let _ = child.wait(); + let stdout = stdout_reader.join().expect("stdout reader"); + let stderr = stderr_reader.join().expect("stderr reader"); + panic!( + "compiled fixture hung waiting for a child stdin callback:\nstdout: {stdout}\nstderr: {stderr}" + ); + } + std::thread::sleep(Duration::from_millis(25)); + }; + + let stdout = stdout_reader.join().expect("stdout reader"); + let stderr = stderr_reader.join().expect("stderr reader"); + assert!( + status.success(), + "compiled fixture failed ({status:?}):\nstdout: {stdout}\nstderr: {stderr}" + ); + for expected in [ + "WRITE_CB_ASYNC:true", + "ENCODED_WRITE_CB_ASYNC:true", + "END_CB_ASYNC:true", + "EXIT:0:null", + "CWD:true", + "ENV:true", + "FRAMES:true", + ] { + assert!( + stdout.contains(expected), + "missing {expected:?} in {stdout:?}" + ); + } +}