From 8a95667e2ff95cf55213ca0c53336b415c494479 Mon Sep 17 00:00:00 2001 From: Anton Date: Thu, 30 Jul 2026 14:25:51 +0300 Subject: [PATCH] feat: add callbacks, Node streams, and deep wire encoding Harden 1.x for sticky stateful actors: destroy({ close }), nested actor refs, coverage in CI, and JSDoc across the lib. --- .github/workflows/ci-cd-dev.yml | 4 + .github/workflows/ci-cd-master.yml | 4 + .github/workflows/pull-request.yml | 4 + README.md | 65 ++- docs/index.md | 65 ++- eslint.config.js | 1 + package.json | 4 + src/examples/actors/streamer.ts | 37 ++ src/examples/bench.ts | 68 +++ src/examples/callbacks.ts | 18 + src/examples/streams.ts | 17 + src/lib/actor-meta.ts | 22 + src/lib/index.ts | 7 + src/lib/protocol/callback-registry.ts | 121 ++++ src/lib/protocol/messages.ts | 178 ++++++ src/lib/protocol/plain.ts | 26 + src/lib/protocol/refs.ts | 99 +++- src/lib/protocol/serializer.ts | 204 ++++++- src/lib/protocol/stream-bridge.ts | 592 ++++++++++++++++++++ src/lib/proxy/proxy.ts | 15 + src/lib/runtime/host-stream-transport.ts | 109 ++++ src/lib/runtime/module-dir.ts | 5 +- src/lib/runtime/registry.ts | 20 + src/lib/runtime/runtime.ts | 62 ++- src/lib/runtime/scheduler.ts | 17 + src/lib/runtime/stream-router.ts | 64 +++ src/lib/runtime/worker-node.ts | 673 ++++++++++++++++++++--- src/lib/types.ts | 71 ++- src/lib/worker/runtime-worker.ts | 473 ++++++++++++++-- src/test/fixtures/callback-actor.ts | 33 ++ src/test/fixtures/closable.ts | 15 + src/test/fixtures/nested-actors.ts | 17 + src/test/fixtures/stream-actor.ts | 64 +++ src/test/runtime.unit.spec.ts | 144 ++++- vitest.unit.config.js | 15 + 35 files changed, 3160 insertions(+), 173 deletions(-) create mode 100644 src/examples/actors/streamer.ts create mode 100644 src/examples/bench.ts create mode 100644 src/examples/callbacks.ts create mode 100644 src/examples/streams.ts create mode 100644 src/lib/protocol/callback-registry.ts create mode 100644 src/lib/protocol/plain.ts create mode 100644 src/lib/protocol/stream-bridge.ts create mode 100644 src/lib/runtime/host-stream-transport.ts create mode 100644 src/lib/runtime/stream-router.ts create mode 100644 src/test/fixtures/callback-actor.ts create mode 100644 src/test/fixtures/nested-actors.ts create mode 100644 src/test/fixtures/stream-actor.ts diff --git a/.github/workflows/ci-cd-dev.yml b/.github/workflows/ci-cd-dev.yml index 2b92d56..faeb3ec 100644 --- a/.github/workflows/ci-cd-dev.yml +++ b/.github/workflows/ci-cd-dev.yml @@ -31,6 +31,10 @@ jobs: - name: Run Tests run: npm test + - name: Coverage thresholds + if: matrix.node-version == '24' + run: npm run test:coverage + release: name: Release needs: test diff --git a/.github/workflows/ci-cd-master.yml b/.github/workflows/ci-cd-master.yml index 3e8506b..15b3db2 100644 --- a/.github/workflows/ci-cd-master.yml +++ b/.github/workflows/ci-cd-master.yml @@ -31,6 +31,10 @@ jobs: - name: Run Tests run: npm test + - name: Coverage thresholds + if: matrix.node-version == '24' + run: npm run test:coverage + release: name: Release needs: test diff --git a/.github/workflows/pull-request.yml b/.github/workflows/pull-request.yml index b628b6e..3ff1daf 100644 --- a/.github/workflows/pull-request.yml +++ b/.github/workflows/pull-request.yml @@ -30,3 +30,7 @@ jobs: - name: Run Tests run: npm test + + - name: Coverage thresholds + if: matrix.node-version == '24' + run: npm run test:coverage diff --git a/README.md b/README.md index b65d3ba..8e77a47 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,7 @@ const { Runtime, actor } = require("@js-ak/remote-objects"); // CJS - **Methods are always async** from the caller’s side (even if the class method is sync) - **Actors are sticky** — an instance stays on one worker until `destroy` / `dispose` - **`return this`** becomes an actor reference (same proxy identity), not a cloned object +- **Callbacks and Node.js streams** can cross the boundary as args/results - **Workers are an implementation detail** — the public API is objects and methods ## Binding classes @@ -80,13 +81,14 @@ new Runtime({ }); ``` -Debug events include `register`, `spawn`, `destroy`, `call:start`, `call:end`, `dispose`. Spawn/call events carry `actorId` as `"workerId:objectId"`. +Debug events include `register`, `spawn`, `destroy`, `call:start`, `call:end`, `call:timeout`, `worker:error`, `bridge:call`, `bridge:result`, `dispose`. Spawn/call events carry `actorId` as `"workerId:objectId"`. ## Lifecycle ```ts -await runtime.destroy(counter); // drop one actor -await runtime.dispose(); // close actors (dispose/close if present), drain, terminate workers +await runtime.destroy(counter); // close (dispose/close) then drop one actor +await runtime.destroy(counter, { close: false }); // drop without close +await runtime.dispose(); // close actors, drain, terminate workers await runtime.dispose({ closeActors: false }); ``` @@ -94,13 +96,64 @@ After `dispose`, further `spawn` / method calls fail with a clear error. ## Passing actors as arguments -Proxies can be passed into methods (including across workers): +Proxies can be passed into methods (including across workers), including nested inside plain objects/arrays: ```ts await linker.link(counter); await linker.readOther(); + +const wrapped = await nested.wrap(counter); // { counter, label } +await nested.readWrapped(wrapped); ``` +## Callbacks + +Functions may be passed as arguments or returned from methods. Remote invocations are always async. + +```ts +await actor.withProgress(10, async (n) => { + console.log("progress", n); +}); + +const add = await actor.makeAdder(10); +await add(5); // 15 +``` + +- Callbacks in **args** live until that method call finishes (plus in-flight invokes) +- Callbacks in **return values** live until the owning actor is destroyed +- Errors inside callbacks reject the remote invoke + +## Streams + +Node.js `Readable` / `Writable` / `Duplex` can be args or results (objectMode preserved; backpressure via pause/resume). + +```ts +const stream = await actor.query(100); +for await (const row of stream) { + // ... +} +``` + +## Isolation patterns + +Use sticky actors when a worker should own long-lived state (DB pools, SDK clients, caches): + +1. Put the client behind an actor class; bind with `actor(Class, import.meta)` +2. Prefer method calls over sharing mutable state across threads +3. Use callbacks for progress / notifications; streams for row batches or ingest +4. Call `destroy(proxy)` (default `close: true`) or `dispose()` so `close`/`dispose` on the actor runs + +Same-actor calls are serialized (mailbox). Different actors may run in parallel on the pool. + +## Compared to similar tools + +| | remote-objects | Comlink | Piscina | +|--|----------------|---------|---------| +| Model | Sticky class actors + proxies | RPC proxies | Task pool | +| Best for | Stateful isolation (DB/SDK) | General worker RPC | Stateless jobs | +| Callbacks / streams | Yes (Node streams) | Callbacks / proxies | Per-task message | +| Migration between workers | No | N/A | N/A | + ## Helpers ```ts @@ -111,10 +164,12 @@ const handle = getActorHandle(counter); // { workerId, objectId } | undefined ## Limits -- Arguments and results must be structured-clone compatible (plus actor refs) +- Arguments and results must be structured-clone compatible (plus actor / callback / stream refs) +- Deep encoding walks arrays and plain objects only (not arbitrary class instances) - You cannot read instance fields through the proxy — only call methods - Overlapping calls to the **same** actor are queued (mailbox); different actors may run in parallel - Actors are not migrated between workers; identity is fixed at spawn +- Circular structures in encoded plain objects are rejected with a clear error ## License diff --git a/docs/index.md b/docs/index.md index b65d3ba..8e77a47 100644 --- a/docs/index.md +++ b/docs/index.md @@ -42,6 +42,7 @@ const { Runtime, actor } = require("@js-ak/remote-objects"); // CJS - **Methods are always async** from the caller’s side (even if the class method is sync) - **Actors are sticky** — an instance stays on one worker until `destroy` / `dispose` - **`return this`** becomes an actor reference (same proxy identity), not a cloned object +- **Callbacks and Node.js streams** can cross the boundary as args/results - **Workers are an implementation detail** — the public API is objects and methods ## Binding classes @@ -80,13 +81,14 @@ new Runtime({ }); ``` -Debug events include `register`, `spawn`, `destroy`, `call:start`, `call:end`, `dispose`. Spawn/call events carry `actorId` as `"workerId:objectId"`. +Debug events include `register`, `spawn`, `destroy`, `call:start`, `call:end`, `call:timeout`, `worker:error`, `bridge:call`, `bridge:result`, `dispose`. Spawn/call events carry `actorId` as `"workerId:objectId"`. ## Lifecycle ```ts -await runtime.destroy(counter); // drop one actor -await runtime.dispose(); // close actors (dispose/close if present), drain, terminate workers +await runtime.destroy(counter); // close (dispose/close) then drop one actor +await runtime.destroy(counter, { close: false }); // drop without close +await runtime.dispose(); // close actors, drain, terminate workers await runtime.dispose({ closeActors: false }); ``` @@ -94,13 +96,64 @@ After `dispose`, further `spawn` / method calls fail with a clear error. ## Passing actors as arguments -Proxies can be passed into methods (including across workers): +Proxies can be passed into methods (including across workers), including nested inside plain objects/arrays: ```ts await linker.link(counter); await linker.readOther(); + +const wrapped = await nested.wrap(counter); // { counter, label } +await nested.readWrapped(wrapped); ``` +## Callbacks + +Functions may be passed as arguments or returned from methods. Remote invocations are always async. + +```ts +await actor.withProgress(10, async (n) => { + console.log("progress", n); +}); + +const add = await actor.makeAdder(10); +await add(5); // 15 +``` + +- Callbacks in **args** live until that method call finishes (plus in-flight invokes) +- Callbacks in **return values** live until the owning actor is destroyed +- Errors inside callbacks reject the remote invoke + +## Streams + +Node.js `Readable` / `Writable` / `Duplex` can be args or results (objectMode preserved; backpressure via pause/resume). + +```ts +const stream = await actor.query(100); +for await (const row of stream) { + // ... +} +``` + +## Isolation patterns + +Use sticky actors when a worker should own long-lived state (DB pools, SDK clients, caches): + +1. Put the client behind an actor class; bind with `actor(Class, import.meta)` +2. Prefer method calls over sharing mutable state across threads +3. Use callbacks for progress / notifications; streams for row batches or ingest +4. Call `destroy(proxy)` (default `close: true`) or `dispose()` so `close`/`dispose` on the actor runs + +Same-actor calls are serialized (mailbox). Different actors may run in parallel on the pool. + +## Compared to similar tools + +| | remote-objects | Comlink | Piscina | +|--|----------------|---------|---------| +| Model | Sticky class actors + proxies | RPC proxies | Task pool | +| Best for | Stateful isolation (DB/SDK) | General worker RPC | Stateless jobs | +| Callbacks / streams | Yes (Node streams) | Callbacks / proxies | Per-task message | +| Migration between workers | No | N/A | N/A | + ## Helpers ```ts @@ -111,10 +164,12 @@ const handle = getActorHandle(counter); // { workerId, objectId } | undefined ## Limits -- Arguments and results must be structured-clone compatible (plus actor refs) +- Arguments and results must be structured-clone compatible (plus actor / callback / stream refs) +- Deep encoding walks arrays and plain objects only (not arbitrary class instances) - You cannot read instance fields through the proxy — only call methods - Overlapping calls to the **same** actor are queued (mailbox); different actors may run in parallel - Actors are not migrated between workers; identity is fixed at spawn +- Circular structures in encoded plain objects are rejected with a clear error ## License diff --git a/eslint.config.js b/eslint.config.js index 87bc67e..0957562 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -9,6 +9,7 @@ export default tseslint.config( { ignores: [ "build/**", + "coverage/**", "docs/**", "eslint.config.js", "vitest.*.js", diff --git a/package.json b/package.json index 3a472af..9f74314 100644 --- a/package.json +++ b/package.json @@ -46,10 +46,14 @@ "postbuild:cjs": "node scripts/write-cjs-package.js && node scripts/fix-cjs-import-meta.js", "test": "npm run build && npm run test:unit && node --test src/test/cjs-require.test.js", "test:unit": "vitest run --config vitest.unit.config.js", + "test:coverage": "npm run build && vitest run --config vitest.unit.config.js --coverage", + "bench": "npm run build && node build/esm/examples/bench.js", "example:counter": "npm run build && node build/esm/examples/counter.js", "example:return-this": "npm run build && node build/esm/examples/return-this.js", "example:db": "npm run build && node build/esm/examples/db.js", "example:fn": "npm run build && node build/esm/examples/fn.js", + "example:callbacks": "npm run build && node build/esm/examples/callbacks.js", + "example:streams": "npm run build && node build/esm/examples/streams.js", "example:cjs": "npm run build && node src/examples/cjs/counter.cjs" }, "engines": { diff --git a/src/examples/actors/streamer.ts b/src/examples/actors/streamer.ts new file mode 100644 index 0000000..c0739b5 --- /dev/null +++ b/src/examples/actors/streamer.ts @@ -0,0 +1,37 @@ +import { Readable } from "node:stream"; + +import { actor } from "../../index.js"; + +export class Streamer { + /** Emit `count` rows via a Node.js Readable (objectMode). */ + query(count: number): Readable { + let i = 0; + + return new Readable({ + objectMode: true, + read() { + if (i >= count) { + this.push(null); + + return; + } + const n = i++; + + this.push({ id: n, value: n * n }); + }, + }); + } + + async forEachRow( + count: number, + onRow: (row: { id: number; value: number; }) => void | Promise, + ): Promise { + for (let i = 0; i < count; i++) { + await onRow({ id: i, value: i * i }); + } + + return count; + } +} + +actor(Streamer, import.meta); diff --git a/src/examples/bench.ts b/src/examples/bench.ts new file mode 100644 index 0000000..1055f05 --- /dev/null +++ b/src/examples/bench.ts @@ -0,0 +1,68 @@ +/* eslint-disable no-console, sort-imports */ +import { Readable } from "node:stream"; +import { Worker } from "node:worker_threads"; + +import { Runtime } from "../index.js"; +import { Counter } from "./actors/counter.js"; +import { Streamer } from "./actors/streamer.js"; + +async function time(label: string, fn: () => Promise): Promise { + const start = performance.now(); + + await fn(); + const ms = performance.now() - start; + + console.log(`${label}: ${ms.toFixed(1)}ms`); +} + +const runtime = new Runtime({ workers: 1 }); +const counter = await runtime.spawn(Counter, 0); +const streamer = await runtime.spawn(Streamer); + +const CALLS = 2_000; + +await time(`remote call x${CALLS}`, async () => { + for (let i = 0; i < CALLS; i++) { + await counter.inc(); + } +}); + +await time("callback round-trip x500", async () => { + await streamer.forEachRow(500, async () => undefined); +}); + +await time("stream 5k rows", async () => { + const stream = await streamer.query(5_000) as Readable; + let n = 0; + + for await (const _row of stream) { + n += 1; + } + if (n !== 5_000) throw new Error(`expected 5000, got ${n}`); +}); + +await time("baseline worker postMessage x2000", async () => { + await new Promise((resolve, reject) => { + const worker = new Worker( + ` + const { parentPort } = require("node:worker_threads"); + parentPort.on("message", (msg) => parentPort.postMessage(msg)); + `, + { eval: true }, + ); + let left = CALLS; + + worker.on("message", () => { + left -= 1; + if (left === 0) { + void worker.terminate().then(() => resolve()); + } + }); + worker.on("error", reject); + for (let i = 0; i < CALLS; i++) { + worker.postMessage(i); + } + }); +}); + +await runtime.dispose(); diff --git a/src/examples/callbacks.ts b/src/examples/callbacks.ts new file mode 100644 index 0000000..1460842 --- /dev/null +++ b/src/examples/callbacks.ts @@ -0,0 +1,18 @@ +/* eslint-disable no-console */ +import { Runtime } from "../index.js"; +import { Streamer } from "./actors/streamer.js"; + +const runtime = new Runtime({ workers: 1 }); +const streamer = await runtime.spawn(Streamer); + +try { + const rows: Array<{ id: number; value: number; }> = []; + + const n = await streamer.forEachRow(5, async (row) => { + rows.push(row); + }); + + console.log("rows processed:", n, rows); +} finally { + await runtime.dispose(); +} diff --git a/src/examples/streams.ts b/src/examples/streams.ts new file mode 100644 index 0000000..297ccdd --- /dev/null +++ b/src/examples/streams.ts @@ -0,0 +1,17 @@ +/* eslint-disable no-console */ + +import { Runtime } from "../index.js"; +import { Streamer } from "./actors/streamer.js"; + +const runtime = new Runtime({ workers: 1 }); +const streamer = await runtime.spawn(Streamer); + +try { + const stream = await streamer.query(5); + + for await (const row of stream) { + console.log("stream row", row); + } +} finally { + await runtime.dispose(); +} diff --git a/src/lib/actor-meta.ts b/src/lib/actor-meta.ts index 8c72a0d..5a7dfa1 100644 --- a/src/lib/actor-meta.ts +++ b/src/lib/actor-meta.ts @@ -3,23 +3,35 @@ import { pathToFileURL } from "node:url"; import type { AnyActorClass } from "./types.js"; +/** Actor class with optional module metadata attached by {@link actor}. */ export type ActorClassWithMeta = AnyActorClass & { [ACTOR_META]?: ActorMeta; }; +/** Module location + export name needed to load a class inside a worker. */ export type ActorMeta = { moduleUrl: string; exportName: string; }; +/** + * Module location accepted by {@link actor}. + * Use `import.meta` (ESM), `__filename` / string path (CJS), or `{ url }` / `{ filename }`. + */ export type ActorModuleRef = | ImportMeta | { url: string; } | { filename: string; } | string; +/** Well-known symbol storing {@link ActorMeta} on a bound class. */ export const ACTOR_META = Symbol.for("remote-objects.actorMeta"); +/** + * Resolves a module location to a `file:` / `data:` URL for worker `import()`. + * @param meta - `import.meta`, `__filename`, path string, or `{ url }` / `{ filename }` + * @returns Absolute module URL + */ function resolveModuleUrl(meta: ActorModuleRef): string { if (typeof meta === "string") { if (meta.startsWith("file:") || meta.startsWith("data:")) { @@ -50,6 +62,11 @@ function resolveModuleUrl(meta: ActorModuleRef): string { * * CJS: * actor(Counter, __filename); + * + * @param Class - Actor class to bind + * @param meta - Module location (`import.meta`, `__filename`, etc.) + * @param exportName - Named export to load in the worker (default: `Class.name`) + * @returns The same class (for chaining / re-export) */ export function actor( Class: T, @@ -64,6 +81,11 @@ export function actor( return Class; } +/** + * Reads module metadata previously attached by {@link actor}. + * @param Class - Actor class + * @returns Binding metadata, or `undefined` if the class was never bound + */ export function getActorMeta(Class: AnyActorClass): ActorMeta | undefined { return (Class as ActorClassWithMeta)[ACTOR_META]; } diff --git a/src/lib/index.ts b/src/lib/index.ts index 3bf5f30..87730b8 100644 --- a/src/lib/index.ts +++ b/src/lib/index.ts @@ -1,3 +1,9 @@ +/** + * Public API for `@js-ak/remote-objects`. + * + * Actor-style remote objects on Node.js worker threads — write normal classes, + * bind them with {@link actor}, spawn via {@link Runtime}, call through typed proxies. + */ export type { ActorClass, ActorHandle, @@ -5,6 +11,7 @@ export type { AnyActorClass, DebugEvent, DebugHandler, + DestroyOptions, RuntimeDebug, RuntimeOptions, } from "./types.js"; diff --git a/src/lib/protocol/callback-registry.ts b/src/lib/protocol/callback-registry.ts new file mode 100644 index 0000000..12e606b --- /dev/null +++ b/src/lib/protocol/callback-registry.ts @@ -0,0 +1,121 @@ +import type { CallbackRef } from "./messages.js"; +import { callbackRef } from "./refs.js"; + +/** One registered local function that was sent across the wire. */ +export type CallbackEntry = { + fn: (...args: unknown[]) => unknown; + /** Actor that returned this callback (return-scoped). */ + boundObjectId?: number; + /** Release after the hosting call finishes (arg-scoped). */ + callScoped?: boolean; +}; + +/** Side that owns the real function: main thread or a worker id. */ +export type CallbackOwner = "host" | number; + +/** + * Stores local functions that were sent across the wire as callback_ref. + */ +export class CallbackRegistry { + private readonly entries = new Map(); + private nextId = 1; + private readonly owner: CallbackOwner; + + /** + * @param owner - `"host"` or this worker's id (embedded in issued refs) + */ + constructor(owner: CallbackOwner) { + this.owner = owner; + } + + /** + * Registers a local function and returns a wire {@link CallbackRef}. + * @param fn - Function to invoke when the remote side calls the stub + * @param options - Lifetime: call-scoped and/or bound to an actor objectId + * @returns Wire tag for the remote side + */ + register( + fn: (...args: unknown[]) => unknown, + options?: { boundObjectId?: number; callScoped?: boolean; }, + ): CallbackRef { + const callbackId = this.nextId++; + const entry: CallbackEntry = { fn }; + + if (options?.boundObjectId !== undefined) { + entry.boundObjectId = options.boundObjectId; + } + if (options?.callScoped) { + entry.callScoped = true; + } + + this.entries.set(callbackId, entry); + + return callbackRef(this.owner, callbackId); + } + + /** + * Looks up a registered entry without invoking it. + * @param callbackId - Id from a {@link CallbackRef} + * @returns Entry, or `undefined` if unknown + */ + get(callbackId: number): CallbackEntry | undefined { + return this.entries.get(callbackId); + } + + /** + * Invokes a registered callback (awaits thenables). + * @param callbackId - Id from a {@link CallbackRef} + * @param args - Decoded arguments from the remote side + * @returns Callback result + */ + async invoke(callbackId: number, args: unknown[]): Promise { + const entry = this.entries.get(callbackId); + + if (!entry) { + throw new Error(`Unknown callback ${callbackId}`); + } + + return entry.fn(...args); + } + + /** + * Drops the given callback ids unconditionally. + * @param callbackIds - Ids to remove + */ + release(callbackIds: Iterable): void { + for (const id of callbackIds) { + this.entries.delete(id); + } + } + + /** + * Drops only call-scoped entries among the given ids (after a method returns). + * @param callbackIds - Candidate ids from the finished call + */ + releaseCallScoped(callbackIds: Iterable): void { + for (const id of callbackIds) { + const entry = this.entries.get(id); + + if (entry?.callScoped) { + this.entries.delete(id); + } + } + } + + /** + * Drops callbacks returned by a given actor (on destroy). + * @param objectId - Actor that owned the returned callbacks + */ + releaseBoundToObject(objectId: number): void { + for (const [id, entry] of this.entries) { + if (entry.boundObjectId === objectId) { + this.entries.delete(id); + } + } + } + + /** Removes every registered callback. */ + clear(): void { + this.entries.clear(); + } +} diff --git a/src/lib/protocol/messages.ts b/src/lib/protocol/messages.ts index a41b1dc..3debf7e 100644 --- a/src/lib/protocol/messages.ts +++ b/src/lib/protocol/messages.ts @@ -5,6 +5,10 @@ export type ActorRef = { objectId: number; }; +/** + * Cross-worker actor method call, routed through the parent {@link WorkerNode}. + * Posted by a worker that holds a stub for an actor on another worker. + */ export type BridgeCallMessage = { bridge: "call"; id: number; @@ -14,6 +18,36 @@ export type BridgeCallMessage = { args: unknown[]; }; +/** + * Request to invoke a callback owned by the host or another worker. + * Posted by a worker that received a {@link CallbackRef} stub. + */ +export type BridgeCallbackInvokeMessage = { + bridge: "callback_invoke"; + id: number; + owner: "host" | number; + callbackId: number; + args: unknown[]; +}; + +/** Reply to {@link BridgeCallbackInvokeMessage}. */ +export type BridgeCallbackResultMessage = + | { + bridge: "callback_result"; + id: number; + result: unknown; + } + | { + bridge: "callback_result"; + id: number; + error: { + message: string; + name?: string; + stack?: string; + }; + }; + +/** Reply to {@link BridgeCallMessage}. */ export type BridgeResultMessage = | { bridge: "result"; @@ -30,6 +64,92 @@ export type BridgeResultMessage = }; }; +/** + * Stream control / data messages for {@link StreamRef} bridges. + * `direction` distinguishes owner→consumer (`from_owner`) vs consumer→owner (`to_owner`). + */ +export type BridgeStreamMessage = + | { + bridge: "stream_data"; + owner: "host" | number; + streamId: number; + chunk: unknown; + direction: "to_owner" | "from_owner"; + } + | { + bridge: "stream_end"; + owner: "host" | number; + streamId: number; + direction: "to_owner" | "from_owner"; + } + | { + bridge: "stream_error"; + owner: "host" | number; + streamId: number; + direction: "to_owner" | "from_owner"; + error: { + message: string; + name?: string; + stack?: string; + }; + } + | { + bridge: "stream_pause"; + owner: "host" | number; + streamId: number; + } + | { + bridge: "stream_resume"; + owner: "host" | number; + streamId: number; + } + | { + bridge: "stream_write_ack"; + owner: "host" | number; + streamId: number; + id: number; + } + | { + bridge: "stream_write"; + owner: "host" | number; + streamId: number; + id: number; + chunk: unknown; + } + | { + bridge: "stream_write_end"; + owner: "host" | number; + streamId: number; + id: number; + } + | { + bridge: "stream_write_result"; + owner: "host" | number; + streamId: number; + id: number; + error?: { + message: string; + name?: string; + stack?: string; + }; + } + | { + bridge: "stream_close"; + owner: "host" | number; + streamId: number; + }; + +/** Wire tag for a callable living on the host or a worker. */ +export type CallbackRef = { + type: "callback_ref"; + owner: "host" | number; + callbackId: number; +}; + +/** + * Host → worker command messages (register / create / call / destroy / callbacks). + * Each carries a correlation `id` answered by {@link ProtocolResponse}. + */ export type ProtocolMessage = | { command: "register"; @@ -56,12 +176,25 @@ export type ProtocolMessage = command: "destroy"; id: number; objectId: number; + close?: boolean; } | { command: "close_all"; id: number; + } + | { + command: "callback_invoke"; + id: number; + callbackId: number; + args: unknown[]; + } + | { + command: "callback_release"; + id: number; + callbackIds: number[]; }; +/** Worker → host reply to a {@link ProtocolMessage}. */ export type ProtocolResponse = | { id: number; @@ -75,3 +208,48 @@ export type ProtocolResponse = stack?: string; }; }; + +/** Wire tag for a Node.js stream living on the host or a worker. */ +export type StreamRef = { + type: "stream_ref"; + owner: "host" | number; + streamId: number; + mode: "readable" | "writable" | "duplex"; + objectMode: boolean; +}; + +/** Structured-clone-safe error payload for wire messages. */ +export type WireError = { + message: string; + name?: string; + stack?: string; +}; + +/** + * Rebuilds an `Error` from a wire payload (preserves name/stack when present). + * @param error - Serialized error fields + * @returns Reconstructed Error instance + */ +export function fromWireError(error: WireError): Error { + const err = new Error(error.message); + + if (error.name) err.name = error.name; + if (error.stack) err.stack = error.stack; + + return err; +} + +/** + * Serializes any thrown value into a {@link WireError} for postMessage. + * @param err - Caught error or arbitrary throw value + * @returns Structured-clone-safe error payload + */ +export function toWireError(err: unknown): WireError { + const error = err instanceof Error ? err : new Error(String(err)); + + return { + message: error.message, + name: error.name, + ...(error.stack ? { stack: error.stack } : {}), + }; +} diff --git a/src/lib/protocol/plain.ts b/src/lib/protocol/plain.ts new file mode 100644 index 0000000..b31a2c0 --- /dev/null +++ b/src/lib/protocol/plain.ts @@ -0,0 +1,26 @@ +/** + * Builds a clear error message when `postMessage` / structured clone fails. + * @param err - Underlying clone/postMessage error + * @returns Human-readable error string for wrapping + */ +export function cloneErrorMessage(err: unknown): string { + const message = err instanceof Error ? err.message : String(err); + + return ( + "Value is not structured-clone compatible for worker_threads " + + `(and is not an actor/callback/stream ref). ${message}` + ); +} + +/** + * True for JSON-like plain objects (not class instances, arrays, or null). + * Used by the serializer to decide what to deep-walk. + * @param value - Value under test + * @returns Whether `value` is a plain object + */ +export function isPlainObject(value: unknown): value is Record { + if (typeof value !== "object" || value === null) return false; + const proto = Object.getPrototypeOf(value); + + return proto === Object.prototype || proto === null; +} diff --git a/src/lib/protocol/refs.ts b/src/lib/protocol/refs.ts index f5946c6..e1d5f94 100644 --- a/src/lib/protocol/refs.ts +++ b/src/lib/protocol/refs.ts @@ -1,13 +1,45 @@ -import type { ActorRef } from "./messages.js"; +import type { + ActorRef, CallbackRef, StreamRef, +} from "./messages.js"; +/** + * Builds an {@link ActorRef} wire tag. + * @param workerId - Worker that owns the actor + * @param objectId - Actor id on that worker + * @returns Wire actor reference + */ export function actorRef(workerId: number, objectId: number): ActorRef { return { objectId, type: "actor_ref", workerId }; } +/** + * Builds a {@link CallbackRef} wire tag. + * @param owner - `"host"` or owning worker id + * @param callbackId - Id in that side's {@link CallbackRegistry} + * @returns Wire callback reference + */ +export function callbackRef( + owner: "host" | number, + callbackId: number, +): CallbackRef { + return { callbackId, owner, type: "callback_ref" }; +} + +/** + * Formats a stable actor id string (`"workerId:objectId"`). + * @param workerId - Worker index + * @param objectId - Object id on that worker + * @returns Stable actor id for logs and debug events + */ export function formatActorId(workerId: number, objectId: number): string { return `${workerId}:${objectId}`; } +/** + * Type guard for {@link ActorRef}. + * @param value - Value under test + * @returns Whether `value` is an actor wire tag + */ export function isActorRef(value: unknown): value is ActorRef { return ( typeof value === "object" @@ -17,3 +49,68 @@ export function isActorRef(value: unknown): value is ActorRef { && typeof (value as ActorRef).objectId === "number" ); } + +/** + * Type guard for {@link CallbackRef}. + * @param value - Value under test + * @returns Whether `value` is a callback wire tag + */ +export function isCallbackRef(value: unknown): value is CallbackRef { + return ( + typeof value === "object" + && value !== null + && (value as CallbackRef).type === "callback_ref" + && typeof (value as CallbackRef).callbackId === "number" + && ( + (value as CallbackRef).owner === "host" + || typeof (value as CallbackRef).owner === "number" + ) + ); +} + +/** + * Type guard for {@link StreamRef}. + * @param value - Value under test + * @returns Whether `value` is a stream wire tag + */ +export function isStreamRef(value: unknown): value is StreamRef { + return ( + typeof value === "object" + && value !== null + && (value as StreamRef).type === "stream_ref" + && typeof (value as StreamRef).streamId === "number" + && typeof (value as StreamRef).objectMode === "boolean" + && ( + (value as StreamRef).owner === "host" + || typeof (value as StreamRef).owner === "number" + ) + && ( + (value as StreamRef).mode === "readable" + || (value as StreamRef).mode === "writable" + || (value as StreamRef).mode === "duplex" + ) + ); +} + +/** + * Builds a {@link StreamRef} wire tag. + * @param owner - `"host"` or owning worker id + * @param streamId - Id in that side's {@link StreamBridge} + * @param mode - Readable / writable / duplex + * @param objectMode - Whether chunks are objects (vs Buffers/strings) + * @returns Wire stream reference + */ +export function streamRef( + owner: "host" | number, + streamId: number, + mode: StreamRef["mode"], + objectMode: boolean, +): StreamRef { + return { + mode, + objectMode, + owner, + streamId, + type: "stream_ref", + }; +} diff --git a/src/lib/protocol/serializer.ts b/src/lib/protocol/serializer.ts index 51d4dee..a0651fa 100644 --- a/src/lib/protocol/serializer.ts +++ b/src/lib/protocol/serializer.ts @@ -1,43 +1,211 @@ -import { actorRef, isActorRef } from "./refs.js"; -import type { ActorRef } from "./messages.js"; +import type { + ActorRef, CallbackRef, StreamRef, +} from "./messages.js"; +import { + isActorRef, isCallbackRef, isStreamRef, +} from "./refs.js"; +import { isPlainObject } from "./plain.js"; +/** Hooks used when turning wire tags back into local values / stubs. */ +export type DecodeContext = { + resolveActorRef: (ref: ActorRef) => unknown; + resolveCallbackRef?: (ref: CallbackRef) => unknown; + resolveStreamRef?: (ref: StreamRef) => unknown; +}; + +/** + * Hooks used when turning local values into wire tags before `postMessage`. + * Actors, callbacks, and streams are replaced with refs; arrays/plain objects are walked. + */ export type EncodeContext = { - workerId: number; - currentObject: object; - currentObjectId: number; + workerId?: number; + currentObject?: object; + currentObjectId?: number; /** Optional map of known live actors in this worker. */ actors?: Map; + /** Host-side: resolve actor proxies to handles. */ + resolveProxy?: (value: unknown) => { workerId: number; objectId: number; } | undefined; + registerCallback?: (fn: (...args: never[]) => unknown) => CallbackRef; + registerStream?: (value: object) => StreamRef | undefined; }; /** - * Encodes method return values before crossing the worker boundary. - * `return this` becomes an actor_ref, not a state dump. + * Encodes / decodes values crossing the worker boundary. + * Deep-walks arrays and plain objects; tags actors, callbacks, and streams. */ export class Serializer { - encode(value: unknown, context: EncodeContext): unknown { - if (value === context.currentObject) { - return actorRef(context.workerId, context.currentObjectId); + /** + * Encodes a value for the wire (deep-walk + refs). + * @param value - Return value or argument tree + * @param context - Actor / callback / stream registration hooks + * @returns Value safe to send via `postMessage` + */ + encode(value: unknown, context: EncodeContext = {}): unknown { + return this.walkEncode(value, context, new WeakSet()); + } + + /** + * Decodes a wire value into local actors, callback stubs, or stream proxies. + * @param value - Value received over `postMessage` + * @param context - Resolvers for each ref kind + * @returns Local value or stub/proxy + */ + decode(value: unknown, context: DecodeContext): unknown { + return this.walkDecode(value, context, new WeakSet()); + } + + /** + * Recursive encode walk. Throws on circular plain-object graphs. + * @param value - Current node + * @param context - Encode hooks + * @param seen - Cycle detection set + * @returns Encoded node + */ + private walkEncode( + value: unknown, + context: EncodeContext, + seen: WeakSet, + ): unknown { + if (value === null || typeof value !== "object") { + if (typeof value === "function") { + if (!context.registerCallback) { + throw new Error( + "Functions cannot cross the worker boundary without callback support", + ); + } + + return context.registerCallback( + value as (...args: never[]) => unknown, + ); + } + + return value; + } + + if ( + context.currentObject !== undefined + && value === context.currentObject + && context.currentObjectId !== undefined + && context.workerId !== undefined + ) { + return { + objectId: context.currentObjectId, + type: "actor_ref", + workerId: context.workerId, + } satisfies ActorRef; } if (context.actors) { - const objectId = context.actors.get(value as object); + const objectId = context.actors.get(value); - if (objectId !== undefined) { - return actorRef(context.workerId, objectId); + if (objectId !== undefined && context.workerId !== undefined) { + return { + objectId, + type: "actor_ref", + workerId: context.workerId, + } satisfies ActorRef; } } - return value; + const proxyHandle = context.resolveProxy?.(value); + + if (proxyHandle) { + return { + objectId: proxyHandle.objectId, + type: "actor_ref", + workerId: proxyHandle.workerId, + } satisfies ActorRef; + } + + if (context.registerStream) { + const ref = context.registerStream(value); + + if (ref) return ref; + } + + if (seen.has(value)) { + throw new Error( + "Circular references are not supported when encoding values for remote calls", + ); + } + + if (Array.isArray(value)) { + seen.add(value); + + return value.map((item) => this.walkEncode(item, context, seen)); + } + + if (!isPlainObject(value)) { + return value; + } + + seen.add(value); + const out: Record = {}; + + for (const [key, item] of Object.entries(value)) { + out[key] = this.walkEncode(item, context, seen); + } + + return out; } - decode( + /** + * Recursive decode walk for arrays and plain objects. + * @param value - Current wire node + * @param context - Decode hooks + * @param seen - Cycle detection set + * @returns Decoded node + */ + private walkDecode( value: unknown, - resolveRef: (ref: ActorRef) => unknown, + context: DecodeContext, + seen: WeakSet, ): unknown { if (isActorRef(value)) { - return resolveRef(value); + return context.resolveActorRef(value); + } + + if (isCallbackRef(value)) { + if (!context.resolveCallbackRef) { + throw new Error("Callback refs are not supported in this context"); + } + + return context.resolveCallbackRef(value); + } + + if (isStreamRef(value)) { + if (!context.resolveStreamRef) { + throw new Error("Stream refs are not supported in this context"); + } + + return context.resolveStreamRef(value); + } + + if (value === null || typeof value !== "object") { + return value; + } + + if (seen.has(value)) { + return value; + } + + if (Array.isArray(value)) { + seen.add(value); + + return value.map((item) => this.walkDecode(item, context, seen)); + } + + if (!isPlainObject(value)) { + return value; + } + + seen.add(value); + const out: Record = {}; + + for (const [key, item] of Object.entries(value)) { + out[key] = this.walkDecode(item, context, seen); } - return value; + return out; } } diff --git a/src/lib/protocol/stream-bridge.ts b/src/lib/protocol/stream-bridge.ts new file mode 100644 index 0000000..fa5be96 --- /dev/null +++ b/src/lib/protocol/stream-bridge.ts @@ -0,0 +1,592 @@ +import { + Duplex, + Readable, + Writable, + isReadable as nodeIsReadable, + isWritable as nodeIsWritable, +} from "node:stream"; + +import type { StreamRef } from "./messages.js"; +import { streamRef } from "./refs.js"; + +/** Side that owns the real Node.js stream: main thread or a worker id. */ +export type StreamOwner = "host" | number; + +/** + * Low-level send hooks used by {@link StreamBridge} to emit bridge stream messages. + * Implemented differently on host ({@link createHostStreamTransport}) vs worker. + */ +export type StreamTransport = { + sendData: (chunk: unknown) => void; + sendEnd: () => void; + sendError: (err: Error) => void; + sendPause: () => void; + sendResume: () => void; + sendWrite: (id: number, chunk: unknown) => void; + sendWriteEnd: (id: number) => void; + sendWriteResult: (id: number, error?: Error) => void; + sendClose: () => void; +}; + +/** Pending writable acknowledgements keyed by write id. */ +type WritePending = Map< + number, + { resolve: () => void; reject: (err: Error) => void; } +>; + +/** Bookkeeping for one local or proxy stream in this bridge. */ +type LocalStreamEntry = { + ref: StreamRef; + local: Readable | Writable | Duplex; + proxy?: Readable | Writable | Duplex; + writePending: WritePending; + closed: boolean; + /** True when we own the real stream (encoded locally). */ + isOwner: boolean; +}; + +/** + * @param value - Candidate value to test + * @returns Whether `value` is a Node.js Readable (including Duplex) + */ +function isNodeReadable(value: object): value is Readable { + return Boolean(nodeIsReadable(value as Readable)); +} + +/** + * @param value - Candidate value to test + * @returns Whether `value` is a Node.js Writable (including Duplex) + */ +function isNodeWritable(value: object): value is Writable { + return Boolean(nodeIsWritable(value as Writable)); +} + +/** + * Bridges Node.js streams across worker_threads via stream_ref messages. + */ +export class StreamBridge { + private readonly entries = new Map(); + private nextId = 1; + private readonly owner: StreamOwner; + private readonly createTransport: (ref: StreamRef) => StreamTransport; + + /** + * @param owner - `"host"` or this worker's id (embedded in issued refs) + * @param createTransport - Factory for per-ref send hooks + */ + constructor( + owner: StreamOwner, + createTransport: (ref: StreamRef) => StreamTransport, + ) { + this.owner = owner; + this.createTransport = createTransport; + } + + /** + * If `value` is a Node stream, registers it and returns a {@link StreamRef}. + * Starts forwarding owner events (`data` / `end` / `error`) over the transport. + * @param value - Candidate value during encode + * @returns Wire ref, or `undefined` when not a stream + */ + tryRegisterLocal(value: object): StreamRef | undefined { + const mode = detectStreamMode(value); + + if (!mode) return undefined; + + for (const entry of this.entries.values()) { + if (entry.local === value || entry.proxy === value) { + return entry.ref; + } + } + + const ref = streamRef( + this.owner, + this.nextId++, + mode, + detectObjectMode(value, mode), + ); + const entry: LocalStreamEntry = { + closed: false, + isOwner: true, + local: value as Readable | Writable | Duplex, + ref, + writePending: new Map(), + }; + + this.entries.set(entryKey(ref.owner, ref.streamId), entry); + this.attachOwnerListeners(entry); + + return ref; + } + + /** + * Creates (or reuses) a local Node stream proxy for a remote {@link StreamRef}. + * @param ref - Wire tag from decode + * @returns Local Readable, Writable, or Duplex proxy + */ + createProxy(ref: StreamRef): Readable | Writable | Duplex { + const key = entryKey(ref.owner, ref.streamId); + const existing = this.entries.get(key); + + if (existing?.proxy) { + return existing.proxy; + } + + const writePending: WritePending = new Map(); + const entry: LocalStreamEntry = { + closed: false, + isOwner: false, + local: undefined as unknown as Readable, + ref, + writePending, + }; + + this.entries.set(key, entry); + + const transport = this.createTransport(ref); + let proxy: Readable | Writable | Duplex; + + if (ref.mode === "readable") { + proxy = this.createReadableProxy(ref, transport); + } else if (ref.mode === "writable") { + proxy = this.createWritableProxy(ref, transport, writePending); + } else { + proxy = this.createDuplexProxy(ref, transport, writePending); + } + + entry.local = proxy; + entry.proxy = proxy; + + return proxy; + } + + /** + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + * @returns Whether this bridge tracks the given stream + */ + has(owner: StreamOwner, streamId: number): boolean { + return this.entries.has(entryKey(owner, streamId)); + } + + /** + * Looks up an entry by owner + streamId. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + * @returns Entry, or `undefined` when not tracked + */ + private get( + owner: StreamOwner, + streamId: number, + ): LocalStreamEntry | undefined { + return this.entries.get(entryKey(owner, streamId)); + } + + /** + * Forwards local owner stream events to the remote consumer via transport. + * @param entry - Newly registered owner-side stream + */ + private attachOwnerListeners(entry: LocalStreamEntry): void { + const transport = this.createTransport(entry.ref); + const { local, ref } = entry; + + if (ref.mode === "readable" || ref.mode === "duplex") { + const readable = local as Readable; + + readable.on("data", (chunk: unknown) => { + if (entry.closed) return; + transport.sendData(chunk); + }); + readable.on("end", () => { + if (entry.closed) return; + transport.sendEnd(); + }); + readable.on("error", (err: Error) => { + if (entry.closed) return; + transport.sendError(err); + }); + } + + if (ref.mode === "writable" || ref.mode === "duplex") { + const writable = local as Writable; + + writable.on("error", (err: Error) => { + if (entry.closed) return; + transport.sendError(err); + }); + } + } + + /** + * Builds a Readable proxy that pulls resume/pause across the wire. + * @param ref - Wire stream ref (objectMode, etc.) + * @param transport - Send hooks for this ref + * @returns Local Readable proxy + */ + private createReadableProxy( + ref: StreamRef, + transport: StreamTransport, + ): Readable { + const readable = new Readable({ + objectMode: ref.objectMode, + read: () => { + transport.sendResume(); + }, + }); + + readable.on("pause", () => { + transport.sendPause(); + }); + readable.on("close", () => { + transport.sendClose(); + }); + + return readable; + } + + /** + * Builds a Writable proxy; each write waits for a remote ack. + * @param ref - Wire stream ref (objectMode, etc.) + * @param transport - Send hooks for this ref + * @param pending - Map of write-id acknowledgements + * @returns Local Writable proxy + */ + private createWritableProxy( + ref: StreamRef, + transport: StreamTransport, + pending: WritePending, + ): Writable { + let nextWriteId = 1; + + const writable = new Writable({ + final: (cb) => { + const id = nextWriteId++; + + pending.set(id, { + reject: (err) => cb(err), + resolve: () => cb(), + }); + transport.sendWriteEnd(id); + }, + objectMode: ref.objectMode, + write: (chunk, _enc, cb) => { + const id = nextWriteId++; + + pending.set(id, { + reject: (err) => cb(err), + resolve: () => cb(), + }); + transport.sendWrite(id, chunk); + }, + }); + + writable.on("close", () => { + transport.sendClose(); + }); + + return writable; + } + + /** + * Builds a Duplex proxy (readable + writable halves over the wire). + * @param ref - Wire stream ref (objectMode, etc.) + * @param transport - Send hooks for this ref + * @param pending - Map of write-id acknowledgements + * @returns Local Duplex proxy + */ + private createDuplexProxy( + ref: StreamRef, + transport: StreamTransport, + pending: WritePending, + ): Duplex { + let nextWriteId = 1; + + const duplex = new Duplex({ + final: (cb) => { + const id = nextWriteId++; + + pending.set(id, { + reject: (err) => cb(err), + resolve: () => cb(), + }); + transport.sendWriteEnd(id); + }, + objectMode: ref.objectMode, + read: () => { + transport.sendResume(); + }, + write: (chunk, _enc, cb) => { + const id = nextWriteId++; + + pending.set(id, { + reject: (err) => cb(err), + resolve: () => cb(), + }); + transport.sendWrite(id, chunk); + }, + }); + + duplex.on("pause", () => { + transport.sendPause(); + }); + duplex.on("close", () => { + transport.sendClose(); + }); + + return duplex; + } + + /** + * Pushes a chunk into a local readable proxy (consumer side). + * Sends pause when the proxy buffer is full. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + * @param chunk - Decoded data chunk + */ + onRemoteData(owner: StreamOwner, streamId: number, chunk: unknown): void { + const entry = this.get(owner, streamId); + + if (!entry || entry.closed || !entry.proxy) return; + const readable = entry.proxy as Readable; + + if (typeof readable.push === "function") { + const ok = readable.push(chunk); + + if (!ok) { + this.createTransport(entry.ref).sendPause(); + } + } + } + + /** + * Signals EOF on a local readable proxy. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + */ + onRemoteEnd(owner: StreamOwner, streamId: number): void { + const entry = this.get(owner, streamId); + + if (!entry || entry.closed || !entry.proxy) return; + const readable = entry.proxy as Readable; + + if (typeof readable.push === "function") { + readable.push(null); + } + } + + /** + * Destroys local/proxy streams after a remote error. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + * @param err - Error from the remote side + */ + onRemoteError(owner: StreamOwner, streamId: number, err: Error): void { + const entry = this.get(owner, streamId); + + if (!entry || entry.closed) return; + entry.closed = true; + entry.local.destroy(err); + if (entry.proxy && entry.proxy !== entry.local) { + entry.proxy.destroy(err); + } + } + + /** + * Applies backpressure on the owner-side readable. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + */ + onPause(owner: StreamOwner, streamId: number): void { + const entry = this.get(owner, streamId); + + if (!entry || !entry.isOwner) return; + const readable = entry.local as Readable; + + if (typeof readable.pause === "function") { + readable.pause(); + } + } + + /** + * Resumes the owner-side readable after consumer drain. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + */ + onResume(owner: StreamOwner, streamId: number): void { + const entry = this.get(owner, streamId); + + if (!entry || !entry.isOwner) return; + const readable = entry.local as Readable; + + if (typeof readable.resume === "function") { + readable.resume(); + } + } + + /** + * Owner-side: writes a chunk from a remote writable proxy, then acks. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + * @param id - Correlation id for {@link onWriteResult} + * @param chunk - Chunk to write to the local writable + */ + async onWrite( + owner: StreamOwner, + streamId: number, + id: number, + chunk: unknown, + ): Promise { + const entry = this.get(owner, streamId); + + if (!entry || !entry.isOwner) { + throw new Error(`Unknown stream ${owner}:${streamId}`); + } + + const writable = entry.local as Writable; + const transport = this.createTransport(entry.ref); + + try { + await new Promise((resolve, reject) => { + writable.write(chunk as never, (err) => { + if (err) reject(err); + else resolve(); + }); + }); + transport.sendWriteResult(id); + } catch (err) { + transport.sendWriteResult( + id, + err instanceof Error ? err : new Error(String(err)), + ); + } + } + + /** + * Owner-side: ends the local writable after the remote proxy calls `end`. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + * @param id - Correlation id for {@link onWriteResult} + */ + async onWriteEnd( + owner: StreamOwner, + streamId: number, + id: number, + ): Promise { + const entry = this.get(owner, streamId); + + if (!entry || !entry.isOwner) { + throw new Error(`Unknown stream ${owner}:${streamId}`); + } + + const writable = entry.local as Writable; + const transport = this.createTransport(entry.ref); + + try { + await new Promise((resolve, reject) => { + writable.end((err?: Error | null) => { + if (err) reject(err); + else resolve(); + }); + }); + transport.sendWriteResult(id); + } catch (err) { + transport.sendWriteResult( + id, + err instanceof Error ? err : new Error(String(err)), + ); + } + } + + /** + * Resolves/rejects a pending writable proxy write by correlation id. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + * @param id - Correlation id from the pending write + * @param error - Optional error when the remote write failed + */ + onWriteResult( + owner: StreamOwner, + streamId: number, + id: number, + error?: Error, + ): void { + const entry = this.get(owner, streamId); + const waiter = entry?.writePending.get(id); + + if (!waiter) return; + entry?.writePending.delete(id); + if (error) waiter.reject(error); + else waiter.resolve(); + } + + /** + * Force-destroys a tracked stream and drops its entry. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + */ + onClose(owner: StreamOwner, streamId: number): void { + const key = entryKey(owner, streamId); + const entry = this.entries.get(key); + + if (!entry || entry.closed) return; + entry.closed = true; + entry.local.destroy(); + if (entry.proxy && entry.proxy !== entry.local) { + entry.proxy.destroy(); + } + this.entries.delete(key); + } + + /** Closes every stream tracked by this bridge (runtime dispose). */ + closeAll(): void { + for (const entry of [...this.entries.values()]) { + this.onClose(entry.ref.owner, entry.ref.streamId); + } + } +} + +/** + * Reads objectMode flags from a Node stream for the given mode. + * @param value - Local stream instance + * @param mode - Detected readable / writable / duplex + * @returns Whether the stream uses object mode for that side + */ +function detectObjectMode(value: object, mode: StreamRef["mode"]): boolean { + const stream = value as { + readableObjectMode?: boolean; + writableObjectMode?: boolean; + }; + + if (mode === "readable") return Boolean(stream.readableObjectMode); + if (mode === "writable") return Boolean(stream.writableObjectMode); + + return Boolean(stream.readableObjectMode || stream.writableObjectMode); +} + +/** + * Composite map key so host and worker stream ids do not collide. + * @param owner - Stream owner (`"host"` or worker id) + * @param streamId - Stream id within that owner + * @returns Key of the form `owner:streamId` + */ +function entryKey(owner: StreamOwner, streamId: number): string { + return `${owner}:${streamId}`; +} + +/** + * Detects whether a value is a Node.js stream and which mode it has. + * @param value - Candidate during encode + * @returns `"readable"` | `"writable"` | `"duplex"`, or `undefined` + */ +export function detectStreamMode( + value: object, +): StreamRef["mode"] | undefined { + const readable = isNodeReadable(value); + const writable = isNodeWritable(value); + + if (readable && writable) return "duplex"; + if (readable) return "readable"; + if (writable) return "writable"; + + return undefined; +} diff --git a/src/lib/proxy/proxy.ts b/src/lib/proxy/proxy.ts index 2dfb8b4..6bc4ee8 100644 --- a/src/lib/proxy/proxy.ts +++ b/src/lib/proxy/proxy.ts @@ -1,5 +1,11 @@ import type { ActorHandle } from "../types.js"; +/** + * Routes a proxy method invocation to the owning {@link WorkerNode}. + * @param objectId - Actor id on that worker + * @param method - Method name + * @param args - Encoded-ready argument list (host still encodes before send) + */ export type CallHandler = ( objectId: number, method: string, @@ -11,6 +17,10 @@ const handles = new WeakMap(); /** * Builds a Proxy that looks like a local instance but routes * every method call through the runtime message protocol. + * + * @param handle - Stable actor identity + * @param call - Handler that performs the remote call + * @returns Opaque proxy; use {@link getActorHandle} to recover the handle */ export function createProxy( handle: ActorHandle, @@ -36,6 +46,11 @@ export function createProxy( return proxy; } +/** + * Returns the {@link ActorHandle} for a proxy created by this runtime, if any. + * @param value - Suspected actor proxy + * @returns Handle, or `undefined` when `value` is not a known proxy + */ export function getActorHandle(value: unknown): ActorHandle | undefined { if (typeof value !== "object" || value === null) return undefined; diff --git a/src/lib/runtime/host-stream-transport.ts b/src/lib/runtime/host-stream-transport.ts new file mode 100644 index 0000000..1743239 --- /dev/null +++ b/src/lib/runtime/host-stream-transport.ts @@ -0,0 +1,109 @@ +/* eslint-disable sort-imports */ +import type { BridgeStreamMessage, StreamRef } from "../protocol/messages.js"; +import { toWireError } from "../protocol/messages.js"; +import type { StreamTransport } from "../protocol/stream-bridge.js"; +import type { StreamRouter } from "./stream-router.js"; +import type { WorkerNode } from "./worker-node.js"; + +/** + * Builds host-side {@link StreamTransport} hooks for a single {@link StreamRef}. + * Routes owner events to subscribed workers and consumer control messages to the owner. + * + * @param getWorkers - Snapshot accessor for the runtime's worker pool + * @param streamRouter - Shared subscription table + * @param ref - Stream being transported + * @returns Transport callbacks used by {@link StreamBridge} + */ +export function createHostStreamTransport( + getWorkers: () => WorkerNode[], + streamRouter: StreamRouter, + ref: StreamRef, +): StreamTransport { + const postToConsumers = (msg: BridgeStreamMessage) => { + for (const consumer of streamRouter.consumers(ref.owner, ref.streamId)) { + if (consumer === "host") continue; + getWorkers()[consumer as number]?.postToWorker(msg); + } + }; + + const postToOwner = (msg: BridgeStreamMessage) => { + if (ref.owner === "host") return; + getWorkers()[ref.owner]?.postToWorker(msg); + }; + + return { + sendClose: () => { + postToOwner({ + bridge: "stream_close", + owner: ref.owner, + streamId: ref.streamId, + }); + }, + sendData: (chunk) => { + postToConsumers({ + bridge: "stream_data", + chunk, + direction: "from_owner", + owner: ref.owner, + streamId: ref.streamId, + }); + }, + sendEnd: () => { + postToConsumers({ + bridge: "stream_end", + direction: "from_owner", + owner: ref.owner, + streamId: ref.streamId, + }); + }, + sendError: (err) => { + postToConsumers({ + bridge: "stream_error", + direction: "from_owner", + error: toWireError(err), + owner: ref.owner, + streamId: ref.streamId, + }); + }, + sendPause: () => { + postToOwner({ + bridge: "stream_pause", + owner: ref.owner, + streamId: ref.streamId, + }); + }, + sendResume: () => { + postToOwner({ + bridge: "stream_resume", + owner: ref.owner, + streamId: ref.streamId, + }); + }, + sendWrite: (id, chunk) => { + postToOwner({ + bridge: "stream_write", + chunk, + id, + owner: ref.owner, + streamId: ref.streamId, + }); + }, + sendWriteEnd: (id) => { + postToOwner({ + bridge: "stream_write_end", + id, + owner: ref.owner, + streamId: ref.streamId, + }); + }, + sendWriteResult: (id, error) => { + postToConsumers({ + bridge: "stream_write_result", + id, + owner: ref.owner, + streamId: ref.streamId, + ...(error ? { error: toWireError(error) } : {}), + }); + }, + }; +} diff --git a/src/lib/runtime/module-dir.ts b/src/lib/runtime/module-dir.ts index f449b75..0d68db1 100644 --- a/src/lib/runtime/module-dir.ts +++ b/src/lib/runtime/module-dir.ts @@ -1,7 +1,10 @@ import { fileURLToPath } from "node:url"; import path from "node:path"; -/** Directory of this compiled module (CJS emit patched in scripts/fix-cjs-import-meta.js). */ +/** + * Directory of this compiled module (CJS emit patched in scripts/fix-cjs-import-meta.js). + * @returns Absolute directory path of this file + */ export function getModuleDir(): string { // eslint-disable-next-line @typescript-eslint/ban-ts-comment -- CJS rejects import.meta; postbuild rewrites emit // @ts-ignore diff --git a/src/lib/runtime/registry.ts b/src/lib/runtime/registry.ts index b92c76d..30fbe47 100644 --- a/src/lib/runtime/registry.ts +++ b/src/lib/runtime/registry.ts @@ -1,18 +1,30 @@ import type { AnyActorClass } from "../types.js"; +/** Module binding recorded for a registered actor class. */ type ClassRegistration = { className: string; moduleUrl: string; exportName: string; }; +/** Host-side registry entry: class constructor + load metadata. */ type Entry = ClassRegistration & { Class: AnyActorClass; }; +/** + * Host-side map of actor class name → module URL / export. + * Used to know which classes are already registered before spawn. + */ export class Registry { private readonly classes = new Map(); + /** + * Records a class binding (does not talk to workers by itself). + * @param Class - Actor class + * @param moduleUrl - Absolute URL workers use to `import()` the module + * @param exportName - Named export inside that module (default: `Class.name`) + */ register( Class: AnyActorClass, moduleUrl: string, @@ -26,10 +38,18 @@ export class Registry { }); } + /** + * @param name - Class name key + * @returns Entry if registered, otherwise `undefined` + */ get(name: string): Entry | undefined { return this.classes.get(name); } + /** + * @param name - Class name key + * @returns Whether the class is already recorded + */ has(name: string): boolean { return this.classes.has(name); } diff --git a/src/lib/runtime/runtime.ts b/src/lib/runtime/runtime.ts index 16992a3..d234302 100644 --- a/src/lib/runtime/runtime.ts +++ b/src/lib/runtime/runtime.ts @@ -1,18 +1,29 @@ +/* eslint-disable sort-imports */ import type { ActorProxy, AnyActorClass, DebugHandler, + DestroyOptions, RuntimeOptions, } from "../types.js"; import { formatActorId } from "../protocol/refs.js"; import { getActorHandle } from "../proxy/proxy.js"; import { getActorMeta } from "../actor-meta.js"; +import { CallbackRegistry } from "../protocol/callback-registry.js"; +import { StreamBridge } from "../protocol/stream-bridge.js"; +import { createHostStreamTransport } from "./host-stream-transport.js"; import { Registry } from "./registry.js"; import { Scheduler } from "./scheduler.js"; +import { StreamRouter } from "./stream-router.js"; import { WorkerNode } from "./worker-node.js"; +/** + * Normalizes {@link RuntimeOptions.debug} into a single event handler. + * @param debug - Boolean, function, or `{ onEvent }` form + * @returns Handler to invoke, or `undefined` when debug is off + */ function resolveDebug(debug: RuntimeOptions["debug"]): DebugHandler | undefined { if (!debug) return undefined; @@ -29,13 +40,23 @@ function resolveDebug(debug: RuntimeOptions["debug"]): DebugHandler | undefined return debug.onEvent; } +/** + * Owns a pool of worker threads, schedules sticky actors, and exposes + * typed proxies for remote method calls (including callbacks and streams). + */ export class Runtime { private readonly registry = new Registry(); private readonly workers: WorkerNode[]; private readonly scheduler: Scheduler; private readonly onDebug?: DebugHandler; + private readonly hostCallbacks = new CallbackRegistry("host"); + private readonly streamRouter = new StreamRouter(); + private readonly hostStreams: StreamBridge; private disposed = false; + /** + * @param options - Pool size, debug hooks, and optional call timeout + */ constructor(options: RuntimeOptions = {}) { const count = options.workers ?? 1; const onDebug = resolveDebug(options.debug); @@ -45,16 +66,17 @@ export class Runtime { const workers: WorkerNode[] = []; this.workers = workers; + this.hostStreams = new StreamBridge("host", (ref) => + createHostStreamTransport(() => workers, this.streamRouter, ref), + ); for (let id = 0; id < count; id++) { - const nodeOptions: { - id: number; - getWorker: (workerId: number) => WorkerNode | undefined; - onDebug?: DebugHandler; - callTimeoutMs?: number; - } = { + const nodeOptions: ConstructorParameters[0] = { getWorker: (workerId) => workers[workerId], + hostCallbacks: this.hostCallbacks, + hostStreams: this.hostStreams, id, + streamRouter: this.streamRouter, }; if (onDebug) nodeOptions.onDebug = onDebug; @@ -67,6 +89,10 @@ export class Runtime { this.scheduler = new Scheduler(this.workers); } + /** + * Throws if the runtime has already been disposed. + * @param action - Human-readable action name for the error message + */ private assertOpen(action: string): void { if (this.disposed) { throw new Error(`Runtime is disposed; cannot ${action}`); @@ -77,6 +103,8 @@ export class Runtime { * Registers a class for workers. * Bind in the actor module: `actor(Counter, import.meta)` (ESM) * or `actor(Counter, __filename)` (CJS). + * + * @param Class - Actor class previously bound with {@link actor} */ async register(Class: C): Promise { this.assertOpen("register"); @@ -97,6 +125,14 @@ export class Runtime { ); } + /** + * Spawns an actor on a worker (round-robin) and returns a typed proxy. + * Auto-registers the class on first use if needed. + * + * @param Class - Actor class + * @param args - Constructor arguments (structured-clone + refs) + * @returns Typed proxy; all methods are async from the caller side + */ async spawn( Class: C, ...args: ConstructorParameters @@ -111,8 +147,12 @@ export class Runtime { /** * Removes an actor from its worker. Further method calls on the proxy fail. + * By default calls `dispose`/`close` on the actor first. + * + * @param proxy - Proxy returned by {@link spawn} + * @param options - Pass `{ close: false }` to skip actor cleanup */ - async destroy(proxy: object): Promise { + async destroy(proxy: object, options?: DestroyOptions): Promise { this.assertOpen("destroy"); const handle = getActorHandle(proxy); @@ -128,12 +168,14 @@ export class Runtime { ); } - await worker.destroy(handle.objectId); + await worker.destroy(handle.objectId, options); } /** * Graceful shutdown: stop new work, optionally call dispose/close on actors, * wait for in-flight calls, then terminate workers. + * + * @param options - Pass `{ closeActors: false }` to skip per-actor close */ async dispose(options?: { closeActors?: boolean; }): Promise { if (this.disposed) return; @@ -151,5 +193,9 @@ export class Runtime { }), ), ); + + this.hostStreams.closeAll(); + this.hostCallbacks.clear(); + this.streamRouter.clear(); } } diff --git a/src/lib/runtime/scheduler.ts b/src/lib/runtime/scheduler.ts index 2c6b6ae..e94cbb1 100644 --- a/src/lib/runtime/scheduler.ts +++ b/src/lib/runtime/scheduler.ts @@ -1,15 +1,26 @@ import type { ActorProxy, AnyActorClass } from "../types.js"; import type { WorkerNode } from "./worker-node.js"; +/** + * Round-robin placement of new actors across the worker pool. + * After spawn, an actor stays sticky on the chosen worker. + */ export class Scheduler { private next = 0; + /** + * @param workers - Non-empty list of {@link WorkerNode}s + */ constructor(private readonly workers: WorkerNode[]) { if (workers.length === 0) { throw new Error("Scheduler requires at least one worker"); } } + /** + * Picks the next worker in round-robin order. + * @returns Worker that will host the next spawned actor + */ pick(): WorkerNode { const worker = this.workers[this.next % this.workers.length]; @@ -21,6 +32,12 @@ export class Scheduler { return worker; } + /** + * Spawns an actor on the next worker and returns a typed proxy. + * @param Class - Actor class (already registered on workers) + * @param args - Constructor arguments + * @returns Typed proxy for the new actor + */ async create( Class: C, args: ConstructorParameters, diff --git a/src/lib/runtime/stream-router.ts b/src/lib/runtime/stream-router.ts new file mode 100644 index 0000000..0cf2a68 --- /dev/null +++ b/src/lib/runtime/stream-router.ts @@ -0,0 +1,64 @@ +/** + * Tracks which sides hold a proxy for a given stream_ref. + * Used by the host to fan-out `stream_data` / write acks to the right consumers. + */ +export class StreamRouter { + private readonly subs = new Map>(); + + /** + * Composite key for owner + streamId. + * @param owner - `"host"` or worker id that owns the real stream + * @param streamId - Stream id on that owner + * @returns Map key string + */ + private key(owner: "host" | number, streamId: number): string { + return `${owner}:${streamId}`; + } + + /** + * Records that `consumer` holds a proxy for the given stream. + * @param owner - Stream owner + * @param streamId - Stream id + * @param consumer - `"host"` or worker id that should receive events + */ + subscribe( + owner: "host" | number, + streamId: number, + consumer: "host" | number, + ): void { + const key = this.key(owner, streamId); + let set = this.subs.get(key); + + if (!set) { + set = new Set(); + this.subs.set(key, set); + } + set.add(consumer); + } + + /** + * @param owner - Stream owner + * @param streamId - Stream id + * @returns All consumers currently subscribed to this stream + */ + consumers( + owner: "host" | number, + streamId: number, + ): Array<"host" | number> { + return [...(this.subs.get(this.key(owner, streamId)) ?? [])]; + } + + /** + * Drops all consumers for a stream (on close). + * @param owner - Stream owner + * @param streamId - Stream id + */ + unsubscribeAll(owner: "host" | number, streamId: number): void { + this.subs.delete(this.key(owner, streamId)); + } + + /** Clears every subscription (runtime dispose). */ + clear(): void { + this.subs.clear(); + } +} diff --git a/src/lib/runtime/worker-node.ts b/src/lib/runtime/worker-node.ts index 3fc1b20..ee6d62a 100644 --- a/src/lib/runtime/worker-node.ts +++ b/src/lib/runtime/worker-node.ts @@ -1,31 +1,52 @@ +/* eslint-disable sort-imports */ import { Worker } from "node:worker_threads"; import path from "node:path"; -import type { ActorHandle, DebugHandler } from "../types.js"; +import type { + ActorHandle, DebugHandler, DestroyOptions, +} from "../types.js"; import type { ActorRef, BridgeCallMessage, + BridgeCallbackInvokeMessage, + BridgeCallbackResultMessage, BridgeResultMessage, + BridgeStreamMessage, + CallbackRef, ProtocolMessage, ProtocolResponse, + StreamRef, } from "../protocol/messages.js"; -import { - actorRef, formatActorId, isActorRef, -} from "../protocol/refs.js"; +import { fromWireError, toWireError } from "../protocol/messages.js"; +import { formatActorId, isStreamRef } from "../protocol/refs.js"; import { createProxy, getActorHandle } from "../proxy/proxy.js"; import { Serializer } from "../protocol/serializer.js"; +import { CallbackRegistry } from "../protocol/callback-registry.js"; +import { StreamBridge } from "../protocol/stream-bridge.js"; +import { cloneErrorMessage } from "../protocol/plain.js"; import { getModuleDir } from "./module-dir.js"; +import type { StreamRouter } from "./stream-router.js"; +/** Pending host↔worker request waiting for a {@link ProtocolResponse}. */ type Pending = { reject: (reason?: unknown) => void; resolve: (value: unknown) => void; }; +/** Construction options for {@link WorkerNode}. */ export type WorkerNodeOptions = { + /** Worker index in the runtime pool. */ id: number; onDebug?: DebugHandler; callTimeoutMs?: number; + /** Lookup other workers for bridge routing. */ getWorker: (workerId: number) => WorkerNode | undefined; + /** Shared host-side callback registry. */ + hostCallbacks: CallbackRegistry; + /** Shared host-side stream bridge. */ + hostStreams: StreamBridge; + /** Shared stream consumer subscription table. */ + streamRouter: StreamRouter; }; // Always load the ESM worker: CJS emit rewrites `import()` to `require()`, @@ -41,6 +62,11 @@ const workerEntry = path.join( "runtime-worker.js", ); +/** + * Host-side handle for one worker thread. + * Owns the mailbox, pending requests, encode/decode, and bridge routing + * for actors sticky to this worker. + */ export class WorkerNode { readonly id: number; private readonly worker: Worker; @@ -49,6 +75,9 @@ export class WorkerNode { private readonly onDebug?: DebugHandler; private readonly callTimeoutMs?: number; private readonly getWorker: (workerId: number) => WorkerNode | undefined; + private readonly hostCallbacks: CallbackRegistry; + private readonly hostStreams: StreamBridge; + private readonly streamRouter: StreamRouter; private nextRequestId = 1; private nextObjectId = 1; private closed = false; @@ -57,9 +86,16 @@ export class WorkerNode { private readonly mailboxes = new Map>(); private drainWaiters: Array<() => void> = []; + /** + * Spawns the ESM worker entry and wires message / error handlers. + * @param options - Pool identity, shared registries, timeouts + */ constructor(options: WorkerNodeOptions) { this.id = options.id; this.getWorker = options.getWorker; + this.hostCallbacks = options.hostCallbacks; + this.hostStreams = options.hostStreams; + this.streamRouter = options.streamRouter; if (options.onDebug) this.onDebug = options.onDebug; if (options.callTimeoutMs !== undefined) { this.callTimeoutMs = options.callTimeoutMs; @@ -69,51 +105,88 @@ export class WorkerNode { workerData: { workerId: this.id }, }); - this.worker.on("message", (msg: ProtocolResponse | BridgeCallMessage) => { - if ( - typeof msg === "object" - && msg !== null - && "bridge" in msg - && msg.bridge === "call" - ) { - void this.handleBridgeCall(msg); + this.worker.on("message", (msg: unknown) => { + void this.onWorkerMessage(msg); + }); - return; + this.worker.on("error", (err: Error) => { + this.onDebug?.({ + error: err.message, + type: "worker:error", + workerId: this.id, + }); + for (const [, p] of this.pending) { + p.reject(err); } + this.pending.clear(); + this.notifyDrain(); + }); + } - const response = msg as ProtocolResponse; - const pending = this.pending.get(response.id); + /** + * Dispatches inbound worker messages: protocol responses and bridge traffic. + * @param msg - Raw `worker_threads` message + */ + private async onWorkerMessage(msg: unknown): Promise { + if (typeof msg !== "object" || msg === null) return; - if (!pending) return; - this.pending.delete(response.id); - this.notifyDrain(); + if ("bridge" in msg) { + const bridge = (msg as { bridge: string; }).bridge; + + if (bridge === "call") { + void this.handleBridgeCall(msg as BridgeCallMessage); - if ("error" in response) { - const err = new Error(response.error.message); + return; + } - if (response.error.name) err.name = response.error.name; - if (response.error.stack) err.stack = response.error.stack; - pending.reject(err); + if (bridge === "callback_invoke") { + void this.handleBridgeCallbackInvoke( + msg as BridgeCallbackInvokeMessage, + ); return; } - pending.resolve( - this.serializer.decode(response.result, (ref) => - this.resolveActorRef(ref), - ), - ); - }); + if (bridge === "callback_result") { + this.handleBridgeCallbackResult(msg as BridgeCallbackResultMessage); - this.worker.on("error", (err) => { - for (const [, p] of this.pending) { - p.reject(err); + return; } - this.pending.clear(); - this.notifyDrain(); - }); + + if (typeof bridge === "string" && bridge.startsWith("stream_")) { + void this.handleBridgeStream(msg as BridgeStreamMessage); + + return; + } + + return; + } + + const response = msg as ProtocolResponse; + const pending = this.pending.get(response.id); + + if (!pending) return; + this.pending.delete(response.id); + this.notifyDrain(); + + if ("error" in response) { + pending.reject(fromWireError(response.error)); + + return; + } + + try { + pending.resolve(this.decodeValue(response.result)); + } catch (err) { + pending.reject(err); + } } + /** + * Resolves an {@link ActorRef} to a host-side proxy on the owning worker. + * @param ref - Wire actor tag from decode + * @returns Host-side proxy for the actor on its owning worker + */ private resolveActorRef(ref: ActorRef): unknown { const target = this.getWorker(ref.workerId); @@ -126,6 +199,106 @@ export class WorkerNode { return target.createLocalProxy(ref.objectId); } + /** + * Builds an async stub that invokes a callback on its owner side. + * @param ref - Wire callback tag from decode + * @returns Async function that invokes the callback on its owner side + */ + private resolveCallbackRef(ref: CallbackRef): unknown { + if (ref.owner === "host") { + return (...args: unknown[]) => + this.hostCallbacks.invoke(ref.callbackId, args); + } + + if (ref.owner === this.id) { + return (...args: unknown[]) => + this.invokeWorkerCallback(ref.callbackId, args); + } + + const target = this.getWorker(ref.owner); + + if (!target) { + throw new Error( + `Unknown worker ${ref.owner} for callback ${ref.callbackId}`, + ); + } + + return (...args: unknown[]) => + target.invokeWorkerCallback(ref.callbackId, args); + } + + /** + * Subscribes the host as consumer and creates a local stream proxy. + * @param ref - Wire stream tag from decode + * @returns Local stream proxy for the remote stream + */ + private resolveStreamRef(ref: StreamRef): unknown { + this.streamRouter.subscribe(ref.owner, ref.streamId, "host"); + + return this.hostStreams.createProxy(ref); + } + + /** + * Decodes a wire value using this node's resolvers. + * @param value - Encoded wire value + * @returns Decoded local value (proxies, callbacks, streams) + */ + private decodeValue(value: unknown): unknown { + return this.serializer.decode(value, { + resolveActorRef: (ref) => this.resolveActorRef(ref), + resolveCallbackRef: (ref) => this.resolveCallbackRef(ref), + resolveStreamRef: (ref) => this.resolveStreamRef(ref), + }); + } + + /** + * Encodes a host-side value for the wire (actors, callbacks, streams). + * @param value - Args tree or return value + * @param options - Optional call-scoped callback tracking / bound actor id + * @returns Encoded wire value + */ + private encodeValue( + value: unknown, + options?: { + callScopedCallbacks?: number[]; + boundObjectId?: number; + }, + ): unknown { + return this.serializer.encode(value, { + registerCallback: (fn) => { + const ref = this.hostCallbacks.register( + fn as (...args: unknown[]) => unknown, + { + callScoped: options?.callScopedCallbacks !== undefined, + ...(options?.boundObjectId !== undefined + ? { boundObjectId: options.boundObjectId } + : {}), + }, + ); + + options?.callScopedCallbacks?.push(ref.callbackId); + + return ref; + }, + registerStream: (obj) => { + const ref = this.hostStreams.tryRegisterLocal(obj); + + if (ref) { + this.streamRouter.subscribe(ref.owner, ref.streamId, this.id); + } + + return ref; + }, + resolveProxy: (v) => getActorHandle(v), + workerId: this.id, + }); + } + + /** + * Builds a typed proxy for an actor sticky to this worker. + * @param objectId - Local object id on this worker + * @returns Typed host-side proxy for the actor + */ createLocalProxy(objectId: number): T { const handle: ActorHandle = { objectId, workerId: this.id }; @@ -134,19 +307,237 @@ export class WorkerNode { ) as T; } - private encodeArgs(args: unknown[]): unknown[] { - return args.map((arg) => { - const handle = getActorHandle(arg); + /** + * Invokes a callback registered inside this worker. + * @param callbackId - Id in the worker's callback registry + * @param args - Already-local (or to-be-encoded) arguments + * @returns Decoded callback result + */ + async invokeWorkerCallback( + callbackId: number, + args: unknown[], + ): Promise { + const encodedArgs = this.encodeValue(args) as unknown[]; + + return this.request({ + args: encodedArgs, + callbackId, + command: "callback_invoke", + id: this.nextRequestId++, + }); + } + + /** + * Handles a worker-posted callback invoke (host or cross-worker owner). + * @param msg - Bridge callback invoke from this worker + */ + private async handleBridgeCallbackInvoke( + msg: BridgeCallbackInvokeMessage, + ): Promise { + try { + const decodedArgs = this.decodeValue(msg.args) as unknown[]; + let result: unknown; + + if (msg.owner === "host") { + result = await this.hostCallbacks.invoke(msg.callbackId, decodedArgs); + } else { + const target = this.getWorker(msg.owner); - if (handle) { - return actorRef(handle.workerId, handle.objectId); + if (!target) { + throw new Error(`Unknown worker ${msg.owner}`); + } + result = await target.invokeWorkerCallback( + msg.callbackId, + decodedArgs, + ); } - return arg; - }); + const reply: BridgeCallbackResultMessage = { + bridge: "callback_result", + id: msg.id, + result: this.encodeValue(result), + }; + + this.post(reply); + } catch (err) { + const reply: BridgeCallbackResultMessage = { + bridge: "callback_result", + error: toWireError(err), + id: msg.id, + }; + + this.post(reply); + } + } + + /** + * Resolves a host-pending bridge callback result (rarely used; worker holds bridgePending). + * @param msg - Callback result message + */ + private handleBridgeCallbackResult(msg: BridgeCallbackResultMessage): void { + const pending = this.pending.get(msg.id); + + if (!pending) return; + this.pending.delete(msg.id); + this.notifyDrain(); + + if ("error" in msg) { + pending.reject(fromWireError(msg.error)); + + return; + } + + pending.resolve(this.decodeValue(msg.result)); } + /** + * Routes stream bridge messages between host streams and worker peers. + * @param msg - Stream control/data message from this worker + */ + private async handleBridgeStream(msg: BridgeStreamMessage): Promise { + const { owner, streamId } = msg; + + if (msg.bridge === "stream_data" && msg.direction === "from_owner") { + for (const consumer of this.streamRouter.consumers(owner, streamId)) { + if (consumer === "host") { + this.hostStreams.onRemoteData(owner, streamId, msg.chunk); + } else if (consumer !== this.id) { + this.getWorker(consumer)?.postToWorker(msg); + } else { + this.postToWorker(msg); + } + } + + return; + } + + if (msg.bridge === "stream_end" && msg.direction === "from_owner") { + for (const consumer of this.streamRouter.consumers(owner, streamId)) { + if (consumer === "host") { + this.hostStreams.onRemoteEnd(owner, streamId); + } else if (consumer !== this.id) { + this.getWorker(consumer)?.postToWorker(msg); + } else { + this.postToWorker(msg); + } + } + + return; + } + + if (msg.bridge === "stream_error") { + const err = fromWireError(msg.error); + + if (msg.direction === "from_owner") { + for (const consumer of this.streamRouter.consumers(owner, streamId)) { + if (consumer === "host") { + this.hostStreams.onRemoteError(owner, streamId, err); + } else if (consumer !== this.id) { + this.getWorker(consumer)?.postToWorker(msg); + } else { + this.postToWorker(msg); + } + } + } else if (owner === "host") { + this.hostStreams.onRemoteError(owner, streamId, err); + } else { + this.getWorker(owner)?.postToWorker(msg); + } + + return; + } + + if (msg.bridge === "stream_pause") { + if (owner === "host") this.hostStreams.onPause(owner, streamId); + else if (owner === this.id) this.postToWorker(msg); + else this.getWorker(owner)?.postToWorker(msg); + + return; + } + + if (msg.bridge === "stream_resume") { + if (owner === "host") this.hostStreams.onResume(owner, streamId); + else if (owner === this.id) this.postToWorker(msg); + else this.getWorker(owner)?.postToWorker(msg); + + return; + } + + if (msg.bridge === "stream_write") { + if (owner === "host") { + await this.hostStreams.onWrite(owner, streamId, msg.id, msg.chunk); + } else if (owner === this.id) { + this.postToWorker(msg); + } else { + this.getWorker(owner)?.postToWorker(msg); + } + + return; + } + + if (msg.bridge === "stream_write_end") { + if (owner === "host") { + await this.hostStreams.onWriteEnd(owner, streamId, msg.id); + } else if (owner === this.id) { + this.postToWorker(msg); + } else { + this.getWorker(owner)?.postToWorker(msg); + } + + return; + } + + if (msg.bridge === "stream_write_result") { + if (this.hostStreams.has(owner, streamId)) { + this.hostStreams.onWriteResult( + owner, + streamId, + msg.id, + msg.error ? fromWireError(msg.error) : undefined, + ); + } else { + for (const consumer of this.streamRouter.consumers(owner, streamId)) { + if (consumer !== "host") { + this.getWorker(consumer)?.postToWorker(msg); + } + } + } + + return; + } + + if (msg.bridge === "stream_close") { + if (this.hostStreams.has(owner, streamId)) { + this.hostStreams.onClose(owner, streamId); + } + this.streamRouter.unsubscribeAll(owner, streamId); + if (owner === this.id) this.postToWorker(msg); + else if (owner !== "host") this.getWorker(owner)?.postToWorker(msg); + } + } + + /** + * Posts an arbitrary message into this worker (bridge fan-out helper). + * @param msg - Message payload + */ + postToWorker(msg: unknown): void { + this.post(msg); + } + + /** + * Forwards a cross-worker actor call to the target {@link WorkerNode}. + * @param msg - Bridge call from this worker's stub + */ private async handleBridgeCall(msg: BridgeCallMessage): Promise { + this.onDebug?.({ + method: msg.method, + objectId: msg.objectId, + requestId: msg.id, + targetWorkerId: msg.targetWorkerId, + type: "bridge:call", + workerId: this.id, + }); + try { const target = this.getWorker(msg.targetWorkerId); @@ -154,19 +545,11 @@ export class WorkerNode { throw new Error(`Unknown worker ${msg.targetWorkerId}`); } - const decodedArgs = msg.args.map((arg) => { - if (isActorRef(arg)) { - return this.resolveActorRef(arg); - } - - return arg; - }); - + const decodedArgs = this.decodeValue(msg.args) as unknown[]; const result = await target.call(msg.objectId, msg.method, decodedArgs); - const handle = getActorHandle(result); - const encoded = handle - ? actorRef(handle.workerId, handle.objectId) - : result; + const encoded = target.encodeForBridge(result); + + this.subscribeStreamsIn(encoded, this.id); const reply: BridgeResultMessage = { bridge: "result", @@ -174,29 +557,81 @@ export class WorkerNode { result: encoded, }; - this.worker.postMessage(reply); + this.post(reply); + this.onDebug?.({ + requestId: msg.id, + targetWorkerId: msg.targetWorkerId, + type: "bridge:result", + workerId: this.id, + }); } catch (err) { - const error = err instanceof Error ? err : new Error(String(err)); const reply: BridgeResultMessage = { bridge: "result", - error: { - message: error.message, - name: error.name, - ...(error.stack ? { stack: error.stack } : {}), - }, + error: toWireError(err), id: msg.id, }; - this.worker.postMessage(reply); + this.post(reply); + this.onDebug?.({ + error: err instanceof Error ? err.message : String(err), + requestId: msg.id, + targetWorkerId: msg.targetWorkerId, + type: "bridge:result", + workerId: this.id, + }); + } + } + + /** + * Encode a host-side value for a bridge reply (may include stream/callback refs). + * @param value - Call result to send back to the requesting worker + * @returns Encoded value for the bridge reply + */ + encodeForBridge(value: unknown): unknown { + return this.encodeValue(value); + } + + /** + * Walks an encoded tree and subscribes `consumer` to any nested stream refs. + * @param value - Encoded (or partially decoded) tree + * @param consumer - Side that will hold proxies for those streams + */ + private subscribeStreamsIn( + value: unknown, + consumer: "host" | number, + ): void { + if (isStreamRef(value)) { + this.streamRouter.subscribe(value.owner, value.streamId, consumer); + + return; + } + if (!value || typeof value !== "object") return; + if (Array.isArray(value)) { + for (const item of value) this.subscribeStreamsIn(item, consumer); + + return; + } + for (const item of Object.values(value)) { + this.subscribeStreamsIn(item, consumer); } } + /** + * Throws if this worker node has been closed by dispose. + * @param action - Human-readable action for the error message + */ assertOpen(action: string): void { if (this.closed) { throw new Error(`Runtime is disposed; cannot ${action}`); } } + /** + * Tells the worker to load and register an actor class module. + * @param className - Registry key / class name + * @param moduleUrl - Absolute URL for worker `import()` + * @param exportName - Named export inside the module + */ async register( className: string, moduleUrl: string, @@ -218,22 +653,35 @@ export class WorkerNode { }); } + /** + * Creates an actor instance on this worker and returns a host proxy. + * @param className - Registered class name + * @param args - Constructor arguments (encoded before send) + * @returns Host-side proxy for the new actor + */ async create( className: string, args: unknown[], ): Promise { this.assertOpen("spawn"); const objectId = this.nextObjectId++; + const callScoped: number[] = []; this.localObjects.add(objectId); - await this.request({ - args: this.encodeArgs(args), - className, - command: "create", - id: this.nextRequestId++, - objectId, - }); + try { + await this.request({ + args: this.encodeValue(args, { + callScopedCallbacks: callScoped, + }) as unknown[], + className, + command: "create", + id: this.nextRequestId++, + objectId, + }); + } finally { + this.hostCallbacks.releaseCallScoped(callScoped); + } this.onDebug?.({ actorId: formatActorId(this.id, objectId), @@ -246,7 +694,15 @@ export class WorkerNode { return this.createLocalProxy(objectId); } - async destroy(objectId: number): Promise { + /** + * Destroys an actor on this worker (optionally calling dispose/close first). + * @param objectId - Local actor id + * @param options - `{ close: false }` skips actor cleanup + */ + async destroy( + objectId: number, + options?: DestroyOptions, + ): Promise { this.assertOpen("destroy"); if (!this.localObjects.has(objectId)) { throw new Error( @@ -254,8 +710,11 @@ export class WorkerNode { ); } + const close = options?.close !== false; + await this.enqueue(objectId, async () => { await this.request({ + close, command: "destroy", id: this.nextRequestId++, objectId, @@ -264,6 +723,7 @@ export class WorkerNode { this.localObjects.delete(objectId); this.mailboxes.delete(objectId); + this.hostCallbacks.releaseBoundToObject(objectId); this.onDebug?.({ actorId: formatActorId(this.id, objectId), @@ -273,6 +733,13 @@ export class WorkerNode { }); } + /** + * Enqueues a method call on an actor's mailbox. + * @param objectId - Local actor id + * @param method - Method name + * @param args - Host-side arguments (encoded before send) + * @returns Decoded method result + */ async call( objectId: number, method: string, @@ -283,6 +750,12 @@ export class WorkerNode { return this.enqueue(objectId, () => this.doCall(objectId, method, args)); } + /** + * Chains a task onto the per-actor mailbox promise. + * @param objectId - Actor whose mailbox to use + * @param task - Async work to run serially for this actor + * @returns Result of the enqueued task + */ private enqueue(objectId: number, task: () => Promise): Promise { const prev = this.mailboxes.get(objectId) ?? Promise.resolve(); const next = prev.then(task, task); @@ -298,6 +771,13 @@ export class WorkerNode { return next; } + /** + * Performs one remote call with debug events, timeout, and call-scoped callbacks. + * @param objectId - Local actor id + * @param method - Method name + * @param args - Host-side arguments + * @returns Decoded call result + */ private async doCall( objectId: number, method: string, @@ -306,6 +786,7 @@ export class WorkerNode { const requestId = this.nextRequestId++; const started = performance.now(); const actorId = formatActorId(this.id, objectId); + const callScoped: number[] = []; this.onDebug?.({ actorId, @@ -319,7 +800,9 @@ export class WorkerNode { try { const result = await this.request( { - args: this.encodeArgs(args), + args: this.encodeValue(args, { + callScopedCallbacks: callScoped, + }) as unknown[], command: "call", id: requestId, method, @@ -327,6 +810,7 @@ export class WorkerNode { }, this.callTimeoutMs, method, + { actorId, objectId }, ); this.onDebug?.({ @@ -352,13 +836,24 @@ export class WorkerNode { workerId: this.id, }); throw err; + } finally { + this.hostCallbacks.releaseCallScoped(callScoped); } } + /** + * Sends a {@link ProtocolMessage} and waits for the matching response. + * @param message - Outbound command (must carry a unique `id`) + * @param timeoutMs - Optional call timeout + * @param methodForTimeout - Method name for timeout error text / debug + * @param timeoutMeta - Actor identity for `call:timeout` debug events + * @returns Decoded response value + */ private request( message: ProtocolMessage, timeoutMs?: number, methodForTimeout?: string, + timeoutMeta?: { actorId: string; objectId: number; }, ): Promise { return new Promise((resolve, reject) => { let timer: ReturnType | undefined; @@ -382,6 +877,19 @@ export class WorkerNode { if (!this.pending.has(message.id)) return; this.pending.delete(message.id); this.notifyDrain(); + this.onDebug?.({ + ...(timeoutMeta?.actorId + ? { actorId: timeoutMeta.actorId } + : {}), + ...(timeoutMeta?.objectId !== undefined + ? { objectId: timeoutMeta.objectId } + : {}), + ...(methodForTimeout ? { method: methodForTimeout } : {}), + requestId: message.id, + timeoutMs, + type: "call:timeout", + workerId: this.id, + }); reject( new Error( `Call timed out after ${timeoutMs}ms` @@ -391,10 +899,23 @@ export class WorkerNode { }, timeoutMs); } - this.worker.postMessage(message); + this.post(message); }); } + /** + * `postMessage` with a clearer error when structured clone fails. + * @param message - Payload to send into the worker + */ + private post(message: unknown): void { + try { + this.worker.postMessage(message); + } catch (err) { + throw new Error(cloneErrorMessage(err)); + } + } + + /** Wakes {@link drain} waiters when no requests are pending. */ private notifyDrain(): void { if (this.pending.size > 0) return; const waiters = this.drainWaiters; @@ -403,6 +924,7 @@ export class WorkerNode { for (const wait of waiters) wait(); } + /** Waits until all in-flight requests on this worker have settled. */ async drain(): Promise { if (this.pending.size === 0) return; await new Promise((resolve) => { @@ -410,6 +932,9 @@ export class WorkerNode { }); } + /** + * Asks the worker to call dispose/close on every live actor (best-effort). + */ async closeAllActors(): Promise { if (this.closed) return; await this.request({ @@ -418,6 +943,10 @@ export class WorkerNode { }); } + /** + * Closes the node: optional actor close, drain, reject pendings, terminate thread. + * @param options - `{ closeActors: false }` skips per-actor close + */ async gracefulTerminate(options?: { closeActors?: boolean; }): Promise { diff --git a/src/lib/types.ts b/src/lib/types.ts index d963db2..52f6df2 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -5,17 +5,26 @@ export type ActorClass = { name: string; }; -/** Stable identity of an actor across the runtime. */ +/** Stable identity of an actor across the runtime (`workerId:objectId`). */ export type ActorHandle = { workerId: number; objectId: number; }; /** - * Remote view of an actor: methods are async, `return this` stays a proxy. + * Maps a local function type to the async stub seen across the worker boundary. + * @internal + */ +type RemoteFunction = F extends (...args: infer A) => infer R + ? (...args: A) => Promise> + : never; + +/** + * Remote view of an actor: methods are always async, `return this` stays a proxy. * * Already-async methods are kept as-is so generic type parameters survive * (mapped `infer` would erase them). Sync methods are wrapped in `Promise`. + * Returned functions become async stubs. */ export type ActorProxy = { [K in keyof TInstance as TInstance[K] extends (...args: never) => unknown @@ -40,6 +49,10 @@ export type AnyActorClass = { name: string; }; +/** + * Runtime observability events emitted when `RuntimeOptions.debug` is set. + * Spawn/call events carry `actorId` as `"workerId:objectId"`. + */ export type DebugEvent = | { type: "register"; @@ -77,21 +90,71 @@ export type DebugEvent = durationMs: number; error?: string; } + | { + type: "call:timeout"; + workerId: number; + objectId?: number; + actorId?: string; + method?: string; + requestId: number; + timeoutMs: number; + } + | { + type: "worker:error"; + workerId: number; + error: string; + } + | { + type: "bridge:call"; + workerId: number; + targetWorkerId: number; + objectId: number; + method: string; + requestId: number; + } + | { + type: "bridge:result"; + workerId: number; + targetWorkerId: number; + requestId: number; + error?: string; + } | { type: "dispose"; workers: number; }; +/** Callback invoked for each {@link DebugEvent}. */ export type DebugHandler = (event: DebugEvent) => void; -/** If method returns `this` / same instance, proxy keeps identity. */ +/** + * If method returns `this` / same instance, proxy keeps identity. + * Functions become async stubs; everything else is left as-is. + * @internal + */ type RemoteValue = R extends TInstance ? ActorProxy - : R; + : R extends (...args: never[]) => unknown + ? RemoteFunction + : R; +/** Options for {@link Runtime.destroy}. */ +export type DestroyOptions = { + /** Call actor `dispose`/`close` before removing (default true). */ + close?: boolean; +}; + +/** + * Debug configuration for {@link Runtime}. + * - `true` — log events to stderr + * - function — custom handler + * - `{ onEvent }` — same as function, object form + */ export type RuntimeDebug = boolean | DebugHandler | { onEvent: DebugHandler; }; +/** Options for constructing a {@link Runtime}. */ export interface RuntimeOptions { + /** Number of worker threads in the pool (default `1`). */ workers?: number; /** Log runtime events to stderr, or pass a custom handler. */ debug?: RuntimeDebug; diff --git a/src/lib/worker/runtime-worker.ts b/src/lib/worker/runtime-worker.ts index ab7b2fa..4375615 100644 --- a/src/lib/worker/runtime-worker.ts +++ b/src/lib/worker/runtime-worker.ts @@ -1,14 +1,35 @@ +/** + * Worker-thread entry for `@js-ak/remote-objects`. + * + * Loads actor modules, hosts sticky instances, encodes/decodes wire values + * (actors, callbacks, streams), and talks to the parent {@link WorkerNode} + * via protocol + bridge messages. + * + * Always loaded as ESM (`build/esm/lib/worker/runtime-worker.js`) even when + * the host package is consumed as CJS. + */ +/* eslint-disable sort-imports */ import { parentPort, workerData } from "node:worker_threads"; import { pathToFileURL } from "node:url"; import type { + ActorRef, BridgeCallMessage, + BridgeCallbackInvokeMessage, + BridgeCallbackResultMessage, BridgeResultMessage, + BridgeStreamMessage, + CallbackRef, ProtocolMessage, + StreamRef, } from "../protocol/messages.js"; import type { AnyActorClass } from "../types.js"; +import { fromWireError, toWireError } from "../protocol/messages.js"; +import { actorRef } from "../protocol/refs.js"; import { Serializer } from "../protocol/serializer.js"; -import { isActorRef } from "../protocol/refs.js"; +import { CallbackRegistry } from "../protocol/callback-registry.js"; +import { StreamBridge, type StreamTransport } from "../protocol/stream-bridge.js"; +import { cloneErrorMessage } from "../protocol/plain.js"; if (!parentPort) { throw new Error("runtime-worker must run inside a Worker"); @@ -19,7 +40,9 @@ const workerId = (workerData as { workerId: number; }).workerId; const registry = new Map(); const objects = new Map(); const objectIds = new Map(); +const stubRefs = new WeakMap(); const serializer = new Serializer(); +const callbacks = new CallbackRegistry(workerId); let nextBridgeId = 1; const bridgePending = new Map< number, @@ -29,6 +52,108 @@ const bridgePending = new Map< } >(); +/** + * Posts to the parent port with a clearer error on structured-clone failure. + * @param message - Protocol or bridge payload + */ +function post(message: unknown): void { + try { + port.postMessage(message); + } catch (err) { + throw new Error(cloneErrorMessage(err)); + } +} + +/** + * Builds worker-side {@link StreamTransport} that posts bridge stream messages to the host. + * @param ref - Stream being transported from this worker + * @returns Stream transport that posts bridge stream messages + */ +function createStreamTransport(ref: StreamRef): StreamTransport { + return { + sendClose: () => { + post({ + bridge: "stream_close", + owner: ref.owner, + streamId: ref.streamId, + } satisfies BridgeStreamMessage); + }, + sendData: (chunk) => { + post({ + bridge: "stream_data", + chunk, + direction: "from_owner", + owner: ref.owner, + streamId: ref.streamId, + } satisfies BridgeStreamMessage); + }, + sendEnd: () => { + post({ + bridge: "stream_end", + direction: "from_owner", + owner: ref.owner, + streamId: ref.streamId, + } satisfies BridgeStreamMessage); + }, + sendError: (err) => { + post({ + bridge: "stream_error", + direction: "from_owner", + error: toWireError(err), + owner: ref.owner, + streamId: ref.streamId, + } satisfies BridgeStreamMessage); + }, + sendPause: () => { + post({ + bridge: "stream_pause", + owner: ref.owner, + streamId: ref.streamId, + } satisfies BridgeStreamMessage); + }, + sendResume: () => { + post({ + bridge: "stream_resume", + owner: ref.owner, + streamId: ref.streamId, + } satisfies BridgeStreamMessage); + }, + sendWrite: (id, chunk) => { + post({ + bridge: "stream_write", + chunk, + id, + owner: ref.owner, + streamId: ref.streamId, + } satisfies BridgeStreamMessage); + }, + sendWriteEnd: (id) => { + post({ + bridge: "stream_write_end", + id, + owner: ref.owner, + streamId: ref.streamId, + } satisfies BridgeStreamMessage); + }, + sendWriteResult: (id, error) => { + post({ + bridge: "stream_write_result", + id, + owner: ref.owner, + streamId: ref.streamId, + ...(error ? { error: toWireError(error) } : {}), + } satisfies BridgeStreamMessage); + }, + }; +} + +const streams = new StreamBridge(workerId, createStreamTransport); + +/** + * Normalizes a module path to a URL suitable for dynamic `import()`. + * @param moduleUrl - Absolute path or already-qualified `file:` / `data:` URL + * @returns URL string suitable for dynamic `import()` + */ function toImportUrl(moduleUrl: string): string { if (moduleUrl.startsWith("file:") || moduleUrl.startsWith("data:")) { return moduleUrl; @@ -37,6 +162,13 @@ function toImportUrl(moduleUrl: string): string { return pathToFileURL(moduleUrl).href; } +/** + * Resolves a named (or default) class export from a loaded actor module. + * @param mod - Module namespace object from `import()` + * @param exportName - Expected export name + * @param moduleUrl - For error messages + * @returns Actor class export + */ function resolveExport( mod: Record, exportName: string, @@ -66,8 +198,136 @@ function resolveExport( ); } +/** + * Encodes a worker-local value for the wire (actors, stubs, callbacks, streams). + * @param value - Return value or argument tree + * @param options - Current actor context and optional call-scoped callback list + * @returns Wire-safe encoded value + */ +function encodeValue( + value: unknown, + options?: { + currentObject?: object; + currentObjectId?: number; + callScopedCallbacks?: number[]; + boundObjectId?: number; + }, +): unknown { + return serializer.encode(value, { + actors: objectIds, + ...(options?.currentObject !== undefined + ? { currentObject: options.currentObject } + : {}), + ...(options?.currentObjectId !== undefined + ? { currentObjectId: options.currentObjectId } + : {}), + registerCallback: (fn) => { + const ref = callbacks.register(fn as (...args: unknown[]) => unknown, { + callScoped: options?.callScopedCallbacks !== undefined, + ...(options?.boundObjectId !== undefined + ? { boundObjectId: options.boundObjectId } + : options?.currentObjectId !== undefined + ? { boundObjectId: options.currentObjectId } + : {}), + }); + + options?.callScopedCallbacks?.push(ref.callbackId); + + return ref; + }, + registerStream: (obj) => streams.tryRegisterLocal(obj), + resolveProxy: (value) => { + const ref = stubRefs.get(value as object); + + if (!ref) return undefined; + + return { objectId: ref.objectId, workerId: ref.workerId }; + }, + workerId, + }); +} + +/** + * Builds an async stub for a callback owned elsewhere (host or another worker). + * @param ref - Wire callback tag + * @returns Local invoker or async bridge stub for the callback + */ +function resolveCallbackRef(ref: CallbackRef): unknown { + if (ref.owner === workerId) { + return (...args: unknown[]) => callbacks.invoke(ref.callbackId, args); + } + + return (...args: unknown[]) => { + const id = nextBridgeId++; + const encodedArgs = encodeValue(args); + + return new Promise((resolve, reject) => { + bridgePending.set(id, { reject, resolve }); + const msg: BridgeCallbackInvokeMessage = { + args: encodedArgs as unknown[], + bridge: "callback_invoke", + callbackId: ref.callbackId, + id, + owner: ref.owner, + }; + + post(msg); + }); + }; +} + +/** + * Creates a local stream proxy for a remote {@link StreamRef}. + * @param ref - Wire stream tag + * @returns Local stream proxy for the remote stream + */ +function resolveStreamRef(ref: StreamRef): unknown { + return streams.createProxy(ref); +} + +/** + * Decodes a wire value into local actors, stubs, callbacks, or stream proxies. + * @param value - Encoded wire value + * @returns Local actors, stubs, callbacks, or stream proxies + */ +function decodeValue(value: unknown): unknown { + return serializer.decode(value, { + resolveActorRef: (ref) => decodeActorRef(ref), + resolveCallbackRef: (ref) => resolveCallbackRef(ref), + resolveStreamRef: (ref) => resolveStreamRef(ref), + }); +} + +/** + * Resolves an actor ref to a local instance or a cross-worker stub. + * @param ref - Actor identity from the wire + * @returns Local actor instance or cross-worker stub + */ +function decodeActorRef(ref: { + workerId: number; + objectId: number; +}): unknown { + if (ref.workerId === workerId) { + const local = objects.get(ref.objectId); + + if (!local) { + throw new Error(`Unknown local actor ${workerId}:${ref.objectId}`); + } + + return local; + } + + return createRemoteStub(ref.workerId, ref.objectId); +} + +/** + * Proxy stub for an actor living on another worker; calls go through the bridge. + * @param targetWorkerId - Worker that owns the actor + * @param objectId - Object id on that worker + * @returns Proxy stub that bridges method calls to the remote actor + */ function createRemoteStub(targetWorkerId: number, objectId: number): object { - return new Proxy( + const stub = new Proxy( {}, { get(_target, prop) { @@ -77,20 +337,12 @@ function createRemoteStub(targetWorkerId: number, objectId: number): object { return (...args: unknown[]) => { const id = nextBridgeId++; - const encodedArgs = args.map((arg) => { - const localId = objectIds.get(arg as object); - - if (localId !== undefined) { - return { objectId: localId, type: "actor_ref", workerId }; - } - - return arg; - }); + const encodedArgs = encodeValue(args); return new Promise((resolve, reject) => { bridgePending.set(id, { reject, resolve }); const msg: BridgeCallMessage = { - args: encodedArgs, + args: encodedArgs as unknown[], bridge: "call", id, method: prop, @@ -98,30 +350,22 @@ function createRemoteStub(targetWorkerId: number, objectId: number): object { targetWorkerId, }; - port.postMessage(msg); + post(msg); }); }; }, }, ); -} - -function decodeArg(arg: unknown): unknown { - if (!isActorRef(arg)) return arg; - if (arg.workerId === workerId) { - const local = objects.get(arg.objectId); - - if (!local) { - throw new Error(`Unknown local actor ${workerId}:${arg.objectId}`); - } - - return local; - } + stubRefs.set(stub, actorRef(targetWorkerId, objectId)); - return createRemoteStub(arg.workerId, arg.objectId); + return stub; } +/** + * Best-effort call to instance `dispose` or `close` if present. + * @param instance - Actor instance + */ async function closeActor(instance: object): Promise { const record = instance as Record; const method @@ -135,33 +379,124 @@ async function closeActor(instance: object): Promise { await (record[method] as () => unknown).call(instance); } +/** + * Applies an inbound stream bridge message to the local {@link StreamBridge}. + * @param msg - Stream control/data message from the host + */ +async function handleStreamMessage(msg: BridgeStreamMessage): Promise { + const { owner, streamId } = msg; + + if (msg.bridge === "stream_data" && msg.direction === "from_owner") { + streams.onRemoteData(owner, streamId, msg.chunk); + + return; + } + + if (msg.bridge === "stream_end" && msg.direction === "from_owner") { + streams.onRemoteEnd(owner, streamId); + + return; + } + + if (msg.bridge === "stream_error") { + streams.onRemoteError(owner, streamId, fromWireError(msg.error)); + + return; + } + + if (msg.bridge === "stream_pause") { + streams.onPause(owner, streamId); + + return; + } + + if (msg.bridge === "stream_resume") { + streams.onResume(owner, streamId); + + return; + } + + if (msg.bridge === "stream_write") { + await streams.onWrite(owner, streamId, msg.id, msg.chunk); + + return; + } + + if (msg.bridge === "stream_write_end") { + await streams.onWriteEnd(owner, streamId, msg.id); + + return; + } + + if (msg.bridge === "stream_write_result") { + streams.onWriteResult( + owner, + streamId, + msg.id, + msg.error ? fromWireError(msg.error) : undefined, + ); + + return; + } + + if (msg.bridge === "stream_close") { + streams.onClose(owner, streamId); + } +} + port.on( "message", - async (msg: ProtocolMessage | BridgeResultMessage) => { + async ( + msg: + | ProtocolMessage + | BridgeResultMessage + | BridgeCallbackResultMessage + | BridgeStreamMessage, + ) => { if ( typeof msg === "object" && msg !== null && "bridge" in msg - && msg.bridge === "result" ) { - const pending = bridgePending.get(msg.id); + if (msg.bridge === "result") { + const pending = bridgePending.get(msg.id); + + if (!pending) return; + bridgePending.delete(msg.id); + + if ("error" in msg) { + pending.reject(fromWireError(msg.error)); + + return; + } + + pending.resolve(decodeValue(msg.result)); + + return; + } - if (!pending) return; - bridgePending.delete(msg.id); + if (msg.bridge === "callback_result") { + const pending = bridgePending.get(msg.id); - if ("error" in msg) { - const err = new Error(msg.error.message); + if (!pending) return; + bridgePending.delete(msg.id); - if (msg.error.name) err.name = msg.error.name; - if (msg.error.stack) err.stack = msg.error.stack; - pending.reject(err); + if ("error" in msg) { + pending.reject(fromWireError(msg.error)); + + return; + } + + pending.resolve(decodeValue(msg.result)); return; } - pending.resolve( - serializer.decode(msg.result, (ref) => decodeArg(ref)), - ); + if (typeof msg.bridge === "string" && msg.bridge.startsWith("stream_")) { + await handleStreamMessage(msg as BridgeStreamMessage); + + return; + } return; } @@ -181,7 +516,7 @@ port.on( ); registry.set(request.className, Class); - port.postMessage({ id: request.id, result: null }); + post({ id: request.id, result: null }); return; } @@ -193,13 +528,13 @@ port.on( throw new Error(`Unknown class: ${request.className}`); } - const args = request.args.map(decodeArg); + const args = decodeValue(request.args) as unknown[]; const instance = new Class(...args); objects.set(request.objectId, instance); objectIds.set(instance, request.objectId); - port.postMessage({ id: request.id, result: null }); + post({ id: request.id, result: null }); return; } @@ -208,10 +543,18 @@ port.on( const instance = objects.get(request.objectId); if (instance) { + if (request.close !== false) { + try { + await closeActor(instance); + } catch { + // Best-effort close before destroy. + } + } objectIds.delete(instance); objects.delete(request.objectId); + callbacks.releaseBoundToObject(request.objectId); } - port.postMessage({ id: request.id, result: null }); + post({ id: request.id, result: null }); return; } @@ -226,7 +569,28 @@ port.on( // Best-effort cleanup during graceful shutdown. } } - port.postMessage({ id: request.id, result: null }); + streams.closeAll(); + callbacks.clear(); + post({ id: request.id, result: null }); + + return; + } + + if (request.command === "callback_invoke") { + const args = decodeValue(request.args) as unknown[]; + const result = await callbacks.invoke(request.callbackId, args); + + post({ + id: request.id, + result: encodeValue(result), + }); + + return; + } + + if (request.command === "callback_release") { + callbacks.release(request.callbackIds); + post({ id: request.id, result: null }); return; } @@ -244,30 +608,23 @@ port.on( throw new Error(`Unknown method: ${request.method}`); } - const args = request.args.map(decodeArg); + const args = decodeValue(request.args) as unknown[]; const result = await fn.apply(object, args); - const encoded = serializer.encode(result, { - actors: objectIds, + const encoded = encodeValue(result, { + boundObjectId: request.objectId, currentObject: object, currentObjectId: request.objectId, - workerId, }); - port.postMessage({ id: request.id, result: encoded }); + post({ id: request.id, result: encoded }); return; } throw new Error("Unknown command"); } catch (err) { - const error = err instanceof Error ? err : new Error(String(err)); - - port.postMessage({ - error: { - message: error.message, - name: error.name, - ...(error.stack ? { stack: error.stack } : {}), - }, + post({ + error: toWireError(err), id: request.id, }); } diff --git a/src/test/fixtures/callback-actor.ts b/src/test/fixtures/callback-actor.ts new file mode 100644 index 0000000..393440b --- /dev/null +++ b/src/test/fixtures/callback-actor.ts @@ -0,0 +1,33 @@ +import { actor } from "../../index.js"; + +export class CallbackActor { + async withProgress( + n: number, + onProgress: (value: number) => void | Promise, + ): Promise { + let sum = 0; + + for (let i = 1; i <= n; i++) { + sum += i; + await onProgress(i); + } + + return sum; + } + + async callMaybe( + fn: () => number | Promise, + ): Promise { + return fn(); + } + + makeAdder(base: number): (x: number) => number { + return (x) => base + x; + } + + async boom(fn: () => void | Promise): Promise { + await fn(); + } +} + +actor(CallbackActor, import.meta); diff --git a/src/test/fixtures/closable.ts b/src/test/fixtures/closable.ts index 3a42b8b..afb47be 100644 --- a/src/test/fixtures/closable.ts +++ b/src/test/fixtures/closable.ts @@ -1,10 +1,14 @@ import { actor } from "../../index.js"; +/** Module-level probe shared by actors on the same worker. */ +let lastClosed = false; + export class Closable { closed = false; close(): void { this.closed = true; + lastClosed = true; } isClosed(): boolean { @@ -12,4 +16,15 @@ export class Closable { } } +export class ClosableProbe { + reset(): void { + lastClosed = false; + } + + wasLastClosed(): boolean { + return lastClosed; + } +} + actor(Closable, import.meta); +actor(ClosableProbe, import.meta); diff --git a/src/test/fixtures/nested-actors.ts b/src/test/fixtures/nested-actors.ts new file mode 100644 index 0000000..87c7732 --- /dev/null +++ b/src/test/fixtures/nested-actors.ts @@ -0,0 +1,17 @@ +import { actor } from "../../index.js"; + +export type HasValue = { + getValue: () => number | Promise; +}; + +export class NestedActors { + wrap(counter: HasValue): { counter: HasValue; label: string; } { + return { counter, label: "wrapped" }; + } + + async readWrapped(payload: { counter: HasValue; }): Promise { + return payload.counter.getValue(); + } +} + +actor(NestedActors, import.meta); diff --git a/src/test/fixtures/stream-actor.ts b/src/test/fixtures/stream-actor.ts new file mode 100644 index 0000000..51ceda2 --- /dev/null +++ b/src/test/fixtures/stream-actor.ts @@ -0,0 +1,64 @@ +import { Readable, Writable } from "node:stream"; + +import { actor } from "../../index.js"; + +export class StreamActor { + numbers(count: number): Readable { + let i = 0; + + return new Readable({ + objectMode: true, + read() { + if (i >= count) { + this.push(null); + + return; + } + this.push(i++); + }, + }); + } + + async collect(input: Readable): Promise { + const items: unknown[] = []; + + for await (const chunk of input) { + items.push(chunk); + } + + return items; + } + + sink(): { stream: Writable; done: () => Promise; } { + const items: unknown[] = []; + let resolveDone: (value: unknown[]) => void; + const done = new Promise((resolve) => { + resolveDone = resolve; + }); + + const stream = new Writable({ + final(cb) { + resolveDone!(items); + cb(); + }, + objectMode: true, + write(chunk, _enc, cb) { + items.push(chunk); + cb(); + }, + }); + + return { done: () => done, stream }; + } + + failingReadable(): Readable { + return new Readable({ + objectMode: true, + read() { + this.destroy(new Error("stream boom")); + }, + }); + } +} + +actor(StreamActor, import.meta); diff --git a/src/test/runtime.unit.spec.ts b/src/test/runtime.unit.spec.ts index cd75712..aeebfbd 100644 --- a/src/test/runtime.unit.spec.ts +++ b/src/test/runtime.unit.spec.ts @@ -1,3 +1,6 @@ +/* eslint-disable sort-imports */ +import { Readable } from "node:stream"; + import { afterAll, describe, @@ -6,11 +9,14 @@ import { } from "vitest"; import { Runtime, getActorHandle } from "../index.js"; -import { Closable } from "./fixtures/closable.js"; +import { CallbackActor } from "./fixtures/callback-actor.js"; +import { Closable, ClosableProbe } from "./fixtures/closable.js"; import { Counter } from "./fixtures/counter.js"; import { Linker } from "./fixtures/linker.js"; import { MailboxActor } from "./fixtures/mailbox-actor.js"; +import { NestedActors } from "./fixtures/nested-actors.js"; import { SlowActor } from "./fixtures/slow-actor.js"; +import { StreamActor } from "./fixtures/stream-actor.js"; describe("remote-objects", () => { const runtimes: Runtime[] = []; @@ -85,6 +91,28 @@ describe("remote-objects", () => { await expect(counter.inc()).rejects.toThrow(/Unknown object/); }); + it("closes actor on destroy by default", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const probe = await runtime.spawn(ClosableProbe); + + await probe.reset(); + const actor = await runtime.spawn(Closable); + + await runtime.destroy(actor); + expect(await probe.wasLastClosed()).toBe(true); + }); + + it("skips close on destroy when close:false", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const probe = await runtime.spawn(ClosableProbe); + + await probe.reset(); + const actor = await runtime.spawn(Closable); + + await runtime.destroy(actor, { close: false }); + expect(await probe.wasLastClosed()).toBe(false); + }); + it("rejects work after dispose", async () => { const runtime = new Runtime({ workers: 1 }); const counter = await runtime.spawn(Counter, 0); @@ -106,8 +134,18 @@ describe("remote-objects", () => { it("times out slow calls", async () => { const runtime = track(new Runtime({ callTimeoutMs: 50, workers: 1 })); const slow = await runtime.spawn(SlowActor); + const events: string[] = []; + + const runtime2 = track(new Runtime({ + callTimeoutMs: 50, + debug: (e) => events.push(e.type), + workers: 1, + })); + const slow2 = await runtime2.spawn(SlowActor); await expect(slow.wait(200)).rejects.toThrow(/timed out/); + await expect(slow2.wait(200)).rejects.toThrow(/timed out/); + expect(events).toContain("call:timeout"); }); it("passes actor refs as method arguments", async () => { @@ -128,6 +166,17 @@ describe("remote-objects", () => { expect(await linker.readOther()).toBe(7); }); + it("deep-encodes nested actor refs in plain objects", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const counter = await runtime.spawn(Counter, 3); + const nested = await runtime.spawn(NestedActors); + const wrapped = await nested.wrap(counter); + + expect(wrapped.label).toBe("wrapped"); + expect(await nested.readWrapped(wrapped)).toBe(3); + expect(await wrapped.counter.getValue()).toBe(3); + }); + it("serializes calls per actor mailbox", async () => { const runtime = track(new Runtime({ workers: 1 })); const actor = await runtime.spawn(MailboxActor); @@ -139,4 +188,97 @@ describe("remote-objects", () => { expect(results).toEqual([1, 2, 3]); }); + + it("invokes progress callbacks passed as arguments", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const actor = await runtime.spawn(CallbackActor); + const seen: number[] = []; + + const sum = await actor.withProgress(4, async (v) => { + seen.push(v); + }); + + expect(sum).toBe(10); + expect(seen).toEqual([1, 2, 3, 4]); + }); + + it("propagates errors from callbacks", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const actor = await runtime.spawn(CallbackActor); + + await expect(actor.boom(async () => { + throw new Error("cb fail"); + })).rejects.toThrow(/cb fail/); + }); + + it("returns callable callbacks from actors", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const actor = await runtime.spawn(CallbackActor); + const add = await actor.makeAdder(10); + + expect(await add(5)).toBe(15); + }); + + it("invokes callbacks across workers", async () => { + const runtime = track(new Runtime({ workers: 2 })); + const actor = await runtime.spawn(CallbackActor); // worker 0 + const linker = await runtime.spawn(Linker); // worker 1 — just to use 2 workers + + void linker; + + const seen: number[] = []; + const sum = await actor.withProgress(3, (v) => { + seen.push(v); + }); + + expect(sum).toBe(6); + expect(seen).toEqual([1, 2, 3]); + }); + + it("streams readable results from actors", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const actor = await runtime.spawn(StreamActor); + const stream = await actor.numbers(5); + const items: unknown[] = []; + + for await (const chunk of stream as Readable) { + items.push(chunk); + } + + expect(items).toEqual([0, 1, 2, 3, 4]); + }); + + it("accepts readable streams as arguments", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const actor = await runtime.spawn(StreamActor); + const input = Readable.from([10, 20, 30], { objectMode: true }); + const items = await actor.collect(input); + + expect(items).toEqual([10, 20, 30]); + }); + + it("accepts writable streams as nested return values", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const actor = await runtime.spawn(StreamActor); + const { done, stream } = await actor.sink(); + + stream.write(1); + stream.write(2); + stream.end(); + expect(await done()).toEqual([1, 2]); + }); + + it("rejects circular structures with a clear error", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const counter = await runtime.spawn(Counter, 0); + const circular: { self?: unknown; } = {}; + + circular.self = circular; + + await expect( + (counter as unknown as { add: (n: unknown) => Promise; }).add( + circular, + ), + ).rejects.toThrow(/Circular references/); + }); }); diff --git a/vitest.unit.config.js b/vitest.unit.config.js index 488f470..8f10bf8 100644 --- a/vitest.unit.config.js +++ b/vitest.unit.config.js @@ -6,5 +6,20 @@ export default defineConfig({ // Runtime/worker tests need compiled ESM (worker loads build/esm + actor file: URLs). include: ["build/esm/**/*.unit.spec.js"], setupFiles: ["./vitest.setup.js"], + coverage: { + provider: "v8", + include: ["build/esm/lib/**/*.js"], + exclude: [ + "build/esm/lib/worker/**", + "build/esm/lib/index.js", + ], + reporter: ["text", "lcov"], + thresholds: { + lines: 65, + functions: 75, + branches: 50, + statements: 65, + }, + }, }, });