Run Process.run through one worker on Bun and Node

bend2/base.bend

@@ -201,6 +201,8 @@ def IO.args_bytes() -> IO(List<List<&2, U32>>):
   import "./effs/args.c"
   import "./effs/args.js"
 
+# Runs an argv subprocess on C, Bun and Node, without a shell.
+# max_output bounds combined stdout/stderr bytes; timeout_ms includes pipe draining.
 def Process.run(program: String, args: List<String>, input: String,
   max_output: U32, timeout_ms: U32) ->
   IO(Result<&1, &1, U32 & String, U32 & (String & String)>):

bend2/effs/process_run.js

@@ -3,10 +3,12 @@
 
 const PROCESS_WORKER = `
 const { errno, signals } = require("node:os").constants;
+const { parentPort } = require("node:worker_threads");
+const { spawn } = require("node:child_process");
 let port;
 let flag;
 
-self.onmessage = async ({ data }) => {
+parentPort.on("message", async (data) => {
   if (data.port !== undefined) {
     ({ port, flag } = data);
     return;
@@ -21,40 +23,56 @@ self.onmessage = async ({ data }) => {
   port.postMessage(got);
   Atomics.store(flag, 0, 1);
   Atomics.notify(flag, 0);
-};
+});
 
-async function run({ argv, input, env, max, ms }) {
-  const proc = Bun.spawn({ cmd: argv, stdin: input, env, stdout: "pipe",
-    stderr: "pipe" });
-  const readers = [proc.stdout.getReader(), proc.stderr.getReader()];
-  let size = 0;
-  const read = async (reader) => {
-    const chunks = [];
-    for (let r; !(r = await reader.read()).done;) {
-      chunks.push(r.value);
-      size += r.value.length;
-      if (size > max) {
-        throw "EFBIG";
-      }
-    }
-    return Buffer.concat(chunks).toString("utf8");
-  };
-  let timer;
-  try {
-    const [out, err] = await Promise.race([
-      Promise.all([...readers.map(read), proc.exited]),
-      new Promise((_, no) => timer = setTimeout(no, Math.min(ms, 2 ** 31 - 1),
-        "ETIMEDOUT"))]);
-    const sig = proc.signalCode === null ? 0 : signals[proc.signalCode] ?? 0;
-    return { code: proc.exitCode ?? 128 + sig, out, err };
-  } catch (e) {
-    proc.kill("SIGKILL");
-    readers.forEach((r) => r.cancel().catch(() => {}));
-    await proc.exited;
-    throw e;
-  } finally {
-    clearTimeout(timer);
-  }
+function run({ argv, input, env, max, ms }) {
+  return new Promise((yes, no) => {
+    const proc = spawn(argv[0], argv.slice(1), { env,
+      stdio: ["pipe", "pipe", "pipe"] });
+    // Node destroys proc.stdin at exit; a descendant may still read it.
+    const stdin = proc.stdin;
+    proc.stdin = null;
+    const chunks = [[], []];
+    let size = 0;
+    let error;
+    const fail = (e) => {
+      if (error !== undefined) return;
+      error = e;
+      proc.kill("SIGKILL");
+      stdin.destroy();
+      proc.stdout.destroy();
+      proc.stderr.destroy();
+    };
+    proc.once("error", fail);
+    stdin.on("error", (e) => {
+      if (e.code !== "EPIPE") fail(e);
+    });
+    [proc.stdout, proc.stderr].forEach((stream, i) => {
+      stream.on("error", fail);
+      stream.on("data", (chunk) => {
+        if (error !== undefined) return;
+        size += chunk.length;
+        if (size > max) fail("EFBIG");
+        else chunks[i].push(chunk);
+      });
+    });
+    // Feed stdin until the child exits and its stdout and stderr close.
+    let live = 3;
+    const done = () => --live || stdin.destroy();
+    proc.once("exit", done);
+    proc.stdout.once("close", done);
+    proc.stderr.once("close", done);
+    const timer = setTimeout(() => fail("ETIMEDOUT"),
+      Math.min(ms, 2 ** 31 - 1));
+    proc.once("close", (code, signal) => {
+      clearTimeout(timer);
+      if (error !== undefined) no(error);
+      else yes({ code: code ?? 128 + (signals[signal] ?? 0),
+        out: Buffer.concat(chunks[0]).toString("utf8"),
+        err: Buffer.concat(chunks[1]).toString("utf8") });
+    });
+    stdin.end(input);
+  });
 }
 `;
 
@@ -70,10 +88,10 @@ function process_run(program, args, input, maxOutput, timeoutMs) {
     return io_fail(22);
   }
   if (process_worker === null) {
-    const { MessageChannel, receiveMessageOnPort } = require("node:worker_threads");
+    const { Worker, MessageChannel, receiveMessageOnPort } = require("node:worker_threads");
     const { port1, port2 } = new MessageChannel();
     const flag = new Int32Array(new SharedArrayBuffer(4));
-    const worker = new Worker(URL.createObjectURL(new Blob([PROCESS_WORKER])));
+    const worker = new Worker(PROCESS_WORKER, { eval: true });
     worker.unref();
     worker.postMessage({ port: port2, flag }, [port2]);
     process_worker = { worker, port: port1, flag, receive: receiveMessageOnPort };

tests/io/process_run.bend

@@ -91,6 +91,23 @@ def main() -> IO(Unit):
     r1 : Result<&1, &1, U32 & String, U32 & (String & String)> <-
       Process.run("/bin/echo", ["hi"], "", 128, 4294967295)
     IO.print("largest timeout: " ++ answer(r1, 0, "hi\n", ""))
+    partial : Result<&1, &1, U32 & String, U32 & (String & String)> <-
+      Process.run("/usr/bin/head", ["-c", "3"],
+        String.repeat(String.repeat("z", 128n), 1000n), 128, 3000)
+    IO.print("partial stdin: " ++ answer(partial, 0, "zzz", ""))
+    unread : Result<&1, &1, U32 & String, U32 & (String & String)> <-
+      Process.run("/usr/bin/true", [],
+        String.repeat(String.repeat("z", 128n), 1000n), 128, 3000)
+    IO.print("unread stdin: " ++ answer(unread, 0, "", ""))
+    held : Result<&1, &1, U32 & String, U32 & (String & String)> <-
+      Process.run("/bin/sh", ["-c", "sleep 2 <&0 >/dev/null 2>&1 & exit 4"],
+        String.repeat(String.repeat("z", 256n), 1200n), 128, 500)
+    IO.print("descendant stdin: " ++ answer(held, 4, "", ""))
+    late : Result<&1, &1, U32 & String, U32 & (String & String)> <-
+      Process.run("/bin/sh", ["-c", "(sleep 0.2; cat) <&0 & exit 0"],
+        String.repeat(String.repeat("z", 128n), 1000n), 200000, 3000)
+    IO.print("late stdin reader: " ++ answer(late, 0,
+      String.repeat(String.repeat("z", 128n), 1000n), ""))
 
 #|stdin: True
 #|literal argv: True
@@ -111,3 +128,7 @@ def main() -> IO(Unit):
 #|runaway descendant: True
 #|closed stdio: True
 #|largest timeout: True
+#|partial stdin: True
+#|unread stdin: True
+#|descendant stdin: True
+#|late stdin reader: True