From 5ff8c02708079cbd348c78cdb43a987df1229f4b Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 4 Nov 2026 21:18:17 +0000 Subject: [PATCH] feat(chord): workers start from the base code the server booted with createWorkers bundles the worker entry once (esbuild, Node builtins and installed packages external) and forks every worker from that snapshot, so an edit of the app's source while the server runs reaches workers only through a restart. The kit watches the bundle's inputs as base code. p2 creates its session workers with SessionWorkers.create and keeps the worker base's inputs as a dev restart baseline. --- apps/p2/src/hosts/server.ts | 14 +- apps/p2/src/hosts/session-workers.ts | 29 ++- packages/chord/CHANGELOG.md | 1 + packages/chord/docs/design.md | 41 +++- packages/chord/docs/http-workers/book.md | 2 +- packages/chord/docs/http-workers/index.html | 2 +- packages/chord/src/node/app/app.ts | 32 ++- packages/chord/src/node/surfaces/bundle.ts | 10 +- packages/chord/src/node/workers/README.md | 47 +++-- packages/chord/src/node/workers/base.ts | 196 ++++++++++++++++++ packages/chord/src/node/workers/workers.ts | 39 +++- packages/chord/test/bench/workers.ts | 2 +- .../chord/test/examples/41-workers/main.ts | 2 +- .../test/examples/42-worker-crash/main.ts | 2 +- .../chord/test/examples/43-rollout/main.ts | 2 +- packages/chord/test/node/app/app.test.ts | 30 ++- packages/chord/test/node/workers/base.test.ts | 179 ++++++++++++++++ packages/chord/test/node/workers/support.ts | 8 +- .../chord/test/node/workers/workers.test.ts | 63 +++++- 19 files changed, 627 insertions(+), 74 deletions(-) create mode 100644 packages/chord/src/node/workers/base.ts create mode 100644 packages/chord/test/node/workers/base.test.ts diff --git a/apps/p2/src/hosts/server.ts b/apps/p2/src/hosts/server.ts index 165bb3197..bb02da159 100644 --- a/apps/p2/src/hosts/server.ts +++ b/apps/p2/src/hosts/server.ts @@ -1,6 +1,6 @@ // Server host: HTTP + WebSocket accept, routes `session:` hosts to session workers, loads the server facets. import { randomUUID } from "node:crypto"; -import { mkdirSync } from "node:fs"; +import { mkdirSync, readFileSync } from "node:fs"; import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; import { homedir } from "node:os"; import { join } from "node:path"; @@ -42,7 +42,7 @@ import { } from "./bundles.ts"; import { devControl } from "./dev.ts"; import { readDevCommand, sendDevCommand, serveDevChannel } from "./dev-channel.ts"; -import { LoadedModules, loadedModuleHashes, loadedPluginFiles } from "./dev-modules.ts"; +import { contentHash, LoadedModules, loadedModuleHashes, loadedPluginFiles } from "./dev-modules.ts"; import { createP2Plugins, listPlugins, PLUGINS_DIRECTORY, SERVER_PACKAGES, selectPlugins } from "./plugins.ts"; import { contentSecurityPolicy, @@ -174,7 +174,7 @@ export async function startServer(options: ServerOptions = {}): Promise !file.startsWith(PLUGINS_DIRECTORY)) + .map((file) => ({ file, hash: contentHash(readFileSync(file)) })), + ); // Capture before plugin resources open (e.g. provider SDK startup), not only after the logs facet activates. const stopLogs = captureProcessLogs("server"); const serveWeb = options.web ?? true; diff --git a/apps/p2/src/hosts/session-workers.ts b/apps/p2/src/hosts/session-workers.ts index adc315811..375a7fa57 100644 --- a/apps/p2/src/hosts/session-workers.ts +++ b/apps/p2/src/hosts/session-workers.ts @@ -37,7 +37,8 @@ const CRASH_RESTARTS = 5; const CRASH_WINDOW_MS = 60_000; export class SessionWorkers implements SessionWorkerControl { - readonly workers: Workers; + /** Chord's workers component, made by `create` (its worker base bundled first). */ + #workers: Workers | undefined; readonly #options: SessionWorkersOptions; readonly #generations = new Map(); readonly #operations = new Map>(); @@ -53,14 +54,35 @@ export class SessionWorkers implements SessionWorkerControl { #states: Readonly> = {}; #generation = 0; - constructor(options: SessionWorkersOptions, context: Context) { + private constructor(options: SessionWorkersOptions) { this.#options = options; processLogs.redactor.addDirectory(options.dataDir); for (const [key, value] of Object.entries(options.env ?? {})) { if (/(?:key|token|secret|password|cookie|auth)/i.test(key)) processLogs.redactor.addSecret(value); } + } + + /** + * The session workers: Chord's workers component created (it bundles the worker entry once, so every worker runs the + * code the server booted with; a broken entry rejects), then followed. + */ + static async create(options: SessionWorkersOptions, context: Context): Promise { + const made = new SessionWorkers(options); + made.#workers = await made.#createWorkers(context); + void made.#follow(); + return made; + } + + /** Chord's workers component for the sessions. */ + get workers(): Workers { + // Set by `create` before the instance is handed out. + return this.#workers!; + } + + #createWorkers(context: Context): Promise> { + const options = this.#options; const dataDir = realpathSync(options.dataDir); - this.workers = createWorkers( + return createWorkers( options.host, Session, { @@ -99,7 +121,6 @@ export class SessionWorkers implements SessionWorkerControl { }, context, ); - void this.#follow(); } /** Generation of the session's latest worker (0: never started). */ diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 8ed3adde0..f3419f729 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -7,6 +7,7 @@ - Chord v2 is the package's root: `@earendil-works/chord` exports v2 (was `@earendil-works/chord/next`). Chord v1 moved unchanged to `@earendil-works/chord/old` (was the root) and `@earendil-works/chord/old/node` (was `/node`); it stays until p2 runs on v2. The foundations v1's root re-exported have their own subpaths: `@earendil-works/chord/json` (`JsonValue`, `JsonRepresentation`, `copyJson`, `isJsonValue`) and `@earendil-works/chord/watch` (`Watch`, `NativeWatch` and the watch types), beside `/context` and `/delta`. Migration: imports of `@earendil-works/chord/next` drop `/next`; v1's imports of `@earendil-works/chord` become `@earendil-works/chord/old`, and `/node` becomes `/old/node`; imports of `Context`, `Draft`, `JsonValue`, `copyJson` or the watch types from the root move to `/context`, `/delta`, `/json` and `/watch`. In the repository, v2's `src/next` is now `src`, `src/next-node` is `src/node`, `src/next-web`, `src/next-react` and `src/next-tui` are `src/web`, `src/react` and `src/tui`, v1 is in `src/old`, and tests and docs moved likewise (`test/next` to `test`, `docs/next` to `docs`, v1's to `test/old` and `docs/old`). - Chord v1 (`@earendil-works/chord/old`, `@earendil-works/chord/old/node`) removed, with its tests and docs (`src/old`, `test/old`, `docs/old`). Apps still on v1 pin `@earendil-works/chord@1.5.0`. - Chord v2's `startWorker` no longer takes `telemetry` (it was added unreleased in slice 9): with a telemetry store on the server, the kit's workers send their spans and metrics to it, and no worker opens the store's file. `startServer`'s `telemetry` is a store with `ingest` (`createSqliteTelemetry`'s). Migration: drop `telemetry` from `startWorker` and the store's path from the worker's environment. +- Chord v2's `createWorkers` returns a promise and rejects when the worker base does not build (decision "Workers start from the base code the server booted with"): it bundles the worker kind's `entry` and the base code it imports once with esbuild (`--conditions` from `execArgv`; Node builtins and packages resolved under `node_modules` external, by their files; workspace packages linked into `node_modules` inlined; `import.meta.url` and `__dirname` naming each source file; an inline source map, forked with `--enable-source-maps`) into a private temporary directory removed at `close`, and forks every worker from that snapshot: an edit of the app's source while the server runs (an agent half way through a change) no longer reaches workers forked later, which kept failing to start until the edit was fixed; a server restart applies it. `workers.inputs` lists the files the base was built from; the app kit's `startServer` builds each worker kind's base before loading (a broken one rejects the start, everything closed) and marks `base` when one of those files changes (it marked only the entry). ### Added diff --git a/packages/chord/docs/design.md b/packages/chord/docs/design.md index fcfa15886..e0d4f6bd3 100644 --- a/packages/chord/docs/design.md +++ b/packages/chord/docs/design.md @@ -2056,7 +2056,7 @@ worker (the reload suite), not only by unit tests. "The page" is any client over | a contract change the new code still `accepts` | the old page keeps working through the mapping | answered through the mapping | unchanged | | a contract change not accepted, or a type removed | the reference waits with its last values (`failed { waiting }`) until the page's code updates or the type returns | calls fail at once with the cause (`contract_mismatch`, `unavailable` not served; §7) | unchanged | | a web or terminal plugin | the other plugins untouched, scroll kept; the reloaded plugin's components remount | the reloaded plugin's calls as above | component state starts over; `usePageState`/`scope.pageState` survive | -| base code (the app's client entry, Chord, React) | applied by restarting the server (the kit builds base bundles at start; no reload applies them); the page reconnects to a new base and reloads (the terminal restarts its sandbox): the only case losing what is on screen | calls in flight end with the link | page state lost; in-memory server and worker state lost; stores unchanged | +| base code (the app's client and worker entries, Chord, React) | applied by restarting the server (the kit builds base bundles at start, the workers component its worker base when it is created; no reload applies them, and a worker forked meanwhile runs the server's boot code); the page reconnects to a new base and reloads (the terminal restarts its sandbox): the only case losing what is on screen | calls in flight end with the link | page state lost; in-memory server and worker state lost; stores unchanged | Order: the server, then every running worker by the same rules, then the clients' scripts. @@ -2374,17 +2374,36 @@ route's checks, never instead of them. ### Workers (`@earendil-works/chord/workers`) -`createWorkers(host, Kind, { kind, entry, execArgv?, env?, authorize, principal, start, lock, idleMs?, plugins, -onOutput?, onError }, ctx)` routes `Kind` itself (`host.route`, keeping its `RouteHandle`): a forked worker per -address over IPC, the link made at the fork, a start message `{ address, start, scripts, lock }` with the app's start -data and the kind's complete current scripts (`plugins.scripts(kind)`); the 60 s start timer covers fork, lock and -connect only (Q4). `workers.state` (a JSON watchable per address key), `stop`, `retry`, `step(name)`, `close`; -`workerMain(fn, { hostPackages, redact, onError })` is the child's side (`fn({ address, start, parent, plugins, -onClose }, ctx)`, `plugins` a `WorkerPlugins` evaluating the shipped scripts, closers in reverse order). One live -worker per address (`processLock`: an abstract socket on Linux, a named pipe on Windows, a socket file elsewhere that -fails closed when stale, the stale file cleared by the component before a start and after a child's exit by a signal, -decision "Stale lock files are cleared by the starter"; a start waits for a draining old holder without polling). +`await createWorkers(host, Kind, { kind, entry, execArgv?, env?, authorize, principal, start, lock, idleMs?, plugins, +onOutput?, onError }, ctx)` bundles the kind's worker base, then routes `Kind` itself (`host.route`, keeping its +`RouteHandle`): a forked worker per address over IPC, the link made at the fork, a start message `{ address, start, +scripts, lock }` with the app's start data and the kind's complete current scripts (`plugins.scripts(kind)`); the 60 s +start timer covers fork, lock and connect only (Q4). `workers.state` (a JSON watchable per address key), `inputs`, +`stop`, `retry`, `step(name)`, `close`; `workerMain(fn, { hostPackages, redact, onError })` is the child's side +(`fn({ address, start, parent, plugins, onClose }, ctx)`, `plugins` a `WorkerPlugins` evaluating the shipped scripts, +closers in reverse order). One live worker per address (`processLock`: an abstract socket on Linux, a named pipe on +Windows, a socket file elsewhere that fails closed when stale, the stale file cleared by the component before a start +and after a child's exit by a signal, decision "Stale lock files are cleared by the starter"; a start waits for a +draining old holder without polling). +- **The worker base** (decision "Workers start from the base code the server booted with"): `entry` and the base code + it imports are bundled once by esbuild when the component is created (`workers.inputs`: the files read), as the + surfaces build their base bundles at start, and every worker of the kind is forked from that snapshot: an edit of the + app's source while the server runs (an agent half way through a change) reaches no worker; a server restart applies + it (the kit gives the inputs to the watcher's `base` mark). A base that does not build rejects `createWorkers` (the + server does not start with it). What it holds: the app's files and workspace packages linked into `node_modules` (by + their real paths); external, loaded by Node at each fork from where the build resolved them: Node builtins and every + package resolved under a `node_modules` directory (native addons and packages reading their own files work as + installed; they change only through an install, which needs a restart anyway). Resolution as the child's Node does it + (its `execArgv` `--conditions`, `main` only, `tsconfig.json` paths as esbuild reads them); every imported module runs + (`sideEffects: false` drops nothing); `import.meta.url`, `.dirname`, `.filename`, `__dirname` and `__filename` name + each module's source file. One ES module with an inline source map, forked with `--enable-source-maps` before the + app's `execArgv` (stack traces name the source files); `execArgv` and `env` apply to every fork as before (modules an + `--import` names load from disk at each fork). The file lives in a private temporary directory (0700, written again + into a new one when a cleaner removed it), removed at `close`. Plugins are unaffected: workers evaluate the server's + built scripts. Cost: a build at start (the kit's test worker 55-100 ms, p2's session worker 215-250 ms, 4.2 MB of + code); a fork then loads one file (the kit's test worker 60 ms instead of 360 ms from source, p2's 350 ms instead of + 760 ms). - **Keep-alive** (D9, R3, Q5): the child reports when `host.activity`'s references, jobs and lingering objects and the app's `busy` counts flip between all zero and not (local calls do not count; the first state with `ready`, which also carries the load report); the server stops a worker idle for `idleMs` (5 min). diff --git a/packages/chord/docs/http-workers/book.md b/packages/chord/docs/http-workers/book.md index 93cba687b..0c8ebe390 100644 --- a/packages/chord/docs/http-workers/book.md +++ b/packages/chord/docs/http-workers/book.md @@ -75,7 +75,7 @@ const plugins = createPlugins({ onError, }, ctx); const http = createHttp(server, { listen: { host: "127.0.0.1", port: 8788 }, onError }, ctx); // settings are arguments -const workers = createWorkers(server, Session, { +const workers = await createWorkers(server, Session, { kind: "worker", entry: "./worker.ts", plugins, principal: (a) => ({ kind: "worker", session: a.params.id }), start: (address, ctx) => ({ id: address.params.id, dir: sessionDir(address) }), lock: (address) => `chat:${dataDir}:${address.key}`, authorize: mayReach, onError, diff --git a/packages/chord/docs/http-workers/index.html b/packages/chord/docs/http-workers/index.html index 00512c54b..0be7a891c 100644 --- a/packages/chord/docs/http-workers/index.html +++ b/packages/chord/docs/http-workers/index.html @@ -185,7 +185,7 @@ workers component, then loads its plugins:

onError, }, ctx); const http = createHttp(server, { listen: { host: "127.0.0.1", port: 8788 }, onError }, ctx); // settings are arguments -const workers = createWorkers(server, Session, { +const workers = await createWorkers(server, Session, { kind: "worker", entry: "./worker.ts", plugins, principal: (a) => ({ kind: "worker", session: a.params.id }), start: (address, ctx) => ({ id: address.params.id, dir: sessionDir(address) }), lock: (address) => `chat:${dataDir}:${address.key}`, authorize: mayReach, onError, diff --git a/packages/chord/src/node/app/app.ts b/packages/chord/src/node/app/app.ts index 039c910b2..43a03d76b 100644 --- a/packages/chord/src/node/app/app.ts +++ b/packages/chord/src/node/app/app.ts @@ -72,7 +72,10 @@ export type AppKind = { readonly platform: "node" | "browser"; readonly packages export type AppWorkerKind

= { /** The host kind (`hostKind("project", { id: str() })`). */ readonly kind: KindShape; - /** The worker's entry module (it calls `startWorker`), its Node flags and environment. */ + /** + * The worker's entry module (it calls `startWorker`), bundled with the base code it imports at start (every worker of + * the kind runs that snapshot until the server restarts); its Node flags and environment. + */ readonly entry: string; readonly execArgv?: readonly string[]; readonly env?: Readonly>; @@ -137,7 +140,8 @@ export type AppOptions

= { readonly principalAttributes?: PrincipalAttributes

; /** * The app's base code beside what the kit knows (files or directories: its API modules, its start script): a change - * sets `plugins.state.base`. The kit adds each worker kind's `entry` and every file its base bundles were built from. + * sets `plugins.state.base`. The kit adds every file its base bundles were built from: the web's, the terminal's and + * each worker kind's. */ readonly base?: readonly string[]; /** The `closeTimeout` and `readyTimeout` of every host (the server's and each worker's): 30 s and 60 s by default. */ @@ -260,9 +264,7 @@ export async function startServer

(options: AppOptions

, c context, ); // Base code by kind: directories are watched as trees, files one by one (`watchBase`). - const base = [...(options.base ?? []), ...(options.workers ?? []).map(({ entry }) => entry)].map((path) => - resolve(path), - ); + const base = (options.base ?? []).map((path) => resolve(path)); const baseDirs = base.filter((path) => statSync(path, { throwIfNoEntry: false })?.isDirectory() === true); const plugins = createPlugins( { @@ -284,9 +286,6 @@ export async function startServer

(options: AppOptions

, c const workers = new Map>(); // The address key of each worker by its principal's key: the principal its link stamps on its telemetry batches. const workerHosts = new Map(); - for (const kind of options.workers ?? []) { - workers.set(kind.kind.kind, workersOf(host, kind, options, layout.dataDir, plugins, workerHosts, context)); - } const link = new OwnerLink(); const core: PluginFacet

[] = [ { name: "chord.http", facet: http.facet }, @@ -319,12 +318,20 @@ export async function startServer

(options: AppOptions

, c const surfaces: Surface

[] = []; const app = new RunningApp(host, http, workers, plugins, surfaces, { store, lock }, layout, options); try { + // Each worker kind's base bundled now: one that does not build fails the start, everything closed. + for (const kind of options.workers ?? []) { + workers.set( + kind.kind.kind, + await workersOf(host, kind, options, layout.dataDir, plugins, workerHosts, context), + ); + } for (const surface of await surfacesOf(host, options, plugins, context)) { surfaces.push(surface); core.push({ name: `chord.${surface.name}`, facet: surface.facet }); } const files = base.filter((path) => !baseDirs.includes(path)); - plugins.watchBase([...files, ...surfaces.flatMap(({ inputs }) => inputs)]); + const bundled = [...surfaces, ...workers.values()].flatMap(({ inputs }) => inputs); + plugins.watchBase([...files, ...bundled]); await plugins.load(host, { kind: "server", core, modules: modulesOf(api, options.kinds.server) }, context); await ownerTokenFile(layout.ownerToken, host, context); app.started(link.renew(), await http.listening(context)); @@ -401,7 +408,10 @@ function modulesOf(api: AppApi, kind: AppKind | undefined): HostPackages { return { ...api.modules, ...kind?.packages }; } -/** The workers component of one worker kind: started with the kit's part of the start data beside the app's. */ +/** + * The workers component of one worker kind, its base bundled: started with the kit's part of the start data beside the + * app's. + */ function workersOf

( host: Host

, kind: AppWorkerKind

, @@ -410,7 +420,7 @@ function workersOf

( plugins: Plugins, workerHosts: Map, context: Context, -): Workers { +): Promise> { const facet = kind.facet ?? "worker"; const serverKey = canonicalJson(options.principal.server as unknown as JsonValue, context); const isServer = (principal: P): boolean => canonicalJson(principal as unknown as JsonValue, context) === serverKey; diff --git a/packages/chord/src/node/surfaces/bundle.ts b/packages/chord/src/node/surfaces/bundle.ts index 0e0b6e7c3..65b57385f 100644 --- a/packages/chord/src/node/surfaces/bundle.ts +++ b/packages/chord/src/node/surfaces/bundle.ts @@ -5,7 +5,7 @@ import { createHash } from "node:crypto"; import { basename, extname, resolve } from "node:path"; -import { build } from "esbuild"; +import { build, type Metafile } from "esbuild"; import { CSS, JAVASCRIPT, mimeOf, type ServedFile } from "./served.ts"; /** @@ -69,9 +69,13 @@ export async function buildBase(entry: string, platform: BasePlatform, prefix: s files.set(`${prefix}assets/${basename(output.path)}`, { mime: mimeOf(output.path), bytes: output.contents }); } } + return { hash: digest, script, style, files, inputs: metafileInputs(result.metafile) }; +} + +/** The files a build read, absolute (a base bundle's and the worker base's inputs). */ +export function metafileInputs(metafile: Metafile): string[] { // Metafile inputs are relative to the working directory; a namespaced one (`ns:path`) is no file. - const inputs = Object.keys(result.metafile.inputs) + return Object.keys(metafile.inputs) .filter((input) => !/^[a-z-]+:/.test(input)) .map((input) => resolve(input)); - return { hash: digest, script, style, files, inputs }; } diff --git a/packages/chord/src/node/workers/README.md b/packages/chord/src/node/workers/README.md index fee223bb1..d8640a238 100644 --- a/packages/chord/src/node/workers/README.md +++ b/packages/chord/src/node/workers/README.md @@ -1,15 +1,15 @@ # workers The workers component (slice 10 §3, §4, design §23, decision "Slice 10's points"): a forked worker process per address -of one host kind, started on demand by the route it owns, stopped when idle, never restarted by itself; its plugins are -built by the server and shipped. `workerMain` is the worker's side. Node-only. Exported as -`@earendil-works/chord/workers`. +of one host kind, started on demand by the route it owns, stopped when idle, never restarted by itself; forked from the +kind's worker base, bundled when the component is created; its plugins are built by the server and shipped. +`workerMain` is the worker's side. Node-only. Exported as `@earendil-works/chord/workers`. ## For apps (`index.ts`) ```ts // server.ts -const workers = createWorkers(server, Session, { +const workers = await createWorkers(server, Session, { // rejects when the worker base does not build kind: "worker", entry: "./worker.ts", execArgv: ["--import", resolver], env: { … }, // bootstrap authorize: (principal, address) => app.mayReach(principal, address), // the route's (synchronous) principal: (address) => ({ kind: "worker", session: address.params.id }), @@ -21,6 +21,7 @@ const workers = createWorkers(server, Session, { onError: (error) => log(error), // WorkersError }, ctx); workers.state // { [address key]: WorkerState }, a JSON watchable +workers.inputs // the files the worker base was built from (base code to watch) await workers.stop(Session({ id }), ctx); // suspended while held, then stopped await workers.retry(Session({ id }), ctx); // a failed or suspended address starts again (if held) await plugins.rollout([plugins.step("server", server), workers.step("workers")], built, ctx); @@ -40,18 +41,31 @@ workerMain(async ({ address, start, parent, plugins, onClose }, ctx) => { `resolve` rejects at once with its stored failure (its holders see `failed { waiting }`, nothing forks); the kernel reports that rejection (`resolve_failed`) once per new hold, so every tab opening a failed session adds one `onError` entry. +- **The worker base** (decision "Workers start from the base code the server booted with"): `createWorkers` bundles + `entry` and the base code it imports once (esbuild, before routing) and forks every worker from that snapshot, so an + edit of the app's source while the server runs (half done, maybe) reaches no worker until the server restarts. A + base that does not build rejects `createWorkers`. Inlined: the app's files and workspace packages linked into + `node_modules` (by their real paths). External, loaded by Node at each fork from the file the build resolved: Node + builtins and every package resolved under `node_modules` (native addons, packages reading their own files). Imports + resolve as the child's Node would (`--conditions` from `execArgv`, `main` only, `tsconfig.json` paths); every + imported module runs (`sideEffects: false` drops none); `import.meta.url`, `.dirname`, `.filename`, `__dirname` and + `__filename` name each module's source file. One ES module with an inline source map, forked with + `--enable-source-maps` ahead of the app's `execArgv` (stack traces name the source files). Its file is in a private + temporary directory, written again into a new one when a cleaner removed it, removed at `close`. `inputs` lists the + files it was built from: the kit marks `base` when one changes. - **Per address:** a child's events (`started`, `ready`, `idle`, `fatal`, its exit, its link closed by this side) are recorded on that child as they happen (a late event of an old child changes nothing); starts and stops run one after another. A start waits for the previous child's exit and rejects with a crash recorded meanwhile; a stop records its intent before `shutdown`, so the exit it causes is no crash. - **A start:** the app's `start(address, ctx)` (aborted by `close`), then the fork (`serialization: "advanced"`, a - detached process group, the server's environment with `env` over it: a worker inherits every variable the server - has, secrets included, unless the app starts the server with a trimmed one; stdout and stderr piped: the stderr tail - kept, both mirrored to this process's own without `onOutput`), its IPC link made at once, the start message `{ - address, start, scripts, lock }` (`scripts`: the kind's current set, `plugins.scripts(kind)`). The child takes the - lock (waiting for an old holder without polling), makes its parent link and sends `started` when the app's host - connects; `resolve` returns the link then. The start timer (60 s) covers fork, lock and connect only. `ready` (after - the app's function returned, with the idle state and the load report) makes the worker `running`. + detached process group, the worker base's file with `--enable-source-maps` and `execArgv`, the server's environment + with `env` over it: a worker inherits every variable the server has, secrets included, unless the app starts the + server with a trimmed one; stdout and stderr piped: the stderr tail kept, both mirrored to this process's own + without `onOutput`), its IPC link made at once, the start message `{ address, start, scripts, lock }` (`scripts`: + the kind's current set, `plugins.scripts(kind)`). The child takes the lock (waiting for an old holder without + polling), makes its parent link and sends `started` when the app's host connects; `resolve` returns the link then. + The start timer (60 s) covers fork, lock and connect only. `ready` (after the app's function returned, with the idle + state and the load report) makes the worker `running`. - **Failures stay** (R2): a crash (an exit nobody asked for), a `fatal` or a failed start leaves the address `failed` (`worker_failed` once, `worker_exited` for the exit) until `retry` or a rollout reaching workers. - **A lost link is recovered from** (decision "A lost worker link is recovered from"): a link closed by this side (the @@ -73,8 +87,9 @@ workerMain(async ({ address, start, parent, plugins, onClose }, ctx) => { `authorize_failed`, its error private too. - **`stop(address, ctx)`:** `RouteHandle.suspend`, then the stop: holders see `failed { waiting }` ("suspended") until their last release; a later hold starts it on demand; `retry` lifts it at once. -- **`close(ctx)`:** starts waiting in `start(address)` are aborted, every child stopped, new starts reject - `released/closed`; the host's close (the route's context) closes it too, and no exit after that is a failure. +- **`close(ctx)`:** starts waiting in `start(address)` are aborted, the worker base's file removed, every child + stopped, new starts reject `released/closed`; the host's close (the route's context) closes it too, and no exit + after that is a failure. - **`step(name)`** (§4): clears the failures from before it and retries the held ones (listed `{ host, report }` with their load report once ready, or `{ host, error }`), then ships the kind's current set to every worker whose last set is older, in parallel, one reload outstanding per child: `{ host, report }`, `{ host, stopped: true }` (stopped @@ -97,6 +112,6 @@ stays failed, Retry brings it back), `43-rollout` (an edit rolled out to the ser ## Inside the directory -`control.ts` (control messages and errors as they cross), `lock.ts` (process locks), `process.ts` (one forked worker -as the server sees it), `workers.ts` (the component: the route, per-address records, idle stops, the rollout step, -the state), `main.ts` (`workerMain`). +`base.ts` (the worker base: the entry bundled once), `control.ts` (control messages and errors as they cross), +`lock.ts` (process locks), `process.ts` (one forked worker as the server sees it), `workers.ts` (the component: the +route, per-address records, idle stops, the rollout step, the state), `main.ts` (`workerMain`). diff --git a/packages/chord/src/node/workers/base.ts b/packages/chord/src/node/workers/base.ts new file mode 100644 index 000000000..e94c1ef36 --- /dev/null +++ b/packages/chord/src/node/workers/base.ts @@ -0,0 +1,196 @@ +// The worker base (decision "Workers start from the base code the server booted with"): a worker kind's entry and the +// base code it imports, bundled once by esbuild when the workers component is created, as the surfaces build their base +// bundles at start. Every worker of the kind is forked from this snapshot, so an edit of the app's source while the +// server runs reaches workers only through a restart of the server. Packages installed under `node_modules` (native +// addons, packages reading their own files) and Node builtins stay external: Node loads them at each fork from where +// the build resolved them. Everything else, the app's files and workspace packages linked into `node_modules`, is +// inlined. The bundle is one ES module with an inline source map in a private temporary directory. + +import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { readFile } from "node:fs/promises"; +import { isBuiltin } from "node:module"; +import { tmpdir } from "node:os"; +import { dirname, extname, join, parse, resolve, sep } from "node:path"; +import { pathToFileURL } from "node:url"; +import { build, type Loader, type Plugin } from "esbuild"; +import { metafileInputs } from "../surfaces/bundle.ts"; + +/** A worker kind's bundled base: the file workers are forked from, and the files it was built from. */ +export interface WorkerBase { + /** The files esbuild read for it (absolute): the app's base code a change of which outdates it. */ + readonly inputs: readonly string[]; + /** The bundle's path, written again into a new private directory when a temporary-file cleaner removed it. */ + file(): string; + /** Removes the bundle's directory. */ + remove(): void; +} + +/** + * What `import.meta.url`, `.dirname` and `.filename` (and a CommonJS file's free `__dirname` and `__filename`) become: a + * module's own constants (`moduleLocations`). + */ +const MODULE_URL = "__chordModuleUrl"; +const MODULE_DIRNAME = "__chordModuleDirname"; +const MODULE_FILENAME = "__chordModuleFilename"; + +/** + * Bundles `entry` for workers forked with `execArgv` (its `--conditions` resolve imports as the child's Node would). + * Rejects with esbuild's failure: the server does not start with a broken worker base. + */ +export async function buildWorkerBase(entry: string, execArgv: readonly string[]): Promise { + // Built as if at the file system's root: the source map's paths relative to it, made absolute by `sourceRoot`. + const root = parse(resolve(entry)).root; + const result = await build({ + entryPoints: [entry], + outfile: join(root, "worker.mjs"), + write: false, + bundle: true, + format: "esm", + platform: "node", + // Node's own resolution: the child's conditions, `main` only (no `module` condition or field). + conditions: conditionsOf(execArgv), + mainFields: ["main"], + // Every imported module runs, as Node runs it: `sideEffects: false` and pure annotations drop none (the entry's + // own package may say it, as Chord's does). + ignoreAnnotations: true, + // Function and class names as the source has them (`name` stays what unbundled code sees). + keepNames: true, + sourcemap: "inline", + sourcesContent: false, + sourceRoot: pathToFileURL(root).href, + // A CommonJS file of the base code requires through this; external packages by their absolute paths. + banner: { + js: 'import { createRequire as __chordCreateRequire } from "node:module"; const require = __chordCreateRequire(import.meta.url);', + }, + define: { + "import.meta.url": MODULE_URL, + "import.meta.dirname": MODULE_DIRNAME, + "import.meta.filename": MODULE_FILENAME, + __dirname: MODULE_DIRNAME, + __filename: MODULE_FILENAME, + }, + plugins: [installedPackages(), moduleLocations()], + logLevel: "silent", + metafile: true, + }); + // One entry, one output file. + const [output] = result.outputFiles; + return new BundleFile(output!.contents, metafileInputs(result.metafile)); +} + +/** The conditions of `--conditions=`, `--conditions `, `-C ` and `-C=` in `execArgv`. */ +function conditionsOf(execArgv: readonly string[]): string[] { + const conditions: string[] = []; + for (let i = 0; i < execArgv.length; i++) { + const flag = execArgv[i]!; + const inline = /^(?:--conditions|-C)=(.*)$/.exec(flag); + if (inline !== null) conditions.push(inline[1]!); + else if ((flag === "--conditions" || flag === "-C") && i + 1 < execArgv.length) conditions.push(execArgv[++i]!); + } + return conditions; +} + +/** Marks the resolver's own call, so the plugin does not resolve it again. */ +const RESOLVING = Symbol("chord.worker-base.resolving"); + +/** + * Keeps a package resolved under a `node_modules` directory external, by its absolute file (a URL for an `import`, a + * path for a `require`): the bundle lives elsewhere, so a bare specifier would not resolve from it. A workspace + * package resolves to its real path outside `node_modules` and is inlined. A specifier that does not resolve is left to + * esbuild (an error, or a warning for a `require` in a `try`). + */ +function installedPackages(): Plugin { + return { + name: "chord-worker-installed-packages", + setup(builder) { + builder.onResolve({ filter: /^[^./]/ }, async (args) => { + if (args.pluginData === RESOLVING || isBuiltin(args.path)) return undefined; + const resolved = await builder.resolve(args.path, { + kind: args.kind, + importer: args.importer, + resolveDir: args.resolveDir, + pluginData: RESOLVING, + }); + if (resolved.errors.length > 0) return undefined; + if (!resolved.path.split(sep).includes("node_modules")) return resolved; + const path = args.kind.startsWith("require") ? resolved.path : pathToFileURL(resolved.path).href; + return { path, external: true }; + }); + }, + }; +} + +/** Whether a source names its own location. */ +const LOCATIONS = /import\.meta|__dirname|__filename/; + +/** The loaders of the source files `moduleLocations` reads: every extension its filter matches. */ +const LOADERS: Readonly> = { + ".js": "js", + ".mjs": "js", + ".cjs": "js", + ".jsx": "jsx", + ".ts": "ts", + ".mts": "ts", + ".cts": "ts", + ".tsx": "tsx", +}; + +/** + * Keeps `import.meta.url`, `.dirname` and `.filename`, `__dirname` and `__filename` naming each module's source file, not + * the bundle (base code finds files beside it this way: p2's plugins directory): a module using them gets its own + * constants, which `define` puts in their place. Declared on its first line, so the source map's lines stay. + */ +function moduleLocations(): Plugin { + return { + name: "chord-worker-module-locations", + setup(builder) { + builder.onLoad({ filter: /\.(?:[cm]?[jt]s|[jt]sx)$/ }, async (args) => { + const source = await readFile(args.path, "utf8"); + if (!LOCATIONS.test(source)) return undefined; + const constants = [ + [MODULE_URL, pathToFileURL(args.path).href], + [MODULE_DIRNAME, dirname(args.path)], + [MODULE_FILENAME, args.path], + ].map(([name, value]) => `${name} = ${JSON.stringify(value)}`); + // The filter matched one of `LOADERS`' extensions. + const loader = LOADERS[extname(args.path)]!; + return { contents: `var ${constants.join(", ")};${source}`, loader }; + }); + }, + }; +} + +/** The bundle as a file: written into a private temporary directory, again into a new one when it is gone. */ +class BundleFile implements WorkerBase { + readonly inputs: readonly string[]; + readonly #contents: Uint8Array; + #dir = ""; + #path = ""; + + constructor(contents: Uint8Array, inputs: readonly string[]) { + this.#contents = contents; + this.inputs = inputs; + this.#write(); + } + + file(): string { + // A long-running server's temporary directory may be cleaned (systemd-tmpfiles, macOS): a fork needs the file. A + // new directory, never the old name: another user may have made that one since. + if (!existsSync(this.#path)) { + this.remove(); + this.#write(); + } + return this.#path; + } + + remove(): void { + rmSync(this.#dir, { recursive: true, force: true }); + } + + /** Writes the bundle into a new private directory (mode 0700, the file 0600). */ + #write(): void { + this.#dir = mkdtempSync(join(tmpdir(), "chord-worker-")); + this.#path = join(this.#dir, "worker.mjs"); + writeFileSync(this.#path, this.#contents, { mode: 0o600 }); + } +} diff --git a/packages/chord/src/node/workers/workers.ts b/packages/chord/src/node/workers/workers.ts index f08af6199..7520625ed 100644 --- a/packages/chord/src/node/workers/workers.ts +++ b/packages/chord/src/node/workers/workers.ts @@ -3,6 +3,8 @@ // `fatal` or a failed start leaves the address failed until `retry` or a rollout. A running worker's lost link is a // stop its holders start a new worker after, once a minute (decision "A lost worker link is recovered from"). // Per address, a child's events are recorded on that child as they happen; starts and stops run one after another. +// Every worker is forked from the kind's worker base, bundled when the component is created (decision "Workers start +// from the base code the server booted with"). import type { Context } from "../../context/index.ts"; import { awaitWithContext, BACKGROUND_CONTEXT, withAbortSignal } from "../../context/index.ts"; @@ -15,6 +17,7 @@ import { Batch, type BatchError, Source } from "../../state/internal.ts"; import type { Watchable } from "../../watch.ts"; import type { Plugins, RolloutHost, RolloutStep } from "../plugins/plugins.ts"; import type { PluginScript } from "../plugins/runtime.ts"; +import { buildWorkerBase, type WorkerBase } from "./base.ts"; import { type CrossedError, type CrossedReport, type FromWorker, uncrossed, uncrossedReport } from "./control.ts"; import { clearStaleLock } from "./lock.ts"; import { type Exit, WorkerProcess } from "./process.ts"; @@ -23,7 +26,10 @@ import { type Exit, WorkerProcess } from "./process.ts"; export type WorkersOptions

= { /** The facet kind its workers load (`"worker"`), shipped by the plugins component. */ readonly kind: string; - /** The worker's entry module (it calls `workerMain`), its Node flags and its environment. */ + /** + * The worker's entry module (it calls `workerMain`), bundled with the base code it imports when the component is + * created; its Node flags (their `--conditions` resolve the bundle's imports too) and its environment. + */ readonly entry: string; readonly execArgv?: readonly string[]; readonly env?: Readonly>; @@ -84,25 +90,31 @@ export type WorkersError = export interface Workers { /** Per address key its `WorkerState`; no entry when stopped or never started. A JSON watchable. */ readonly state: Watchable>>; + /** The files its worker base was built from: the app's base code for its workers. */ + readonly inputs: readonly string[]; /** Suspends `address` while it has holders (`RouteHandle.suspend`), then stops its worker. */ stop(address: Address, context: Context): Promise; /** Clears `address`'s failure and suspension: it starts again now if held (`RouteHandle.retry`). */ retry(address: Address, context: Context): Promise; /** A rollout step shipping the kind's current set to every worker, one result per worker (§4). */ step(name: string): RolloutStep; - /** Cancels starts, stops every worker; new starts reject `released/closed`. */ + /** Cancels starts, stops every worker, removes the worker base's file; new starts reject `released/closed`. */ close(context: Context): Promise; } -/** Creates the workers component on `host`: it routes `Kind` itself (`host.route`, before `load`). */ -export function createWorkers

( +/** + * Bundles the kind's worker base, then creates the workers component on `host`: it routes `Kind` itself (`host.route`, + * before `load`). Rejects with esbuild's failure when the base does not build: the server does not start with it. + */ +export async function createWorkers

( host: Host

, Kind: HostKind, options: WorkersOptions>, context: Context, -): Workers> { +): Promise>> { + const base = await buildWorkerBase(options.entry, options.execArgv ?? []); // Its addresses are `Address>`, as the route hands them over. - const component = new WorkersComponent(options as unknown as WorkersOptions, context); + const component = new WorkersComponent(options as unknown as WorkersOptions, base, context); const route = host.route( Kind, { @@ -221,6 +233,7 @@ class WorkersComponent implements Workers { readonly state: Watchable>>; readonly #state: WorkersState; readonly #options: WorkersOptions; + readonly #base: WorkerBase; readonly #addresses = new Map(); /** Aborted at `close`: starts waiting in the app's `start` or for `started` end. */ readonly #closing = new AbortController(); @@ -230,14 +243,19 @@ class WorkersComponent implements Workers { #routing = false; #idleTimer: ReturnType | undefined; - constructor(options: WorkersOptions, context: Context) { + constructor(options: WorkersOptions, base: WorkerBase, context: Context) { this.#options = options; + this.#base = base; this.#state = new WorkersState(this.#addresses, new Batch((error) => options.onError(error), undefined)); // Its value is a `Record`. this.state = this.#state as unknown as Watchable>>; this.#closeWith(context); } + get inputs(): readonly string[] { + return this.#base.inputs; + } + /** * The route's `resolve`: a failed address rejects at once with its (public) failure, nothing forks; otherwise a * start, after the previous child's exit. @@ -285,6 +303,8 @@ class WorkersComponent implements Workers { async close(context: Context): Promise { if (!this.#closing.signal.aborted) this.#closing.abort(closed()); + // Nothing forks once closing: running workers loaded their module already. + this.#base.remove(); clearInterval(this.#idleTimer); this.#idleTimer = undefined; // Each stop, then what its exit queued (the clear of a killed child's lock file). @@ -415,8 +435,9 @@ class WorkersComponent implements Workers { let child!: Child; const process = new WorkerProcess( { - entry: this.#options.entry, - ...(this.#options.execArgv === undefined ? {} : { execArgv: this.#options.execArgv }), + entry: this.#base.file(), + // Stack traces name the base code's files; the app's flags come after (`--no-enable-source-maps` wins). + execArgv: ["--enable-source-maps", ...(this.#options.execArgv ?? [])], ...(this.#options.env === undefined ? {} : { env: this.#options.env }), ...(onOutput === undefined ? {} : { onOutput: (stream, chunk) => onOutput(address, stream, chunk) }), }, diff --git a/packages/chord/test/bench/workers.ts b/packages/chord/test/bench/workers.ts index a70084bd5..ef67c91ba 100644 --- a/packages/chord/test/bench/workers.ts +++ b/packages/chord/test/bench/workers.ts @@ -74,7 +74,7 @@ console.log( // Memory per idle worker on the server: its records, process handle, link and remote references. const server = createHost({ address: hostKind("server")(), principal: { kind: "server" }, onError: () => {} }); -const workers = createWorkers( +const workers = await createWorkers( server, Session, { diff --git a/packages/chord/test/examples/41-workers/main.ts b/packages/chord/test/examples/41-workers/main.ts index 08562b532..896e336e8 100644 --- a/packages/chord/test/examples/41-workers/main.ts +++ b/packages/chord/test/examples/41-workers/main.ts @@ -22,7 +22,7 @@ const plugins = createPlugins( // 1. The server, and its workers: one forked process per session, stopped after 300 ms without use. const server = createHost({ address: hostKind("server")(), principal: { kind: "server" }, onError: console.error }); -const workers = createWorkers( +const workers = await createWorkers( server, Session, { diff --git a/packages/chord/test/examples/42-worker-crash/main.ts b/packages/chord/test/examples/42-worker-crash/main.ts index 2d1d85f33..5c2ecf500 100644 --- a/packages/chord/test/examples/42-worker-crash/main.ts +++ b/packages/chord/test/examples/42-worker-crash/main.ts @@ -22,7 +22,7 @@ const plugins = createPlugins( context, ); const server = createHost({ address: hostKind("server")(), principal: { kind: "server" }, onError: () => {} }); -const workers = createWorkers( +const workers = await createWorkers( server, Session, { diff --git a/packages/chord/test/examples/43-rollout/main.ts b/packages/chord/test/examples/43-rollout/main.ts index 5a2184b00..44315ab14 100644 --- a/packages/chord/test/examples/43-rollout/main.ts +++ b/packages/chord/test/examples/43-rollout/main.ts @@ -36,7 +36,7 @@ const plugins = createPlugins( context, ); const server = createHost({ address: hostKind("server")(), principal: { kind: "server" }, onError: console.error }); -const workers = createWorkers( +const workers = await createWorkers( server, Session, { diff --git a/packages/chord/test/node/app/app.test.ts b/packages/chord/test/node/app/app.test.ts index d97d84c28..ba12be31b 100644 --- a/packages/chord/test/node/app/app.test.ts +++ b/packages/chord/test/node/app/app.test.ts @@ -1,6 +1,7 @@ // Tests of `startServer` (slice 13 §1, §4, §5): the layout and the start-data object, start and close, a server without -// workers, two worker kinds, `applyChanges`, a required root failing, and export and import per address (documents, -// tables and files round-trip; an existing address refused). +// workers, two worker kinds, `applyChanges`, a required root failing, a worker entry that does not build, the base +// marks (a worker base's inputs among them), and export and import per address (documents, tables and files +// round-trip; an existing address refused). import { chmod, mkdir, readdir, stat, symlink, writeFile } from "node:fs/promises"; import { homedir } from "node:os"; @@ -109,6 +110,17 @@ describe("starting and closing", () => { // Closed: the data directory's lock is free again (a second start fails only on the same plugin). expect(await rejection(testApp({ root, dataDir }))).toMatchObject({ name: "PluginError" }); }); + + it("rejects when a worker kind's entry does not build, with everything closed", async () => { + const base = await tempDir(); + await writeFile(join(base, "worker.ts"), "export const broken = ;"); + const worker = { kind: Project, entry: join(base, "worker.ts"), start: () => null, authorize: () => true }; + const dataDir = join(base, "data"); + expect(String(await rejection(testApp({ dataDir, workers: [worker] })))).toContain('Unexpected ";"'); + // Closed: the data directory's lock is free again. + const { app } = await testApp({ dataDir }); + await app.close(context); + }); }); describe("workers", () => { @@ -232,14 +244,22 @@ describe("applying changes one after another", () => { }); describe("base code", () => { - it("marks base for a change of `base`'s files and directories and of a worker kind's entry", async () => { + it("marks base for a change of `base`'s files and directories and of a file a worker kind's base was built from", async () => { const dir = await tempDir(); await mkdir(join(dir, "api")); await writeFile(join(dir, "api", "v1.ts"), "export {};"); await writeFile(join(dir, "start.ts"), "export {};"); - await writeFile(join(dir, "worker.ts"), "export {};"); + await writeFile(join(dir, "worker.ts"), 'import "./shared.ts";'); + await writeFile(join(dir, "shared.ts"), "export {};"); const worker = { kind: Project, entry: join(dir, "worker.ts"), start: () => null, authorize: () => true }; - for (const changed of [join(dir, "api", "v1.ts"), join(dir, "start.ts"), join(dir, "worker.ts")]) { + // The entry last: its change drops its import. + const changes = [ + join(dir, "api", "v1.ts"), + join(dir, "start.ts"), + join(dir, "shared.ts"), + join(dir, "worker.ts"), + ]; + for (const changed of changes) { const { app } = await testApp({ base: [join(dir, "api"), join(dir, "start.ts")], workers: [worker] }); try { expect(app.plugins.state.value.base).toBe(false); diff --git a/packages/chord/test/node/workers/base.test.ts b/packages/chord/test/node/workers/base.test.ts new file mode 100644 index 000000000..205662600 --- /dev/null +++ b/packages/chord/test/node/workers/base.test.ts @@ -0,0 +1,179 @@ +// Tests of the worker base (decision "Workers start from the base code the server booted with"): the entry and its base +// code bundled once (an edit after the build runs nowhere, every imported module runs), installed packages external and +// Node builtins kept, a workspace package linked into `node_modules` inlined, the child's `--conditions`, module +// locations and stack traces naming the source files, a broken entry rejected, and the file in a private directory +// restored when a cleaner removed it. + +import { execFileSync, spawnSync } from "node:child_process"; +import { existsSync, statSync } from "node:fs"; +import { mkdir, mkdtemp, readFile, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { pathToFileURL } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import { buildWorkerBase, type WorkerBase } from "../../../src/node/workers/base.ts"; + +const made: string[] = []; +const bases: WorkerBase[] = []; + +afterEach(async () => { + for (const base of bases.splice(0)) base.remove(); + for (const dir of made.splice(0)) await rm(dir, { recursive: true, force: true }); +}); + +/** A temporary directory with `files` (relative path to contents), removed after the test. */ +async function project(files: Readonly>): Promise { + const dir = await mkdtemp(join(tmpdir(), "chord-worker-base-")); + made.push(dir); + for (const [path, contents] of Object.entries(files)) { + await mkdir(dirname(join(dir, path)), { recursive: true }); + await writeFile(join(dir, path), contents); + } + return dir; +} + +/** `entry`'s base, removed after the test. */ +async function built(entry: string, execArgv: readonly string[] = []): Promise { + const base = await buildWorkerBase(entry, execArgv); + bases.push(base); + return base; +} + +/** What the bundle prints when Node runs it. */ +function run(base: WorkerBase): string { + return execFileSync(process.execPath, [base.file()], { encoding: "utf8" }).trim(); +} + +describe("the snapshot", () => { + it("bundles the entry and its imports once: an edit after the build runs nowhere; inputs name both files", async () => { + const dir = await project({ + "entry.ts": 'import { label } from "./label.ts";\nconst typed: string = label;\nconsole.log(typed);\n', + "label.ts": 'export const label = "v1";\n', + }); + const base = await built(join(dir, "entry.ts")); + await writeFile(join(dir, "label.ts"), 'export const label = "v2";\n'); + expect(run(base)).toBe("v1"); + expect([...base.inputs].sort()).toEqual([join(dir, "entry.ts"), join(dir, "label.ts")]); + }); + + it("runs every imported module, as Node does: sideEffects: false drops no side-effect import", async () => { + const dir = await project({ + "package.json": '{ "name": "app", "type": "module", "sideEffects": false }', + "entry.ts": 'import "./effect.ts";\nconsole.log("entry");\n', + "effect.ts": 'console.log("effect");\n', + }); + expect(run(await built(join(dir, "entry.ts")))).toBe("effect\nentry"); + }); + + it("a broken entry rejects with esbuild's failure", async () => { + const dir = await project({ "entry.ts": 'import { label } from "./missing.ts";\nconsole.log(label);\n' }); + await expect(buildWorkerBase(join(dir, "entry.ts"), [])).rejects.toThrow(/Could not resolve "\.\/missing\.ts"/); + }); +}); + +describe("what stays external", () => { + it("keeps installed packages and builtins external by their files; inlines a workspace package linked into node_modules", async () => { + const dir = await project({ + "entry.ts": [ + 'import { join } from "node:path";', + 'import { where as installed } from "installed";', + 'import { where as linked } from "linked";', + 'const required = require("required");', + "console.log(JSON.stringify({ installed, linked, required: required.where, join: typeof join }));", + ].join("\n"), + "node_modules/installed/package.json": '{ "name": "installed", "type": "module", "exports": "./index.js" }', + "node_modules/installed/index.js": "export const where = import.meta.url;\n", + "node_modules/required/package.json": '{ "name": "required", "main": "index.js" }', + "node_modules/required/index.js": "exports.where = __filename;\n", + "workspace/linked/package.json": '{ "name": "linked", "type": "module", "exports": "./index.js" }', + "workspace/linked/index.js": 'export const where = "linked v1";\n', + }); + await symlink(join(dir, "workspace", "linked"), join(dir, "node_modules", "linked"), "dir"); + const base = await built(join(dir, "entry.ts")); + const bundle = await readFile(base.file(), "utf8"); + expect(bundle).toContain(`from "${pathToFileURL(join(dir, "node_modules", "installed", "index.js")).href}"`); + expect(bundle).toContain(JSON.stringify(join(dir, "node_modules", "required", "index.js"))); + expect(bundle).toContain('from "node:path"'); + expect(base.inputs).toContain(join(dir, "workspace", "linked", "index.js")); + // The linked package is base code (in the snapshot); installed packages load from disk at each fork. + await writeFile(join(dir, "workspace", "linked", "index.js"), 'export const where = "linked v2";\n'); + await writeFile(join(dir, "node_modules", "required", "index.js"), 'exports.where = "required v2";\n'); + expect(JSON.parse(run(base))).toEqual({ + installed: pathToFileURL(join(dir, "node_modules", "installed", "index.js")).href, + linked: "linked v1", + required: "required v2", + join: "function", + }); + }); + + it("resolves with the child's --conditions (both spellings)", async () => { + const dir = await project({ + "entry.ts": 'import { from } from "conditional";\nconsole.log(from);\n', + "workspace/conditional/package.json": JSON.stringify({ + name: "conditional", + type: "module", + exports: { ".": { source: "./source.js", default: "./index.js" } }, + }), + "workspace/conditional/source.js": 'export const from = "source";\n', + "workspace/conditional/index.js": 'export const from = "default";\n', + }); + await mkdir(join(dir, "node_modules")); + await symlink(join(dir, "workspace", "conditional"), join(dir, "node_modules", "conditional"), "dir"); + const entry = join(dir, "entry.ts"); + expect(run(await built(entry))).toBe("default"); + expect(run(await built(entry, ["--no-warnings", "--conditions=source"]))).toBe("source"); + expect(run(await built(entry, ["-C", "source"]))).toBe("source"); + }); +}); + +describe("module locations and stack traces", () => { + it("import.meta and __dirname name each source file, a local __dirname stays the module's own", async () => { + const dir = await project({ + "entry.ts": [ + 'import { here } from "./sub/here.ts";', + 'import cjs from "./sub/legacy.cjs";', + "console.log(JSON.stringify({ here, cjs, url: import.meta.url }));", + ].join("\n"), + "sub/here.ts": [ + 'import { fileURLToPath } from "node:url";', + 'import { dirname } from "node:path";', + "const __dirname = dirname(fileURLToPath(import.meta.url));", + "export const here = { dirname: __dirname, meta: import.meta.dirname, file: import.meta.filename };", + ].join("\n"), + "sub/legacy.cjs": "module.exports = { dirname: __dirname, filename: __filename };\n", + }); + const sub = join(dir, "sub"); + expect(JSON.parse(run(await built(join(dir, "entry.ts"))))).toEqual({ + here: { dirname: sub, meta: sub, file: join(sub, "here.ts") }, + cjs: { dirname: sub, filename: join(sub, "legacy.cjs") }, + url: pathToFileURL(join(dir, "entry.ts")).href, + }); + }); + + it("maps stack traces to the source files and lines with --enable-source-maps", async () => { + const dir = await project({ + "entry.ts": 'import { fail } from "./fail.ts";\nfail();\n', + "fail.ts": + "const at: string = import.meta.url;\n\nexport function fail(): void {\n\tthrow new Error(at);\n}\n", + }); + const base = await built(join(dir, "entry.ts")); + const { stderr } = spawnSync(process.execPath, ["--enable-source-maps", base.file()], { encoding: "utf8" }); + expect(stderr).toContain(`${join(dir, "fail.ts")}:4:`); + expect(stderr).toContain(`${join(dir, "entry.ts")}:2:`); + }); +}); + +describe("the bundle's file", () => { + it("lives in a private directory; written again into a new one when a cleaner removed it; removed by remove", async () => { + const dir = await project({ "entry.ts": 'console.log("ran");\n' }); + const base = await buildWorkerBase(join(dir, "entry.ts"), []); + const first = base.file(); + expect([statSync(dirname(first)).mode & 0o777, statSync(first).mode & 0o777]).toEqual([0o700, 0o600]); + await rm(dirname(first), { recursive: true, force: true }); + const second = base.file(); + expect(second).not.toBe(first); + expect(run(base)).toBe("ran"); + base.remove(); + expect(existsSync(dirname(second))).toBe(false); + }); +}); diff --git a/packages/chord/test/node/workers/support.ts b/packages/chord/test/node/workers/support.ts index 15c68fab8..d26e16f78 100644 --- a/packages/chord/test/node/workers/support.ts +++ b/packages/chord/test/node/workers/support.ts @@ -32,7 +32,8 @@ export type TestPrincipal = | { readonly kind: "worker"; readonly id: string } | { readonly kind: "guest"; readonly id: string }; -const ENTRY = fileURLToPath(new URL("../fixtures/workers/worker.ts", import.meta.url)); +/** The fixture worker's entry. */ +export const ENTRY = fileURLToPath(new URL("../fixtures/workers/worker.ts", import.meta.url)); const PLUGINS = fileURLToPath(new URL("../fixtures/workers/plugins/", import.meta.url)); @@ -67,6 +68,7 @@ export type Fleet = { /** Starts a server routing `Session` to fixture workers; `options` override the workers' defaults. */ export async function fleet( options: { + readonly entry?: string; readonly idleMs?: number; readonly heartbeat?: { readonly ping?: number; readonly close?: number }; readonly start?: (address: { readonly key: string }, context: Context) => Promise; @@ -92,12 +94,12 @@ export async function fleet( onError: (error) => hostErrors.push(error), ...(options.heartbeat === undefined ? {} : { heartbeat: options.heartbeat }), }); - const workers = createWorkers( + const workers = await createWorkers( server, Session, { kind: "worker", - entry: ENTRY, + entry: options.entry ?? ENTRY, execArgv: ["--conditions=source", "--experimental-strip-types", "--no-warnings"], authorize: options.authorize ?? (() => true), principal: options.principal ?? ((address) => ({ kind: "worker", id: address.params.id }) as const), diff --git a/packages/chord/test/node/workers/workers.test.ts b/packages/chord/test/node/workers/workers.test.ts index 638d1bec3..ce2f48114 100644 --- a/packages/chord/test/node/workers/workers.test.ts +++ b/packages/chord/test/node/workers/workers.test.ts @@ -2,9 +2,11 @@ // `connect`, the start timer covering only the start), failures that stay (a crash, a `fatal`, a failed start: no fork // for a new hold, holders `failed { waiting }`) until `retry` or a rollout, a running worker's lost link recovered from // once a minute, idle stops and `busy`, `stop` and its suspension, the per-address queue, `close`, the rollout step, -// stale lock files cleared, and nothing kept of 1,000 addresses. +// stale lock files cleared, nothing kept of 1,000 addresses, and the worker base (an edit after the component was +// created runs in no worker; a broken entry rejects its creation). import { spawnSync } from "node:child_process"; +import { existsSync } from "node:fs"; import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -13,10 +15,11 @@ import { describe, expect, it, vi } from "vitest"; import { withAbortSignal } from "../../../src/context/index.ts"; import { createHost } from "../../../src/host/index.ts"; import { memoryLinkPair } from "../../../src/link/index.ts"; -import type { RolloutHost } from "../../../src/node/plugins/index.ts"; +import { createPlugins, type RolloutHost } from "../../../src/node/plugins/index.ts"; +import { createWorkers } from "../../../src/node/workers/index.ts"; import { expectCollected } from "../../support.ts"; import { Control } from "../fixtures/workers/control.ts"; -import { context, fleet, Notes, Session, type TestPrincipal, until } from "./support.ts"; +import { context, ENTRY, fleet, Notes, Server, Session, type TestPrincipal, until } from "./support.ts"; /** A step's hosts as `[host, what]`. */ function listed(hosts: readonly RolloutHost[]) { @@ -55,6 +58,60 @@ describe("starts on demand", () => { }); }); +describe('the worker base (decision "Workers start from the base code the server booted with")', () => { + it("a worker started after an edit of the entry's imported source runs the code the component was created with", async () => { + const dir = await mkdtemp(join(tmpdir(), "chord-workers-base-")); + // The fixture worker, plus a line naming its label and the file it was forked from. + await writeFile(join(dir, "label.ts"), 'export const label = "boot";\n'); + await writeFile( + join(dir, "entry.ts"), + [ + `import ${JSON.stringify(ENTRY)};`, + 'import { label } from "./label.ts";', + 'process.stdout.write(["label", label, process.argv[1]].join(" ") + "\\n");', + ].join("\n"), + ); + const { server, workers, output } = await fleet({ entry: join(dir, "entry.ts") }); + expect(workers.inputs).toEqual(expect.arrayContaining([join(dir, "entry.ts"), join(dir, "label.ts"), ENTRY])); + // Half way through an edit: the source on disk no longer even parses. + await writeFile(join(dir, "label.ts"), "export const label = ;\n"); + const notes = await server.get(Notes, { id: "S1" }, context); + expect(notes.label.value).toBe("S1 v1"); + const line = output.find((each) => each.startsWith("session|S1 label ")); + expect(line).toMatch(/^session\|S1 label boot /); + notes.release(); + // The worker ran from the bundle's file, which the component's close removes. + const forkedFrom = line!.split(" ")[3]!; + expect(existsSync(forkedFrom)).toBe(true); + await workers.close(context); + expect(existsSync(forkedFrom)).toBe(false); + await rm(dir, { recursive: true, force: true }); + }); + + it("a broken entry rejects createWorkers: nothing is routed, the kind can be routed again", async () => { + const dir = await mkdtemp(join(tmpdir(), "chord-workers-base-")); + await writeFile(join(dir, "entry.ts"), "export const broken = ;\n"); + const server = createHost({ address: Server(), principal: { kind: "server" }, onError: () => {} }); + const plugins = createPlugins({ roots: [dir], kinds: {}, onError: () => {} }, context); + const options = { + kind: "worker", + entry: join(dir, "entry.ts"), + authorize: () => true, + principal: () => ({ kind: "server" }) as const, + start: () => ({}), + lock: (address: { readonly key: string }) => `chord-test:${dir}:${address.key}`, + plugins, + onError: () => {}, + }; + await expect(createWorkers(server, Session, options, context)).rejects.toThrow(/Unexpected ";"/); + const workers = await createWorkers(server, Session, { ...options, entry: ENTRY }, context); + await workers.close(context); + await server.close(context); + plugins.close(); + await rm(dir, { recursive: true, force: true }); + }); +}); + describe("failures stay until retry or a rollout (R2)", () => { it("a start that throws fails the address: holders fail waiting with it, a new hold forks nothing; retry starts it", async () => { let fail = true;