diff --git a/apps/server/src/process/SpawnWorker.test.ts b/apps/server/src/process/SpawnWorker.test.ts new file mode 100644 index 0000000000000..210f550c7c4ad --- /dev/null +++ b/apps/server/src/process/SpawnWorker.test.ts @@ -0,0 +1,157 @@ +// @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"; +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 HostProcess from "@t3tools/shared/HostProcess"; + +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 the server's Windows children on a launcher thread, not the server thread", () => + Effect.gen(function* () { + 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", () => + 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); + } + // 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" } }, + }), + ); + const [fd3, extraCode] = yield* Effect.all([text(extra.getOutputFd(3)), extra.exitCode], { + concurrency: "unbounded", + }); + 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()))"), + ), + ); + 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, detached or not, 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..81f72ff00739d --- /dev/null +++ b/apps/server/src/process/SpawnWorker.ts @@ -0,0 +1,507 @@ +// @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"; +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. +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: "destroy"; readonly id: number; readonly fd: number } + | { 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; + } + 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 }); + else stream.end(() => post({ type: "finished", id, fd: request.fd })); + return; + case "flow": + if (request.paused) stream?.pause(); + else stream?.resume(); + return; + case "destroy": + stream?.destroy(); + return; + case "kill": + try { + child.kill(request.signal); + } catch {} + 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, + ) => { + 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); + 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 }); + }; + // 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 { + 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: 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) { + if (stream && "push" in stream && !stream.readableEnded) stream.push(null); + else if (stream && !("push" in stream)) 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], + // 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; + } + + 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": + child.emit("close", child.exitCode, child.signalCode); + 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, so try to end the whole tree before reporting the exit. + const finish = () => { + child.exit(null, "SIGKILL"); + child.abandon(); + }; + if (child.pid === undefined) finish(); + else this.pool.killTree(child.pid, finish); + } + 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(); + readonly platform: NodeJS.Platform; + #nextId = 0; + + constructor(platform: NodeJS.Platform) { + this.platform = platform; + } + + 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); + }; + + killTree(pid: number, done: () => void): void { + const killPid = () => { + try { + process.kill(pid, "SIGKILL"); + } catch {} + }; + if (this.platform !== "win32") { + killPid(); + done(); + return; + } + let settled = false; + const finish = (killed: boolean) => { + if (settled) return; + settled = true; + // As Effect's own kill does when taskkill fails, end at least 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 { + 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.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); + }), +); + +/** 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/server.ts b/apps/server/src/server.ts index 5c9ab9c433ffc..ae485c8741a37 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -15,6 +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 SpawnWorker from "./process/SpawnWorker.ts"; import { EnvironmentHttpApi, type RepositoryIdentity } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Duration from "effect/Duration"; @@ -338,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; @@ -374,12 +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)), + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer, layerSpawner)), ); const layerPersistence = Layer.empty.pipe(Layer.provideMerge(SqlitePersistence.layerConfig)); @@ -1207,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); 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..6e957eac98f41 --- /dev/null +++ b/patches/@effect__platform-node-shared@4.1.2.patch @@ -0,0 +1,93 @@ +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..fbecacc8b39a9eab7725a799aa5019c36bcc32ab 100644 +--- a/dist/NodeChildProcessSpawner.js ++++ b/dist/NodeChildProcessSpawner.js +@@ -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 +345,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 +675,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 f56b9fb3f0a7b..72734e68b8b0b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -356,6 +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': 06952843c6f431ca83c78762fd18d23cda76883801049ee311feced2438ecf14 '@effect/vitest@4.1.2': 299510b71c36537a832a67e9d160a5f112a5f326a29e95c41cb45fc8234e47c7 '@expo/fingerprint@0.21.5': 8fba80d6bbce46c72de427fe8c7907462d262a0bfe50c6b2b7ee9c4e0c61e5e1 '@expo/metro-config@58.0.14': 90fa980e8b6480631fb2fbb04559834e3c612db40c550cd828f11be297d8d760 @@ -837,7 +838,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=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 +13679,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=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(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 +13696,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=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 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index d64fc1a679f0c..f9e8d89a09eeb 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -262,6 +262,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 # A newline in a copied table cell becomes
, so Copy as Markdown keeps the row whole. react-native-enriched-markdown@1.2.0-t3.16: patches/react-native-enriched-markdown@1.2.0-t3.16.patch # A failed Artifacts.cached effect hangs every later caller: https://github.com/alchemy-run/alchemy/issues/2981