Handle Deno subprocess stdin failures (#1418)

* Handle Deno subprocess stdin failures * Read the start message with raw fd reads so the fixture can close stdin without aborting ---------

morluto committed 3 months ago

Commit d3610e31 parent 8c717445 files · +193 −37Browse files

Unified · Split

5 files

.changeset/handle-deno-stdin-failures.md +5 −0
packages/kernel/runtime-deno-subprocess/src/deno-worker-process.ts +47 −2
packages/kernel/runtime-deno-subprocess/src/fixtures/close-stdin-worker.mjs +37 −0
packages/kernel/runtime-deno-subprocess/src/index.test.ts +23 −0
packages/kernel/runtime-deno-subprocess/src/index.ts +81 −35

.changeset/handle-deno-stdin-failures.md+5−0

@@ -0,0 +1,5 @@
	1	+---
	2	+"executor": patch
	3	+---
	4	+
	5	+Return an execution error when a Deno subprocess closes stdin instead of emitting an unhandled write failure.

packages/kernel/runtime-deno-subprocess/src/deno-worker-process.ts+47−2

@@ -25,11 +25,13 @@ export type DenoWorkerProcessCallbacks = { · 24 unmodified lines
25	25	onStdoutLine: (line: string) => void;
26	26	onStderr: (chunk: string) => void;
27	27	onError: (error: Error) => void;
	28	+ onStdinError: (error: Error) => void;
28	29	onExit: (code: number \| null, signal: NodeJS.Signals \| null) => void;
29	30	};
30	31
31	32	export type DenoWorkerProcess = {
32		− stdin: NodeJS.WritableStream;
	33	+ ready: Promise<void>;
	34	+ write: (data: string) => Promise<void>;
33	35	dispose: () => void;
34	36	};
35	37
@@ -81,6 +83,20 @@ export const spawnDenoWorkerProcess = ( · 45 unmodified lines
81	83	throw new Error("Failed to create piped stdio for Deno worker subprocess");
82	84	}
83	85
	86	+ const ready = new Promise<void>((resolve, reject) => {
	87	+ const onSpawn = () => {
	88	+ child.removeListener("error", onSpawnError);
	89	+ resolve();
	90	+ };
	91	+ const onSpawnError = (cause: unknown) => {
	92	+ child.removeListener("spawn", onSpawn);
	93	+ reject(normalizeError(cause));
	94	+ };
	95	+
	96	+ child.once("spawn", onSpawn);
	97	+ child.once("error", onSpawnError);
	98	+ });
	99	+
84	100	child.stdout.setEncoding("utf8");
85	101	child.stderr.setEncoding("utf8");
86	102
@@ -109,17 +125,44 @@ export const spawnDenoWorkerProcess = ( · 22 unmodified lines
109	125	callbacks.onError(normalizeError(cause));
110	126	};
111	127
	128	+ const onStdinError = (cause: unknown) => {
	129	+ callbacks.onStdinError(normalizeError(cause));
	130	+ };
	131	+
	132	+ const onStdinClose = () => {
	133	+ child.stdin.removeListener("error", onStdinError);
	134	+ };
	135	+
112	136	const onExit = (code: number \| null, signal: NodeJS.Signals \| null) => {
113	137	callbacks.onExit(code, signal);
114	138	};
115	139
116	140	child.stdout.on("data", onStdoutData);
117	141	child.stderr.on("data", onStderrData);
	142	+ child.stdin.on("error", onStdinError);
	143	+ child.stdin.once("close", onStdinClose);
118	144	child.on("error", onError);
119	145	child.on("exit", onExit);
120	146
121	147	let disposed = false;
122	148
	149	+ const write = (data: string): Promise<void> =>
	150	+ new Promise((resolve, reject) => {
	151	+ if (disposed \|\| !child.stdin.writable) {
	152	+ reject(new Error("Deno subprocess stdin is not writable"));
	153	+ return;
	154	+ }
	155	+
	156	+ child.stdin.write(data, (cause: Error \| null \| undefined) => {
	157	+ if (cause) {
	158	+ reject(cause);
	159	+ return;
	160	+ }
	161	+
	162	+ resolve();
	163	+ });
	164	+ });
	165	+
123	166	const dispose = () => {
124	167	if (disposed) {
125	168	return;
@@ -130,6 +173,7 @@ export const spawnDenoWorkerProcess = ( · 4 unmodified lines
130	173	child.stderr!.removeListener("data", onStderrData);
131	174	child.removeListener("error", onError);
132	175	child.removeListener("exit", onExit);
	176	+ child.stdin.destroy();
133	177
134	178	if (!child.killed) {
135	179	child.kill("SIGKILL");
@@ -137,7 +181,8 @@ export const spawnDenoWorkerProcess = ( · 1 unmodified line
137	181	};
138	182
139	183	return {
140		− stdin: child.stdin,
	184	+ ready,
	185	+ write,
141	186	dispose,
142	187	};
143	188	};

packages/kernel/runtime-deno-subprocess/src/fixtures/close-stdin-worker.mjs+37−0

@@ -0,0 +1,37 @@
	1	+#!/usr/bin/env node
	2	+
	3	+// Reads the start message with raw fd reads instead of a stdin stream: only
	4	+// an unwatched fd can be closed here, since libuv aborts the process when
	5	+// fd 0 is closed while a stream watcher is attached, and a stream destroy()
	6	+// never closes the underlying stdio fd, so the host's writes would keep
	7	+// succeeding instead of failing with EPIPE.
	8	+
	9	+import { closeSync, readSync } from "node:fs";
	10	+
	11	+let buffered = "";
	12	+const chunk = Buffer.alloc(4096);
	13	+while (!buffered.includes("\n")) {
	14	+ try {
	15	+ const bytesRead = readSync(0, chunk, 0, chunk.length);
	16	+ if (bytesRead <= 0) break;
	17	+ buffered += chunk.toString("utf8", 0, bytesRead);
	18	+ } catch (error) {
	19	+ if (error.code === "EAGAIN") continue;
	20	+ throw error;
	21	+ }
	22	+}
	23	+
	24	+const { nonce } = JSON.parse(buffered.slice(0, buffered.indexOf("\n")));
	25	+
	26	+closeSync(0);
	27	+process.stdout.write(
	28	+ `@@executor-ipc@@${JSON.stringify({
	29	+ type: "tool_call",
	30	+ nonce,
	31	+ requestId: "request-1",
	32	+ toolPath: "test.call",
	33	+ args: {},
	34	+ })}\n`,
	35	+);
	36	+
	37	+setTimeout(() => process.exit(0), 5_000);

packages/kernel/runtime-deno-subprocess/src/index.test.ts+23−0

@@ -1,3 +1,5 @@
	1	+import { fileURLToPath } from "node:url";
	2	+
1	3	import { describe, expect, it } from "@effect/vitest";
2	4	import * as Data from "effect/Data";
3	5	import * as Effect from "effect/Effect";
@@ -39,6 +41,27 @@ it.effect("returns an actionable error when Deno is missing", () => · 35 unmodified lines
39	41	}),
40	42	);
41	43
	44	+it.effect.skipIf(process.platform === "win32")(
	45	+ "returns an execution error when the subprocess closes stdin",
	46	+ () =>
	47	+ Effect.gen(function* () {
	48	+ const executor = makeDenoSubprocessExecutor({
	49	+ denoExecutable: fileURLToPath(
	50	+ new URL("./fixtures/close-stdin-worker.mjs", import.meta.url),
	51	+ ),
	52	+ timeoutMs: 5_000,
	53	+ });
	54	+ const toolInvoker = makeTestInvoker({
	55	+ "test.call": () => "done",
	56	+ });
	57	+
	58	+ const output = yield* executor.execute("return tools.test.call({});", toolInvoker);
	59	+
	60	+ expect(output.result).toBeNull();
	61	+ expect(output.error).toContain("Failed to write to Deno subprocess stdin");
	62	+ }),
	63	+);
	64	+
42	65	describe.skipIf(!isDenoAvailable())("runtime-deno-subprocess", () => {
43	66	it.effect("executes simple code and returns result", () =>
44	67	Effect.gen(function* () {

packages/kernel/runtime-deno-subprocess/src/index.ts+81−35

@@ -16,7 +16,11 @@ import * as Option from "effect/Option"; · 15 unmodified lines
16	16	import * as Queue from "effect/Queue";
17	17	import * as Schema from "effect/Schema";
18	18
19		−import { type DenoPermissions, spawnDenoWorkerProcess } from "./deno-worker-process";
	19	+import {
	20	+ type DenoPermissions,
	21	+ type DenoWorkerProcess,
	22	+ spawnDenoWorkerProcess,
	23	+} from "./deno-worker-process";
20	24
21	25	export type { DenoPermissions };
22	26
@@ -49,6 +53,15 @@ class DenoSpawnError extends Data.TaggedError("DenoSpawnError")<{ · 26 unmodified lines
49	53	}
50	54	}
51	55
	56	+class DenoProcessWriteError extends Data.TaggedError("DenoProcessWriteError")<{
	57	+ readonly reason: unknown;
	58	+}> {
	59	+ override get message() {
	60	+ const detail = this.reason instanceof Error ? this.reason.message : String(this.reason);
	61	+ return `Failed to write to Deno subprocess stdin: ${detail}`;
	62	+ }
	63	+}
	64	+
52	65	// ---------------------------------------------------------------------------
53	66	// IPC schemas
54	67	// ---------------------------------------------------------------------------
@@ -149,9 +162,14 @@ const causeMessage = (cause: Cause.Cause<unknown>): string => { · 94 unmodified lines
149	162	return String(squashed);
150	163	};
151	164
152		−const writeMessage = (stdin: NodeJS.WritableStream, message: HostToWorkerMessage): void => {
153		− stdin.write(`${JSON.stringify(message)}\n`);
154		−};
	165	+const writeMessage = (
	166	+ worker: DenoWorkerProcess,
	167	+ message: HostToWorkerMessage,
	168	+): Effect.Effect<void, DenoProcessWriteError> =>
	169	+ Effect.tryPromise({
	170	+ try: () => worker.write(`${JSON.stringify(message)}\n`),
	171	+ catch: (reason) => new DenoProcessWriteError({ reason }),
	172	+ });
155	173
156	174	// ---------------------------------------------------------------------------
157	175	// Core execution
@@ -217,6 +235,14 @@ const executeInDeno = ( · 59 unmodified lines
217	235	}),
218	236	);
219	237	},
	238	+ onStdinError: (cause) => {
	239	+ runSync(
	240	+ completeWith({
	241	+ result: null,
	242	+ error: new DenoProcessWriteError({ reason: cause }).message,
	243	+ }),
	244	+ );
	245	+ },
220	246	onExit: (exitCode, signal) => {
221	247	runSync(
222	248	completeWith({
@@ -234,8 +260,21 @@ const executeInDeno = ( · 11 unmodified lines
234	260	let inFlight = 0;
235	261	let lastReturnedAt = start;
236	262
237		− // Send code to the subprocess
238		− writeMessage(worker.stdin, { type: "start", code: recoveredBody, nonce });
	263	+ const startError = yield* Effect.gen(function* () {
	264	+ yield* Effect.tryPromise({
	265	+ try: () => worker.ready,
	266	+ catch: (reason) => new DenoSpawnError({ executable: denoExecutable, reason }),
	267	+ });
	268	+ yield* writeMessage(worker, { type: "start", code: recoveredBody, nonce });
	269	+ }).pipe(
	270	+ Effect.as<Error \| undefined>(undefined),
	271	+ Effect.catch((error) => Effect.succeed(error)),
	272	+ Effect.onInterrupt(() => Effect.sync(worker.dispose)),
	273	+ );
	274	+ if (startError) {
	275	+ worker.dispose();
	276	+ return { result: null, error: startError.message };
	277	+ }
239	278
240	279	// Measure only continuous autonomous subprocess compute. Tool dispatches
241	280	// suspend the clock and each return grants a fresh timeout budget.
@@ -269,35 +308,42 @@ const executeInDeno = ( · 27 unmodified lines
269	308	// must run even if the invoke or write dies, or the watchdog's
270	309	// in-flight guard would suspend the timeout forever.
271	310	inFlight += 1;
272		− yield* toolInvoker.invoke({ path: msg.toolPath, args: msg.args }).pipe(
273		− Effect.map(
274		− (value): HostToWorkerMessage => ({
275		− type: "tool_result",
276		− nonce,
277		− requestId: msg.requestId,
278		− ok: true,
279		− value,
280		− }),
281		− ),
282		− Effect.catchCause((cause) =>
283		− Effect.succeed<HostToWorkerMessage>({
284		− type: "tool_result",
285		− nonce,
286		− requestId: msg.requestId,
287		− ok: false,
288		− error: causeMessage(cause),
289		− }),
290		− ),
291		− Effect.flatMap((toolResult) =>
292		− Effect.sync(() => writeMessage(worker.stdin, toolResult)),
293		− ),
294		− Effect.ensuring(
295		− Effect.sync(() => {
296		− inFlight -= 1;
297		− lastReturnedAt = Date.now();
298		− }),
299		− ),
300		− );
	311	+ const writeSucceeded = yield* toolInvoker
	312	+ .invoke({ path: msg.toolPath, args: msg.args })
	313	+ .pipe(
	314	+ Effect.map(
	315	+ (value): HostToWorkerMessage => ({
	316	+ type: "tool_result",
	317	+ nonce,
	318	+ requestId: msg.requestId,
	319	+ ok: true,
	320	+ value,
	321	+ }),
	322	+ ),
	323	+ Effect.catchCause((cause) =>
	324	+ Effect.succeed<HostToWorkerMessage>({
	325	+ type: "tool_result",
	326	+ nonce,
	327	+ requestId: msg.requestId,
	328	+ ok: false,
	329	+ error: causeMessage(cause),
	330	+ }),
	331	+ ),
	332	+ Effect.flatMap((toolResult) => writeMessage(worker, toolResult)),
	333	+ Effect.as(true),
	334	+ Effect.catchTag("DenoProcessWriteError", (error) =>
	335	+ completeWith({ result: null, error: error.message }).pipe(Effect.as(false)),
	336	+ ),
	337	+ Effect.ensuring(
	338	+ Effect.sync(() => {
	339	+ inFlight -= 1;
	340	+ lastReturnedAt = Date.now();
	341	+ }),
	342	+ ),
	343	+ );
	344	+ if (!writeSucceeded) {
	345	+ return;
	346	+ }
301	347	break;
302	348	}
303	349

Handle Deno subprocess stdin failures (#1418)

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code