diff --git a/README.md b/README.md index 8e77a47..4f20bfd 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,7 @@ const { Runtime, actor } = require("@js-ak/remote-objects"); // CJS ## Core ideas - **`runtime.spawn(Class, ...args)`** creates an actor on a worker and returns a typed proxy +- **`runtime.getOrSpawn(key, Class, ...args)`** returns one actor per key (same proxy until `destroy`); placement is `hash(key) % workers` - **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 @@ -69,7 +70,7 @@ actor(Database, __filename); module.exports = { Database }; ``` -`spawn` auto-registers the class on first use (or call `runtime.register(Database)` explicitly). +`spawn` and `getOrSpawn` auto-register the class on first use (or call `runtime.register(Database)` explicitly). ## Options @@ -81,7 +82,7 @@ new Runtime({ }); ``` -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"`. +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"`. **`getOrSpawn` emits the same `spawn` event** when it creates a new actor (cache hits do not spawn again). ## Lifecycle @@ -92,7 +93,7 @@ await runtime.dispose(); // close actors, drain, terminate workers await runtime.dispose({ closeActors: false }); ``` -After `dispose`, further `spawn` / method calls fail with a clear error. +After `dispose`, further `spawn`, `getOrSpawn`, or method calls fail with a clear error. ## Passing actors as arguments @@ -145,6 +146,54 @@ Use sticky actors when a worker should own long-lived state (DB pools, SDK clien Same-actor calls are serialized (mailbox). Different actors may run in parallel on the pool. +## When to use what + +You do not need to predict every app up front — pick a pattern from what you are doing: + +| You want… | Use | +|-----------|-----| +| Long-lived client state (DB pool, SDK session, in-memory cache) | One actor per resource; or **`getOrSpawn(key, ...)`** for one actor per tenant/shard | +| More throughput on CPU or I/O | `workers: 2+` and **separate** actor instances (each spawn → least-loaded worker) | +| Strict ordering for one object | One actor — overlapping calls on the same proxy are queued (mailbox) | +| Parallel work on the same class | Multiple `spawn`s, or `getOrSpawn` + extra `spawn`s (see `examples/db.ts`) | +| Progress / one-off handlers during a call | Callback in **args** (released when the call finishes) | +| Long-lived handler returned from a method | Callback in **return value** (released on `destroy` of the owning actor) | +| Many rows or chunked I/O | Streams as args or results (backpressure built in) | +| Compose actors (even on different workers) | Pass actor proxies as method arguments | +| Shut down one resource | `destroy(proxy)` — runs `close`/`dispose` on the actor by default | +| Shut down the whole runtime | `dispose()` — drain in-flight work, then terminate workers | + +**Worker count** + +- **`workers: 1`** — simplest mental model; all actors share one thread. Good for getting started or when isolation from the main thread is enough. +- **`workers: 2+`** — use when you want parallel actors on separate threads. Each new spawn is placed on the worker with the fewest live actors and in-flight requests; after that the actor stays sticky on that worker. + +**One actor vs many** + +- **One proxy, many calls** — state and side effects stay in one place; calls do not overlap on that instance. +- **Many proxies, same class** — independent state and true parallelism (e.g. two query actors, two connection pools). + +**One actor per tenant / key** + +Use **`getOrSpawn(key, Class, ...args)`** — the runtime keeps one proxy per key until `destroy`. The worker is chosen by `hash(key) % workers` and stays stable for that key. Reusing a key with a different class or constructor args throws. + +```ts +const db = await runtime.getOrSpawn(`tenant:${tenantId}`, Database, creds); +// later, same process → same proxy / same pool +const same = await runtime.getOrSpawn(`tenant:${tenantId}`, Database, creds); +expect(same).toBe(db); + +await runtime.destroy(db); // drops the key; next getOrSpawn creates a fresh actor +``` + +For **many independent instances** of the same class (parallel pools), use **`spawn`** instead. **`spawn` does not register a key** — `spawn(Database, creds)` and `getOrSpawn("db", Database, creds)` are always separate actors. + +**Not a fit** + +- Fire-and-forget stateless jobs on a pile of plain data → a task pool (e.g. Piscina) may be simpler. +- Shared mutable state on the main thread with no isolation → you do not need actors at all. +- Moving an existing actor to another worker after spawn → not supported; spawn again if you need a new placement. + ## Compared to similar tools | | remote-objects | Comlink | Piscina | @@ -170,6 +219,10 @@ const handle = getActorHandle(counter); // { workerId, objectId } | undefined - 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 +- **`getOrSpawn` keys are per-runtime (in-memory)** — not shared across processes or runtimes +- **`spawn` and `getOrSpawn` use separate registries** — a keyed name does not attach to an actor created with `spawn` +- **Key placement is `hash(key) % workers`** — changing the worker pool size can move a key to a different worker on the next create (after `destroy`) +- **Repeated `getOrSpawn` compares constructor args** using deep equality for primitives, arrays, and plain objects only (not `Date`, `Map`, class instances, etc.) ## License diff --git a/docs/index.md b/docs/index.md index 8e77a47..4f20bfd 100644 --- a/docs/index.md +++ b/docs/index.md @@ -39,6 +39,7 @@ const { Runtime, actor } = require("@js-ak/remote-objects"); // CJS ## Core ideas - **`runtime.spawn(Class, ...args)`** creates an actor on a worker and returns a typed proxy +- **`runtime.getOrSpawn(key, Class, ...args)`** returns one actor per key (same proxy until `destroy`); placement is `hash(key) % workers` - **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 @@ -69,7 +70,7 @@ actor(Database, __filename); module.exports = { Database }; ``` -`spawn` auto-registers the class on first use (or call `runtime.register(Database)` explicitly). +`spawn` and `getOrSpawn` auto-register the class on first use (or call `runtime.register(Database)` explicitly). ## Options @@ -81,7 +82,7 @@ new Runtime({ }); ``` -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"`. +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"`. **`getOrSpawn` emits the same `spawn` event** when it creates a new actor (cache hits do not spawn again). ## Lifecycle @@ -92,7 +93,7 @@ await runtime.dispose(); // close actors, drain, terminate workers await runtime.dispose({ closeActors: false }); ``` -After `dispose`, further `spawn` / method calls fail with a clear error. +After `dispose`, further `spawn`, `getOrSpawn`, or method calls fail with a clear error. ## Passing actors as arguments @@ -145,6 +146,54 @@ Use sticky actors when a worker should own long-lived state (DB pools, SDK clien Same-actor calls are serialized (mailbox). Different actors may run in parallel on the pool. +## When to use what + +You do not need to predict every app up front — pick a pattern from what you are doing: + +| You want… | Use | +|-----------|-----| +| Long-lived client state (DB pool, SDK session, in-memory cache) | One actor per resource; or **`getOrSpawn(key, ...)`** for one actor per tenant/shard | +| More throughput on CPU or I/O | `workers: 2+` and **separate** actor instances (each spawn → least-loaded worker) | +| Strict ordering for one object | One actor — overlapping calls on the same proxy are queued (mailbox) | +| Parallel work on the same class | Multiple `spawn`s, or `getOrSpawn` + extra `spawn`s (see `examples/db.ts`) | +| Progress / one-off handlers during a call | Callback in **args** (released when the call finishes) | +| Long-lived handler returned from a method | Callback in **return value** (released on `destroy` of the owning actor) | +| Many rows or chunked I/O | Streams as args or results (backpressure built in) | +| Compose actors (even on different workers) | Pass actor proxies as method arguments | +| Shut down one resource | `destroy(proxy)` — runs `close`/`dispose` on the actor by default | +| Shut down the whole runtime | `dispose()` — drain in-flight work, then terminate workers | + +**Worker count** + +- **`workers: 1`** — simplest mental model; all actors share one thread. Good for getting started or when isolation from the main thread is enough. +- **`workers: 2+`** — use when you want parallel actors on separate threads. Each new spawn is placed on the worker with the fewest live actors and in-flight requests; after that the actor stays sticky on that worker. + +**One actor vs many** + +- **One proxy, many calls** — state and side effects stay in one place; calls do not overlap on that instance. +- **Many proxies, same class** — independent state and true parallelism (e.g. two query actors, two connection pools). + +**One actor per tenant / key** + +Use **`getOrSpawn(key, Class, ...args)`** — the runtime keeps one proxy per key until `destroy`. The worker is chosen by `hash(key) % workers` and stays stable for that key. Reusing a key with a different class or constructor args throws. + +```ts +const db = await runtime.getOrSpawn(`tenant:${tenantId}`, Database, creds); +// later, same process → same proxy / same pool +const same = await runtime.getOrSpawn(`tenant:${tenantId}`, Database, creds); +expect(same).toBe(db); + +await runtime.destroy(db); // drops the key; next getOrSpawn creates a fresh actor +``` + +For **many independent instances** of the same class (parallel pools), use **`spawn`** instead. **`spawn` does not register a key** — `spawn(Database, creds)` and `getOrSpawn("db", Database, creds)` are always separate actors. + +**Not a fit** + +- Fire-and-forget stateless jobs on a pile of plain data → a task pool (e.g. Piscina) may be simpler. +- Shared mutable state on the main thread with no isolation → you do not need actors at all. +- Moving an existing actor to another worker after spawn → not supported; spawn again if you need a new placement. + ## Compared to similar tools | | remote-objects | Comlink | Piscina | @@ -170,6 +219,10 @@ const handle = getActorHandle(counter); // { workerId, objectId } | undefined - 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 +- **`getOrSpawn` keys are per-runtime (in-memory)** — not shared across processes or runtimes +- **`spawn` and `getOrSpawn` use separate registries** — a keyed name does not attach to an actor created with `spawn` +- **Key placement is `hash(key) % workers`** — changing the worker pool size can move a key to a different worker on the next create (after `destroy`) +- **Repeated `getOrSpawn` compares constructor args** using deep equality for primitives, arrays, and plain objects only (not `Date`, `Map`, class instances, etc.) ## License diff --git a/src/examples/db.ts b/src/examples/db.ts index 6a822aa..9694841 100644 --- a/src/examples/db.ts +++ b/src/examples/db.ts @@ -11,7 +11,8 @@ const creds: DbCreds = { const runtime = new Runtime({ debug: true, workers: 2 }); -const db = await runtime.spawn(Database, creds); +// one pool for the app — reuse via getOrSpawn("main") from anywhere in-process +const db = await runtime.getOrSpawn("main", Database, creds); try { const one = await db.queryOne<{ n: number; }>("SELECT 1::int AS n"); @@ -22,7 +23,7 @@ try { console.log("version ->", version?.version); - // two actors → round-robin across workers, each with its own pool + // second spawn → separate pool for parallel queries (load-balanced worker) const db2 = await runtime.spawn(Database, creds); const [a, b] = await Promise.all([ db.queryOne("SELECT 42::int AS worker_probe"), @@ -31,6 +32,11 @@ try { console.log("parallel ->", { a, b }); + // same key → same proxy as db (spawn above does not register "main") + const same = await runtime.getOrSpawn("main", Database, creds); + + console.log("getOrSpawn reuse ->", same === db); + await db2.close(); } finally { await db.close(); diff --git a/src/lib/index.ts b/src/lib/index.ts index 87730b8..a925230 100644 --- a/src/lib/index.ts +++ b/src/lib/index.ts @@ -2,7 +2,8 @@ * 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. + * bind them with {@link actor}, spawn or {@link Runtime.getOrSpawn} via + * {@link Runtime}, call through typed proxies. */ export type { ActorClass, diff --git a/src/lib/protocol/callback-registry.ts b/src/lib/protocol/callback-registry.ts index 12e606b..f52a859 100644 --- a/src/lib/protocol/callback-registry.ts +++ b/src/lib/protocol/callback-registry.ts @@ -18,6 +18,8 @@ export type CallbackOwner = "host" | number; */ export class CallbackRegistry { private readonly entries = new Map(); + /** Reverse index: actor objectId → callback ids bound to it. */ + private readonly boundByObject = new Map>(); private nextId = 1; private readonly owner: CallbackOwner; @@ -49,6 +51,9 @@ export class CallbackRegistry { } this.entries.set(callbackId, entry); + if (entry.boundObjectId !== undefined) { + this.trackBound(entry.boundObjectId, callbackId); + } return callbackRef(this.owner, callbackId); } @@ -84,7 +89,7 @@ export class CallbackRegistry { */ release(callbackIds: Iterable): void { for (const id of callbackIds) { - this.entries.delete(id); + this.removeEntry(id); } } @@ -97,25 +102,71 @@ export class CallbackRegistry { const entry = this.entries.get(id); if (entry?.callScoped) { - this.entries.delete(id); + this.removeEntry(id); } } } /** * Drops callbacks returned by a given actor (on destroy). + * Uses a reverse index for O(k) cleanup where k is callbacks bound to the actor. * @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); - } + const ids = this.boundByObject.get(objectId); + + if (!ids) return; + for (const id of ids) { + this.entries.delete(id); } + this.boundByObject.delete(objectId); } /** Removes every registered callback. */ clear(): void { this.entries.clear(); + this.boundByObject.clear(); + } + + /** + * Drops one entry and keeps {@link boundByObject} in sync. + * @param callbackId - Id to remove + */ + private removeEntry(callbackId: number): void { + const entry = this.entries.get(callbackId); + + if (!entry) return; + if (entry.boundObjectId !== undefined) { + this.untrackBound(entry.boundObjectId, callbackId); + } + this.entries.delete(callbackId); + } + + /** + * @param objectId - Actor that owns returned callbacks + * @param callbackId - Registered callback id + */ + private trackBound(objectId: number, callbackId: number): void { + let ids = this.boundByObject.get(objectId); + + if (!ids) { + ids = new Set(); + this.boundByObject.set(objectId, ids); + } + ids.add(callbackId); + } + + /** + * @param objectId - Actor that owned the callback + * @param callbackId - Registered callback id + */ + private untrackBound(objectId: number, callbackId: number): void { + const ids = this.boundByObject.get(objectId); + + if (!ids) return; + ids.delete(callbackId); + if (ids.size === 0) { + this.boundByObject.delete(objectId); + } } } diff --git a/src/lib/runtime/runtime.ts b/src/lib/runtime/runtime.ts index d234302..c4ff8c2 100644 --- a/src/lib/runtime/runtime.ts +++ b/src/lib/runtime/runtime.ts @@ -11,6 +11,7 @@ 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 { isPlainObject } from "../protocol/plain.js"; import { StreamBridge } from "../protocol/stream-bridge.js"; import { createHostStreamTransport } from "./host-stream-transport.js"; @@ -19,6 +20,53 @@ import { Scheduler } from "./scheduler.js"; import { StreamRouter } from "./stream-router.js"; import { WorkerNode } from "./worker-node.js"; +/** Host-side singleton entry for {@link Runtime.getOrSpawn}. */ +type KeyedEntry = { + proxy: object; + className: string; + args: unknown[]; +}; + +/** + * Deep equality for structured-clone constructor args (primitives, arrays, plain objects). + * @param a - First value + * @param b - Second value + * @returns Whether both values are deeply equal + */ +function valueEqual(a: unknown, b: unknown): boolean { + if (a === b) return true; + if (a === null || b === null) return false; + if (typeof a !== typeof b) return false; + + if (Array.isArray(a) && Array.isArray(b)) { + if (a.length !== b.length) return false; + + return a.every((item, index) => valueEqual(item, b[index])); + } + + if (isPlainObject(a) && isPlainObject(b)) { + const keysA = Object.keys(a); + const keysB = Object.keys(b); + + if (keysA.length !== keysB.length) return false; + + return keysA.every((key) => valueEqual(a[key], b[key])); + } + + return false; +} + +/** + * @param a - Args from an existing keyed entry + * @param b - Args from a new {@link Runtime.getOrSpawn} call + * @returns Whether constructor argument lists match + */ +function argsEqual(a: unknown[], b: unknown[]): boolean { + if (a.length !== b.length) return false; + + return a.every((item, index) => valueEqual(item, b[index])); +} + /** * Normalizes {@link RuntimeOptions.debug} into a single event handler. * @param debug - Boolean, function, or `{ onEvent }` form @@ -43,6 +91,8 @@ function resolveDebug(debug: RuntimeOptions["debug"]): DebugHandler | undefined /** * Owns a pool of worker threads, schedules sticky actors, and exposes * typed proxies for remote method calls (including callbacks and streams). + * Use {@link spawn} for new actors (load-balanced) or {@link getOrSpawn} for + * one actor per key until {@link destroy}. */ export class Runtime { private readonly registry = new Registry(); @@ -53,6 +103,8 @@ export class Runtime { private readonly streamRouter = new StreamRouter(); private readonly hostStreams: StreamBridge; private disposed = false; + /** One actor proxy per spawn key ({@link Runtime.getOrSpawn}). */ + private readonly keyed = new Map(); /** * @param options - Pool size, debug hooks, and optional call timeout @@ -126,7 +178,7 @@ export class Runtime { } /** - * Spawns an actor on a worker (round-robin) and returns a typed proxy. + * Spawns an actor on the least-loaded worker and returns a typed proxy. * Auto-registers the class on first use if needed. * * @param Class - Actor class @@ -145,11 +197,75 @@ export class Runtime { return this.scheduler.create(Class, args); } + /** + * Returns an existing actor for `key`, or spawns one on the worker chosen by + * `hash(key) % workers`. The same key always maps to the same proxy until + * {@link destroy} removes it. Reusing a key with a different class or + * constructor args throws. + * + * Auto-registers the class on first use if needed. + * + * @param key - Non-empty affinity key (tenant id, shard name, etc.) + * @param Class - Actor class + * @param args - Constructor arguments (must match on later calls for this key) + * @returns Typed proxy; all methods are async from the caller side + * @throws If `key` is empty, or the key is already bound to another class or args + */ + async getOrSpawn( + key: string, + Class: C, + ...args: ConstructorParameters + ): Promise>> { + this.assertOpen("getOrSpawn"); + + if (key.length === 0) { + throw new Error("getOrSpawn key must be a non-empty string"); + } + + const existing = this.keyed.get(key); + + if (existing) { + if (existing.className !== Class.name) { + throw new Error( + `getOrSpawn key "${key}" is already bound to ${existing.className}, not ${Class.name}`, + ); + } + + if (!argsEqual(existing.args, args as unknown[])) { + throw new Error( + `getOrSpawn key "${key}" was created with different constructor arguments`, + ); + } + + return existing.proxy as ActorProxy>; + } + + if (!this.registry.has(Class.name)) { + await this.register(Class); + } + + const worker = this.scheduler.pickForKey(key); + const proxy = await this.scheduler.createOn( + worker, + Class, + args, + ); + + this.keyed.set(key, { + args: args as unknown[], + className: Class.name, + proxy, + }); + + return proxy; + } + /** * Removes an actor from its worker. Further method calls on the proxy fail. * By default calls `dispose`/`close` on the actor first. + * Also drops a {@link getOrSpawn} entry when the destroyed proxy matches. * - * @param proxy - Proxy returned by {@link spawn} + * @param proxy - Proxy returned by {@link spawn} or {@link getOrSpawn} * @param options - Pass `{ close: false }` to skip actor cleanup */ async destroy(proxy: object, options?: DestroyOptions): Promise { @@ -169,6 +285,21 @@ export class Runtime { } await worker.destroy(handle.objectId, options); + this.releaseKeyedProxy(proxy); + } + + /** + * Drops a {@link getOrSpawn} entry when its proxy is destroyed. + * @param proxy - Actor proxy being removed + */ + private releaseKeyedProxy(proxy: object): void { + for (const [key, entry] of this.keyed) { + if (entry.proxy === proxy) { + this.keyed.delete(key); + + return; + } + } } /** @@ -197,5 +328,6 @@ export class Runtime { this.hostStreams.closeAll(); this.hostCallbacks.clear(); this.streamRouter.clear(); + this.keyed.clear(); } } diff --git a/src/lib/runtime/scheduler.ts b/src/lib/runtime/scheduler.ts index e94cbb1..77d14d1 100644 --- a/src/lib/runtime/scheduler.ts +++ b/src/lib/runtime/scheduler.ts @@ -1,13 +1,24 @@ import type { ActorProxy, AnyActorClass } from "../types.js"; import type { WorkerNode } from "./worker-node.js"; +/** FNV-1a 32-bit hash for stable worker placement by spawn key. */ +function hashKey(key: string): number { + let h = 2_166_136_261; + + for (let i = 0; i < key.length; i++) { + h ^= key.charCodeAt(i); + h = Math.imul(h, 1_677_761_9); + } + + return h >>> 0; +} + /** - * Round-robin placement of new actors across the worker pool. - * After spawn, an actor stays sticky on the chosen worker. + * Load-aware placement of new actors across the worker pool. + * Picks the worker with the fewest live actors and in-flight requests; + * ties break on the lowest worker id. After spawn, an actor stays sticky. */ export class Scheduler { - private next = 0; - /** * @param workers - Non-empty list of {@link WorkerNode}s */ @@ -18,13 +29,42 @@ export class Scheduler { } /** - * Picks the next worker in round-robin order. + * Picks the least-loaded worker for the next spawn. * @returns Worker that will host the next spawned actor */ pick(): WorkerNode { - const worker = this.workers[this.next % this.workers.length]; + let best = this.workers[0]; + + if (!best) { + throw new Error("No worker available"); + } + + let bestLoad = best.getSchedulingLoad(); + + for (let i = 1; i < this.workers.length; i++) { + const worker = this.workers[i]; + + if (!worker) continue; + const load = worker.getSchedulingLoad(); + + if (load < bestLoad || (load === bestLoad && worker.id < best.id)) { + best = worker; + bestLoad = load; + } + } + + return best; + } + + /** + * Picks a worker deterministically from a spawn key (stable across the pool size). + * @param key - Non-empty affinity key (tenant id, shard name, etc.) + * @returns Worker that will host actors for this key + */ + pickForKey(key: string): WorkerNode { + const index = hashKey(key) % this.workers.length; + const worker = this.workers[index]; - this.next += 1; if (!worker) { throw new Error("No worker available"); } @@ -46,4 +86,19 @@ export class Scheduler { return worker.create(Class.name, args as unknown[]); } + + /** + * Spawns an actor on a specific worker and returns a typed proxy. + * @param worker - Target {@link WorkerNode} + * @param Class - Actor class (already registered on workers) + * @param args - Constructor arguments + * @returns Typed proxy for the new actor + */ + async createOn( + worker: WorkerNode, + Class: C, + args: ConstructorParameters, + ): Promise>> { + return worker.create(Class.name, args as unknown[]); + } } diff --git a/src/lib/runtime/worker-node.ts b/src/lib/runtime/worker-node.ts index ee6d62a..68bb7c1 100644 --- a/src/lib/runtime/worker-node.ts +++ b/src/lib/runtime/worker-node.ts @@ -86,6 +86,15 @@ export class WorkerNode { private readonly mailboxes = new Map>(); private drainWaiters: Array<() => void> = []; + /** + * Live actors plus in-flight host↔worker requests on this node. + * Used by {@link Scheduler} for load-aware spawn placement. + * @returns Scheduling weight (lower = preferred for new actors) + */ + getSchedulingLoad(): number { + return this.localObjects.size + this.pending.size; + } + /** * Spawns the ESM worker entry and wires message / error handlers. * @param options - Pool identity, shared registries, timeouts diff --git a/src/test/callback-registry.unit.spec.ts b/src/test/callback-registry.unit.spec.ts new file mode 100644 index 0000000..9f2c9c3 --- /dev/null +++ b/src/test/callback-registry.unit.spec.ts @@ -0,0 +1,51 @@ +import { + describe, expect, it, +} from "vitest"; + +import { CallbackRegistry } from "../lib/protocol/callback-registry.js"; + +describe("CallbackRegistry", () => { + it("releaseBoundToObject drops only callbacks bound to that actor", () => { + const registry = new CallbackRegistry("host"); + const keep = registry.register(() => undefined); + const dropA1 = registry.register(() => undefined, { boundObjectId: 1 }); + const dropA2 = registry.register(() => undefined, { boundObjectId: 1 }); + const dropB = registry.register(() => undefined, { boundObjectId: 2 }); + + registry.releaseBoundToObject(1); + + expect(registry.get(keep.callbackId)).toBeDefined(); + expect(registry.get(dropA1.callbackId)).toBeUndefined(); + expect(registry.get(dropA2.callbackId)).toBeUndefined(); + expect(registry.get(dropB.callbackId)).toBeDefined(); + }); + + it("release and releaseCallScoped keep the bound index consistent", () => { + const registry = new CallbackRegistry("host"); + const scoped = registry.register(() => undefined, { + boundObjectId: 7, + callScoped: true, + }); + const bound = registry.register(() => undefined, { boundObjectId: 7 }); + + registry.releaseCallScoped([scoped.callbackId]); + expect(registry.get(bound.callbackId)).toBeDefined(); + + registry.release([bound.callbackId]); + registry.releaseBoundToObject(7); + expect(registry.get(bound.callbackId)).toBeUndefined(); + }); + + it("clear removes every entry", () => { + const registry = new CallbackRegistry(0); + const a = registry.register(() => undefined, { boundObjectId: 3 }); + const b = registry.register(() => undefined); + + registry.clear(); + + expect(registry.get(a.callbackId)).toBeUndefined(); + expect(registry.get(b.callbackId)).toBeUndefined(); + registry.releaseBoundToObject(3); + expect(registry.get(b.callbackId)).toBeUndefined(); + }); +}); diff --git a/src/test/runtime.unit.spec.ts b/src/test/runtime.unit.spec.ts index aeebfbd..9b7efec 100644 --- a/src/test/runtime.unit.spec.ts +++ b/src/test/runtime.unit.spec.ts @@ -40,7 +40,7 @@ describe("remote-objects", () => { expect(await counter.getValue()).toBe(15); }); - it("keeps actors sticky and round-robins spawn", async () => { + it("keeps actors sticky and load-balances spawn", async () => { const runtime = track(new Runtime({ workers: 2 })); const a = await runtime.spawn(Counter, 0); const b = await runtime.spawn(Counter, 0); @@ -60,6 +60,78 @@ describe("remote-objects", () => { expect(getActorHandle(b)?.workerId).toBe(1); }); + it("getOrSpawn returns the same proxy for the same key", async () => { + const runtime = track(new Runtime({ workers: 2 })); + const a = await runtime.getOrSpawn("tenant-a", Counter, 10); + const b = await runtime.getOrSpawn("tenant-a", Counter, 10); + + expect(a).toBe(b); + expect(getActorHandle(a)?.objectId).toBe(getActorHandle(b)?.objectId); + await a.inc(); + expect(await b.getValue()).toBe(11); + }); + + it("getOrSpawn pins a key to a stable worker", async () => { + const runtime = track(new Runtime({ workers: 2 })); + const first = await runtime.getOrSpawn("shard-1", Counter, 0); + const handle = getActorHandle(first); + + expect(handle).toBeTruthy(); + await runtime.destroy(first); + const second = await runtime.getOrSpawn("shard-1", Counter, 0); + + expect(getActorHandle(second)?.workerId).toBe(handle?.workerId); + expect(getActorHandle(second)?.objectId).not.toBe(handle?.objectId); + }); + + it("getOrSpawn rejects mismatched class or args for a key", async () => { + const runtime = track(new Runtime({ workers: 1 })); + + await runtime.getOrSpawn("k", Counter, 1); + + await expect(runtime.getOrSpawn("k", Counter, 2)).rejects.toThrow( + /different constructor arguments/, + ); + await expect( + runtime.getOrSpawn("k", MailboxActor), + ).rejects.toThrow(/already bound to Counter/); + }); + + it("destroy drops getOrSpawn entries", async () => { + const runtime = track(new Runtime({ workers: 1 })); + const first = await runtime.getOrSpawn("tenant", Counter, 0); + + await runtime.destroy(first); + const second = await runtime.getOrSpawn("tenant", Counter, 0); + + expect(second).not.toBe(first); + expect(getActorHandle(second)?.objectId).not.toBe( + getActorHandle(first)?.objectId, + ); + }); + + it("getOrSpawn rejects an empty key", async () => { + const runtime = track(new Runtime({ workers: 1 })); + + await expect(runtime.getOrSpawn("", Counter, 0)).rejects.toThrow( + /non-empty string/, + ); + }); + + it("spawn and getOrSpawn with the same key are separate actors", async () => { + const runtime = track(new Runtime({ workers: 2 })); + const spawned = await runtime.spawn(Counter, 0); + const keyed = await runtime.getOrSpawn("db", Counter, 0); + + expect(spawned).not.toBe(keyed); + const spawnedHandle = getActorHandle(spawned); + const keyedHandle = getActorHandle(keyed); + + expect( + `${spawnedHandle?.workerId}:${spawnedHandle?.objectId}`, + ).not.toBe(`${keyedHandle?.workerId}:${keyedHandle?.objectId}`); + }); + it("returns proxy for return this", async () => { const runtime = track(new Runtime({ workers: 1 })); const counter = await runtime.spawn(Counter, 0); diff --git a/src/test/scheduler.unit.spec.ts b/src/test/scheduler.unit.spec.ts new file mode 100644 index 0000000..49c0532 --- /dev/null +++ b/src/test/scheduler.unit.spec.ts @@ -0,0 +1,63 @@ +import { + describe, expect, it, +} from "vitest"; + +import { Scheduler } from "../lib/runtime/scheduler.js"; +import type { WorkerNode } from "../lib/runtime/worker-node.js"; + +function mockWorker(id: number, load: number): WorkerNode { + return { getSchedulingLoad: () => load, id } as WorkerNode; +} + +describe("Scheduler", () => { + it("picks the worker with the lowest scheduling load", () => { + const scheduler = new Scheduler([ + mockWorker(0, 3), + mockWorker(1, 1), + mockWorker(2, 2), + ]); + + expect(scheduler.pick().id).toBe(1); + }); + + it("breaks load ties on the lowest worker id", () => { + const scheduler = new Scheduler([ + mockWorker(2, 1), + mockWorker(0, 1), + mockWorker(1, 1), + ]); + + expect(scheduler.pick().id).toBe(0); + }); + + it("throws when the worker pool is empty", () => { + expect(() => new Scheduler([])).toThrow(/at least one worker/); + }); + + it("pickForKey is stable for the same key", () => { + const scheduler = new Scheduler([ + mockWorker(0, 0), + mockWorker(1, 0), + mockWorker(2, 0), + ]); + + expect(scheduler.pickForKey("tenant-42").id).toBe( + scheduler.pickForKey("tenant-42").id, + ); + }); + + it("pickForKey can target different workers for different keys", () => { + const scheduler = new Scheduler([ + mockWorker(0, 0), + mockWorker(1, 0), + mockWorker(2, 0), + ]); + const ids = new Set( + ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"].map( + (key) => scheduler.pickForKey(key).id, + ), + ); + + expect(ids.size).toBeGreaterThan(1); + }); +});