feat(dev): isolate requests in parent-owned workers

.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.

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",

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<string, unknown>;
+  readonly hooks?: WorkerHooks;
+  readonly name: string;
+  readonly workerEntry: string;
+}
+
+export class BaseEnvRunner {
+  readonly closed: boolean;
+  readonly ready: boolean;
+  protected readonly _data: Record<string, unknown> | undefined;
+  protected readonly _name: string;
+  protected readonly _workerEntry: string;
+
+  constructor(options: BaseEnvRunnerOptions);
+
+  close(cause?: unknown): Promise<void>;
+  fetch(input: string | URL | Request, init?: RequestInit): Promise<Response>;
+  sendMessage(message: unknown): void;
+  upgrade(context: {
+    node: { head: Uint8Array; req: IncomingMessage; socket: Socket };
+  }): Promise<void>;
+  waitForReady(timeout?: number): Promise<void>;
+
+  protected _closeRuntime(): Promise<void>;
+  protected _handleMessage(message: unknown): void;
+  protected _hasRuntime(): boolean;
+  protected _initWithVirtualData(initialize: () => void): void;
+  protected _runtimeType(): string;
+}

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",
+    },
+  ],
+};

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,

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.
  */

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 () => {

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

packages/eve/src/internal/nitro/host/create-application-nitro.ts

@@ -753,6 +753,10 @@ export async function createDevelopmentApplicationNitro(
 ): Promise<Nitro> {
   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,

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<Uint8Array>) : 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<void> {
+  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<void>((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<void> {
+  if (!server.listening) {
+    return Promise.resolve();
+  }
+
+  const closed = new Promise<void>((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<void> {
+  await new Promise<void>((resolve, reject) => {
+    response.once("error", reject);
+    response.end(resolve);
+  });
+}