Skip to content

Commit ef272b1

Browse files
reoringreoring
committed
fix(sentinel): use Promise.race for probe timeout to avoid pipe hang
Instead of relying solely on proc.kill() to close pipes (which may fail when shell child processes keep pipes open), race the probe work against a timeout promise. This guarantees run() returns within timeoutMs regardless of process cleanup behavior.
1 parent 23a427c commit ef272b1

1 file changed

Lines changed: 76 additions & 66 deletions

File tree

‎src/workflow/sentinel/probe-runner.ts‎

Lines changed: 76 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -55,80 +55,30 @@ export class ProbeRunner {
5555
env: { ...process.env, ...(this.env ?? {}) },
5656
});
5757

58-
// Hard timeout: kill the probe if it exceeds the deadline.
59-
// Kill the entire process group (negative PID) with SIGKILL to ensure
60-
// child processes spawned by the shell are also terminated — otherwise
61-
// orphaned children keep the pipe open and block stream reads.
62-
const timeoutId = setTimeout(() => {
63-
try {
64-
process.kill(-proc.pid, "SIGKILL");
65-
} catch {
66-
// Process group kill may fail (e.g. process already exited, or not
67-
// a process group leader). Fall back to killing the direct child.
68-
try { proc.kill(9); } catch { /* already dead */ }
69-
}
70-
}, this.timeoutMs);
71-
72-
// Read stdout and stderr concurrently to avoid deadlocks.
73-
// readBoundedDrain reads up to maxBytes into memory, then drains the
74-
// remainder without storing it. This prevents pipe-full blocking when
75-
// the process emits more data than the limit.
76-
const [stdoutChunks, stderrText] = await Promise.all([
77-
this.readBoundedDrain(
78-
proc.stdout as ReadableStream<Uint8Array>,
79-
MAX_PROBE_OUTPUT_BYTES,
80-
proc,
81-
),
82-
this.readBoundedStringDrain(
83-
proc.stderr as ReadableStream<Uint8Array>,
84-
MAX_PROBE_STDERR_BYTES,
85-
proc,
86-
),
87-
]);
88-
89-
// Wait for process to exit and get exit code, then clear the timeout.
90-
// Clearing AFTER exit ensures the hard timeout stays active until the
91-
// process is fully reaped (avoids hanging on proc.exited).
92-
const exitCode = await proc.exited;
93-
clearTimeout(timeoutId);
94-
95-
const bounded = new TextDecoder()
96-
.decode(Buffer.concat(stdoutChunks.chunks))
97-
.slice(0, MAX_PROBE_OUTPUT_BYTES);
98-
99-
// Probe MUST emit valid JSON.
100-
let output: ProbeOutput;
101-
try {
102-
output = JSON.parse(bounded.trim());
103-
} catch {
104-
return {
105-
success: false,
106-
digest: "",
107-
error: "probe output is not valid JSON",
108-
exitCode,
109-
stderr: stderrText || undefined,
110-
ts,
111-
};
112-
}
58+
// Race the actual work against a hard deadline. Even if process kill
59+
// fails to close pipes (e.g. orphaned children keep them open), the
60+
// timeout branch resolves immediately so the caller never hangs.
61+
const TIMEOUT_SENTINEL = Symbol("timeout");
62+
const timeoutPromise = new Promise<typeof TIMEOUT_SENTINEL>((resolve) => {
63+
setTimeout(() => resolve(TIMEOUT_SENTINEL), this.timeoutMs);
64+
});
11365

114-
// Compute digest: prefer explicit `digest` field from the probe, fall
115-
// back to a hash of the normalised JSON.
116-
const digest = output.digest ?? this.computeDigest(output);
66+
const workPromise = this.runInner(proc, ts);
67+
const result = await Promise.race([workPromise, timeoutPromise]);
11768

118-
// Check exit code when require_zero_exit is enabled
119-
if (this.requireZeroExit && exitCode !== 0) {
69+
if (result === TIMEOUT_SENTINEL) {
70+
// Best-effort kill — don't await anything that might hang.
71+
try { process.kill(-proc.pid, "SIGKILL"); } catch { /* ignore */ }
72+
try { proc.kill(9); } catch { /* ignore */ }
12073
return {
12174
success: false,
122-
output, // Include parsed output for diagnostic purposes
123-
digest,
124-
error: `probe exited with non-zero code ${exitCode} (require_zero_exit is enabled)`,
125-
exitCode,
126-
stderr: stderrText || undefined,
75+
digest: "",
76+
error: "probe timed out",
12777
ts,
12878
};
12979
}
13080

131-
return { success: true, output, digest, exitCode, ts };
81+
return result;
13282
} catch (err) {
13383
return {
13484
success: false,
@@ -139,6 +89,66 @@ export class ProbeRunner {
13989
}
14090
}
14191

92+
/** Inner implementation that reads stdout/stderr and waits for exit. */
93+
private async runInner(
94+
proc: ReturnType<typeof Bun.spawn>,
95+
ts: number,
96+
): Promise<ProbeResult> {
97+
// Read stdout and stderr concurrently to avoid deadlocks.
98+
const [stdoutChunks, stderrText] = await Promise.all([
99+
this.readBoundedDrain(
100+
proc.stdout as ReadableStream<Uint8Array>,
101+
MAX_PROBE_OUTPUT_BYTES,
102+
proc,
103+
),
104+
this.readBoundedStringDrain(
105+
proc.stderr as ReadableStream<Uint8Array>,
106+
MAX_PROBE_STDERR_BYTES,
107+
proc,
108+
),
109+
]);
110+
111+
const exitCode = await proc.exited;
112+
113+
const bounded = new TextDecoder()
114+
.decode(Buffer.concat(stdoutChunks.chunks))
115+
.slice(0, MAX_PROBE_OUTPUT_BYTES);
116+
117+
// Probe MUST emit valid JSON.
118+
let output: ProbeOutput;
119+
try {
120+
output = JSON.parse(bounded.trim());
121+
} catch {
122+
return {
123+
success: false,
124+
digest: "",
125+
error: "probe output is not valid JSON",
126+
exitCode,
127+
stderr: stderrText || undefined,
128+
ts,
129+
};
130+
}
131+
132+
// Compute digest: prefer explicit `digest` field from the probe, fall
133+
// back to a hash of the normalised JSON.
134+
const digest = output.digest ?? this.computeDigest(output);
135+
136+
// Check exit code when require_zero_exit is enabled
137+
if (this.requireZeroExit && exitCode !== 0) {
138+
return {
139+
success: false,
140+
output, // Include parsed output for diagnostic purposes
141+
digest,
142+
error: `probe exited with non-zero code ${exitCode} (require_zero_exit is enabled)`,
143+
exitCode,
144+
stderr: stderrText || undefined,
145+
ts,
146+
};
147+
}
148+
149+
return { success: true, output, digest, exitCode, ts };
150+
}
151+
142152
/**
143153
* Read up to `maxBytes` from a stream into memory, then drain any remaining
144154
* data without storing it. If the process is still writing after `maxBytes`,

0 commit comments

Comments
 (0)