diff --git a/.changeset/stable-dev-worker-transport.md b/.changeset/stable-dev-worker-transport.md new file mode 100644 index 000000000..143a49c3f --- /dev/null +++ b/.changeset/stable-dev-worker-transport.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Keep the `eve dev` listener stable while ready-gated workers are replaced. In-flight HTTP streams, request bodies, keep-alive connections, WebSockets, client addresses, and dev control requests now retain their admitted worker or cancel cleanly during reload and shutdown. diff --git a/packages/eve/package.json b/packages/eve/package.json index 952e17b31..822d983f1 100644 --- a/packages/eve/package.json +++ b/packages/eve/package.json @@ -337,6 +337,7 @@ "chokidar": "5.0.0", "commander": "14.0.3", "emulate": "0.6.0", + "env-runner": "0.1.12", "eventsource-parser": "3.0.8", "experimental-ai-sdk-code-mode": "1.0.16", "gray-matter": "4.0.3", diff --git a/packages/eve/scripts/vendor-compiled/declarations/env-runner.d.ts b/packages/eve/scripts/vendor-compiled/declarations/env-runner.d.ts new file mode 100644 index 000000000..69e8964c2 --- /dev/null +++ b/packages/eve/scripts/vendor-compiled/declarations/env-runner.d.ts @@ -0,0 +1,38 @@ +import type { IncomingMessage } from "node:http"; +import type { Socket } from "node:net"; + +export interface WorkerHooks { + onClose?(runner: BaseEnvRunner, cause?: unknown): void; + onReady?(runner: BaseEnvRunner): void; +} + +export interface BaseEnvRunnerOptions { + readonly data?: Record; + readonly hooks?: WorkerHooks; + readonly name: string; + readonly workerEntry: string; +} + +export class BaseEnvRunner { + readonly closed: boolean; + readonly ready: boolean; + protected readonly _data: Record | undefined; + protected readonly _name: string; + protected readonly _workerEntry: string; + + constructor(options: BaseEnvRunnerOptions); + + close(cause?: unknown): Promise; + fetch(input: string | URL | Request, init?: RequestInit): Promise; + sendMessage(message: unknown): void; + upgrade(context: { + node: { head: Uint8Array; req: IncomingMessage; socket: Socket }; + }): Promise; + waitForReady(timeout?: number): Promise; + + protected _closeRuntime(): Promise; + protected _handleMessage(message: unknown): void; + protected _hasRuntime(): boolean; + protected _initWithVirtualData(initialize: () => void): void; + protected _runtimeType(): string; +} diff --git a/packages/eve/scripts/vendor-compiled/env-runner.mjs b/packages/eve/scripts/vendor-compiled/env-runner.mjs new file mode 100644 index 000000000..90a95d69a --- /dev/null +++ b/packages/eve/scripts/vendor-compiled/env-runner.mjs @@ -0,0 +1,18 @@ +import { loadDeclaration } from "./_shared.mjs"; + +export default { + packageName: "env-runner", + compiledPath: "env-runner", + entries: [ + { + entry: "dist/_chunks/common-base-runner.mjs", + outputPath: "index", + declaration: await loadDeclaration("env-runner.d.ts"), + }, + { + entry: "dist/runners/node-worker/worker.mjs", + outputPath: "node-worker", + declaration: "export {};\n", + }, + ], +}; diff --git a/packages/eve/scripts/vendor-compiled/index.mjs b/packages/eve/scripts/vendor-compiled/index.mjs index 747c26258..027905d89 100644 --- a/packages/eve/scripts/vendor-compiled/index.mjs +++ b/packages/eve/scripts/vendor-compiled/index.mjs @@ -32,6 +32,7 @@ import chokidar from "./chokidar.mjs"; import commander from "./commander.mjs"; import experimentalAiSdkCodeMode from "./experimental-ai-sdk-code-mode.mjs"; import eventsourceParserStream from "./eventsource-parser-stream.mjs"; +import envRunner from "./env-runner.mjs"; import grayMatter from "./gray-matter.mjs"; import jose from "./jose.mjs"; import jsoncParser from "./jsonc-parser.mjs"; @@ -52,6 +53,7 @@ export const MODULES = [ commander, experimentalAiSdkCodeMode, eventsourceParserStream, + envRunner, google, grayMatter, jose, diff --git a/packages/eve/src/internal/nitro/dev-runtime-artifacts.ts b/packages/eve/src/internal/nitro/dev-runtime-artifacts.ts index decaa00f6..4f0832070 100644 --- a/packages/eve/src/internal/nitro/dev-runtime-artifacts.ts +++ b/packages/eve/src/internal/nitro/dev-runtime-artifacts.ts @@ -40,6 +40,11 @@ export interface DevelopmentRuntimeArtifactsSnapshot { readonly sourceRoot: string; } +export interface ActiveDevelopmentRuntimeArtifactsSnapshot { + readonly runtimeAppRoot: string; + readonly snapshotRoot: string; +} + /** * Resolves the dev-server pointer that records the latest runtime artifact * snapshot for new sessions. @@ -138,6 +143,21 @@ export function readDevelopmentRuntimeArtifactsSnapshotRoot( return pointer.runtimeAppRoot; } +export function readActiveDevelopmentRuntimeArtifactsSnapshot( + appRoot: string, +): ActiveDevelopmentRuntimeArtifactsSnapshot | undefined { + const pointer = readDevelopmentRuntimeArtifactsPointer( + resolveDevelopmentRuntimeArtifactsPointerPath(appRoot), + ); + if (pointer === undefined || pointer.version === 1) { + return undefined; + } + return { + runtimeAppRoot: pointer.runtimeAppRoot, + snapshotRoot: pointer.snapshotRoot, + }; +} + /** * Reads a revision token for the latest dev runtime artifact snapshot. */ diff --git a/packages/eve/src/internal/nitro/host/configure-nitro-routes.test.ts b/packages/eve/src/internal/nitro/host/configure-nitro-routes.test.ts index 6a3d5db1a..e816b5d1b 100644 --- a/packages/eve/src/internal/nitro/host/configure-nitro-routes.test.ts +++ b/packages/eve/src/internal/nitro/host/configure-nitro-routes.test.ts @@ -310,28 +310,21 @@ describe("Nitro route configuration", () => { expect(source).toContain("authored-module-map-loader.js"); }); - it("registers the dev runtime artifact revision route only in dev mode", async () => { + it("does not register parent-owned development control routes in Nitro", async () => { const devNitro = createNitroStub({ dev: true }); const prodNitro = createNitroStub({ dev: false }); await configureDevelopmentNitroRoutes(devNitro, createPreparedHost()); await configureProductionNitroRoutes(prodNitro, createPreparedHost(), "app"); - expect(devNitro.options.handlers).toContainEqual({ - handler: "#eve-route/eve/v1/dev/runtime-artifacts", - method: "GET", - route: "/eve/v1/dev/runtime-artifacts", - }); - expect(devNitro.options.handlers).not.toContainEqual( - expect.objectContaining({ - route: "/eve/v1/dev/runtime-artifacts/rebuild", - }), - ); - expect(prodNitro.options.handlers).not.toContainEqual( - expect.objectContaining({ - route: "/eve/v1/dev/runtime-artifacts", - }), - ); + for (const nitro of [devNitro, prodNitro]) { + expect(nitro.options.handlers).not.toContainEqual( + expect.objectContaining({ route: "/eve/v1/dev/runtime-artifacts" }), + ); + expect(nitro.options.handlers).not.toContainEqual( + expect.objectContaining({ route: "/eve/v1/dev/runtime-artifacts/rebuild" }), + ); + } }); it("registers the agent info route for dev and production app builds", async () => { diff --git a/packages/eve/src/internal/nitro/host/configure-nitro-routes.ts b/packages/eve/src/internal/nitro/host/configure-nitro-routes.ts index 089ca80c6..c36b6a733 100644 --- a/packages/eve/src/internal/nitro/host/configure-nitro-routes.ts +++ b/packages/eve/src/internal/nitro/host/configure-nitro-routes.ts @@ -4,7 +4,6 @@ import { dirname, join, relative } from "node:path"; import type { Nitro } from "nitro/types"; import { EVE_DEV_DISPATCH_SCHEDULE_ROUTE_PATTERN, - EVE_DEV_RUNTIME_ARTIFACTS_ROUTE_PATH, EVE_HEALTH_ROUTE_PATH, } from "#protocol/routes.js"; import { @@ -316,13 +315,6 @@ function registerDevelopmentControlRoutes( nitro: Nitro, artifactsConfig: DevelopmentNitroArtifactsConfig, ): void { - addFrameworkVirtualHandler(nitro, { - args: JSON.stringify({ appRoot: artifactsConfig.appRoot }), - handlerExport: "handleDevRuntimeArtifactsRequest", - method: "GET", - modulePath: resolvePackageSourceFilePath("src/internal/nitro/routes/dev-runtime-artifacts.ts"), - route: EVE_DEV_RUNTIME_ARTIFACTS_ROUTE_PATH, - }); addFrameworkVirtualHandler(nitro, { // The complete config is resolved here, in the unbundled host process, // and baked into the handler: resolving the module-map loader path from diff --git a/packages/eve/src/internal/nitro/host/create-application-nitro.ts b/packages/eve/src/internal/nitro/host/create-application-nitro.ts index 423653854..4b8aede28 100644 --- a/packages/eve/src/internal/nitro/host/create-application-nitro.ts +++ b/packages/eve/src/internal/nitro/host/create-application-nitro.ts @@ -753,6 +753,10 @@ export async function createDevelopmentApplicationNitro( ): Promise { const nitroBuildDir = resolveNitroBuildDirectory(preparedHost.appRoot); const bundler = createApplicationNitroBundlerConfiguration(preparedHost, undefined); + const plugins = createApplicationNitroPlugins(preparedHost); + plugins.push( + resolvePackageSourceFilePath("src/internal/nitro/host/dev-worker-metadata-plugin.ts"), + ); await prepareEveVersionedCacheDirectory(nitroBuildDir); const nitro = await createNitro( @@ -762,7 +766,7 @@ export async function createDevelopmentApplicationNitro( dev: true, features: { websocket: true }, logLevel: 1, - plugins: createApplicationNitroPlugins(preparedHost), + plugins, publicAssets: [], scanDirs: [resolvePackageSourceDirectoryPath("src/execution")], rolldownConfig: bundler.nitroRolldownConfig, diff --git a/packages/eve/src/internal/nitro/host/dev-worker-http.ts b/packages/eve/src/internal/nitro/host/dev-worker-http.ts new file mode 100644 index 000000000..e79ef9c9b --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-worker-http.ts @@ -0,0 +1,169 @@ +import type { IncomingMessage, Server, ServerResponse } from "node:http"; +import type { Socket } from "node:net"; +import { Readable } from "node:stream"; + +import { + DEVELOPMENT_WORKER_METADATA_HEADER, + encodeDevelopmentWorkerMetadata, +} from "#internal/nitro/host/dev-worker-metadata.js"; +import type { DevelopmentWorkerGeneration } from "#internal/nitro/host/dev-worker-server.js"; +import { toErrorMessage } from "#shared/errors.js"; + +export function createPublicRequest(request: IncomingMessage, signal: AbortSignal): Request { + const authority = request.headers.host ?? "localhost"; + const url = new URL(request.url ?? "/", `http://${authority}`); + const headers = new Headers(); + for (const [name, value] of Object.entries(request.headers)) { + if (Array.isArray(value)) { + for (const item of value) { + headers.append(name, item); + } + } else if (value !== undefined) { + headers.set(name, value); + } + } + headers.delete(DEVELOPMENT_WORKER_METADATA_HEADER); + + const hasBody = request.method !== "GET" && request.method !== "HEAD"; + return new Request(url, { + body: hasBody ? (Readable.toWeb(request) as ReadableStream) : undefined, + duplex: hasBody ? "half" : undefined, + headers, + method: request.method, + signal, + } as RequestInit & { duplex?: "half" }); +} + +export function createWorkerRequest(input: { + readonly generation: DevelopmentWorkerGeneration; + readonly request: Request; + readonly secret: string; + readonly socket: Socket; +}): Request { + const headers = new Headers(input.request.headers); + headers.set( + DEVELOPMENT_WORKER_METADATA_HEADER, + encodeDevelopmentWorkerMetadata({ + metadata: { + clientAddress: normalizeClientAddress(input.socket.remoteAddress), + generationId: input.generation.id, + runtimeAppRoot: input.generation.runtimeAppRoot, + }, + secret: input.secret, + }), + ); + return new Request(input.request, { headers, signal: input.request.signal }); +} + +export function stampWorkerUpgradeRequest(input: { + readonly generation: DevelopmentWorkerGeneration; + readonly request: IncomingMessage; + readonly secret: string; +}): void { + input.request.headers[DEVELOPMENT_WORKER_METADATA_HEADER] = encodeDevelopmentWorkerMetadata({ + metadata: { + clientAddress: normalizeClientAddress(input.request.socket.remoteAddress), + generationId: input.generation.id, + runtimeAppRoot: input.generation.runtimeAppRoot, + }, + secret: input.secret, + }); +} + +export async function writeResponse( + response: ServerResponse, + webResponse: Response, + signal: AbortSignal, +): Promise { + if (response.destroyed) { + await webResponse.body?.cancel(); + return; + } + + response.statusCode = webResponse.status; + response.statusMessage = webResponse.statusText; + const getSetCookie = ( + webResponse.headers as Headers & { + getSetCookie?: () => readonly string[]; + } + ).getSetCookie; + const setCookies = getSetCookie?.call(webResponse.headers) ?? []; + webResponse.headers.forEach((value, name) => { + if (name.toLowerCase() !== "set-cookie") { + response.setHeader(name, value); + } + }); + if (setCookies.length > 0) { + response.setHeader("set-cookie", [...setCookies]); + } + + if (webResponse.body === null) { + await endResponse(response); + return; + } + + const body = Readable.fromWeb(webResponse.body as import("node:stream/web").ReadableStream); + const cancelBody = () => body.destroy(signal.reason as Error | undefined); + signal.addEventListener("abort", cancelBody, { once: true }); + try { + await new Promise((resolve, reject) => { + const onClose = () => { + if (!response.writableEnded) { + reject(new Error("Development client disconnected.")); + } + }; + body.once("error", reject); + response.once("error", reject); + response.once("close", onClose); + response.once("finish", resolve); + body.pipe(response); + }); + } finally { + signal.removeEventListener("abort", cancelBody); + if (!body.destroyed && (signal.aborted || response.destroyed)) { + body.destroy(); + } + } +} + +export function writeRequestError(response: ServerResponse, error: unknown): void { + if (!response.headersSent) { + response.statusCode = 503; + response.setHeader("content-type", "application/json; charset=utf-8"); + } + if (!response.writableEnded && !response.destroyed) { + response.end(JSON.stringify({ error: toErrorMessage(error) })); + } +} + +export function closeServer(server: Server): Promise { + if (!server.listening) { + return Promise.resolve(); + } + + const closed = new Promise((resolve, reject) => { + server.close((error) => { + if (error === undefined) { + resolve(); + } else { + reject(error); + } + }); + }); + server.closeIdleConnections?.(); + return closed; +} + +function normalizeClientAddress(address: string | undefined): string | null { + if (address === undefined || address.length === 0) { + return null; + } + return address.startsWith("::ffff:") ? address.slice("::ffff:".length) : address; +} + +async function endResponse(response: ServerResponse): Promise { + await new Promise((resolve, reject) => { + response.once("error", reject); + response.end(resolve); + }); +} diff --git a/packages/eve/src/internal/nitro/host/dev-worker-metadata-plugin.integration.test.ts b/packages/eve/src/internal/nitro/host/dev-worker-metadata-plugin.integration.test.ts new file mode 100644 index 000000000..126bebdd2 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-worker-metadata-plugin.integration.test.ts @@ -0,0 +1,106 @@ +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { useTemporaryDirectories } from "#internal/testing/use-temporary-app-roots.js"; +import { + DEVELOPMENT_WORKER_APP_ROOT_ENV, + DEVELOPMENT_WORKER_METADATA_HEADER, + DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV, + encodeDevelopmentWorkerMetadata, + readDevelopmentWorkerRequestMetadata, +} from "#internal/nitro/host/dev-worker-metadata.js"; +import installDevelopmentWorkerMetadataPlugin from "#internal/nitro/host/dev-worker-metadata-plugin.js"; + +const createScratchDirectory = useTemporaryDirectories(); + +describe("development worker metadata plugin", () => { + afterEach(() => { + delete process.env[DEVELOPMENT_WORKER_APP_ROOT_ENV]; + delete process.env[DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV]; + }); + + it("removes the transport header and installs only validated parent metadata", async () => { + const appRoot = await createScratchDirectory("eve-dev-worker-metadata-"); + const generationId = "generation-a"; + const runtimeAppRoot = join( + appRoot, + ".eve", + "dev-runtime", + "snapshots", + generationId, + "source", + "app", + ); + const secret = "parent-secret"; + await mkdir(join(runtimeAppRoot, ".eve", "compile"), { recursive: true }); + await writeFile( + join(runtimeAppRoot, ".eve", "compile", "compiled-agent-manifest.json"), + "{}\n", + ); + process.env[DEVELOPMENT_WORKER_APP_ROOT_ENV] = appRoot; + process.env[DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV] = secret; + + let requestHook: ((event: { readonly req: Request }) => void) | undefined; + installDevelopmentWorkerMetadataPlugin({ + hooks: { + hook(_name, handler) { + requestHook = handler; + }, + }, + }); + const request = new Request("http://worker.test/", { + headers: { + [DEVELOPMENT_WORKER_METADATA_HEADER]: encodeDevelopmentWorkerMetadata({ + metadata: { + clientAddress: "192.0.2.25", + generationId, + runtimeAppRoot, + }, + secret, + }), + }, + }); + + requestHook?.({ req: request }); + + expect(request.headers.has(DEVELOPMENT_WORKER_METADATA_HEADER)).toBe(false); + expect(readDevelopmentWorkerRequestMetadata(request)).toEqual({ + clientAddress: "192.0.2.25", + generationId, + runtimeAppRoot, + }); + }); + + it("rejects a signed runtime root outside the named generation", async () => { + const appRoot = await createScratchDirectory("eve-dev-worker-metadata-outside-"); + const outsideRuntimeRoot = await createScratchDirectory("eve-dev-worker-metadata-target-"); + const secret = "parent-secret"; + process.env[DEVELOPMENT_WORKER_APP_ROOT_ENV] = appRoot; + process.env[DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV] = secret; + + let requestHook: ((event: { readonly req: Request }) => void) | undefined; + installDevelopmentWorkerMetadataPlugin({ + hooks: { + hook(_name, handler) { + requestHook = handler; + }, + }, + }); + const request = new Request("http://worker.test/", { + headers: { + [DEVELOPMENT_WORKER_METADATA_HEADER]: encodeDevelopmentWorkerMetadata({ + metadata: { + clientAddress: "192.0.2.25", + generationId: "generation-a", + runtimeAppRoot: outsideRuntimeRoot, + }, + secret, + }), + }, + }); + + expect(() => requestHook?.({ req: request })).toThrow("outside its generation"); + }); +}); diff --git a/packages/eve/src/internal/nitro/host/dev-worker-metadata-plugin.ts b/packages/eve/src/internal/nitro/host/dev-worker-metadata-plugin.ts new file mode 100644 index 000000000..05f841f84 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-worker-metadata-plugin.ts @@ -0,0 +1,81 @@ +import { existsSync, realpathSync } from "node:fs"; +import { basename, join, relative, resolve, sep } from "node:path"; + +import { + decodeDevelopmentWorkerMetadata, + DEVELOPMENT_WORKER_APP_ROOT_ENV, + DEVELOPMENT_WORKER_METADATA_HEADER, + DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV, + installDevelopmentWorkerRequestMetadata, +} from "#internal/nitro/host/dev-worker-metadata.js"; + +interface NitroRequestHookApp { + readonly hooks: { + hook(name: "request", handler: (event: { readonly req: Request }) => void): unknown; + }; +} + +export default function installDevelopmentWorkerMetadataPlugin( + nitroApp: NitroRequestHookApp, +): void { + const appRoot = readRequiredEnvironment(DEVELOPMENT_WORKER_APP_ROOT_ENV); + const secret = readRequiredEnvironment(DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV); + + nitroApp.hooks.hook("request", (event) => { + const metadata = decodeDevelopmentWorkerMetadata({ + header: event.req.headers.get(DEVELOPMENT_WORKER_METADATA_HEADER), + secret, + }); + validateDevelopmentWorkerRuntimeRoot({ + appRoot, + generationId: metadata.generationId, + runtimeAppRoot: metadata.runtimeAppRoot, + }); + event.req.headers.delete(DEVELOPMENT_WORKER_METADATA_HEADER); + installDevelopmentWorkerRequestMetadata(event.req, metadata); + }); +} + +function validateDevelopmentWorkerRuntimeRoot(input: { + readonly appRoot: string; + readonly generationId: string; + readonly runtimeAppRoot: string; +}): void { + if (basename(input.generationId) !== input.generationId) { + throw new Error("Development worker generation id is invalid."); + } + + const snapshotRoot = canonicalize( + join(input.appRoot, ".eve", "dev-runtime", "snapshots", input.generationId), + ); + const runtimeAppRoot = canonicalize(input.runtimeAppRoot); + const relativePath = relative(snapshotRoot, runtimeAppRoot); + if ( + relativePath.length === 0 || + relativePath === ".." || + relativePath.startsWith(`..${sep}`) || + relativePath.startsWith(sep) + ) { + throw new Error("Development worker runtime root is outside its generation."); + } + + if (!existsSync(join(runtimeAppRoot, ".eve", "compile", "compiled-agent-manifest.json"))) { + throw new Error(`Development worker generation "${input.generationId}" is incomplete.`); + } +} + +function canonicalize(path: string): string { + try { + return realpathSync(path); + } catch { + return resolve(path); + } +} + +function readRequiredEnvironment(name: string): string { + const value = process.env[name]; + if (value === undefined || value.length === 0) { + throw new Error(`Development worker environment is missing ${name}.`); + } + return value; +} diff --git a/packages/eve/src/internal/nitro/host/dev-worker-metadata.test.ts b/packages/eve/src/internal/nitro/host/dev-worker-metadata.test.ts new file mode 100644 index 000000000..e6478d5ba --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-worker-metadata.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "vitest"; + +import { + decodeDevelopmentWorkerMetadata, + encodeDevelopmentWorkerMetadata, +} from "#internal/nitro/host/dev-worker-metadata.js"; + +describe("development worker metadata", () => { + it("accepts metadata signed by the parent transport", () => { + const metadata = { + clientAddress: "192.0.2.10", + generationId: "generation-a", + runtimeAppRoot: "/app/.eve/dev-runtime/snapshots/generation-a/source/app", + }; + const header = encodeDevelopmentWorkerMetadata({ metadata, secret: "parent-secret" }); + + expect(decodeDevelopmentWorkerMetadata({ header, secret: "parent-secret" })).toEqual(metadata); + }); + + it("rejects metadata that a public caller cannot sign", () => { + const header = encodeDevelopmentWorkerMetadata({ + metadata: { + clientAddress: "203.0.113.50", + generationId: "generation-a", + runtimeAppRoot: "/app/.eve/dev-runtime/snapshots/generation-a/source/app", + }, + secret: "public-secret", + }); + + expect(() => decodeDevelopmentWorkerMetadata({ header, secret: "parent-secret" })).toThrow( + "not trusted", + ); + }); +}); diff --git a/packages/eve/src/internal/nitro/host/dev-worker-metadata.ts b/packages/eve/src/internal/nitro/host/dev-worker-metadata.ts new file mode 100644 index 000000000..c5d911f26 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-worker-metadata.ts @@ -0,0 +1,100 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; + +export const DEVELOPMENT_WORKER_METADATA_HEADER = "x-eve-dev-worker-metadata"; +export const DEVELOPMENT_WORKER_APP_ROOT_ENV = "EVE_DEV_WORKER_APP_ROOT"; +export const DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV = "EVE_DEV_WORKER_TRANSPORT_SECRET"; + +const DEVELOPMENT_WORKER_REQUEST_METADATA = Symbol.for("eve.dev.worker-request-metadata"); + +export interface DevelopmentWorkerMetadata { + readonly clientAddress: string | null; + readonly generationId: string; + readonly runtimeAppRoot: string; +} + +interface RequestWithDevelopmentWorkerMetadata extends Request { + [DEVELOPMENT_WORKER_REQUEST_METADATA]?: DevelopmentWorkerMetadata; +} + +export function encodeDevelopmentWorkerMetadata(input: { + readonly metadata: DevelopmentWorkerMetadata; + readonly secret: string; +}): string { + const payload = Buffer.from(JSON.stringify(input.metadata), "utf8").toString("base64url"); + const signature = signDevelopmentWorkerMetadata(payload, input.secret); + return `${payload}.${signature}`; +} + +export function decodeDevelopmentWorkerMetadata(input: { + readonly header: string | null; + readonly secret: string; +}): DevelopmentWorkerMetadata { + if (input.header === null) { + throw new Error("Development worker request is missing trusted metadata."); + } + + const separatorIndex = input.header.indexOf("."); + if (separatorIndex <= 0 || separatorIndex === input.header.length - 1) { + throw new Error("Development worker request metadata is malformed."); + } + + const payload = input.header.slice(0, separatorIndex); + const signature = input.header.slice(separatorIndex + 1); + const expectedSignature = signDevelopmentWorkerMetadata(payload, input.secret); + const received = Buffer.from(signature, "base64url"); + const expected = Buffer.from(expectedSignature, "base64url"); + + if (received.length !== expected.length || !timingSafeEqual(received, expected)) { + throw new Error("Development worker request metadata is not trusted."); + } + + let value: unknown; + try { + value = JSON.parse(Buffer.from(payload, "base64url").toString("utf8")); + } catch { + throw new Error("Development worker request metadata is malformed."); + } + + if (!isDevelopmentWorkerMetadata(value)) { + throw new Error("Development worker request metadata is malformed."); + } + + return value; +} + +export function installDevelopmentWorkerRequestMetadata( + request: Request, + metadata: DevelopmentWorkerMetadata, +): void { + Object.defineProperty(request, DEVELOPMENT_WORKER_REQUEST_METADATA, { + configurable: false, + enumerable: false, + value: metadata, + writable: false, + }); +} + +export function readDevelopmentWorkerRequestMetadata( + request: Request, +): DevelopmentWorkerMetadata | undefined { + return (request as RequestWithDevelopmentWorkerMetadata)[DEVELOPMENT_WORKER_REQUEST_METADATA]; +} + +function signDevelopmentWorkerMetadata(payload: string, secret: string): string { + return createHmac("sha256", secret).update(payload).digest("base64url"); +} + +function isDevelopmentWorkerMetadata(value: unknown): value is DevelopmentWorkerMetadata { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + return false; + } + + const record = value as Record; + return ( + (record.clientAddress === null || typeof record.clientAddress === "string") && + typeof record.generationId === "string" && + record.generationId.length > 0 && + typeof record.runtimeAppRoot === "string" && + record.runtimeAppRoot.length > 0 + ); +} diff --git a/packages/eve/src/internal/nitro/host/dev-worker-runner.ts b/packages/eve/src/internal/nitro/host/dev-worker-runner.ts new file mode 100644 index 000000000..0ea9faba2 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-worker-runner.ts @@ -0,0 +1,126 @@ +import { existsSync } from "node:fs"; +import { Worker } from "node:worker_threads"; + +import { BaseEnvRunner } from "#compiled/env-runner/index.js"; +import { resolvePackageCompiledFilePath } from "#internal/application/package.js"; +import { + DEVELOPMENT_WORKER_APP_ROOT_ENV, + DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV, +} from "#internal/nitro/host/dev-worker-metadata.js"; +import type { + DevelopmentWorkerRunner, + DevelopmentWorkerRunnerFactory, +} from "#internal/nitro/host/dev-worker-server.js"; + +class NodeDevelopmentWorkerRunner extends BaseEnvRunner implements DevelopmentWorkerRunner { + #closeCause: unknown; + readonly #environment: NodeJS.ProcessEnv; + #worker: Worker | undefined; + + constructor(input: Parameters[0]) { + const workerEntry = resolvePackageCompiledFilePath("src/compiled/env-runner/node-worker.js"); + super({ + data: { + entry: input.entry, + ...input.workerData, + }, + hooks: { + onClose: (_runner, cause) => input.onClose(cause), + }, + name: input.name, + workerEntry, + }); + this.#environment = { + ...process.env, + [DEVELOPMENT_WORKER_APP_ROOT_ENV]: input.appRoot, + [DEVELOPMENT_WORKER_TRANSPORT_SECRET_ENV]: input.transportSecret, + }; + this._initWithVirtualData(() => this.#startWorker()); + } + + override sendMessage(message: unknown): void { + if (this.#worker === undefined) { + throw new Error("Development worker is not initialized."); + } + this.#worker.postMessage(message); + } + + override async waitForReady(timeout: number): Promise { + try { + await super.waitForReady(timeout); + } catch (error) { + if (this.#closeCause === undefined) { + throw error; + } + throw new Error( + `Development worker failed before readiness: ${this.#closeCause instanceof Error ? this.#closeCause.message : String(this.#closeCause)}`, + { cause: this.#closeCause }, + ); + } + } + + protected override _hasRuntime(): boolean { + return this.#worker !== undefined; + } + + protected override _runtimeType(): string { + return "worker"; + } + + protected override async _closeRuntime(): Promise { + const worker = this.#worker; + if (worker === undefined) { + return; + } + + this.#worker = undefined; + worker.removeAllListeners(); + await worker.terminate(); + } + + protected override _handleMessage(message: unknown): void { + if (isWorkerInitializationError(message)) { + this.#closeCause = new Error(message.error); + } + super._handleMessage(message); + } + + #startWorker(): void { + if (!existsSync(this._workerEntry)) { + void this.close(`Development worker entry not found at "${this._workerEntry}".`); + return; + } + + const worker = new Worker(this._workerEntry, { + env: this.#environment, + workerData: { + name: this._name, + ...this._data, + }, + }); + this.#worker = worker; + worker.once("error", (error) => { + this.#closeCause = error; + void this.close(error); + }); + worker.once("exit", (code) => { + const error = new Error(`Development worker exited with code ${String(code)}.`); + this.#closeCause ??= error; + void this.close(error); + }); + worker.on("message", (message: unknown) => this._handleMessage(message)); + } +} + +export const createNodeDevelopmentWorkerRunner: DevelopmentWorkerRunnerFactory = (input) => + new NodeDevelopmentWorkerRunner(input); + +function isWorkerInitializationError( + value: unknown, +): value is { readonly error: string; readonly event: "init-error" } { + if (typeof value !== "object" || value === null) { + return false; + } + const record = value as Record; + return record.event === "init-error" && typeof record.error === "string"; +} diff --git a/packages/eve/src/internal/nitro/host/dev-worker-server.scenario.test.ts b/packages/eve/src/internal/nitro/host/dev-worker-server.scenario.test.ts new file mode 100644 index 000000000..df0440a80 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-worker-server.scenario.test.ts @@ -0,0 +1,610 @@ +import { Agent, request as requestHttp } from "node:http"; +import { connect } from "node:net"; + +import { describe, expect, it, vi } from "vitest"; + +import { + createDevelopmentWorkerServer, + type DevelopmentWorkerGeneration, + type DevelopmentWorkerRunner, + type DevelopmentWorkerRunnerFactory, +} from "#internal/nitro/host/dev-worker-server.js"; +import { decodeDevelopmentWorkerMetadata } from "#internal/nitro/host/dev-worker-metadata.js"; + +const TEST_DEADLINE_MS = 5_000; + +const FIRST_GENERATION: DevelopmentWorkerGeneration = { + id: "first", + runtimeAppRoot: "/tmp/eve-dev-runtime/first/source/app", +}; +const SECOND_GENERATION: DevelopmentWorkerGeneration = { + id: "second", + runtimeAppRoot: "/tmp/eve-dev-runtime/second/source/app", +}; + +function createDeferred(): { + readonly promise: Promise; + reject(error: unknown): void; + resolve(value: T): void; +} { + let rejectPromise: ((error: unknown) => void) | undefined; + let resolvePromise: ((value: T) => void) | undefined; + const promise = new Promise((resolve, reject) => { + rejectPromise = reject; + resolvePromise = resolve; + }); + + return { + promise, + reject(error) { + rejectPromise?.(error); + }, + resolve(value) { + resolvePromise?.(value); + }, + }; +} + +interface TestRunner extends DevelopmentWorkerRunner { + readonly closeMock: ReturnType; + readonly waitForReady: ReturnType Promise>>; + crash(error: Error): void; +} + +function createRunnerFactory( + fetchHandler: (request: Request, runnerIndex: number, secret: string) => Promise, + waitForReady: (runnerIndex: number) => Promise = async () => undefined, + upgradeHandler: ( + input: Parameters[0], + runnerIndex: number, + secret: string, + ) => Promise = async () => undefined, +): { + readonly createRunner: DevelopmentWorkerRunnerFactory; + readonly runners: TestRunner[]; +} { + const runners: TestRunner[] = []; + const createRunner: DevelopmentWorkerRunnerFactory = (input) => { + const runnerIndex = runners.length; + let closed = false; + const closeMock = vi.fn(async () => { + closed = true; + input.onClose(); + }); + const runner: TestRunner = { + close: closeMock, + closeMock, + crash(error) { + closed = true; + input.onClose(error); + }, + fetch: async (request) => await fetchHandler(request, runnerIndex, input.transportSecret), + get closed() { + return closed; + }, + upgrade: vi.fn( + async (upgradeInput) => + await upgradeHandler(upgradeInput, runnerIndex, input.transportSecret), + ), + waitForReady: vi.fn(async () => await waitForReady(runnerIndex)), + }; + runners.push(runner); + return runner; + }; + + return { createRunner, runners }; +} + +async function listen(createRunner: DevelopmentWorkerRunnerFactory): Promise<{ + readonly server: ReturnType; + readonly url: string; +}> { + return await listenWithGenerationResolver(createRunner, (generation) => generation); +} + +async function listenWithGenerationResolver( + createRunner: DevelopmentWorkerRunnerFactory, + resolveAdmissionGeneration: ( + workerGeneration: DevelopmentWorkerGeneration, + ) => DevelopmentWorkerGeneration, +): Promise<{ + readonly server: ReturnType; + readonly url: string; +}> { + const server = createDevelopmentWorkerServer({ + appRoot: "/tmp/eve-dev-worker-test", + createRunner, + resolveAdmissionGeneration, + }); + const listener = server.listen({ hostname: "127.0.0.1", port: 0 }); + await listener.ready(); + if (listener.url === undefined) { + throw new Error("Development worker listener did not expose a URL."); + } + + return { server, url: listener.url }; +} + +async function closeWithinDeadline(close: () => Promise): Promise { + let timer: NodeJS.Timeout | undefined; + const deadline = new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error("Development worker server close timed out.")), + TEST_DEADLINE_MS, + ); + }); + + try { + await Promise.race([close(), deadline]); + } finally { + clearTimeout(timer); + } +} + +async function withinDeadline(operation: Promise, message: string): Promise { + let timer: NodeJS.Timeout | undefined; + const deadline = new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(message)), TEST_DEADLINE_MS); + }); + + try { + return await Promise.race([operation, deadline]); + } finally { + clearTimeout(timer); + } +} + +describe("development worker server", () => { + it("keeps the active worker while a candidate is unready or fails readiness", async () => { + const candidateReadiness = createDeferred(); + const { createRunner, runners } = createRunnerFactory( + async (_request, runnerIndex) => new Response(runnerIndex === 0 ? "active" : "candidate"), + async (runnerIndex) => { + if (runnerIndex === 1) { + await candidateReadiness.promise; + } + }, + ); + const { server, url } = await listen(createRunner); + const first = await server.prepareCandidate({ + entry: "/tmp/first.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(first); + server.setControlHandler(async (request) => { + const pathname = new URL(request.url).pathname; + return pathname === "/control" ? new Response("parent") : undefined; + }); + + const pendingCandidate = server.prepareCandidate({ + entry: "/tmp/second.mjs", + generation: SECOND_GENERATION, + workerData: {}, + }); + await vi.waitFor(() => { + expect(runners[1]?.waitForReady).toHaveBeenCalledOnce(); + }); + await expect( + fetch(new URL("/control", url)).then(async (response) => await response.text()), + ).resolves.toBe("parent"); + await expect(fetch(url).then(async (response) => await response.text())).resolves.toBe( + "active", + ); + + candidateReadiness.reject(new Error("candidate failed readiness")); + await expect(pendingCandidate).rejects.toThrow("candidate failed readiness"); + await expect(fetch(url).then(async (response) => await response.text())).resolves.toBe( + "active", + ); + expect(runners[0]?.closeMock).not.toHaveBeenCalled(); + + await closeWithinDeadline(() => server.close()); + }); + + it("restarts an active worker after a crash without moving its generation", async () => { + const { createRunner, runners } = createRunnerFactory(async (request, runnerIndex, secret) => { + const metadata = decodeDevelopmentWorkerMetadata({ + header: request.headers.get("x-eve-dev-worker-metadata"), + secret, + }); + return new Response(`${String(runnerIndex)}:${metadata.generationId}`); + }); + const { server, url } = await listen(createRunner); + const first = await server.prepareCandidate({ + entry: "/tmp/first.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(first); + + runners[0]?.crash(new Error("worker crashed")); + await vi.waitFor(async () => { + await expect(fetch(url).then(async (response) => await response.text())).resolves.toBe( + "1:first", + ); + }); + + await closeWithinDeadline(() => server.close()); + }); + + it("admits later requests on a new runtime generation without replacing the worker", async () => { + let admissionGeneration = FIRST_GENERATION; + const { createRunner, runners } = createRunnerFactory(async (request, runnerIndex, secret) => { + const metadata = decodeDevelopmentWorkerMetadata({ + header: request.headers.get("x-eve-dev-worker-metadata"), + secret, + }); + return new Response(`${String(runnerIndex)}:${metadata.generationId}`); + }); + const { server, url } = await listenWithGenerationResolver( + createRunner, + () => admissionGeneration, + ); + const candidate = await server.prepareCandidate({ + entry: "/tmp/first.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(candidate); + + await expect(fetch(url).then(async (response) => await response.text())).resolves.toBe( + "0:first", + ); + admissionGeneration = SECOND_GENERATION; + await expect(fetch(url).then(async (response) => await response.text())).resolves.toBe( + "0:second", + ); + expect(runners).toHaveLength(1); + + await closeWithinDeadline(() => server.close()); + }); + + it("terminates an active client response when its worker crashes", async () => { + const responseStarted = createDeferred(); + const { createRunner, runners } = createRunnerFactory( + async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("first\n")); + responseStarted.resolve(); + }, + }), + ), + ); + const { server, url } = await listen(createRunner); + const candidate = await server.prepareCandidate({ + entry: "/tmp/stream.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(candidate); + + const response = await fetch(url); + const reader = response.body?.getReader(); + await responseStarted.promise; + await expect(reader?.read()).resolves.toEqual(expect.objectContaining({ done: false })); + runners[0]?.crash(new Error("worker crashed")); + + await expect( + withinDeadline( + reader?.read() ?? Promise.reject(new Error("Missing response body.")), + "Client response remained open after its worker crashed.", + ), + ).rejects.toThrow(); + await closeWithinDeadline(() => server.close()); + }); + + it("releases a retired worker when the client disconnects from its response", async () => { + const bodyCancelled = createDeferred(); + const firstChunk = createDeferred(); + const { createRunner, runners } = createRunnerFactory(async (_request, runnerIndex) => { + if (runnerIndex !== 0) { + return new Response("next"); + } + + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("first\n")); + firstChunk.resolve(); + }, + cancel() { + bodyCancelled.resolve(); + }, + }), + ); + }); + const { server, url } = await listen(createRunner); + const first = await server.prepareCandidate({ + entry: "/tmp/first.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(first); + + const response = await fetch(url); + await firstChunk.promise; + const second = await server.prepareCandidate({ + entry: "/tmp/second.mjs", + generation: SECOND_GENERATION, + workerData: {}, + }); + await server.promote(second); + + expect(runners[0]?.closeMock).not.toHaveBeenCalled(); + await response.body?.cancel(); + await withinDeadline( + bodyCancelled.promise, + "Worker response body was not cancelled after the client disconnected.", + ); + await vi.waitFor(() => { + expect(runners[0]?.closeMock).toHaveBeenCalledOnce(); + }); + + await closeWithinDeadline(() => server.close()); + }); + + it("releases a retired worker when the client aborts a request body", async () => { + const bodyAborted = createDeferred(); + const bodyStarted = createDeferred(); + const { createRunner, runners } = createRunnerFactory(async (request, runnerIndex) => { + if (runnerIndex !== 0) { + return new Response("next"); + } + bodyStarted.resolve(); + await request.text().catch(() => bodyAborted.resolve()); + return new Response("unexpected"); + }); + const { server, url } = await listen(createRunner); + const first = await server.prepareCandidate({ + entry: "/tmp/first.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(first); + + const target = new URL(url); + const request = requestHttp({ + headers: { "content-type": "text/plain", "transfer-encoding": "chunked" }, + host: target.hostname, + method: "POST", + path: "/", + port: Number(target.port), + }); + request.on("error", () => undefined); + request.write("partial body"); + await bodyStarted.promise; + const second = await server.prepareCandidate({ + entry: "/tmp/second.mjs", + generation: SECOND_GENERATION, + workerData: {}, + }); + await server.promote(second); + expect(runners[0]?.closeMock).not.toHaveBeenCalled(); + + request.destroy(); + await withinDeadline( + bodyAborted.promise, + "Worker request body did not observe the client cancellation.", + ); + await vi.waitFor(() => { + expect(runners[0]?.closeMock).toHaveBeenCalledOnce(); + }); + + await closeWithinDeadline(() => server.close()); + }); + + it("keeps the public HTTP connection alive across worker promotion", async () => { + const { createRunner } = createRunnerFactory( + async (_request, runnerIndex) => new Response(String(runnerIndex)), + ); + const { server, url } = await listen(createRunner); + const first = await server.prepareCandidate({ + entry: "/tmp/first.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(first); + const agent = new Agent({ keepAlive: true, maxSockets: 1 }); + + try { + const firstResponse = await requestWithAgent(url, agent); + const second = await server.prepareCandidate({ + entry: "/tmp/second.mjs", + generation: SECOND_GENERATION, + workerData: {}, + }); + await server.promote(second); + const secondResponse = await requestWithAgent(url, agent); + + expect(firstResponse.body).toBe("0"); + expect(secondResponse.body).toBe("1"); + expect(secondResponse.localPort).toBe(firstResponse.localPort); + } finally { + agent.destroy(); + await closeWithinDeadline(() => server.close()); + } + }); + + it("holds a websocket lease until close and stamps the parent socket address", async () => { + let clientAddress: string | null | undefined; + const upgraded = createDeferred(); + const { createRunner, runners } = createRunnerFactory( + async () => new Response("not used"), + async () => undefined, + async (input, _runnerIndex, secret) => { + const metadata = decodeDevelopmentWorkerMetadata({ + header: input.node.req.headers["x-eve-dev-worker-metadata"] as string, + secret, + }); + clientAddress = metadata.clientAddress; + input.node.socket.write( + "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n", + ); + upgraded.resolve(); + }, + ); + const { server, url } = await listen(createRunner); + const first = await server.prepareCandidate({ + entry: "/tmp/first.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(first); + const target = new URL(url); + const socket = connect({ host: target.hostname, port: Number(target.port) }); + socket.on("error", () => undefined); + socket.write( + [ + "GET /socket HTTP/1.1", + `Host: ${target.host}`, + "Connection: Upgrade", + "Upgrade: websocket", + "x-eve-dev-worker-metadata: public-spoof", + "", + "", + ].join("\r\n"), + ); + await upgraded.promise; + const second = await server.prepareCandidate({ + entry: "/tmp/second.mjs", + generation: SECOND_GENERATION, + workerData: {}, + }); + await server.promote(second); + + expect(clientAddress).toBe("127.0.0.1"); + expect(runners[0]?.closeMock).not.toHaveBeenCalled(); + socket.destroy(); + await vi.waitFor(() => { + expect(runners[0]?.closeMock).toHaveBeenCalledOnce(); + }); + + await closeWithinDeadline(() => server.close()); + }); + + it("stops the worker producing an active stream before waiting for listener shutdown", async () => { + const responseStarted = createDeferred(); + const { createRunner, runners } = createRunnerFactory( + async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: ready\n\n")); + responseStarted.resolve(); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ), + ); + const { server, url } = await listen(createRunner); + const candidate = await server.prepareCandidate({ + entry: "/tmp/stream.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(candidate); + + const response = await fetch(url); + const reader = response.body?.getReader(); + await responseStarted.promise; + await expect(reader?.read()).resolves.toEqual(expect.objectContaining({ done: false })); + await closeWithinDeadline(() => server.close()); + + expect(runners[0]?.closeMock).toHaveBeenCalledOnce(); + await expect(reader?.read()).rejects.toThrow(); + }); + + it("preserves the socket client address in signed metadata and replaces public spoofing", async () => { + const { createRunner } = createRunnerFactory(async (request, _runnerIndex, secret) => { + const metadata = decodeDevelopmentWorkerMetadata({ + header: request.headers.get("x-eve-dev-worker-metadata"), + secret, + }); + return Response.json({ + clientAddress: metadata.clientAddress, + runtimeAppRoot: metadata.runtimeAppRoot, + }); + }); + const { server, url } = await listen(createRunner); + const candidate = await server.prepareCandidate({ + entry: "/tmp/metadata.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(candidate); + + const body = await new Promise((resolve, reject) => { + const target = new URL(url); + const request = requestHttp( + { + headers: { "x-eve-dev-worker-metadata": "public-spoof" }, + host: target.hostname, + path: "/", + port: Number(target.port), + }, + (response) => { + const chunks: Buffer[] = []; + response.on("data", (chunk: Buffer) => chunks.push(chunk)); + response.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); + }, + ); + request.on("error", reject); + request.end(); + }); + + expect(JSON.parse(body)).toEqual({ + clientAddress: "127.0.0.1", + runtimeAppRoot: FIRST_GENERATION.runtimeAppRoot, + }); + + await closeWithinDeadline(() => server.close()); + }); + + it("closes workers and listeners only once when close is repeated", async () => { + const { createRunner, runners } = createRunnerFactory(async () => new Response("ok")); + const { server } = await listen(createRunner); + const candidate = await server.prepareCandidate({ + entry: "/tmp/close.mjs", + generation: FIRST_GENERATION, + workerData: {}, + }); + await server.promote(candidate); + + await closeWithinDeadline(async () => { + await Promise.all([server.close(), server.close()]); + await server.close(); + }); + + expect(runners[0]?.closeMock).toHaveBeenCalledOnce(); + }); +}); + +async function requestWithAgent( + url: string, + agent: Agent, +): Promise<{ readonly body: string; readonly localPort: number | undefined }> { + return await new Promise((resolve, reject) => { + const target = new URL(url); + const request = requestHttp( + { + agent, + host: target.hostname, + path: "/", + port: Number(target.port), + }, + (response) => { + const chunks: Buffer[] = []; + const localPort = response.socket.localPort; + response.on("data", (chunk: Buffer) => chunks.push(chunk)); + response.on("end", () => { + resolve({ body: Buffer.concat(chunks).toString("utf8"), localPort }); + }); + }, + ); + request.on("error", reject); + request.end(); + }); +} diff --git a/packages/eve/src/internal/nitro/host/dev-worker-server.ts b/packages/eve/src/internal/nitro/host/dev-worker-server.ts new file mode 100644 index 000000000..321c05656 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-worker-server.ts @@ -0,0 +1,515 @@ +import { randomBytes } from "node:crypto"; +import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http"; +import type { Socket } from "node:net"; + +import { + closeServer, + createPublicRequest, + createWorkerRequest, + stampWorkerUpgradeRequest, + writeRequestError, + writeResponse, +} from "#internal/nitro/host/dev-worker-http.js"; +import { toErrorMessage } from "#shared/errors.js"; + +const DEVELOPMENT_WORKER_READY_TIMEOUT_MS = 60_000; + +export interface DevelopmentWorkerGeneration { + readonly id: string; + readonly runtimeAppRoot: string; +} + +export interface DevelopmentWorkerRunner { + readonly closed: boolean; + close(cause?: unknown): Promise; + fetch(request: Request): Promise; + upgrade(input: { + readonly node: { + readonly head: Buffer; + readonly req: IncomingMessage; + readonly socket: Socket; + }; + }): Promise; + waitForReady(timeout: number): Promise; +} + +export interface DevelopmentWorkerRunnerFactoryInput { + readonly appRoot: string; + readonly entry: string; + readonly name: string; + readonly onClose: (cause?: unknown) => void; + readonly transportSecret: string; + readonly workerData: Readonly>; +} + +export type DevelopmentWorkerRunnerFactory = ( + input: DevelopmentWorkerRunnerFactoryInput, +) => DevelopmentWorkerRunner; + +type WorkerState = "candidate" | "active" | "retired" | "closed"; + +interface WorkerSlot { + readonly entry: string; + readonly generation: DevelopmentWorkerGeneration; + leases: number; + readonly runner: DevelopmentWorkerRunner; + state: WorkerState; + readonly workerData: Readonly>; +} + +export interface DevelopmentWorkerCandidate { + readonly slot: WorkerSlot; +} + +export interface DevelopmentWorkerListener { + close(): Promise; + readonly node: { readonly server: Server }; + ready(): Promise; + readonly url: string | undefined; +} + +interface RequestLease { + readonly generation: DevelopmentWorkerGeneration; + release(): void; + readonly slot: WorkerSlot; +} + +interface ActiveExchange { + cancel(): void; + readonly slot: WorkerSlot; +} + +interface ListenerState { + readonly beginClose: () => Promise; + destroySockets(): void; +} + +export interface DevelopmentWorkerServer { + close(): Promise; + discardCandidate(candidate: DevelopmentWorkerCandidate): Promise; + listen(input: { readonly hostname: string; readonly port: number }): DevelopmentWorkerListener; + prepareCandidate(input: { + readonly entry: string; + readonly generation: DevelopmentWorkerGeneration; + readonly workerData: Readonly>; + }): Promise; + promote(candidate: DevelopmentWorkerCandidate): Promise; + setControlHandler(handler: (request: Request) => Promise): void; +} + +export function createDevelopmentWorkerServer(input: { + readonly appRoot: string; + readonly createRunner: DevelopmentWorkerRunnerFactory; + readonly resolveAdmissionGeneration: ( + workerGeneration: DevelopmentWorkerGeneration, + ) => DevelopmentWorkerGeneration; +}): DevelopmentWorkerServer { + return new ParentDevelopmentWorkerServer(input); +} + +class ParentDevelopmentWorkerServer implements DevelopmentWorkerServer { + readonly #activeExchanges = new Set(); + readonly #activeWaiters = new Set<() => void>(); + readonly #appRoot: string; + readonly #createRunner: DevelopmentWorkerRunnerFactory; + readonly #listeners = new Set(); + readonly #resolveAdmissionGeneration: ( + workerGeneration: DevelopmentWorkerGeneration, + ) => DevelopmentWorkerGeneration; + readonly #slots = new Set(); + readonly #transportSecret = randomBytes(32).toString("base64url"); + #accepting = true; + #activeSlot: WorkerSlot | undefined; + #closePromise: Promise | undefined; + #controlHandler: ((request: Request) => Promise) | undefined; + #promotion: Promise = Promise.resolve(); + #workerCounter = 0; + + constructor(input: { + readonly appRoot: string; + readonly createRunner: DevelopmentWorkerRunnerFactory; + readonly resolveAdmissionGeneration: ( + workerGeneration: DevelopmentWorkerGeneration, + ) => DevelopmentWorkerGeneration; + }) { + this.#appRoot = input.appRoot; + this.#createRunner = input.createRunner; + this.#resolveAdmissionGeneration = input.resolveAdmissionGeneration; + } + + setControlHandler(handler: (request: Request) => Promise): void { + this.#controlHandler = handler; + } + + listen(input: { readonly hostname: string; readonly port: number }): DevelopmentWorkerListener { + if (!this.#accepting) { + throw new Error("Development worker server is closed."); + } + + const sockets = new Set(); + const server = createServer((request, response) => { + void this.#handleRequest(request, response); + }); + server.on("connection", (socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + }); + server.on("upgrade", (request, socket, head) => { + void this.#handleUpgrade(request, socket as Socket, head); + }); + + let url: string | undefined; + const ready = new Promise((resolve, reject) => { + const onError = (error: Error) => { + server.off("listening", onListening); + reject(error); + }; + const onListening = () => { + server.off("error", onError); + const address = server.address(); + if (address === null || typeof address === "string") { + reject(new Error("Development worker server did not expose a TCP address.")); + return; + } + const urlHost = input.hostname.includes(":") ? `[${input.hostname}]` : input.hostname; + url = `http://${urlHost}:${String(address.port)}/`; + resolve(); + }; + server.once("error", onError); + server.once("listening", onListening); + server.listen({ host: input.hostname, port: input.port }); + }); + let listenerClosePromise: Promise | undefined; + const beginClose = (): Promise => { + listenerClosePromise ??= closeServer(server).finally(() => { + this.#listeners.delete(listenerState); + }); + return listenerClosePromise; + }; + const destroySockets = () => { + for (const socket of sockets) { + socket.destroy(); + } + }; + const listenerState: ListenerState = { beginClose, destroySockets }; + this.#listeners.add(listenerState); + + return { + async close() { + const closed = beginClose(); + destroySockets(); + await closed; + }, + node: { server }, + ready: async () => await ready, + get url() { + return url; + }, + }; + } + + async prepareCandidate(input: { + readonly entry: string; + readonly generation: DevelopmentWorkerGeneration; + readonly workerData: Readonly>; + }): Promise { + if (!this.#accepting) { + throw new Error("Development worker server is closed."); + } + + let slot: WorkerSlot | undefined; + const workerNumber = this.#workerCounter++; + const runner = this.#createRunner({ + appRoot: this.#appRoot, + entry: input.entry, + name: `eve-dev-${String(workerNumber)}`, + onClose: (cause) => { + if (slot !== undefined) { + void this.#handleWorkerClose(slot, cause); + } + }, + transportSecret: this.#transportSecret, + workerData: input.workerData, + }); + slot = { + entry: input.entry, + generation: input.generation, + leases: 0, + runner, + state: "candidate", + workerData: input.workerData, + }; + this.#slots.add(slot); + + try { + await runner.waitForReady(DEVELOPMENT_WORKER_READY_TIMEOUT_MS); + } catch (error) { + await this.#closeSlot(slot, error); + throw error; + } + + if (!this.#accepting) { + await this.#closeSlot(slot); + throw new Error("Development worker server closed before candidate readiness."); + } + + return { slot }; + } + + async promote(candidate: DevelopmentWorkerCandidate): Promise { + const previousPromotion = this.#promotion; + let finishPromotion: (() => void) | undefined; + this.#promotion = new Promise((resolve) => { + finishPromotion = resolve; + }); + + await previousPromotion; + try { + if (!this.#accepting) { + throw new Error("Development worker server is closed."); + } + if (candidate.slot.state !== "candidate" || candidate.slot.runner.closed) { + throw new Error("Development worker candidate is not ready for promotion."); + } + + const previousSlot = this.#activeSlot; + candidate.slot.state = "active"; + this.#activeSlot = candidate.slot; + this.#wakeActiveWaiters(); + + if (previousSlot !== undefined && previousSlot !== candidate.slot) { + previousSlot.state = "retired"; + await this.#closeRetiredSlot(previousSlot); + } + } finally { + finishPromotion?.(); + } + } + + async discardCandidate(candidate: DevelopmentWorkerCandidate): Promise { + if (candidate.slot.state !== "candidate") { + throw new Error("Only an unpromoted development worker candidate can be discarded."); + } + await this.#closeSlot(candidate.slot); + } + + close(): Promise { + this.#closePromise ??= this.#close(); + return this.#closePromise; + } + + async #close(): Promise { + this.#accepting = false; + this.#wakeActiveWaiters(); + + const listeners = [...this.#listeners]; + const listenerClosePromises = listeners.map((listener) => listener.beginClose()); + + for (const exchange of this.#activeExchanges) { + exchange.cancel(); + } + + await Promise.all([...this.#slots].map(async (slot) => await this.#closeSlot(slot))); + + for (const listener of listeners) { + listener.destroySockets(); + } + + await Promise.all(listenerClosePromises); + } + + async #handleRequest(request: IncomingMessage, response: ServerResponse): Promise { + const requestAbort = new AbortController(); + const abortRequest = (cause?: unknown) => { + if (!requestAbort.signal.aborted) { + requestAbort.abort(cause); + } + }; + request.once("aborted", abortRequest); + request.once("error", abortRequest); + response.once("close", () => { + if (!response.writableEnded) { + abortRequest(); + } + }); + + let lease: RequestLease | undefined; + let exchange: ActiveExchange | undefined; + try { + const publicRequest = createPublicRequest(request, requestAbort.signal); + const controlResponse = await this.#controlHandler?.(publicRequest); + if (controlResponse !== undefined) { + await writeResponse(response, controlResponse, requestAbort.signal); + return; + } + + lease = await this.#admit(); + const workerRequest = createWorkerRequest({ + generation: lease.generation, + request: publicRequest, + secret: this.#transportSecret, + socket: request.socket, + }); + const admittedLease = lease; + exchange = { + cancel() { + abortRequest(new Error("Development worker request was cancelled.")); + if (!response.destroyed) { + response.destroy(); + } + admittedLease.release(); + }, + slot: lease.slot, + }; + this.#activeExchanges.add(exchange); + response.once("finish", admittedLease.release); + response.once("error", admittedLease.release); + response.once("close", admittedLease.release); + request.once("aborted", admittedLease.release); + + const workerResponse = await lease.slot.runner.fetch(workerRequest); + await writeResponse(response, workerResponse, requestAbort.signal); + } catch (error) { + if (!requestAbort.signal.aborted) { + writeRequestError(response, error); + } + } finally { + if (exchange !== undefined) { + this.#activeExchanges.delete(exchange); + } + lease?.release(); + request.off("aborted", abortRequest); + request.off("error", abortRequest); + } + } + + async #handleUpgrade(request: IncomingMessage, socket: Socket, head: Buffer): Promise { + let lease: RequestLease | undefined; + let exchange: ActiveExchange | undefined; + try { + lease = await this.#admit(); + stampWorkerUpgradeRequest({ + generation: lease.generation, + request, + secret: this.#transportSecret, + }); + const admittedLease = lease; + exchange = { + cancel() { + if (!socket.destroyed) { + socket.destroy(); + } + admittedLease.release(); + }, + slot: lease.slot, + }; + this.#activeExchanges.add(exchange); + const release = () => { + admittedLease.release(); + if (exchange !== undefined) { + this.#activeExchanges.delete(exchange); + exchange = undefined; + } + }; + socket.once("close", release); + socket.once("error", release); + await lease.slot.runner.upgrade({ node: { head, req: request, socket } }); + } catch { + if (exchange !== undefined) { + exchange.cancel(); + } else { + // Failure before the exchange existed: no socket listener releases + // the lease, so release it here or the retired worker never closes. + lease?.release(); + } + if (!socket.destroyed) { + socket.destroy(); + } + } + } + + async #admit(): Promise { + await this.#promotion; + while (this.#accepting && this.#activeSlot === undefined) { + await new Promise((resolve) => this.#activeWaiters.add(resolve)); + await this.#promotion; + } + + const slot = this.#activeSlot; + if (!this.#accepting || slot === undefined || slot.state !== "active") { + throw new Error("Development worker is unavailable."); + } + + const generation = this.#resolveAdmissionGeneration(slot.generation); + slot.leases += 1; + let released = false; + return { + generation, + release: () => { + if (released) { + return; + } + released = true; + slot.leases -= 1; + void this.#closeRetiredSlot(slot).catch((error) => { + console.error(`[eve:dev] failed to close retired worker: ${toErrorMessage(error)}`); + }); + }, + slot, + }; + } + + async #handleWorkerClose(slot: WorkerSlot, cause: unknown): Promise { + if (slot.state !== "active" || !this.#accepting || this.#activeSlot !== slot) { + return; + } + + slot.state = "closed"; + this.#slots.delete(slot); + this.#activeSlot = undefined; + for (const exchange of this.#activeExchanges) { + if (exchange.slot === slot) { + exchange.cancel(); + this.#activeExchanges.delete(exchange); + } + } + + try { + const candidate = await this.prepareCandidate({ + entry: slot.entry, + generation: slot.generation, + workerData: slot.workerData, + }); + await this.promote(candidate); + } catch (error) { + console.error( + `[eve:dev] worker restart failed after ${toErrorMessage(cause)}: ${toErrorMessage(error)}`, + ); + } + } + + async #closeRetiredSlot(slot: WorkerSlot): Promise { + if (slot.state === "retired" && slot.leases === 0) { + await this.#closeSlot(slot); + } + } + + async #closeSlot(slot: WorkerSlot, cause?: unknown): Promise { + if (slot.state === "closed") { + return; + } + slot.state = "closed"; + this.#slots.delete(slot); + if (this.#activeSlot === slot) { + this.#activeSlot = undefined; + } + await slot.runner.close(cause); + } + + #wakeActiveWaiters(): void { + for (const wake of this.#activeWaiters) { + wake(); + } + this.#activeWaiters.clear(); + } +} diff --git a/packages/eve/src/internal/nitro/host/dispatch-schedule-in-dev.ts b/packages/eve/src/internal/nitro/host/dispatch-schedule-in-dev.ts index 09326e5fb..87831dad5 100644 --- a/packages/eve/src/internal/nitro/host/dispatch-schedule-in-dev.ts +++ b/packages/eve/src/internal/nitro/host/dispatch-schedule-in-dev.ts @@ -1,7 +1,7 @@ import type { DevelopmentNitroArtifactsConfig } from "#internal/nitro/routes/runtime-artifacts.js"; -import { createDiskRuntimeCompiledArtifactsSource } from "#runtime/compiled-artifacts-source.js"; import { createScheduleRegistrations } from "#runtime/schedules/register.js"; import { loadResolvedCompiledSchedules } from "#runtime/schedules/resolve-schedule.js"; +import { resolveNitroRequestCompiledArtifactsSource } from "#internal/nitro/routes/runtime-artifacts.js"; /** * Result of dispatching one authored schedule via the dev-only HTTP route. @@ -58,6 +58,7 @@ export class UnknownDevScheduleError extends Error { */ export async function dispatchScheduleInDev(input: { readonly artifactsConfig: DevelopmentNitroArtifactsConfig; + readonly request: Request; readonly scheduleId: string; }): Promise { const { appRoot, moduleMapLoaderPath } = input.artifactsConfig; @@ -67,9 +68,10 @@ export async function dispatchScheduleInDev(input: { ); } - const compiledArtifactsSource = createDiskRuntimeCompiledArtifactsSource(appRoot, { - moduleMapLoaderPath, - }); + const compiledArtifactsSource = resolveNitroRequestCompiledArtifactsSource( + input.artifactsConfig, + input.request, + ); const schedules = await loadResolvedCompiledSchedules({ compiledArtifactsSource }); const registrations = createScheduleRegistrations(schedules); const registration = registrations.find((candidate) => candidate.scheduleId === input.scheduleId); @@ -81,8 +83,12 @@ export async function dispatchScheduleInDev(input: { ); } - const { dispatchScheduleTask } = await import("#internal/nitro/routes/schedule-task.js"); - const result = await dispatchScheduleTask(registration.taskName, input.artifactsConfig); + const { dispatchScheduleTaskFromArtifacts } = + await import("#internal/nitro/routes/schedule-task.js"); + const result = await dispatchScheduleTaskFromArtifacts( + registration.taskName, + compiledArtifactsSource, + ); return { scheduleId: result.scheduleId, diff --git a/packages/eve/src/internal/nitro/host/nitro-development-worker-server.test.ts b/packages/eve/src/internal/nitro/host/nitro-development-worker-server.test.ts new file mode 100644 index 000000000..20675f00c --- /dev/null +++ b/packages/eve/src/internal/nitro/host/nitro-development-worker-server.test.ts @@ -0,0 +1,82 @@ +import type { Nitro } from "nitro/types"; +import { describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + core: { + close: vi.fn(async () => undefined), + discardCandidate: vi.fn(async () => undefined), + listen: vi.fn(), + prepareCandidate: vi.fn(), + promote: vi.fn(async () => undefined), + setControlHandler: vi.fn(), + }, +})); + +vi.mock("#internal/nitro/host/dev-worker-server.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + createDevelopmentWorkerServer: vi.fn(() => mocks.core), + }; +}); + +vi.mock("#internal/nitro/host/dev-worker-runner.js", () => ({ + createNodeDevelopmentWorkerRunner: vi.fn(), +})); + +function createDeferred(): { + readonly promise: Promise; + resolve(value: T): void; +} { + let resolvePromise: ((value: T) => void) | undefined; + const promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + return { + promise, + resolve(value) { + resolvePromise?.(value); + }, + }; +} + +describe("NitroDevelopmentWorkerServer", () => { + it("discards a candidate that becomes ready after Nitro rejects its build", async () => { + const hooks = new Map unknown>(); + const candidateReady = createDeferred(); + const candidate = {}; + mocks.core.prepareCandidate.mockReturnValueOnce(candidateReady.promise); + const hook = ((name: string, handler: (...args: never[]) => unknown) => { + hooks.set(name, handler); + }) as Nitro["hooks"]["hook"]; + const nitro = { + hooks: { + hook, + }, + logger: { error: vi.fn() }, + options: { + output: { dir: "/tmp/nitro", serverDir: "server" }, + }, + }; + const { NitroDevelopmentWorkerServer } = + await import("#internal/nitro/host/nitro-development-worker-server.js"); + const server = new NitroDevelopmentWorkerServer({ appRoot: "/tmp/app", nitro }); + const buildError = new Error("Nitro build failed"); + + const result = server.buildCandidate({ + generation: { id: "generation-a", runtimeAppRoot: "/tmp/generation-a" }, + trigger: async () => { + const reload = hooks.get("dev:reload")?.({ + entry: "/tmp/index.mjs", + workerData: {}, + } as never); + hooks.get("dev:error")?.(buildError as never); + candidateReady.resolve(candidate); + await reload; + }, + }); + + await expect(result).rejects.toThrow("Nitro build failed"); + expect(mocks.core.discardCandidate).toHaveBeenCalledWith(candidate); + }); +}); diff --git a/packages/eve/src/internal/nitro/host/nitro-development-worker-server.ts b/packages/eve/src/internal/nitro/host/nitro-development-worker-server.ts new file mode 100644 index 000000000..6c94b2898 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/nitro-development-worker-server.ts @@ -0,0 +1,205 @@ +import { basename, resolve } from "node:path"; + +import type { Nitro } from "nitro/types"; + +import { readActiveDevelopmentRuntimeArtifactsSnapshot } from "#internal/nitro/dev-runtime-artifacts.js"; +import { + createDevelopmentWorkerServer, + type DevelopmentWorkerCandidate, + type DevelopmentWorkerGeneration, + type DevelopmentWorkerListener, + type DevelopmentWorkerServer, +} from "#internal/nitro/host/dev-worker-server.js"; +import { createNodeDevelopmentWorkerRunner } from "#internal/nitro/host/dev-worker-runner.js"; +import { toErrorMessage } from "#shared/errors.js"; + +interface PendingCandidate { + readonly generation: DevelopmentWorkerGeneration; + readonly promise: Promise; + reject(error: unknown): void; + resolve(candidate: DevelopmentWorkerCandidate): void; +} + +interface NitroDevelopmentWorkerHost { + readonly hooks: Pick; + readonly logger: { + error(message: unknown, ...details: unknown[]): unknown; + }; + readonly options: { + readonly output: Pick; + }; +} + +export class NitroDevelopmentWorkerServer { + readonly #appRoot: string; + readonly #nitro: NitroDevelopmentWorkerHost; + #pendingCandidate: PendingCandidate | undefined; + readonly #server: DevelopmentWorkerServer; + + constructor(input: { readonly appRoot: string; readonly nitro: NitroDevelopmentWorkerHost }) { + this.#appRoot = input.appRoot; + this.#nitro = input.nitro; + this.#server = createDevelopmentWorkerServer({ + appRoot: input.appRoot, + createRunner: createNodeDevelopmentWorkerRunner, + resolveAdmissionGeneration: () => { + const generation = readActiveGeneration(this.#appRoot); + if (generation === undefined) { + throw new Error("Development runtime generation is unavailable for request admission."); + } + return generation; + }, + }); + + input.nitro.hooks.hook("dev:reload", async (payload) => { + await this.#handleReload({ + entry: payload?.entry ?? this.#resolveDefaultEntry(), + workerData: payload?.workerData ?? {}, + }); + }); + input.nitro.hooks.hook("dev:error", (error) => { + const pending = this.#pendingCandidate; + if (pending !== undefined) { + this.#pendingCandidate = undefined; + pending.reject(error); + } + }); + input.nitro.hooks.hook("close", async () => await this.close()); + } + + listen(input: { readonly hostname: string; readonly port: number }): DevelopmentWorkerListener { + return this.#server.listen(input); + } + + setControlHandler(handler: (request: Request) => Promise): void { + this.#server.setControlHandler(handler); + } + + async buildCandidate(input: { + readonly generation: DevelopmentWorkerGeneration; + readonly trigger: () => Promise; + }): Promise { + if (this.#pendingCandidate !== undefined) { + throw new Error("A development worker candidate is already pending."); + } + + const pending = createPendingCandidate(input.generation); + void pending.promise.catch(() => undefined); + this.#pendingCandidate = pending; + try { + await input.trigger(); + return await pending.promise; + } catch (error) { + if (this.#pendingCandidate === pending) { + this.#pendingCandidate = undefined; + pending.reject(error); + } + const candidate = await pending.promise.catch(() => undefined); + if (candidate !== undefined) { + await this.#server.discardCandidate(candidate); + } + throw error; + } + } + + async promote(candidate: DevelopmentWorkerCandidate): Promise { + await this.#server.promote(candidate); + } + + async close(): Promise { + const pending = this.#pendingCandidate; + if (pending !== undefined) { + this.#pendingCandidate = undefined; + pending.reject(new Error("Development worker server closed.")); + } + await this.#server.close(); + } + + async #handleReload(input: { + readonly entry: string; + readonly workerData: Readonly>; + }): Promise { + const pending = this.#pendingCandidate; + const generation = pending?.generation ?? readActiveGeneration(this.#appRoot); + if (generation === undefined) { + const error = new Error("Nitro produced a worker without an active development generation."); + if (pending !== undefined) { + this.#pendingCandidate = undefined; + pending.reject(error); + return; + } + throw error; + } + + try { + const candidate = await this.#server.prepareCandidate({ + entry: input.entry, + generation, + workerData: input.workerData, + }); + if (pending !== undefined && this.#pendingCandidate === pending) { + this.#pendingCandidate = undefined; + pending.resolve(candidate); + return; + } + if (pending !== undefined) { + await this.#server.discardCandidate(candidate); + return; + } + await this.#server.promote(candidate); + } catch (error) { + if (pending !== undefined && this.#pendingCandidate === pending) { + this.#pendingCandidate = undefined; + pending.reject(error); + return; + } + this.#nitro.logger.error(`[eve:dev] candidate worker failed: ${toErrorMessage(error)}`); + } + } + + #resolveDefaultEntry(): string { + return resolve( + this.#nitro.options.output.dir, + this.#nitro.options.output.serverDir, + "index.mjs", + ); + } +} + +export function toDevelopmentWorkerGeneration(input: { + readonly runtimeAppRoot: string; + readonly snapshotRoot: string; +}): DevelopmentWorkerGeneration { + return { + id: basename(input.snapshotRoot), + runtimeAppRoot: input.runtimeAppRoot, + }; +} + +function readActiveGeneration(appRoot: string): DevelopmentWorkerGeneration | undefined { + const snapshot = readActiveDevelopmentRuntimeArtifactsSnapshot(appRoot); + if (snapshot === undefined) { + return undefined; + } + return toDevelopmentWorkerGeneration(snapshot); +} + +function createPendingCandidate(generation: DevelopmentWorkerGeneration): PendingCandidate { + let rejectPromise: ((error: unknown) => void) | undefined; + let resolvePromise: ((candidate: DevelopmentWorkerCandidate) => void) | undefined; + const promise = new Promise((resolve, reject) => { + rejectPromise = reject; + resolvePromise = resolve; + }); + + return { + generation, + promise, + reject(error) { + rejectPromise?.(error); + }, + resolve(candidate) { + resolvePromise?.(candidate); + }, + }; +} diff --git a/packages/eve/src/internal/nitro/host/prepare-application-host.ts b/packages/eve/src/internal/nitro/host/prepare-application-host.ts index c341cb8c4..75e37ee38 100644 --- a/packages/eve/src/internal/nitro/host/prepare-application-host.ts +++ b/packages/eve/src/internal/nitro/host/prepare-application-host.ts @@ -26,7 +26,10 @@ import { discardDevelopmentGeneration, stageDevelopmentGeneration, } from "#internal/nitro/development-generation.js"; -import type { PreparedApplicationHost } from "#internal/nitro/host/types.js"; +import type { + PreparedApplicationHost, + PreparedDevelopmentApplicationHost, +} from "#internal/nitro/host/types.js"; /** * Compiles one authored app in place and stages the package-owned artifacts @@ -35,7 +38,7 @@ import type { PreparedApplicationHost } from "#internal/nitro/host/types.js"; */ export async function prepareDevelopmentApplicationHost( startPath: string, -): Promise { +): Promise { const compileResult = await compileAgent({ startPath, }); @@ -56,12 +59,15 @@ export async function prepareDevelopmentApplicationHost( appRoot: compileResult.project.appRoot, generation, }); - return createPreparedApplicationHost({ - compileResult, - compiledArtifacts, - schedules, - workflowBuildDir: resolveWorkflowBuildDirectory(compileResult.project.appRoot), - }); + return { + ...createPreparedApplicationHost({ + compileResult, + compiledArtifacts, + schedules, + workflowBuildDir: resolveWorkflowBuildDirectory(compileResult.project.appRoot), + }), + generation, + }; } catch (error) { await discardDevelopmentGeneration(generation); throw error; diff --git a/packages/eve/src/internal/nitro/host/start-development-server.test.ts b/packages/eve/src/internal/nitro/host/start-development-server.test.ts index abc70fb2e..9b3947c32 100644 --- a/packages/eve/src/internal/nitro/host/start-development-server.test.ts +++ b/packages/eve/src/internal/nitro/host/start-development-server.test.ts @@ -1,6 +1,3 @@ -import { EventEmitter } from "node:events"; -import type { IncomingMessage } from "node:http"; -import type { Socket } from "node:net"; import { join } from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -27,9 +24,14 @@ const mocks = vi.hoisted(() => { url: "http://localhost:2000/", }; const devServer = { + buildCandidate: vi.fn(async (input: { trigger: () => Promise }) => { + await input.trigger(); + return {}; + }), close: vi.fn(async () => undefined), listen: vi.fn(() => listenerServer), - upgrade: vi.fn(async (_req: unknown, _socket: unknown, _head: unknown) => undefined), + promote: vi.fn(async () => undefined), + setControlHandler: vi.fn(), }; const files = new Map(); const devHandlers: Nitro["options"]["devHandlers"] = []; @@ -51,7 +53,6 @@ const mocks = vi.hoisted(() => { authoredSourceWatcher, buildNitro: vi.fn(async () => undefined), createDevelopmentApplicationNitro: vi.fn(async () => nitro), - createDevServer: vi.fn(() => devServer), devServer, fetch: vi.fn(async () => new Response(null, { status: 200 })), files, @@ -59,7 +60,16 @@ const mocks = vi.hoisted(() => { listenerServer, mkdir: vi.fn(async () => undefined), nitro, - prepareDevelopmentApplicationHost: vi.fn(async () => ({ appRoot: "/tmp/eve-test" })), + prepareDevelopmentApplicationHost: vi.fn(async () => ({ + appRoot: "/tmp/eve-test", + generation: { + fingerprint: "test", + runtimeAppRoot: "/tmp/eve-test/.eve/dev-runtime/snapshots/test/source/app", + snapshotRoot: "/tmp/eve-test/.eve/dev-runtime/snapshots/test", + snapshotSourceRoot: "/tmp/eve-test/.eve/dev-runtime/snapshots/test/source", + sourceRoot: "/tmp/eve-test", + }, + })), prepareNitro: vi.fn(async () => undefined), readFile: vi.fn(async (path: string) => { if ( @@ -115,10 +125,16 @@ vi.mock("node:fs/promises", () => ({ vi.mock("nitro/builder", () => ({ build: mocks.buildNitro, - createDevServer: mocks.createDevServer, prepare: mocks.prepareNitro, })); +vi.mock("./nitro-development-worker-server.js", () => ({ + NitroDevelopmentWorkerServer: function NitroDevelopmentWorkerServer() { + return mocks.devServer; + }, + toDevelopmentWorkerGeneration: vi.fn((generation) => generation), +})); + vi.mock("./create-application-nitro.js", () => ({ createDevelopmentApplicationNitro: mocks.createDevelopmentApplicationNitro, })); @@ -153,31 +169,6 @@ vi.mock("#execution/sandbox/bindings/local.js", async (importOriginal) => { }; }); -function createRequest(): IncomingMessage { - return { - headers: { - upgrade: "websocket", - }, - method: "GET", - } as IncomingMessage; -} - -function createSocket(): Socket { - const socket = new EventEmitter() as Socket; - Object.defineProperty(socket, "destroyed", { - configurable: true, - value: false, - }); - socket.destroy = vi.fn(() => { - Object.defineProperty(socket, "destroyed", { - configurable: true, - value: true, - }); - return socket; - }); - return socket; -} - const developmentServerStatePath = join("/tmp/eve-test", ".eve", "dev-server-state.v1.json"); function readStateRecord( @@ -244,15 +235,6 @@ async function loadStartDevelopmentServer(): Promise< }; } -async function startServer(): Promise<{ - close(): Promise; - url: string; -}> { - const startDevelopmentServer = await loadStartDevelopmentServer(); - - return await startDevelopmentServer("/tmp/eve-test"); -} - describe("normalizeDevelopmentServerClientUrl", () => { it("rewrites the IPv6 wildcard listen hostname to IPv6 loopback", async () => { const { normalizeDevelopmentServerClientUrl } = await import("./start-development-server.js"); @@ -324,9 +306,6 @@ describe("createDevelopmentServer", () => { delete process.env.PORT; delete process.env.EVE_DEVELOPMENT_SANDBOX_RUN_ID; mocks.files.clear(); - mocks.devServer.upgrade = vi.fn( - async (_req: unknown, _socket: unknown, _head: unknown) => undefined, - ); Object.assign(mocks.nitro.options, { devHandlers: [], experimental: {}, @@ -393,7 +372,6 @@ describe("createDevelopmentServer", () => { expect(mocks.devServer.listen).toHaveBeenCalledWith({ hostname: "127.0.0.1", port: 2000, - silent: true, }); await server.close(); @@ -434,12 +412,10 @@ describe("createDevelopmentServer", () => { expect(mocks.devServer.listen).toHaveBeenNthCalledWith(1, { hostname: "127.0.0.1", port: 2000, - silent: true, }); expect(mocks.devServer.listen).toHaveBeenNthCalledWith(2, { hostname: "127.0.0.1", port: 2001, - silent: true, }); expect(server.url).toBe("http://127.0.0.1:2001/"); @@ -459,27 +435,31 @@ describe("createDevelopmentServer", () => { expect(mocks.files.has(developmentServerStatePath)).toBe(false); }); - async function callDevHandler( - handler: Nitro["options"]["devHandlers"][number]["handler"], - url: string, - ): Promise { - if (typeof handler !== "function") throw new Error("Expected a callable dev handler."); - return await handler({ - node: { req: { url } }, - } as Parameters[0]); + async function callControlHandler(url: string): Promise { + const call = mocks.devServer.setControlHandler.mock.calls.at(-1); + const handler = call?.[0] as ((request: Request) => Promise) | undefined; + if (handler === undefined) throw new Error("Missing runtime rebuild handler."); + return await handler(new Request(url)); } - it("registers a host-owned runtime rebuild handler that forces the live watcher", async () => { + it("serves runtime artifact state from the parent-owned control handler", async () => { const startDevelopmentServer = await loadStartDevelopmentServer(); const server = await startDevelopmentServer("/tmp/eve-test"); - const rebuildHandler = mocks.nitro.options.devHandlers.find( - (handler) => handler.route === "/eve/v1/dev/runtime-artifacts/rebuild", - ); - if (rebuildHandler === undefined) throw new Error("Missing runtime rebuild handler."); + const response = await callControlHandler("http://localhost/eve/v1/dev/runtime-artifacts"); + + if (!(response instanceof Response)) throw new Error("Expected a Response."); + await expect(response.json()).resolves.toEqual({ revision: "/tmp/eve-test" }); + expect(mocks.authoredSourceWatcher.flush).not.toHaveBeenCalled(); + expect(mocks.authoredSourceWatcher.rebuild).not.toHaveBeenCalled(); - const response = await callDevHandler( - rebuildHandler.handler, - "/eve/v1/dev/runtime-artifacts/rebuild?force=1", + await server.close(); + }); + + it("registers a host-owned runtime rebuild handler that forces the live watcher", async () => { + const startDevelopmentServer = await loadStartDevelopmentServer(); + const server = await startDevelopmentServer("/tmp/eve-test"); + const response = await callControlHandler( + "http://localhost/eve/v1/dev/runtime-artifacts/rebuild?force=1", ); expect(mocks.authoredSourceWatcher.rebuild).toHaveBeenCalledOnce(); @@ -493,12 +473,7 @@ describe("createDevelopmentServer", () => { it("registers a host-owned runtime rebuild handler that flushes queued changes", async () => { const startDevelopmentServer = await loadStartDevelopmentServer(); const server = await startDevelopmentServer("/tmp/eve-test"); - const rebuildHandler = mocks.nitro.options.devHandlers.find( - (handler) => handler.route === "/eve/v1/dev/runtime-artifacts/rebuild", - ); - if (rebuildHandler === undefined) throw new Error("Missing runtime rebuild handler."); - - await callDevHandler(rebuildHandler.handler, "/eve/v1/dev/runtime-artifacts/rebuild"); + await callControlHandler("http://localhost/eve/v1/dev/runtime-artifacts/rebuild"); expect(mocks.authoredSourceWatcher.flush).toHaveBeenCalledOnce(); expect(mocks.authoredSourceWatcher.rebuild).not.toHaveBeenCalled(); @@ -807,76 +782,4 @@ describe("createDevelopmentServer", () => { /Invalid PORT environment variable/, ); }); - - it("swallows websocket upgrade rejections from the Nitro dev server", async () => { - const originalUpgrade = vi.fn( - async (_req: unknown, _socket: unknown, _head: unknown): Promise => { - throw new Error("Upstream server did not upgrade the connection"); - }, - ); - Object.assign(mocks.nitro.options.features, { websocket: true }); - mocks.devServer.upgrade = originalUpgrade; - - const server = await startServer(); - - try { - const socket = createSocket(); - await expect( - mocks.devServer.upgrade(createRequest(), socket, Buffer.alloc(0)), - ).resolves.toBeUndefined(); - - expect(originalUpgrade).toHaveBeenCalledTimes(1); - expect(socket.destroy).toHaveBeenCalledTimes(1); - } finally { - await server.close(); - } - }); - - it("rejects websocket upgrades before Nitro proxying when websocket support is disabled", async () => { - const originalUpgrade = vi.fn( - async (_req: unknown, _socket: unknown, _head: unknown): Promise => undefined, - ); - mocks.devServer.upgrade = originalUpgrade; - - const server = await startServer(); - - try { - const socket = createSocket(); - await expect( - mocks.devServer.upgrade(createRequest(), socket, Buffer.alloc(0)), - ).resolves.toBeUndefined(); - - expect(originalUpgrade).not.toHaveBeenCalled(); - expect(socket.destroy).toHaveBeenCalledTimes(1); - } finally { - await server.close(); - } - }); - - it("handles socket errors emitted during websocket upgrade handling", async () => { - const originalUpgrade = vi.fn( - async (_req: unknown, socket: unknown, _head: unknown): Promise => { - const upgradeSocket = socket as Socket; - - upgradeSocket.emit("error", new Error("socket failure")); - throw new Error("socket failure"); - }, - ); - Object.assign(mocks.nitro.options.features, { websocket: true }); - mocks.devServer.upgrade = originalUpgrade; - - const server = await startServer(); - - try { - const socket = createSocket(); - await expect( - mocks.devServer.upgrade(createRequest(), socket, Buffer.alloc(0)), - ).resolves.toBeUndefined(); - - expect(originalUpgrade).toHaveBeenCalledTimes(1); - expect(socket.destroy).toHaveBeenCalledTimes(1); - } finally { - await server.close(); - } - }); }); diff --git a/packages/eve/src/internal/nitro/host/start-development-server.ts b/packages/eve/src/internal/nitro/host/start-development-server.ts index 4d3f84ece..09beb8569 100644 --- a/packages/eve/src/internal/nitro/host/start-development-server.ts +++ b/packages/eve/src/internal/nitro/host/start-development-server.ts @@ -1,16 +1,16 @@ -import type { IncomingMessage } from "node:http"; -import type { Socket } from "node:net"; - import { EVE_DEV_ENV_FLAG } from "#internal/application/optional-package-install.js"; -import { build as buildNitro, createDevServer, prepare } from "nitro/builder"; +import { build as buildNitro, prepare } from "nitro/builder"; import type { Nitro } from "nitro/types"; import { createDevelopmentApplicationNitro } from "#internal/nitro/host/create-application-nitro.js"; import { createDevelopmentNitroArtifactsConfig } from "#internal/nitro/host/artifacts-config.js"; import type { AuthoredSourceWatcherHandle } from "#internal/nitro/host/dev-authored-source-watcher.js"; import { prepareDevelopmentApplicationHost } from "#internal/nitro/host/prepare-application-host.js"; -import { EVE_DEV_RUNTIME_ARTIFACTS_REBUILD_ROUTE_PATH } from "#protocol/routes.js"; +import { + EVE_DEV_RUNTIME_ARTIFACTS_REBUILD_ROUTE_PATH, + EVE_DEV_RUNTIME_ARTIFACTS_ROUTE_PATH, +} from "#protocol/routes.js"; import { resolveDiscoveryProject } from "#discover/project.js"; import { DevelopmentServerState } from "#internal/nitro/host/dev-server-state.js"; import { toErrorMessage } from "#shared/errors.js"; @@ -43,6 +43,10 @@ import { import { detectPackageManager, type PackageManagerKind } from "#setup/package-manager.js"; import { eveDevArguments } from "#setup/primitives/index.js"; import { devBootPhase } from "#internal/dev-boot-progress.js"; +import { + NitroDevelopmentWorkerServer, + toDevelopmentWorkerGeneration, +} from "#internal/nitro/host/nitro-development-worker-server.js"; const MAX_ALLOWED_DEVELOPMENT_SERVER_PORT = 65_535; const WORKFLOW_LOCAL_BASE_URL_ENV = "WORKFLOW_LOCAL_BASE_URL"; @@ -108,8 +112,7 @@ function isAddressInUseError(error: unknown): error is NodeJS.ErrnoException { return error instanceof Error && "code" in error && error.code === "EADDRINUSE"; } -type NitroDevelopmentServer = ReturnType; -type NitroDevelopmentServerUpgrade = NitroDevelopmentServer["upgrade"]; +type NitroDevelopmentServer = NitroDevelopmentWorkerServer; function resolveDevelopmentServerPort(port: number | string | undefined): number { const resolvedPort = @@ -227,73 +230,25 @@ function installWorkflowLocalQueueEnvironment(serverUrl: string): () => void { }; } -function attachTemporarySocketErrorHandler(socket: Socket): () => void { - // Keep early socket failures from becoming uncaught EventEmitter errors - // while Nitro/httpxy installs its own upgrade-path listeners. - const onSocketError = () => {}; - - socket.once("error", onSocketError); - - return () => { - socket.off("error", onSocketError); - }; -} - -function shouldProxyDevelopmentServerWebSocketUpgrades(nitro: Nitro): boolean { - return nitro.options.features.websocket === true || nitro.options.experimental.websocket === true; -} - -function guardDevelopmentServerWebSocketUpgrades( - nitro: Nitro, - devServer: NitroDevelopmentServer, -): void { - const originalUpgrade = devServer.upgrade.bind(devServer) as NitroDevelopmentServerUpgrade; - const websocketEnabled = shouldProxyDevelopmentServerWebSocketUpgrades(nitro); - const guardedUpgrade: NitroDevelopmentServerUpgrade = async ( - req: IncomingMessage, - socket: Socket, - head: unknown, - ) => { - if (!websocketEnabled) { - if (!socket.destroyed) { - socket.destroy(); - } - return; - } - - const removeSocketErrorHandler = attachTemporarySocketErrorHandler(socket); - - try { - await originalUpgrade(req, socket, head); - } catch { - if (!socket.destroyed) { - socket.destroy(); - } - } finally { - removeSocketErrorHandler(); - } - }; - - devServer.upgrade = guardedUpgrade; -} - function addDevelopmentRuntimeArtifactsRebuildHandler(input: { readonly appRoot: string; - readonly nitro: Nitro; + readonly devServer: NitroDevelopmentWorkerServer; readonly watcher: AuthoredSourceWatcherHandle; }): void { - input.nitro.options.devHandlers.push({ - route: EVE_DEV_RUNTIME_ARTIFACTS_REBUILD_ROUTE_PATH, - handler: async (event) => { - const requestUrl = event.node?.req.url ?? ""; - const url = new URL(requestUrl, "http://localhost"); - if (url.searchParams.get("force") === "1") { - await input.watcher.rebuild(); - } else { - await input.watcher.flush(); - } + input.devServer.setControlHandler(async (request) => { + const url = new URL(request.url); + if (url.pathname === EVE_DEV_RUNTIME_ARTIFACTS_ROUTE_PATH && request.method === "GET") { return handleDevRuntimeArtifactsRequest({ appRoot: input.appRoot }); - }, + } + if (url.pathname !== EVE_DEV_RUNTIME_ARTIFACTS_REBUILD_ROUTE_PATH || request.method !== "GET") { + return undefined; + } + if (url.searchParams.get("force") === "1") { + await input.watcher.rebuild(); + } else { + await input.watcher.flush(); + } + return handleDevRuntimeArtifactsRequest({ appRoot: input.appRoot }); }); } @@ -314,12 +269,12 @@ async function closeDevelopmentServerResources(input: { } }; + const devServer = input.devServer; + const listenerClosed = devServer === undefined ? true : await attempt(() => devServer.close()); const authoredSourceWatcher = input.authoredSourceWatcher; if (authoredSourceWatcher !== undefined) { await attempt(() => authoredSourceWatcher.close()); } - const devServer = input.devServer; - const listenerClosed = devServer === undefined ? true : await attempt(() => devServer.close()); const nitro = input.nitro; if (nitro !== undefined) { await attempt(() => nitro.close()); @@ -365,7 +320,7 @@ function createDevelopmentServerStartupCleanupError( async function listenForDevelopmentServer(input: { readonly devServer: NitroDevelopmentServer; - readonly host?: string; + readonly host: string; readonly port: number | string | undefined; readonly retryOnAddressInUse: boolean; }) { @@ -379,7 +334,6 @@ async function listenForDevelopmentServer(input: { const server = input.devServer.listen({ hostname: input.host, port, - silent: true, }); try { @@ -476,9 +430,11 @@ async function startNitroDevelopmentServer( options.onBootProgress, ); nitro = activeNitro; - devServer = createDevServer(activeNitro); + devServer = new NitroDevelopmentWorkerServer({ + appRoot: preparedHost.appRoot, + nitro: activeNitro, + }); const activeDevServer = devServer; - guardDevelopmentServerWebSocketUpgrades(activeNitro, devServer); const hostname = options.host ?? activeNitro.options.devServer.hostname ?? DEFAULT_DEVELOPMENT_SERVER_HOST; const retryOnAddressInUse = requestedPort === undefined; @@ -503,8 +459,14 @@ async function startNitroDevelopmentServer( await devBootPhase( "building dev bundle", async () => { - await prepare(activeNitro); - await buildNitro(activeNitro); + const candidate = await activeDevServer.buildCandidate({ + generation: toDevelopmentWorkerGeneration(preparedHost.generation), + trigger: async () => { + await prepare(activeNitro); + await buildNitro(activeNitro); + }, + }); + await activeDevServer.promote(candidate); }, options.onBootProgress, ); @@ -523,7 +485,7 @@ async function startNitroDevelopmentServer( ); addDevelopmentRuntimeArtifactsRebuildHandler({ appRoot: project.appRoot, - nitro: activeNitro, + devServer: activeDevServer, watcher: authoredSourceWatcher, }); await state.write(serverUrl); diff --git a/packages/eve/src/internal/nitro/host/types.ts b/packages/eve/src/internal/nitro/host/types.ts index 46e0a0df5..5c328a395 100644 --- a/packages/eve/src/internal/nitro/host/types.ts +++ b/packages/eve/src/internal/nitro/host/types.ts @@ -3,6 +3,7 @@ import type { ScheduleRegistration } from "#runtime/schedules/register.js"; import type { ResolvedScheduleDefinition } from "#runtime/types.js"; import type { GeneratedCompiledArtifactsFiles } from "#internal/application/compiled-artifacts.js"; import type { DevBootProgressReporter } from "#internal/dev-boot-progress.js"; +import type { DevelopmentGeneration } from "#internal/nitro/development-generation.js"; /** * Route surface included in one programmatic Nitro host build. @@ -70,3 +71,7 @@ export interface PreparedApplicationHost { schedules: readonly ResolvedScheduleDefinition[]; workflowBuildDir: string; } + +export interface PreparedDevelopmentApplicationHost extends PreparedApplicationHost { + generation: DevelopmentGeneration; +} diff --git a/packages/eve/src/internal/nitro/routes/agent-info/load-agent-info-data.ts b/packages/eve/src/internal/nitro/routes/agent-info/load-agent-info-data.ts index 6ccabcb9b..099658d90 100644 --- a/packages/eve/src/internal/nitro/routes/agent-info/load-agent-info-data.ts +++ b/packages/eve/src/internal/nitro/routes/agent-info/load-agent-info-data.ts @@ -3,6 +3,7 @@ import type { RuntimeCompiledArtifactsSource } from "#runtime/compiled-artifacts import { type NitroArtifactsConfig, resolveNitroCompiledArtifactsSource, + resolveNitroRequestCompiledArtifactsSource, } from "#internal/nitro/routes/runtime-artifacts.js"; import { loadCompiledManifest } from "#runtime/loaders/manifest.js"; import { loadCompiledModuleMap } from "#runtime/loaders/module-map.js"; @@ -70,6 +71,13 @@ export function resolveAgentInfoCompiledArtifactsSource( return resolveNitroCompiledArtifactsSource(input); } +export function resolveAgentInfoRequestCompiledArtifactsSource( + input: NitroArtifactsConfig, + request: Request, +): RuntimeCompiledArtifactsSource { + return resolveNitroRequestCompiledArtifactsSource(input, request); +} + async function loadAgentInfoDataFromArtifacts( compiledArtifactsSource: RuntimeCompiledArtifactsSource, ): Promise { diff --git a/packages/eve/src/internal/nitro/routes/channel-dispatch.test.ts b/packages/eve/src/internal/nitro/routes/channel-dispatch.test.ts index 92782b9c2..617d2d3fb 100644 --- a/packages/eve/src/internal/nitro/routes/channel-dispatch.test.ts +++ b/packages/eve/src/internal/nitro/routes/channel-dispatch.test.ts @@ -2,7 +2,7 @@ import type { H3Event } from "nitro"; import { describe, expect, it, vi } from "vitest"; import { CHANNEL_SENTINEL, type CompiledChannel } from "#channel/compiled-channel.js"; -import type { RouteHandlerArgs } from "#channel/routes.js"; +import type { RouteHandlerArgs, WebSocketPeer } from "#channel/routes.js"; import type { DeliverInput, RunInput, Runtime } from "#channel/types.js"; import { readVercelProjectLink } from "#internal/vercel/project-link.js"; import type { RouteContext } from "#public/definitions/channel.js"; @@ -13,6 +13,7 @@ import { dispatchChannelWebSocketRequest, } from "#internal/nitro/routes/channel-dispatch.js"; import { resolveNitroChannelRuntimeBundle } from "#internal/nitro/routes/runtime-stack.js"; +import { installDevelopmentWorkerRequestMetadata } from "#internal/nitro/host/dev-worker-metadata.js"; vi.mock("#internal/nitro/routes/runtime-stack.js", () => ({ resolveNitroChannelRuntimeBundle: vi.fn(), @@ -50,12 +51,20 @@ function createDeferred() { function createEvent(input?: { readonly headers?: Record; readonly requestIp?: string; + readonly trustedRequestIp?: string | null; readonly waitUntil?: (task: Promise) => void; }): H3Event { const request = new Request("https://eve.test/slack", { headers: input?.headers }); Object.assign(request, { ip: input?.requestIp ?? "127.0.0.1", }); + if (input?.trustedRequestIp !== undefined) { + installDevelopmentWorkerRequestMetadata(request, { + clientAddress: input.trustedRequestIp, + generationId: "generation-a", + runtimeAppRoot: "/app/.eve/dev-runtime/snapshots/generation-a/source/app", + }); + } return { context: { @@ -275,6 +284,40 @@ describe("dispatchChannelRequest", () => { ); }); + it("uses the parent-captured client address instead of the worker socket address", async () => { + let requestIp: string | null | undefined; + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [ + { + handler: async (_request, args) => { + requestIp = args.requestIp; + return new Response("ok"); + }, + fetch: async () => new Response("not used"), + logicalPath: "agent/channels/webhook.ts", + method: "POST", + name: "webhook", + sourceId: "channel-webhook", + sourceKind: "module", + urlPath: "/webhook", + } satisfies ResolvedChannelDefinition, + ], + runtime, + }); + + await dispatchChannelRequest( + createEvent({ + requestIp: "127.0.0.1", + trustedRequestIp: "192.0.2.80", + waitUntil: vi.fn(), + }), + "POST /webhook", + {} as never, + ); + + expect(requestIp).toBe("192.0.2.80"); + }); + it("does not invent a channel request id when Vercel did not send one", async () => { const runtimeForTest: Runtime = { deliver: vi.fn().mockResolvedValue({ sessionId: "sess_route" }), @@ -423,6 +466,61 @@ describe("dispatchChannelRequest", () => { expect(waitUntil).toHaveBeenCalledTimes(1); }); + it("preserves the parent-captured client address on websocket peers", async () => { + let peerAddress: string | undefined; + let transportHeaderVisible: boolean | undefined; + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [ + { + fetch: async () => new Response("not used"), + websocket: async () => ({ + open(peer) { + peerAddress = peer.remoteAddress; + transportHeaderVisible = peer.request.headers.has("x-eve-dev-worker-metadata"); + }, + }), + logicalPath: "agent/channels/voice.ts", + method: "WEBSOCKET", + name: "voice", + sourceId: "channel-voice", + sourceKind: "module", + urlPath: "/voice", + } satisfies ResolvedChannelDefinition, + ], + runtime, + }); + const hooks = await dispatchChannelWebSocketRequest( + createEvent({ + requestIp: "127.0.0.1", + trustedRequestIp: "192.0.2.90", + waitUntil: vi.fn(), + }), + "WEBSOCKET /voice", + {} as never, + ); + const peer = { + close: vi.fn(), + context: {}, + id: "peer-1", + namespace: "voice", + publish: vi.fn(), + remoteAddress: "127.0.0.1", + request: new Request("https://eve.test/voice", { + headers: { "x-eve-dev-worker-metadata": "signed-parent-value" }, + }), + send: vi.fn(), + subscribe: vi.fn(), + terminate: vi.fn(), + topics: new Set(), + unsubscribe: vi.fn(), + } satisfies WebSocketPeer; + + await hooks.open?.(peer); + + expect(peerAddress).toBe("192.0.2.90"); + expect(transportHeaderVisible).toBe(false); + }); + it("rejects websocket upgrades when no websocket channel matches", async () => { mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ channels: [], diff --git a/packages/eve/src/internal/nitro/routes/channel-dispatch.ts b/packages/eve/src/internal/nitro/routes/channel-dispatch.ts index a01e2b290..2c9a5a507 100644 --- a/packages/eve/src/internal/nitro/routes/channel-dispatch.ts +++ b/packages/eve/src/internal/nitro/routes/channel-dispatch.ts @@ -5,7 +5,7 @@ import { toCrossChannelTargets, } from "#channel/cross-channel-receive.js"; import type { DeliverInput, RunInput, Runtime } from "#channel/types.js"; -import type { RouteHandlerArgs, WebSocketRouteHooks } from "#channel/routes.js"; +import type { RouteHandlerArgs, WebSocketPeer, WebSocketRouteHooks } from "#channel/routes.js"; import { createSendFn } from "#channel/send.js"; import { createGetSessionFn } from "#channel/session.js"; import { createLogger, logError } from "#internal/logging.js"; @@ -14,6 +14,10 @@ import type { NitroArtifactsConfig } from "#internal/nitro/routes/runtime-artifa import { resolveNitroChannelRuntimeBundle } from "#internal/nitro/routes/runtime-stack.js"; import { readVercelProjectLink } from "#internal/vercel/project-link.js"; import { withVercelOidcProjectResolver } from "#runtime/governance/auth/vercel-oidc-project.js"; +import { + DEVELOPMENT_WORKER_METADATA_HEADER, + readDevelopmentWorkerRequestMetadata, +} from "#internal/nitro/host/dev-worker-metadata.js"; const log = createLogger("channel.dispatch"); @@ -45,7 +49,7 @@ export async function dispatchChannelRequest( routeKey: string, config: NitroArtifactsConfig, ): Promise { - const bundle = await resolveNitroChannelRuntimeBundle(config); + const bundle = await resolveNitroChannelRuntimeBundle(config, event.req); const matchedChannel = bundle.channels.find( (channel) => `${channel.method.toUpperCase()} ${channel.urlPath}` === routeKey, @@ -100,7 +104,7 @@ export async function dispatchChannelWebSocketRequest( routeKey: string, config: NitroArtifactsConfig, ): Promise { - const bundle = await resolveNitroChannelRuntimeBundle(config); + const bundle = await resolveNitroChannelRuntimeBundle(config, event.req); const matchedChannel = bundle.channels.find( (channel) => `${channel.method.toUpperCase()} ${channel.urlPath}` === routeKey, @@ -123,7 +127,7 @@ export async function dispatchChannelWebSocketRequest( async () => await websocket(event.req, routeArgs.args), ); flushBackgroundTasks(event, routeArgs.backgroundTasks, routeKey, matchedChannel.name); - return hooks; + return protectWebSocketPeerMetadata(hooks, routeArgs.args.requestIp); } catch (error) { const errorId = logError(log, "channel websocket handler threw", error, { routeKey, @@ -198,8 +202,8 @@ function buildRouteArgs( requestIp, }, async () => { - const { handleAgentInfoRequest } = await import("#internal/nitro/routes/info.js"); - return await handleAgentInfoRequest(config); + const { handleAgentInfoRequestForRequest } = await import("#internal/nitro/routes/info.js"); + return await handleAgentInfoRequestForRequest(config, event.req); }, ); @@ -271,6 +275,71 @@ function flushBackgroundTasks( } function extractSocketIp(event: H3Event): string | null { + const trustedAddress = readDevelopmentWorkerRequestMetadata(event.req)?.clientAddress; + if (trustedAddress !== undefined) { + return trustedAddress; + } + const ip = event.req.ip; return typeof ip === "string" && ip.length > 0 ? ip : null; } + +function protectWebSocketPeerMetadata( + hooks: WebSocketRouteHooks, + requestIp: string | null, +): WebSocketRouteHooks { + const peers = new WeakMap(); + const wrap = (peer: WebSocketPeer): WebSocketPeer => { + const existing = peers.get(peer); + if (existing !== undefined) { + return existing; + } + + const requestHeaders = new Headers(peer.request.headers); + requestHeaders.delete(DEVELOPMENT_WORKER_METADATA_HEADER); + const publicRequest = new Proxy(peer.request, { + get(target, property) { + if (property === "headers") { + return requestHeaders; + } + const value = Reflect.get(target, property, target) as unknown; + return typeof value === "function" ? value.bind(target) : value; + }, + }); + const wrapped = new Proxy(peer, { + get(target, property) { + if (property === "remoteAddress") { + return requestIp ?? undefined; + } + if (property === "request") { + return publicRequest; + } + const value = Reflect.get(target, property, target) as unknown; + return typeof value === "function" ? value.bind(target) : value; + }, + }); + peers.set(peer, wrapped); + return wrapped; + }; + const preserved: WebSocketRouteHooks = {}; + if (hooks.close !== undefined) { + const close = hooks.close; + preserved.close = async (peer, details) => await close(wrap(peer), details); + } + if (hooks.error !== undefined) { + const error = hooks.error; + preserved.error = async (peer, cause) => await error(wrap(peer), cause); + } + if (hooks.message !== undefined) { + const message = hooks.message; + preserved.message = async (peer, value) => await message(wrap(peer), value); + } + if (hooks.open !== undefined) { + const open = hooks.open; + preserved.open = async (peer) => await open(wrap(peer)); + } + if (hooks.upgrade !== undefined) { + preserved.upgrade = hooks.upgrade; + } + return preserved; +} diff --git a/packages/eve/src/internal/nitro/routes/dev-schedule-dispatch.test.ts b/packages/eve/src/internal/nitro/routes/dev-schedule-dispatch.test.ts index f7ed24707..7f0e9c845 100644 --- a/packages/eve/src/internal/nitro/routes/dev-schedule-dispatch.test.ts +++ b/packages/eve/src/internal/nitro/routes/dev-schedule-dispatch.test.ts @@ -51,6 +51,7 @@ describe("handleDevScheduleDispatchRequest", () => { }); expect(mocks.dispatchScheduleInDev).toHaveBeenCalledWith({ artifactsConfig: ARTIFACTS_CONFIG, + request: expect.any(Request), scheduleId: "heartbeat", }); }); @@ -66,6 +67,7 @@ describe("handleDevScheduleDispatchRequest", () => { expect(response.status).toBe(200); expect(mocks.dispatchScheduleInDev).toHaveBeenCalledWith({ artifactsConfig: ARTIFACTS_CONFIG, + request: expect.any(Request), scheduleId: "weird/name", }); }); diff --git a/packages/eve/src/internal/nitro/routes/dev-schedule-dispatch.ts b/packages/eve/src/internal/nitro/routes/dev-schedule-dispatch.ts index 3919eceef..74d2849c6 100644 --- a/packages/eve/src/internal/nitro/routes/dev-schedule-dispatch.ts +++ b/packages/eve/src/internal/nitro/routes/dev-schedule-dispatch.ts @@ -54,6 +54,7 @@ export async function handleDevScheduleDispatchRequest( try { const result = await dispatchScheduleInDev({ artifactsConfig: input, + request, scheduleId, }); return Response.json({ diff --git a/packages/eve/src/internal/nitro/routes/info.ts b/packages/eve/src/internal/nitro/routes/info.ts index b657fcc12..06a5389bc 100644 --- a/packages/eve/src/internal/nitro/routes/info.ts +++ b/packages/eve/src/internal/nitro/routes/info.ts @@ -3,14 +3,19 @@ import { buildAgentInfoResponseFromManifest } from "#internal/nitro/routes/agent import { loadAgentInfoManifestData, resolveAgentInfoCompiledArtifactsSource, + resolveAgentInfoRequestCompiledArtifactsSource, } from "#internal/nitro/routes/agent-info/load-agent-info-data.js"; import type { GatewayCredentialPresence } from "#internal/resolve-model-endpoint-status.js"; import type { NitroArtifactsConfig } from "#internal/nitro/routes/runtime-artifacts.js"; +import type { RuntimeCompiledArtifactsSource } from "#runtime/compiled-artifacts-source.js"; import type { ModelRouting } from "#shared/agent-definition.js"; -async function createAgentInfoPayload(input: NitroArtifactsConfig) { +async function createAgentInfoPayload( + input: NitroArtifactsConfig, + compiledArtifactsSource: RuntimeCompiledArtifactsSource, +) { const data = await loadAgentInfoManifestData({ - compiledArtifactsSource: resolveAgentInfoCompiledArtifactsSource(input), + compiledArtifactsSource, }); return buildAgentInfoResponseFromManifest(data, { @@ -19,6 +24,21 @@ async function createAgentInfoPayload(input: NitroArtifactsConfig) { }); } +async function createAgentInfoResponse( + input: NitroArtifactsConfig, + compiledArtifactsSource: RuntimeCompiledArtifactsSource, +): Promise { + return new Response( + JSON.stringify(await createAgentInfoPayload(input, compiledArtifactsSource)), + { + headers: { + "cache-control": "no-store", + "content-type": "application/json; charset=utf-8", + }, + }, + ); +} + function hasEnvValue(value: string | undefined): boolean { return value !== undefined && value.trim() !== ""; } @@ -49,10 +69,15 @@ async function resolveGatewayCredentialPresence( * Builds the package-owned JSON inspection response for the current agent. */ export async function handleAgentInfoRequest(input: NitroArtifactsConfig): Promise { - return new Response(JSON.stringify(await createAgentInfoPayload(input)), { - headers: { - "cache-control": "no-store", - "content-type": "application/json; charset=utf-8", - }, - }); + return await createAgentInfoResponse(input, resolveAgentInfoCompiledArtifactsSource(input)); +} + +export async function handleAgentInfoRequestForRequest( + input: NitroArtifactsConfig, + request: Request, +): Promise { + return await createAgentInfoResponse( + input, + resolveAgentInfoRequestCompiledArtifactsSource(input, request), + ); } diff --git a/packages/eve/src/internal/nitro/routes/runtime-artifacts.test.ts b/packages/eve/src/internal/nitro/routes/runtime-artifacts.test.ts index ba594dea4..e0ac78406 100644 --- a/packages/eve/src/internal/nitro/routes/runtime-artifacts.test.ts +++ b/packages/eve/src/internal/nitro/routes/runtime-artifacts.test.ts @@ -3,7 +3,11 @@ import { describe, expect, it } from "vitest"; import { COMPILED_AGENT_MANIFEST_VERSION } from "#compiler/manifest.js"; import { installBundledCompiledArtifacts } from "#runtime/loaders/bundled-artifacts.js"; import { createRuntimeSession, withRuntimeSession } from "#runtime/sessions/runtime-session.js"; -import { resolveNitroCompiledArtifactsSource } from "#internal/nitro/routes/runtime-artifacts.js"; +import { + resolveNitroCompiledArtifactsSource, + resolveNitroRequestCompiledArtifactsSource, +} from "#internal/nitro/routes/runtime-artifacts.js"; +import { installDevelopmentWorkerRequestMetadata } from "#internal/nitro/host/dev-worker-metadata.js"; /** * Installs an empty compiled-artifact snapshot on the currently active runtime @@ -84,6 +88,45 @@ describe("resolveNitroCompiledArtifactsSource", () => { }); }); + it("resolves a development request from its admitted generation", () => { + const request = new Request("http://worker.test/"); + installDevelopmentWorkerRequestMetadata(request, { + clientAddress: "192.0.2.10", + generationId: "generation-a", + runtimeAppRoot: "/tmp/dev-app/.eve/dev-runtime/snapshots/generation-a/source/app", + }); + + expect( + resolveNitroRequestCompiledArtifactsSource( + { + appRoot: "/tmp/dev-app", + devRuntimeArtifactsPointerPath: "/tmp/dev-app/.eve/dev-runtime/current.json", + kind: "development", + moduleMapLoaderPath: "/package/src/internal/authored-module-map-loader.ts", + }, + request, + ), + ).toMatchObject({ + appRoot: "/tmp/dev-app/.eve/dev-runtime/snapshots/generation-a/source/app", + kind: "disk", + sandboxAppRoot: "/tmp/dev-app", + }); + }); + + it("rejects a development request without an admitted generation", () => { + expect(() => + resolveNitroRequestCompiledArtifactsSource( + { + appRoot: "/tmp/dev-app", + devRuntimeArtifactsPointerPath: "/tmp/dev-app/.eve/dev-runtime/current.json", + kind: "development", + moduleMapLoaderPath: "/package/src/internal/authored-module-map-loader.ts", + }, + new Request("http://worker.test/"), + ), + ).toThrow("missing its admitted artifact generation"); + }); + it("uses bundled artifacts outside development mode when they exist", async () => { await withScopedRuntimeSession(() => { installEmptyBundledArtifacts(); diff --git a/packages/eve/src/internal/nitro/routes/runtime-artifacts.ts b/packages/eve/src/internal/nitro/routes/runtime-artifacts.ts index a4e65679f..baa6bab95 100644 --- a/packages/eve/src/internal/nitro/routes/runtime-artifacts.ts +++ b/packages/eve/src/internal/nitro/routes/runtime-artifacts.ts @@ -5,6 +5,7 @@ import { } from "#runtime/compiled-artifacts-source.js"; import { readBundledCompiledArtifacts } from "#runtime/loaders/bundled-artifacts.js"; import { readDevelopmentRuntimeArtifactsSnapshotRoot } from "#internal/nitro/dev-runtime-artifacts.js"; +import { readDevelopmentWorkerRequestMetadata } from "#internal/nitro/host/dev-worker-metadata.js"; /** * Configuration values needed to resolve the compiled-artifact source for @@ -48,3 +49,23 @@ export function resolveNitroCompiledArtifactsSource( throw new Error("eve Nitro production requires bundled artifacts."); } + +/** Resolves the immutable artifact generation stamped on one admitted request. */ +export function resolveNitroRequestCompiledArtifactsSource( + config: NitroArtifactsConfig, + request: Request, +): RuntimeCompiledArtifactsSource { + if (config.kind === "production") { + return resolveNitroCompiledArtifactsSource(config); + } + + const metadata = readDevelopmentWorkerRequestMetadata(request); + if (metadata === undefined) { + throw new Error("Development request is missing its admitted artifact generation."); + } + + return createDiskRuntimeCompiledArtifactsSource(metadata.runtimeAppRoot, { + moduleMapLoaderPath: config.moduleMapLoaderPath, + sandboxAppRoot: config.appRoot, + }); +} diff --git a/packages/eve/src/internal/nitro/routes/runtime-stack.ts b/packages/eve/src/internal/nitro/routes/runtime-stack.ts index f655325c8..1910797b8 100644 --- a/packages/eve/src/internal/nitro/routes/runtime-stack.ts +++ b/packages/eve/src/internal/nitro/routes/runtime-stack.ts @@ -4,7 +4,7 @@ import { getCompiledRuntimeAgentBundle } from "#runtime/sessions/compiled-agent- import type { ResolvedChannelDefinition } from "#runtime/types.js"; import { type NitroArtifactsConfig, - resolveNitroCompiledArtifactsSource, + resolveNitroRequestCompiledArtifactsSource, } from "#internal/nitro/routes/runtime-artifacts.js"; /** @@ -32,8 +32,9 @@ export interface NitroChannelRuntimeBundle { */ export async function resolveNitroChannelRuntimeBundle( config: NitroArtifactsConfig, + request: Request, ): Promise { - const compiledArtifactsSource = resolveNitroCompiledArtifactsSource(config); + const compiledArtifactsSource = resolveNitroRequestCompiledArtifactsSource(config, request); const bundle = await getCompiledRuntimeAgentBundle({ compiledArtifactsSource, }); diff --git a/packages/eve/src/internal/nitro/routes/schedule-task.ts b/packages/eve/src/internal/nitro/routes/schedule-task.ts index a0958d235..cda3e030a 100644 --- a/packages/eve/src/internal/nitro/routes/schedule-task.ts +++ b/packages/eve/src/internal/nitro/routes/schedule-task.ts @@ -5,6 +5,7 @@ import { loadResolvedCompiledScheduleByTaskName } from "#runtime/schedules/resol import { getCompiledRuntimeAgentBundle } from "#runtime/sessions/compiled-agent-cache.js"; import type { NitroArtifactsConfig } from "#internal/nitro/routes/runtime-artifacts.js"; import { resolveNitroCompiledArtifactsSource } from "#internal/nitro/routes/runtime-artifacts.js"; +import type { RuntimeCompiledArtifactsSource } from "#runtime/compiled-artifacts-source.js"; /** * Dispatches one eve authored schedule via the execution engine. @@ -17,7 +18,16 @@ export async function dispatchScheduleTask( taskName: string, config: NitroArtifactsConfig, ): Promise<{ scheduleId: string; sessionIds: readonly string[] }> { - const compiledArtifactsSource = resolveNitroCompiledArtifactsSource(config); + return await dispatchScheduleTaskFromArtifacts( + taskName, + resolveNitroCompiledArtifactsSource(config), + ); +} + +export async function dispatchScheduleTaskFromArtifacts( + taskName: string, + compiledArtifactsSource: RuntimeCompiledArtifactsSource, +): Promise<{ scheduleId: string; sessionIds: readonly string[] }> { const schedule = await loadResolvedCompiledScheduleByTaskName(taskName, { compiledArtifactsSource, }); diff --git a/packages/eve/test/scenarios/agent-info-route.scenario.test.ts b/packages/eve/test/scenarios/agent-info-route.scenario.test.ts index 1e092861f..c92179d11 100644 --- a/packages/eve/test/scenarios/agent-info-route.scenario.test.ts +++ b/packages/eve/test/scenarios/agent-info-route.scenario.test.ts @@ -6,6 +6,7 @@ import { describe, expect, it } from "vitest"; import { compileAgent } from "../../src/compiler/compile-agent.js"; import { createDevelopmentNitroArtifactsConfig } from "../../src/internal/nitro/host/artifacts-config.js"; +import { installDevelopmentWorkerRequestMetadata } from "../../src/internal/nitro/host/dev-worker-metadata.js"; import type { AgentInfoResponse } from "../../src/internal/nitro/routes/agent-info/build-agent-info-response.js"; import { dispatchChannelRequest } from "../../src/internal/nitro/routes/channel-dispatch.js"; import { EVE_CREATE_SESSION_ROUTE_PATH, EVE_INFO_ROUTE_PATH } from "../../src/protocol/routes.js"; @@ -70,8 +71,14 @@ function createInfoEvent(request: Request): H3Event { } async function requestAgentInfo(appRoot: string, request: Request): Promise { + const admittedRequest = new Request(request); + installDevelopmentWorkerRequestMetadata(admittedRequest, { + clientAddress: "127.0.0.1", + generationId: "scenario-generation", + runtimeAppRoot: appRoot, + }); return await dispatchChannelRequest( - createInfoEvent(request), + createInfoEvent(admittedRequest), INFO_ROUTE_KEY, createDevelopmentNitroArtifactsConfig({ appRoot }), ); diff --git a/packages/eve/test/scenarios/dev-server.scenario.test.ts b/packages/eve/test/scenarios/dev-server.scenario.test.ts index be149c9fe..25c470be1 100644 --- a/packages/eve/test/scenarios/dev-server.scenario.test.ts +++ b/packages/eve/test/scenarios/dev-server.scenario.test.ts @@ -28,11 +28,40 @@ const scenarioApp = useScenarioApp(); const DEV_SERVER_SCENARIO_TIMEOUT_MS = 360_000; const DEV_SERVER_AGENT_DESCRIPTOR: ScenarioAppDescriptor = { ...WEATHER_AGENT_DESCRIPTOR, - files: Object.fromEntries( - Object.entries(WEATHER_AGENT_DESCRIPTOR.files).filter( - ([path]) => !path.startsWith("agent/channels/"), + files: { + ...Object.fromEntries( + Object.entries(WEATHER_AGENT_DESCRIPTOR.files).filter( + ([path]) => !path.startsWith("agent/channels/"), + ), ), - ), + "agent/channels/dev-generation.ts": [ + 'import { defineChannel, GET } from "eve/channels";', + "", + "export default defineChannel({", + ' routes: [GET("/dev-generation", () => new Response(process.env.EVE_SCENARIO_RELOAD ?? process.env.EVE_WEBSOCKET_RELOAD ?? "initial"))],', + "});", + "", + ].join("\n"), + }, +}; +const WEBSOCKET_DEV_SERVER_DESCRIPTOR: ScenarioAppDescriptor = { + ...DEV_SERVER_AGENT_DESCRIPTOR, + files: { + ...DEV_SERVER_AGENT_DESCRIPTOR.files, + "agent/channels/socket.ts": [ + 'import { defineChannel, WS } from "eve/channels";', + "", + "export default defineChannel({", + ' routes: [WS("/socket", () => ({', + " message(peer, message) {", + ' const transportHeader = peer.request.headers.has("x-eve-dev-worker-metadata") ? "exposed" : "hidden";', + ' peer.send(`${transportHeader}:${peer.remoteAddress ?? "missing"}:${message.text()}`);', + " },", + " }))],", + "});", + "", + ].join("\n"), + }, }; interface RunningEveDev { @@ -63,10 +92,12 @@ function hasUnsupportedWindowsEsmImport(text: string): boolean { ); } -function hasKnownDevBundlingFailure(text: string): boolean { +function hasKnownDevServerFailure(text: string): boolean { return ( hasUnsupportedWindowsEsmImport(text) || text.includes("UNRESOLVED_IMPORT") || + text.includes("ECONNRESET") || + text.includes("socket hang up") || (text.includes("ERR_MODULE_NOT_FOUND") && text.includes("authored-module-map-loader")) ); } @@ -84,13 +115,13 @@ async function wait(ms: number): Promise { } async function waitForCondition( - condition: () => boolean, + condition: () => boolean | Promise, failureMessage: string, timeoutMs: number = 60_000, ): Promise { const deadline = Date.now() + timeoutMs; - while (!condition()) { + while (!(await condition())) { if (Date.now() >= deadline) { throw new Error(failureMessage); } @@ -152,11 +183,11 @@ async function waitForServerUrl(input: { const output = input.getOutput(); const combinedOutput = `${output.stdout}\n${output.stderr}`; - if (hasKnownDevBundlingFailure(combinedOutput)) { + if (hasKnownDevServerFailure(combinedOutput)) { settleReject( new Error( [ - "eve dev emitted a known generated dev bundle import failure.", + "eve dev emitted a known reload or generated-bundle failure.", `stdout:\n${output.stdout}`, `stderr:\n${output.stderr}`, ].join("\n\n"), @@ -194,6 +225,46 @@ async function waitForServerUrl(input: { }); } +async function waitForWebSocketOpen(socket: WebSocket): Promise { + await waitForWebSocketEvent(socket, "open", () => undefined); +} + +async function waitForWebSocketMessage(socket: WebSocket): Promise { + return await waitForWebSocketEvent(socket, "message", (event) => String(event.data)); +} + +async function waitForWebSocketClose(socket: WebSocket): Promise { + await waitForWebSocketEvent(socket, "close", () => undefined); +} + +async function waitForWebSocketEvent( + socket: WebSocket, + eventName: "close" | "message" | "open", + select: (event: MessageEvent) => T, +): Promise { + return await new Promise((resolve, reject) => { + const deadline = setTimeout(() => { + cleanup(); + reject(new Error(`Timed out waiting for WebSocket ${eventName}.`)); + }, 10_000); + const cleanup = () => { + clearTimeout(deadline); + socket.removeEventListener(eventName, onEvent as EventListener); + socket.removeEventListener("error", onError); + }; + const onEvent = (event: Event) => { + cleanup(); + resolve(select(event as MessageEvent)); + }; + const onError = () => { + cleanup(); + reject(new Error(`WebSocket failed while waiting for ${eventName}.`)); + }; + socket.addEventListener(eventName, onEvent as EventListener, { once: true }); + socket.addEventListener("error", onError, { once: true }); + }); +} + async function startEveDev(appRoot: string): Promise { const eveBinPath = join(appRoot, "node_modules", "eve", "bin", "eve.js"); const child = spawn( @@ -256,6 +327,42 @@ async function startEveDev(appRoot: string): Promise { } describe("eve dev server", () => { + it( + "keeps an admitted websocket on its worker while a ready replacement is promoted", + async () => { + const app = await scenarioApp(WEBSOCKET_DEV_SERVER_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + const socketUrl = new URL("/socket", server.url); + socketUrl.protocol = "ws:"; + const socket = new WebSocket(socketUrl); + + try { + await waitForWebSocketOpen(socket); + const firstMessage = waitForWebSocketMessage(socket); + socket.send("before"); + await expect(firstMessage).resolves.toBe("hidden:127.0.0.1:before"); + + await writeFile(join(app.appRoot, ".env.local"), "EVE_WEBSOCKET_RELOAD=1\n"); + await waitForCondition(async () => { + const generation = await fetch(new URL("/dev-generation", server.url)); + return (await generation.text()) === "1"; + }, `Timed out waiting for websocket worker replacement.\n\nstdout:\n${server.stdout()}\n\nstderr:\n${server.stderr()}`); + + const nextMessage = waitForWebSocketMessage(socket); + socket.send("after"); + await expect(nextMessage).resolves.toBe("hidden:127.0.0.1:after"); + } finally { + if (socket.readyState === WebSocket.OPEN) { + const closed = waitForWebSocketClose(socket); + socket.close(); + await closed; + } + await server.stop(); + } + }, + DEV_SERVER_SCENARIO_TIMEOUT_MS, + ); + it( "rebuilds after its startup runtime generation is force-pruned and completes a streamed turn", async () => { @@ -312,8 +419,10 @@ describe("eve dev server", () => { const currentRuntimeRoot = readDevelopmentRuntimeArtifactsSnapshotRoot(pointerPath); return currentRuntimeRoot !== undefined && currentRuntimeRoot !== authoredRuntimeRoot; }, `Timed out waiting for the structural reload snapshot.\n\nstdout:\n${server.stdout()}\n\nstderr:\n${server.stderr()}`); - // Nitro exposes no replacement-ready signal while it owns the public listener. - await wait(2_000); + await waitForCondition(async () => { + const generation = await fetch(new URL("/dev-generation", server.url)); + return (await generation.text()) === "1"; + }, `Timed out waiting for a ready replacement worker.\n\nstdout:\n${server.stdout()}\n\nstderr:\n${server.stderr()}`); let messageResult: Awaited>; try { messageResult = await sendDevelopmentMessage({ @@ -342,7 +451,7 @@ describe("eve dev server", () => { ].join("\n\n"), ).toBe(true); const output = `${server.stdout()}\n${server.stderr()}`; - expect(hasKnownDevBundlingFailure(output)).toBe(false); + expect(hasKnownDevServerFailure(output)).toBe(false); } finally { await server.stop(); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a1d962c99..3f669b995 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1087,6 +1087,9 @@ importers: emulate: specifier: 0.6.0 version: 0.6.0 + env-runner: + specifier: 0.1.12 + version: 0.1.12 eventsource-parser: specifier: 3.0.8 version: 3.0.8