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;