1/**
2 * spawn and manage other programs as sub-processes.
3 * @module
4 */
5
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;
28}
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.ipcChildEnvironment(progressFd) },
46 stdio,
47 });
48
49 const onAbort = () => void child.kill();
50 signal?.addEventListener("abort", onAbort);
51
52 const report = handoff
53 && progress.decodeAttachToRef(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>;
91}
92
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>;
99}
100
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 /** for a subprocess using `@clo/lib/process`, setup progress IPC. */
122 progress?: progress.Ref;
123}
124
125/** disposing kills the process if it is still running, and waits for it to exit */
126export interface Subprocess<O extends Options = Options> extends AsyncDisposable {
127 stdin: Piped<O["stdin"], WritableStream<Uint8Array>>;
128 stdout: Piped<O["stdout"], ReadableStream<Uint8Array>>;
129 stderr: Piped<O["stderr"], ReadableStream<Uint8Array>>;
130 /** resolves once the process exits with code zero. any other exit rejects with an {@linkcode ExitError} */
131 wait(): Promise<void>;
132 kill(signal?: NodeJS.Signals): void;
133
134 /** send a message over the `ipc` channel. `callback` hears if it could not be */
135 send(message: unknown, callback?: (error: Error | null) => void): void;
136 on(event: "message", listener: (message: unknown) => void): void;
137 on(event: "disconnect", listener: () => void): void;
138 off(event: "message", listener: (message: unknown) => void): void;
139 off(event: "disconnect", listener: () => void): void;
140}
141
142type Piped<S, T> = S extends "pipe" ? T : null;
143
144/** how a process ended, when that was not with exit code zero */
145export class ExitError extends Error {
146 /** as it would be typed into a shell */
147 cmd: string;
148 code: number | null;
149 signal: NodeJS.Signals | null;
150 /** only when {@linkcode execute} ran the command */
151 declare stderr?: string;
152
153 constructor(cmd: Command, code: number | null, signal: NodeJS.Signals | null) {
154 // windows prints exit codes signed, such as -1073741819 for an access violation
155 if (code !== null && code > 2 ** 31) code |= 0;
156 super(`${path.basename(cmd[0])} failed with ${signal ? `signal ${signal}` : `code ${code}`}`);
157
158 this.cmd = cmd.map(string.escapeShellArgument).join(" ");
159 this.code = code;
160 this.signal = signal;
161 }
162}
163
164async function findInodeForPort(port: number) {
165 const hexPort = port.toString(16).toUpperCase().padStart(4, "0");
166 for (const f of ["/proc/net/tcp", "/proc/net/tcp6"]) {
167 try {
168 const lines = (await fs.readFile(f, "utf8")).split("\n").slice(1);
169 for (const line of lines) {
170 const parts = line.trim().split(/\s+/);
171 if (parts[1]?.split(":")[1] === hexPort) return parts[9];
172 }
173 } catch {}
174 }
175 return null;
176}
177
178async function findPidForInode(inode: string) {
179 const target = `socket:[${inode}]`;
180 for (const pid of await fs.readdir("/proc")) {
181 if (!/^\d+$/.test(pid)) continue;
182 try {
183 for (const fd of await fs.readdir(`/proc/${pid}/fd`)) {
184 try {
185 if (await fs.readlink(`/proc/${pid}/fd/${fd}`) === target) return pid;
186 } catch {}
187 }
188 } catch {}
189 }
190 return null;
191}
192
193async function commandFromPortLinux(port: number) {
194 const inode = await findInodeForPort(port);
195 if (!inode) throw new Error(`no process on port ${port}`);
196 const pid = await findPidForInode(inode);
197 if (!pid) throw new Error(`no pid for inode ${inode}`);
198 const raw = await fs.readFile(`/proc/${pid}/cmdline`, "utf8");
199 return { pid: Number(pid), command: raw.split("\0").filter(Boolean).join(" ") };
200}
201
202async function commandFromPortMac(port: number) {
203 const lsofOut = await execute(["lsof", "-ti", `tcp:${port}`]);
204 const pid = lsofOut.trim().split("\n")[0];
205 if (!pid) throw new Error(`no process on port ${port}`);
206 const psOut = await execute(["ps", "-p", pid, "-o", "args="]);
207 return { pid: Number(pid), command: psOut.trim() };
208}
209
210async function commandFromPortWindows(port: number) {
211 const stdout = await execute(["netstat", "-ano"]);
212 const line = stdout.split("\n").find(l => {
213 const m = l.match(/\s+[\d.:]+:(\d+)\s+/);
214 return m && parseInt(m[1]!) === port && l.includes("LISTENING");
215 });
216 if (!line) throw new Error(`no process on port ${port}`);
217 const pid = line.trim().split(/\s+/).at(-1);
218 const psOut = await execute([
219 "powershell",
220 "-NoProfile",
221 "-Command",
222 `(Get-CimInstance Win32_Process -Filter "ProcessId=${pid}").CommandLine`,
223 ]);
224 const cmdline = psOut.trim();
225 if (!cmdline) throw new Error(`no cmdline for pid ${pid}`);
226 return { pid: Number(pid), command: cmdline };
227}
228
229/** Racily searches for what process is taking up a given port. */
230export function lookupProcessOnPort(port: number): Promise<ProcessLookup> {
231 switch (os.platform()) {
232 case "linux":
233 return commandFromPortLinux(port);
234 case "darwin":
235 return commandFromPortMac(port);
236 case "win32":
237 return commandFromPortWindows(port);
238 default:
239 throw new Error(`unsupported platform: ${os.platform()}`);
240 }
241}
242
243interface ProcessLookup {
244 pid: number;
245 command: string;
246}
247
248import * as child_process from "node:child_process";
249import * as fs from "node:fs/promises";
250import * as os from "node:os";
251import * as path from "node:path";
252import * as process from "node:process";
253import * as stream from "node:stream";
254import { ASSERT } from "./assert.ts";
255import * as progress from "./progress.ts";
256import * as string from "./string.ts";