From d3185018db6284979ba12696d8b8e34be6985f39 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Mon, 12 Oct 2026 15:33:50 -0700 Subject: [PATCH] feat(lib/subprocess): rewrite deprecated apis see #20 Assisted-by: claude-opus-5.5 --- framework/watch.ts | 32 +-- lib/examples/subprocess.ts | 30 +++ lib/node.ts | 3 + lib/progress.ts | 157 +++++++++--- lib/readme.changes.md | 11 + lib/stream.test.ts | 50 ++++ lib/stream.ts | 25 ++ lib/subprocess.test.ts | 245 ++++++++++++++++++ lib/subprocess.ts | 261 +++++++++++++------- lib/subprocess/ffmpeg.ts | 58 ++--- src/bin/deploy.ts | 17 +- src/file-viewer/indexer/processors/media.ts | 12 +- src/file-viewer/indexer/scrub.ts | 21 +- 13 files changed, 713 insertions(+), 209 deletions(-) create mode 100644 lib/examples/subprocess.ts create mode 100644 lib/subprocess.test.ts diff --git a/framework/watch.ts b/framework/watch.ts index c2dcdb20ccf325bc9f3735949300fef3e5c548d5..3f27c04b90bb7d292eecc129cc85610ecbbd4bc0 100644 --- a/framework/watch.ts +++ b/framework/watch.ts @@ -2,10 +2,7 @@ const debounceMilliseconds = 25; -let server: child_process.ChildProcess | null = null; -process.on("beforeExit", () => { - server?.removeListener("close", onSubprocessClose); -}); +let server: subprocess.Subprocess | null = null; let watch: Watch; @@ -18,22 +15,18 @@ export async function main() { } function serve() { - if (server) { - server.removeListener("close", onSubprocessClose); - server.kill(); - } - server = subprocess.fork({ - cmd: [".clover/o/backend.js", "--development"], - stdio: "inherit", - execArgv: ["--no-warnings"], + server?.kill(); + const backend = server = subprocess.spawn( + [process.execPath, "--no-warnings", ".clover/o/backend.js", "--development"], + { stdin: "inherit", ipc: true }, + ); + + backend.wait().then(() => "", (err) => `: ${exception.message(err)}`).then((reason) => { + // a backend that was replaced is not the one to report + if (server !== backend) return; + server = null; + console.error(`Backend process exited${reason}`); }); - server.on("close", onSubprocessClose); -} - -function onSubprocessClose(code: number | null, signal: string | null) { - server = null; - const status = code != null ? `code ${code}` : `signal ${signal}`; - console.error(`Backend process exited with ${status}`); } let isRunning = false; @@ -182,6 +175,7 @@ class Watch { } } +import * as exception from "@clo/lib/exception"; import * as subprocess from "@clo/lib/subprocess"; import * as child_process from "node:child_process"; import * as fs from "node:fs"; diff --git a/lib/examples/subprocess.ts b/lib/examples/subprocess.ts new file mode 100644 index 0000000000000000000000000000000000000000..87d658f5204009ed4cc4a40e8c86de002fe18a10 --- /dev/null +++ b/lib/examples/subprocess.ts @@ -0,0 +1,30 @@ +// a build that fans out to worker processes. each worker is this same file, +// started with a node of the build as its global progress root: what it +// starts and logs is drawn by the one process that has the terminal. +const [worker] = process.argv.slice(2); +const steps = ["parse", "transform", "minify"]; + +if (worker) { + log.info(`${worker}: found ${steps.length} steps`); + for (const step of steps) { + using task = progress.start(step, { total: 30 }); + for (let i = 0; i < 30; i += 1) { + await async.delay(20 + Math.random() * 60); + task.value += 1; + if (i === 15) task.log.warn(`${step} is taking a while`); + } + } + log.info(`${worker}: done`); +} else { + using build = progress.start("build"); + await Promise.all(["pages", "styles", "scripts"].map(async (name) => { + using node = build.start(name); + await subprocess.spawn([process.execPath, import.meta.filename, name], { progress: node }).wait(); + })); + log.info("built"); +} + +import * as async from "../async.ts"; +import * as log from "../log.ts"; +import * as progress from "../progress.ts"; +import * as subprocess from "../subprocess.ts"; diff --git a/lib/node.ts b/lib/node.ts index 3e35d3132d5eda8c1109622f940b4ce61860a78b..61823672474a27a206ed219659008618c1bac78c 100644 --- a/lib/node.ts +++ b/lib/node.ts @@ -27,7 +27,9 @@ interface Process { removeListener(event: string, callback: () => void): this; listenerCount(event: string): number; kill(pid: number, signal: string): boolean; + env: Partial>; pid: number; + ppid: number; platform: string; versions: { node?: string; bun?: string; deno?: string }; stdin: TtyInput; @@ -72,6 +74,7 @@ interface Builtins { readFileSync(file: string): Uint8Array; writeFileSync(file: string, data: string, options?: { mode?: number }): void; appendFileSync(file: string, data: string, options?: { mode?: number }): void; + writeSync(fd: number, data: string): number; mkdirSync(dir: string, options: { recursive: true }): void; }; "inspector": { diff --git a/lib/progress.ts b/lib/progress.ts index 00af0fc2d9d918d126e1d73eddf49c9d15910e92..03e14c0d05cafae19fc9f01fb6777c203e2d3a58 100644 --- a/lib/progress.ts +++ b/lib/progress.ts @@ -441,7 +441,7 @@ function newNode( showTotal: opts.showTotal ?? true, passive: opts.passive ?? false, hidden: opts.hidden ?? false, - logs: [], + logs: [...opts.messages ?? []], children: [], parent: null, sortChildren: opts.sortChildren ?? null, @@ -957,40 +957,7 @@ export function encodeEventStream< isFirst = false; } - // TODO: why does ts require two args - stack.use(root.on("node-start", (newNode, _) => { - if (!newNode.parent) s.rootChildrenUpdated = true; - s.changed.set(newNode, new Set()); - emitSoon(); - })); - stack.use(root.on("node-change", (node, key) => { - let set = s.changed.get(node); - if (!set) s.changed.set(node, set = new Set()); - set.add(key); - emitSoon(); - })); - stack.use(root.on("node-end", (node) => { - if (!node.parent) s.rootChildrenUpdated = true; - s.changed.delete(node); - s.deleted.add(node); // handle add and remove in same frame - emitSoon(); - })); - // custom events pass through the stream verbatim - stack.use(root.onAny((channel, args) => { - if (rootEvents.has(channel)) return; - s.pendingEvents.push([channel as string, ...args as ts.Json[]]); - emitSoon(); - })); - stack.use(root.on("end", (result) => { - s.pendingEvents.push(["end", result]); - s.ended = true; - emitNow(); - })); - stack.use(root.on("error", (error) => { - s.pendingEvents.push(["error", exceptions.message(error)]); - s.ended = true; - emitNow(); - })); + stack.use(s.observe(emitSoon, emitNow)); }, async pull(controller) { await ready.until((x) => x === true); @@ -1087,6 +1054,51 @@ class Encoder< } } + /** + * fill `changed`, `deleted`, and `pendingEvents` from the events of the + * root. `soon` is called when there is a delta worth sending, and `now` + * when it is the last one. + */ + observe(soon: () => void, now: () => void): DisposableStack { + const { root } = this; + const stack = new DisposableStack(); + // TODO: why does ts require two args + stack.use(root.on("node-start", (newNode, _) => { + if (!newNode.parent) this.rootChildrenUpdated = true; + this.changed.set(newNode, new Set()); + soon(); + })); + stack.use(root.on("node-change", (node, key) => { + let set = this.changed.get(node); + if (!set) this.changed.set(node, set = new Set()); + set.add(key); + soon(); + })); + stack.use(root.on("node-end", (node) => { + if (!node.parent) this.rootChildrenUpdated = true; + this.changed.delete(node); + this.deleted.add(node); // handle add and remove in same frame + soon(); + })); + // custom events pass through the stream verbatim + stack.use(root.onAny((channel, args) => { + if (rootEvents.has(channel)) return; + this.pendingEvents.push([channel as string, ...args as ts.Json[]]); + soon(); + })); + stack.use(root.on("end", (result) => { + this.pendingEvents.push(["end", result]); + this.ended = true; + now(); + })); + stack.use(root.on("error", (error) => { + this.pendingEvents.push(["error", exceptions.message(error)]); + this.ended = true; + now(); + })); + return stack; + } + getDelta(isFirst = false) { const deleted = Array.from( this.deleted, @@ -1830,7 +1842,79 @@ type EncodedKey = number & { brand: typeof kNode }; const globalKeyPool = new KeyPool(); -const global: Ref = /** @__PURE__ */ ((root = new Root()) => (attachToScreen(root, log), root))(); +/** names the pipe to a parent process that handed this one a place in its progress */ +const parentEnv = "CLOVER_PROGRESS"; + +/** + * the environment for a child process whose global root and log are to be + * read by {@linkcode attachChild}. `fd` is where the child has the pipe. + */ +export function childEnv(fd: number): Record { + return { [parentEnv]: `${fd}:${UNWRAP(node.process).pid}` }; +} + +/** + * read what a child process reports through the pipe that {@linkcode childEnv} + * named to it. everything on its global root is started on `target`, and what + * it logs globally is logged globally here. resolves once the child closes + * the pipe, which ends all it had started. + * + * `subprocess.spawn` does this with its `progress` option. + */ +export async function attachChild(readable: ReadableStream, target: Ref): Promise { + const decoder = new Decoder(target, new Events(), noop); + try { + for await (const line of stream.lines(readable)) { + const record: StreamEvent | log.Message = JSON.parse(line); + if (Array.isArray(record)) decoder.processEvent(record); + else log.writeMessage(record); + } + } finally { + decoder.reset(); + } +} + +/** + * the other end of {@linkcode attachChild}. writes are synchronous, so that + * the last words of a process are not lost when it exits. + */ +function tryAttachToParent(root: Root): boolean { + const { process } = node; + const fs = node.builtin("fs"); + const [fd, parent] = (process?.env[parentEnv] ?? "").split(":").map(Number); + // every descendant inherits the variable, but only the child has the pipe + if (!process || !fs || !fd || process.ppid !== parent) return false; + + const write = (record: StreamEvent | log.Message) => { + try { + fs.writeSync(fd, JSON.stringify(record) + "\n"); + } catch { + // the parent is gone, and so is anyone to tell + } + }; + + const encoder = new Encoder(root); + let timer: async.Cancelable | null = null; + const flush = () => { + timer?.cancel(); + timer = null; + write(encoder.getDelta()); + }; + const flushSoon = () => { + if (timer) return; + (timer = root.delay(1000 / 30)).then(flush); + }; + + encoder.observe(flushSoon, flush); + process.addListener("exit", () => timer && flush()); + // what changed on a node before a message is said stays before it + log.replaceGlobalMessageDestination((m) => (timer && flush(), write({ ...m, stack: undefined }))); + return true; +} + +const global: Ref = /** @__PURE__ */ ((root = new Root()) => ( + tryAttachToParent(root) || attachToScreen(root, log), root +))(); /** * a {@linkcode Ref} to the global progress root. unlike referencing the @@ -1861,6 +1945,7 @@ import { Events } from "./Events.ts"; import * as exceptions from "./exception.ts"; import * as log from "./log.ts"; import * as stack from "./log/stack.ts"; +import * as node from "./node.ts"; import * as stream from "./stream.ts"; import * as string from "./string.ts"; import * as ansi from "./string/ansi.ts"; diff --git a/lib/readme.changes.md b/lib/readme.changes.md index 55e20bf0d115f7394fe11cd2e55bf2e8e969e6f7..6a8ebf9dd08b4fc7c4d505e164292ed9d3d5f9e8 100644 --- a/lib/readme.changes.md +++ b/lib/readme.changes.md @@ -2,9 +2,18 @@ ## v8 +### breaking + +- `subprocess` is rewritten around web streams. + ### features - new library `repl`: an interactive prompt to add to long running programs. +- `stream.lines` reads a byte stream as lines of text. +- `subprocess`'s `progress` option hands a `progress.Ref` to a child process, + which becomes the global root of `progress` there. the child's global log is + written by the parent as well. `progress.childEnv` and `progress.attachChild` + are the two halves, for spawning by other means. - `log` - widgets can now read keyboard input. - finally fixes resize corruptions for long running widgets @@ -15,6 +24,8 @@ - `async.DedupeConcurrent` no longer causes unhandled rejections with correct caller error handling. +- `progress`: `StartOptions.messages` is no longer ignored, so a decoded stream + keeps the logs that a node had when it started. ## v7 diff --git a/lib/stream.test.ts b/lib/stream.test.ts index d5f4d7e0a0696c239339bfd746a14e18ddedd4dc..7e102cc1a046aede942df77e28bca2d0fe42f268 100644 --- a/lib/stream.test.ts +++ b/lib/stream.test.ts @@ -225,6 +225,56 @@ describe("BufferedReader", () => { }); }); +describe("lines", () => { + async function collect(chunks: Array) { + const readable = new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(typeof chunk === "string" ? string.encodeUtf8(chunk) : chunk); + } + controller.close(); + }, + }); + return await Array.fromAsync(stream.lines(readable)); + } + + test("every kind of line ending", async () => { + assert.deepEqual(await collect(["a\nb\r\nc\rd\n"]), ["a", "b", "c", "d"]); + assert.deepEqual(await collect(["a\n\nb\r\n\r\nc"]), ["a", "", "b", "", "c"]); + assert.deepEqual(await collect([]), []); + assert.deepEqual(await collect(["\n"]), [""]); + }); + + test("lines split between chunks", async () => { + assert.deepEqual(await collect(["hel", "lo\nwor", "ld"]), ["hello", "world"]); + }); + + test("a line feed completes a carriage return from the chunk before", async () => { + assert.deepEqual(await collect(["a\r", "\nb"]), ["a", "b"]); + assert.deepEqual(await collect(["a\r", "\n", "\nb"]), ["a", "", "b"]); + assert.deepEqual(await collect(["a\r", "b\r", "\r\n"]), ["a", "b", ""]); + }); + + test("a character split between chunks", async () => { + const bytes = string.encodeUtf8("caf\u00e9\n"); + assert.deepEqual(await collect([bytes.subarray(0, 4), bytes.subarray(4)]), ["caf\u00e9"]); + }); + + test("leaving early cancels the stream", async () => { + let cancelled = false; + const readable = new ReadableStream({ + pull: (controller) => controller.enqueue(string.encodeUtf8("again\n")), + cancel: () => void (cancelled = true), + }); + for await (const line of stream.lines(readable)) { + assert.equal(line, "again"); + break; + } + assert.ok(cancelled); + }); +}); + import { strict as assert } from "node:assert"; import { describe, test } from "vitest"; import * as stream from "./stream.ts"; +import * as string from "./string.ts"; diff --git a/lib/stream.ts b/lib/stream.ts index 6c89ea46d5e9158542120383e6d3c8dac4d956d8..633d60785376a07481b065917c78d441875964c3 100644 --- a/lib/stream.ts +++ b/lib/stream.ts @@ -474,5 +474,30 @@ export function readWritePair(): [ return [reader, unbuffered]; } +/** + * decode utf-8 and yield each line without its ending, which is any of `\n`, + * `\r\n`, or the lone `\r` that command line programs redraw a status with. + */ +export async function* lines(readable: ReadableStream): AsyncGenerator { + const decoder = new TextDecoder(); + let rest = ""; + let endsInCarriageReturn = false; + + for await (const bytes of readable) { + let chunk = decoder.decode(bytes, { stream: true }); + if (chunk === "") continue; + // the `\n` of a `\r\n` that arrived in two chunks + if (endsInCarriageReturn && chunk[0] === "\n") chunk = chunk.slice(1); + endsInCarriageReturn = chunk.endsWith("\r"); + + const parts = (rest + chunk).split(/\r\n|\r|\n/); + rest = UNWRAP(parts.pop()); + yield* parts; + } + + rest += decoder.decode(); + if (rest) yield rest; +} + import { ASSERT, UNWRAP } from "./assert.ts"; import * as string from "./string.ts"; diff --git a/lib/subprocess.test.ts b/lib/subprocess.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..c2b318f744d603b5ebf6cb0484344b05c86c3fd9 --- /dev/null +++ b/lib/subprocess.test.ts @@ -0,0 +1,245 @@ +const node = (script: string): subprocess.Command => [process.execPath, "-e", script]; +/** like `node`, for a script that imports the library */ +const module = (script: string): subprocess.Command => [ + process.execPath, + "--no-warnings", + "--input-type=module", + "-e", + script, +]; +const lib = (file: string) => JSON.stringify(new URL(file, import.meta.url).href); + +const messages: log.Message[] = []; +log.replaceGlobalMessageDestination((message) => messages.push(message)); + +async function until(condition: () => unknown) { + while (!condition()) await async.delay(5); +} + +describe("execute", () => { + test("returns stdout", async () => { + assert.equal(await subprocess.execute(node(`console.log("hello")`)), "hello\n"); + }); + + test("does not stall on output larger than a pipe buffer", async () => { + const script = `process.stderr.write("e".repeat(300000)); process.stdout.write("o".repeat(300000))`; + assert.equal((await subprocess.execute(node(script))).length, 300000); + }); + + test("a failure carries the exit code, the command, and stderr", async () => { + const cmd = node(`console.error("it broke"); process.exit(3)`); + const err = await subprocess.execute(cmd).catch((err) => err); + + assert.ok(err instanceof subprocess.ExitError); + assert.equal(err.message, `${path.basename(process.execPath)} failed with code 3`); + assert.equal(err.code, 3); + assert.equal(err.signal, null); + assert.equal(err.stderr, "it broke\n"); + assert.equal(err.cmd, `${process.execPath} -e 'console.error("it broke"); process.exit(3)'`); + }); + + test("a missing program rejects with ENOENT", async () => { + const err = await subprocess.execute(["clover-lib-missing-program"]).catch((err) => err); + assert.equal(exception.code(err), "ENOENT"); + }); + + test("env is added to the environment, and cwd is used", async () => { + const script = `console.log(process.env.CLOVER_TEST, !!process.env.PATH, process.cwd())`; + const cwd = fs.realpathSync(os.tmpdir()); + const out = await subprocess.execute(node(script), { env: { CLOVER_TEST: "yes" }, cwd }); + assert.equal(out, `yes true ${cwd}\n`); + }); +}); + +describe("spawn", () => { + test("wait resolves on exit code zero", async () => { + await subprocess.spawn(node("")).wait(); + }); + + test("a failure nobody waits for is not an unhandled rejection", async () => { + const proc = subprocess.spawn(node(`process.exit(1)`)); + await async.delay(300); + await assert.rejects(proc.wait(), { message: /failed with code 1$/ }); + }); + + test("stdin and stdout are web streams when piped", async () => { + const proc = subprocess.spawn(node(`process.stdin.pipe(process.stdout)`), { + stdin: "pipe", + stdout: "pipe", + }); + const writer = proc.stdin.getWriter(); + await writer.write(string.encodeUtf8("round trip")); + await writer.close(); + + assert.equal(await new Response(proc.stdout).text(), "round trip"); + assert.equal(proc.stderr, null); + await proc.wait(); + }); + + test("a status that redraws itself is read before its next line is written", async () => { + const script = ` + process.stderr.write("frame=1\\r"); + process.stdin.resume().on("end", () => process.stderr.write("frame=2\\rdone\\n")); + `; + const proc = subprocess.spawn(node(script), { stdin: "pipe", stderr: "pipe" }); + const seen: string[] = []; + for await (const line of stream.lines(proc.stderr)) { + seen.push(line); + if (line === "frame=1") await proc.stdin.close(); + } + await proc.wait(); + + assert.deepEqual(seen, ["frame=1", "frame=2", "done"]); + }); + + test("kill is reported as the signal", async () => { + const proc = subprocess.spawn(node(`setInterval(() => {}, 1000)`)); + proc.kill("SIGKILL"); + const err = await proc.wait().catch((err) => err); + + assert.ok(err instanceof subprocess.ExitError); + assert.equal(err.signal, "SIGKILL"); + assert.equal(err.code, null); + assert.match(err.message, /failed with signal SIGKILL$/); + }); + + test("aborting kills the process and rejects with the reason", async () => { + const ctrl = new AbortController(); + const reason = new Error("rescheduled"); + const proc = subprocess.spawn(node(`setInterval(() => {}, 1000)`), { signal: ctrl.signal }); + ctrl.abort(reason); + + assert.equal(await proc.wait().catch((err) => err), reason); + assert.throws(() => subprocess.spawn(node(""), { signal: ctrl.signal }), reason); + await assert.rejects(subprocess.execute(node(""), { signal: ctrl.signal }), reason); + }); + + test("disposing kills a running process, without having read its pipe", async () => { + let escaped: subprocess.Subprocess; + { + await using proc = subprocess.spawn( + node(`process.stdout.write("x".repeat(300000)); setInterval(() => {}, 1000)`), + { stdout: "pipe" }, + ); + escaped = proc; + } + await assert.rejects(escaped.wait(), { signal: "SIGTERM" }); + }); +}); + +describe("ipc", () => { + test("messages go both ways, and the end of the channel is heard", async () => { + const proc = subprocess.spawn(node(`process.on("message", (m) => process.send({ echo: m }))`), { + ipc: true, + }); + const reply = new Promise((resolve) => proc.on("message", resolve)); + const disconnect = new Promise((resolve) => proc.on("disconnect", resolve)); + proc.send("hello"); + + assert.deepEqual(await reply, { echo: "hello" }); + proc.kill(); + await disconnect; + const error = await new Promise((resolve) => proc.send("too late", resolve)); + assert.equal(exception.code(error), "ERR_IPC_CHANNEL_CLOSED"); + }); + + test("send needs the option", async () => { + await using proc = subprocess.spawn(node("")); + assert.throws(() => proc.send("hello"), /ipc/); + }); + + test("a prompt here can evaluate in the child", async () => { + const script = ` + import * as repl from ${lib("./repl.ts")}; + repl.serve(repl.javascript({ globals: { answer: 42 } }), process); + `; + await using proc = subprocess.spawn(module(script), { ipc: true }); + const printed: string[] = []; + await repl.remote(proc).eval("answer + 1", (value) => printed.push(String(value))); + + assert.deepEqual(printed.map(ansi.strip), ["43"]); + }); +}); + +describe("progress", () => { + const script = ` + import * as log from ${lib("./log.ts")}; + import * as progress from ${lib("./progress.ts")}; + log.info("from the child"); + const task = progress.start("child task", { total: 2 }); + task.log.warn("on the task"); + task.value = 1; + process.on("message", () => { + progress.start("started last").log.error("on a node at exit"); + log.error("last words"); + process.exit(1); + }); + `; + + test("the child starts on the node it was handed, and logs through this process", async () => { + const root = new progress.Root(); + const owner = root.start("owner"); + const proc = subprocess.spawn(module(script), { progress: owner, ipc: true }); + const children = () => UNWRAP(root.active[0]).children; + await until(() => children()[0]?.value === 1); + + const task = UNWRAP(children()[0]); + assert.deepEqual([task.text, task.total], ["child task", 2]); + assert.deepEqual(task.logs.map((m) => [m.level, m.text]), [["warn", "on the task"]]); + assert.deepEqual(messages.map((m) => [m.level, m.text]), [["info", "from the child"]]); + + // what is said right before exiting is in by the time `wait` is over + const ended: string[] = []; + root.on("node-end", (node) => ended.push(`${node.text}: ${node.logs.map((m) => m.text)}`)); + proc.send("finish"); + await assert.rejects(proc.wait(), subprocess.ExitError); + + assert.deepEqual(messages.at(-1)?.text, "last words"); + assert.deepEqual(ended.sort(), ["child task: on the task", "started last: on a node at exit"]); + assert.deepEqual(children(), []); + }); + + test("only the child that was handed the node reports to it", async () => { + // the grandchild inherits the environment, but not the pipe + const between = ` + import * as child_process from "node:child_process"; + const args = ["--no-warnings", "--input-type=module", "-e", process.argv[1]]; + child_process.spawnSync(process.execPath, args, { stdio: "inherit" }); + `; + const grandchild = ` + import * as log from ${lib("./log.ts")}; + import * as progress from ${lib("./progress.ts")}; + progress.start("grandchild task"); + log.info("from the grandchild"); + `; + const root = new progress.Root(); + const started: string[] = []; + root.on("node-start", (node) => started.push(node.text)); + const proc = subprocess.spawn([...module(between), "--", grandchild], { + progress: root, + stdout: "pipe", + }); + const stdout = await new Response(proc.stdout).text(); + await proc.wait(); + + assert.match(stdout, /from the grandchild/); + assert.deepEqual(started, []); + }); +}); + +import { strict as assert } from "node:assert"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import * as process from "node:process"; +import { describe, test } from "vitest"; +import { UNWRAP } from "./assert.ts"; +import * as async from "./async.ts"; +import * as exception from "./exception.ts"; +import * as log from "./log.ts"; +import * as progress from "./progress.ts"; +import * as repl from "./repl.ts"; +import * as stream from "./stream.ts"; +import * as string from "./string.ts"; +import * as ansi from "./string/ansi.ts"; +import * as subprocess from "./subprocess.ts"; diff --git a/lib/subprocess.ts b/lib/subprocess.ts index 851a9733a294d82bd56b1b80e02cf30eec3794be..80b5e1b22af54c76c226c2dc3dd7963ec7a1e92c 100644 --- a/lib/subprocess.ts +++ b/lib/subprocess.ts @@ -1,95 +1,170 @@ /** - * This file is expecting a full rewrite to abstract the Node API away - * entirely, exposing web streams I/O. - * + * spawn and manage other programs as sub-processes. * @module */ -const execFileRaw: typeof child_process.execFile.__promisify__ = util.promisify( - child_process.execFile, -); -/*** @deprecated */ -export const exec: typeof execFileRaw = (( - ...args: Parameters -) => - execFileRaw(...args).catch((e: any) => { - if (e?.message?.startsWith?.("Command failed")) { - if (e.code > 2 ** 31) e.code |= 0; - const code = e.signal ? `signal ${e.signal}` : `code ${e.code}`; - e.message = `${e.cmd.split(" ")[0]} failed with ${code}`; - } - throw e; - })) as any; - -// const activePids = new Set(); - -/*** @deprecated */ -export interface SpawnOptions extends child_process.SpawnOptions { - cmd: string[]; -} -/*** @deprecated */ -export function spawn( - { cmd, ...rest }: SpawnOptions, -): child_process.ChildProcess { - // rest.detached ??= true; - const process = child_process.spawn(cmd[0]!, cmd.slice(1), rest); - // const pid = rest.detached && process.pid ? -process.pid : process.pid; - // if (pid) { - // activePids.add(pid); - // process.on("exit", () => void activePids.delete(pid)); - // } - return process; -} - -/*** @deprecated */ -export interface ForkOptions extends child_process.ForkOptions { - cmd: string[]; -} - -/*** @deprecated */ -export function fork( - { cmd, ...rest }: ForkOptions, -): child_process.ChildProcess { - // rest.detached ??= true; - const process = child_process.fork(cmd[0]!, cmd.slice(1), rest); - // const pid = rest.detached && process.pid ? -process.pid : process.pid; - // if (pid) { - // activePids.add(pid); - // process.on("exit", () => void activePids.delete(pid)); - // } - return process; -} - -/*** @deprecated */ -export async function spawnAndWait( - opts: SpawnOptions, -): Promise<{ stdout: Buffer | null; stderr: Buffer | null }> { - opts.stdio ??= ["ignore", "inherit", "inherit"]; - const process = spawn(opts); - const output = Promise.all([ - process.stdout ? Array.fromAsync(process.stdout).then(Buffer.concat) : null, - process.stderr ? Array.fromAsync(process.stderr).then(Buffer.concat) : null, +/** + * execute a subprocess and wait for it to exit, returning what it wrote to + * stdout. stderr is captured too, shown via {@linkcode ExitError} on failure. + * this is, essentially, a simpler version of {@linkcode spawn}. + */ +export async function execute( + cmd: Command, + opts?: Omit, +): Promise { + await using proc = spawn(cmd, { ...opts, stdout: "pipe", stderr: "pipe" }); + const [stdout, stderr] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), ]); - let [code, signal] = (await events.EventEmitter.once(process, "exit")) as [ - number | null, - string | null, - ]; - if (code !== 0) { + + try { + await proc.wait(); + } catch (err) { + if (err instanceof ExitError) err.stderr = stderr; + throw err; + } + return stdout; +} + +/** + * start a command and return at once. how it ends, which includes not being + * able to start, is found by {@linkcode Subprocess.wait}. + */ +export function spawn(cmd: Command, opts?: O): Subprocess { + const { stdin = "ignore", stdout = "inherit", stderr = "inherit" }: Options = opts ?? {}; + const { cwd, env, signal, ipc, progress: handoff }: Options = opts ?? {}; + signal?.throwIfAborted(); + + const stdio: Array = [stdin, stdout, stderr]; + if (handoff) stdio.push("pipe"); + if (ipc) stdio.push("ipc"); + + const child = child_process.spawn(cmd[0], cmd.slice(1), { + cwd, + env: { ...process.env, ...env, ...handoff && progress.childEnv(progressFd) }, + stdio, + }); + + const onAbort = () => void child.kill(); + signal?.addEventListener("abort", onAbort); + + const report = handoff + && progress.attachChild(toWeb(child.stdio[progressFd] as stream.Readable), handoff); + + const exit = new Promise((resolve, reject) => { + child.on("error", reject); + child.once("exit", (code, exitSignal) => { + if (signal?.aborted) reject(signal.reason); + else if (code === 0) resolve(); + else reject(new ExitError(cmd, code, exitSignal)); + }); + }) + // the last of what the child reported is in before `wait` is over + .finally(() => report) + .finally(() => signal?.removeEventListener("abort", onAbort)); + + // a failure is only an error to those who `wait` for it + exit.catch(() => {}); + report?.catch(() => {}); + + const proc: Subprocess = { + stdin: child.stdin && stream.Writable.toWeb(child.stdin), + stdout: child.stdout && toWeb(child.stdout), + stderr: child.stderr && toWeb(child.stderr), + wait: () => exit, + kill: (killSignal) => void child.kill(killSignal), + send(message, callback) { + ASSERT(ipc, "`send` is for a process started with the `ipc` option"); + child.send(message as child_process.Serializable, callback); + }, + on: (event: string, listener: (message: unknown) => void) => void child.on(event, listener), + off: (event: string, listener: (message: unknown) => void) => void child.off(event, listener), + async [Symbol.asyncDispose]() { + child.kill(); + await exit.catch(() => {}); + // a pipe that was not read to its end holds a file descriptor + for (const pipe of child.stdio) pipe?.destroy(); + }, + }; + return proc as Subprocess; +} + +/** where a child has the pipe for its `progress`: after stdin, stdout, and stderr */ +const progressFd = 3; + +/** node.js types its web streams apart from the global ones */ +function toWeb(pipe: stream.Readable) { + return stream.Readable.toWeb(pipe) as ReadableStream; +} + +/** a program followed by its arguments */ +export type Command = readonly [string, ...string[]]; + +/** `"pipe"` puts a stream on the {@linkcode Subprocess} */ +export type Stdio = "pipe" | "inherit" | "ignore"; + +export interface Options { + cwd?: string; + /** added to the environment of this process */ + env?: Record; + /** kills the process when aborted, and `wait` rejects with its reason */ + signal?: AbortSignal; + /** @default "ignore" */ + stdin?: Stdio; + /** @default "inherit" */ + stdout?: Stdio; + /** @default "inherit" */ + stderr?: Stdio; + /** open a channel for messages with a child that is node.js too, where it is `process` */ + ipc?: boolean; + /** + * hand over a place in the progress of this process. a child using + * `@clo/lib/progress` has it as its global root, and logs through this + * process instead of writing to the terminal by itself. + */ + progress?: progress.Ref; +} + +/** disposing kills the process if it is still running, and waits for it to exit */ +export interface Subprocess extends AsyncDisposable { + stdin: Piped>; + stdout: Piped>; + stderr: Piped>; + /** resolves once the process exits with code zero. any other exit rejects with an {@linkcode ExitError} */ + wait(): Promise; + kill(signal?: NodeJS.Signals): void; + + /** send a message over the `ipc` channel. `callback` hears if it could not be */ + send(message: unknown, callback?: (error: Error | null) => void): void; + on(event: "message", listener: (message: unknown) => void): void; + on(event: "disconnect", listener: () => void): void; + off(event: "message", listener: (message: unknown) => void): void; + off(event: "disconnect", listener: () => void): void; +} + +type Piped = S extends "pipe" ? T : null; + +/** how a process ended, when that was not with exit code zero */ +export class ExitError extends Error { + /** as it would be typed into a shell */ + cmd: string; + code: number | null; + signal: NodeJS.Signals | null; + /** only when {@linkcode execute} ran the command */ + declare stderr?: string; + + constructor(cmd: Command, code: number | null, signal: NodeJS.Signals | null) { + // windows prints exit codes signed, such as -1073741819 for an access violation if (code !== null && code > 2 ** 31) code |= 0; - const exit = signal ? `signal ${signal}` : `code ${code}`; - throw new Error(`${opts.cmd[0]} failed with ${exit}`); + super(`${path.basename(cmd[0])} failed with ${signal ? `signal ${signal}` : `code ${code}`}`); + + this.cmd = cmd.map(string.escapeShellArgument).join(" "); + this.code = code; + this.signal = signal; } - const [stdout, stderr] = await output; - return { stdout, stderr }; } -// process.on("exit", () => { -// for (const pid of activePids) { -// console.info(`Kill ${Math.abs(pid)}`); -// process.kill(pid, 9); -// } -// }); - async function findInodeForPort(port: number) { const hexPort = port.toString(16).toUpperCase().padStart(4, "0"); for (const f of ["/proc/net/tcp", "/proc/net/tcp6"]) { @@ -129,22 +204,23 @@ async function commandFromPortLinux(port: number) { } async function commandFromPortMac(port: number) { - const { stdout: lsofOut } = await exec("lsof", ["-ti", `tcp:${port}`]); + const lsofOut = await execute(["lsof", "-ti", `tcp:${port}`]); const pid = lsofOut.trim().split("\n")[0]; if (!pid) throw new Error(`no process on port ${port}`); - const { stdout: psOut } = await exec("ps", ["-p", pid, "-o", "args="]); + const psOut = await execute(["ps", "-p", pid, "-o", "args="]); return { pid: Number(pid), command: psOut.trim() }; } async function commandFromPortWindows(port: number) { - const { stdout } = await exec("netstat", ["-ano"]); + const stdout = await execute(["netstat", "-ano"]); const line = stdout.split("\n").find(l => { const m = l.match(/\s+[\d.:]+:(\d+)\s+/); return m && parseInt(m[1]!) === port && l.includes("LISTENING"); }); if (!line) throw new Error(`no process on port ${port}`); const pid = line.trim().split(/\s+/).at(-1); - const { stdout: psOut } = await exec("powershell", [ + const psOut = await execute([ + "powershell", "-NoProfile", "-Command", `(Get-CimInstance Win32_Process -Filter "ProcessId=${pid}").CommandLine`, @@ -155,7 +231,7 @@ async function commandFromPortWindows(port: number) { } /** Racily searches for what process is taking up a given port. */ -export async function lookupProcessOnPort(port: number): Promise { +export function lookupProcessOnPort(port: number): Promise { switch (os.platform()) { case "linux": return commandFromPortLinux(port); @@ -173,9 +249,12 @@ interface ProcessLookup { command: string; } -import { Buffer } from "node:buffer"; import * as child_process from "node:child_process"; -import * as events from "node:events"; import * as fs from "node:fs/promises"; import * as os from "node:os"; -import * as util from "node:util"; +import * as path from "node:path"; +import * as process from "node:process"; +import * as stream from "node:stream"; +import { ASSERT } from "./assert.ts"; +import * as progress from "./progress.ts"; +import * as string from "./string.ts"; diff --git a/lib/subprocess/ffmpeg.ts b/lib/subprocess/ffmpeg.ts index b672f57051db20bd4c520f05bdc751419ace3d1c..76ff6c5ba56871a516aa5ef8501196ee32f7d0ee 100644 --- a/lib/subprocess/ffmpeg.ts +++ b/lib/subprocess/ffmpeg.ts @@ -32,23 +32,21 @@ export async function spawn(options: SpawnOptions) { const { ffmpeg = "ffmpeg", args, cwd } = options; using node = options.progress; const title = node.text; - const proc = child_process.spawn(ffmpeg, [...defaultExtraOptions, ...args], { - stdio: ["ignore", "inherit", "pipe"], - env: { ...process.env, SVT_LOG: "2" }, - cwd: cwd?.toString(), + await using proc = subprocess.spawn([ffmpeg, ...defaultExtraOptions, ...args], { + stderr: "pipe", + env: { SVT_LOG: "2" }, + cwd, signal: options.signal, }); + const parser = new Parse(); - let running = true; - const splitter = readline.createInterface({ input: proc.stderr }); - splitter.on("line", (line) => { + for await (const line of stream.lines(proc.stderr)) { const result = parser.onLine(line); if (result.kind === "ignore") { - return; + continue; } else if (result.kind === "log") { node.log[result.level](result.message); } else if (result.kind === "progress") { - if (!running) return; const { frame, totalFrames, fps, speed } = result; node.value = frame; node.total = totalFrames; @@ -59,17 +57,8 @@ export async function spawn(options: SpawnOptions) { ].filter(Boolean).join(", "); node.text = `${title}${extras.length > 0 ? ` (${extras})` : ""}`; } else result satisfies never; - }); - const [code, signal] = await events.once(proc, "close"); - running = false; - if (code !== 0) { - const fmt = code ? `code ${code}` : `signal ${signal}`; - const e: any = new Error(`ffmpeg failed with ${fmt}`); - e.args = [ffmpeg, ...args].map(string.escapeShellArgument).join(" "); - e.code = code; - e.signal = signal; - throw e; } + await proc.wait(); } export interface Streams { @@ -78,21 +67,18 @@ export interface Streams { } export async function probeStreams(path: string): Promise { - const { stdout } = await subprocess.spawnAndWait({ - cmd: [ - "ffprobe", - "-v", - "error", - "-show_entries", - "stream=codec_type", - path.toString(), - "-of", - "default=noprint_wrappers=1:nokey=1", - ], - stdio: "pipe", - }); + const stdout = await subprocess.execute([ + "ffprobe", + "-v", + "error", + "-show_entries", + "stream=codec_type", + path.toString(), + "-of", + "default=noprint_wrappers=1:nokey=1", + ]); - const lines = new TextDecoder().decode(stdout!).trim().split("\n"); + const lines = stdout.trim().split("\n"); return { audio: lines.includes("audio"), video: lines.includes("video"), @@ -201,11 +187,7 @@ export class Parse { } } -import * as child_process from "node:child_process"; -import events from "node:events"; import * as path from "node:path"; -import * as process from "node:process"; -import * as readline from "node:readline"; import type * as progress from "../progress.ts"; -import * as string from "../string.ts"; +import * as stream from "../stream.ts"; import * as subprocess from "../subprocess.ts"; diff --git a/src/bin/deploy.ts b/src/bin/deploy.ts index 56d46864876c21d010f1acce4fc29c7005721725..8059dfa347e5e1729775ea3d6a37db4a6b253856 100644 --- a/src/bin/deploy.ts +++ b/src/bin/deploy.ts @@ -41,16 +41,13 @@ export async function main() { ], progress: progress.start("copy site files"), }); - await subprocess.spawnAndWait({ - stdio: "inherit", - cmd: [ - "ssh", - "paperclover.net", - "cd paperclover.net && " - + "npm i &&" - + "pm2 restart site", - ], - }); + await subprocess.spawn([ + "ssh", + "paperclover.net", + "cd paperclover.net && " + + "npm i &&" + + "pm2 restart site", + ], { stdin: "inherit" }).wait(); } import { Path } from "#sitegen/path"; diff --git a/src/file-viewer/indexer/processors/media.ts b/src/file-viewer/indexer/processors/media.ts index ef5f78b97b8a7eb8a272fb5be6c9d5cdfd6cf647..cf2498a361e7e999a50bde4f25faf0a904b67da8 100644 --- a/src/file-viewer/indexer/processors/media.ts +++ b/src/file-viewer/indexer/processors/media.ts @@ -14,7 +14,8 @@ const procDuration: Processor = { include: rules.extsDuration, cores: 1, async run({ path, mediaFile, signal }) { - const { stdout } = await subprocess.exec(ffprobeBin!, [ + const stdout = await subprocess.execute([ + ffprobeBin!, "-v", "error", "-show_entries", @@ -55,7 +56,8 @@ const procDimensions: Processor = { } } else if (rules.extsImage.has(ext)) { // Use magick to observe streams - const { stdout } = await subprocess.exec("magick", [ + const stdout = await subprocess.execute([ + "magick", "identify", "-auto-orient", "-format", @@ -68,7 +70,8 @@ const procDimensions: Processor = { } } else { // Use ffprobe to observe streams - const { stdout } = await subprocess.exec("ffprobe", [ + const stdout = await subprocess.execute([ + "ffprobe", "-v", "error", "-select_streams", @@ -127,7 +130,8 @@ const procImageSubsets: Processor = { cores: 2, subkey: `${size}${ext}`, async producer(dir) { - await subprocess.exec(ffmpegBin!, [ + await subprocess.execute([ + ffmpegBin!, ...ffmpegOptions, "-i", path.toString(), diff --git a/src/file-viewer/indexer/scrub.ts b/src/file-viewer/indexer/scrub.ts index b5f823e17ff0becd7d46167431ca144fed563404..c8177eb9022f35b71809712aeb654b737b0c4160 100644 --- a/src/file-viewer/indexer/scrub.ts +++ b/src/file-viewer/indexer/scrub.ts @@ -4,10 +4,6 @@ // first); modifies the file in place while preserving timestamps. const exiftoolBin = testProgram("exiftool"); -// `-ee` on an iphone video dumps per-frame embedded metadata, which easily -// exceeds execFile's default 1MB stdout buffer -const execOptions = { maxBuffer: 64 * 1024 * 1024 }; - export async function scrubLocationMetadata( path: Path, stats: fs.Stats, @@ -30,21 +26,23 @@ export async function scrubLocationMetadata( case ".jpg": case ".jpeg": case ".png": - const { stdout: gpsCheck } = await subprocess.exec("exiftool", [ + const gpsCheck = await subprocess.execute([ + "exiftool", "-gps:all", path.toString(), - ], execOptions); + ]); hasLocation = gpsCheck.trim().length > 0; args = ["-gps:all=", path.toString(), "-o", tempOutput.toString()]; break; case ".mov": case ".mp4": - const { stdout: videoCheck } = await subprocess.exec("exiftool", [ + const videoCheck = await subprocess.execute([ + "exiftool", "-ee", "-G3", "-s", path.toString(), - ], execOptions); + ]); hasLocation = videoCheck.includes("GPS") || videoCheck.includes("Location"); args = [ @@ -56,12 +54,13 @@ export async function scrubLocationMetadata( ]; break; case ".m4a": - const { stdout: m4aCheck } = await subprocess.exec("exiftool", [ + const m4aCheck = await subprocess.execute([ + "exiftool", "-ee", "-G3", "-s", path.toString(), - ], execOptions); + ]); hasLocation = m4aCheck.includes("GPS") || m4aCheck.includes("Location") || m4aCheck.includes("Filename") @@ -102,7 +101,7 @@ export async function scrubLocationMetadata( await tempOutput.delete({ force: true }); // Remove metadata - await subprocess.exec("exiftool", args, execOptions); + await subprocess.execute(["exiftool", ...args]); if (!tempOutput.ifExistsSync()) { throw new Error(`Failed to create output file: ${tempOutput}`); } -- 2.54.0