feat: add callbacks, Node streams, and deep wire encoding

.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

.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

.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

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,27 +81,79 @@ 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 });
 ```
 
 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
 

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,27 +81,79 @@ 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 });
 ```
 
 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
 

eslint.config.js

@@ -9,6 +9,7 @@ export default tseslint.config(
 	{
 		ignores: [
 			"build/**",
+			"coverage/**",
 			"docs/**",
 			"eslint.config.js",
 			"vitest.*.js",

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": {

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<void>,
+	): Promise<number> {
+		for (let i = 0; i < count; i++) {
+			await onRow({ id: i, value: i * i });
+		}
+
+		return count;
+	}
+}
+
+actor(Streamer, import.meta);

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<void>): Promise<void> {
+	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<void>((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();

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();
+}