Skip to content

Commit 49ef234

Browse files
mdev34-labmdev34-lab
authored
fix(core): handle detached child processes holding stdio open
Separate process exit from stdio close so child process lifecycle does not hang when detached descendants inherit stdio. - resolve exitCode/isRunning from the exit event - make kill() wait for process exit instead of stream close - add best-effort descendant reaping after process exit - drain post-exit output until the stream closes or stays quiet - avoid the post-exit delay for normal commands - add regression tests for detached children and kill teardown Fixes anomalyco#24731
2 parents 9b0dd36 + 030477a commit 49ef234

6 files changed

Lines changed: 404 additions & 67 deletions

File tree

‎packages/core/src/cross-spawn-spawner.ts‎

Lines changed: 79 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -265,24 +265,30 @@ export const make = Effect.gen(function* () {
265265
}
266266

267267
const spawn = (command: ChildProcess.StandardCommand, opts: NodeChildProcess.SpawnOptions) =>
268-
Effect.callback<readonly [NodeChildProcess.ChildProcess, ExitSignal], PlatformError.PlatformError>((resume) => {
269-
const signal = Deferred.makeUnsafe<readonly [code: number | null, signal: NodeJS.Signals | null]>()
268+
Effect.callback<
269+
readonly [NodeChildProcess.ChildProcess, ExitSignal, ExitSignal],
270+
PlatformError.PlatformError
271+
>((resume) => {
272+
const exited = Deferred.makeUnsafe<readonly [code: number | null, signal: NodeJS.Signals | null]>()
273+
const closed = Deferred.makeUnsafe<readonly [code: number | null, signal: NodeJS.Signals | null]>()
270274
const proc = launch(command.command, command.args, opts)
271-
let end = false
272275
let exit: readonly [code: number | null, signal: NodeJS.Signals | null] | undefined
273276
proc.on("error", (err) => {
274277
resume(Effect.fail(toPlatformError("spawn", err, command)))
275278
})
279+
// A detached descendant can hold the inherited stdio open after this process
280+
// exits. Report the exit promptly, while keeping "close" for kill/escalation
281+
// and for failed spawns where no exit event is emitted.
276282
proc.on("exit", (...args) => {
277283
exit = args
284+
Deferred.doneUnsafe(exited, Exit.succeed(args))
278285
})
279286
proc.on("close", (...args) => {
280-
if (end) return
281-
end = true
282-
Deferred.doneUnsafe(signal, Exit.succeed(exit ?? args))
287+
Deferred.doneUnsafe(closed, Exit.succeed(exit ?? args))
288+
Deferred.doneUnsafe(exited, Exit.succeed(exit ?? args))
283289
})
284290
proc.on("spawn", () => {
285-
resume(Effect.succeed([proc, signal]))
291+
resume(Effect.succeed([proc, exited, closed]))
286292
})
287293
return Effect.sync(() => {
288294
proc.kill("SIGTERM")
@@ -321,26 +327,62 @@ export const make = Effect.gen(function* () {
321327
return Effect.fail(toPlatformError("kill", new Error("Failed to kill child process"), command))
322328
})
323329

324-
const timeout =
325-
(
326-
proc: NodeChildProcess.ChildProcess,
327-
command: ChildProcess.StandardCommand,
328-
opts: ChildProcess.KillOptions | undefined,
329-
) =>
330-
<A, E, R>(
331-
f: (
332-
command: ChildProcess.StandardCommand,
333-
proc: NodeChildProcess.ChildProcess,
334-
signal: NodeJS.Signals,
335-
) => Effect.Effect<A, E, R>,
336-
) => {
337-
const signal = opts?.killSignal ?? "SIGTERM"
338-
if (Predicate.isUndefined(opts?.forceKillAfter)) return f(command, proc, signal)
339-
return Effect.timeoutOrElse(f(command, proc, signal), {
340-
duration: opts.forceKillAfter,
341-
orElse: () => f(command, proc, "SIGKILL"),
342-
})
343-
}
330+
const terminate = (
331+
command: ChildProcess.StandardCommand,
332+
proc: NodeChildProcess.ChildProcess,
333+
exited: ExitSignal,
334+
opts?: ChildProcess.KillOptions,
335+
) => {
336+
const sig = opts?.killSignal ?? "SIGTERM"
337+
const send = (s: NodeJS.Signals) => Effect.catch(killGroup(command, proc, s), () => killOne(command, proc, s))
338+
// Wait for the process to exit, not for its stdio to close: a detached
339+
// descendant can keep the inherited stdio open long after the process is
340+
// gone, so waiting on `closed` here would hang kill()/teardown forever.
341+
const attempt = send(sig).pipe(Effect.andThen(Deferred.await(exited)), Effect.asVoid)
342+
const grace = opts?.forceKillAfter
343+
if (!grace) return attempt
344+
return Effect.timeoutOrElse(attempt, {
345+
duration: grace,
346+
// SIGKILL is the last resort: signal the group, then wait for the process
347+
// to exit only up to the same grace. The process should die immediately on
348+
// SIGKILL, but never block teardown if it somehow does not.
349+
orElse: () =>
350+
send("SIGKILL").pipe(
351+
Effect.andThen(
352+
Effect.timeoutOrElse(Deferred.await(exited), {
353+
duration: grace,
354+
orElse: () => Effect.void,
355+
}),
356+
),
357+
Effect.asVoid,
358+
),
359+
})
360+
}
361+
362+
// Best-effort cleanup for a process that has already exited. Signal its group
363+
// to sweep any descendants it left behind, but never block on `closed`: a
364+
// descendant that escaped the group (or ignores the signal) keeps the
365+
// inherited stdio open, which would otherwise hang scope teardown forever.
366+
const reap = (
367+
command: ChildProcess.StandardCommand,
368+
proc: NodeChildProcess.ChildProcess,
369+
closed: ExitSignal,
370+
opts?: ChildProcess.KillOptions,
371+
) => {
372+
const sig = opts?.killSignal ?? "SIGTERM"
373+
const send = (s: NodeJS.Signals) => Effect.catch(killGroup(command, proc, s), () => killOne(command, proc, s))
374+
if (!opts?.forceKillAfter) return send(sig)
375+
// Give the group a grace period to drain, then force-kill stragglers.
376+
return send(sig).pipe(
377+
Effect.andThen(
378+
Effect.timeoutOrElse(Deferred.await(closed), {
379+
duration: opts.forceKillAfter,
380+
orElse: () => send("SIGKILL"),
381+
}),
382+
),
383+
Effect.asVoid,
384+
)
385+
}
344386

345387
const source = (handle: ChildProcessHandle, from: ChildProcess.PipeFromOption | undefined) => {
346388
const opt = from ?? "stdout"
@@ -370,7 +412,7 @@ export const make = Effect.gen(function* () {
370412
const extra = fds(command.options)
371413
const dir = yield* cwd(command.options)
372414

373-
const [proc, signal] = yield* Effect.acquireRelease(
415+
const [proc, exited] = yield* Effect.acquireRelease(
374416
spawn(command, {
375417
cwd: dir,
376418
env: env(command.options),
@@ -379,26 +421,16 @@ export const make = Effect.gen(function* () {
379421
shell: command.options.shell,
380422
windowsHide: process.platform === "win32",
381423
}),
382-
Effect.fnUntraced(function* ([proc, signal]) {
383-
const done = yield* Deferred.isDone(signal)
384-
const kill = timeout(proc, command, command.options)
424+
Effect.fnUntraced(function* ([proc, exited, closed]) {
425+
const done = yield* Deferred.isDone(exited)
385426
if (done) {
386-
const [code] = yield* Deferred.await(signal)
427+
const [code] = yield* Deferred.await(exited)
387428
if (process.platform === "win32") return yield* Effect.void
388-
if (code !== 0 && Predicate.isNotNull(code)) return yield* Effect.ignore(kill(killGroup))
429+
if (code !== 0 && Predicate.isNotNull(code))
430+
return yield* Effect.ignore(reap(command, proc, closed, command.options))
389431
return yield* Effect.void
390432
}
391-
const send = (s: NodeJS.Signals) =>
392-
Effect.catch(killGroup(command, proc, s), () => killOne(command, proc, s))
393-
const sig = command.options.killSignal ?? "SIGTERM"
394-
const attempt = send(sig).pipe(Effect.andThen(Deferred.await(signal)), Effect.asVoid)
395-
const escalated = command.options.forceKillAfter
396-
? Effect.timeoutOrElse(attempt, {
397-
duration: command.options.forceKillAfter,
398-
orElse: () => send("SIGKILL").pipe(Effect.andThen(Deferred.await(signal)), Effect.asVoid),
399-
})
400-
: attempt
401-
return yield* Effect.ignore(escalated)
433+
return yield* Effect.ignore(terminate(command, proc, exited, command.options))
402434
}),
403435
)
404436

@@ -413,8 +445,8 @@ export const make = Effect.gen(function* () {
413445
all: out.all,
414446
getInputFd: fd.getInputFd,
415447
getOutputFd: fd.getOutputFd,
416-
isRunning: Effect.map(Deferred.isDone(signal), (done) => !done),
417-
exitCode: Effect.flatMap(Deferred.await(signal), ([code, signal]) => {
448+
isRunning: Effect.map(Deferred.isDone(exited), (done) => !done),
449+
exitCode: Effect.flatMap(Deferred.await(exited), ([code, signal]) => {
418450
if (Predicate.isNotNull(code)) return Effect.succeed(ExitCode(code))
419451
return Effect.fail(
420452
toPlatformError(
@@ -424,17 +456,7 @@ export const make = Effect.gen(function* () {
424456
),
425457
)
426458
}),
427-
kill: (opts?: ChildProcess.KillOptions) => {
428-
const sig = opts?.killSignal ?? "SIGTERM"
429-
const send = (s: NodeJS.Signals) =>
430-
Effect.catch(killGroup(command, proc, s), () => killOne(command, proc, s))
431-
const attempt = send(sig).pipe(Effect.andThen(Deferred.await(signal)), Effect.asVoid)
432-
if (!opts?.forceKillAfter) return attempt
433-
return Effect.timeoutOrElse(attempt, {
434-
duration: opts.forceKillAfter,
435-
orElse: () => send("SIGKILL").pipe(Effect.andThen(Deferred.await(signal)), Effect.asVoid),
436-
})
437-
},
459+
kill: (opts?: ChildProcess.KillOptions) => terminate(command, proc, exited, opts),
438460
unref: Effect.sync(() => {
439461
if (ref) {
440462
proc.unref()

‎packages/core/src/process.ts‎

Lines changed: 57 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,9 @@
11
import { Context, Duration, Effect, Fiber, Layer, Schema, Stream } from "effect"
22
import type { PlatformError } from "effect/PlatformError"
3+
import type * as Scope from "effect/Scope"
34
import { ChildProcess } from "effect/unstable/process"
45
import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"
6+
import type { ExitCode } from "effect/unstable/process/ChildProcessSpawner"
57
import { CrossSpawnSpawner } from "./cross-spawn-spawner"
68
import { makeGlobalNode } from "./effect/app-node"
79

@@ -136,6 +138,58 @@ export const collectStream = (stream: Stream.Stream<Uint8Array, PlatformError>,
136138
},
137139
).pipe(Effect.map((x) => ({ buffer: Buffer.concat(x.chunks), truncated: x.truncated })))
138140

141+
const POST_EXIT_QUIET = Duration.millis(500)
142+
143+
/**
144+
* Collects a child process output stream without waiting for it to close.
145+
*
146+
* A detached descendant can inherit the child's stdio and keep the pipe open
147+
* after the child exits, so waiting for EOF/`close` hangs forever. Instead,
148+
* once the process exits, keep draining until the stream has been quiet for
149+
* `POST_EXIT_QUIET`, then stop even if the stream never closes.
150+
*/
151+
const collectUntilQuiet = (
152+
stream: Stream.Stream<Uint8Array, PlatformError>,
153+
maxOutputBytes: number | undefined,
154+
exit: Effect.Effect<ExitCode, PlatformError>,
155+
): Effect.Effect<{ buffer: Buffer; truncated: boolean }, PlatformError, Scope.Scope> =>
156+
Effect.gen(function* () {
157+
const acc = { chunks: [] as Uint8Array[], bytes: 0, truncated: false }
158+
let activity = -1
159+
const output = yield* Effect.forkScoped(
160+
Stream.runForEach(stream, (chunk) =>
161+
Effect.sync(() => {
162+
activity++
163+
if (maxOutputBytes === undefined) {
164+
acc.chunks.push(chunk)
165+
acc.bytes += chunk.length
166+
return
167+
}
168+
const remaining = maxOutputBytes - acc.bytes
169+
if (remaining > 0) acc.chunks.push(remaining >= chunk.length ? chunk : chunk.slice(0, remaining))
170+
acc.bytes += chunk.length
171+
acc.truncated = acc.truncated || acc.bytes > maxOutputBytes
172+
}),
173+
),
174+
)
175+
yield* exit
176+
// Prefer the stream closing: a normal command closes its pipe right after
177+
// exiting, so `Fiber.join(output)` wins the race and we return immediately
178+
// (no idle delay on the happy path). The quiet window only kicks in when the
179+
// stream stays open after exit — a detached descendant holding the pipe.
180+
const quiet = Effect.gen(function* () {
181+
let last = activity
182+
while (true) {
183+
yield* Effect.sleep(POST_EXIT_QUIET)
184+
const current = activity
185+
if (current === last) return
186+
last = current
187+
}
188+
})
189+
yield* Effect.raceFirst(Fiber.join(output), quiet)
190+
return { buffer: Buffer.concat(acc.chunks), truncated: acc.truncated }
191+
})
192+
139193
const layer = Layer.effect(
140194
Service,
141195
Effect.gen(function* () {
@@ -148,7 +202,7 @@ const layer = Layer.effect(
148202
const handle = yield* spawner.spawn(command)
149203
if (options?.combineOutput) {
150204
const [output, exitCode] = yield* Effect.all(
151-
[collectStream(handle.all, options.maxOutputBytes), handle.exitCode],
205+
[collectUntilQuiet(handle.all, options.maxOutputBytes, handle.exitCode), handle.exitCode],
152206
{ concurrency: "unbounded" },
153207
)
154208
return {
@@ -164,8 +218,8 @@ const layer = Layer.effect(
164218
}
165219
const [stdout, stderr, exitCode] = yield* Effect.all(
166220
[
167-
collectStream(handle.stdout, options?.maxOutputBytes),
168-
collectStream(handle.stderr, options?.maxErrorBytes),
221+
collectUntilQuiet(handle.stdout, options?.maxOutputBytes, handle.exitCode),
222+
collectUntilQuiet(handle.stderr, options?.maxErrorBytes, handle.exitCode),
169223
handle.exitCode,
170224
],
171225
{ concurrency: "unbounded" },

‎packages/core/test/effect/cross-spawn-spawner.test.ts‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,20 @@ async function gone(pid: number, timeout = 5_000) {
5959
return !alive(pid)
6060
}
6161

62+
async function readPid(file: string, timeout = 5_000) {
63+
const end = Date.now() + timeout
64+
while (Date.now() < end) {
65+
try {
66+
const value = await fs.readFile(file, "utf-8")
67+
if (value) return Number(value)
68+
} catch (error) {
69+
if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error
70+
}
71+
await new Promise((resolve) => setTimeout(resolve, 20))
72+
}
73+
throw new Error(`Process did not write its pid to ${file}`)
74+
}
75+
6276
describe("cross-spawn spawner", () => {
6377
describe("basic spawning", () => {
6478
fx.effect(
@@ -287,6 +301,94 @@ describe("cross-spawn spawner", () => {
287301
)
288302
})
289303

304+
describe("detached children (issue #24731)", () => {
305+
fx.live(
306+
"exitCode resolves when the main process exits even if a detached child keeps stdio open",
307+
Effect.gen(function* () {
308+
const tmp = yield* Effect.acquireRelease(
309+
Effect.promise(() => tmpdir()),
310+
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
311+
)
312+
const pidFile = path.join(tmp.path, "daemon.pid")
313+
314+
// Mimics playwright-cli / a backgrounded server: spawn a long-lived detached
315+
// child that inherits stdio (keeping the parent's stdout pipe write end open),
316+
// then exit the main process immediately.
317+
const code = [
318+
'const cp = require("node:child_process")',
319+
'const fs = require("node:fs")',
320+
'const daemon = cp.spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { detached: true, stdio: "inherit" })',
321+
"daemon.unref()",
322+
"fs.writeFileSync(process.argv[1], String(daemon.pid))",
323+
'process.stdout.write("started")',
324+
"process.exit(0)",
325+
].join("\n")
326+
327+
const handle = yield* ChildProcess.make(process.execPath, ["-e", code, pidFile])
328+
329+
const result = yield* Effect.raceAll([
330+
handle.exitCode.pipe(Effect.map((exit) => ({ kind: "exit" as const, code: exit }))),
331+
Effect.sleep("5 seconds").pipe(Effect.as({ kind: "timeout" as const, code: null })),
332+
])
333+
334+
// Reap the detached daemon so it does not leak and so the spawner's
335+
// "close" can fire during scope teardown.
336+
const pid = yield* Effect.promise(() => readPid(pidFile))
337+
yield* Effect.sync(() => {
338+
try {
339+
process.kill(pid, "SIGKILL")
340+
} catch {}
341+
})
342+
343+
expect(result.kind).toBe("exit")
344+
expect(result.code).toBe(ChildProcessSpawner.ExitCode(0))
345+
}),
346+
15_000,
347+
)
348+
349+
fx.live(
350+
"kill without forceKillAfter returns when the parent exits even if a detached child keeps stdio open",
351+
Effect.gen(function* () {
352+
if (process.platform === "win32") return
353+
354+
const tmp = yield* Effect.acquireRelease(
355+
Effect.promise(() => tmpdir()),
356+
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
357+
)
358+
const pidFile = path.join(tmp.path, "daemon.pid")
359+
360+
// The parent keeps running after spawning a detached child that inherits
361+
// stdio, so the parent's stdio pipe stays open after the parent dies.
362+
// kill() must return on the parent's exit, not hang on its `close`.
363+
const code = [
364+
'const cp = require("node:child_process")',
365+
'const fs = require("node:fs")',
366+
'const daemon = cp.spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { detached: true, stdio: "inherit" })',
367+
"daemon.unref()",
368+
"fs.writeFileSync(process.argv[1], String(daemon.pid))",
369+
"setInterval(() => {}, 1000)",
370+
].join("\n")
371+
372+
const handle = yield* ChildProcess.make(process.execPath, ["-e", code, pidFile])
373+
const pid = yield* Effect.promise(() => readPid(pidFile))
374+
375+
const start = Date.now()
376+
yield* handle.kill()
377+
const elapsed = Date.now() - start
378+
379+
// Reap the detached daemon so it does not leak.
380+
yield* Effect.sync(() => {
381+
try {
382+
process.kill(pid, "SIGKILL")
383+
} catch {}
384+
})
385+
386+
expect(elapsed).toBeLessThan(5_000)
387+
}),
388+
15_000,
389+
)
390+
})
391+
290392
describe("error handling", () => {
291393
fx.effect(
292394
"fails for invalid command",

0 commit comments

Comments
 (0)