Open History to see the last commit
1import { fileURLToPath } from "node:url";
2import * as Cache from "effect/Cache";
3import type * as Cause from "effect/Cause";
4import * as Config from "effect/Config";
5import * as Context from "effect/Context";
6import * as Data from "effect/Data";
7import * as Deferred from "effect/Deferred";
8import * as Duration from "effect/Duration";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as HttpClient from "effect/http/HttpClient";
12import * as HttpServer from "effect/http/HttpServer";
13import { HttpServerRequest } from "effect/http/HttpServerRequest";
14import * as HttpServerResponse from "effect/http/HttpServerResponse";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import type { PlatformError } from "effect/PlatformError";
18import * as ChildProcess from "effect/process/ChildProcess";
19import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
20import * as Queue from "effect/Queue";
21import * as Schedule from "effect/Schedule";
22import * as Schema from "effect/Schema";
23import * as Scope from "effect/Scope";
24import * as Stream from "effect/Stream";
25import { LOG_FORMAT_ENV, logChildLine } from "../Util/LogRecord.ts";
26import { nodeLoaderArgs } from "../Util/Node.ts";
27import { httpServer } from "../Util/PlatformServices.ts";
28import { timedShutdownStep } from "../Util/Shutdown.ts";
29import { pipedColorEnv } from "../Util/Terminal.ts";
30import { traceParentEnv } from "../Util/TraceParent.ts";
31import { SPAWNER_URL_ENV_KEY } from "./RpcProviderProxy.ts";
32import { RPC_SERVER_ENVIRONMENT_KEY, type RpcServerEnvironment } from "./RpcServerEnvironment.ts";
33
34export class RpcSpawner extends Context.Service<
35 RpcSpawner,
36 {
37 readonly url: string;
38 }
39>()("alchemy/Local/RpcSpawner") {}
40
41class InvalidServerEntryUrl extends Data.TaggedError("InvalidServerEntryUrl")<{
42 cause: unknown;
43}> {}
44
45/**
46 * The spawner forks one child per distinct `serverEntryUrl`. Stack-specific
47 * context (stack name/stage, AlchemyContext) is NOT part of the spawn key —
48 * it travels per RPC session (see `SESSION_ENV_PARAM`), so many stacks
49 * (e.g. every test file in a run) share a single sidecar process.
50 */
51export const RpcSpawnPayload = Schema.Struct({ serverEntryUrl: Schema.String });
52export type RpcSpawnPayload = typeof RpcSpawnPayload.Type;
53
54/**
55 * One line of sidecar child output, tagged with the channel it arrived on.
56 * Streamed as NDJSON over the spawner's {@link LOGS_PATH} endpoint.
57 */
58export interface SidecarLogLine {
59 readonly channel: "stdout" | "stderr";
60 readonly line: string;
61}
62
63export const HeartbeatFrame = Schema.Struct({
64 channel: Schema.Literal("heartbeat"),
65});
66export type HeartbeatFrame = typeof HeartbeatFrame.Type;
67
68const SidecarLogLineSchema = Schema.Struct({
69 channel: Schema.Literals(["stdout", "stderr"]),
70 line: Schema.String,
71});
72
73/**
74 * Path on the spawner's HTTP server that streams sidecar output as NDJSON
75 * ({@link SidecarLogLine} per line). Consumed by {@link forwardSidecarLogs}
76 * from the exec child, which owns the terminal renderer.
77 */
78export const LOGS_PATH = "/logs";
79
80/** Options for the {@link RpcSpawner} that forks sidecar processes. */
81export type RpcSpawnerOptions = Pick<RpcServerEnvironment, "profile" | "envFile"> & {
82 /**
83 * Prefixes of environment variables withheld from sidecar processes (and,
84 * through them, from their own children). Sidecars otherwise inherit this
85 * process's entire environment.
86 */
87 readonly omitEnv?: ReadonlyArray<string>;
88};
89
90export const make = Effect.fn(function* ({ profile, envFile, omitEnv = [] }: RpcSpawnerOptions) {
91 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
92 const scope = yield* Effect.scope;
93 const cache = yield* Cache.make({
94 lookup: (serverEntryUrl: string) => spawn(serverEntryUrl).pipe(Scope.provide(scope)),
95 capacity: Infinity,
96 });
97
98 // Sidecar output hub. During `alchemy dev` this process (the outer dev
99 // command) shares the tty with the exec child, and the exec child owns the
100 // repainting progress renderer. Printing sidecar lines RAW from here
101 // interleaves with the renderer's repaints and corrupts the region
102 // (stacked/duplicated frames). So: when an exec child is subscribed via
103 // the /logs endpoint, hand lines to it and let its logger route them into
104 // the renderer; only log from this process as a fallback when no
105 // subscriber is connected (e.g. during a --watch restart gap, when no
106 // renderer is alive either).
107 const subscribers = new Set<(line: SidecarLogLine) => void>();
108 const publish = (line: SidecarLogLine): Effect.Effect<void> =>
109 Effect.suspend(() => {
110 if (subscribers.size > 0) {
111 for (const notify of subscribers) notify(line);
112 return Effect.void;
113 }
114 // Fallback path (no exec child subscribed): render the sidecar's
115 // records through this process's loggers.
116 return logChildLine(line.line, { stream: line.channel, process: "sidecar" });
117 });
118
119 const spawn = Effect.fn(function* (serverEntryUrl: string) {
120 const bin = typeof globalThis.Bun !== "undefined" ? "bun" : "node";
121 const main = fileURLToPath(serverEntryUrl);
122 // The sidecar runs on THIS process's runtime binary, not whatever PATH
123 // resolves: a `.ts` entry (checkout) needs the dev-mode loader hooks,
124 // which are gated on this node's version, and a published `.js` entry
125 // must not silently switch Node versions between supervisor and sidecar.
126 const program = bin === "bun" ? "bun" : process.execPath;
127 // Stack-specific context is deliberately absent: sessions carry their
128 // own SessionEnvironment, so one child serves every stack.
129 const environment: RpcServerEnvironment = {
130 profile,
131 envFile,
132 };
133 // Sidecar stdio is piped, so toolchains down the chain (vite, workerd,
134 // pretty loggers) detect a non-TTY and drop ANSI colors — but their
135 // output ultimately renders on THIS process's terminal. Force color
136 // through the pipe when that terminal supports it, unless the user
137 // already decided (NO_COLOR / FORCE_COLOR). `extendEnv` propagates it
138 // from the sidecar to its own children (dev servers, workerd).
139 const command = ChildProcess.make(
140 program,
141 {
142 bun: ["run", main],
143 // Under Node the sidecar starts with alchemy's Oxc loader hooks, the
144 // same way `dev.ts` starts the exec child: a `.ts` entry (src/ in a
145 // checkout) also gets src-condition resolution, a `.js` entry (lib/
146 // in a published install) gets the loader alone.
147 node: [...nodeLoaderArgs(main), main],
148 }[bin],
149 {
150 stdout: "pipe",
151 // Piped (NOT inherited) so the child's output routes through the
152 // Effect Console service: raw writes to the parent's fd corrupt the
153 // test runner's reporter/TUI. The drain below is mandatory — an
154 // unread pipe eventually fills and blocks the child.
155 stderr: "pipe",
156 // `detached` deliberately unset: the spawner defaults to a new process
157 // group on POSIX (and none on Windows, where a detached console child
158 // has its own quirks). A terminal Ctrl+C must NOT reach the sidecar
159 // directly — its lifecycle belongs to this process's finalizer (the
160 // bounded kill below), and a sidecar tearing itself down concurrently
161 // yanks local providers out from under the exec child mid-shutdown.
162 // As a group leader the kill also reaps the sidecar's own children
163 // (workerd, emulators) instead of orphaning them.
164 env: {
165 ...(omitEnv.length > 0 ? yield* inheritedEnv(omitEnv) : {}),
166 [RPC_SERVER_ENVIRONMENT_KEY]: JSON.stringify(environment),
167 // Its startup joins the trace of the call that spawned it; each
168 // provider call then carries its caller's span (RpcSerialization).
169 ...(yield* traceParentEnv),
170 // Log records, not rendered text: see `forwardSidecarLogs`.
171 [LOG_FORMAT_ENV]: "json",
172 // Tools further down (vite, workerd) still style their own output;
173 // records keep that text as written.
174 ...pipedColorEnv(),
175 },
176 // With variables to withhold, the full environment is built above.
177 extendEnv: omitEnv.length === 0,
178 },
179 );
180 const handle = yield* spawner.spawn(command);
181 const stderrReader = yield* handle.stderr.pipe(
182 Stream.decodeText,
183 Stream.splitLines,
184 Stream.runForEach((line) => publish({ channel: "stderr", line })),
185 Effect.ignore,
186 Effect.forkScoped,
187 );
188 const stdout = yield* readRpcOutput(handle.stdout, (line) =>
189 publish({ channel: "stdout", line }),
190 );
191 // This scope is the child handle's sole owner. Graceful shutdown runs this
192 // finalizer; abrupt parent loss closes the RPC parent connection and the
193 // child self-terminates (both paths are covered by RpcSpawnerCleanup).
194 // Command cleanup allows 1s for TERM and 1s for KILL; reserve 1s for sidecar teardown.
195 const kill = handle.kill({ forceKillAfter: "3 seconds" });
196 // Registered after the readers, so it runs before they are interrupted:
197 // the sidecar's own shutdown records drain until its pipes close.
198 yield* Effect.addFinalizer(() =>
199 kill.pipe(
200 Effect.ignore,
201 Effect.andThen(
202 Fiber.joinAll([stderrReader, stdout.reader]).pipe(
203 Effect.timeout("1 second"),
204 Effect.ignore,
205 ),
206 ),
207 timedShutdownStep(`stopped sidecar pid ${handle.pid}`),
208 ),
209 );
210 const url = yield* Deferred.await(stdout.address);
211 const ws = yield* Effect.acquireRelease(
212 Effect.sync(() => {
213 const ws = new WebSocket(new URL("/parent", url));
214 // Keep the parent link active so no idle timer on either end can
215 // close it; the sidecar ignores messages on this link.
216 const heartbeat = setInterval(() => {
217 if (ws.readyState === WebSocket.OPEN) ws.send("heartbeat");
218 }, 30_000);
219 heartbeat.unref?.();
220 ws.addEventListener("close", () => clearInterval(heartbeat));
221 return ws;
222 }),
223 (ws) => Effect.sync(() => ws.close()),
224 );
225 return {
226 url,
227 isRunning: Effect.zipWith(
228 handle.isRunning,
229 Effect.sync(
230 () => ws.readyState === WebSocket.CONNECTING || ws.readyState === WebSocket.OPEN,
231 ),
232 (a, b) => a && b,
233 { concurrent: true },
234 ),
235 kill,
236 };
237 });
238
239 const register = Effect.fn(function* (
240 serverEntryUrl: string,
241 attempt = 0,
242 ): Effect.fn.Return<string, PlatformError> {
243 const child = yield* Cache.get(cache, serverEntryUrl);
244 if (yield* child.isRunning) {
245 return child.url;
246 }
247 if (attempt > 3) {
248 return yield* Effect.die(
249 new Error(`Failed to spawn RPC server for "${serverEntryUrl}" after ${attempt} attempts.`),
250 );
251 }
252 yield* Effect.logWarning(
253 `RPC sidecar for ${serverEntryUrl} is no longer running; respawning (attempt ${attempt + 1})`,
254 );
255 // The sidecar is usually already gone, and killing an exited process fails.
256 yield* child.kill.pipe(Effect.ignore);
257 yield* Cache.invalidate(cache, serverEntryUrl);
258 return yield* register(serverEntryUrl, attempt + 1);
259 });
260
261 const server = yield* HttpServer.HttpServer;
262
263 const encoder = new TextEncoder();
264 // The first heartbeat flushes the response headers immediately; the
265 // periodic ones defeat idle timeouts (Bun kills sockets that stay silent
266 // for ~10s). Entries without a `line` are skipped by the client.
267 const heartbeat: HeartbeatFrame = { channel: "heartbeat" };
268 const HEARTBEAT = encoder.encode(`${JSON.stringify(heartbeat)}\n`);
269 const heartbeats = Stream.make(HEARTBEAT).pipe(
270 Stream.concat(
271 Schedule.spaced(Duration.seconds(5)).pipe(
272 Stream.fromSchedule,
273 Stream.map(() => HEARTBEAT),
274 ),
275 ),
276 );
277
278 yield* server.serve(
279 Effect.gen(function* () {
280 const request = yield* HttpServerRequest;
281 // `request.url` is relative under Node and absolute under Bun —
282 // normalize to the pathname before matching.
283 const pathname = new URL(request.url, "http://localhost").pathname;
284 if (pathname === LOGS_PATH) {
285 // Long-lived NDJSON stream of sidecar output. The subscriber (exec
286 // child) routes these lines through its logger, which the renderer
287 // inserts above the progress region instead of tearing it. Client
288 // disconnect interrupts the stream and unregisters the subscriber.
289 const queue = yield* Queue.make<Uint8Array, Cause.Done>();
290 const notify = (line: SidecarLogLine) => {
291 Queue.offerUnsafe(queue, encoder.encode(`${JSON.stringify(line)}\n`));
292 };
293 subscribers.add(notify);
294 return HttpServerResponse.stream(
295 Stream.merge(
296 Stream.fromQueue(queue).pipe(
297 Stream.ensuring(Effect.sync(() => subscribers.delete(notify))),
298 ),
299 heartbeats,
300 ),
301 { contentType: "application/x-ndjson" },
302 );
303 }
304 const decoded = yield* Effect.result(
305 request.json.pipe(Effect.flatMap(Schema.decodeUnknownEffect(RpcSpawnPayload))),
306 );
307 if (decoded._tag === "Failure") {
308 return HttpServerResponse.text("Invalid RPC spawn payload.", {
309 status: 400,
310 });
311 }
312 const entry = yield* Effect.result(
313 Effect.try({
314 try: () => new URL(decoded.success.serverEntryUrl),
315 catch: (cause) => new InvalidServerEntryUrl({ cause }),
316 }),
317 );
318 if (entry._tag === "Failure" || entry.success.protocol !== "file:") {
319 return HttpServerResponse.text("serverEntryUrl must be a valid file URL.", { status: 400 });
320 }
321 const url = yield* register(entry.success.href);
322 return HttpServerResponse.text(url);
323 }),
324 );
325
326 return RpcSpawner.of({
327 url: HttpServer.formatAddress(server.address),
328 });
329});
330
331/** This process's environment without the variables matching `omitEnv`. */
332const inheritedEnv = (omitEnv: ReadonlyArray<string>) =>
333 Effect.sync(() =>
334 Object.fromEntries(
335 Object.entries(process.env).filter(
336 ([key]) => !omitEnv.some((prefix) => key.startsWith(prefix)),
337 ),
338 ),
339 );
340
341export const layerServer = (environment: RpcSpawnerOptions) =>
342 Layer.effect(RpcSpawner, make(environment)).pipe(
343 // `/logs` is deliberately long-lived. On shutdown the exec child still
344 // owns that response while this scope owns the exec child, so waiting for
345 // Node's default 20-second graceful HTTP drain creates a finalizer cycle.
346 // Stop accepting/serving immediately so the watcher finalizer can close
347 // the child; all actual sidecars still use their own bounded finalizers.
348 Layer.provide(httpServer(0, "127.0.0.1", { gracefulShutdownTimeout: 0 })),
349 );
350
351const RPC_ADDRESS_REGEX = /(<ALCHEMY_RPC_ADDRESS>)(.+)(<\/ALCHEMY_RPC_ADDRESS>)/;
352
353/**
354 * Read the sidecar's stdout: its log lines and the RPC address line. The
355 * reader runs until the pipe closes, so a caller can drain it on shutdown.
356 */
357const readRpcOutput = (
358 stdout: Stream.Stream<Uint8Array, PlatformError>,
359 publish: (line: string) => Effect.Effect<void>,
360) =>
361 Effect.gen(function* () {
362 const address = yield* Deferred.make<string>();
363 // Every line but the address is published, including what the sidecar
364 // logs while starting: a layer that fails to build says why there.
365 const reader = yield* stdout.pipe(
366 Stream.decodeText,
367 Stream.splitLines,
368 Stream.runForEach((line) => {
369 const match = line.match(RPC_ADDRESS_REGEX);
370 return match ? Deferred.succeed(address, match[2]) : publish(line);
371 }),
372 Effect.ignore,
373 Effect.forkScoped,
374 );
375 return { address, reader };
376 });
377
378export const parseSidecarLogLine = (raw: string): SidecarLogLine | undefined => {
379 try {
380 return Schema.decodeUnknownOption(SidecarLogLineSchema)(JSON.parse(raw)).pipe(
381 Option.getOrUndefined,
382 );
383 } catch {
384 return undefined;
385 }
386};
387
388/**
389 * Pull the sidecars' log records from the spawner (the outer `alchemy dev`
390 * process) into THIS process's loggers. The exec child owns the terminal
391 * renderer, so the records render above the repainting progress region
392 * instead of racing it on the shared tty; it forwards them on to the
393 * supervisor, which writes the run log. Forks in the ambient scope and never fails: when no spawner is
394 * configured (`ALCHEMY_RPC_SPAWNER_URL` absent — plain deploy/destroy) it is
395 * a no-op, and if the connection drops the spawner's own fallback printing
396 * takes over.
397 */
398export const forwardSidecarLogs: Effect.Effect<void, never, HttpClient.HttpClient | Scope.Scope> =
399 Config.String(SPAWNER_URL_ENV_KEY).pipe(
400 Effect.flatMap((spawnerUrl) => {
401 const streamOnce = Effect.gen(function* () {
402 const client = yield* HttpClient.HttpClient;
403 const response = yield* client.get(new URL(LOGS_PATH, spawnerUrl).toString());
404 yield* response.stream.pipe(
405 Stream.decodeText,
406 Stream.splitLines,
407 Stream.runForEach((raw) =>
408 Effect.suspend(() => {
409 const parsed = parseSidecarLogLine(raw);
410 if (parsed === undefined) return Effect.void;
411 // Sidecar lines are log records: replay them through this
412 // process's loggers, which render them (or pass them on as JSON).
413 return logChildLine(parsed.line, { stream: parsed.channel, process: "sidecar" });
414 }),
415 ),
416 );
417 });
418 // Keep the subscription alive for the whole dev session: reconnect
419 // (paced) if the stream ends or errors. While disconnected the spawner's
420 // fallback printing covers the gap; the loop dies with the ambient scope.
421 return streamOnce.pipe(
422 Effect.ignore,
423 Effect.andThen(Effect.sleep("1 second")),
424 Effect.forever,
425 );
426 }),
427 Effect.ignore,
428 Effect.forkScoped,
429 Effect.asVoid,
430 );