Open History to see the last commit
1import * as Context from "effect/Context";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Option from "effect/Option";
6import * as PlatformError from "effect/PlatformError";
7import * as Schema from "effect/Schema";
8import * as Scope from "effect/Scope";
9import * as Stream from "effect/Stream";
10import * as ChildProcess from "effect/process/ChildProcess";
11import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
12import { resolveSpawnCommand } from "@t3tools/shared/shell";
13import {
14 collectUint8StreamText,
15 decodeUtf8,
16 type CollectedUint8StreamText,
17} from "@t3tools/provider-core/server/collectStreamText";
18
19export interface ProcessRunInput {
20 readonly command: string;
21 readonly args: ReadonlyArray<string>;
22 readonly cwd?: string | undefined;
23 readonly spawnCwd?: string | undefined;
24 readonly timeout?: Duration.Input | undefined;
25 readonly env?: NodeJS.ProcessEnv | undefined;
26 readonly stdin?: string | undefined;
27 /** Receives every stdout chunk, including bytes beyond the buffered output limit. */
28 readonly onStdoutChunk?: ((chunk: Uint8Array) => void) | undefined;
29 readonly maxOutputBytes?: number | undefined;
30 readonly outputMode?: "error" | "truncate" | undefined;
31 readonly truncatedMarker?: string | undefined;
32 /** Timed-out results discard partial stdout/stderr. */
33 readonly timeoutBehavior?: "error" | "timedOutResult" | undefined;
34}
35
36export interface ProcessRunOutput {
37 readonly stdout: string;
38 readonly stderr: string;
39 readonly code: ChildProcessSpawner.ExitCode | null;
40 readonly timedOut: boolean;
41 readonly stdoutTruncated: boolean;
42 readonly stderrTruncated: boolean;
43 readonly stdoutInvalidUtf8: boolean;
44 readonly stderrInvalidUtf8: boolean;
45}
46
47const ProcessInvocationFields = {
48 command: Schema.String,
49 argumentCount: Schema.Number,
50 cwd: Schema.optional(Schema.String),
51 spawnCwd: Schema.optional(Schema.String),
52};
53
54const formatProcessInvocation = (input: {
55 readonly command: string;
56 readonly cwd?: string | undefined;
57 readonly spawnCwd?: string | undefined;
58}): string => {
59 const executionCwd = input.spawnCwd ?? input.cwd;
60 return executionCwd === undefined
61 ? `'${input.command}'`
62 : `'${input.command}' in '${executionCwd}'`;
63};
64
65export class ProcessSpawnError extends Schema.TaggedError<ProcessSpawnError>()(
66 "ProcessSpawnError",
67 {
68 ...ProcessInvocationFields,
69 resolvedCommand: Schema.optional(Schema.String),
70 resolvedArgumentCount: Schema.optional(Schema.Number),
71 shell: Schema.optional(Schema.Boolean),
72 cause: Schema.Defect(),
73 },
74) {
75 override get message(): string {
76 return `Failed to spawn process ${formatProcessInvocation(this)}`;
77 }
78}
79
80export class ProcessStdinError extends Schema.TaggedError<ProcessStdinError>()(
81 "ProcessStdinError",
82 {
83 ...ProcessInvocationFields,
84 stdinBytes: Schema.Number,
85 cause: Schema.Defect(),
86 },
87) {
88 override get message(): string {
89 return `Failed to write stdin for process ${formatProcessInvocation(this)}`;
90 }
91}
92
93export class ProcessOutputLimitError extends Schema.TaggedError<ProcessOutputLimitError>()(
94 "ProcessOutputLimitError",
95 {
96 ...ProcessInvocationFields,
97 stream: Schema.Literals(["stdout", "stderr"]),
98 maxBytes: Schema.Number,
99 observedBytes: Schema.Number,
100 },
101) {
102 override get message(): string {
103 return `Process ${formatProcessInvocation(this)} ${this.stream} produced ${this.observedBytes} bytes, exceeding the ${this.maxBytes} byte limit`;
104 }
105}
106
107export class ProcessReadError extends Schema.TaggedError<ProcessReadError>()("ProcessReadError", {
108 ...ProcessInvocationFields,
109 stream: Schema.Literals(["stdout", "stderr", "exitCode"]),
110 cause: Schema.Defect(),
111}) {
112 override get message(): string {
113 return `Failed to read ${this.stream} for process ${formatProcessInvocation(this)}`;
114 }
115}
116
117export class ProcessTimeoutError extends Schema.TaggedError<ProcessTimeoutError>()(
118 "ProcessTimeoutError",
119 {
120 ...ProcessInvocationFields,
121 timeoutMs: Schema.Number,
122 },
123) {
124 override get message(): string {
125 return `Process ${formatProcessInvocation(this)} timed out after ${this.timeoutMs}ms`;
126 }
127}
128
129export const ProcessRunError = Schema.Union([
130 ProcessSpawnError,
131 ProcessStdinError,
132 ProcessOutputLimitError,
133 ProcessReadError,
134 ProcessTimeoutError,
135]);
136export type ProcessRunError = typeof ProcessRunError.Type;
137
138export class ProcessRunner extends Context.Service<
139 ProcessRunner,
140 {
141 readonly run: (input: ProcessRunInput) => Effect.Effect<ProcessRunOutput, ProcessRunError>;
142 }
143>()("t3/processRunner") {}
144
145const DEFAULT_TIMEOUT = "60 seconds";
146// Keep SIGTERM-ignoring helpers from blocking timeouts or graceful shutdown.
147const FORCE_KILL_AFTER = "1 second";
148const DEFAULT_MAX_OUTPUT_BYTES = 8 * 1024 * 1024;
149
150const collectText = Effect.fnUntraced(function* (input: {
151 readonly command: string;
152 readonly args: ReadonlyArray<string>;
153 readonly cwd?: string | undefined;
154 readonly spawnCwd?: string | undefined;
155 readonly streamName: "stdout" | "stderr";
156 readonly stream: Stream.Stream<Uint8Array, PlatformError.PlatformError>;
157 readonly maxOutputBytes: number;
158 readonly outputMode: "error" | "truncate";
159 readonly truncatedMarker: string;
160}) {
161 const stream = input.stream.pipe(
162 Stream.mapError(
163 (cause) =>
164 new ProcessReadError({
165 command: input.command,
166 argumentCount: input.args.length,
167 cwd: input.cwd,
168 spawnCwd: input.spawnCwd,
169 stream: input.streamName,
170 cause,
171 }),
172 ),
173 );
174
175 if (input.outputMode === "truncate") {
176 return yield* collectUint8StreamText({
177 stream,
178 maxBytes: input.maxOutputBytes,
179 truncatedMarker: input.truncatedMarker,
180 });
181 }
182
183 return yield* stream.pipe(
184 Stream.runFoldEffect<
185 {
186 readonly chunks: Uint8Array<ArrayBufferLike>[];
187 readonly bytes: number;
188 },
189 Uint8Array<ArrayBufferLike>,
190 ProcessOutputLimitError | ProcessReadError,
191 never
192 >(
193 () => ({ chunks: [], bytes: 0 }),
194 (state, chunk) => {
195 const remainingBytes = input.maxOutputBytes - state.bytes;
196 if (chunk.byteLength > remainingBytes) {
197 return Effect.fail(
198 new ProcessOutputLimitError({
199 command: input.command,
200 argumentCount: input.args.length,
201 cwd: input.cwd,
202 spawnCwd: input.spawnCwd,
203 stream: input.streamName,
204 maxBytes: input.maxOutputBytes,
205 observedBytes: state.bytes + chunk.byteLength,
206 }),
207 );
208 }
209
210 state.chunks.push(chunk);
211 return Effect.succeed({
212 chunks: state.chunks,
213 bytes: state.bytes + chunk.byteLength,
214 });
215 },
216 ),
217 Effect.map((state): CollectedUint8StreamText => ({
218 ...decodeUtf8(Buffer.concat(state.chunks, state.bytes)),
219 bytes: state.bytes,
220 truncated: false,
221 })),
222 );
223});
224
225function finalizeRunProcess<R>(
226 effect: Effect.Effect<ProcessRunOutput, ProcessRunError, R | Scope.Scope>,
227 input: ProcessRunInput,
228): Effect.Effect<ProcessRunOutput, ProcessRunError, Exclude<R, Scope.Scope>> {
229 const timeout = Duration.fromInputUnsafe(input.timeout ?? DEFAULT_TIMEOUT);
230 const timeoutBehavior = input.timeoutBehavior ?? "error";
231
232 return effect.pipe(
233 Effect.scoped,
234 Effect.timeoutOption(timeout),
235 Effect.flatMap((result) => {
236 if (Option.isSome(result)) {
237 return Effect.succeed(result.value);
238 }
239 if (timeoutBehavior === "timedOutResult") {
240 return Effect.succeed({
241 stdout: "",
242 stderr: "",
243 code: null,
244 timedOut: true,
245 stdoutTruncated: false,
246 stderrTruncated: false,
247 stdoutInvalidUtf8: false,
248 stderrInvalidUtf8: false,
249 } satisfies ProcessRunOutput);
250 }
251 return Effect.fail(
252 new ProcessTimeoutError({
253 command: input.command,
254 argumentCount: input.args.length,
255 cwd: input.cwd,
256 spawnCwd: input.spawnCwd,
257 timeoutMs: Duration.toMillis(timeout),
258 }),
259 );
260 }),
261 );
262}
263
264const commandName = (command: string) => command.replace(/^.*[\\/]/, "");
265
266const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
267 spawner: ChildProcessSpawner.ChildProcessSpawner["Service"],
268 input: ProcessRunInput,
269): Effect.fn.Return<ProcessRunOutput, ProcessRunError, Scope.Scope> {
270 yield* Effect.annotateCurrentSpan("process.command", commandName(input.command));
271 const maxOutputBytes = input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES;
272 const outputMode = input.outputMode ?? "error";
273 const truncatedMarker = input.truncatedMarker ?? "";
274 const extendEnv = input.env !== undefined;
275 const spawnCommand = yield* resolveSpawnCommand(
276 input.command,
277 input.args,
278 input.env === undefined ? {} : { env: input.env, extendEnv },
279 );
280
281 const toSpawnError = (cause: unknown) =>
282 new ProcessSpawnError({
283 command: input.command,
284 argumentCount: input.args.length,
285 cwd: input.cwd,
286 spawnCwd: input.spawnCwd,
287 resolvedCommand: spawnCommand.command,
288 resolvedArgumentCount: spawnCommand.args.length,
289 shell: spawnCommand.shell,
290 cause,
291 });
292
293 const child = yield* spawner
294 .spawn(
295 ChildProcess.make(spawnCommand.command, spawnCommand.args, {
296 ...((input.spawnCwd ?? input.cwd) ? { cwd: input.spawnCwd ?? input.cwd } : {}),
297 ...(input.env !== undefined
298 ? {
299 env: input.env,
300 extendEnv,
301 }
302 : {}),
303 shell: spawnCommand.shell,
304 forceKillAfter: FORCE_KILL_AFTER,
305 }),
306 )
307 .pipe(
308 Effect.mapError(toSpawnError),
309 // Synchronous spawn throws arrive as defects; recover them too so callers see typed spawn failures.
310 Effect.catchDefect((cause) => Effect.fail(toSpawnError(cause))),
311 );
312
313 const stdin = input.stdin;
314 const onStdoutChunk = input.onStdoutChunk;
315 const writeStdin =
316 stdin === undefined
317 ? Effect.void
318 : Stream.run(Stream.encodeText(Stream.make(stdin)), child.stdin).pipe(
319 Effect.mapError(
320 (cause) =>
321 new ProcessStdinError({
322 command: input.command,
323 argumentCount: input.args.length,
324 cwd: input.cwd,
325 spawnCwd: input.spawnCwd,
326 stdinBytes: Buffer.byteLength(stdin),
327 cause,
328 }),
329 ),
330 );
331
332 const [stdout, stderr] = yield* Effect.all(
333 [
334 collectText({
335 command: input.command,
336 args: input.args,
337 cwd: input.cwd,
338 spawnCwd: input.spawnCwd,
339 streamName: "stdout",
340 stream: onStdoutChunk
341 ? child.stdout.pipe(Stream.tap((chunk) => Effect.sync(() => onStdoutChunk(chunk))))
342 : child.stdout,
343 maxOutputBytes,
344 outputMode,
345 truncatedMarker,
346 }),
347 collectText({
348 command: input.command,
349 args: input.args,
350 cwd: input.cwd,
351 spawnCwd: input.spawnCwd,
352 streamName: "stderr",
353 stream: child.stderr,
354 maxOutputBytes,
355 outputMode,
356 truncatedMarker,
357 }),
358 writeStdin,
359 ],
360 { concurrency: "unbounded" },
361 );
362
363 const exitCode = yield* child.exitCode.pipe(
364 Effect.mapError(
365 (cause) =>
366 new ProcessReadError({
367 command: input.command,
368 argumentCount: input.args.length,
369 cwd: input.cwd,
370 spawnCwd: input.spawnCwd,
371 stream: "exitCode",
372 cause,
373 }),
374 ),
375 );
376
377 return {
378 stdout: stdout.text,
379 stderr: stderr.text,
380 code: exitCode,
381 timedOut: false,
382 stdoutTruncated: stdout.truncated,
383 stderrTruncated: stderr.truncated,
384 stdoutInvalidUtf8: stdout.invalidUtf8,
385 stderrInvalidUtf8: stderr.invalidUtf8,
386 } satisfies ProcessRunOutput;
387});
388
389/** @public */
390export const make = Effect.fn("ProcessRunner.make")(function* () {
391 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
392
393 const run: ProcessRunner["Service"]["run"] = (input) =>
394 finalizeRunProcess(runProcessCore(spawner, input), input);
395
396 return ProcessRunner.of({
397 run,
398 });
399});
400
401export const layer = Layer.effect(ProcessRunner, make());