From fec57703817ca3c5a4e8d33b8917a80ed76e5f60 Mon Sep 17 00:00:00 2001 From: sidreyas Date: Sat, 7 Nov 2026 04:54:48 +0530 Subject: [PATCH 1/2] fix(server): Windows Git launches no longer block RPC --- apps/server/src/process/WindowsSpawn.ts | 92 ++++++++ .../process/processCreation.test-fixture.ts | 140 ++++++++++++ .../src/process/processCreationProbe.ts | 82 +++++++ .../src/process/windowsProcessWorker.test.ts | 142 ++++++++++++ .../src/process/windowsProcessWorker.ts | 209 ++++++++++++++++++ apps/server/src/processRunner.test.ts | 34 ++- apps/server/src/processRunner.ts | 55 +++-- apps/server/src/server.ts | 2 + apps/server/src/vcs/GitVcsDriverCore.ts | 42 ++-- 9 files changed, 753 insertions(+), 45 deletions(-) create mode 100644 apps/server/src/process/WindowsSpawn.ts create mode 100644 apps/server/src/process/processCreation.test-fixture.ts create mode 100644 apps/server/src/process/processCreationProbe.ts create mode 100644 apps/server/src/process/windowsProcessWorker.test.ts create mode 100644 apps/server/src/process/windowsProcessWorker.ts diff --git a/apps/server/src/process/WindowsSpawn.ts b/apps/server/src/process/WindowsSpawn.ts new file mode 100644 index 0000000000000..02bea772e0765 --- /dev/null +++ b/apps/server/src/process/WindowsSpawn.ts @@ -0,0 +1,92 @@ +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as PlatformError from "effect/PlatformError"; +import * as Scope from "effect/Scope"; +import * as Sink from "effect/Sink"; +import * as Redacted from "effect/Redacted"; +import * as Stream from "effect/Stream"; +import type * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { makePool } from "./windowsProcessWorker.ts"; + +type Spawn = ( + command: ChildProcess.StandardCommand, +) => Effect.Effect< + Pick, + PlatformError.PlatformError, + Scope.Scope +>; + +export const WindowsSpawn = Context.Reference("t3/WindowsSpawn", { + defaultValue: () => undefined, +}); + +export const layer = Layer.effect( + WindowsSpawn, + Effect.gen(function* () { + if ((yield* HostProcess.Platform) !== "win32") return undefined; + const fs = yield* FileSystem.FileSystem; + const pool = yield* Effect.acquireRelease(Effect.sync(makePool), (pool) => + Effect.promise(() => pool.close()), + ); + const error = (cause: unknown) => + PlatformError.systemError({ + _tag: + typeof cause === "object" && cause !== null && "code" in cause + ? cause.code === "ENOENT" + ? "NotFound" + : cause.code === "EACCES" || cause.code === "EPERM" + ? "PermissionDenied" + : "Unknown" + : "Unknown", + module: "ChildProcess", + method: "spawn", + description: cause instanceof Error ? cause.message : String(cause), + cause, + }); + const spawn: Spawn = Effect.fn("WindowsSpawn.spawn")(function* (command) { + if (command.options.cwd !== undefined) yield* fs.access(command.options.cwd); + const job = yield* Effect.acquireRelease( + Effect.try({ + try: () => + pool.start({ + command: command.command, + args: command.args.map((arg) => + Redacted.isRedacted(arg) ? Redacted.value(arg) : arg, + ), + cwd: command.options.cwd, + env: Object.fromEntries( + Object.entries({ + ...(command.options.env === undefined || command.options.extendEnv === true + ? process.env + : {}), + ...command.options.env, + }).map(([key, value]) => [ + key, + Redacted.isRedacted(value) ? Redacted.value(value) : value, + ]), + ), + shell: command.options.shell ?? false, + detached: command.options.detached, + windowsHide: command.options.windowsHide ?? true, + }), + catch: error, + }), + (job) => Effect.promise(() => job.dispose()), + ); + const streams = yield* Effect.tryPromise({ try: () => job.started, catch: error }); + return { + stdin: Sink.fromWritableStream({ evaluate: () => streams.stdin, onError: error }), + stdout: Stream.fromReadableStream({ evaluate: () => streams.stdout, onError: error }), + stderr: Stream.fromReadableStream({ evaluate: () => streams.stderr, onError: error }), + exitCode: Effect.tryPromise({ try: () => job.exit, catch: error }).pipe( + Effect.map(ChildProcessSpawner.ExitCode), + ), + }; + }); + return spawn; + }), +); diff --git a/apps/server/src/process/processCreation.test-fixture.ts b/apps/server/src/process/processCreation.test-fixture.ts new file mode 100644 index 0000000000000..281cdcf1f4768 --- /dev/null +++ b/apps/server/src/process/processCreation.test-fixture.ts @@ -0,0 +1,140 @@ +/* oxlint-disable unicorn/require-post-message-target-origin -- Node worker ports have no browser origin. */ +// @effect-diagnostics nodeBuiltinImport:off -- an independent worker gates the real native process-creation boundary. +import * as NodeChildProcess from "node:child_process"; +import * as NodeHttp from "node:http"; +import * as NodeModule from "node:module"; +import * as NodeWorkerThreads from "node:worker_threads"; + +const { control, nativeOnly } = NodeWorkerThreads.workerData as { + control: SharedArrayBuffer; + nativeOnly?: boolean; +}; +const barrier = new Int32Array(control); +const parent = NodeWorkerThreads.parentPort!; +const require = NodeModule.createRequire(import.meta.url); +const childProcesses = require("node:child_process") as typeof NodeChildProcess; +const workers = require("node:worker_threads") as typeof NodeWorkerThreads; +const spawn = childProcesses.spawn; +const Worker = workers.Worker; +const postMessage = workers.MessagePort.prototype.postMessage; +workers.MessagePort.prototype.postMessage = function (...args: Parameters) { + if (args[0] === "dispose") parent.postMessage({ type: "disposing" }); + return Reflect.apply(postMessage, this, args); +}; + +// The same native gate follows an eval worker without adding a production test hook. +const workerPrelude = ` +{ + const cp = require("node:child_process"); + const { parentPort, workerData } = require("node:worker_threads"); + const spawn = cp.spawn; + cp.spawn = function(command, args, options) { + if (Array.isArray(args) && args.includes("t3-25245-spawn-gate")) { + parentPort.postMessage({ type: "creation-blocked", serverThread: false }); + Atomics.wait(new Int32Array(workerData.creationGate), 0, 0); + } + const child = spawn(command, args, options); + if (Array.isArray(args) && args.includes("t3-25245-spawn-gate")) { + child.once("exit", () => parentPort.postMessage({ type: "child-exited" })); + } + return child; + }; + require("node:module").syncBuiltinESMExports(); +} +`; + +childProcesses.spawn = ((...args: Parameters) => { + if (Array.isArray(args[1]) && args[1].includes("t3-25245-spawn-gate")) { + parent.postMessage({ type: "creation-blocked", serverThread: true }); + Atomics.wait(barrier, 0, 0); + } + return Reflect.apply(spawn, childProcesses, args); +}) as typeof spawn; + +workers.Worker = class extends Worker { + constructor(filename: string | URL, options?: NodeWorkerThreads.WorkerOptions) { + super( + options?.eval && typeof filename === "string" ? workerPrelude + filename : filename, + options?.eval + ? { ...options, workerData: { ...options.workerData, creationGate: control } } + : options, + ); + this.on("message", (event: { type?: string }) => { + if (event.type === "creation-blocked" || event.type === "child-exited") + parent.postMessage(event); + }); + } +}; +NodeModule.syncBuiltinESMExports(); + +const command = { + command: process.execPath, + args: ["-e", "process.stdout.write('finished')", "t3-25245-spawn-gate"], +}; + +const runner = await (async () => { + if (nativeOnly) { + return { + run: () => + new Promise<{ stdout: string; code: number | null }>((resolve, reject) => { + const child = childProcesses.spawn(command.command, command.args, { windowsHide: true }); + let stdout = ""; + child.stdout.on("data", (chunk: Buffer) => (stdout += chunk.toString())); + child.once("error", reject); + child.once("close", (code) => resolve({ stdout, code })); + }), + dispose: async () => {}, + cancel: async () => {}, + }; + } + const [Effect, Layer, ManagedRuntime, NodeServices, ProcessRunner, Fiber] = await Promise.all([ + import("effect/Effect"), + import("effect/Layer"), + import("effect/ManagedRuntime"), + import("@effect/platform-node/NodeServices"), + import("../processRunner.ts"), + import("effect/Fiber"), + ]); + const runtime = ManagedRuntime.make(ProcessRunner.layer.pipe(Layer.provide(NodeServices.layer))); + let running: ReturnType | undefined; + return { + run: () => { + const fiber = runtime.runFork( + Effect.flatMap(ProcessRunner.ProcessRunner, (service) => service.run(command)), + ); + running = fiber; + return runtime.runPromise(Fiber.join(fiber)); + }, + cancel: () => (running ? runtime.runPromise(Fiber.interrupt(running)) : Promise.resolve()), + dispose: () => runtime.dispose(), + }; +})(); + +const server = NodeHttp.createServer((request, response) => { + if (request.url === "/cancel") { + void runner.cancel().then(() => response.end("cancelled")); + return; + } + if (request.url === "/probe") { + response.end(JSON.stringify({ gateClosed: Atomics.load(barrier, 0) === 0 })); + return; + } + void runner.run().then( + (result) => response.end(JSON.stringify(result)), + (error: unknown) => { + response.statusCode = 500; + response.end(String(error)); + }, + ); +}); +parent.once("message", () => { + void runner.dispose().finally(() => { + server.closeAllConnections(); + server.close(() => parent.close()); + }); +}); +server.listen(0, "127.0.0.1", () => { + const address = server.address(); + if (address === null || typeof address === "string") throw new Error("Missing fixture address"); + parent.postMessage({ type: "listening", port: address.port }); +}); diff --git a/apps/server/src/process/processCreationProbe.ts b/apps/server/src/process/processCreationProbe.ts new file mode 100644 index 0000000000000..e91701ec77933 --- /dev/null +++ b/apps/server/src/process/processCreationProbe.ts @@ -0,0 +1,82 @@ +/* oxlint-disable unicorn/require-post-message-target-origin -- Node worker ports have no browser origin. */ +// @effect-diagnostics nodeBuiltinImport:off -- the test controller must remain independent of a synchronously blocked server. +import * as NodeHttp from "node:http"; +import * as NodeWorkerThreads from "node:worker_threads"; + +const request = (port: number, path: string) => + new Promise((resolve, reject) => { + const request = NodeHttp.get( + { hostname: "127.0.0.1", port, path, agent: false }, + (response) => { + let body = ""; + response.setEncoding("utf8"); + response.on("data", (chunk: string) => (body += chunk)); + response.once("end", () => + response.statusCode === 200 ? resolve(body) : reject(new Error(body)), + ); + response.once("error", reject); + }, + ); + request.once("error", reject); + }); + +export async function probeProcessCreation(nativeOnly = false, cancel = false) { + const control = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT); + const barrier = new Int32Array(control); + const listening = Promise.withResolvers(); + const blocked = Promise.withResolvers(); + const stopped = Promise.withResolvers(); + const disposing = Promise.withResolvers(); + const childExited = Promise.withResolvers(); + const worker = new NodeWorkerThreads.Worker( + new URL("./processCreation.test-fixture.ts", import.meta.url), + { + workerData: { control, nativeOnly }, + execArgv: [], + }, + ); + worker.on("message", (event: { type: string; port: number; serverThread: boolean }) => { + if (event.type === "listening") listening.resolve(event.port); + if (event.type === "creation-blocked") blocked.resolve(event.serverThread); + if (event.type === "disposing") disposing.resolve(); + if (event.type === "child-exited") childExited.resolve(); + }); + worker.once("error", (error) => { + listening.reject(error); + blocked.reject(error); + }); + worker.once("exit", () => stopped.resolve()); + const release = () => { + Atomics.store(barrier, 0, 1); + Atomics.notify(barrier, 0); + }; + let running: Promise | undefined; + // A startup failure can happen before the controller starts waiting for the gate. + void blocked.promise.catch(() => {}); + try { + const port = await listening.promise; + running = request(port, "/run"); + void running.catch(blocked.reject); + const serverThread = await blocked.promise; + // Native creation on the server cannot service the probe; release it so main fails without a timeout. + if (serverThread) release(); + const probe = JSON.parse(await request(port, "/probe")) as { gateClosed: boolean }; + if (cancel && !serverThread) { + const cancellation = request(port, "/cancel"); + await disposing.promise; + release(); + await cancellation; + await childExited.promise; + await running.catch(() => {}); + return { serverThread, ...probe, stdout: "cancelled", code: null }; + } + release(); + const result = JSON.parse(await running) as { stdout: string; code: number | null }; + return { serverThread, ...probe, stdout: result.stdout, code: result.code }; + } finally { + release(); + await running?.catch(() => {}); + worker.postMessage("stop"); + await stopped.promise; + } +} diff --git a/apps/server/src/process/windowsProcessWorker.test.ts b/apps/server/src/process/windowsProcessWorker.test.ts new file mode 100644 index 0000000000000..f3934ed213076 --- /dev/null +++ b/apps/server/src/process/windowsProcessWorker.test.ts @@ -0,0 +1,142 @@ +// @effect-diagnostics nodeBuiltinImport:off -- verifies real child streams and process-tree ownership. +import * as NodeNet from "node:net"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Redacted from "effect/Redacted"; +import * as Stream from "effect/Stream"; +import * as ChildProcess from "effect/process/ChildProcess"; +import { makePool } from "./windowsProcessWorker.ts"; +import * as WindowsSpawn from "./WindowsSpawn.ts"; + +async function text(stream: ReadableStream) { + const chunks: Uint8Array[] = []; + for await (const chunk of stream) chunks.push(chunk); + return Buffer.concat(chunks).toString(); +} + +// quarantine: #25245 Windows process-tree termination uses taskkill. +describe.skipIf(HostProcess.Platform.defaultValue() !== "win32")("Windows process worker", () => { + it.live("preserves inherited, replaced and removed environment values", () => + Effect.gen(function* () { + const spawn = yield* WindowsSpawn.WindowsSpawn; + if (spawn === undefined) return expect.fail("Missing Windows spawn service"); + const environment = yield* HostProcess.Environment; + expect(environment.NODE_ENV).toBeDefined(); + const env = { T3_WORKER_TEST: Redacted.make("child") }; + for (const [options, expected] of [ + [{}, [environment.NODE_ENV, null]], + [{ env }, [null, "child"]], + [{ env, extendEnv: true }, [environment.NODE_ENV, "child"]], + [{ env: { ...env, NODE_ENV: undefined }, extendEnv: true }, [null, "child"]], + ] as const) { + yield* Effect.scoped( + Effect.gen(function* () { + const child = yield* spawn( + ChildProcess.make( + process.execPath, + [ + "-e", + "process.stdout.write(JSON.stringify([process.env.NODE_ENV, process.env.T3_WORKER_TEST]))", + ], + options, + ), + ); + const stdout = yield* Stream.mkString(Stream.decodeText(child.stdout)); + expect(JSON.parse(stdout)).toEqual(expected); + expect(yield* child.exitCode).toBe(0); + }), + ); + } + }).pipe(Effect.provide(WindowsSpawn.layer.pipe(Layer.provide(NodeServices.layer)))), + ); + + it.live("preserves concurrent streams, stdin, exit status and reuse after spawn errors", () => + Effect.promise(async () => { + const pool = makePool(); + const input = { command: process.execPath, args: [], env: process.env, shell: false }; + try { + const missing = pool.start({ ...input, command: "t3-missing-executable-25245" }); + await expect(missing.started).rejects.toMatchObject({ code: "ENOENT" }); + await missing.dispose(); + await Promise.all( + [7, 9].map(async (code) => { + const payload = "x".repeat(2 * 1024 * 1024); + const job = pool.start({ + ...input, + args: [ + "-e", + `process.stdin.pipe(process.stdout); process.stdin.once('end', () => { process.stderr.write(process.env.T3_WORKER_TEST); process.exitCode = ${code}; });`, + ], + env: { ...process.env, T3_WORKER_TEST: "stderr" }, + }); + try { + const streams = await job.started; + const output = text(streams.stdout); + const error = text(streams.stderr); + const writer = streams.stdin.getWriter(); + await writer.write(Buffer.from(payload)); + await writer.close(); + writer.releaseLock(); + expect(await output).toBe(payload); + expect(await error).toBe("stderr"); + expect(await job.exit).toBe(code); + } finally { + await job.dispose(); + } + }), + ); + } finally { + await pool.close(); + } + }), + ); + + it.live("waits for tree cleanup when disposing a running child", () => + Effect.promise(async () => { + const pool = makePool(); + const connected = Promise.withResolvers(); + const disconnected = Promise.withResolvers(); + const sockets = new Set(); + const server = NodeNet.createServer((socket) => { + sockets.add(socket); + socket.on("error", () => {}); + socket.once("data", () => connected.resolve()); + socket.once("close", () => { + sockets.delete(socket); + disconnected.resolve(); + }); + }); + server.listen(0, "127.0.0.1"); + await new Promise((resolve, reject) => { + server.once("listening", resolve); + server.once("error", reject); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing test address"); + const descendant = `const s=require('node:net').connect(${address.port},'127.0.0.1',()=>s.write('ready')); s.once('close',()=>process.exit());`; + const job = pool.start({ + command: process.execPath, + args: [ + "-e", + `require('node:child_process').spawn(process.execPath,['-e',${JSON.stringify(descendant)}],{stdio:'ignore',windowsHide:true}); process.stdin.resume();`, + ], + env: process.env, + shell: false, + }); + try { + await job.started; + await connected.promise; + await job.dispose(); + await disconnected.promise; + } finally { + for (const socket of sockets) socket.destroy(); + await job.dispose(); + await pool.close(); + await new Promise((resolve) => server.close(() => resolve())); + } + }), + ); +}); diff --git a/apps/server/src/process/windowsProcessWorker.ts b/apps/server/src/process/windowsProcessWorker.ts new file mode 100644 index 0000000000000..6a9445c0c0427 --- /dev/null +++ b/apps/server/src/process/windowsProcessWorker.ts @@ -0,0 +1,209 @@ +/* oxlint-disable unicorn/require-post-message-target-origin -- Node worker ports have no browser origin. */ +// @effect-diagnostics nodeBuiltinImport:off -- native creation and tree cleanup must stay off the server event loop. +import * as NodeWorkerThreads from "node:worker_threads"; + +interface SpawnInput { + readonly command: string; + readonly args: ReadonlyArray; + readonly cwd?: string | undefined; + readonly env: NodeJS.ProcessEnv; + readonly shell: boolean | string; + readonly detached?: boolean | undefined; + readonly windowsHide?: boolean | undefined; +} + +interface Streams { + readonly stdin: WritableStream; + readonly stdout: ReadableStream; + readonly stderr: ReadableStream; +} + +type Reply = + | ({ readonly type: "spawned" } & Streams) + | { readonly type: "exit"; readonly code: number | null; readonly signal: string | null } + | { readonly type: "error"; readonly error: Error; readonly code?: string } + | { readonly type: "released" }; + +// Inline source also runs in Electron, bundled CLI and single-executable builds. +const source = ` +const { parentPort } = require("node:worker_threads"); +const { spawn, execFile } = require("node:child_process"); +const { Readable, Writable } = require("node:stream"); +parentPort.on("message", ({ input, port, control, stop }) => { + if (stop) { parentPort.close(); return; } + const state = new Int32Array(control); + let child; + let disposing = false; + let exited; + const exit = new Promise(resolve => { exited = resolve; }); + const report = error => port.postMessage({ type: "error", error, code: error.code }); + const dispose = async () => { + if (disposing) return; + disposing = true; + if (child && child.pid && child.exitCode === null && child.signalCode === null) { + await new Promise(resolve => { + execFile("taskkill", ["/pid", String(child.pid), "/T", "/F"], { windowsHide: true }, error => { + if (error && child.exitCode === null && child.signalCode === null) child.kill(); + resolve(); + }); + }); + await exit; + } + child?.stdin?.destroy(); + child?.stdout?.destroy(); + child?.stderr?.destroy(); + port.postMessage({ type: "released" }); + port.close(); + }; + port.on("message", () => { void dispose().catch(report); }); + try { + child = spawn(input.command, input.args, { + cwd: input.cwd, env: input.env, shell: input.shell, windowsHide: input.windowsHide ?? true, + detached: input.detached, + }); + Atomics.store(state, 0, child.pid || 0); + child.once("exit", (code, signal) => { + Atomics.store(state, 0, 0); + exited(); + port.postMessage({ type: "exit", code, signal }); + }); + child.on("error", report); + child.stdin.on("error", () => {}); + child.once("spawn", () => { + if (Atomics.load(state, 1)) { void dispose().catch(report); return; } + const options = { strategy: { highWaterMark: 65536, size: chunk => chunk.byteLength } }; + const stdout = Readable.toWeb(child.stdout, options); + const stderr = Readable.toWeb(child.stderr, options); + const stdin = Writable.toWeb(child.stdin); + port.postMessage({ type: "spawned", stdin, stdout, stderr }, [stdin, stdout, stderr]); + }); + } catch (error) { report(error); } +}); +`; + +const cleanupSource = ` +const { workerData, parentPort } = require("node:worker_threads"); +const state = new Int32Array(workerData); +const pid = Atomics.load(state, 0); +if (pid) { + require("node:child_process").execFile("taskkill", ["/pid", String(pid), "/T", "/F"], { windowsHide: true }, error => { + parentPort.postMessage(error || null); + parentPort.close(); + }); +} else { parentPort.postMessage(null); parentPort.close(); } +`; + +const cleanupOrphan = (control: SharedArrayBuffer) => + new Promise((resolve, reject) => { + const worker = new NodeWorkerThreads.Worker(cleanupSource, { eval: true, workerData: control }); + let failure: Error | undefined; + worker.on("message", (error: Error | null) => { + failure = error ?? undefined; + }); + worker.once("error", reject); + worker.once("exit", () => (failure ? reject(failure) : resolve())); + }); + +export function makePool() { + const jobs = new Set<{ dispose: () => Promise }>(); + let idle: NodeWorkerThreads.Worker | undefined; + let closed = false; + + const stop = (worker: NodeWorkerThreads.Worker) => + new Promise((resolve) => { + worker.once("exit", () => resolve()); + worker.ref(); + worker.postMessage({ stop: true }); + }); + + const start = (input: SpawnInput) => { + if (closed) throw new Error("Process worker pool is closed"); + const worker = idle ?? new NodeWorkerThreads.Worker(source, { eval: true }); + if (!idle) { + worker.on("error", () => {}); + worker.once("exit", () => { + if (idle === worker) idle = undefined; + }); + } + idle = undefined; + worker.ref(); + const { port1, port2 } = new NodeWorkerThreads.MessageChannel(); + const control = new SharedArrayBuffer(2 * Int32Array.BYTES_PER_ELEMENT); + const state = new Int32Array(control); + const started = Promise.withResolvers(); + const exit = Promise.withResolvers(); + const released = Promise.withResolvers(); + // Failure can precede acquisition of either result by the Effect consumer. + void started.promise.catch(() => {}); + void exit.promise.catch(() => {}); + void released.promise.catch(() => {}); + let dead = false; + let disposal: Promise | undefined; + const fail = (error: Error) => { + started.reject(error); + exit.reject(error); + }; + const onError = (error: Error) => fail(error); + const onExit = (code: number) => { + dead = true; + fail(new Error(`Process worker exited unexpectedly (${code})`)); + void cleanupOrphan(control).then(released.resolve, released.reject); + }; + worker.on("error", onError); + worker.once("exit", onExit); + port1.on("message", (reply: Reply) => { + switch (reply.type) { + case "spawned": + started.resolve(reply); + break; + case "exit": + if (reply.code === null) exit.reject(new Error(`Process interrupted by ${reply.signal}`)); + else exit.resolve(reply.code); + break; + case "error": + fail(Object.assign(reply.error, { code: reply.code })); + break; + case "released": + started.reject(new Error("Process creation cancelled")); + released.resolve(); + break; + } + }); + const job = { + started: started.promise, + exit: exit.promise, + dispose: () => + (disposal ??= (async () => { + Atomics.store(state, 1, 1); + port1.postMessage("dispose"); + await released.promise; + port1.close(); + worker.off("error", onError); + worker.off("exit", onExit); + jobs.delete(job); + if (dead) return; + if (closed || idle) { + await stop(worker); + } else { + idle = worker; + worker.unref(); + } + })()), + }; + jobs.add(job); + worker.postMessage({ input, port: port2, control }, [port2]); + return job; + }; + + return { + start, + close: async () => { + closed = true; + await Promise.all([...jobs].map((job) => job.dispose())); + if (idle) { + await stop(idle); + idle = undefined; + } + }, + }; +} diff --git a/apps/server/src/processRunner.test.ts b/apps/server/src/processRunner.test.ts index 3754f25a4c775..c6acbfd9f02af 100644 --- a/apps/server/src/processRunner.test.ts +++ b/apps/server/src/processRunner.test.ts @@ -11,6 +11,7 @@ import * as HostProcess from "@t3tools/shared/HostProcess"; import { SpawnExecutableResolution } from "@t3tools/shared/shell"; import * as ProcessRunner from "./processRunner.ts"; +import { probeProcessCreation } from "./process/processCreationProbe.ts"; type ChildProcessCommand = { readonly command: string; @@ -70,13 +71,44 @@ const runWith = }), ), Effect.provide( - ProcessRunner.layer.pipe( + Layer.effect(ProcessRunner.ProcessRunner, ProcessRunner.make()).pipe( Layer.provide(Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner)), ), ), ); describe("runProcess", () => { + // quarantine: #25245 this regression exercises Windows child ownership during native creation. + it.live.skipIf(HostProcess.Platform.defaultValue() !== "win32")( + "cleans up a child cancelled during Windows native creation", + () => + Effect.promise(() => probeProcessCreation(false, true)).pipe( + Effect.map((result) => { + expect(result).toEqual({ + serverThread: false, + gateClosed: true, + stdout: "cancelled", + code: null, + }); + }), + ), + ); + // quarantine: #25245 this regression exercises the Windows process-creation boundary. + it.live.skipIf(HostProcess.Platform.defaultValue() !== "win32")( + "answers requests while Windows native process creation is blocked", + () => + Effect.promise(() => probeProcessCreation()).pipe( + Effect.map((result) => { + expect(result).toEqual({ + serverThread: false, + gateClosed: true, + stdout: "finished", + code: 0, + }); + }), + ), + ); + it.effect("resolves and escapes Windows command shims before spawning", () => { const spawner = makeSpawner((command) => Effect.sync(() => { diff --git a/apps/server/src/processRunner.ts b/apps/server/src/processRunner.ts index a345c67eca881..2948c39fb2306 100644 --- a/apps/server/src/processRunner.ts +++ b/apps/server/src/processRunner.ts @@ -10,6 +10,7 @@ import * as Stream from "effect/Stream"; import * as ChildProcess from "effect/process/ChildProcess"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import * as WindowsSpawn from "./process/WindowsSpawn.ts"; import { collectUint8StreamText, decodeUtf8, @@ -263,8 +264,16 @@ function finalizeRunProcess( const commandName = (command: string) => command.replace(/^.*[\\/]/, ""); +type Spawn = ( + command: ChildProcess.StandardCommand, +) => Effect.Effect< + Pick, + PlatformError.PlatformError, + Scope.Scope +>; + const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* ( - spawner: ChildProcessSpawner.ChildProcessSpawner["Service"], + spawn: Spawn, input: ProcessRunInput, ): Effect.fn.Return { yield* Effect.annotateCurrentSpan("process.command", commandName(input.command)); @@ -290,25 +299,23 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* ( cause, }); - const child = yield* spawner - .spawn( - ChildProcess.make(spawnCommand.command, spawnCommand.args, { - ...((input.spawnCwd ?? input.cwd) ? { cwd: input.spawnCwd ?? input.cwd } : {}), - ...(input.env !== undefined - ? { - env: input.env, - extendEnv, - } - : {}), - shell: spawnCommand.shell, - forceKillAfter: FORCE_KILL_AFTER, - }), - ) - .pipe( - Effect.mapError(toSpawnError), - // Synchronous spawn throws arrive as defects; recover them too so callers see typed spawn failures. - Effect.catchDefect((cause) => Effect.fail(toSpawnError(cause))), - ); + const child = yield* spawn( + ChildProcess.make(spawnCommand.command, spawnCommand.args, { + ...((input.spawnCwd ?? input.cwd) ? { cwd: input.spawnCwd ?? input.cwd } : {}), + ...(input.env !== undefined + ? { + env: input.env, + extendEnv, + } + : {}), + shell: spawnCommand.shell, + forceKillAfter: FORCE_KILL_AFTER, + }), + ).pipe( + Effect.mapError(toSpawnError), + // Synchronous spawn throws arrive as defects; recover them too so callers see typed spawn failures. + Effect.catchDefect((cause) => Effect.fail(toSpawnError(cause))), + ); const stdin = input.stdin; const onStdoutChunk = input.onStdoutChunk; @@ -389,13 +396,15 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* ( /** @public */ export const make = Effect.fn("ProcessRunner.make")(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const spawn = (yield* WindowsSpawn.WindowsSpawn) ?? spawner.spawn; - const run: ProcessRunner["Service"]["run"] = (input) => - finalizeRunProcess(runProcessCore(spawner, input), input); + const run = Effect.fn("ProcessRunner.run")((input: ProcessRunInput) => + finalizeRunProcess(runProcessCore(spawn, input), input), + ); return ProcessRunner.of({ run, }); }); -export const layer = Layer.effect(ProcessRunner, make()); +export const layer = Layer.effect(ProcessRunner, make()).pipe(Layer.provide(WindowsSpawn.layer)); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 5c9ab9c433ffc..422e636c0c332 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -94,6 +94,7 @@ import * as DesktopBrowserChannel from "./preview/DesktopBrowserChannel.ts"; import * as ServerBrowserStream from "./preview/ServerBrowserStream.ts"; import * as PreviewBrowser from "./preview/PreviewBrowser.ts"; import * as ProcessRunner from "./processRunner.ts"; +import * as WindowsSpawn from "./process/WindowsSpawn.ts"; import * as GitManager from "./git/GitManager.ts"; import * as EnvironmentTheme from "./environmentTheme.ts"; import * as Keybindings from "./keybindings.ts"; @@ -379,6 +380,7 @@ const layerHttpServer = Layer.unwrap( ).pipe(Layer.provide(ManagedPolicy.layer)); const layerPlatformServices = DocumentText.layer.pipe( + Layer.provideMerge(WindowsSpawn.layer), Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), ); diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index 85a7e524b1a63..27c9756a90938 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -20,6 +20,7 @@ import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import * as WindowsSpawn from "../process/WindowsSpawn.ts"; import { GitCommandError, @@ -945,6 +946,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; const commandSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const spawn = (yield* WindowsSpawn.WindowsSpawn) ?? commandSpawner.spawn; const { worktreesDir } = yield* ServerConfig.ServerConfig; const isServerInternalPath = yield* makeIsServerInternalPath; const digest = yield* Digest.Digest; @@ -988,27 +990,25 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const resolved = yield* resolveSpawnCommand("git", [], { env: spawnEnv }); // cmd wrappers truncate multiline commits; leave git.cmd handling to Node lookup. const executable = resolved.shell ? "git" : resolved.command; - const child = yield* commandSpawner - .spawn( - ChildProcess.make(executable, commandInput.args, { - cwd: commandInput.cwd, - env: spawnEnv, - forceKillAfter: input.forceKillAfterMs, - ...(runsGitWithoutConsole(hostPlatform, executable, commandInput.args) - ? { detached: true, windowsHide: true } - : {}), - }), - ) - .pipe( - Effect.mapError( - (cause) => - new GitCommandError({ - ...gitCommandContext(commandInput), - detail: "Failed to spawn Git process.", - cause, - }), - ), - ); + const child = yield* spawn( + ChildProcess.make(executable, commandInput.args, { + cwd: commandInput.cwd, + env: spawnEnv, + forceKillAfter: input.forceKillAfterMs, + ...(runsGitWithoutConsole(hostPlatform, executable, commandInput.args) + ? { detached: true, windowsHide: true } + : {}), + }), + ).pipe( + Effect.mapError( + (cause) => + new GitCommandError({ + ...gitCommandContext(commandInput), + detail: "Failed to spawn Git process.", + cause, + }), + ), + ); const [stdout, stderr, exitCode] = yield* Effect.all( [ From df3b2c4469af04dabac3f8a174c5c5f0bf10e896 Mon Sep 17 00:00:00 2001 From: sidreyas Date: Sat, 7 Nov 2026 06:45:58 +0530 Subject: [PATCH 2/2] fix(server): bound cleanup waits without releasing live workers --- .../process/processCleanup.test-fixture.ts | 166 ++++++++++++++++++ .../process/processCreation.test-fixture.ts | 25 ++- .../src/process/processCreationProbe.ts | 32 +++- .../src/process/windowsProcessWorker.test.ts | 128 +++++++++----- .../src/process/windowsProcessWorker.ts | 47 +++-- apps/server/src/processRunner.test.ts | 32 +++- 6 files changed, 351 insertions(+), 79 deletions(-) create mode 100644 apps/server/src/process/processCleanup.test-fixture.ts diff --git a/apps/server/src/process/processCleanup.test-fixture.ts b/apps/server/src/process/processCleanup.test-fixture.ts new file mode 100644 index 0000000000000..43be7e4fa1a8d --- /dev/null +++ b/apps/server/src/process/processCleanup.test-fixture.ts @@ -0,0 +1,166 @@ +/* oxlint-disable unicorn/require-post-message-target-origin -- Node worker ports have no browser origin. */ +// @effect-diagnostics nodeBuiltinImport:off -- gates native termination in an isolated worker with real children. +import * as NodeModule from "node:module"; +import * as NodeTest from "node:test"; +import * as NodeWorkerThreads from "node:worker_threads"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; + +const { mode, descendant } = NodeWorkerThreads.workerData as { + mode: "normal" | "taskkill" | "child-exit"; + descendant: string; +}; +const parent = NodeWorkerThreads.parentPort!; +const require = NodeModule.createRequire(import.meta.url); +const workers = require("node:worker_threads") as typeof NodeWorkerThreads; +const Worker = workers.Worker; +const MessageChannel = workers.MessageChannel; +const owners: NodeWorkerThreads.Worker[] = []; +const gates: NodeWorkerThreads.MessagePort[] = []; +const channels: NodeWorkerThreads.MessagePort[] = []; +const blocked = Promise.withResolvers(); +void blocked.promise.catch(() => {}); +const prelude = `{ + const cp = require("node:child_process"); + const { parentPort, workerData } = require("node:worker_threads"); + const spawn = cp.spawn, execFile = cp.execFile; + let held, pid; + cp.spawn = function(command, args, options) { + const child = spawn(command, args, options); + if (args.includes("t3-25245-cleanup-gate")) { + pid = child.pid; + if (workerData.mode === "child-exit") child.kill = () => { + parentPort.postMessage("blocked"); + return false; + }; + } + return child; + }; + cp.execFile = function(command, args, options, callback) { + if (command !== "taskkill" || args[1] !== String(pid) || workerData.mode === "normal") + return execFile(command, args, options, callback); + held = () => execFile(command, args, options, workerData.mode === "taskkill" ? callback : () => {}); + if (workerData.mode === "taskkill") parentPort.postMessage("blocked"); + else callback(new Error("Fixture taskkill failed")); + }; + workerData.cleanupControl.on("message", () => { const release = held; held = undefined; release?.(); }); + workerData.cleanupControl.unref(); +}`; +workers.Worker = class extends Worker { + constructor(filename: string | URL, options?: NodeWorkerThreads.WorkerOptions) { + const { port1, port2 } = new MessageChannel(); + super(typeof filename === "string" && options?.eval ? prelude + filename : filename, { + ...options, + workerData: { ...options?.workerData, mode, cleanupControl: port2 }, + transferList: [...(options?.transferList ?? []), port2], + }); + gates.push(port1); + port1.unref(); + owners.push(this); + this.on("message", (message: unknown) => { + if (message === "blocked") blocked.resolve(); + }); + this.once("error", blocked.reject); + this.once("exit", () => blocked.reject(new Error("Cleanup owner exited before termination"))); + } +}; +workers.MessageChannel = class extends MessageChannel { + constructor() { + super(); + channels.push(this.port1); + } +}; +NodeModule.syncBuiltinESMExports(); + +const { makePool } = await import("./windowsProcessWorker.ts"); +const pool = makePool(); +const input = { command: process.execPath, env: process.env, shell: false }; +const job = pool.start({ + ...input, + args: [ + "-e", + `require('node:child_process').spawn(process.execPath,['-e',${JSON.stringify(descendant)}],{stdio:'ignore',windowsHide:true}); process.stdin.resume();`, + "t3-25245-cleanup-gate", + ], +}); +const owner = owners[0]!; +const released = new Promise((resolve) => channels[0]!.once("close", () => resolve())); +const ownerStopped = new Promise((resolve) => owner.once("exit", () => resolve())); +let exited = false; +void job.exit.then( + () => { + exited = true; + }, + () => { + exited = true; + }, +); +let other: ReturnType | undefined; +try { + await job.started; + parent.postMessage({ type: "started" }); + await new Promise((resolve) => parent.once("message", () => resolve())); + const acquired = Promise.withResolvers(); + const running = Effect.runFork( + Effect.acquireRelease( + Effect.sync(() => { + acquired.resolve(); + return job; + }), + (job) => Effect.promise(() => job.dispose()), + ).pipe(Effect.andThen(Effect.never), Effect.scoped), + ); + await acquired.promise; + NodeTest.mock.timers.enable({ apis: ["setTimeout"] }); + let settled = false; + const disposal = Effect.runPromise(Fiber.interrupt(running)).then(() => { + settled = true; + }); + if (mode === "normal") await disposal; + else { + await blocked.promise; + NodeTest.mock.timers.runAll(); + } + NodeTest.mock.timers.reset(); + other = pool.start({ ...input, args: ["-e", "process.stdin.resume()"] }); + await other.started; + const separateOwner = owners.length === 2; + let closed = false; + if (mode === "child-exit" && settled) { + await pool.close(); + closed = true; + } + parent.postMessage({ + type: "disposed", + settled, + exited, + separateOwner, + live: owner.threadId !== -1, + closed, + }); + await new Promise((resolve) => parent.once("message", () => resolve())); + if (mode !== "normal") gates[0]!.postMessage("release"); + await released; + await disposal; + let reused = false; + if (mode === "taskkill") { + const count = owners.length; + const next = pool.start({ ...input, args: ["-e", ""] }); + await next.started; + await next.exit; + reused = owners.length === count; + await next.dispose(); + } + await other.dispose(); + await pool.close(); + await ownerStopped; + parent.postMessage({ type: "cleaned", reused }); +} finally { + NodeTest.mock.timers.reset(); + if (mode !== "normal") gates[0]!.postMessage("release"); + await other?.dispose(); + await job.dispose(); + await pool.close(); + for (const gate of gates) gate.close(); + parent.close(); +} diff --git a/apps/server/src/process/processCreation.test-fixture.ts b/apps/server/src/process/processCreation.test-fixture.ts index 281cdcf1f4768..fb3295f758517 100644 --- a/apps/server/src/process/processCreation.test-fixture.ts +++ b/apps/server/src/process/processCreation.test-fixture.ts @@ -5,20 +5,31 @@ import * as NodeHttp from "node:http"; import * as NodeModule from "node:module"; import * as NodeWorkerThreads from "node:worker_threads"; -const { control, nativeOnly } = NodeWorkerThreads.workerData as { +const { control, nativeOnly, failure } = NodeWorkerThreads.workerData as { control: SharedArrayBuffer; nativeOnly?: boolean; + failure?: { at: "disposing" | "child-exited"; mode: "error" | "exit" }; }; const barrier = new Int32Array(control); const parent = NodeWorkerThreads.parentPort!; +const childExit = Promise.withResolvers(); const require = NodeModule.createRequire(import.meta.url); const childProcesses = require("node:child_process") as typeof NodeChildProcess; const workers = require("node:worker_threads") as typeof NodeWorkerThreads; const spawn = childProcesses.spawn; const Worker = workers.Worker; const postMessage = workers.MessagePort.prototype.postMessage; +const failAt = (at: "disposing" | "child-exited") => { + if (failure?.at !== at) return; + if (failure.mode === "exit") process.exit(17); + throw new Error(`Fixture worker failed at ${at}`); +}; workers.MessagePort.prototype.postMessage = function (...args: Parameters) { - if (args[0] === "dispose") parent.postMessage({ type: "disposing" }); + if (args[0] === "dispose") { + // Throw outside the Effect finalizer so this is a native worker failure. + if (failure?.at === "disposing") queueMicrotask(() => failAt("disposing")); + else parent.postMessage({ type: "disposing" }); + } return Reflect.apply(postMessage, this, args); }; @@ -60,6 +71,10 @@ workers.Worker = class extends Worker { : options, ); this.on("message", (event: { type?: string }) => { + if (event.type === "child-exited" && failure?.at === "child-exited") { + childExit.resolve(); + return; + } if (event.type === "creation-blocked" || event.type === "child-exited") parent.postMessage(event); }); @@ -127,7 +142,11 @@ const server = NodeHttp.createServer((request, response) => { }, ); }); -parent.once("message", () => { +parent.on("message", (message: string) => { + if (message === "fail-child-exit") { + void childExit.promise.then(() => queueMicrotask(() => failAt("child-exited"))); + return; + } void runner.dispose().finally(() => { server.closeAllConnections(); server.close(() => parent.close()); diff --git a/apps/server/src/process/processCreationProbe.ts b/apps/server/src/process/processCreationProbe.ts index e91701ec77933..27f022cc61985 100644 --- a/apps/server/src/process/processCreationProbe.ts +++ b/apps/server/src/process/processCreationProbe.ts @@ -20,7 +20,11 @@ const request = (port: number, path: string) => request.once("error", reject); }); -export async function probeProcessCreation(nativeOnly = false, cancel = false) { +export async function probeProcessCreation( + nativeOnly = false, + cancel = false, + failure?: { at: "disposing" | "child-exited"; mode: "error" | "exit" }, +) { const control = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT); const barrier = new Int32Array(control); const listening = Promise.withResolvers(); @@ -28,10 +32,17 @@ export async function probeProcessCreation(nativeOnly = false, cancel = false) { const stopped = Promise.withResolvers(); const disposing = Promise.withResolvers(); const childExited = Promise.withResolvers(); + const milestones = [listening, blocked, disposing, childExited]; + // A worker failure can precede acquisition of any of these milestones. + for (const milestone of milestones) void milestone.promise.catch(() => {}); + const fail = (error: Error) => { + for (const milestone of milestones) milestone.reject(error); + }; + let stopping = false; const worker = new NodeWorkerThreads.Worker( new URL("./processCreation.test-fixture.ts", import.meta.url), { - workerData: { control, nativeOnly }, + workerData: { control, nativeOnly, failure }, execArgv: [], }, ); @@ -41,18 +52,17 @@ export async function probeProcessCreation(nativeOnly = false, cancel = false) { if (event.type === "disposing") disposing.resolve(); if (event.type === "child-exited") childExited.resolve(); }); - worker.once("error", (error) => { - listening.reject(error); - blocked.reject(error); + worker.once("error", fail); + worker.once("exit", (code) => { + if (!stopping) fail(new Error(`Fixture worker exited unexpectedly (${code})`)); + stopped.resolve(); }); - worker.once("exit", () => stopped.resolve()); const release = () => { Atomics.store(barrier, 0, 1); Atomics.notify(barrier, 0); }; let running: Promise | undefined; - // A startup failure can happen before the controller starts waiting for the gate. - void blocked.promise.catch(() => {}); + let cancellation: Promise | undefined; try { const port = await listening.promise; running = request(port, "/run"); @@ -62,10 +72,12 @@ export async function probeProcessCreation(nativeOnly = false, cancel = false) { if (serverThread) release(); const probe = JSON.parse(await request(port, "/probe")) as { gateClosed: boolean }; if (cancel && !serverThread) { - const cancellation = request(port, "/cancel"); + cancellation = request(port, "/cancel"); + void cancellation.catch(() => {}); await disposing.promise; release(); await cancellation; + if (failure?.at === "child-exited") worker.postMessage("fail-child-exit"); await childExited.promise; await running.catch(() => {}); return { serverThread, ...probe, stdout: "cancelled", code: null }; @@ -76,6 +88,8 @@ export async function probeProcessCreation(nativeOnly = false, cancel = false) { } finally { release(); await running?.catch(() => {}); + await cancellation?.catch(() => {}); + stopping = true; worker.postMessage("stop"); await stopped.promise; } diff --git a/apps/server/src/process/windowsProcessWorker.test.ts b/apps/server/src/process/windowsProcessWorker.test.ts index f3934ed213076..84a08803c197b 100644 --- a/apps/server/src/process/windowsProcessWorker.test.ts +++ b/apps/server/src/process/windowsProcessWorker.test.ts @@ -1,5 +1,6 @@ // @effect-diagnostics nodeBuiltinImport:off -- verifies real child streams and process-tree ownership. import * as NodeNet from "node:net"; +import * as NodeWorkerThreads from "node:worker_threads"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { describe, expect, it } from "@effect/vitest"; @@ -94,49 +95,90 @@ describe.skipIf(HostProcess.Platform.defaultValue() !== "win32")("Windows proces }), ); - it.live("waits for tree cleanup when disposing a running child", () => - Effect.promise(async () => { - const pool = makePool(); - const connected = Promise.withResolvers(); - const disconnected = Promise.withResolvers(); - const sockets = new Set(); - const server = NodeNet.createServer((socket) => { - sockets.add(socket); - socket.on("error", () => {}); - socket.once("data", () => connected.resolve()); - socket.once("close", () => { - sockets.delete(socket); - disconnected.resolve(); + it.live.each(["normal", "taskkill", "child-exit"] as const)( + "retains tree cleanup ownership when disposing a running child (%s)", + (mode) => + Effect.promise(async () => { + const connected = Promise.withResolvers(); + const disconnected = Promise.withResolvers(); + const sockets = new Set(); + const server = NodeNet.createServer((socket) => { + sockets.add(socket); + socket.on("error", () => {}); + socket.once("data", () => connected.resolve()); + socket.once("close", () => { + sockets.delete(socket); + disconnected.resolve(); + }); }); - }); - server.listen(0, "127.0.0.1"); - await new Promise((resolve, reject) => { - server.once("listening", resolve); - server.once("error", reject); - }); - const address = server.address(); - if (!address || typeof address === "string") throw new Error("Missing test address"); - const descendant = `const s=require('node:net').connect(${address.port},'127.0.0.1',()=>s.write('ready')); s.once('close',()=>process.exit());`; - const job = pool.start({ - command: process.execPath, - args: [ - "-e", - `require('node:child_process').spawn(process.execPath,['-e',${JSON.stringify(descendant)}],{stdio:'ignore',windowsHide:true}); process.stdin.resume();`, - ], - env: process.env, - shell: false, - }); - try { - await job.started; - await connected.promise; - await job.dispose(); - await disconnected.promise; - } finally { - for (const socket of sockets) socket.destroy(); - await job.dispose(); - await pool.close(); - await new Promise((resolve) => server.close(() => resolve())); - } - }), + server.listen(0, "127.0.0.1"); + await new Promise((resolve, reject) => { + server.once("listening", resolve); + server.once("error", reject); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing test address"); + const descendant = `const s=require('node:net').connect(${address.port},'127.0.0.1',()=>s.write('ready')); s.once('close',()=>process.exit());`; + const fixture = new NodeWorkerThreads.Worker( + new URL("./processCleanup.test-fixture.ts", import.meta.url), + { + workerData: { mode, descendant }, + execArgv: [], + }, + ); + const cleaned = Promise.withResolvers<{ reused: boolean }>(); + const failed = Promise.withResolvers(); + const fail = (error: Error) => { + cleaned.reject(error); + failed.reject(error); + }; + const stopped = new Promise((resolve) => { + fixture.once("exit", (code) => { + fail(new Error(`Cleanup fixture exited unexpectedly (${code})`)); + resolve(code); + }); + }); + fixture.on("message", (event: { type: string; reused: boolean }) => { + if (event.type === "cleaned") cleaned.resolve(event); + }); + fixture.once("error", fail); + void cleaned.promise.catch(() => {}); + void failed.promise.catch(() => {}); + const message = () => new Promise((resolve) => fixture.once("message", resolve)); + try { + await Promise.race([message(), failed.promise]); + await Promise.race([connected.promise, failed.promise]); + const disposed = message(); + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node worker messages have no target origin. + fixture.postMessage("dispose"); + const result = await Promise.race([disposed, failed.promise]); + expect(result).toMatchObject({ + type: "disposed", + settled: true, + exited: mode === "normal", + }); + if (mode !== "normal") { + expect(result).toMatchObject({ + separateOwner: true, + live: true, + closed: mode === "child-exit", + }); + expect(sockets.size).toBe(1); + } + } finally { + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node worker messages have no target origin. + fixture.postMessage("release"); + try { + const result = await cleaned.promise; + if (mode === "taskkill") expect(result.reused).toBe(true); + expect(await stopped).toBe(0); + await disconnected.promise; + } finally { + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => server.close(() => resolve())); + await stopped; + } + } + }), ); }); diff --git a/apps/server/src/process/windowsProcessWorker.ts b/apps/server/src/process/windowsProcessWorker.ts index 6a9445c0c0427..c335d5d34ac20 100644 --- a/apps/server/src/process/windowsProcessWorker.ts +++ b/apps/server/src/process/windowsProcessWorker.ts @@ -173,22 +173,37 @@ export function makePool() { started: started.promise, exit: exit.promise, dispose: () => - (disposal ??= (async () => { - Atomics.store(state, 1, 1); - port1.postMessage("dispose"); - await released.promise; - port1.close(); - worker.off("error", onError); - worker.off("exit", onExit); - jobs.delete(job); - if (dead) return; - if (closed || idle) { - await stop(worker); - } else { - idle = worker; - worker.unref(); - } - })()), + (disposal ??= new Promise((resolve, reject) => { + // Bound the caller's wait; only release can relinquish child ownership. + // @effect-diagnostics-next-line globalTimers:off -- this Promise adapter bounds callers without cancelling worker-owned cleanup. + const timer = setTimeout(resolve, 1_000); + const cleanup = async () => { + Atomics.store(state, 1, 1); + port1.postMessage("dispose"); + await released.promise; + port1.close(); + worker.off("error", onError); + worker.off("exit", onExit); + jobs.delete(job); + if (dead) return; + if (closed || idle) { + await stop(worker); + } else { + idle = worker; + worker.unref(); + } + }; + void cleanup().then( + () => { + clearTimeout(timer); + resolve(); + }, + (error) => { + clearTimeout(timer); + reject(error); + }, + ); + })), }; jobs.add(job); worker.postMessage({ input, port: port2, control }, [port2]); diff --git a/apps/server/src/processRunner.test.ts b/apps/server/src/processRunner.test.ts index c6acbfd9f02af..81b6379d3a02f 100644 --- a/apps/server/src/processRunner.test.ts +++ b/apps/server/src/processRunner.test.ts @@ -78,12 +78,28 @@ const runWith = ); describe("runProcess", () => { - // quarantine: #25245 this regression exercises Windows child ownership during native creation. - it.live.skipIf(HostProcess.Platform.defaultValue() !== "win32")( - "cleans up a child cancelled during Windows native creation", - () => - Effect.promise(() => probeProcessCreation(false, true)).pipe( - Effect.map((result) => { + for (const failure of [ + undefined, + { at: "disposing", mode: "error" }, + { at: "disposing", mode: "exit" }, + { at: "child-exited", mode: "error" }, + { at: "child-exited", mode: "exit" }, + ] as const) { + // quarantine: #25245 this regression exercises Windows child ownership during native creation. + it.live.skipIf(HostProcess.Platform.defaultValue() !== "win32")( + `cleans up a child cancelled during Windows native creation (${JSON.stringify(failure)})`, + () => + Effect.promise(async () => { + const probe = probeProcessCreation(false, true, failure); + if (failure) { + await expect(probe).rejects.toThrow( + failure.mode === "error" + ? `Fixture worker failed at ${failure.at}` + : "Fixture worker exited unexpectedly (17)", + ); + return; + } + const result = await probe; expect(result).toEqual({ serverThread: false, gateClosed: true, @@ -91,8 +107,8 @@ describe("runProcess", () => { code: null, }); }), - ), - ); + ); + } // quarantine: #25245 this regression exercises the Windows process-creation boundary. it.live.skipIf(HostProcess.Platform.defaultValue() !== "win32")( "answers requests while Windows native process creation is blocked",