From f828f8b76073bd744e6791155ba27e39e52160a5 Mon Sep 17 00:00:00 2001 From: Lakshmi Tanmay Date: Tue, 3 Nov 2026 19:53:19 +0530 Subject: [PATCH 01/14] fix(server): isolate Linux process launches from the RPC event loop --- apps/server/src/bin.ts | 5 +- apps/server/src/process/SpawnHost.test.ts | 265 ++++++++++++++++ apps/server/src/process/SpawnHost.ts | 300 ++++++++++++++++++ .../src/process/spawnHost.test-fixture.ts | 23 ++ apps/server/src/process/spawnHostProcess.ts | 124 ++++++++ apps/server/src/server.ts | 9 +- .../@effect__platform-node-shared@4.1.2.patch | 65 ++++ pnpm-lock.yaml | 9 +- pnpm-workspace.yaml | 1 + 9 files changed, 795 insertions(+), 6 deletions(-) create mode 100644 apps/server/src/process/SpawnHost.test.ts create mode 100644 apps/server/src/process/SpawnHost.ts create mode 100644 apps/server/src/process/spawnHost.test-fixture.ts create mode 100644 apps/server/src/process/spawnHostProcess.ts create mode 100644 patches/@effect__platform-node-shared@4.1.2.patch diff --git a/apps/server/src/bin.ts b/apps/server/src/bin.ts index a6dfe16f066a4..aafb421047f69 100644 --- a/apps/server/src/bin.ts +++ b/apps/server/src/bin.ts @@ -18,7 +18,10 @@ if ( // Guard sockets before either dispatch path can load outbound-capable modules. const { ignoreMacosTypeOfServiceEinval } = await import("./outboundHttp.ts"); ignoreMacosTypeOfServiceEinval(); - if (command === "__document-text") { + if (command === "__spawn-host") { + const { runSpawnHost } = await import("./process/spawnHostProcess.ts"); + runSpawnHost(); + } else if (command === "__document-text") { const { extractDocumentText } = await import("./provider/documentTextWorker.ts"); try { process.stdout.write(await extractDocumentText(process.argv[3]!, process.argv[4]!)); diff --git a/apps/server/src/process/SpawnHost.test.ts b/apps/server/src/process/SpawnHost.test.ts new file mode 100644 index 0000000000000..800303ee74e33 --- /dev/null +++ b/apps/server/src/process/SpawnHost.test.ts @@ -0,0 +1,265 @@ +// @effect-diagnostics nodeBuiltinImport:off - the regression intercepts the native spawn boundary and inspects IPC sockets. +// oxlint-disable-next-line t3code/namespace-node-imports -- Mutating the CJS export is necessary to intercept the native ESM binding. +import NodeChildProcess from "node:child_process"; +import * as NodeModule from "node:module"; +import type * as NodeStream from "node:stream"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import { vi } from "vite-plus/test"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Result from "effect/Result"; +import * as Stream from "effect/Stream"; +import * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; + +import * as SpawnHost from "./SpawnHost.ts"; + +const layer = SpawnHost.layerHost(process.execPath, [ + new URL("../bin.ts", import.meta.url).pathname, + "__spawn-host", +]).pipe(Layer.provide(NodeServices.layer)); +const text = (stream: Stream.Stream) => + stream.pipe(Stream.decodeText(), Stream.mkString); + +// quarantine: #30183 Platform coverage: this layer is installed only on Linux. +describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("SpawnHost", () => { + it.live("starts commands without calling native spawn on the server thread", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const nativeSpawn = NodeChildProcess.spawn; + const helpers: NodeChildProcess.ChildProcess[] = []; + using _spawn = vi + .spyOn(NodeChildProcess, "spawn") + .mockImplementation((command, args, options) => { + if (command === "/bin/echo") throw new Error("Native spawn blocked the server thread"); + const helper = nativeSpawn(command, args, options); + helpers.push(helper); + return helper; + }); + using _restore = { + [Symbol.dispose]() { + _spawn.mockRestore(); + NodeModule.syncBuiltinESMExports(); + }, + }; + NodeModule.syncBuiltinESMExports(); + const child = yield* spawner.spawn(ChildProcess.make("/bin/echo", ["responsive"])); + expect(yield* text(child.stdout)).toBe("responsive\n"); + expect(yield* child.exitCode).toBe(0); + for (let i = 0; i < 2; i++) { + yield* spawner.spawn(ChildProcess.make(process.execPath, ["-e", "process.stdin.resume()"])); + } + expect(helpers).toHaveLength(1); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); + + it.live("lets another command finish while one host is blocked inside spawn", () => + Effect.gen(function* () { + const spawned = yield* Deferred.make(); + const nativeSpawn = NodeChildProcess.spawn; + using _spawn = vi + .spyOn(NodeChildProcess, "spawn") + .mockImplementation((command, args, options) => { + const host = nativeSpawn(command, args, { + ...options, + stdio: ["inherit", "inherit", "inherit", "ipc", "pipe"], + }); + Deferred.doneUnsafe(spawned, Exit.succeed(host)); + return host; + }); + using _restore = { + [Symbol.dispose]() { + _spawn.mockRestore(); + NodeModule.syncBuiltinESMExports(); + }, + }; + NodeModule.syncBuiltinESMExports(); + const fixtureLayer = SpawnHost.layerHost(process.execPath, [ + new URL("./spawnHost.test-fixture.ts", import.meta.url).pathname, + ]).pipe(Layer.provide(NodeServices.layer)); + yield* Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const blocked = yield* spawner + .spawn(ChildProcess.make("/t3-blocked-spawn")) + .pipe(Effect.forkChild); + const host = yield* Deferred.await(spawned); + const gate = host.stdio[4] as NodeStream.Duplex; + yield* Effect.promise( + () => new Promise((resolve) => gate.once("data", () => resolve())), + ); + yield* Effect.gen(function* () { + const healthy = yield* spawner.spawn( + ChildProcess.make("/bin/echo", ["still responding"]), + ); + expect(yield* text(healthy.stdout)).toBe("still responding\n"); + expect(yield* healthy.exitCode).toBe(0); + expect(host.exitCode).toBeNull(); + }).pipe( + Effect.ensuring( + Effect.sync(() => { + gate.write("x"); + }), + ), + ); + const child = yield* Fiber.join(blocked); + expect(yield* child.exitCode).toBe(0); + }).pipe(Effect.scoped, Effect.provide(fixtureLayer)); + }), + ); + + it.live("preserves stdin, fast output, env, cwd, literal args and additional descriptors", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const literal = "quote' \" $HOME; * &"; + const script = ` + const fs = require('node:fs'); + const input = fs.readFileSync(0, 'utf8'); + fs.writeSync(1, JSON.stringify([input, process.cwd(), process.env.SPAWN_HOST_TEST, process.argv[1]]) + '\\n'); + fs.writeSync(1, 'a'.repeat(1024 * 1024)); + fs.writeSync(2, 'b'.repeat(1024 * 1024)); + fs.writeSync(3, 'fd3:' + fs.readFileSync(4, 'utf8')); + `; + for (let i = 0; i < 3; i++) { + const child = yield* spawner.spawn( + ChildProcess.make(process.execPath, ["-e", script, literal], { + cwd: "/tmp", + env: { SPAWN_HOST_TEST: "inherited value" }, + stdin: Stream.encodeText(Stream.make("stdin value")), + additionalFds: { + fd3: { type: "output" }, + fd4: { type: "input", stream: Stream.encodeText(Stream.make("fd4 value")) }, + }, + }), + ); + const [stdout, stderr, fd3, code] = yield* Effect.all( + [text(child.stdout), text(child.stderr), text(child.getOutputFd(3)), child.exitCode], + { concurrency: "unbounded" }, + ); + expect(stdout).toBe( + JSON.stringify(["stdin value", "/tmp", "inherited value", literal]) + + "\n" + + "a".repeat(1024 * 1024), + ); + expect(stderr).toBe("b".repeat(1024 * 1024)); + expect(fd3).toBe("fd3:fd4 value"); + expect(code).toBe(0); + } + const pipeline = ChildProcess.make("/bin/echo", ["piped"]).pipe( + ChildProcess.pipeTo(ChildProcess.make("/usr/bin/tr", ["a-z", "A-Z"])), + ); + const child = yield* spawner.spawn(pipeline); + expect(yield* text(child.stdout)).toBe("PIPED\n"); + expect(yield* child.exitCode).toBe(0); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); + + it.live("retains launch errors and exit codes, and can reuse a host after either", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const result = yield* spawner + .spawn(ChildProcess.make("/t3-missing-command-30183")) + .pipe(Effect.result); + expect(Result.isFailure(result)).toBe(true); + if (Result.isFailure(result)) expect(result.failure.reason._tag).toBe("NotFound"); + const invalid = yield* spawner + .spawn(ChildProcess.make("/bin/true", [], { cwd: "/dev/null" })) + .pipe(Effect.result); + expect(Result.isFailure(invalid)).toBe(true); + const child = yield* spawner.spawn(ChildProcess.make("/bin/sh", ["-c", "exit 23"])); + expect(yield* child.exitCode).toBe(23); + const healthy = yield* spawner.spawn(ChildProcess.make("/bin/true")); + expect(yield* healthy.exitCode).toBe(0); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); + + it.live("kills a hosted process group and reports the signalled exit", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + for (const detached of [true, false]) { + const child = yield* spawner.spawn( + ChildProcess.make( + process.execPath, + [ + "-e", + "process.on('SIGTERM', () => {});process.stdout.write(String(process.pid));process.stdin.resume()", + ], + { detached }, + ), + ); + const pid = yield* child.stdout.pipe(Stream.take(1), text); + expect(Number(pid)).toBe(child.pid); + yield* child.kill({ forceKillAfter: "0 millis" }); + expect(yield* child.isRunning).toBe(false); + expect(Result.isFailure(yield* child.exitCode.pipe(Effect.result))).toBe(true); + } + const started = yield* Deferred.make(); + const running = yield* Effect.gen(function* () { + const scoped = yield* spawner.spawn( + ChildProcess.make(process.execPath, ["-e", "process.stdin.resume()"]), + ); + yield* Deferred.succeed(started, scoped); + return yield* Effect.never; + }).pipe(Effect.scoped, Effect.forkChild); + const scoped = yield* Deferred.await(started); + yield* Fiber.interrupt(running); + expect(yield* scoped.isRunning).toBe(false); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); + + it.live( + "fails a command when its launcher dies and starts the next command in a fresh host", + () => + Effect.gen(function* () { + const spawned = yield* Deferred.make(); + const nativeSpawn = NodeChildProcess.spawn; + using _spawn = vi + .spyOn(NodeChildProcess, "spawn") + .mockImplementation((command, args, options) => { + const host = nativeSpawn(command, args, options); + Deferred.doneUnsafe(spawned, Exit.succeed(host)); + return host; + }); + using _restore = { + [Symbol.dispose]() { + _spawn.mockRestore(); + NodeModule.syncBuiltinESMExports(); + }, + }; + NodeModule.syncBuiltinESMExports(); + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const child = yield* spawner.spawn( + ChildProcess.make(process.execPath, [ + "-e", + "process.stdout.write('ready');process.stdin.resume()", + ]), + ); + yield* child.stdout.pipe(Stream.take(1), Stream.runDrain); + const host = yield* Deferred.await(spawned); + host.kill("SIGKILL"); + expect(Result.isFailure(yield* child.exitCode.pipe(Effect.result))).toBe(true); + expect(yield* child.isRunning).toBe(false); + const healthy = yield* spawner.spawn(ChildProcess.make("/bin/true")); + expect(yield* healthy.exitCode).toBe(0); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); + + it.live("fails launcher startup without retrying the target on the server", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const result = yield* spawner.spawn(ChildProcess.make("/bin/true")).pipe(Effect.result); + expect(Result.isFailure(result)).toBe(true); + }).pipe( + Effect.scoped, + Effect.provide( + SpawnHost.layerHost("/t3-missing-launcher-30183", []).pipe( + Layer.provide(NodeServices.layer), + ), + ), + ), + ); +}); diff --git a/apps/server/src/process/SpawnHost.ts b/apps/server/src/process/SpawnHost.ts new file mode 100644 index 0000000000000..e9c2f8e5e41a3 --- /dev/null +++ b/apps/server/src/process/SpawnHost.ts @@ -0,0 +1,300 @@ +// @effect-diagnostics nodeBuiltinImport:off globalTimers:off - Node process and IPC callbacks implement the platform spawner boundary. +import * as NodeChildProcess from "node:child_process"; +import * as NodeEvents from "node:events"; +import type * as NodeNet from "node:net"; +import type * as NodeStream from "node:stream"; +import * as NodeChildProcessSpawner from "@effect/platform-node-shared/NodeChildProcessSpawner"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { resolveSelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; + +import type { SpawnHostEvent, SpawnHostSpawnRequest } from "./spawnHostProcess.ts"; + +// A blocked native spawn occupies its host; already-running children share available hosts. +const MAX_IDLE_HOSTS = 1; +const HOST_READY_TIMEOUT_MS = 10_000; + +class HostedChild + extends NodeEvents.EventEmitter + implements NodeChildProcessSpawner.SpawnedChildProcess +{ + pid: number | undefined; + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + stdin: NodeStream.Writable | null = null; + stdout: NodeStream.Readable | null = null; + stderr: NodeStream.Readable | null = null; + readonly stdio: Array = []; + referenced = true; + signal: NodeJS.Signals | number | undefined; + exited = false; + + readonly host: SpawnHost; + readonly id: number; + + constructor(host: SpawnHost, id: number) { + super(); + this.host = host; + this.id = id; + } + + kill(signal: NodeJS.Signals | number = "SIGTERM"): boolean { + if (this.exited) return false; + if (this.pid === undefined) { + this.signal = signal; + return true; + } + try { + process.kill(this.pid, signal); + return true; + } catch { + return false; + } + } + + ref(): void { + this.referenced = true; + this.host.setReferenced(this); + } + + unref(): void { + this.referenced = false; + this.host.setReferenced(this); + } + + attach(fd: number, socket: NodeNet.Socket): void { + this.stdio[fd] = socket; + if (fd === 0) { + // Node destroys stdin on exit even after the Effect sink has removed its error listener. + socket.on("error", () => {}); + this.stdin = socket; + } else if (fd === 1) this.stdout = socket; + else if (fd === 2) this.stderr = socket; + } + + exit(code: number | null, signal: NodeJS.Signals | null): void { + this.exited = true; + this.exitCode = code; + this.signalCode = signal; + this.stdin?.destroy(); + this.emit("exit", code, signal); + process.nextTick(() => { + for (const stream of this.stdio) { + if ( + stream && + "resume" in stream && + stream.listenerCount("data") === 0 && + stream.listenerCount("readable") === 0 + ) { + stream.resume(); + } + } + }); + } +} + +class SpawnHost { + readonly process: NodeChildProcess.ChildProcess; + readonly children = new Map(); + ready = false; + closed = false; + request: SpawnHostSpawnRequest | undefined; + #nextId = 0; + readonly #readyTimer: NodeJS.Timeout; + + readonly pool: SpawnPool; + + constructor(pool: SpawnPool, command: string, args: ReadonlyArray) { + this.pool = pool; + const env = { ...process.env }; + delete env.NODE_OPTIONS; + this.process = NodeChildProcess.spawn(command, [...args], { + cwd: "/", + env, + stdio: ["inherit", "inherit", "inherit", "ipc"], + }); + this.process.unref(); + this.#readyTimer = setTimeout(() => { + this.process.kill("SIGKILL"); + this.fail(new Error("The process launcher did not start in time.")); + }, HOST_READY_TIMEOUT_MS); + this.#readyTimer.unref(); + this.process.on("message", (event, socket) => + this.onEvent(event as SpawnHostEvent, socket as NodeNet.Socket | undefined), + ); + this.process.on("error", (error) => this.fail(error)); + this.process.once("close", () => + this.fail(new Error("The process launcher exited unexpectedly.")), + ); + this.process.channel?.unref(); + } + + spawn( + command: string, + args: ReadonlyArray, + options: NodeChildProcess.SpawnOptions, + ): HostedChild { + const child = new HostedChild(this, ++this.#nextId); + this.children.set(child.id, child); + this.request = { + type: "spawn", + id: child.id, + referenced: child.referenced, + command, + args, + options: { ...options, cwd: options.cwd ?? process.cwd(), env: options.env ?? process.env }, + }; + this.syncRef(); + if (this.ready) this.send(); + return child; + } + + syncRef(): void { + if ([...this.children.values()].some((child) => child.referenced)) this.process.channel?.ref(); + else this.process.channel?.unref(); + } + + setReferenced(child: HostedChild): void { + this.syncRef(); + if (this.ready && this.children.has(child.id) && this.process.connected) { + this.process.send({ type: "ref", id: child.id, referenced: child.referenced }, (error) => { + if (error) this.fail(error); + }); + } + } + + send(): void { + this.process.send(this.request!, (error) => { + if (error) this.fail(error); + }); + } + + onEvent(event: SpawnHostEvent, socket: NodeNet.Socket | undefined): void { + if (this.closed) { + socket?.destroy(); + return; + } + if (event.type === "ready") { + clearTimeout(this.#readyTimer); + this.ready = true; + if (this.request) this.send(); + return; + } + const child = this.children.get(event.id); + if (!child) { + socket?.destroy(); + return; + } + switch (event.type) { + case "stdio": + if (socket) child.attach(event.fd, socket); + break; + case "spawn": + this.request = undefined; + child.pid = event.pid; + if (child.signal !== undefined) child.kill(child.signal); + child.emit("spawn"); + break; + case "exit": + child.exit(event.code, event.signal); + this.release(child); + break; + case "error": + if (child.pid === undefined) { + child.exited = true; + this.release(child); + } + child.emit("error", Object.assign(new Error(event.error.message), event.error)); + break; + } + } + + release(child: HostedChild): void { + this.children.delete(child.id); + if (this.request?.id === child.id) this.request = undefined; + this.syncRef(); + this.pool.release(this); + } + + fail(error: Error): void { + if (this.closed) return; + this.close(); + for (const child of this.children.values()) { + if (child.exited) continue; + if (child.pid === undefined) { + child.exited = true; + child.emit("error", error); + } else { + child.exit(null, "SIGKILL"); + } + } + } + + close(): void { + this.closed = true; + clearTimeout(this.#readyTimer); + this.pool.hosts.delete(this); + for (const child of this.children.values()) { + if (!child.referenced || child.pid === undefined || child.exited) continue; + try { + process.kill(-child.pid, "SIGKILL"); + } catch { + child.kill("SIGKILL"); + } + } + if (this.process.connected) this.process.disconnect(); + } +} + +class SpawnPool { + readonly hosts = new Set(); + + readonly command: string; + readonly args: ReadonlyArray; + + constructor(command: string, args: ReadonlyArray) { + this.command = command; + this.args = args; + } + + readonly spawn = ( + command: string, + args: ReadonlyArray, + options: NodeChildProcess.SpawnOptions, + ) => { + let host = [...this.hosts].find((candidate) => candidate.request === undefined); + if (!host) { + host = new SpawnHost(this, this.command, this.args); + this.hosts.add(host); + } + return host.spawn(command, args, options); + }; + + release(host: SpawnHost): void { + if ( + [...this.hosts].filter((candidate) => candidate.children.size === 0).length > MAX_IDLE_HOSTS + ) + host.close(); + } + + close(): void { + for (const host of this.hosts) host.close(); + } +} + +export const layerHost = (command: string, args: ReadonlyArray) => + Layer.unwrap( + Effect.acquireRelease( + Effect.sync(() => new SpawnPool(command, args)), + (pool) => Effect.sync(() => pool.close()), + ).pipe(Effect.map((pool) => NodeChildProcessSpawner.layerWithSpawn(pool.spawn))), + ); + +export const layer = Layer.unwrap( + Effect.gen(function* () { + if ((yield* HostProcess.Platform) !== "linux") return NodeChildProcessSpawner.layer; + const invocation = yield* resolveSelfInvocation(); + return layerHost(invocation.command, selfInvocationArgs(invocation, ["__spawn-host"])); + }), +); diff --git a/apps/server/src/process/spawnHost.test-fixture.ts b/apps/server/src/process/spawnHost.test-fixture.ts new file mode 100644 index 0000000000000..154caf6c8a933 --- /dev/null +++ b/apps/server/src/process/spawnHost.test-fixture.ts @@ -0,0 +1,23 @@ +// @effect-diagnostics nodeBuiltinImport:off - the fixture gates native spawn with a synchronous descriptor read. +// oxlint-disable-next-line t3code/namespace-node-imports -- The fixture replaces the native CJS export before syncing ESM bindings. +import NodeChildProcess from "node:child_process"; +import * as NodeFS from "node:fs"; +import * as NodeModule from "node:module"; +import { runSpawnHost } from "./spawnHostProcess.ts"; + +const spawn = NodeChildProcess.spawn; +const gatedSpawn = ( + command: string, + args: ReadonlyArray = [], + options: NodeChildProcess.SpawnOptions = {}, +) => { + if (command === "/t3-blocked-spawn") { + NodeFS.writeSync(4, "blocked"); + NodeFS.readSync(4, Buffer.alloc(1), 0, 1, null); + return spawn("/bin/true", [], options); + } + return spawn(command, args, options); +}; +NodeChildProcess.spawn = gatedSpawn as typeof NodeChildProcess.spawn; +NodeModule.syncBuiltinESMExports(); +runSpawnHost(); diff --git a/apps/server/src/process/spawnHostProcess.ts b/apps/server/src/process/spawnHostProcess.ts new file mode 100644 index 0000000000000..ed2204fcf19a4 --- /dev/null +++ b/apps/server/src/process/spawnHostProcess.ts @@ -0,0 +1,124 @@ +// @effect-diagnostics nodeBuiltinImport:off - the launcher deliberately loads only Node builtins. +import * as NodeChildProcess from "node:child_process"; +import type * as NodeNet from "node:net"; + +export interface SpawnHostSpawnRequest { + readonly type: "spawn"; + readonly id: number; + readonly command: string; + readonly args: ReadonlyArray; + readonly options: NodeChildProcess.SpawnOptions; + readonly referenced: boolean; +} + +export interface SpawnHostError { + readonly message: string; + readonly code?: string | undefined; + readonly errno?: number | undefined; + readonly syscall?: string | undefined; + readonly path?: string | undefined; + readonly spawnargs?: ReadonlyArray | undefined; +} + +/** Sent in order per child: `stdio` once per piped fd (carrying the socket), `spawn`, then `exit`. */ +export type SpawnHostEvent = + | { readonly type: "ready" } + | { readonly type: "stdio"; readonly id: number; readonly fd: number } + | { readonly type: "spawn"; readonly id: number; readonly pid: number } + | { + readonly type: "exit"; + readonly id: number; + readonly code: number | null; + readonly signal: NodeJS.Signals | null; + } + | { readonly type: "error"; readonly id: number; readonly error: SpawnHostError }; + +const toSpawnHostError = (cause: unknown): SpawnHostError => { + const error = cause as Partial & { spawnargs?: ReadonlyArray }; + return { + message: typeof error?.message === "string" ? error.message : String(cause), + code: error?.code, + errno: error?.errno, + syscall: error?.syscall, + path: error?.path, + spawnargs: error?.spawnargs, + }; +}; + +// Child stdio has no public paused option; use net's pauseOnCreate mechanism. +const stopReading = (socket: NodeNet.Socket) => + (socket as unknown as { _handle?: { readStop?: () => void } | null })._handle?.readStop?.(); + +export function runSpawnHost(): void { + const channel = process.send?.bind(process); + if (!channel) { + process.stderr.write("t3 __spawn-host is started by the T3 server.\n"); + process.exitCode = 1; + return; + } + const active = new Map(); + const send = (event: SpawnHostEvent, socket?: NodeNet.Socket) => channel(event, socket, () => {}); + + // Keep reporting exits during group shutdown until the server closes IPC. + for (const signal of ["SIGINT", "SIGTERM", "SIGHUP"] as const) process.on(signal, () => {}); + process.on("disconnect", () => { + for (const { child, referenced } of active.values()) { + if (child.pid === undefined || !referenced) continue; + try { + process.kill(-child.pid, "SIGKILL"); + } catch { + child.kill("SIGKILL"); + } + } + process.exit(0); + }); + + process.on( + "message", + ( + request: + | SpawnHostSpawnRequest + | { readonly type: "ref"; readonly id: number; readonly referenced: boolean }, + ) => { + if (request.type === "ref") { + const entry = active.get(request.id); + if (entry) entry.referenced = request.referenced; + return; + } + const { id } = request; + let child: NodeChildProcess.ChildProcess; + try { + child = NodeChildProcess.spawn(request.command, [...request.args], request.options); + active.set(id, { child, referenced: request.referenced }); + } catch (cause) { + send({ type: "error", id, error: toSpawnHostError(cause) }); + return; + } + // Stop reads before the loop polls, preserving every byte for the server. + const stdio = child.stdio.map((socket) => { + if (socket) stopReading(socket as NodeNet.Socket); + return socket as NodeNet.Socket | null; + }); + // Prevent Node's exit-time flush/destroy from consuming sockets still queued for IPC transfer. + child.stdin = null; + child.stdout = null; + child.stderr = null; + child.stdio.fill(null); + child.on("error", (cause) => { + if (child.pid === undefined) active.delete(id); + send({ type: "error", id, error: toSpawnHostError(cause) }); + }); + child.once("spawn", () => { + stdio.forEach((socket, fd) => { + if (socket) send({ type: "stdio", id, fd }, socket); + }); + send({ type: "spawn", id, pid: child.pid! }); + }); + child.on("exit", (code, signal) => { + active.delete(id); + send({ type: "exit", id, code, signal }); + }); + }, + ); + send({ type: "ready" }); +} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index a4433b59e8b77..6cbf5f5180d82 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -13,6 +13,7 @@ import * as NodeHttp from "node:http"; import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as SpawnHost from "./process/SpawnHost.ts"; import { EnvironmentHttpApi, type RepositoryIdentity } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Duration from "effect/Duration"; @@ -374,7 +375,13 @@ const layerHttpServer = Layer.unwrap( ).pipe(Layer.provide(ManagedPolicy.layer)); const layerPlatformServices = DocumentText.layer.pipe( - Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + Layer.provideMerge( + Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + SpawnHost.layer.pipe(Layer.provide(NodeServices.layer)), + ), + ), ); const layerPersistence = Layer.empty.pipe(Layer.provideMerge(SqlitePersistence.layerConfig)); diff --git a/patches/@effect__platform-node-shared@4.1.2.patch b/patches/@effect__platform-node-shared@4.1.2.patch new file mode 100644 index 0000000000000..f676127a3ecf7 --- /dev/null +++ b/patches/@effect__platform-node-shared@4.1.2.patch @@ -0,0 +1,65 @@ +diff --git a/dist/NodeChildProcessSpawner.d.ts b/dist/NodeChildProcessSpawner.d.ts +index 213be4fa2647d800b320fa483c9ede801366fd73..bb7a2bb97e976f85af4760a7299db39506318155 100644 +--- a/dist/NodeChildProcessSpawner.d.ts ++++ b/dist/NodeChildProcessSpawner.d.ts +@@ -31,6 +31,8 @@ + * @stability unstable + * @since 4.0.0 + */ ++import type * as NodeChildProcess from "node:child_process"; ++import type { Readable, Writable } from "node:stream"; + import type * as Arr from "effect/Array"; + import * as FileSystem from "effect/FileSystem"; + import * as Layer from "effect/Layer"; +@@ -45,6 +47,19 @@ import { ChildProcessSpawner } from "effect/process/ChildProcessSpawner"; + * @since 4.0.0 + */ + export declare const layer: Layer.Layer; ++export interface SpawnedChildProcess extends NodeJS.EventEmitter { ++ readonly pid?: number | undefined; ++ readonly exitCode: number | null; ++ readonly signalCode: NodeJS.Signals | null; ++ readonly stdin: Writable | null; ++ readonly stdout: Readable | null; ++ readonly stderr: Readable | null; ++ readonly stdio: ReadonlyArray; ++ kill(signal?: NodeJS.Signals | number): boolean; ++ ref(): void; ++ unref(): void; ++} ++export declare const layerWithSpawn: (spawnProcess: (command: string, args: ReadonlyArray, options: NodeChildProcess.SpawnOptions) => SpawnedChildProcess) => Layer.Layer; + /** + * Result of flattening a pipeline of commands. + * +diff --git a/dist/NodeChildProcessSpawner.js b/dist/NodeChildProcessSpawner.js +index 08d50b16dfe564551f1bffc7cae2d3b53805e172..6760a0d0c48f30b1a9aa8bbfdf6b9c6db9c07888 100644 +--- a/dist/NodeChildProcessSpawner.js ++++ b/dist/NodeChildProcessSpawner.js +@@ -90,7 +90,7 @@ const taskkill = (childProcess, onExit = () => {}) => { + windowsHide: true + }, onExit); + }; +-const make = /*#__PURE__*/Effect.gen(function* () { ++const makeWithSpawn = spawnProcess => Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const resolveWorkingDirectory = Effect.fnUntraced(function* (options) { +@@ -335,7 +335,7 @@ const make = /*#__PURE__*/Effect.gen(function* () { + }; + const spawn = (command, spawnOptions) => Effect.callback(resume => { + const deferred = Deferred.makeUnsafe(); +- const handle = NodeChildProcess.spawn(command.command, command.args.map(unwrap), spawnOptions); ++ const handle = spawnProcess(command.command, command.args.map(unwrap), spawnOptions); + handle.on("error", error => { + // Node copies the arguments into `spawnargs`. + if (command.args.some(Redacted.isRedacted)) { +@@ -665,7 +665,8 @@ const make = /*#__PURE__*/Effect.gen(function* () { + * @category layers + * @since 4.0.0 + */ +-export const layer = /*#__PURE__*/Layer.effect(ChildProcessSpawner, make); ++export const layer = /*#__PURE__*/Layer.effect(ChildProcessSpawner, /*#__PURE__*/makeWithSpawn(NodeChildProcess.spawn)); ++export const layerWithSpawn = spawnProcess => Layer.effect(ChildProcessSpawner, makeWithSpawn(spawnProcess)); + /** + * Flattens a `Command` into an array of `StandardCommand`s along with pipe + * options for each connection. diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 755f2f95c2ed1..b3120c0acd639 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -365,6 +365,7 @@ pnpmfileChecksum: sha256-STsbxG7+RxOFI611PrDEj1TonJgqKKllY9hYCFMLYiU= patchedDependencies: '@alchemy.run/test@2.0.0-beta.85': 29c0539ea71d9cc372f0b16558640599beb9bc609605feb2a493da268d72713e '@clerk/expo-native-components@0.4.0': 7a808b92da8460750b85818552e13be3fb73d111691c7db76965970351e124e4 + '@effect/platform-node-shared@4.1.2': 285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e '@effect/vitest@4.1.2': 299510b71c36537a832a67e9d160a5f112a5f326a29e95c41cb45fc8234e47c7 '@expo/fingerprint@0.21.5': 8fba80d6bbce46c72de427fe8c7907462d262a0bfe50c6b2b7ee9c4e0c61e5e1 '@expo/metro-config@58.0.14': 90fa980e8b6480631fb2fbb04559834e3c612db40c550cd828f11be297d8d760 @@ -844,7 +845,7 @@ importers: version: 4.1.2(bufferutil@4.1.0)(effect@4.1.2)(redis@6.2.1)(utf-8-validate@6.0.6) '@effect/platform-node-shared': specifier: 4.1.2 - version: 4.1.2(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) + version: 4.1.2(patch_hash=285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) '@ff-labs/fff-node': specifier: 0.11.0 version: 0.11.0(patch_hash=cc71fcec015f4d39bc96ae2e2bf6b0ca678124c722c66a7841efce9d34b5983c) @@ -13732,13 +13733,13 @@ snapshots: '@effect/platform-bun@4.1.2(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6)': dependencies: - '@effect/platform-node-shared': 4.1.2(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) + '@effect/platform-node-shared': 4.1.2(patch_hash=285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) effect: 4.1.2 transitivePeerDependencies: - bufferutil - utf-8-validate - '@effect/platform-node-shared@4.1.2(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6)': + '@effect/platform-node-shared@4.1.2(patch_hash=285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6)': dependencies: '@types/ws': 8.18.2 effect: 4.1.2 @@ -13749,7 +13750,7 @@ snapshots: '@effect/platform-node@4.1.2(bufferutil@4.1.0)(effect@4.1.2)(redis@6.2.1)(utf-8-validate@6.0.6)': dependencies: - '@effect/platform-node-shared': 4.1.2(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) + '@effect/platform-node-shared': 4.1.2(patch_hash=285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) effect: 4.1.2 redis: 6.2.1 undici: 8.11.2 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index df77b847c3418..2bb3f23082613 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -409,6 +409,7 @@ patchedDependencies: # Observe the owned host exit independently of descendants retaining its stdio. "@muse-code/sdk@1.4.6": patches/@muse-code__sdk@1.4.6.patch app-builder-lib@26.16.1: patches/app-builder-lib@26.16.1.patch + "@effect/platform-node-shared@4.1.2": patches/@effect__platform-node-shared@4.1.2.patch peerDependencyRules: allowAny: From 84356c079882c9dceeee2aa480b1a4d273cebbe2 Mon Sep 17 00:00:00 2001 From: Lakshmi Tanmay Date: Tue, 3 Nov 2026 20:04:00 +0530 Subject: [PATCH 02/14] fix(server): keep hosted commands alive when another launch fails --- apps/server/src/process/SpawnHost.test.ts | 54 +++++++++++++++++++ .../src/process/spawnHost.test-fixture.ts | 15 ++++++ apps/server/src/process/spawnHostProcess.ts | 21 ++++---- 3 files changed, 81 insertions(+), 9 deletions(-) diff --git a/apps/server/src/process/SpawnHost.test.ts b/apps/server/src/process/SpawnHost.test.ts index 800303ee74e33..1ef0ce7912ea8 100644 --- a/apps/server/src/process/SpawnHost.test.ts +++ b/apps/server/src/process/SpawnHost.test.ts @@ -170,6 +170,29 @@ describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("SpawnHost", () .spawn(ChildProcess.make("/bin/true", [], { cwd: "/dev/null" })) .pipe(Effect.result); expect(Result.isFailure(invalid)).toBe(true); + const fdCounts: number[] = []; + for (let pass = 0; pass < 2; pass++) { + if (pass === 1) { + for (let i = 0; i < 12; i++) { + for (const command of ["/t3-missing-command-30183", "/dev/null"]) { + expect( + Result.isFailure( + yield* spawner.spawn(ChildProcess.make(command)).pipe(Effect.result), + ), + ).toBe(true); + } + } + } + const probe = yield* spawner.spawn( + ChildProcess.make(process.execPath, [ + "-e", + "process.stdout.write(String(require('node:fs').readdirSync('/proc/' + process.ppid + '/fd').length))", + ]), + ); + fdCounts.push(Number(yield* text(probe.stdout))); + expect(yield* probe.exitCode).toBe(0); + } + expect(fdCounts[1]).toBe(fdCounts[0]); const child = yield* spawner.spawn(ChildProcess.make("/bin/sh", ["-c", "exit 23"])); expect(yield* child.exitCode).toBe(23); const healthy = yield* spawner.spawn(ChildProcess.make("/bin/true")); @@ -248,6 +271,37 @@ describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("SpawnHost", () }).pipe(Effect.scoped, Effect.provide(layer)), ); + it.live("reports descriptor exhaustion without stopping another command in the same host", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const running = yield* spawner.spawn( + ChildProcess.make(process.execPath, [ + "-e", + "process.stdout.write('ready');process.stdin.resume()", + ]), + ); + yield* running.stdout.pipe(Stream.take(1), Stream.runDrain); + const failed = yield* spawner + .spawn(ChildProcess.make("/t3-exhausted-spawn")) + .pipe(Effect.result); + expect(Result.isFailure(failed)).toBe(true); + expect(yield* running.isRunning).toBe(true); + const healthy = yield* spawner.spawn(ChildProcess.make("/bin/true")); + expect(yield* healthy.exitCode).toBe(0); + }).pipe( + Effect.scoped, + Effect.provide( + SpawnHost.layerHost("/bin/sh", [ + "-c", + 'ulimit -n 64; exec "$@"', + "sh", + process.execPath, + new URL("./spawnHost.test-fixture.ts", import.meta.url).pathname, + ]).pipe(Layer.provide(NodeServices.layer)), + ), + ), + ); + it.live("fails launcher startup without retrying the target on the server", () => Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; diff --git a/apps/server/src/process/spawnHost.test-fixture.ts b/apps/server/src/process/spawnHost.test-fixture.ts index 154caf6c8a933..3dd875a03ff19 100644 --- a/apps/server/src/process/spawnHost.test-fixture.ts +++ b/apps/server/src/process/spawnHost.test-fixture.ts @@ -16,6 +16,21 @@ const gatedSpawn = ( NodeFS.readSync(4, Buffer.alloc(1), 0, 1, null); return spawn("/bin/true", [], options); } + if (command === "/t3-exhausted-spawn") { + const fds: number[] = []; + try { + for (let i = 0; i < 128; i++) { + try { + fds.push(NodeFS.openSync("/dev/null", "r")); + } catch { + break; + } + } + return spawn("/bin/true", [], options); + } finally { + for (const fd of fds) NodeFS.closeSync(fd); + } + } return spawn(command, args, options); }; NodeChildProcess.spawn = gatedSpawn as typeof NodeChildProcess.spawn; diff --git a/apps/server/src/process/spawnHostProcess.ts b/apps/server/src/process/spawnHostProcess.ts index ed2204fcf19a4..a7ae7ec74d2a9 100644 --- a/apps/server/src/process/spawnHostProcess.ts +++ b/apps/server/src/process/spawnHostProcess.ts @@ -94,20 +94,23 @@ export function runSpawnHost(): void { send({ type: "error", id, error: toSpawnHostError(cause) }); return; } - // Stop reads before the loop polls, preserving every byte for the server. - const stdio = child.stdio.map((socket) => { - if (socket) stopReading(socket as NodeNet.Socket); - return socket as NodeNet.Socket | null; + const stdio = (child.stdio ?? []).map((socket) => socket as NodeNet.Socket | null); + child.on("error", (cause) => { + if (child.pid === undefined) { + active.delete(id); + for (const socket of stdio) socket?.destroy(); + } + send({ type: "error", id, error: toSpawnHostError(cause) }); }); + // Stop reads before the loop polls, preserving every byte for the server. + for (const socket of stdio) { + if (socket) stopReading(socket); + } // Prevent Node's exit-time flush/destroy from consuming sockets still queued for IPC transfer. child.stdin = null; child.stdout = null; child.stderr = null; - child.stdio.fill(null); - child.on("error", (cause) => { - if (child.pid === undefined) active.delete(id); - send({ type: "error", id, error: toSpawnHostError(cause) }); - }); + child.stdio?.fill(null); child.once("spawn", () => { stdio.forEach((socket, fd) => { if (socket) send({ type: "stdio", id, fd }, socket); From 9642d50303dc79c4f0818b2a34ee8712bd15a252 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 4 Nov 2026 02:11:15 +0000 Subject: [PATCH 03/14] test(server): decode spawn helper file paths --- apps/server/src/process/SpawnHost.test.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/apps/server/src/process/SpawnHost.test.ts b/apps/server/src/process/SpawnHost.test.ts index 1ef0ce7912ea8..f53a31dcf631c 100644 --- a/apps/server/src/process/SpawnHost.test.ts +++ b/apps/server/src/process/SpawnHost.test.ts @@ -3,6 +3,7 @@ import NodeChildProcess from "node:child_process"; import * as NodeModule from "node:module"; import type * as NodeStream from "node:stream"; +import * as NodeURL from "node:url"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import { vi } from "vite-plus/test"; @@ -20,7 +21,7 @@ import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as SpawnHost from "./SpawnHost.ts"; const layer = SpawnHost.layerHost(process.execPath, [ - new URL("../bin.ts", import.meta.url).pathname, + NodeURL.fileURLToPath(new URL("../bin.ts", import.meta.url)), "__spawn-host", ]).pipe(Layer.provide(NodeServices.layer)); const text = (stream: Stream.Stream) => @@ -80,7 +81,7 @@ describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("SpawnHost", () }; NodeModule.syncBuiltinESMExports(); const fixtureLayer = SpawnHost.layerHost(process.execPath, [ - new URL("./spawnHost.test-fixture.ts", import.meta.url).pathname, + NodeURL.fileURLToPath(new URL("./spawnHost.test-fixture.ts", import.meta.url)), ]).pipe(Layer.provide(NodeServices.layer)); yield* Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; @@ -296,7 +297,7 @@ describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("SpawnHost", () 'ulimit -n 64; exec "$@"', "sh", process.execPath, - new URL("./spawnHost.test-fixture.ts", import.meta.url).pathname, + NodeURL.fileURLToPath(new URL("./spawnHost.test-fixture.ts", import.meta.url)), ]).pipe(Layer.provide(NodeServices.layer)), ), ), From a9f555d012256f2ad6acf472913f907a717c09b4 Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Fri, 6 Nov 2026 19:47:15 -0500 Subject: [PATCH 04/14] perf(server): Windows process launches no longer block the server thread Node's spawn() runs CreateProcess on the calling thread. On Windows, antivirus scans and a busy disk can hold that call for seconds, and every git, gh and provider launch froze RPC and heartbeats for that long (#25245). On Windows the shared Effect spawner now launches through a small pool of worker threads. Each worker owns its children; the server thread gets stand-in handles that relay stdio, exit and kill. Effect's taskkill goes through the same path. --- apps/server/src/process/SpawnHost.ts | 5 +- apps/server/src/process/SpawnWorker.test.ts | 132 ++++++ apps/server/src/process/SpawnWorker.ts | 442 ++++++++++++++++++ .../@effect__platform-node-shared@4.1.2.patch | 40 +- pnpm-lock.yaml | 10 +- 5 files changed, 617 insertions(+), 12 deletions(-) create mode 100644 apps/server/src/process/SpawnWorker.test.ts create mode 100644 apps/server/src/process/SpawnWorker.ts diff --git a/apps/server/src/process/SpawnHost.ts b/apps/server/src/process/SpawnHost.ts index e9c2f8e5e41a3..ada20d1431067 100644 --- a/apps/server/src/process/SpawnHost.ts +++ b/apps/server/src/process/SpawnHost.ts @@ -10,6 +10,7 @@ import * as HostProcess from "@t3tools/shared/HostProcess"; import { resolveSelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; import type { SpawnHostEvent, SpawnHostSpawnRequest } from "./spawnHostProcess.ts"; +import * as SpawnWorker from "./SpawnWorker.ts"; // A blocked native spawn occupies its host; already-running children share available hosts. const MAX_IDLE_HOSTS = 1; @@ -293,7 +294,9 @@ export const layerHost = (command: string, args: ReadonlyArray) => export const layer = Layer.unwrap( Effect.gen(function* () { - if ((yield* HostProcess.Platform) !== "linux") return NodeChildProcessSpawner.layer; + const platform = yield* HostProcess.Platform; + if (platform === "win32") return SpawnWorker.layer; + if (platform !== "linux") return NodeChildProcessSpawner.layer; const invocation = yield* resolveSelfInvocation(); return layerHost(invocation.command, selfInvocationArgs(invocation, ["__spawn-host"])); }), diff --git a/apps/server/src/process/SpawnWorker.test.ts b/apps/server/src/process/SpawnWorker.test.ts new file mode 100644 index 0000000000000..1017189ae2f62 --- /dev/null +++ b/apps/server/src/process/SpawnWorker.test.ts @@ -0,0 +1,132 @@ +// @effect-diagnostics nodeBuiltinImport:off - fixtures need real paths and the process resource list. +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Result from "effect/Result"; +import * as Stream from "effect/Stream"; +import * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; + +import * as SpawnWorker from "./SpawnWorker.ts"; + +const layer = SpawnWorker.layer.pipe(Layer.provide(NodeServices.layer)); +const text = (stream: Stream.Stream) => + stream.pipe(Stream.decodeText(), Stream.mkString); +const node = (script: string, args: ReadonlyArray = [], options = {}) => + ChildProcess.make(process.execPath, ["-e", script, ...args], options); + +describe("SpawnWorker", () => { + it.live("runs children on a launcher thread, not the server thread", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const child = yield* spawner.spawn( + node("process.stdout.write('ready');process.stdin.resume()"), + ); + expect(yield* child.stdout.pipe(Stream.take(1), text)).toBe("ready"); + // The thread that called spawn owns the native process handle. + expect(process.getActiveResourcesInfo()).not.toContain("ProcessWrap"); + expect(yield* child.isRunning).toBe(true); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); + + it.live("preserves stdin, large output, env, cwd, literal args and additional descriptors", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const literal = "quote' \" $HOME; * & %PATH%"; + const cwd = NodeOS.tmpdir(); + const script = ` + const fs = require('node:fs'); + const input = fs.readFileSync(0, 'utf8'); + fs.writeSync(1, JSON.stringify([input, process.cwd(), process.env.SPAWN_WORKER_TEST, process.argv[1]]) + '\\n'); + fs.writeSync(1, 'a'.repeat(1024 * 1024)); + fs.writeSync(2, 'b'.repeat(1024 * 1024)); + `; + for (let i = 0; i < 3; i++) { + const child = yield* spawner.spawn( + node(script, [literal], { + cwd, + env: { SPAWN_WORKER_TEST: "inherited value" }, + extendEnv: true, + stdin: Stream.encodeText(Stream.make("stdin value")), + }), + ); + const [stdout, stderr, code] = yield* Effect.all( + [text(child.stdout), text(child.stderr), child.exitCode], + { concurrency: "unbounded" }, + ); + expect(stdout).toBe( + JSON.stringify(["stdin value", NodePath.resolve(cwd), "inherited value", literal]) + + "\n" + + "a".repeat(1024 * 1024), + ); + expect(stderr).toBe("b".repeat(1024 * 1024)); + expect(code).toBe(0); + } + const extra = yield* spawner.spawn( + node("require('node:fs').writeSync(3, 'fd3 value')", [], { + additionalFds: { fd3: { type: "output" } }, + }), + ); + const [fd3, extraCode] = yield* Effect.all([text(extra.getOutputFd(3)), extra.exitCode], { + concurrency: "unbounded", + }); + expect(fd3).toBe("fd3 value"); + expect(extraCode).toBe(0); + const pipeline = node("process.stdout.write('piped')").pipe( + ChildProcess.pipeTo( + node("process.stdin.on('data', (d) => process.stdout.write(String(d).toUpperCase()))"), + ), + ); + const piped = yield* spawner.spawn(pipeline); + expect(yield* text(piped.stdout)).toBe("PIPED"); + expect(yield* piped.exitCode).toBe(0); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); + + it.live("reports launch errors and exit codes, and keeps launching after either", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const missing = yield* spawner + .spawn(ChildProcess.make("t3-missing-command-25245")) + .pipe(Effect.result); + expect(Result.isFailure(missing)).toBe(true); + if (Result.isFailure(missing)) expect(missing.failure.reason._tag).toBe("NotFound"); + const failed = yield* spawner.spawn(node("process.exit(23)")); + expect(yield* failed.exitCode).toBe(23); + const healthy = yield* spawner.spawn(node("process.stdout.write('ok')")); + expect(yield* text(healthy.stdout)).toBe("ok"); + expect(yield* healthy.exitCode).toBe(0); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); + + it.live("kills running commands, including process trees and on scope interruption", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + for (const detached of [true, false]) { + const child = yield* spawner.spawn( + node("process.stdout.write(String(process.pid));process.stdin.resume()", [], { + detached, + }), + ); + const pid = yield* child.stdout.pipe(Stream.take(1), text); + expect(Number(pid)).toBe(child.pid); + yield* child.kill(); + expect(yield* child.isRunning).toBe(false); + } + const started = yield* Deferred.make(); + const running = yield* Effect.gen(function* () { + const scoped = yield* spawner.spawn(node("process.stdin.resume()")); + yield* Deferred.succeed(started, scoped); + return yield* Effect.never; + }).pipe(Effect.scoped, Effect.forkChild); + const scoped = yield* Deferred.await(started); + yield* Fiber.interrupt(running); + expect(yield* scoped.isRunning).toBe(false); + }).pipe(Effect.scoped, Effect.provide(layer)), + ); +}); diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts new file mode 100644 index 0000000000000..da989e4459659 --- /dev/null +++ b/apps/server/src/process/SpawnWorker.ts @@ -0,0 +1,442 @@ +// @effect-diagnostics nodeBuiltinImport:off - worker threads and Node streams implement the platform spawner boundary. +import * as NodeEvents from "node:events"; +import type * as NodeChildProcess from "node:child_process"; +import * as NodeStream from "node:stream"; +import * as NodeWorkerThreads from "node:worker_threads"; +import * as NodeChildProcessSpawner from "@effect/platform-node-shared/NodeChildProcessSpawner"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +// Windows runs CreateProcess synchronously on the calling thread, and antivirus or a busy disk can +// hold it for seconds. Workers own the children, so only the worker waits. +const MAX_WORKERS = 4; +const MAX_IDLE_WORKERS = 1; + +type ToWorker = + | { + readonly type: "spawn"; + readonly id: number; + readonly command: string; + readonly args: ReadonlyArray; + readonly options: NodeChildProcess.SpawnOptions; + } + | { readonly type: "write"; readonly id: number; readonly fd: number; readonly data: Uint8Array } + | { readonly type: "end"; readonly id: number; readonly fd: number } + | { readonly type: "flow"; readonly id: number; readonly fd: number; readonly paused: boolean } + | { readonly type: "kill"; readonly id: number; readonly signal: NodeJS.Signals | number }; + +interface SerializedError { + readonly message: string; + readonly [key: string]: unknown; +} + +type FromWorker = + | { readonly type: "spawn"; readonly id: number; readonly pid: number | undefined } + | { readonly type: "error"; readonly id: number; readonly error: SerializedError } + | { readonly type: "data"; readonly id: number; readonly fd: number; readonly data: Uint8Array } + | { + readonly type: "written"; + readonly id: number; + readonly fd: number; + readonly error?: SerializedError; + } + | { readonly type: "eof"; readonly id: number; readonly fd: number } + | { readonly type: "finished"; readonly id: number; readonly fd: number } + | { + readonly type: "stream-error"; + readonly id: number; + readonly fd: number; + readonly error: SerializedError; + } + | { + readonly type: "exit"; + readonly id: number; + readonly code: number | null; + readonly signal: NodeJS.Signals | null; + } + | { readonly type: "close"; readonly id: number }; + +// Inline source runs from bundles, asar archives and single-executable builds without module files. +const workerSource = ` +const { parentPort } = require("node:worker_threads"); +const { spawn } = require("node:child_process"); +const children = new Map(); +const post = (message, transfer) => parentPort.postMessage(message, transfer); +const serialize = (error) => { + const out = { message: String(error && error.message) }; + for (const key of ["code", "errno", "syscall", "path", "spawnargs"]) { + if (error && error[key] !== undefined) out[key] = error[key]; + } + return out; +}; +parentPort.on("message", (request) => { + const { id } = request; + if (request.type === "spawn") { + let child; + try { + child = spawn(request.command, request.args, request.options); + } catch (error) { + post({ type: "error", id, error: serialize(error) }); + return; + } + children.set(id, child); + child.on("spawn", () => post({ type: "spawn", id, pid: child.pid })); + child.on("error", (error) => post({ type: "error", id, error: serialize(error) })); + child.on("exit", (code, signal) => post({ type: "exit", id, code, signal })); + // Output can still arrive after exit; close follows the last of it. + child.on("close", () => { + children.delete(id); + post({ type: "close", id }); + }); + child.stdio.forEach((stream, fd) => { + if (!stream) return; + stream.on("error", (error) => post({ type: "stream-error", id, fd, error: serialize(error) })); + if (fd === 0) return; + // Cloning a Buffer copies its whole backing store; copy the bytes once and transfer them. + stream.on("data", (chunk) => { + const data = new Uint8Array(chunk); + post({ type: "data", id, fd, data }, [data.buffer]); + }); + stream.on("end", () => post({ type: "eof", id, fd })); + }); + return; + } + const child = children.get(id); + if (child === undefined) { + if (request.type === "write") post({ type: "written", id, fd: request.fd, error: { message: "write after exit", code: "EPIPE" } }); + if (request.type === "end") post({ type: "finished", id, fd: request.fd }); + return; + } + const stream = child.stdio[request.fd]; + switch (request.type) { + case "write": + if (!stream || stream.destroyed) { + post({ type: "written", id, fd: request.fd, error: { message: "write after end", code: "EPIPE" } }); + return; + } + stream.write(request.data, (error) => post({ type: "written", id, fd: request.fd, error: error ? serialize(error) : undefined })); + return; + case "end": + if (!stream || stream.destroyed || stream.writableFinished) post({ type: "finished", id, fd: request.fd }); + else stream.end(() => post({ type: "finished", id, fd: request.fd })); + return; + case "flow": + if (request.paused) stream?.pause(); + else stream?.resume(); + return; + case "kill": + child.kill(request.signal); + return; + } +}); +`; + +const toError = (serialized: SerializedError) => + Object.assign(new Error(serialized.message), serialized); + +const toBuffer = (data: Uint8Array) => Buffer.from(data.buffer, data.byteOffset, data.byteLength); + +class WorkerChild + extends NodeEvents.EventEmitter + implements NodeChildProcessSpawner.SpawnedChildProcess +{ + pid: number | undefined; + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + stdin: NodeStream.Writable | null = null; + stdout: NodeStream.Readable | null = null; + stderr: NodeStream.Readable | null = null; + readonly stdio: Array = []; + referenced = true; + spawned = false; + exited = false; + readonly #writes = new Map void>>(); + readonly #finals = new Map void>(); + readonly #paused = new Set(); + + readonly worker: SpawnWorker; + readonly id: number; + + constructor(worker: SpawnWorker, id: number, stdio: ReadonlyArray) { + super(); + this.worker = worker; + this.id = id; + stdio.forEach((option, fd) => { + if (option !== "pipe" && option !== "overlapped") return; + const stream = this.#relay(fd); + this.stdio[fd] = stream; + if (fd === 0) this.stdin = stream as NodeStream.Writable; + else if (fd === 1) this.stdout = stream as NodeStream.Readable; + else if (fd === 2) this.stderr = stream as NodeStream.Readable; + }); + } + + #relay(fd: number): NodeStream.Readable | NodeStream.Writable { + const read = () => { + if (this.#paused.delete(fd)) + this.worker.post({ type: "flow", id: this.id, fd, paused: false }); + }; + const write = ( + chunk: Buffer, + _encoding: BufferEncoding, + callback: (error?: Error | null) => void, + ) => { + const pending = this.#writes.get(fd) ?? []; + pending.push(callback); + this.#writes.set(fd, pending); + const data = new Uint8Array(chunk); + this.worker.post({ type: "write", id: this.id, fd, data }, [data.buffer]); + }; + const final = (callback: (error?: Error | null) => void) => { + this.#finals.set(fd, callback); + this.worker.post({ type: "end", id: this.id, fd }); + }; + if (fd === 0) return new NodeStream.Writable({ write, final }); + if (fd <= 2) return new NodeStream.Readable({ read }); + return new NodeStream.Duplex({ read, write, final, allowHalfOpen: true }); + } + + onData(fd: number, data: Uint8Array): void { + const stream = this.stdio[fd]; + if (!stream || !("push" in stream)) return; + if (!stream.push(toBuffer(data)) && !this.#paused.has(fd)) { + this.#paused.add(fd); + this.worker.post({ type: "flow", id: this.id, fd, paused: true }); + } + } + + onEof(fd: number): void { + const stream = this.stdio[fd]; + if (stream && "push" in stream) stream.push(null); + } + + onFinished(fd: number): void { + const callback = this.#finals.get(fd); + this.#finals.delete(fd); + callback?.(); + } + + onWritten(fd: number, error: SerializedError | undefined): void { + const callback = this.#writes.get(fd)?.shift(); + callback?.(error === undefined ? null : toError(error)); + } + + onStreamError(fd: number, error: SerializedError): void { + this.stdio[fd]?.destroy(toError(error)); + } + + kill(signal: NodeJS.Signals | number = "SIGTERM"): boolean { + if (this.exited) return false; + this.worker.post({ type: "kill", id: this.id, signal }); + return true; + } + + ref(): void { + this.referenced = true; + this.worker.syncRef(); + } + + unref(): void { + this.referenced = false; + this.worker.syncRef(); + } + + exit(code: number | null, signal: NodeJS.Signals | null): void { + if (this.exited) return; + this.exited = true; + this.exitCode = code; + this.signalCode = signal; + // Node destroys stdin on exit; the Effect sink may already have removed its error listener. + this.stdin?.on("error", () => {}); + this.stdin?.destroy(); + this.emit("exit", code, signal); + } + + /** The worker is gone, so no further output or acknowledgement can arrive. */ + abandon(): void { + for (const stream of this.stdio) { + stream?.on("error", () => {}); + stream?.destroy(); + } + } +} + +class SpawnWorker { + readonly worker: NodeWorkerThreads.Worker; + readonly children = new Map(); + readonly pool: SpawnWorkerPool; + /** Launches sent but not yet answered; the worker is inside CreateProcess for one of them. */ + launching = 0; + closed = false; + + constructor(pool: SpawnWorkerPool) { + this.pool = pool; + this.worker = new NodeWorkerThreads.Worker(workerSource, { eval: true }); + this.worker.on("message", (message: FromWorker) => this.onMessage(message)); + this.worker.on("error", (error) => this.fail(error)); + this.worker.on("exit", () => this.fail(new Error("The process launcher thread exited."))); + this.worker.unref(); + } + + post(message: ToWorker, transfer?: ReadonlyArray): void { + if (!this.closed) this.worker.postMessage(message, transfer); + } + + spawn( + id: number, + command: string, + args: ReadonlyArray, + options: NodeChildProcess.SpawnOptions, + ): WorkerChild { + const io = options.stdio; + const stdio = + typeof io === "string" || io === undefined ? [io ?? "pipe", io ?? "pipe", io ?? "pipe"] : io; + const child = new WorkerChild(this, id, stdio); + this.children.set(id, child); + this.launching++; + this.syncRef(); + this.post({ + type: "spawn", + id, + command, + args: [...args], + options: { ...options, stdio }, + }); + return child; + } + + syncRef(): void { + if (this.closed) return; + const referenced = + this.launching > 0 || [...this.children.values()].some((child) => child.referenced); + if (referenced) this.worker.ref(); + else this.worker.unref(); + } + + onMessage(message: FromWorker): void { + const child = this.children.get(message.id); + if (child === undefined) return; + switch (message.type) { + case "spawn": + this.launching--; + child.spawned = true; + child.pid = message.pid; + child.emit("spawn"); + break; + case "error": + if (!child.spawned) { + this.launching--; + child.exited = true; + this.release(child); + } + child.emit("error", toError(message.error)); + break; + case "data": + child.onData(message.fd, message.data); + break; + case "written": + child.onWritten(message.fd, message.error); + break; + case "eof": + child.onEof(message.fd); + break; + case "finished": + child.onFinished(message.fd); + break; + case "stream-error": + child.onStreamError(message.fd, message.error); + break; + case "exit": + child.exit(message.code, message.signal); + break; + case "close": + this.release(child); + break; + } + } + + release(child: WorkerChild): void { + this.children.delete(child.id); + this.syncRef(); + this.pool.release(this); + } + + fail(error: Error): void { + if (this.closed) return; + this.close(); + for (const child of this.children.values()) { + if (child.exited) { + child.abandon(); + continue; + } + if (!child.spawned) { + child.exited = true; + child.emit("error", error); + continue; + } + // The worker owned the only handle; best-effort so a crash leaves no orphans. + if (child.pid !== undefined) { + try { + process.kill(child.pid, "SIGKILL"); + } catch {} + } + child.exit(null, "SIGKILL"); + child.abandon(); + } + this.children.clear(); + } + + close(): void { + if (this.closed) return; + this.closed = true; + this.pool.workers.delete(this); + void this.worker.terminate(); + } +} + +class SpawnWorkerPool { + readonly workers = new Set(); + #nextId = 0; + + readonly spawn = ( + command: string, + args: ReadonlyArray, + options: NodeChildProcess.SpawnOptions, + ) => { + const workers = [...this.workers]; + // A worker inside CreateProcess cannot start another launch until it returns. + const worker = + workers.find((candidate) => candidate.launching === 0) ?? + (workers.length < MAX_WORKERS + ? this.#add() + : workers.reduce((best, candidate) => + candidate.launching < best.launching ? candidate : best, + )); + return worker.spawn(++this.#nextId, command, args, options); + }; + + #add(): SpawnWorker { + const worker = new SpawnWorker(this); + this.workers.add(worker); + return worker; + } + + release(worker: SpawnWorker): void { + if (worker.children.size > 0 || worker.launching > 0) return; + const idle = [...this.workers].filter( + (candidate) => candidate.children.size === 0 && candidate.launching === 0, + ); + if (idle.length > MAX_IDLE_WORKERS) worker.close(); + } + + close(): void { + for (const worker of this.workers) worker.close(); + } +} + +export const layer = Layer.unwrap( + Effect.acquireRelease( + Effect.sync(() => new SpawnWorkerPool()), + (pool) => Effect.sync(() => pool.close()), + ).pipe(Effect.map((pool) => NodeChildProcessSpawner.layerWithSpawn(pool.spawn))), +); diff --git a/patches/@effect__platform-node-shared@4.1.2.patch b/patches/@effect__platform-node-shared@4.1.2.patch index f676127a3ecf7..6e957eac98f41 100644 --- a/patches/@effect__platform-node-shared@4.1.2.patch +++ b/patches/@effect__platform-node-shared@4.1.2.patch @@ -32,19 +32,47 @@ index 213be4fa2647d800b320fa483c9ede801366fd73..bb7a2bb97e976f85af4760a7299db395 * Result of flattening a pipeline of commands. * diff --git a/dist/NodeChildProcessSpawner.js b/dist/NodeChildProcessSpawner.js -index 08d50b16dfe564551f1bffc7cae2d3b53805e172..6760a0d0c48f30b1a9aa8bbfdf6b9c6db9c07888 100644 +index 08d50b16dfe564551f1bffc7cae2d3b53805e172..fbecacc8b39a9eab7725a799aa5019c36bcc32ab 100644 --- a/dist/NodeChildProcessSpawner.js +++ b/dist/NodeChildProcessSpawner.js -@@ -90,7 +90,7 @@ const taskkill = (childProcess, onExit = () => {}) => { - windowsHide: true - }, onExit); +@@ -81,16 +81,26 @@ const isProcessAlive = (command, childProcess) => { + return isUnsignallableProcessGroup(error); + } }; +-const taskkill = (childProcess, onExit = () => {}) => { +- if (hasExited(childProcess)) { +- onExit(null); +- return; +- } +- NodeChildProcess.execFile("taskkill", ["/pid", String(childProcess.pid), "/T", "/F"], { +- windowsHide: true +- }, onExit); +-}; -const make = /*#__PURE__*/Effect.gen(function* () { +const makeWithSpawn = spawnProcess => Effect.gen(function* () { ++ // Starts taskkill through the injected spawner so it gets the same off-thread launch as commands. ++ const taskkill = (childProcess, onExit = () => {}) => { ++ if (hasExited(childProcess)) { ++ onExit(null); ++ return; ++ } ++ let done = false; ++ const finish = error => { ++ if (done) return; ++ done = true; ++ onExit(error); ++ }; ++ const killer = spawnProcess("taskkill", ["/pid", String(childProcess.pid), "/T", "/F"], { ++ stdio: "ignore", ++ windowsHide: true ++ }); ++ killer.on("error", finish); ++ killer.on("exit", code => finish(code === 0 ? null : new Error(`taskkill exited with code ${code}`))); ++ }; const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const resolveWorkingDirectory = Effect.fnUntraced(function* (options) { -@@ -335,7 +335,7 @@ const make = /*#__PURE__*/Effect.gen(function* () { +@@ -335,7 +345,7 @@ const make = /*#__PURE__*/Effect.gen(function* () { }; const spawn = (command, spawnOptions) => Effect.callback(resume => { const deferred = Deferred.makeUnsafe(); @@ -53,7 +81,7 @@ index 08d50b16dfe564551f1bffc7cae2d3b53805e172..6760a0d0c48f30b1a9aa8bbfdf6b9c6d handle.on("error", error => { // Node copies the arguments into `spawnargs`. if (command.args.some(Redacted.isRedacted)) { -@@ -665,7 +665,8 @@ const make = /*#__PURE__*/Effect.gen(function* () { +@@ -665,7 +675,8 @@ const make = /*#__PURE__*/Effect.gen(function* () { * @category layers * @since 4.0.0 */ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index efe29f1058b68..54f17c5faea51 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -356,7 +356,7 @@ pnpmfileChecksum: sha256-HxQwAhyWNI/9fnrZxQ0MltAhBMQohaHvLg4GiiAL6JA= patchedDependencies: '@alchemy.run/test@2.0.0-beta.85': 29c0539ea71d9cc372f0b16558640599beb9bc609605feb2a493da268d72713e '@clerk/expo-native-components@0.4.0': 7a808b92da8460750b85818552e13be3fb73d111691c7db76965970351e124e4 - '@effect/platform-node-shared@4.1.2': 285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e + '@effect/platform-node-shared@4.1.2': 06952843c6f431ca83c78762fd18d23cda76883801049ee311feced2438ecf14 '@effect/vitest@4.1.2': 299510b71c36537a832a67e9d160a5f112a5f326a29e95c41cb45fc8234e47c7 '@expo/fingerprint@0.21.5': 8fba80d6bbce46c72de427fe8c7907462d262a0bfe50c6b2b7ee9c4e0c61e5e1 '@expo/metro-config@58.0.14': 90fa980e8b6480631fb2fbb04559834e3c612db40c550cd828f11be297d8d760 @@ -837,7 +837,7 @@ importers: version: 4.1.2(bufferutil@4.1.0)(effect@4.1.2)(redis@6.2.1)(utf-8-validate@6.0.6) '@effect/platform-node-shared': specifier: 4.1.2 - version: 4.1.2(patch_hash=285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) + version: 4.1.2(patch_hash=06952843c6f431ca83c78762fd18d23cda76883801049ee311feced2438ecf14)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) '@ff-labs/fff-node': specifier: 0.11.0 version: 0.11.0(patch_hash=cc71fcec015f4d39bc96ae2e2bf6b0ca678124c722c66a7841efce9d34b5983c) @@ -13678,13 +13678,13 @@ snapshots: '@effect/platform-bun@4.1.2(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6)': dependencies: - '@effect/platform-node-shared': 4.1.2(patch_hash=285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) + '@effect/platform-node-shared': 4.1.2(patch_hash=06952843c6f431ca83c78762fd18d23cda76883801049ee311feced2438ecf14)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) effect: 4.1.2 transitivePeerDependencies: - bufferutil - utf-8-validate - '@effect/platform-node-shared@4.1.2(patch_hash=285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6)': + '@effect/platform-node-shared@4.1.2(patch_hash=06952843c6f431ca83c78762fd18d23cda76883801049ee311feced2438ecf14)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6)': dependencies: '@types/ws': 8.18.2 effect: 4.1.2 @@ -13695,7 +13695,7 @@ snapshots: '@effect/platform-node@4.1.2(bufferutil@4.1.0)(effect@4.1.2)(redis@6.2.1)(utf-8-validate@6.0.6)': dependencies: - '@effect/platform-node-shared': 4.1.2(patch_hash=285f31ebe703ad4303b1dea2d398855178ee29381eae470f76931172052db53e)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) + '@effect/platform-node-shared': 4.1.2(patch_hash=06952843c6f431ca83c78762fd18d23cda76883801049ee311feced2438ecf14)(bufferutil@4.1.0)(effect@4.1.2)(utf-8-validate@6.0.6) effect: 4.1.2 redis: 6.2.1 undici: 8.11.2 From b1f00f33e39dace23d97c99f0cadec46046f8cb0 Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Fri, 6 Nov 2026 19:56:25 -0500 Subject: [PATCH 05/14] fix(server): a dead launcher thread ends its children's output and process trees If a launcher worker exits, readers of its children's output now finish instead of waiting forever, the whole process tree is killed through another worker, and writes fail with EPIPE. Errors inside the worker no longer take it down. --- apps/server/src/process/SpawnWorker.test.ts | 2 +- apps/server/src/process/SpawnWorker.ts | 61 ++++++++++++++++----- 2 files changed, 47 insertions(+), 16 deletions(-) diff --git a/apps/server/src/process/SpawnWorker.test.ts b/apps/server/src/process/SpawnWorker.test.ts index 1017189ae2f62..d6997133f2b76 100644 --- a/apps/server/src/process/SpawnWorker.test.ts +++ b/apps/server/src/process/SpawnWorker.test.ts @@ -104,7 +104,7 @@ describe("SpawnWorker", () => { }).pipe(Effect.scoped, Effect.provide(layer)), ); - it.live("kills running commands, including process trees and on scope interruption", () => + it.live("kills running commands, detached or not, and on scope interruption", () => Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; for (const detached of [true, false]) { diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts index da989e4459659..949b83f4381a3 100644 --- a/apps/server/src/process/SpawnWorker.ts +++ b/apps/server/src/process/SpawnWorker.ts @@ -6,6 +6,7 @@ import * as NodeWorkerThreads from "node:worker_threads"; import * as NodeChildProcessSpawner from "@effect/platform-node-shared/NodeChildProcessSpawner"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as HostProcess from "@t3tools/shared/HostProcess"; // Windows runs CreateProcess synchronously on the calling thread, and antivirus or a busy disk can // hold it for seconds. Workers own the children, so only the worker waits. @@ -114,7 +115,11 @@ parentPort.on("message", (request) => { post({ type: "written", id, fd: request.fd, error: { message: "write after end", code: "EPIPE" } }); return; } - stream.write(request.data, (error) => post({ type: "written", id, fd: request.fd, error: error ? serialize(error) : undefined })); + try { + stream.write(request.data, (error) => post({ type: "written", id, fd: request.fd, error: error ? serialize(error) : undefined })); + } catch (error) { + post({ type: "written", id, fd: request.fd, error: serialize(error) }); + } return; case "end": if (!stream || stream.destroyed || stream.writableFinished) post({ type: "finished", id, fd: request.fd }); @@ -125,7 +130,9 @@ parentPort.on("message", (request) => { else stream?.resume(); return; case "kill": - child.kill(request.signal); + try { + child.kill(request.signal); + } catch {} return; } }); @@ -181,6 +188,10 @@ class WorkerChild _encoding: BufferEncoding, callback: (error?: Error | null) => void, ) => { + if (this.worker.closed) { + callback(Object.assign(new Error("write EPIPE"), { code: "EPIPE" })); + return; + } const pending = this.#writes.get(fd) ?? []; pending.push(callback); this.#writes.set(fd, pending); @@ -252,11 +263,12 @@ class WorkerChild this.emit("exit", code, signal); } - /** The worker is gone, so no further output or acknowledgement can arrive. */ + /** The worker is gone: end output so readers finish, and let the exit report the kill. */ abandon(): void { + // Not destroy(error): Effect leaves an unread stderr's pipe without an error listener. for (const stream of this.stdio) { - stream?.on("error", () => {}); - stream?.destroy(); + if (stream && "push" in stream && !stream.readableEnded) stream.push(null); + else if (stream && !("push" in stream)) stream.destroy(); } } } @@ -350,6 +362,7 @@ class SpawnWorker { child.exit(message.code, message.signal); break; case "close": + child.emit("close", child.exitCode, child.signalCode); this.release(child); break; } @@ -374,12 +387,8 @@ class SpawnWorker { child.emit("error", error); continue; } - // The worker owned the only handle; best-effort so a crash leaves no orphans. - if (child.pid !== undefined) { - try { - process.kill(child.pid, "SIGKILL"); - } catch {} - } + // The worker owned the only handle, so end the whole tree before reporting the exit. + if (child.pid !== undefined) this.pool.killTree(child.pid); child.exit(null, "SIGKILL"); child.abandon(); } @@ -396,8 +405,13 @@ class SpawnWorker { class SpawnWorkerPool { readonly workers = new Set(); + readonly platform: NodeJS.Platform; #nextId = 0; + constructor(platform: NodeJS.Platform) { + this.platform = platform; + } + readonly spawn = ( command: string, args: ReadonlyArray, @@ -415,6 +429,19 @@ class SpawnWorkerPool { return worker.spawn(++this.#nextId, command, args, options); }; + killTree(pid: number): void { + if (this.platform === "win32") { + this.spawn("taskkill", ["/pid", String(pid), "/T", "/F"], { + stdio: "ignore", + windowsHide: true, + }).on("error", () => {}); + return; + } + try { + process.kill(pid, "SIGKILL"); + } catch {} + } + #add(): SpawnWorker { const worker = new SpawnWorker(this); this.workers.add(worker); @@ -435,8 +462,12 @@ class SpawnWorkerPool { } export const layer = Layer.unwrap( - Effect.acquireRelease( - Effect.sync(() => new SpawnWorkerPool()), - (pool) => Effect.sync(() => pool.close()), - ).pipe(Effect.map((pool) => NodeChildProcessSpawner.layerWithSpawn(pool.spawn))), + Effect.gen(function* () { + const platform = yield* HostProcess.Platform; + const pool = yield* Effect.acquireRelease( + Effect.sync(() => new SpawnWorkerPool(platform)), + (pool) => Effect.sync(() => pool.close()), + ); + return NodeChildProcessSpawner.layerWithSpawn(pool.spawn); + }), ); From 9139b35a26830ac99b5e9c6df6c1a832ea8834e9 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 7 Nov 2026 05:19:04 +0000 Subject: [PATCH 06/14] fix(server): report a dead launcher's child exit only after taskkill ends its tree --- apps/server/src/process/SpawnWorker.ts | 41 ++++++++++++++++++-------- 1 file changed, 29 insertions(+), 12 deletions(-) diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts index 949b83f4381a3..8bbed38bc4506 100644 --- a/apps/server/src/process/SpawnWorker.ts +++ b/apps/server/src/process/SpawnWorker.ts @@ -388,9 +388,12 @@ class SpawnWorker { continue; } // The worker owned the only handle, so end the whole tree before reporting the exit. - if (child.pid !== undefined) this.pool.killTree(child.pid); - child.exit(null, "SIGKILL"); - child.abandon(); + const finish = () => { + child.exit(null, "SIGKILL"); + child.abandon(); + }; + if (child.pid === undefined) finish(); + else this.pool.killTree(child.pid, finish); } this.children.clear(); } @@ -429,17 +432,31 @@ class SpawnWorkerPool { return worker.spawn(++this.#nextId, command, args, options); }; - killTree(pid: number): void { - if (this.platform === "win32") { - this.spawn("taskkill", ["/pid", String(pid), "/T", "/F"], { - stdio: "ignore", - windowsHide: true, - }).on("error", () => {}); + killTree(pid: number, done: () => void): void { + const killPid = () => { + try { + process.kill(pid, "SIGKILL"); + } catch {} + }; + if (this.platform !== "win32") { + killPid(); + done(); return; } - try { - process.kill(pid, "SIGKILL"); - } catch {} + let settled = false; + const finish = (killed: boolean) => { + if (settled) return; + settled = true; + // taskkill could not end the tree, so at least end the direct child. + if (!killed) killPid(); + done(); + }; + this.spawn("taskkill", ["/pid", String(pid), "/T", "/F"], { + stdio: "ignore", + windowsHide: true, + }) + .on("error", () => finish(false)) + .on("exit", (code) => finish(code === 0)); } #add(): SpawnWorker { From 93556fef0b20cf5b830e239bd7931301b1af75d2 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 7 Nov 2026 05:30:58 +0000 Subject: [PATCH 07/14] fix(server): keep the launcher alive when a spawn hits descriptor exhaustion --- apps/server/src/process/SpawnWorker.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts index 8bbed38bc4506..50d7bbc4c4bb0 100644 --- a/apps/server/src/process/SpawnWorker.ts +++ b/apps/server/src/process/SpawnWorker.ts @@ -89,7 +89,7 @@ parentPort.on("message", (request) => { children.delete(id); post({ type: "close", id }); }); - child.stdio.forEach((stream, fd) => { + child.stdio?.forEach((stream, fd) => { if (!stream) return; stream.on("error", (error) => post({ type: "stream-error", id, fd, error: serialize(error) })); if (fd === 0) return; From a9549c185d2db19cba8b857f001dbffbb007cf87 Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 7 Nov 2026 13:34:34 -0500 Subject: [PATCH 08/14] refactor(server): drop the Linux launcher so this PR changes Windows only The Linux helper process belongs to #30209. This PR now keeps only the injectable spawner patch, the Windows launcher threads, and the server wiring that picks them on Windows. Claude Opus 5.5 via Claude Code --- apps/server/src/binMain.ts | 5 +- apps/server/src/process/SpawnHost.test.ts | 320 ------------------ apps/server/src/process/SpawnHost.ts | 303 ----------------- apps/server/src/process/SpawnWorker.ts | 9 +- .../src/process/spawnHost.test-fixture.ts | 38 --- apps/server/src/process/spawnHostProcess.ts | 127 ------- apps/server/src/server.ts | 4 +- 7 files changed, 11 insertions(+), 795 deletions(-) delete mode 100644 apps/server/src/process/SpawnHost.test.ts delete mode 100644 apps/server/src/process/SpawnHost.ts delete mode 100644 apps/server/src/process/spawnHost.test-fixture.ts delete mode 100644 apps/server/src/process/spawnHostProcess.ts diff --git a/apps/server/src/binMain.ts b/apps/server/src/binMain.ts index 7d4f761d16e8c..0dabd73873440 100644 --- a/apps/server/src/binMain.ts +++ b/apps/server/src/binMain.ts @@ -8,10 +8,7 @@ export async function runCliCommand() { // Install the socket guard before importing code that can open outbound connections. const { ignoreMacosTypeOfServiceEinval } = await import("./outboundHttp.ts"); ignoreMacosTypeOfServiceEinval(); - if (command === "__spawn-host") { - const { runSpawnHost } = await import("./process/spawnHostProcess.ts"); - runSpawnHost(); - } else if (command === "__document-text") { + if (command === "__document-text") { const { extractDocumentText } = await import("./provider/documentTextWorker.ts"); try { process.stdout.write(await extractDocumentText(process.argv[3]!, process.argv[4]!)); diff --git a/apps/server/src/process/SpawnHost.test.ts b/apps/server/src/process/SpawnHost.test.ts deleted file mode 100644 index f53a31dcf631c..0000000000000 --- a/apps/server/src/process/SpawnHost.test.ts +++ /dev/null @@ -1,320 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off - the regression intercepts the native spawn boundary and inspects IPC sockets. -// oxlint-disable-next-line t3code/namespace-node-imports -- Mutating the CJS export is necessary to intercept the native ESM binding. -import NodeChildProcess from "node:child_process"; -import * as NodeModule from "node:module"; -import type * as NodeStream from "node:stream"; -import * as NodeURL from "node:url"; -import * as NodeServices from "@effect/platform-node/NodeServices"; -import { describe, expect, it } from "@effect/vitest"; -import { vi } from "vite-plus/test"; -import * as HostProcess from "@t3tools/shared/HostProcess"; -import * as Deferred from "effect/Deferred"; -import * as Effect from "effect/Effect"; -import * as Exit from "effect/Exit"; -import * as Fiber from "effect/Fiber"; -import * as Layer from "effect/Layer"; -import * as Result from "effect/Result"; -import * as Stream from "effect/Stream"; -import * as ChildProcess from "effect/process/ChildProcess"; -import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; - -import * as SpawnHost from "./SpawnHost.ts"; - -const layer = SpawnHost.layerHost(process.execPath, [ - NodeURL.fileURLToPath(new URL("../bin.ts", import.meta.url)), - "__spawn-host", -]).pipe(Layer.provide(NodeServices.layer)); -const text = (stream: Stream.Stream) => - stream.pipe(Stream.decodeText(), Stream.mkString); - -// quarantine: #30183 Platform coverage: this layer is installed only on Linux. -describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("SpawnHost", () => { - it.live("starts commands without calling native spawn on the server thread", () => - Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const nativeSpawn = NodeChildProcess.spawn; - const helpers: NodeChildProcess.ChildProcess[] = []; - using _spawn = vi - .spyOn(NodeChildProcess, "spawn") - .mockImplementation((command, args, options) => { - if (command === "/bin/echo") throw new Error("Native spawn blocked the server thread"); - const helper = nativeSpawn(command, args, options); - helpers.push(helper); - return helper; - }); - using _restore = { - [Symbol.dispose]() { - _spawn.mockRestore(); - NodeModule.syncBuiltinESMExports(); - }, - }; - NodeModule.syncBuiltinESMExports(); - const child = yield* spawner.spawn(ChildProcess.make("/bin/echo", ["responsive"])); - expect(yield* text(child.stdout)).toBe("responsive\n"); - expect(yield* child.exitCode).toBe(0); - for (let i = 0; i < 2; i++) { - yield* spawner.spawn(ChildProcess.make(process.execPath, ["-e", "process.stdin.resume()"])); - } - expect(helpers).toHaveLength(1); - }).pipe(Effect.scoped, Effect.provide(layer)), - ); - - it.live("lets another command finish while one host is blocked inside spawn", () => - Effect.gen(function* () { - const spawned = yield* Deferred.make(); - const nativeSpawn = NodeChildProcess.spawn; - using _spawn = vi - .spyOn(NodeChildProcess, "spawn") - .mockImplementation((command, args, options) => { - const host = nativeSpawn(command, args, { - ...options, - stdio: ["inherit", "inherit", "inherit", "ipc", "pipe"], - }); - Deferred.doneUnsafe(spawned, Exit.succeed(host)); - return host; - }); - using _restore = { - [Symbol.dispose]() { - _spawn.mockRestore(); - NodeModule.syncBuiltinESMExports(); - }, - }; - NodeModule.syncBuiltinESMExports(); - const fixtureLayer = SpawnHost.layerHost(process.execPath, [ - NodeURL.fileURLToPath(new URL("./spawnHost.test-fixture.ts", import.meta.url)), - ]).pipe(Layer.provide(NodeServices.layer)); - yield* Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const blocked = yield* spawner - .spawn(ChildProcess.make("/t3-blocked-spawn")) - .pipe(Effect.forkChild); - const host = yield* Deferred.await(spawned); - const gate = host.stdio[4] as NodeStream.Duplex; - yield* Effect.promise( - () => new Promise((resolve) => gate.once("data", () => resolve())), - ); - yield* Effect.gen(function* () { - const healthy = yield* spawner.spawn( - ChildProcess.make("/bin/echo", ["still responding"]), - ); - expect(yield* text(healthy.stdout)).toBe("still responding\n"); - expect(yield* healthy.exitCode).toBe(0); - expect(host.exitCode).toBeNull(); - }).pipe( - Effect.ensuring( - Effect.sync(() => { - gate.write("x"); - }), - ), - ); - const child = yield* Fiber.join(blocked); - expect(yield* child.exitCode).toBe(0); - }).pipe(Effect.scoped, Effect.provide(fixtureLayer)); - }), - ); - - it.live("preserves stdin, fast output, env, cwd, literal args and additional descriptors", () => - Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const literal = "quote' \" $HOME; * &"; - const script = ` - const fs = require('node:fs'); - const input = fs.readFileSync(0, 'utf8'); - fs.writeSync(1, JSON.stringify([input, process.cwd(), process.env.SPAWN_HOST_TEST, process.argv[1]]) + '\\n'); - fs.writeSync(1, 'a'.repeat(1024 * 1024)); - fs.writeSync(2, 'b'.repeat(1024 * 1024)); - fs.writeSync(3, 'fd3:' + fs.readFileSync(4, 'utf8')); - `; - for (let i = 0; i < 3; i++) { - const child = yield* spawner.spawn( - ChildProcess.make(process.execPath, ["-e", script, literal], { - cwd: "/tmp", - env: { SPAWN_HOST_TEST: "inherited value" }, - stdin: Stream.encodeText(Stream.make("stdin value")), - additionalFds: { - fd3: { type: "output" }, - fd4: { type: "input", stream: Stream.encodeText(Stream.make("fd4 value")) }, - }, - }), - ); - const [stdout, stderr, fd3, code] = yield* Effect.all( - [text(child.stdout), text(child.stderr), text(child.getOutputFd(3)), child.exitCode], - { concurrency: "unbounded" }, - ); - expect(stdout).toBe( - JSON.stringify(["stdin value", "/tmp", "inherited value", literal]) + - "\n" + - "a".repeat(1024 * 1024), - ); - expect(stderr).toBe("b".repeat(1024 * 1024)); - expect(fd3).toBe("fd3:fd4 value"); - expect(code).toBe(0); - } - const pipeline = ChildProcess.make("/bin/echo", ["piped"]).pipe( - ChildProcess.pipeTo(ChildProcess.make("/usr/bin/tr", ["a-z", "A-Z"])), - ); - const child = yield* spawner.spawn(pipeline); - expect(yield* text(child.stdout)).toBe("PIPED\n"); - expect(yield* child.exitCode).toBe(0); - }).pipe(Effect.scoped, Effect.provide(layer)), - ); - - it.live("retains launch errors and exit codes, and can reuse a host after either", () => - Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const result = yield* spawner - .spawn(ChildProcess.make("/t3-missing-command-30183")) - .pipe(Effect.result); - expect(Result.isFailure(result)).toBe(true); - if (Result.isFailure(result)) expect(result.failure.reason._tag).toBe("NotFound"); - const invalid = yield* spawner - .spawn(ChildProcess.make("/bin/true", [], { cwd: "/dev/null" })) - .pipe(Effect.result); - expect(Result.isFailure(invalid)).toBe(true); - const fdCounts: number[] = []; - for (let pass = 0; pass < 2; pass++) { - if (pass === 1) { - for (let i = 0; i < 12; i++) { - for (const command of ["/t3-missing-command-30183", "/dev/null"]) { - expect( - Result.isFailure( - yield* spawner.spawn(ChildProcess.make(command)).pipe(Effect.result), - ), - ).toBe(true); - } - } - } - const probe = yield* spawner.spawn( - ChildProcess.make(process.execPath, [ - "-e", - "process.stdout.write(String(require('node:fs').readdirSync('/proc/' + process.ppid + '/fd').length))", - ]), - ); - fdCounts.push(Number(yield* text(probe.stdout))); - expect(yield* probe.exitCode).toBe(0); - } - expect(fdCounts[1]).toBe(fdCounts[0]); - const child = yield* spawner.spawn(ChildProcess.make("/bin/sh", ["-c", "exit 23"])); - expect(yield* child.exitCode).toBe(23); - const healthy = yield* spawner.spawn(ChildProcess.make("/bin/true")); - expect(yield* healthy.exitCode).toBe(0); - }).pipe(Effect.scoped, Effect.provide(layer)), - ); - - it.live("kills a hosted process group and reports the signalled exit", () => - Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - for (const detached of [true, false]) { - const child = yield* spawner.spawn( - ChildProcess.make( - process.execPath, - [ - "-e", - "process.on('SIGTERM', () => {});process.stdout.write(String(process.pid));process.stdin.resume()", - ], - { detached }, - ), - ); - const pid = yield* child.stdout.pipe(Stream.take(1), text); - expect(Number(pid)).toBe(child.pid); - yield* child.kill({ forceKillAfter: "0 millis" }); - expect(yield* child.isRunning).toBe(false); - expect(Result.isFailure(yield* child.exitCode.pipe(Effect.result))).toBe(true); - } - const started = yield* Deferred.make(); - const running = yield* Effect.gen(function* () { - const scoped = yield* spawner.spawn( - ChildProcess.make(process.execPath, ["-e", "process.stdin.resume()"]), - ); - yield* Deferred.succeed(started, scoped); - return yield* Effect.never; - }).pipe(Effect.scoped, Effect.forkChild); - const scoped = yield* Deferred.await(started); - yield* Fiber.interrupt(running); - expect(yield* scoped.isRunning).toBe(false); - }).pipe(Effect.scoped, Effect.provide(layer)), - ); - - it.live( - "fails a command when its launcher dies and starts the next command in a fresh host", - () => - Effect.gen(function* () { - const spawned = yield* Deferred.make(); - const nativeSpawn = NodeChildProcess.spawn; - using _spawn = vi - .spyOn(NodeChildProcess, "spawn") - .mockImplementation((command, args, options) => { - const host = nativeSpawn(command, args, options); - Deferred.doneUnsafe(spawned, Exit.succeed(host)); - return host; - }); - using _restore = { - [Symbol.dispose]() { - _spawn.mockRestore(); - NodeModule.syncBuiltinESMExports(); - }, - }; - NodeModule.syncBuiltinESMExports(); - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const child = yield* spawner.spawn( - ChildProcess.make(process.execPath, [ - "-e", - "process.stdout.write('ready');process.stdin.resume()", - ]), - ); - yield* child.stdout.pipe(Stream.take(1), Stream.runDrain); - const host = yield* Deferred.await(spawned); - host.kill("SIGKILL"); - expect(Result.isFailure(yield* child.exitCode.pipe(Effect.result))).toBe(true); - expect(yield* child.isRunning).toBe(false); - const healthy = yield* spawner.spawn(ChildProcess.make("/bin/true")); - expect(yield* healthy.exitCode).toBe(0); - }).pipe(Effect.scoped, Effect.provide(layer)), - ); - - it.live("reports descriptor exhaustion without stopping another command in the same host", () => - Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const running = yield* spawner.spawn( - ChildProcess.make(process.execPath, [ - "-e", - "process.stdout.write('ready');process.stdin.resume()", - ]), - ); - yield* running.stdout.pipe(Stream.take(1), Stream.runDrain); - const failed = yield* spawner - .spawn(ChildProcess.make("/t3-exhausted-spawn")) - .pipe(Effect.result); - expect(Result.isFailure(failed)).toBe(true); - expect(yield* running.isRunning).toBe(true); - const healthy = yield* spawner.spawn(ChildProcess.make("/bin/true")); - expect(yield* healthy.exitCode).toBe(0); - }).pipe( - Effect.scoped, - Effect.provide( - SpawnHost.layerHost("/bin/sh", [ - "-c", - 'ulimit -n 64; exec "$@"', - "sh", - process.execPath, - NodeURL.fileURLToPath(new URL("./spawnHost.test-fixture.ts", import.meta.url)), - ]).pipe(Layer.provide(NodeServices.layer)), - ), - ), - ); - - it.live("fails launcher startup without retrying the target on the server", () => - Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const result = yield* spawner.spawn(ChildProcess.make("/bin/true")).pipe(Effect.result); - expect(Result.isFailure(result)).toBe(true); - }).pipe( - Effect.scoped, - Effect.provide( - SpawnHost.layerHost("/t3-missing-launcher-30183", []).pipe( - Layer.provide(NodeServices.layer), - ), - ), - ), - ); -}); diff --git a/apps/server/src/process/SpawnHost.ts b/apps/server/src/process/SpawnHost.ts deleted file mode 100644 index ada20d1431067..0000000000000 --- a/apps/server/src/process/SpawnHost.ts +++ /dev/null @@ -1,303 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off globalTimers:off - Node process and IPC callbacks implement the platform spawner boundary. -import * as NodeChildProcess from "node:child_process"; -import * as NodeEvents from "node:events"; -import type * as NodeNet from "node:net"; -import type * as NodeStream from "node:stream"; -import * as NodeChildProcessSpawner from "@effect/platform-node-shared/NodeChildProcessSpawner"; -import * as Effect from "effect/Effect"; -import * as Layer from "effect/Layer"; -import * as HostProcess from "@t3tools/shared/HostProcess"; -import { resolveSelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; - -import type { SpawnHostEvent, SpawnHostSpawnRequest } from "./spawnHostProcess.ts"; -import * as SpawnWorker from "./SpawnWorker.ts"; - -// A blocked native spawn occupies its host; already-running children share available hosts. -const MAX_IDLE_HOSTS = 1; -const HOST_READY_TIMEOUT_MS = 10_000; - -class HostedChild - extends NodeEvents.EventEmitter - implements NodeChildProcessSpawner.SpawnedChildProcess -{ - pid: number | undefined; - exitCode: number | null = null; - signalCode: NodeJS.Signals | null = null; - stdin: NodeStream.Writable | null = null; - stdout: NodeStream.Readable | null = null; - stderr: NodeStream.Readable | null = null; - readonly stdio: Array = []; - referenced = true; - signal: NodeJS.Signals | number | undefined; - exited = false; - - readonly host: SpawnHost; - readonly id: number; - - constructor(host: SpawnHost, id: number) { - super(); - this.host = host; - this.id = id; - } - - kill(signal: NodeJS.Signals | number = "SIGTERM"): boolean { - if (this.exited) return false; - if (this.pid === undefined) { - this.signal = signal; - return true; - } - try { - process.kill(this.pid, signal); - return true; - } catch { - return false; - } - } - - ref(): void { - this.referenced = true; - this.host.setReferenced(this); - } - - unref(): void { - this.referenced = false; - this.host.setReferenced(this); - } - - attach(fd: number, socket: NodeNet.Socket): void { - this.stdio[fd] = socket; - if (fd === 0) { - // Node destroys stdin on exit even after the Effect sink has removed its error listener. - socket.on("error", () => {}); - this.stdin = socket; - } else if (fd === 1) this.stdout = socket; - else if (fd === 2) this.stderr = socket; - } - - exit(code: number | null, signal: NodeJS.Signals | null): void { - this.exited = true; - this.exitCode = code; - this.signalCode = signal; - this.stdin?.destroy(); - this.emit("exit", code, signal); - process.nextTick(() => { - for (const stream of this.stdio) { - if ( - stream && - "resume" in stream && - stream.listenerCount("data") === 0 && - stream.listenerCount("readable") === 0 - ) { - stream.resume(); - } - } - }); - } -} - -class SpawnHost { - readonly process: NodeChildProcess.ChildProcess; - readonly children = new Map(); - ready = false; - closed = false; - request: SpawnHostSpawnRequest | undefined; - #nextId = 0; - readonly #readyTimer: NodeJS.Timeout; - - readonly pool: SpawnPool; - - constructor(pool: SpawnPool, command: string, args: ReadonlyArray) { - this.pool = pool; - const env = { ...process.env }; - delete env.NODE_OPTIONS; - this.process = NodeChildProcess.spawn(command, [...args], { - cwd: "/", - env, - stdio: ["inherit", "inherit", "inherit", "ipc"], - }); - this.process.unref(); - this.#readyTimer = setTimeout(() => { - this.process.kill("SIGKILL"); - this.fail(new Error("The process launcher did not start in time.")); - }, HOST_READY_TIMEOUT_MS); - this.#readyTimer.unref(); - this.process.on("message", (event, socket) => - this.onEvent(event as SpawnHostEvent, socket as NodeNet.Socket | undefined), - ); - this.process.on("error", (error) => this.fail(error)); - this.process.once("close", () => - this.fail(new Error("The process launcher exited unexpectedly.")), - ); - this.process.channel?.unref(); - } - - spawn( - command: string, - args: ReadonlyArray, - options: NodeChildProcess.SpawnOptions, - ): HostedChild { - const child = new HostedChild(this, ++this.#nextId); - this.children.set(child.id, child); - this.request = { - type: "spawn", - id: child.id, - referenced: child.referenced, - command, - args, - options: { ...options, cwd: options.cwd ?? process.cwd(), env: options.env ?? process.env }, - }; - this.syncRef(); - if (this.ready) this.send(); - return child; - } - - syncRef(): void { - if ([...this.children.values()].some((child) => child.referenced)) this.process.channel?.ref(); - else this.process.channel?.unref(); - } - - setReferenced(child: HostedChild): void { - this.syncRef(); - if (this.ready && this.children.has(child.id) && this.process.connected) { - this.process.send({ type: "ref", id: child.id, referenced: child.referenced }, (error) => { - if (error) this.fail(error); - }); - } - } - - send(): void { - this.process.send(this.request!, (error) => { - if (error) this.fail(error); - }); - } - - onEvent(event: SpawnHostEvent, socket: NodeNet.Socket | undefined): void { - if (this.closed) { - socket?.destroy(); - return; - } - if (event.type === "ready") { - clearTimeout(this.#readyTimer); - this.ready = true; - if (this.request) this.send(); - return; - } - const child = this.children.get(event.id); - if (!child) { - socket?.destroy(); - return; - } - switch (event.type) { - case "stdio": - if (socket) child.attach(event.fd, socket); - break; - case "spawn": - this.request = undefined; - child.pid = event.pid; - if (child.signal !== undefined) child.kill(child.signal); - child.emit("spawn"); - break; - case "exit": - child.exit(event.code, event.signal); - this.release(child); - break; - case "error": - if (child.pid === undefined) { - child.exited = true; - this.release(child); - } - child.emit("error", Object.assign(new Error(event.error.message), event.error)); - break; - } - } - - release(child: HostedChild): void { - this.children.delete(child.id); - if (this.request?.id === child.id) this.request = undefined; - this.syncRef(); - this.pool.release(this); - } - - fail(error: Error): void { - if (this.closed) return; - this.close(); - for (const child of this.children.values()) { - if (child.exited) continue; - if (child.pid === undefined) { - child.exited = true; - child.emit("error", error); - } else { - child.exit(null, "SIGKILL"); - } - } - } - - close(): void { - this.closed = true; - clearTimeout(this.#readyTimer); - this.pool.hosts.delete(this); - for (const child of this.children.values()) { - if (!child.referenced || child.pid === undefined || child.exited) continue; - try { - process.kill(-child.pid, "SIGKILL"); - } catch { - child.kill("SIGKILL"); - } - } - if (this.process.connected) this.process.disconnect(); - } -} - -class SpawnPool { - readonly hosts = new Set(); - - readonly command: string; - readonly args: ReadonlyArray; - - constructor(command: string, args: ReadonlyArray) { - this.command = command; - this.args = args; - } - - readonly spawn = ( - command: string, - args: ReadonlyArray, - options: NodeChildProcess.SpawnOptions, - ) => { - let host = [...this.hosts].find((candidate) => candidate.request === undefined); - if (!host) { - host = new SpawnHost(this, this.command, this.args); - this.hosts.add(host); - } - return host.spawn(command, args, options); - }; - - release(host: SpawnHost): void { - if ( - [...this.hosts].filter((candidate) => candidate.children.size === 0).length > MAX_IDLE_HOSTS - ) - host.close(); - } - - close(): void { - for (const host of this.hosts) host.close(); - } -} - -export const layerHost = (command: string, args: ReadonlyArray) => - Layer.unwrap( - Effect.acquireRelease( - Effect.sync(() => new SpawnPool(command, args)), - (pool) => Effect.sync(() => pool.close()), - ).pipe(Effect.map((pool) => NodeChildProcessSpawner.layerWithSpawn(pool.spawn))), - ); - -export const layer = Layer.unwrap( - Effect.gen(function* () { - const platform = yield* HostProcess.Platform; - if (platform === "win32") return SpawnWorker.layer; - if (platform !== "linux") return NodeChildProcessSpawner.layer; - const invocation = yield* resolveSelfInvocation(); - return layerHost(invocation.command, selfInvocationArgs(invocation, ["__spawn-host"])); - }), -); diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts index 50d7bbc4c4bb0..a5bdf70b073c0 100644 --- a/apps/server/src/process/SpawnWorker.ts +++ b/apps/server/src/process/SpawnWorker.ts @@ -447,7 +447,7 @@ class SpawnWorkerPool { const finish = (killed: boolean) => { if (settled) return; settled = true; - // taskkill could not end the tree, so at least end the direct child. + // As Effect's own kill does when taskkill fails, end at least the direct child. if (!killed) killPid(); done(); }; @@ -488,3 +488,10 @@ export const layer = Layer.unwrap( return NodeChildProcessSpawner.layerWithSpawn(pool.spawn); }), ); + +/** Only Windows creates processes slowly enough on the calling thread to need workers. */ +export const layerServer = Layer.unwrap( + Effect.map(HostProcess.Platform, (platform) => + platform === "win32" ? layer : NodeChildProcessSpawner.layer, + ), +); diff --git a/apps/server/src/process/spawnHost.test-fixture.ts b/apps/server/src/process/spawnHost.test-fixture.ts deleted file mode 100644 index 3dd875a03ff19..0000000000000 --- a/apps/server/src/process/spawnHost.test-fixture.ts +++ /dev/null @@ -1,38 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off - the fixture gates native spawn with a synchronous descriptor read. -// oxlint-disable-next-line t3code/namespace-node-imports -- The fixture replaces the native CJS export before syncing ESM bindings. -import NodeChildProcess from "node:child_process"; -import * as NodeFS from "node:fs"; -import * as NodeModule from "node:module"; -import { runSpawnHost } from "./spawnHostProcess.ts"; - -const spawn = NodeChildProcess.spawn; -const gatedSpawn = ( - command: string, - args: ReadonlyArray = [], - options: NodeChildProcess.SpawnOptions = {}, -) => { - if (command === "/t3-blocked-spawn") { - NodeFS.writeSync(4, "blocked"); - NodeFS.readSync(4, Buffer.alloc(1), 0, 1, null); - return spawn("/bin/true", [], options); - } - if (command === "/t3-exhausted-spawn") { - const fds: number[] = []; - try { - for (let i = 0; i < 128; i++) { - try { - fds.push(NodeFS.openSync("/dev/null", "r")); - } catch { - break; - } - } - return spawn("/bin/true", [], options); - } finally { - for (const fd of fds) NodeFS.closeSync(fd); - } - } - return spawn(command, args, options); -}; -NodeChildProcess.spawn = gatedSpawn as typeof NodeChildProcess.spawn; -NodeModule.syncBuiltinESMExports(); -runSpawnHost(); diff --git a/apps/server/src/process/spawnHostProcess.ts b/apps/server/src/process/spawnHostProcess.ts deleted file mode 100644 index a7ae7ec74d2a9..0000000000000 --- a/apps/server/src/process/spawnHostProcess.ts +++ /dev/null @@ -1,127 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off - the launcher deliberately loads only Node builtins. -import * as NodeChildProcess from "node:child_process"; -import type * as NodeNet from "node:net"; - -export interface SpawnHostSpawnRequest { - readonly type: "spawn"; - readonly id: number; - readonly command: string; - readonly args: ReadonlyArray; - readonly options: NodeChildProcess.SpawnOptions; - readonly referenced: boolean; -} - -export interface SpawnHostError { - readonly message: string; - readonly code?: string | undefined; - readonly errno?: number | undefined; - readonly syscall?: string | undefined; - readonly path?: string | undefined; - readonly spawnargs?: ReadonlyArray | undefined; -} - -/** Sent in order per child: `stdio` once per piped fd (carrying the socket), `spawn`, then `exit`. */ -export type SpawnHostEvent = - | { readonly type: "ready" } - | { readonly type: "stdio"; readonly id: number; readonly fd: number } - | { readonly type: "spawn"; readonly id: number; readonly pid: number } - | { - readonly type: "exit"; - readonly id: number; - readonly code: number | null; - readonly signal: NodeJS.Signals | null; - } - | { readonly type: "error"; readonly id: number; readonly error: SpawnHostError }; - -const toSpawnHostError = (cause: unknown): SpawnHostError => { - const error = cause as Partial & { spawnargs?: ReadonlyArray }; - return { - message: typeof error?.message === "string" ? error.message : String(cause), - code: error?.code, - errno: error?.errno, - syscall: error?.syscall, - path: error?.path, - spawnargs: error?.spawnargs, - }; -}; - -// Child stdio has no public paused option; use net's pauseOnCreate mechanism. -const stopReading = (socket: NodeNet.Socket) => - (socket as unknown as { _handle?: { readStop?: () => void } | null })._handle?.readStop?.(); - -export function runSpawnHost(): void { - const channel = process.send?.bind(process); - if (!channel) { - process.stderr.write("t3 __spawn-host is started by the T3 server.\n"); - process.exitCode = 1; - return; - } - const active = new Map(); - const send = (event: SpawnHostEvent, socket?: NodeNet.Socket) => channel(event, socket, () => {}); - - // Keep reporting exits during group shutdown until the server closes IPC. - for (const signal of ["SIGINT", "SIGTERM", "SIGHUP"] as const) process.on(signal, () => {}); - process.on("disconnect", () => { - for (const { child, referenced } of active.values()) { - if (child.pid === undefined || !referenced) continue; - try { - process.kill(-child.pid, "SIGKILL"); - } catch { - child.kill("SIGKILL"); - } - } - process.exit(0); - }); - - process.on( - "message", - ( - request: - | SpawnHostSpawnRequest - | { readonly type: "ref"; readonly id: number; readonly referenced: boolean }, - ) => { - if (request.type === "ref") { - const entry = active.get(request.id); - if (entry) entry.referenced = request.referenced; - return; - } - const { id } = request; - let child: NodeChildProcess.ChildProcess; - try { - child = NodeChildProcess.spawn(request.command, [...request.args], request.options); - active.set(id, { child, referenced: request.referenced }); - } catch (cause) { - send({ type: "error", id, error: toSpawnHostError(cause) }); - return; - } - const stdio = (child.stdio ?? []).map((socket) => socket as NodeNet.Socket | null); - child.on("error", (cause) => { - if (child.pid === undefined) { - active.delete(id); - for (const socket of stdio) socket?.destroy(); - } - send({ type: "error", id, error: toSpawnHostError(cause) }); - }); - // Stop reads before the loop polls, preserving every byte for the server. - for (const socket of stdio) { - if (socket) stopReading(socket); - } - // Prevent Node's exit-time flush/destroy from consuming sockets still queued for IPC transfer. - child.stdin = null; - child.stdout = null; - child.stderr = null; - child.stdio?.fill(null); - child.once("spawn", () => { - stdio.forEach((socket, fd) => { - if (socket) send({ type: "stdio", id, fd }, socket); - }); - send({ type: "spawn", id, pid: child.pid! }); - }); - child.on("exit", (code, signal) => { - active.delete(id); - send({ type: "exit", id, code, signal }); - }); - }, - ); - send({ type: "ready" }); -} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 80f27345baf34..8d80006f294fd 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -15,7 +15,7 @@ import * as NodeHttp from "node:http"; import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import * as SpawnHost from "./process/SpawnHost.ts"; +import * as SpawnWorker from "./process/SpawnWorker.ts"; import { EnvironmentHttpApi, type RepositoryIdentity } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Duration from "effect/Duration"; @@ -384,7 +384,7 @@ const layerPlatformServices = DocumentText.layer.pipe( Layer.mergeAll( NodeServices.layer, NodeCrypto.layer, - SpawnHost.layer.pipe(Layer.provide(NodeServices.layer)), + SpawnWorker.layerServer.pipe(Layer.provide(NodeServices.layer)), ), ), ); From 0ac07330ee987290d26198195413ab0010158b3a Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 7 Nov 2026 13:38:32 -0500 Subject: [PATCH 09/14] fix(server): a request for a child whose launch failed early no longer stops its launcher thread Claude Opus 5.5 via Claude Code --- apps/server/src/process/SpawnWorker.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts index a5bdf70b073c0..3c0b11d728f7e 100644 --- a/apps/server/src/process/SpawnWorker.ts +++ b/apps/server/src/process/SpawnWorker.ts @@ -108,7 +108,7 @@ parentPort.on("message", (request) => { if (request.type === "end") post({ type: "finished", id, fd: request.fd }); return; } - const stream = child.stdio[request.fd]; + const stream = child.stdio?.[request.fd]; switch (request.type) { case "write": if (!stream || stream.destroyed) { From 0caa12dd4c0aeb671410069c802738c681dcb3c7 Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 7 Nov 2026 13:50:11 -0500 Subject: [PATCH 10/14] fix(server): a caller that drops a child's pipe closes it in the launcher too Claude Opus 5.5 via Claude Code --- apps/server/src/process/SpawnWorker.test.ts | 5 +++++ apps/server/src/process/SpawnWorker.ts | 15 ++++++++++++--- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/apps/server/src/process/SpawnWorker.test.ts b/apps/server/src/process/SpawnWorker.test.ts index d6997133f2b76..96d4bc11678fd 100644 --- a/apps/server/src/process/SpawnWorker.test.ts +++ b/apps/server/src/process/SpawnWorker.test.ts @@ -77,6 +77,11 @@ describe("SpawnWorker", () => { }); expect(fd3).toBe("fd3 value"); expect(extraCode).toBe(0); + const reader = yield* spawner.spawn( + node("process.stdin.on('end', () => process.stdout.write('eof')).resume()"), + ); + yield* Stream.fail("abandoned input").pipe(Stream.run(reader.stdin), Effect.ignore); + expect(yield* text(reader.stdout)).toBe("eof"); const pipeline = node("process.stdout.write('piped')").pipe( ChildProcess.pipeTo( node("process.stdin.on('data', (d) => process.stdout.write(String(d).toUpperCase()))"), diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts index 3c0b11d728f7e..173a136402f7b 100644 --- a/apps/server/src/process/SpawnWorker.ts +++ b/apps/server/src/process/SpawnWorker.ts @@ -24,6 +24,7 @@ type ToWorker = | { readonly type: "write"; readonly id: number; readonly fd: number; readonly data: Uint8Array } | { readonly type: "end"; readonly id: number; readonly fd: number } | { readonly type: "flow"; readonly id: number; readonly fd: number; readonly paused: boolean } + | { readonly type: "destroy"; readonly id: number; readonly fd: number } | { readonly type: "kill"; readonly id: number; readonly signal: NodeJS.Signals | number }; interface SerializedError { @@ -129,6 +130,9 @@ parentPort.on("message", (request) => { if (request.paused) stream?.pause(); else stream?.resume(); return; + case "destroy": + stream?.destroy(); + return; case "kill": try { child.kill(request.signal); @@ -202,9 +206,14 @@ class WorkerChild this.#finals.set(fd, callback); this.worker.post({ type: "end", id: this.id, fd }); }; - if (fd === 0) return new NodeStream.Writable({ write, final }); - if (fd <= 2) return new NodeStream.Readable({ read }); - return new NodeStream.Duplex({ read, write, final, allowHalfOpen: true }); + // A caller that destroys its end closes the child pipe, as it would with a real pipe. + const destroy = (error: Error | null, callback: (error?: Error | null) => void) => { + if (!this.exited) this.worker.post({ type: "destroy", id: this.id, fd }); + callback(error); + }; + if (fd === 0) return new NodeStream.Writable({ write, final, destroy }); + if (fd <= 2) return new NodeStream.Readable({ read, destroy }); + return new NodeStream.Duplex({ read, write, final, destroy, allowHalfOpen: true }); } onData(fd: number, data: Uint8Array): void { From fc010c95592c9a47a68387ad832fc8b2737defd5 Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 7 Nov 2026 13:54:26 -0500 Subject: [PATCH 11/14] test(server): the server's spawner runs Windows children on a launcher thread Claude Opus 5.5 via Claude Code --- apps/server/src/process/SpawnWorker.test.ts | 33 ++++++++++++++------- 1 file changed, 23 insertions(+), 10 deletions(-) diff --git a/apps/server/src/process/SpawnWorker.test.ts b/apps/server/src/process/SpawnWorker.test.ts index 96d4bc11678fd..762bb8de9dc6c 100644 --- a/apps/server/src/process/SpawnWorker.test.ts +++ b/apps/server/src/process/SpawnWorker.test.ts @@ -11,6 +11,7 @@ import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; import * as ChildProcess from "effect/process/ChildProcess"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; +import * as HostProcess from "@t3tools/shared/HostProcess"; import * as SpawnWorker from "./SpawnWorker.ts"; @@ -21,17 +22,29 @@ const node = (script: string, args: ReadonlyArray = [], options = {}) => ChildProcess.make(process.execPath, ["-e", script, ...args], options); describe("SpawnWorker", () => { - it.live("runs children on a launcher thread, not the server thread", () => + it.live("runs the server's Windows children on a launcher thread, not the server thread", () => Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const child = yield* spawner.spawn( - node("process.stdout.write('ready');process.stdin.resume()"), - ); - expect(yield* child.stdout.pipe(Stream.take(1), text)).toBe("ready"); - // The thread that called spawn owns the native process handle. - expect(process.getActiveResourcesInfo()).not.toContain("ProcessWrap"); - expect(yield* child.isRunning).toBe(true); - }).pipe(Effect.scoped, Effect.provide(layer)), + for (const [platform, onServerThread] of [ + ["win32", false], + ["linux", true], + ] as const) { + const launched = Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const child = yield* spawner.spawn( + node("process.stdout.write('ready');process.stdin.resume()"), + ); + expect(yield* child.stdout.pipe(Stream.take(1), text)).toBe("ready"); + // The thread that called spawn owns the native process handle. + expect(process.getActiveResourcesInfo().includes("ProcessWrap")).toBe(onServerThread); + expect(yield* child.isRunning).toBe(true); + }); + yield* launched.pipe( + Effect.scoped, + Effect.provide(SpawnWorker.layerServer.pipe(Layer.provide(NodeServices.layer))), + Effect.provideService(HostProcess.Platform, platform), + ); + } + }), ); it.live("preserves stdin, large output, env, cwd, literal args and additional descriptors", () => From fc986f5c78e900743649e5e2413939c8467efe5c Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 7 Nov 2026 14:15:40 -0500 Subject: [PATCH 12/14] fix(server): every server launch on Windows reaches the launcher threads NodeHttpServer.layer also outputs NodeServices, so Node's own spawner replaced the worker spawner for every layer above the HTTP server, and the outermost ProcessRunner was built before the server's spawner. Claude Opus 5.5 via Claude Code --- apps/server/src/server.ts | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 8d80006f294fd..ae485c8741a37 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -339,6 +339,9 @@ const layerRelayClient = Layer.unwrap( }), ).pipe(Layer.provide(ManagedPolicy.layer)); +// NodeServices and NodeHttpServer each output Node's own spawner, so each is followed by this one. +const layerSpawner = SpawnWorker.layerServer.pipe(Layer.provide(NodeServices.layer)); + const layerHttpServer = Layer.unwrap( Effect.gen(function* () { const config = yield* ServerConfig.ServerConfig; @@ -375,18 +378,12 @@ const layerHttpServer = Layer.unwrap( // `threshold` when context takeover is disabled). websocket: { perMessageDeflate: true }, }, - ); + ).pipe(Layer.merge(layerSpawner)); }), ).pipe(Layer.provide(ManagedPolicy.layer)); const layerPlatformServices = DocumentText.layer.pipe( - Layer.provideMerge( - Layer.mergeAll( - NodeServices.layer, - NodeCrypto.layer, - SpawnWorker.layerServer.pipe(Layer.provide(NodeServices.layer)), - ), - ), + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer, layerSpawner)), ); const layerPersistence = Layer.empty.pipe(Layer.provideMerge(SqlitePersistence.layerConfig)); @@ -1214,6 +1211,6 @@ const layerMakeServer = Layer.unwrap( Layer.provideMerge(layerPlatformServices), ); }), -).pipe(Layer.provide(ProcessRunner.layer)); +).pipe(Layer.provide(ProcessRunner.layer), Layer.provide(layerSpawner)); export const runServer = Layer.launch(layerMakeServer); From 6877d5e0a15525a016521e13f01b633e7fb33178 Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 7 Nov 2026 14:21:01 -0500 Subject: [PATCH 13/14] docs(server): say a dead launcher's cleanup tries to end the tree Claude Opus 5.5 via Claude Code --- apps/server/src/process/SpawnWorker.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts index 173a136402f7b..b43de9c18ffbd 100644 --- a/apps/server/src/process/SpawnWorker.ts +++ b/apps/server/src/process/SpawnWorker.ts @@ -396,7 +396,7 @@ class SpawnWorker { child.emit("error", error); continue; } - // The worker owned the only handle, so end the whole tree before reporting the exit. + // The worker owned the only handle, so try to end the whole tree before reporting the exit. const finish = () => { child.exit(null, "SIGKILL"); child.abandon(); From 2004169cb9a4b865b15561316b8e371fe2d732fe Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 7 Nov 2026 15:04:53 -0500 Subject: [PATCH 14/14] fix(server): launched children get the server's current environment, not a launcher thread's copy Claude Opus 5.5 via Claude Code --- apps/server/src/process/SpawnWorker.test.ts | 9 ++++++++- apps/server/src/process/SpawnWorker.ts | 3 ++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/apps/server/src/process/SpawnWorker.test.ts b/apps/server/src/process/SpawnWorker.test.ts index 762bb8de9dc6c..210f550c7c4ad 100644 --- a/apps/server/src/process/SpawnWorker.test.ts +++ b/apps/server/src/process/SpawnWorker.test.ts @@ -1,4 +1,4 @@ -// @effect-diagnostics nodeBuiltinImport:off - fixtures need real paths and the process resource list. +// @effect-diagnostics nodeBuiltinImport:off processEnvInEffect:off - fixtures need real paths, the process resource list and the live environment. import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -80,6 +80,13 @@ describe("SpawnWorker", () => { expect(stderr).toBe("b".repeat(1024 * 1024)); expect(code).toBe(0); } + // The launcher threads already exist, so this needs the server's environment at launch time. + process.env.SPAWN_WORKER_LATE = "set after start"; + const late = yield* spawner.spawn( + node("process.stdout.write(process.env.SPAWN_WORKER_LATE)"), + ); + expect(yield* text(late.stdout)).toBe("set after start"); + delete process.env.SPAWN_WORKER_LATE; const extra = yield* spawner.spawn( node("require('node:fs').writeSync(3, 'fd3 value')", [], { additionalFds: { fd3: { type: "output" } }, diff --git a/apps/server/src/process/SpawnWorker.ts b/apps/server/src/process/SpawnWorker.ts index b43de9c18ffbd..81f72ff00739d 100644 --- a/apps/server/src/process/SpawnWorker.ts +++ b/apps/server/src/process/SpawnWorker.ts @@ -321,7 +321,8 @@ class SpawnWorker { id, command, args: [...args], - options: { ...options, stdio }, + // A worker's process.env is a copy from when it started; children get the server's current one. + options: { ...options, stdio, env: options.env ?? { ...process.env } }, }); return child; }