fix(child_process): unblock OpenCode writable pipelines

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.

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)),

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<f64> {
+    [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
 }
 

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.

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<boolean> {
+  let synchronous = true;
+  const asynchronous = await new Promise<boolean>((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:?}"
+        );
+    }
+}