1import { buildSpawnOptions } from "@effect/platform-node-shared/internal/nodeChildProcessSpawner"
2import * as NodeChildProcessSpawner from "@effect/platform-node-shared/NodeChildProcessSpawner"
3import * as NodeFileSystem from "@effect/platform-node-shared/NodeFileSystem"
4import * as NodePath from "@effect/platform-node-shared/NodePath"
5import { assert, describe, it } from "@effect/vitest"
6import * as ChildProcessSpawnerTest from "effect-test/process/ChildProcessSpawnerTest"
7import * as ByteSize from "effect/ByteSize"
8import * as Deferred from "effect/Deferred"
9import * as Effect from "effect/Effect"
10import * as Exit from "effect/Exit"
11import * as FileSystem from "effect/FileSystem"
12import * as Layer from "effect/Layer"
13import * as ChildProcess from "effect/process/ChildProcess"
14import * as Scope from "effect/Scope"
15import * as Stream from "effect/Stream"
16import { join } from "node:path"
17
18const NodeServices = NodeChildProcessSpawner.layer.pipe(
19 Layer.provideMerge(Layer.mergeAll(
20 NodeFileSystem.layer,
21 NodePath.layer
22 ))
23)
24
25ChildProcessSpawnerTest.suite("NodeChildProcessSpawner", NodeServices, {
26 processGroups: true
27})
28
29describe("buildSpawnOptions", () => {
30 const base = { stdio: "pipe" } as const
31
32 it("defaults to hiding non-detached Windows children", () => {
33 assert.deepStrictEqual(buildSpawnOptions({}, base, "win32"), {
34 stdio: "pipe",
35 detached: false,
36 shell: undefined,
37 windowsHide: true
38 })
39 assert.deepStrictEqual(buildSpawnOptions({ detached: true }, base, "win32"), {
40 stdio: "pipe",
41 detached: true,
42 shell: undefined,
43 windowsHide: false
44 })
45 assert.deepStrictEqual(buildSpawnOptions({ detached: false }, base, "win32"), {
46 stdio: "pipe",
47 detached: false,
48 shell: undefined,
49 windowsHide: true
50 })
51 })
52
53 it("allows windowsHide to be configured independently of detached", () => {
54 assert.deepStrictEqual(
55 buildSpawnOptions({ detached: false, windowsHide: false }, base, "win32"),
56 {
57 stdio: "pipe",
58 detached: false,
59 shell: undefined,
60 windowsHide: false
61 }
62 )
63 assert.deepStrictEqual(
64 buildSpawnOptions({ detached: true, windowsHide: true }, base, "win32"),
65 {
66 stdio: "pipe",
67 detached: true,
68 shell: undefined,
69 windowsHide: true
70 }
71 )
72 })
73})
74
75it.live("kills every process in a pipeline", () =>
76 Effect.gen(function*() {
77 const fs = yield* FileSystem.FileSystem
78 const directory = yield* fs.makeTempDirectoryScoped()
79 const rootHeartbeat = `${directory}/root-heartbeat`
80 const childHeartbeat = `${directory}/child-heartbeat`
81 const handle = yield* ChildProcess.make(
82 "sh",
83 [
84 "-c",
85 "printf x >> \"$1\"; printf 'ROOT_READY\\n'; while :; do printf x >> \"$1\"; sleep 0.01; done",
86 "pipeline-root",
87 rootHeartbeat
88 ]
89 ).pipe(ChildProcess.pipeTo(ChildProcess.make(
90 "sh",
91 [
92 "-c",
93 "printf x >> \"$1\"; printf 'CHILD_READY\\n'; read -r ready; printf '%s\\n' \"$ready\"; while :; do printf x >> \"$1\"; sleep 0.01; done",
94 "pipeline-child",
95 childHeartbeat
96 ]
97 )))
98 const readyLines = yield* handle.stdout.pipe(
99 Stream.decodeText,
100 Stream.splitLines,
101 Stream.take(2),
102 Stream.runCollect
103 )
104 assert.deepStrictEqual(readyLines, ["CHILD_READY", "ROOT_READY"])
105 yield* handle.kill({ killSignal: "SIGKILL" })
106 const rootSizeAfterKill = (yield* fs.stat(rootHeartbeat)).size
107 const childSizeAfterKill = (yield* fs.stat(childHeartbeat)).size
108 yield* Effect.sleep("100 millis")
109 const rootFinalSize = (yield* fs.stat(rootHeartbeat)).size
110 const childFinalSize = (yield* fs.stat(childHeartbeat)).size
111
112 assert.strictEqual(ByteSize.toBigInt(rootFinalSize), ByteSize.toBigInt(rootSizeAfterKill))
113 assert.strictEqual(ByteSize.toBigInt(childFinalSize), ByteSize.toBigInt(childSizeAfterKill))
114 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
115
116const processGroupFixture = join(__dirname, "fixtures", "process-group.ts")
117
118// Use native timers under TestClock.
119const liveSleep = (millis: number) =>
120 Effect.callback<void>((resume) => {
121 const timer = setTimeout(() => resume(Effect.void), millis)
122 return Effect.sync(() => clearTimeout(timer))
123 })
124
125const liveTimeout = (millis: number) => <A, E, R>(effect: Effect.Effect<A, E, R>) =>
126 Effect.raceFirst(effect, liveSleep(millis).pipe(Effect.andThen(Effect.die(new Error("timed out")))))
127
128const startProcessGroup = (mode: "exit-on-signal" | "ignore-signal", options?: ChildProcess.CommandOptions) =>
129 Effect.gen(function*() {
130 const fs = yield* FileSystem.FileSystem
131 const directory = yield* fs.makeTempDirectoryScoped()
132 const marker = `${directory}/marker`
133 const scope = yield* Scope.fork(yield* Effect.scope)
134 const handle = yield* Scope.provide(scope)(ChildProcess.make(
135 process.execPath,
136 [processGroupFixture, "leader", mode, marker],
137 { stdin: "ignore", ...options }
138 ))
139 const ready = yield* Deferred.make<number>()
140 yield* handle.stdout.pipe(
141 Stream.decodeText,
142 Stream.splitLines,
143 Stream.runForEach((line) =>
144 line.startsWith("READY ") ? Deferred.succeed(ready, Number(line.slice("READY ".length))) : Effect.void
145 ),
146 Effect.forkScoped
147 )
148 const descendantPid = yield* Deferred.await(ready).pipe(liveTimeout(5_000))
149 return { handle, descendantPid, marker, scope }
150 })
151
152const killDescendant = (pid: number) =>
153 Effect.sync(() => {
154 try {
155 process.kill(pid, "SIGKILL")
156 } catch {}
157 })
158
159const assertHeartbeatStopped = (marker: string) =>
160 Effect.gen(function*() {
161 const fs = yield* FileSystem.FileSystem
162 const sizeAfterKill = (yield* fs.stat(marker)).size
163 yield* liveSleep(100)
164 const finalSize = (yield* fs.stat(marker)).size
165 assert.strictEqual(finalSize, sizeAfterKill)
166 })
167
168const timed = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
169 Effect.gen(function*() {
170 const start = Date.now()
171 yield* effect
172 return Date.now() - start
173 })
174
175describe.skipIf(process.platform === "win32")("process group cleanup", () => {
176 it.live("scope release cleans descendants after the leader exits successfully", () =>
177 Effect.gen(function*() {
178 const fs = yield* FileSystem.FileSystem
179 const { descendantPid, handle, marker, scope } = yield* startProcessGroup("exit-on-signal", { stdin: "pipe" })
180 yield* Effect.addFinalizer(() => killDescendant(descendantPid))
181
182 yield* Stream.run(Stream.make(new TextEncoder().encode("exit\n")), handle.stdin)
183 assert.strictEqual(yield* handle.exitCode, 0)
184 assert.doesNotThrow(() => process.kill(descendantPid, 0), "descendant must still be alive after the leader exits")
185 assert.isFalse(yield* fs.exists(marker))
186
187 yield* Scope.close(scope, Exit.void)
188
189 assert.strictEqual(yield* fs.readFileString(marker), "exited")
190 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
191
192 it.live("scope release force kills descendants after the leader exits with a non-zero code", () =>
193 Effect.gen(function*() {
194 const { descendantPid, handle, marker, scope } = yield* startProcessGroup("ignore-signal", {
195 stdin: "pipe",
196 forceKillAfter: "200 millis"
197 })
198 yield* Effect.addFinalizer(() => killDescendant(descendantPid))
199
200 yield* Stream.run(Stream.make(new TextEncoder().encode("exit 1\n")), handle.stdin)
201 assert.strictEqual(yield* handle.exitCode, 1)
202
203 yield* Scope.close(scope, Exit.void).pipe(liveTimeout(2_000))
204
205 yield* assertHeartbeatStopped(marker)
206 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
207
208 it.live("scope release cleans descendants after the leader is killed externally", () =>
209 Effect.gen(function*() {
210 const fs = yield* FileSystem.FileSystem
211 const { descendantPid, handle, marker, scope } = yield* startProcessGroup("exit-on-signal")
212 yield* Effect.addFinalizer(() => killDescendant(descendantPid))
213
214 process.kill(handle.pid, "SIGKILL")
215 assert.isTrue(Exit.isFailure(yield* Effect.exit(handle.exitCode)))
216 assert.doesNotThrow(() => process.kill(descendantPid, 0), "descendant must survive the leader's SIGKILL")
217 assert.isFalse(yield* fs.exists(marker))
218
219 yield* Scope.close(scope, Exit.void)
220
221 assert.isTrue(yield* fs.exists(marker), "scope release must wait for the descendant's exit marker")
222 assert.strictEqual(yield* fs.readFileString(marker), "exited")
223 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
224
225 it.live("scope release waits for descendants that outlive the leader", () =>
226 Effect.gen(function*() {
227 const fs = yield* FileSystem.FileSystem
228 const { handle, marker, scope } = yield* startProcessGroup("exit-on-signal")
229
230 yield* Scope.close(scope, Exit.void)
231
232 assert.isFalse(yield* handle.isRunning)
233 assert.isTrue(yield* fs.exists(marker))
234 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
235
236 it.live("scope release force kills descendants that ignore the kill signal", () =>
237 Effect.gen(function*() {
238 const { marker, scope } = yield* startProcessGroup("ignore-signal", { forceKillAfter: "200 millis" })
239
240 yield* Scope.close(scope, Exit.void)
241
242 yield* assertHeartbeatStopped(marker)
243 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
244
245 it.live("kill force kills descendants that ignore the kill signal", () =>
246 Effect.gen(function*() {
247 const { handle, marker, scope } = yield* startProcessGroup("ignore-signal")
248
249 yield* handle.kill({ forceKillAfter: "200 millis" })
250
251 assert.isFalse(yield* handle.isRunning)
252 yield* assertHeartbeatStopped(marker)
253 yield* Scope.close(scope, Exit.void)
254 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
255
256 it.effect("forceKillAfter escalation does not depend on the Effect clock", () =>
257 Effect.gen(function*() {
258 const { marker, scope } = yield* startProcessGroup("ignore-signal", { forceKillAfter: "200 millis" })
259
260 const releaseMillis = yield* timed(Scope.close(scope, Exit.void))
261
262 assert.isBelow(releaseMillis, 2_000)
263 yield* assertHeartbeatStopped(marker)
264 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
265
266 it.live("scope release returns when a descendant holds the inherited pipe without forceKillAfter", () =>
267 Effect.gen(function*() {
268 const { descendantPid, handle, scope } = yield* startProcessGroup("ignore-signal")
269
270 yield* Effect.gen(function*() {
271 const releaseMillis = yield* timed(Scope.close(scope, Exit.void))
272
273 assert.isFalse(yield* handle.isRunning)
274 assert.isAtLeast(releaseMillis, 1_000)
275 assert.isBelow(releaseMillis, 3_000)
276 }).pipe(Effect.ensuring(killDescendant(descendantPid)))
277 }).pipe(Effect.scoped, Effect.provide(NodeServices)))
278})
279
280it.live("scope release returns when stdout is unread and backpressured", () =>
281 Effect.gen(function*() {
282 const releaseMillis = yield* timed(Effect.scoped(Effect.gen(function*() {
283 yield* ChildProcess.make(process.execPath, [
284 "-e",
285 "process.stdout.write(\"x\".repeat(1024 * 1024)); setInterval(() => {}, 1000)"
286 ], { stdin: "ignore" })
287 yield* Effect.sleep("100 millis")
288 })))
289
290 assert.isBelow(releaseMillis, 2_000)
291 }).pipe(Effect.provide(NodeServices)))