authorclover caruso <git@paperclover.net> 2026-10-12 15:33:50-07:00
committerclover caruso <git@paperclover.net> 2026-10-12 16:58:29-07:00
logd3185018db6284979ba12696d8b8e34be6985f39
treedc8f2a653e9d794487482d01b2e7a4347bb2d2e9
parentfe94d78d812c70545e45828e38f6a49282e61ba0
signature Signed by SSH key SHA256:cOKiuRFOeSRxne6EWgHtdQQSlBxjOXm2hOCFnCdLQbQ (~clover)

feat(lib/subprocess): rewrite deprecated apis

see #20 Assisted-by: claude-opus-5.5

13 files changed, 707 insertions(+), 203 deletions(-)

framework/watch.ts+13-19
......@@ -2,10 +2,7 @@
22
33const debounceMilliseconds = 25;
44
5let server: child_process.ChildProcess | null = null;
6process.on("beforeExit", () => {
7 server?.removeListener("close", onSubprocessClose);
8});
5let server: subprocess.Subprocess | null = null;
96
107let watch: Watch;
118
......@@ -18,22 +15,18 @@ export async function main() {
1815}
1916
2017function serve() {
21 if (server) {
22 server.removeListener("close", onSubprocessClose);
23 server.kill();
24 }
25 server = subprocess.fork({
26 cmd: [".clover/o/backend.js", "--development"],
27 stdio: "inherit",
28 execArgv: ["--no-warnings"],
29 });
30 server.on("close", onSubprocessClose);
31}
18 server?.kill();
19 const backend = server = subprocess.spawn(
20 [process.execPath, "--no-warnings", ".clover/o/backend.js", "--development"],
21 { stdin: "inherit", ipc: true },
22 );
3223
33function onSubprocessClose(code: number | null, signal: string | null) {
34 server = null;
35 const status = code != null ? `code ${code}` : `signal ${signal}`;
36 console.error(`Backend process exited with ${status}`);
24 backend.wait().then(() => "", (err) => `: ${exception.message(err)}`).then((reason) => {
25 // a backend that was replaced is not the one to report
26 if (server !== backend) return;
27 server = null;
28 console.error(`Backend process exited${reason}`);
29 });
3730}
3831
3932let isRunning = false;
......@@ -182,6 +175,7 @@ class Watch {
182175 }
183176}
184177
178import * as exception from "@clo/lib/exception";
185179import * as subprocess from "@clo/lib/subprocess";
186180import * as child_process from "node:child_process";
187181import * as fs from "node:fs";
lib/examples/subprocess.ts created+30
......@@ -0,0 +1,30 @@
1// a build that fans out to worker processes. each worker is this same file,
2// started with a node of the build as its global progress root: what it
3// starts and logs is drawn by the one process that has the terminal.
4const [worker] = process.argv.slice(2);
5const steps = ["parse", "transform", "minify"];
6
7if (worker) {
8 log.info(`${worker}: found ${steps.length} steps`);
9 for (const step of steps) {
10 using task = progress.start(step, { total: 30 });
11 for (let i = 0; i < 30; i += 1) {
12 await async.delay(20 + Math.random() * 60);
13 task.value += 1;
14 if (i === 15) task.log.warn(`${step} is taking a while`);
15 }
16 }
17 log.info(`${worker}: done`);
18} else {
19 using build = progress.start("build");
20 await Promise.all(["pages", "styles", "scripts"].map(async (name) => {
21 using node = build.start(name);
22 await subprocess.spawn([process.execPath, import.meta.filename, name], { progress: node }).wait();
23 }));
24 log.info("built");
25}
26
27import * as async from "../async.ts";
28import * as log from "../log.ts";
29import * as progress from "../progress.ts";
30import * as subprocess from "../subprocess.ts";
lib/node.ts+3
......@@ -27,7 +27,9 @@ interface Process {
2727 removeListener(event: string, callback: () => void): this;
2828 listenerCount(event: string): number;
2929 kill(pid: number, signal: string): boolean;
30 env: Partial<Record<string, string>>;
3031 pid: number;
32 ppid: number;
3133 platform: string;
3234 versions: { node?: string; bun?: string; deno?: string };
3335 stdin: TtyInput;
......@@ -72,6 +74,7 @@ interface Builtins {
7274 readFileSync(file: string): Uint8Array;
7375 writeFileSync(file: string, data: string, options?: { mode?: number }): void;
7476 appendFileSync(file: string, data: string, options?: { mode?: number }): void;
77 writeSync(fd: number, data: string): number;
7578 mkdirSync(dir: string, options: { recursive: true }): void;
7679 };
7780 "inspector": {
lib/progress.ts+121-36
......@@ -441,7 +441,7 @@ function newNode<Result, EventMap extends Events.Map>(
441441 showTotal: opts.showTotal ?? true,
442442 passive: opts.passive ?? false,
443443 hidden: opts.hidden ?? false,
444 logs: [],
444 logs: [...opts.messages ?? []],
445445 children: [],
446446 parent: null,
447447 sortChildren: opts.sortChildren ?? null,
......@@ -957,40 +957,7 @@ export function encodeEventStream<
957957 isFirst = false;
958958 }
959959
960 // TODO: why does ts require two args
961 stack.use(root.on("node-start", (newNode, _) => {
962 if (!newNode.parent) s.rootChildrenUpdated = true;
963 s.changed.set(newNode, new Set());
964 emitSoon();
965 }));
966 stack.use(root.on("node-change", (node, key) => {
967 let set = s.changed.get(node);
968 if (!set) s.changed.set(node, set = new Set());
969 set.add(key);
970 emitSoon();
971 }));
972 stack.use(root.on("node-end", (node) => {
973 if (!node.parent) s.rootChildrenUpdated = true;
974 s.changed.delete(node);
975 s.deleted.add(node); // handle add and remove in same frame
976 emitSoon();
977 }));
978 // custom events pass through the stream verbatim
979 stack.use(root.onAny((channel, args) => {
980 if (rootEvents.has(channel)) return;
981 s.pendingEvents.push([channel as string, ...args as ts.Json[]]);
982 emitSoon();
983 }));
984 stack.use(root.on("end", (result) => {
985 s.pendingEvents.push(["end", result]);
986 s.ended = true;
987 emitNow();
988 }));
989 stack.use(root.on("error", (error) => {
990 s.pendingEvents.push(["error", exceptions.message(error)]);
991 s.ended = true;
992 emitNow();
993 }));
960 stack.use(s.observe(emitSoon, emitNow));
994961 },
995962 async pull(controller) {
996963 await ready.until((x) => x === true);
......@@ -1087,6 +1054,51 @@ class Encoder<
10871054 }
10881055 }
10891056
1057 /**
1058 * fill `changed`, `deleted`, and `pendingEvents` from the events of the
1059 * root. `soon` is called when there is a delta worth sending, and `now`
1060 * when it is the last one.
1061 */
1062 observe(soon: () => void, now: () => void): DisposableStack {
1063 const { root } = this;
1064 const stack = new DisposableStack();
1065 // TODO: why does ts require two args
1066 stack.use(root.on("node-start", (newNode, _) => {
1067 if (!newNode.parent) this.rootChildrenUpdated = true;
1068 this.changed.set(newNode, new Set());
1069 soon();
1070 }));
1071 stack.use(root.on("node-change", (node, key) => {
1072 let set = this.changed.get(node);
1073 if (!set) this.changed.set(node, set = new Set());
1074 set.add(key);
1075 soon();
1076 }));
1077 stack.use(root.on("node-end", (node) => {
1078 if (!node.parent) this.rootChildrenUpdated = true;
1079 this.changed.delete(node);
1080 this.deleted.add(node); // handle add and remove in same frame
1081 soon();
1082 }));
1083 // custom events pass through the stream verbatim
1084 stack.use(root.onAny((channel, args) => {
1085 if (rootEvents.has(channel)) return;
1086 this.pendingEvents.push([channel as string, ...args as ts.Json[]]);
1087 soon();
1088 }));
1089 stack.use(root.on("end", (result) => {
1090 this.pendingEvents.push(["end", result]);
1091 this.ended = true;
1092 now();
1093 }));
1094 stack.use(root.on("error", (error) => {
1095 this.pendingEvents.push(["error", exceptions.message(error)]);
1096 this.ended = true;
1097 now();
1098 }));
1099 return stack;
1100 }
1101
10901102 getDelta(isFirst = false) {
10911103 const deleted = Array.from(
10921104 this.deleted,
......@@ -1830,7 +1842,79 @@ type EncodedKey = number & { brand: typeof kNode };
18301842
18311843const globalKeyPool = new KeyPool<number>();
18321844
1833const global: Ref = /** @__PURE__ */ ((root = new Root()) => (attachToScreen(root, log), root))();
1845/** names the pipe to a parent process that handed this one a place in its progress */
1846const parentEnv = "CLOVER_PROGRESS";
1847
1848/**
1849 * the environment for a child process whose global root and log are to be
1850 * read by {@linkcode attachChild}. `fd` is where the child has the pipe.
1851 */
1852export function childEnv(fd: number): Record<string, string> {
1853 return { [parentEnv]: `${fd}:${UNWRAP(node.process).pid}` };
1854}
1855
1856/**
1857 * read what a child process reports through the pipe that {@linkcode childEnv}
1858 * named to it. everything on its global root is started on `target`, and what
1859 * it logs globally is logged globally here. resolves once the child closes
1860 * the pipe, which ends all it had started.
1861 *
1862 * `subprocess.spawn` does this with its `progress` option.
1863 */
1864export async function attachChild(readable: ReadableStream<Uint8Array>, target: Ref): Promise<void> {
1865 const decoder = new Decoder(target, new Events(), noop);
1866 try {
1867 for await (const line of stream.lines(readable)) {
1868 const record: StreamEvent | log.Message = JSON.parse(line);
1869 if (Array.isArray(record)) decoder.processEvent(record);
1870 else log.writeMessage(record);
1871 }
1872 } finally {
1873 decoder.reset();
1874 }
1875}
1876
1877/**
1878 * the other end of {@linkcode attachChild}. writes are synchronous, so that
1879 * the last words of a process are not lost when it exits.
1880 */
1881function tryAttachToParent(root: Root): boolean {
1882 const { process } = node;
1883 const fs = node.builtin("fs");
1884 const [fd, parent] = (process?.env[parentEnv] ?? "").split(":").map(Number);
1885 // every descendant inherits the variable, but only the child has the pipe
1886 if (!process || !fs || !fd || process.ppid !== parent) return false;
1887
1888 const write = (record: StreamEvent | log.Message) => {
1889 try {
1890 fs.writeSync(fd, JSON.stringify(record) + "\n");
1891 } catch {
1892 // the parent is gone, and so is anyone to tell
1893 }
1894 };
1895
1896 const encoder = new Encoder(root);
1897 let timer: async.Cancelable<void> | null = null;
1898 const flush = () => {
1899 timer?.cancel();
1900 timer = null;
1901 write(encoder.getDelta());
1902 };
1903 const flushSoon = () => {
1904 if (timer) return;
1905 (timer = root.delay(1000 / 30)).then(flush);
1906 };
1907
1908 encoder.observe(flushSoon, flush);
1909 process.addListener("exit", () => timer && flush());
1910 // what changed on a node before a message is said stays before it
1911 log.replaceGlobalMessageDestination((m) => (timer && flush(), write({ ...m, stack: undefined })));
1912 return true;
1913}
1914
1915const global: Ref = /** @__PURE__ */ ((root = new Root()) => (
1916 tryAttachToParent(root) || attachToScreen(root, log), root
1917))();
18341918
18351919/**
18361920 * a {@linkcode Ref} to the global progress root. unlike referencing the
......@@ -1861,6 +1945,7 @@ import { Events } from "./Events.ts";
18611945import * as exceptions from "./exception.ts";
18621946import * as log from "./log.ts";
18631947import * as stack from "./log/stack.ts";
1948import * as node from "./node.ts";
18641949import * as stream from "./stream.ts";
18651950import * as string from "./string.ts";
18661951import * as ansi from "./string/ansi.ts";
lib/readme.changes.md+11
......@@ -2,9 +2,18 @@
22
33## v8
44
5### breaking
6
7- `subprocess` is rewritten around web streams.
8
59### features
610
711- new library `repl`: an interactive prompt to add to long running programs.
12- `stream.lines` reads a byte stream as lines of text.
13- `subprocess`'s `progress` option hands a `progress.Ref` to a child process,
14 which becomes the global root of `progress` there. the child's global log is
15 written by the parent as well. `progress.childEnv` and `progress.attachChild`
16 are the two halves, for spawning by other means.
817- `log`
918 - widgets can now read keyboard input.
1019 - finally fixes resize corruptions for long running widgets
......@@ -15,6 +24,8 @@
1524
1625- `async.DedupeConcurrent` no longer causes unhandled rejections with correct
1726 caller error handling.
27- `progress`: `StartOptions.messages` is no longer ignored, so a decoded stream
28 keeps the logs that a node had when it started.
1829
1930## v7
2031
lib/stream.test.ts+50
......@@ -225,6 +225,56 @@ describe("BufferedReader", () => {
225225 });
226226});
227227
228describe("lines", () => {
229 async function collect(chunks: Array<string | Uint8Array>) {
230 const readable = new ReadableStream<Uint8Array>({
231 start(controller) {
232 for (const chunk of chunks) {
233 controller.enqueue(typeof chunk === "string" ? string.encodeUtf8(chunk) : chunk);
234 }
235 controller.close();
236 },
237 });
238 return await Array.fromAsync(stream.lines(readable));
239 }
240
241 test("every kind of line ending", async () => {
242 assert.deepEqual(await collect(["a\nb\r\nc\rd\n"]), ["a", "b", "c", "d"]);
243 assert.deepEqual(await collect(["a\n\nb\r\n\r\nc"]), ["a", "", "b", "", "c"]);
244 assert.deepEqual(await collect([]), []);
245 assert.deepEqual(await collect(["\n"]), [""]);
246 });
247
248 test("lines split between chunks", async () => {
249 assert.deepEqual(await collect(["hel", "lo\nwor", "ld"]), ["hello", "world"]);
250 });
251
252 test("a line feed completes a carriage return from the chunk before", async () => {
253 assert.deepEqual(await collect(["a\r", "\nb"]), ["a", "b"]);
254 assert.deepEqual(await collect(["a\r", "\n", "\nb"]), ["a", "", "b"]);
255 assert.deepEqual(await collect(["a\r", "b\r", "\r\n"]), ["a", "b", ""]);
256 });
257
258 test("a character split between chunks", async () => {
259 const bytes = string.encodeUtf8("caf\u00e9\n");
260 assert.deepEqual(await collect([bytes.subarray(0, 4), bytes.subarray(4)]), ["caf\u00e9"]);
261 });
262
263 test("leaving early cancels the stream", async () => {
264 let cancelled = false;
265 const readable = new ReadableStream<Uint8Array>({
266 pull: (controller) => controller.enqueue(string.encodeUtf8("again\n")),
267 cancel: () => void (cancelled = true),
268 });
269 for await (const line of stream.lines(readable)) {
270 assert.equal(line, "again");
271 break;
272 }
273 assert.ok(cancelled);
274 });
275});
276
228277import { strict as assert } from "node:assert";
229278import { describe, test } from "vitest";
230279import * as stream from "./stream.ts";
280import * as string from "./string.ts";
lib/stream.ts+25
......@@ -474,5 +474,30 @@ export function readWritePair(): [
474474 return [reader, unbuffered];
475475}
476476
477/**
478 * decode utf-8 and yield each line without its ending, which is any of `\n`,
479 * `\r\n`, or the lone `\r` that command line programs redraw a status with.
480 */
481export async function* lines(readable: ReadableStream<Uint8Array>): AsyncGenerator<string> {
482 const decoder = new TextDecoder();
483 let rest = "";
484 let endsInCarriageReturn = false;
485
486 for await (const bytes of readable) {
487 let chunk = decoder.decode(bytes, { stream: true });
488 if (chunk === "") continue;
489 // the `\n` of a `\r\n` that arrived in two chunks
490 if (endsInCarriageReturn && chunk[0] === "\n") chunk = chunk.slice(1);
491 endsInCarriageReturn = chunk.endsWith("\r");
492
493 const parts = (rest + chunk).split(/\r\n|\r|\n/);
494 rest = UNWRAP(parts.pop());
495 yield* parts;
496 }
497
498 rest += decoder.decode();
499 if (rest) yield rest;
500}
501
477502import { ASSERT, UNWRAP } from "./assert.ts";
478503import * as string from "./string.ts";
lib/subprocess.test.ts created+245
......@@ -0,0 +1,245 @@
1const node = (script: string): subprocess.Command => [process.execPath, "-e", script];
2/** like `node`, for a script that imports the library */
3const module = (script: string): subprocess.Command => [
4 process.execPath,
5 "--no-warnings",
6 "--input-type=module",
7 "-e",
8 script,
9];
10const lib = (file: string) => JSON.stringify(new URL(file, import.meta.url).href);
11
12const messages: log.Message[] = [];
13log.replaceGlobalMessageDestination((message) => messages.push(message));
14
15async function until(condition: () => unknown) {
16 while (!condition()) await async.delay(5);
17}
18
19describe("execute", () => {
20 test("returns stdout", async () => {
21 assert.equal(await subprocess.execute(node(`console.log("hello")`)), "hello\n");
22 });
23
24 test("does not stall on output larger than a pipe buffer", async () => {
25 const script = `process.stderr.write("e".repeat(300000)); process.stdout.write("o".repeat(300000))`;
26 assert.equal((await subprocess.execute(node(script))).length, 300000);
27 });
28
29 test("a failure carries the exit code, the command, and stderr", async () => {
30 const cmd = node(`console.error("it broke"); process.exit(3)`);
31 const err = await subprocess.execute(cmd).catch((err) => err);
32
33 assert.ok(err instanceof subprocess.ExitError);
34 assert.equal(err.message, `${path.basename(process.execPath)} failed with code 3`);
35 assert.equal(err.code, 3);
36 assert.equal(err.signal, null);
37 assert.equal(err.stderr, "it broke\n");
38 assert.equal(err.cmd, `${process.execPath} -e 'console.error("it broke"); process.exit(3)'`);
39 });
40
41 test("a missing program rejects with ENOENT", async () => {
42 const err = await subprocess.execute(["clover-lib-missing-program"]).catch((err) => err);
43 assert.equal(exception.code(err), "ENOENT");
44 });
45
46 test("env is added to the environment, and cwd is used", async () => {
47 const script = `console.log(process.env.CLOVER_TEST, !!process.env.PATH, process.cwd())`;
48 const cwd = fs.realpathSync(os.tmpdir());
49 const out = await subprocess.execute(node(script), { env: { CLOVER_TEST: "yes" }, cwd });
50 assert.equal(out, `yes true ${cwd}\n`);
51 });
52});
53
54describe("spawn", () => {
55 test("wait resolves on exit code zero", async () => {
56 await subprocess.spawn(node("")).wait();
57 });
58
59 test("a failure nobody waits for is not an unhandled rejection", async () => {
60 const proc = subprocess.spawn(node(`process.exit(1)`));
61 await async.delay(300);
62 await assert.rejects(proc.wait(), { message: /failed with code 1$/ });
63 });
64
65 test("stdin and stdout are web streams when piped", async () => {
66 const proc = subprocess.spawn(node(`process.stdin.pipe(process.stdout)`), {
67 stdin: "pipe",
68 stdout: "pipe",
69 });
70 const writer = proc.stdin.getWriter();
71 await writer.write(string.encodeUtf8("round trip"));
72 await writer.close();
73
74 assert.equal(await new Response(proc.stdout).text(), "round trip");
75 assert.equal(proc.stderr, null);
76 await proc.wait();
77 });
78
79 test("a status that redraws itself is read before its next line is written", async () => {
80 const script = `
81 process.stderr.write("frame=1\\r");
82 process.stdin.resume().on("end", () => process.stderr.write("frame=2\\rdone\\n"));
83 `;
84 const proc = subprocess.spawn(node(script), { stdin: "pipe", stderr: "pipe" });
85 const seen: string[] = [];
86 for await (const line of stream.lines(proc.stderr)) {
87 seen.push(line);
88 if (line === "frame=1") await proc.stdin.close();
89 }
90 await proc.wait();
91
92 assert.deepEqual(seen, ["frame=1", "frame=2", "done"]);
93 });
94
95 test("kill is reported as the signal", async () => {
96 const proc = subprocess.spawn(node(`setInterval(() => {}, 1000)`));
97 proc.kill("SIGKILL");
98 const err = await proc.wait().catch((err) => err);
99
100 assert.ok(err instanceof subprocess.ExitError);
101 assert.equal(err.signal, "SIGKILL");
102 assert.equal(err.code, null);
103 assert.match(err.message, /failed with signal SIGKILL$/);
104 });
105
106 test("aborting kills the process and rejects with the reason", async () => {
107 const ctrl = new AbortController();
108 const reason = new Error("rescheduled");
109 const proc = subprocess.spawn(node(`setInterval(() => {}, 1000)`), { signal: ctrl.signal });
110 ctrl.abort(reason);
111
112 assert.equal(await proc.wait().catch((err) => err), reason);
113 assert.throws(() => subprocess.spawn(node(""), { signal: ctrl.signal }), reason);
114 await assert.rejects(subprocess.execute(node(""), { signal: ctrl.signal }), reason);
115 });
116
117 test("disposing kills a running process, without having read its pipe", async () => {
118 let escaped: subprocess.Subprocess;
119 {
120 await using proc = subprocess.spawn(
121 node(`process.stdout.write("x".repeat(300000)); setInterval(() => {}, 1000)`),
122 { stdout: "pipe" },
123 );
124 escaped = proc;
125 }
126 await assert.rejects(escaped.wait(), { signal: "SIGTERM" });
127 });
128});
129
130describe("ipc", () => {
131 test("messages go both ways, and the end of the channel is heard", async () => {
132 const proc = subprocess.spawn(node(`process.on("message", (m) => process.send({ echo: m }))`), {
133 ipc: true,
134 });
135 const reply = new Promise((resolve) => proc.on("message", resolve));
136 const disconnect = new Promise<void>((resolve) => proc.on("disconnect", resolve));
137 proc.send("hello");
138
139 assert.deepEqual(await reply, { echo: "hello" });
140 proc.kill();
141 await disconnect;
142 const error = await new Promise((resolve) => proc.send("too late", resolve));
143 assert.equal(exception.code(error), "ERR_IPC_CHANNEL_CLOSED");
144 });
145
146 test("send needs the option", async () => {
147 await using proc = subprocess.spawn(node(""));
148 assert.throws(() => proc.send("hello"), /ipc/);
149 });
150
151 test("a prompt here can evaluate in the child", async () => {
152 const script = `
153 import * as repl from ${lib("./repl.ts")};
154 repl.serve(repl.javascript({ globals: { answer: 42 } }), process);
155 `;
156 await using proc = subprocess.spawn(module(script), { ipc: true });
157 const printed: string[] = [];
158 await repl.remote(proc).eval("answer + 1", (value) => printed.push(String(value)));
159
160 assert.deepEqual(printed.map(ansi.strip), ["43"]);
161 });
162});
163
164describe("progress", () => {
165 const script = `
166 import * as log from ${lib("./log.ts")};
167 import * as progress from ${lib("./progress.ts")};
168 log.info("from the child");
169 const task = progress.start("child task", { total: 2 });
170 task.log.warn("on the task");
171 task.value = 1;
172 process.on("message", () => {
173 progress.start("started last").log.error("on a node at exit");
174 log.error("last words");
175 process.exit(1);
176 });
177 `;
178
179 test("the child starts on the node it was handed, and logs through this process", async () => {
180 const root = new progress.Root();
181 const owner = root.start("owner");
182 const proc = subprocess.spawn(module(script), { progress: owner, ipc: true });
183 const children = () => UNWRAP(root.active[0]).children;
184 await until(() => children()[0]?.value === 1);
185
186 const task = UNWRAP(children()[0]);
187 assert.deepEqual([task.text, task.total], ["child task", 2]);
188 assert.deepEqual(task.logs.map((m) => [m.level, m.text]), [["warn", "on the task"]]);
189 assert.deepEqual(messages.map((m) => [m.level, m.text]), [["info", "from the child"]]);
190
191 // what is said right before exiting is in by the time `wait` is over
192 const ended: string[] = [];
193 root.on("node-end", (node) => ended.push(`${node.text}: ${node.logs.map((m) => m.text)}`));
194 proc.send("finish");
195 await assert.rejects(proc.wait(), subprocess.ExitError);
196
197 assert.deepEqual(messages.at(-1)?.text, "last words");
198 assert.deepEqual(ended.sort(), ["child task: on the task", "started last: on a node at exit"]);
199 assert.deepEqual(children(), []);
200 });
201
202 test("only the child that was handed the node reports to it", async () => {
203 // the grandchild inherits the environment, but not the pipe
204 const between = `
205 import * as child_process from "node:child_process";
206 const args = ["--no-warnings", "--input-type=module", "-e", process.argv[1]];
207 child_process.spawnSync(process.execPath, args, { stdio: "inherit" });
208 `;
209 const grandchild = `
210 import * as log from ${lib("./log.ts")};
211 import * as progress from ${lib("./progress.ts")};
212 progress.start("grandchild task");
213 log.info("from the grandchild");
214 `;
215 const root = new progress.Root();
216 const started: string[] = [];
217 root.on("node-start", (node) => started.push(node.text));
218 const proc = subprocess.spawn([...module(between), "--", grandchild], {
219 progress: root,
220 stdout: "pipe",
221 });
222 const stdout = await new Response(proc.stdout).text();
223 await proc.wait();
224
225 assert.match(stdout, /from the grandchild/);
226 assert.deepEqual(started, []);
227 });
228});
229
230import { strict as assert } from "node:assert";
231import * as fs from "node:fs";
232import * as os from "node:os";
233import * as path from "node:path";
234import * as process from "node:process";
235import { describe, test } from "vitest";
236import { UNWRAP } from "./assert.ts";
237import * as async from "./async.ts";
238import * as exception from "./exception.ts";
239import * as log from "./log.ts";
240import * as progress from "./progress.ts";
241import * as repl from "./repl.ts";
242import * as stream from "./stream.ts";
243import * as string from "./string.ts";
244import * as ansi from "./string/ansi.ts";
245import * as subprocess from "./subprocess.ts";
lib/subprocess.ts+164-85
......@@ -1,95 +1,170 @@
11/**
2 * This file is expecting a full rewrite to abstract the Node API away
3 * entirely, exposing web streams I/O.
4 *
2 * spawn and manage other programs as sub-processes.
53 * @module
64 */
75
8const execFileRaw: typeof child_process.execFile.__promisify__ = util.promisify(
9 child_process.execFile,
10);
11/*** @deprecated */
12export const exec: typeof execFileRaw = ((
13 ...args: Parameters<typeof execFileRaw>
14) =>
15 execFileRaw(...args).catch((e: any) => {
16 if (e?.message?.startsWith?.("Command failed")) {
17 if (e.code > 2 ** 31) e.code |= 0;
18 const code = e.signal ? `signal ${e.signal}` : `code ${e.code}`;
19 e.message = `${e.cmd.split(" ")[0]} failed with ${code}`;
20 }
21 throw e;
22 })) as any;
23
24// const activePids = new Set<number>();
25
26/*** @deprecated */
27export interface SpawnOptions extends child_process.SpawnOptions {
28 cmd: string[];
6/**
7 * execute a subprocess and wait for it to exit, returning what it wrote to
8 * stdout. stderr is captured too, shown via {@linkcode ExitError} on failure.
9 * this is, essentially, a simpler version of {@linkcode spawn}.
10 */
11export async function execute(
12 cmd: Command,
13 opts?: Omit<Options, "stdout" | "stderr">,
14): Promise<string> {
15 await using proc = spawn(cmd, { ...opts, stdout: "pipe", stderr: "pipe" });
16 const [stdout, stderr] = await Promise.all([
17 new Response(proc.stdout).text(),
18 new Response(proc.stderr).text(),
19 ]);
20
21 try {
22 await proc.wait();
23 } catch (err) {
24 if (err instanceof ExitError) err.stderr = stderr;
25 throw err;
26 }
27 return stdout;
2928}
30/*** @deprecated */
31export function spawn(
32 { cmd, ...rest }: SpawnOptions,
33): child_process.ChildProcess {
34 // rest.detached ??= true;
35 const process = child_process.spawn(cmd[0]!, cmd.slice(1), rest);
36 // const pid = rest.detached && process.pid ? -process.pid : process.pid;
37 // if (pid) {
38 // activePids.add(pid);
39 // process.on("exit", () => void activePids.delete(pid));
40 // }
41 return process;
29
30/**
31 * start a command and return at once. how it ends, which includes not being
32 * able to start, is found by {@linkcode Subprocess.wait}.
33 */
34export function spawn<const O extends Options>(cmd: Command, opts?: O): Subprocess<O> {
35 const { stdin = "ignore", stdout = "inherit", stderr = "inherit" }: Options = opts ?? {};
36 const { cwd, env, signal, ipc, progress: handoff }: Options = opts ?? {};
37 signal?.throwIfAborted();
38
39 const stdio: Array<Stdio | "ipc"> = [stdin, stdout, stderr];
40 if (handoff) stdio.push("pipe");
41 if (ipc) stdio.push("ipc");
42
43 const child = child_process.spawn(cmd[0], cmd.slice(1), {
44 cwd,
45 env: { ...process.env, ...env, ...handoff && progress.childEnv(progressFd) },
46 stdio,
47 });
48
49 const onAbort = () => void child.kill();
50 signal?.addEventListener("abort", onAbort);
51
52 const report = handoff
53 && progress.attachChild(toWeb(child.stdio[progressFd] as stream.Readable), handoff);
54
55 const exit = new Promise<void>((resolve, reject) => {
56 child.on("error", reject);
57 child.once("exit", (code, exitSignal) => {
58 if (signal?.aborted) reject(signal.reason);
59 else if (code === 0) resolve();
60 else reject(new ExitError(cmd, code, exitSignal));
61 });
62 })
63 // the last of what the child reported is in before `wait` is over
64 .finally(() => report)
65 .finally(() => signal?.removeEventListener("abort", onAbort));
66
67 // a failure is only an error to those who `wait` for it
68 exit.catch(() => {});
69 report?.catch(() => {});
70
71 const proc: Subprocess = {
72 stdin: child.stdin && stream.Writable.toWeb(child.stdin),
73 stdout: child.stdout && toWeb(child.stdout),
74 stderr: child.stderr && toWeb(child.stderr),
75 wait: () => exit,
76 kill: (killSignal) => void child.kill(killSignal),
77 send(message, callback) {
78 ASSERT(ipc, "`send` is for a process started with the `ipc` option");
79 child.send(message as child_process.Serializable, callback);
80 },
81 on: (event: string, listener: (message: unknown) => void) => void child.on(event, listener),
82 off: (event: string, listener: (message: unknown) => void) => void child.off(event, listener),
83 async [Symbol.asyncDispose]() {
84 child.kill();
85 await exit.catch(() => {});
86 // a pipe that was not read to its end holds a file descriptor
87 for (const pipe of child.stdio) pipe?.destroy();
88 },
89 };
90 return proc as Subprocess<O>;
4291}
4392
44/*** @deprecated */
45export interface ForkOptions extends child_process.ForkOptions {
46 cmd: string[];
93/** where a child has the pipe for its `progress`: after stdin, stdout, and stderr */
94const progressFd = 3;
95
96/** node.js types its web streams apart from the global ones */
97function toWeb(pipe: stream.Readable) {
98 return stream.Readable.toWeb(pipe) as ReadableStream<Uint8Array>;
4799}
48100
49/*** @deprecated */
50export function fork(
51 { cmd, ...rest }: ForkOptions,
52): child_process.ChildProcess {
53 // rest.detached ??= true;
54 const process = child_process.fork(cmd[0]!, cmd.slice(1), rest);
55 // const pid = rest.detached && process.pid ? -process.pid : process.pid;
56 // if (pid) {
57 // activePids.add(pid);
58 // process.on("exit", () => void activePids.delete(pid));
59 // }
60 return process;
101/** a program followed by its arguments */
102export type Command = readonly [string, ...string[]];
103
104/** `"pipe"` puts a stream on the {@linkcode Subprocess} */
105export type Stdio = "pipe" | "inherit" | "ignore";
106
107export interface Options {
108 cwd?: string;
109 /** added to the environment of this process */
110 env?: Record<string, string>;
111 /** kills the process when aborted, and `wait` rejects with its reason */
112 signal?: AbortSignal;
113 /** @default "ignore" */
114 stdin?: Stdio;
115 /** @default "inherit" */
116 stdout?: Stdio;
117 /** @default "inherit" */
118 stderr?: Stdio;
119 /** open a channel for messages with a child that is node.js too, where it is `process` */
120 ipc?: boolean;
121 /**
122 * hand over a place in the progress of this process. a child using
123 * `@clo/lib/progress` has it as its global root, and logs through this
124 * process instead of writing to the terminal by itself.
125 */
126 progress?: progress.Ref;
61127}
62128
63/*** @deprecated */
64export async function spawnAndWait(
65 opts: SpawnOptions,
66): Promise<{ stdout: Buffer | null; stderr: Buffer | null }> {
67 opts.stdio ??= ["ignore", "inherit", "inherit"];
68 const process = spawn(opts);
69 const output = Promise.all([
70 process.stdout ? Array.fromAsync(process.stdout).then(Buffer.concat) : null,
71 process.stderr ? Array.fromAsync(process.stderr).then(Buffer.concat) : null,
72 ]);
73 let [code, signal] = (await events.EventEmitter.once(process, "exit")) as [
74 number | null,
75 string | null,
76 ];
77 if (code !== 0) {
129/** disposing kills the process if it is still running, and waits for it to exit */
130export interface Subprocess<O extends Options = Options> extends AsyncDisposable {
131 stdin: Piped<O["stdin"], WritableStream<Uint8Array>>;
132 stdout: Piped<O["stdout"], ReadableStream<Uint8Array>>;
133 stderr: Piped<O["stderr"], ReadableStream<Uint8Array>>;
134 /** resolves once the process exits with code zero. any other exit rejects with an {@linkcode ExitError} */
135 wait(): Promise<void>;
136 kill(signal?: NodeJS.Signals): void;
137
138 /** send a message over the `ipc` channel. `callback` hears if it could not be */
139 send(message: unknown, callback?: (error: Error | null) => void): void;
140 on(event: "message", listener: (message: unknown) => void): void;
141 on(event: "disconnect", listener: () => void): void;
142 off(event: "message", listener: (message: unknown) => void): void;
143 off(event: "disconnect", listener: () => void): void;
144}
145
146type Piped<S, T> = S extends "pipe" ? T : null;
147
148/** how a process ended, when that was not with exit code zero */
149export class ExitError extends Error {
150 /** as it would be typed into a shell */
151 cmd: string;
152 code: number | null;
153 signal: NodeJS.Signals | null;
154 /** only when {@linkcode execute} ran the command */
155 declare stderr?: string;
156
157 constructor(cmd: Command, code: number | null, signal: NodeJS.Signals | null) {
158 // windows prints exit codes signed, such as -1073741819 for an access violation
78159 if (code !== null && code > 2 ** 31) code |= 0;
79 const exit = signal ? `signal ${signal}` : `code ${code}`;
80 throw new Error(`${opts.cmd[0]} failed with ${exit}`);
160 super(`${path.basename(cmd[0])} failed with ${signal ? `signal ${signal}` : `code ${code}`}`);
161
162 this.cmd = cmd.map(string.escapeShellArgument).join(" ");
163 this.code = code;
164 this.signal = signal;
81165 }
82 const [stdout, stderr] = await output;
83 return { stdout, stderr };
84166}
85167
86// process.on("exit", () => {
87// for (const pid of activePids) {
88// console.info(`Kill ${Math.abs(pid)}`);
89// process.kill(pid, 9);
90// }
91// });
92
93168async function findInodeForPort(port: number) {
94169 const hexPort = port.toString(16).toUpperCase().padStart(4, "0");
95170 for (const f of ["/proc/net/tcp", "/proc/net/tcp6"]) {
......@@ -129,22 +204,23 @@ async function commandFromPortLinux(port: number) {
129204}
130205
131206async function commandFromPortMac(port: number) {
132 const { stdout: lsofOut } = await exec("lsof", ["-ti", `tcp:${port}`]);
207 const lsofOut = await execute(["lsof", "-ti", `tcp:${port}`]);
133208 const pid = lsofOut.trim().split("\n")[0];
134209 if (!pid) throw new Error(`no process on port ${port}`);
135 const { stdout: psOut } = await exec("ps", ["-p", pid, "-o", "args="]);
210 const psOut = await execute(["ps", "-p", pid, "-o", "args="]);
136211 return { pid: Number(pid), command: psOut.trim() };
137212}
138213
139214async function commandFromPortWindows(port: number) {
140 const { stdout } = await exec("netstat", ["-ano"]);
215 const stdout = await execute(["netstat", "-ano"]);
141216 const line = stdout.split("\n").find(l => {
142217 const m = l.match(/\s+[\d.:]+:(\d+)\s+/);
143218 return m && parseInt(m[1]!) === port && l.includes("LISTENING");
144219 });
145220 if (!line) throw new Error(`no process on port ${port}`);
146221 const pid = line.trim().split(/\s+/).at(-1);
147 const { stdout: psOut } = await exec("powershell", [
222 const psOut = await execute([
223 "powershell",
148224 "-NoProfile",
149225 "-Command",
150226 `(Get-CimInstance Win32_Process -Filter "ProcessId=${pid}").CommandLine`,
......@@ -155,7 +231,7 @@ async function commandFromPortWindows(port: number) {
155231}
156232
157233/** Racily searches for what process is taking up a given port. */
158export async function lookupProcessOnPort(port: number): Promise<ProcessLookup> {
234export function lookupProcessOnPort(port: number): Promise<ProcessLookup> {
159235 switch (os.platform()) {
160236 case "linux":
161237 return commandFromPortLinux(port);
......@@ -173,9 +249,12 @@ interface ProcessLookup {
173249 command: string;
174250}
175251
176import { Buffer } from "node:buffer";
177252import * as child_process from "node:child_process";
178import * as events from "node:events";
179253import * as fs from "node:fs/promises";
180254import * as os from "node:os";
181import * as util from "node:util";
255import * as path from "node:path";
256import * as process from "node:process";
257import * as stream from "node:stream";
258import { ASSERT } from "./assert.ts";
259import * as progress from "./progress.ts";
260import * as string from "./string.ts";
lib/subprocess/ffmpeg.ts+20-38
......@@ -32,23 +32,21 @@ export async function spawn(options: SpawnOptions) {
3232 const { ffmpeg = "ffmpeg", args, cwd } = options;
3333 using node = options.progress;
3434 const title = node.text;
35 const proc = child_process.spawn(ffmpeg, [...defaultExtraOptions, ...args], {
36 stdio: ["ignore", "inherit", "pipe"],
37 env: { ...process.env, SVT_LOG: "2" },
38 cwd: cwd?.toString(),
35 await using proc = subprocess.spawn([ffmpeg, ...defaultExtraOptions, ...args], {
36 stderr: "pipe",
37 env: { SVT_LOG: "2" },
38 cwd,
3939 signal: options.signal,
4040 });
41
4142 const parser = new Parse();
42 let running = true;
43 const splitter = readline.createInterface({ input: proc.stderr });
44 splitter.on("line", (line) => {
43 for await (const line of stream.lines(proc.stderr)) {
4544 const result = parser.onLine(line);
4645 if (result.kind === "ignore") {
47 return;
46 continue;
4847 } else if (result.kind === "log") {
4948 node.log[result.level](result.message);
5049 } else if (result.kind === "progress") {
51 if (!running) return;
5250 const { frame, totalFrames, fps, speed } = result;
5351 node.value = frame;
5452 node.total = totalFrames;
......@@ -59,17 +57,8 @@ export async function spawn(options: SpawnOptions) {
5957 ].filter(Boolean).join(", ");
6058 node.text = `${title}${extras.length > 0 ? ` (${extras})` : ""}`;
6159 } else result satisfies never;
62 });
63 const [code, signal] = await events.once(proc, "close");
64 running = false;
65 if (code !== 0) {
66 const fmt = code ? `code ${code}` : `signal ${signal}`;
67 const e: any = new Error(`ffmpeg failed with ${fmt}`);
68 e.args = [ffmpeg, ...args].map(string.escapeShellArgument).join(" ");
69 e.code = code;
70 e.signal = signal;
71 throw e;
7260 }
61 await proc.wait();
7362}
7463
7564export interface Streams {
......@@ -78,21 +67,18 @@ export interface Streams {
7867}
7968
8069export async function probeStreams(path: string): Promise<Streams> {
81 const { stdout } = await subprocess.spawnAndWait({
82 cmd: [
83 "ffprobe",
84 "-v",
85 "error",
86 "-show_entries",
87 "stream=codec_type",
88 path.toString(),
89 "-of",
90 "default=noprint_wrappers=1:nokey=1",
91 ],
92 stdio: "pipe",
93 });
70 const stdout = await subprocess.execute([
71 "ffprobe",
72 "-v",
73 "error",
74 "-show_entries",
75 "stream=codec_type",
76 path.toString(),
77 "-of",
78 "default=noprint_wrappers=1:nokey=1",
79 ]);
9480
95 const lines = new TextDecoder().decode(stdout!).trim().split("\n");
81 const lines = stdout.trim().split("\n");
9682 return {
9783 audio: lines.includes("audio"),
9884 video: lines.includes("video"),
......@@ -201,11 +187,7 @@ export class Parse {
201187 }
202188}
203189
204import * as child_process from "node:child_process";
205import events from "node:events";
206190import * as path from "node:path";
207import * as process from "node:process";
208import * as readline from "node:readline";
209191import type * as progress from "../progress.ts";
210import * as string from "../string.ts";
192import * as stream from "../stream.ts";
211193import * as subprocess from "../subprocess.ts";
src/bin/deploy.ts+7-10
......@@ -41,16 +41,13 @@ export async function main() {
4141 ],
4242 progress: progress.start("copy site files"),
4343 });
44 await subprocess.spawnAndWait({
45 stdio: "inherit",
46 cmd: [
47 "ssh",
48 "paperclover.net",
49 "cd paperclover.net && "
50 + "npm i &&"
51 + "pm2 restart site",
52 ],
53 });
44 await subprocess.spawn([
45 "ssh",
46 "paperclover.net",
47 "cd paperclover.net && "
48 + "npm i &&"
49 + "pm2 restart site",
50 ], { stdin: "inherit" }).wait();
5451}
5552
5653import { Path } from "#sitegen/path";
src/file-viewer/indexer/processors/media.ts+8-4
......@@ -14,7 +14,8 @@ const procDuration: Processor = {
1414 include: rules.extsDuration,
1515 cores: 1,
1616 async run({ path, mediaFile, signal }) {
17 const { stdout } = await subprocess.exec(ffprobeBin!, [
17 const stdout = await subprocess.execute([
18 ffprobeBin!,
1819 "-v",
1920 "error",
2021 "-show_entries",
......@@ -55,7 +56,8 @@ const procDimensions: Processor = {
5556 }
5657 } else if (rules.extsImage.has(ext)) {
5758 // Use magick to observe streams
58 const { stdout } = await subprocess.exec("magick", [
59 const stdout = await subprocess.execute([
60 "magick",
5961 "identify",
6062 "-auto-orient",
6163 "-format",
......@@ -68,7 +70,8 @@ const procDimensions: Processor = {
6870 }
6971 } else {
7072 // Use ffprobe to observe streams
71 const { stdout } = await subprocess.exec("ffprobe", [
73 const stdout = await subprocess.execute([
74 "ffprobe",
7275 "-v",
7376 "error",
7477 "-select_streams",
......@@ -127,7 +130,8 @@ const procImageSubsets: Processor = {
127130 cores: 2,
128131 subkey: `${size}${ext}`,
129132 async producer(dir) {
130 await subprocess.exec(ffmpegBin!, [
133 await subprocess.execute([
134 ffmpegBin!,
131135 ...ffmpegOptions,
132136 "-i",
133137 path.toString(),
src/file-viewer/indexer/scrub.ts+10-11
......@@ -4,10 +4,6 @@
44// first); modifies the file in place while preserving timestamps.
55const exiftoolBin = testProgram("exiftool");
66
7// `-ee` on an iphone video dumps per-frame embedded metadata, which easily
8// exceeds execFile's default 1MB stdout buffer
9const execOptions = { maxBuffer: 64 * 1024 * 1024 };
10
117export async function scrubLocationMetadata(
128 path: Path,
139 stats: fs.Stats,
......@@ -30,21 +26,23 @@ export async function scrubLocationMetadata(
3026 case ".jpg":
3127 case ".jpeg":
3228 case ".png":
33 const { stdout: gpsCheck } = await subprocess.exec("exiftool", [
29 const gpsCheck = await subprocess.execute([
30 "exiftool",
3431 "-gps:all",
3532 path.toString(),
36 ], execOptions);
33 ]);
3734 hasLocation = gpsCheck.trim().length > 0;
3835 args = ["-gps:all=", path.toString(), "-o", tempOutput.toString()];
3936 break;
4037 case ".mov":
4138 case ".mp4":
42 const { stdout: videoCheck } = await subprocess.exec("exiftool", [
39 const videoCheck = await subprocess.execute([
40 "exiftool",
4341 "-ee",
4442 "-G3",
4543 "-s",
4644 path.toString(),
47 ], execOptions);
45 ]);
4846 hasLocation = videoCheck.includes("GPS")
4947 || videoCheck.includes("Location");
5048 args = [
......@@ -56,12 +54,13 @@ export async function scrubLocationMetadata(
5654 ];
5755 break;
5856 case ".m4a":
59 const { stdout: m4aCheck } = await subprocess.exec("exiftool", [
57 const m4aCheck = await subprocess.execute([
58 "exiftool",
6059 "-ee",
6160 "-G3",
6261 "-s",
6362 path.toString(),
64 ], execOptions);
63 ]);
6564 hasLocation = m4aCheck.includes("GPS")
6665 || m4aCheck.includes("Location")
6766 || m4aCheck.includes("Filename")
......@@ -102,7 +101,7 @@ export async function scrubLocationMetadata(
102101 await tempOutput.delete({ force: true });
103102
104103 // Remove metadata
105 await subprocess.exec("exiftool", args, execOptions);
104 await subprocess.execute(["exiftool", ...args]);
106105 if (!tempOutput.ifExistsSync()) {
107106 throw new Error(`Failed to create output file: ${tempOutput}`);
108107 }