# PowerPool A small, dependency-free worker pool that wraps underlying Worker instances. It encodes plain object messages to transferable `Uint8Array` for efficient transfer, decodes incoming binary messages back to objects, and provides queuing / grow / reaper behavior. ## Quick paths ```js const pool = new PowerPool(workerSource, { size: 2, maxSize: 4 }); const result = await pool.request({ task: 'thumbnail', imageId: 42 }, { timeout: 5000 }); ``` Use `request(message, options)` when a response is required. Use `postMessage(message, transfer, options)` for fire-and-forget work; its positional transfer argument is retained for compatibility. Call `await pool.drain()` before shutdown when queued work must finish. ## Constructor | option | type | default | description | | ------------------------------------------------------- | --------------------------------------------------: | ---------------------------------------------: | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `workerSource` | `Function \| string` | — | Either a Worker factory/constructor (callable) or a relative path string passed to `new Worker(new URL(path, import.meta.url))`. | | `options.size` | `number` | `min(navigator.hardwareConcurrency \|\| 2, 2)` | Initial number of workers to spawn. | | `options.minSize` | `number` | `1` | Minimum workers to keep alive. | | `options.maxSize` | `number` | `Math.max(size, hwConcurrency)` | Maximum workers allowed in the pool. | | `options.workerOptions` | `Object` | `{}` | Options forwarded to the Worker constructor when using a string `workerSource`. | | `options.maxTasksPerWorker` | `number` | `Infinity` | Soft capacity per worker before it is considered busy. | | `options.idleTimeout` | `number` | `60000` | Milliseconds after which idle workers (beyond `minSize`) are terminated. | | `options.messageCodec` | `'framed' \| 'legacy' \| 'negotiated'` | `'framed'` | Wire protocol for object messages. `'framed'` posts a `PowerMessageCodec` envelope; `'legacy'` restores the 1.x bare-JSON body; `'negotiated'` behaves exactly like `'framed'` until a worker advertises the native carrier, then sends that carrier to that worker alone. See [Migrating to the framed protocol](#migrating-to-the-framed-protocol-breaking-change-in-20) and [Protocol negotiation](#protocol-negotiation-negotiated). | | `options.taskQueue` | `boolean` | `true` | Whether to queue tasks when pool is saturated. | | `options.queuePolicy` | `'enqueue'\|'drop-oldest'\|'drop-newest'\|'reject'` | `enqueue` | Policy to apply when the pool is saturated and the queue would otherwise grow. See the queue policy section below. | | `options.maxQueueLength` | `number`\|`'auto'` | `Infinity` | **Hard cap on queued tasks.** With the default `'enqueue'` policy and no cap, a saturated pool grows its queue until the process runs out of memory — the failure mode this option exists to make observable. Set a finite cap and see [Bounding the queue](#bounding-the-queue). | | `options.priority` | `number` | `0` | Default task priority for queued messages. Higher values are dispatched before lower values when the pool is saturated. Tasks with the same priority maintain FIFO order. Can be overridden per-message via `postMessage(msg, transfer, { priority })`. See [Task priority](#task-priority). | | `options.priorityAgingMs` | `number` | `0` | Waiting milliseconds required to gain one effective priority point. Set above `0` to prevent low-priority queued work from starving during sustained high-priority traffic. | | `options.recordTiming` | `boolean` | `false` | Opt in to percentile telemetry. When `true`, task duration and queue wait are recorded into `PowerHistogram`s and `getStats().performance` reports `percentiles` (p50/p90/p99/p99.9), `queueWaitPercentiles` and `utilization`. Off by default, and the default path allocates no histogram and pays one boolean test per recording site. See [Percentile telemetry](#percentile-telemetry). | | `options.maxDrainWaiters` | `number` | `100` | Maximum number of `drain()` calls that may be _waiting_ at once. Beyond it, `drain()` rejects with `ERR_POOL_DRAIN_TOO_MANY_WAITERS` instead of accumulating an unbounded number of `idle` listeners. | | `options.lazy` | `boolean` | `true` | When `true` defer creating workers up to `size` until demand; only `minSize` workers are created at construction. Use this for low-load deployments to avoid unnecessary worker startup cost. | | `options.listenerMaxListeners` / `options.maxListeners` | `number` | `0` (unlimited) | Maximum listeners per internal pool event (see notes). `0` means unlimited. If set to a positive number the pool will throw when registering additional listeners beyond that limit. | | `options.weakListeners` | `boolean` | `false` | When `true` the pool stores listeners as weak references (when supported by the runtime). This avoids retaining large closures but requires `FinalizationRegistry`/`WeakRef` support; you can call `pool._bus.cleanup()` to force cleanup of dead weak refs in environments without deterministic GC (primarily useful for tests). | | `options.autoScale` | `boolean \| Object` | `false` | When provided (or `true`), enables autoscaling. Supply `true` to use defaults, or an object to tune behavior. See the **Autoscaling** section below for properties and tuning recommendations. | | `options.idempotencyTtlMs` | `number` | `0` | Enables the idempotency ledger: `postMessage(msg, transfer, { idempotencyKey })` refuses to dispatch the same key twice. `0` means off and costs nothing. See [Idempotency keys](#idempotency-keys-idempotencyttlms). | ### Idempotency keys (`idempotencyTtlMs`) **A retried task can be dispatched twice, and nothing on the path can tell you so.** `PowerRetry` retries operations that may already have taken effect, and pool messages are exactly-once only _from the pool's side_: when a caller retries across a timeout, the first attempt may still be running. `ALGO-006` defers request hedging for exactly this reason — there was no idempotency story. Opt in with a retention window, then pass a key per post: ```js const pool = new PowerPool(workerFactory, { idempotencyTtlMs: 60_000 }); pool.postMessage({ task: 'charge', n: 1 }, undefined, { idempotencyKey: 'inv-1' }); // -> true pool.postMessage({ task: 'charge', n: 1 }, undefined, { idempotencyKey: 'inv-1' }); // -> false refused; the worker never saw it twice ``` `false` is the pool's existing "refused" answer, so a call site already handles it. Keys are coerced with `String()`, so `42` and `'42'` are one task. **It is opt-in because it costs a lookup per posted message, and that has to be earned.** `idempotencyTtlMs: 0` — the default — allocates no Map and takes no branch; `getStats().idempotency.lookups` stays `0`. With it on, `lookups` counts **every** post, keyed or not, because the cost being paid is a lookup per post and the only honest way to report it is lookups ÷ posts. A pool that passed no keys would otherwise make the feature look free for most of its traffic. **Why a TTL and not a boolean.** A boolean would need a default retention window invented for it, and a ledger that outlives its usefulness is a leak. Settled keys are forgotten after `idempotencyTtlMs`, so it must exceed the longest window in which a caller might retry — a caller that retries after the TTL is refused by nothing and applies its side effect twice, exactly as it would with no ledger at all. Expiry runs a **bounded** sweep (32 entries per post) off the back of the next post, so opting in does not make each post cost more the longer the process runs. #### In-flight versus settled The ledger has two states, and the difference is the whole design: | state | means | a repeat is | | ----------- | --------------------------------------------- | -------------------------- | | `in-flight` | claimed, not yet dispatched — nothing has run | a **concurrent duplicate** | | `settled` | dispatched; may already have applied | a **retry** | Both are refused, and both are counted separately in `getStats().idempotency.duplicatesInFlight` / `duplicatesSettled`. Conflating them into one `seen` set — the usual shape — loses the ability to say which happened, and loses the release below. A key moves to `settled` when the post is **accepted for dispatch**, including when it returns an `awaitResponse` promise. That is deliberate: a promise that later times out means the task was very likely running, and refusing the retry is the entire point of the feature. **A post the pool _refused_ releases its claim instead of settling it.** The task never ran, so leaving the key claimed would turn a transient queue-full into a permanent one — every retry of a task that never happened would be blocked until the TTL expired. In-flight claims are never expired by the sweep, for the same reason in the other direction: expiring one would let a still-running task be posted twice. #### What this does not cover `postMessage` only. `broadcast`, `postMessageBatch` and `stopThePress` have their own paths and do not consult the ledger, so a duplicate sent through `broadcast` is still dispatched twice. `getStats().idempotency` reports `enabled: false` on a pool that has not opted in, which is the first thing to check if a key appears to be ignored. ## API - `request(message, options)` — Send one message and await its worker response. It accepts the `postMessage` options plus `transfer`, and always returns a `Promise`. Use `postMessage` for fire-and-forget work or when the positional transfer-list API is preferable. - `postMessage(message, transfer, options)` — Dispatch a single message to the pool. Returns `true` when dispatched/queued successfully, or when `options.awaitResponse` (or `options.correlationId`) is present returns a `Promise` that resolves with the worker response. When sending plain objects the pool encodes them into a transferable `Uint8Array` automatically, wrapped in a [`PowerMessageCodec`](powerMessageCodec.md) envelope. **See [Migrating to the framed protocol](#migrating-to-the-framed-protocol-breaking-change-in-20) — this changed in 2.0 and requires a one-line edit in every worker.** - Pass `options.workerId` to route the message to a specific worker id; targeting a missing or saturated worker will fail (returns `false` or a rejected Promise). - When `options.taskQueue` is enabled, `options.queuePolicy` controls overload behavior: - `'enqueue'` (default) queues all overflow tasks — **without a `maxQueueLength` cap, without bound.** - `'drop-oldest'` drops the oldest queued task when new work arrives. - `'drop-newest'` drops the newest incoming task when there is already queued backlog. - `'reject'` rejects new overflow tasks immediately instead of queueing. - A refused task returns `false`, or — when the caller passed `awaitResponse` or an explicit `correlationId` — rejects with `code === 'ERR_POOL_QUEUE_FULL'`. See [Bounding the queue](#bounding-the-queue). - Note: `options.awaitResponse` requires the outgoing `message` to be a plain-object (not a TypedArray/ArrayBuffer). The implementation augments the object with a `correlationId` and will throw if a non-plain-object is supplied when `awaitResponse` is requested. - `options.workerId` may be a `number` or `string` (the pool coerces ids to strings internally for correlation handling). - `broadcast(message, transfer)` — Send `message` to every worker in the pool. Each worker receives either the provided transferable or an independently encoded `Uint8Array` when `transfer` is omitted and a plain object is provided. Broadcasting increments each worker's `tasks` counter. - `postMessageBatch(items, options)` — Enqueue or dispatch a batch of messages in a single call. `items` is an array of `{ message, transfer? }`. Returns an array of per-item results (booleans or Promises when `awaitResponse` is requested). Use this to amortize queue push overhead for many items. - When `options.awaitResponse` is enabled, each batch item is handled through the same internal `postMessage()` path as a single-item request. That means the batch preserves correlation behavior and returns Promises for response-waiting entries. - For stable per-item identity in response mode, pass `options.correlationIdFactory(index, item)` to generate a unique `correlationId` for each batch entry. - **The factory is resolved once, up front, and validated before anything is dispatched.** If it returns a duplicate id — within this batch, or one that is already in flight from an earlier call — the whole `postMessageBatch` throws `code === 'ERR_POOL_DUPLICATE_CORRELATION_ID'` and **nothing is sent**. This is deliberately stronger than the alternative: dispatching item-by-item would leave the first caller holding a promise that the second registration already rejected, with two messages already on the wire. The error names the duplicated id and its item index. Because the ids are resolved once, the factory is called exactly `items.length` times even if it is impure. - A fixed `options.correlationId` may only be used when the batch contains a single item. For multiple items the API throws because the pool cannot safely reuse one identifier for many pending responses. - Specifying `options.workerId` targets the batch to a single worker. Targeted batch dispatch is fail-fast: a missing or busy worker will not queue the batch item, and the corresponding return value is `false`. - When `options.taskQueue` is enabled, `options.queuePolicy` also applies to batch enqueue behavior in the fire-and-forget path. - The return array always matches `items.length`. - `prepareBuffers(items, { clone = false })` — Prepare an array of normalized `{ message, transfer }` entries for use with `postMessageBatch`. **The return value's shape depends on `clone` and on `messageCodec`** — see [what it returns](#what-preparebuffers-returns-and-why-it-is-not-a-uint8array). Under a framing codec the default returns each item unencoded and marked `deferred`, because a cached encode body cannot be both shared and correct on the wire; `clone: true` frames immediately. - `stopThePressBatch(items, options)` — Atomically clear the queue, terminate (and optionally recreate) inflight workers, reject pending awaitResponse promises, then forward the provided batch. Returns per-item results like `postMessageBatch`. Useful for emergency replacement of queued work with a new batch. - `stopThePress(message, transfer, options)` — Clear the internal queue, terminate running workers (rejecting pending response Promises), and send the provided `message` through the same dispatch semantics as `postMessage`. By default the pool will recreate replacement workers; pass `options.recreateWorkers = false` to keep the pool reduced. - `pauseQueue()` / `resumeQueue()` — Temporarily pause and resume dispatching tasks from the internal queue. Use `pauseQueue()` when downstream consumers are overloaded or a transient outage occurs; queued tasks are retained and resumed later. - `pause()` / `resume()` — Ergonomic aliases for `pauseQueue()` and `resumeQueue()`, respectively. - `queuePaused` — Read-only boolean property indicating whether queued dispatch is currently paused. - `addWorker()` / `removeWorker()` — Programmatically create or terminate a single worker from the pool. - Event APIs: `addEventListener(type, cb)` / `removeEventListener(type, cb)` — Manage listeners for `'message'`, `'error'`, `'messageerror'`, and `'idle'`. `idle` listeners are invoked immediately if the pool is currently idle. - `terminate()` — Immediately terminate all workers, clear queues, and stop the reaper interval. - `close({ force, timeout })` — **New in 2.1.0.** Graceful shutdown as a single entry point: wait for in-flight work, then terminate. A composition of `drain()` + `shutdown()`, both of which stay exactly as they are. See [Closing the pool](#closing-the-pool). - Disposal hooks: `[Symbol.dispose]()` calls `terminate()` synchronously; `[Symbol.asyncDispose]()` awaits `drain()` then terminates. - `getStats()` — Return a snapshot `{ status: Array<{id,tasks,lastActive}>, performance: Object, queueDepth, queuePressure }` with per-worker status and aggregated performance metrics (EWMA/time-per-task stats). `queuePressure` is normalized to `0..1` when `maxQueueLength` is finite. This is useful for logging and autoscale decisions. - `drain({ signal, timeout })` — Resolves with the pool's stats once the queue is empty and no task is in flight. The wait is bounded three ways, and **every one of them abandons the wait, never the work**: the pool keeps dispatching and keeps serving every other caller, because someone who stopped watching a drain does not get to stop the work. - `signal` — an `AbortSignal` rejects with its reason. An `AbortError` unless you aborted with your own `Error`. An already-aborted signal rejects without waiting. - `timeout` — rejects with `code === 'ERR_POOL_DRAIN_TIMEOUT'` after `timeout` ms. Without it, a drain against a wedged worker waits forever, which is indistinguishable from a hang. - `maxDrainWaiters` (constructor option) — rejects with `code === 'ERR_POOL_DRAIN_TOO_MANY_WAITERS'` once that many are already waiting. Each waiting drain holds an `idle` listener, so a caller that drains in a loop would otherwise accumulate one per call. Whatever ends the wait, the `idle` listener is detached and the waiter slot released — there is no path that leaves either behind. ### Await-response and targeted worker semantics When `options.awaitResponse` is requested, the pool tracks the outgoing request with a generated or provided `correlationId`. Generated ids look like `k3f9qz-1a2b` — a process-unique base-36 tag, a dash, and a process-monotonic base-36 sequence. They are unique across every pool in the process, so a shared log or a shared worker cannot confuse two pools' ids, and they carry no timestamp, so they do not leak anything about when a message was sent. **Do not parse them** — pass `options.correlationId` or `options.correlationIdFactory` if you need a shape you control. The returned Promise resolves only when a worker replies with a matching response payload. If the response never arrives, the Promise rejects when the optional `timeout` expires, and the internal pending entry is removed. When `options.workerId` is supplied, the pool routes the message to that worker only. Targeting a missing or currently saturated worker fails immediately rather than silently queuing the request. For `awaitResponse` callers this means the returned Promise rejects with an immediate failure instead of waiting in the queue. If a worker is terminated while it still has pending `awaitResponse` requests, the pool rejects those Promises and removes the associated pending state. This ensures there are no leaked Promise entries after worker teardown or pool shutdown. ## Bounding the queue `options.queuePolicy` decides what happens when the pool is saturated; `options.maxQueueLength` decides _whether that situation can keep going_. They are different questions, and the defaults answer the second one with "forever". With the default `queuePolicy: 'enqueue'` and no cap, a pool whose workers are slower than its producers accumulates queued tasks without limit. There is no error, no event, and no threshold that stops it — the failure is the process running out of memory, which by then is a long way from the cause. If your producers are not rate-limited against your consumers, set a cap: ```js const pool = new PowerPool(ImageWorker, { minSize: 2, maxSize: 8, taskQueue: true, queuePolicy: 'enqueue', maxQueueLength: 500, // refuse the 501st task rather than growing until OOM }); ``` **Which task gets refused.** With a finite cap, the _incoming_ task is the one that does not fit. A bound the caller asked for is a statement about the newest arrival, and refusing it keeps the work already accepted — with one deliberate exception: - `'drop-oldest'` keeps its documented meaning and evicts the oldest to make room for the newest, so the newest work is the work that runs. Note this policy was already self-bounding before `maxQueueLength` existed: it evicts one and admits one, so the cap never turns it into a refusal. **The steady number it holds is one.** It evicts whenever the queue is non-empty, not only when the queue is at the cap, so `maxQueueLength` does not raise it — if you set `maxQueueLength: 100` with this policy you will still see a queue of length 1. If you want a bounded backlog that _fills_ before it starts dropping, use `'enqueue'` with a finite cap and accept that the newest arrival is what gets refused. - `'drop-newest'` and `'reject'` refuse the incoming task, exactly as they already did. **What refusal looks like.** A refused task returns `false` from `postMessage` / the corresponding slot in a `postMessageBatch` result array. If the caller passed `awaitResponse` or an explicit `correlationId`, the returned Promise rejects with `code === 'ERR_POOL_QUEUE_FULL'` instead, so you can tell "the queue was full" from "the worker failed": ```js try { const result = await pool.postMessage(payload, undefined, { awaitResponse: true }); } catch (err) { if (err.code === 'ERR_POOL_QUEUE_FULL') { // shed load: 429, drop the request, or back off } throw err; } ``` A batch is capped as a group: if 8 of 10 items do not fit, the first 2 are queued and the remaining 8 slots in the result array are `false`. The pool does not queue what fits and then silently drop the rest. **`maxQueueLength: 'auto'` — a starting point, not a default.** **New in 2.1.0.** Pass the string `'auto'` and the cap resolves to `maxSize ** 2`, which is Piscina's affordance for the most common misconfiguration with this option: leaving it unbounded and discovering that as an OOM rather than as a refusal. ```js const pool = new PowerPool(ImageWorker, { minSize: 2, maxSize: 8, maxQueueLength: 'auto', // resolves to 64 }); pool.getStats().maxQueueLength; // 64 — the resolved number, not the string ``` Two things to know: - **`Infinity` is still the default.** A cap that silently appears is a cap nobody chose, and the whole value of a bound is that it is a decision. `'auto'` is something you opt into. - **It is resolved once, at construction.** A later `resize()` does not re-derive it, because a bound that moves under a running pool is a bound nobody can reason about. If you want a different number, pass a number. If you would rather be **told** about pressure without refusing anything, set `options.queueHighThreshold` instead. It emits `pool:queue:high` on the first crossing (and not again until the queue drains back below it) and changes nothing else — see [Events and handlers](#events-and-handlers). The two compose: use the threshold for alerting, the cap for safety. ## Task priority When the pool is saturated and tasks are queued, `PowerPool` dispatches them in priority order rather than pure FIFO. Set a default for all posts with `options.priority` in the constructor, or override per-message with the third argument to `postMessage`: ```js const pool = new PowerPool(WorkerScript, { taskQueue: true, priority: 0, // default for every post }); // high-priority work jumps ahead of the default-priority backlog pool.postMessage(urgentTask, undefined, { priority: 10 }); pool.postMessage(normalTask); // priority 0 ``` **Rules:** - Higher numbers run first. `priority: 10` jumps ahead of `priority: 0`. - Equal-priority tasks keep FIFO order. Two posts with the same priority run in the order they arrived. - The default is `0`. Omitting `priority` is the same as `0`. - Priority is a property of the queued item, not the pool. Every post can choose its own priority regardless of the constructor default. - Priority composes with `queuePolicy`. `drop-oldest` still evicts the oldest entry when over `maxQueueLength`; `drop-newest` and `reject` still refuse the newcomer. Priority only changes the order in which queued items are _dispatched_, not which items are _admitted_ or _dropped_. **How it is dispatched, and why aging is the exception. New in 2.1.0.** With `priorityAgingMs` at its default of `0` the effective priority is static, so the queue is a **binary heap** and each dispatch is O(log n). Before this it was an O(n) scan of every queued item, which made draining a saturated queue O(n²) — under sustained saturation, the difference between a pool that keeps up and one that falls behind progressively. It matters more here than in a Node-only pool: a long pause on a browser main thread is a dropped frame. `priorityAgingMs` above `0` makes the effective priority `priority + (now - enqueuedAt) / priorityAgingMs`, which **drifts with the clock**. A heap keys on the priority at push time, so it would order by a key that is already stale; aging therefore stays on the O(n) scan. The two paths dispatch in the same order for the same input, and both keep FIFO among equal priorities — that equivalence is pinned by a test, so swapping the structure is not something a future change can quietly get wrong. **Batch posts.** `postMessageBatch` accepts `options.priority` and applies it to every item in the batch. Individual items cannot carry different priorities in the same batch — if you need mixed priorities, post them in separate batches or individual posts. ## Aborting a single task **New in 2.1.0.** `postMessage`'s third argument accepts a `signal` — a standard `AbortSignal`, which is why this works the same way in every realm the pool runs in. ```js const ac = new AbortController(); const pending = pool.postMessage(task, undefined, { awaitResponse: true, signal: ac.signal }); ac.abort(); // the caller stops waiting, immediately ``` Three cases, and the third is the one worth understanding: - **Already aborted at post time** — refused before any queue work. A caller holding an aborted signal has already changed their mind, so queuing the task would be work done on their behalf after they cancelled it. Fire-and-forget returns `false` rather than throwing, matching every other refusal on that path. - **Aborted while queued** — the pending response is rejected _at abort time_, not at dispatch time. The item is deliberately **not** pulled out of the queue: neither queue implementation has an efficient remove-by-value, and the dispatch loop already walks past items it is going to discard — the same trade `deadlineAt` makes. The cost is one held slot until dispatch; the alternative is an O(n) scan on every abort. - **Aborted while in flight** — **the work runs to completion and the waiter is released now.** **Why not terminate the worker, the way Piscina does.** Under `maxTasksPerWorker > 1` a worker carries several in-flight tasks, so terminating it to abandon one kills every other task on that worker — a blast radius far larger than the task being aborted, and one the callers of those other tasks did nothing to earn. It also discards a warm worker and its negotiated protocol state. So the _work_ finishes and the _waiter_ is released immediately, with the response discarded when it arrives. That is the honest reading of what an abort usually means — "stop waiting for this". A caller who genuinely needs the work undone has to arrange that in the worker, because no pool can undo a side effect. **`skipQueue` is deliberately not an option.** The pool already dispatches immediately when a worker is free; the queue is only used under saturation; and a `skipQueue` that jumped ahead of queued work would be a priority mechanism by another name when `priority` already exists. ## Percentile telemetry **New in 2.1.0.** `options.recordTiming` opts in to tail-shape statistics that the existing O(1) accumulators cannot express. ```js const pool = new PowerPool(WorkerScript, { recordTiming: true }); // … after some work … const perf = pool.getStats().performance; perf.percentiles; // { p50, p90, p99, p999 } — task duration, ms perf.queueWaitPercentiles; // { p50, p90, p99, p999 } — time spent queued, ms perf.utilization; // 0..1 — fraction of pool lifetime with work in flight ``` **Why a sketch and not more counters.** `timePerTask` and `queueWait` already report exact mean, stddev, min and max in O(1) memory, and `percentSlowTasks` counts exactly against a threshold. None of those can answer "what does the tail look like" — a mean of 12 ms is a different pool at p99 = 13 ms than at p99 = 400 ms, and only the second one has a problem. `PowerHistogram` is a DDSketch, which bounds _relative_ error across decades of magnitude, so a 3 ms task and a 3 s task land in the same sketch without one swamping the other. **It is `PowerHistogram`, not Piscina's histogram.** Piscina's is `node:perf_hooks`, which does not exist in a browser; this pool runs in six realms. The class already exists here and `bench/claims.js latency` already validates its quantile accuracy. Three things to know: - **Off by default, and the default path pays nothing.** With `recordTiming` off the two histograms are never allocated, and each recording site is a single boolean test. A pool that never asks for percentiles should not carry a sketch it never fills. - **`percentiles` and `queueWaitPercentiles` are `null` when off, not zeros.** A pool that never opted in has no sketch, and `p50: 0` would read as "the median task took no time" rather than "nobody is measuring". `null` is the honest answer to "is this on", which is the first thing to check. - **`utilization` is not gated.** It is two comparisons on transitions that already happen on every dispatch and every completion, so it costs nothing and is always reported. It is interval-based rather than sampled, because a sampled utilization is a function of the sampling rate: a pool busy for 1 ms in every 10 reads as idle at a 100 ms sample and saturated at a 1 ms one. **The Welford statistics are additive, not replaced.** `timePerTask`, `queueWait` and `percentSlowTasks` keep their exact closed forms; the sketches add the tail shape beside them. Cross-reference ADR 0014. ## Choosing the worker **New in 2.1.0.** `options.loadBalancer` replaces the _choice_ of worker, not the policy. Omit it and the built-in least-loaded-plus-EWMA rule runs, exactly as before — that rule beats a plain least-loaded count for heterogeneous tasks, because a worker that has been answering fast is a better bet than one that is merely idle. ```js const pool = new PowerPool(WorkerScript, { loadBalancer: (workers) => { // `workers` is a frozen array of { id, tasks, latencyEwma, lastActive } const quietest = workers.reduce((a, b) => (b.tasks < a.tasks ? b : a)); return quietest.id; // or the object itself }, }); ``` **The view is a plain frozen object, not the live worker.** This is the part worth understanding before you write a balancer. A balancer is _your_ code running inside the pool, invoked from the pool's own send path — so it is handed a snapshot of four fields rather than the worker object, which would otherwise expose `terminate()`, the event wrapper, and everything reachable from them. `Object.freeze` makes an attempt to write to the view loud rather than silent. `latencyEwma` is in the view explicitly, because the whole point of the view is that reaching past it is not possible. Three things to know: - **Return the id or the object, either works.** A balancer that just read the view has the id in hand; one that kept a reference has the object. - **A bad return falls back, it does not throw.** `null`, `undefined`, or an id that is not in the pool selects the default rule. The alternative turns your bug into every other caller's outage, because the throw would come from inside the pool's send path. A balancer that _throws_ is logged and falls back the same way. - **A non-function is rejected at construction**, once, rather than silently ignored on every dispatch. The same frozen view is reported per worker by `getStats().workers`, so debugging a custom policy reads exactly what the policy read. ## Worker readiness **New in 2.1.0.** `options.ready` holds a worker out of dispatch until it signals it has finished async init. ```js const pool = new PowerPool(WorkerScript, { ready: true }); // in the worker: postMessage({ ready: true }); // once async setup is done ``` Without this, a task can be dispatched to a worker still running its constructor's async work. The worker receives the message, cannot answer it, and the caller waits for a response that will never come — which surfaces as a timeout rather than as a failure to start. Three things to know: - **The field name is yours.** Pass a string to name it (`ready: 'initialised'`), or `true` for the default `'ready'`. A worker posting `{ ready: true, version: 3 }` has signalled; the extra fields are ignored. A bare `true` works too. - **The signal is pool-internal traffic.** It is consumed by the pool and deliberately _not_ forwarded to `message` listeners, which never sent it and would have to recognise it to be correct. It is also not a task, so it does not touch the task accounting. - **`readyTimeout` fails open.** A worker that never signals is dispatched anyway after that many ms, so a misbehaving worker degrades to the old behaviour rather than wedging the pool forever. `0` (the default) waits forever. **What happens to a post that arrives while nothing is ready.** With the task queue on — the default — the work is _held_, not refused: the caller is not told "no" for a condition that resolves in milliseconds. With `taskQueue: false` and no ready worker, the post is refused instead, because the round-robin fallback would otherwise dispatch straight past the gate. ## Autoscaling `PowerPool` supports an optional autoscaling mode that grows or shrinks the worker pool based on recent observed task latency (EWMA) and queue pressure. Enable it by passing `options.autoScale` to the constructor. When `autoScale` is a boolean `true` the pool uses sensible defaults. For production workloads pass an object to tune behavior: ```js const pool = new PowerPool(WorkerScript, { minSize: 1, maxSize: 16, autoScale: { intervalMs: 1000, // evaluation interval (ms) targetMs: 50, // target per-task latency (ms) alpha: 0.2, // EWMA smoothing factor (0..1) cooldownMs: 5000, // minimum time between scale actions (ms) hysteresis: 0.2, // fractional hysteresis (0..1) to avoid flapping }, }); ``` Behavior summary: - The pool maintains a pool-level EWMA of recent task durations. - Every `intervalMs` the pool evaluates scaling decisions: - Scale up when EWMA exceeds `targetMs * (1 + hysteresis)` or when queue pressure is high. - Scale down when EWMA falls below `targetMs * (1 - hysteresis)` and the queue is empty. - `cooldownMs` prevents rapid oscillation by requiring a minimum delay between scale actions. - The **direction** of a scale action is that latency EWMA against `targetMs`; its **size** is closed-loop. `stepUp` / `stepDown` (default `1`) are _ceilings_ on workers changed per tick, and a [`PowerServo`](powerServo.md) PI controller on the relative error picks how much of that budget to use. At the default ceiling of 1 nothing changes. See [How big a step](autoscale.md#how-big-a-step). Tuning tips: - Increase `targetMs` for longer-running tasks. - Lower `alpha` to smooth noisy workloads; increase it to react faster. - Use `cooldownMs` (e.g., 3–10s) to avoid repeated add/remove cycles. - `hysteresis` values of 0.1–0.3 are typically effective at preventing flapping. See [autoscale guide](autoscale.md) for more details and examples. ### Adaptive concurrency policies The default controller above is a **latency-threshold heuristic**: it compares one EWMA against a fixed `targetMs`. That is a reasonable first cut, but it is a guess about the right fleet size rather than a measurement of it. `autoScale.policy` swaps in a real feedback loop — a concurrency controller that treats the worker count as a congestion window. The pool already tracks every signal these need, so nothing new has to be instrumented: | policy | behaviour | | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `'ewma'` | _(default, unchanged)_ the `targetMs` heuristic above. Reports `concurrencyLimit: null`. | | `'aimd'` | Additive increase while healthy, multiplicative decrease (`aimdBeta`, default `0.7`) on a congestion signal. Simplest and most robust. | | `'vegas'` | Estimates the bottleneck queue as `limit * (1 - minRtt / currentRtt)` and moves by `alpha`/`beta` (`3*log10(limit)` / `6*log10(limit)`), as in the reference implementation. | | `'gradient2'` | `gradient = clamp(longRtt / currentRtt, 0.5, 1)`, then `limit = gradient * limit + queueSize`. Unlike Vegas it does not use the window _minimum_ latency, which biases the estimate. | ```js const pool = new PowerPool(WorkerScript, { minSize: 1, maxSize: 16, autoScale: { policy: 'gradient2', limitMin: 1, limitMax: 16, longWindowAlpha: 0.05, }, }); ``` Extra options: `policy` (`'ewma'|'aimd'|'vegas'|'gradient2'`), `limitMin`, `limitMax`, `longWindowAlpha` (smoothing for the long-window RTT EWMA, default `0.05`), `aimdBeta`. Observability — `getStats().performance` gains: - `concurrencyLimit` — the controller's current limit, or `null` for the `ewma` policy. - `autoScalePolicy` — the policy name **while its interval is ticking**, or `null` when no controller is running. - `congestion` — whether the controller currently believes it is over-provisioned. - `idleReapingActive` — whether the idle-worker reaper's interval is running. - `queueWait` — count and Welford statistics for time spent queued before dispatch; `timePerTask` remains worker service time. - **`concurrencyLimit` is reported, not enforced.** The adaptive controller's limit is exposed via `getStats().performance.concurrencyLimit` for observability, but it is **not** used to gate admission: `bench/claims.js concurrency` measured that enforcing it does not beat a well-chosen constant cap (within the noise floor, -1.6 %). Wiring it into the dispatch path would change how work is admitted for a controller that cannot earn the getter, so the limit stays informational. Treat it as a diagnostic, not a throttle. field reported the _stored configuration_, and `stopThePress(..., { recreateWorkers: false })` clears the interval without clearing the policy — correctly, since a pool stopped and recreated gets a working controller again. So a pool could report `autoScalePolicy: 'aimd'` with an interval that would never fire again, and an operator reading it would conclude adaptation was running. It now answers the only question it can answer honestly. - **`concurrencyLimit` is deliberately retained when the controller stops.** `_adaptiveLimit` keeps its value, so the field remains observable even though `autoScalePolicy` becomes `null` and no further controller ticks run. - **`idleReapingActive` exists because nothing reported it at all.** The reaper's interval is cleared by the same two calls, and neither is restorable, so a pool can accumulate idle workers past `idleTimeout` for the rest of its life with no signal in `getStats()`. It is independent of `autoScale` — with autoscaling off there is no policy field to read, leaving nothing to notice a stopped reaper by. - `'aimd'` and `'vegas'` only move the limit once there is at least one latency sample. A pool that has never completed a task holds its seed value rather than guessing. - Vegas's `alpha`/`beta` scale with `log10(limit)`, so at small limits the queue estimate lands in a neutral band and the limit holds steady. It needs a limit above roughly 3 before it will step down. That is the algorithm's behaviour, not a stall. - `'gradient2'` has no queue-pressure term to grow from, so an idle pool with an empty queue correctly holds its limit steady. Depth is what drives it up. - The limit is a **float**, smoothed by 0.2 each tick so a single noisy sample cannot swing the fleet. `getStats()` rounds it to two decimals. - This is a separate signal from the worker add/remove step, which still runs the existing `cooldown`/`backoff` logic. The worker add/remove step acts on **EWMA latency**; the adaptive policies separately gate new admissions. - **The signal is end-to-end task latency; Netflix's controllers track queueing delay.** The two are the same quantity only for a uniform workload. On a pool whose tasks vary in cost, `vegas`' `minRtt / currentRtt` and `aimd`'s short/long RTT test cannot tell "queueing appeared" from "a heavier task ran", so the controller cuts concurrency for work that was merely expensive. Uniform-cost workloads are the case these port well to. - **`'aimd'` here is delay-shaped despite the name.** Netflix's `AIMDLimit` is loss-based; this branches on RTT divergence, the same shape as `'gradient2'`. The loss-based AIMD in this library is `PowerBackpressure`'s adaptive refill, and it is not interchangeable with this one — a permit gate has no round trip to measure. - **The limit is enforced for `'aimd'`, `'vegas'`, and `'gradient2'`.** `concurrencyLimit` is written by the controller and read by both `getStats()` and the dispatch paths. When the limit is reached, new work is queued if `taskQueue` is enabled; otherwise it is refused. In-flight work is allowed to complete when the controller lowers the limit. - Two consequences are worth stating plainly. Autoscaling this pool still controls **worker count** independently; the adaptive policies control **in-flight admissions**. `maxQueueLength` and `queuePolicy` remain the controls for bounding or refusing queued overflow. ## Events and handlers `onmessage`, `onerror`, `onidle` — setter/getter properties for convenient handlers. ### What a `message` event is `onmessage` and `'message'` listeners receive an event object with `data` set to the worker's reply. That event is the pool's own, **not** the platform's `MessageEvent`: - `data` — the decoded reply. - `correlationId`, `duration` — present when the reply arrived on the native envelope carrier. - `originalEvent` — the platform event the reply arrived on, when there was one. In a browser that is a real `MessageEvent`, so `event.originalEvent.origin` still works. In Node there is no wrapper and this is the posted value. This changed in 2.0. The pool used to forward the browser's `MessageEvent` itself and write `correlationId` onto it, which meant a second listener on the same event saw fields the pool had invented. Nothing is lost except that one hop: read `originalEvent` where you used to read the event's own `origin`, `target` or `lastEventId`. `error` and `messageerror` are forwarded **as the platform delivered them** — only `message` is normalised, because only `message` carries a payload to normalise. `onidle` and `'idle'` listeners receive an event with `data.type === 'pool:idle'` and two separate payloads: - `data.workers` — the per-worker snapshot: an array of `{ id, tasks, lastActive }`. This is the pool's account of _which_ workers it believes are idle. A worker with `tasks !== 0` here means the pool's active-task accounting has drifted from the per-worker counts, so this is the field to log when investigating a drain that never settles. - `data.stats` — the aggregate `getStats()` summary: `{ status, performance, queueLength, activeTasks, workerCount, ... }`. Both are computed lazily, so an idle transition costs nothing when no listener reads them. They used to be one field called `stats` that the documentation described as an array while the code produced a summary — the split names each for what it actually is, and `stats` keeps its key so existing listeners are unaffected. `pool:queue:high` — emitted on the internal event bus when the internal task queue length crosses the configured `options.queueHighThreshold`. Payload: `{ length, threshold }`. Configure `queueHighThreshold` in constructor options to enable this event. This is a **notification, not a limit** — nothing is refused when the threshold is crossed. Use `options.maxQueueLength` to actually bound the queue. `pool:scale` — emitted when workers are added or removed. Payloads vary by origin: when workers are created the payload is `{ action: 'add', id, minSize, maxSize }`; when workers are terminated the payload contains `{ action: 'remove', terminated: [ids], count }`. The existing `resize` event is still emitted for API compatibility. `pool:worker:exit` — **New in 2.1.0.** emitted on the internal event bus when a worker dies. Payload: `{ id, code, completedTask, remaining }`, where `code` is the exit code the runtime reported (or `null` where it reports none) and `completedTask` says whether that worker had ever finished a task. See [Worker death](#worker-death). ### Worker death **New in 2.1.0.** A worker that dies is now detected, settled and replaced. Before this, a dead worker was simply absent, and three things followed — all observable from outside as "the pool stopped working": - `drain()` hung, waiting for a response that would never arrive; - every `awaitResponse` promise timed out instead of rejecting, so a caller waited the full timeout to learn what had already happened; - the pool sat below `minSize` permanently, because nothing noticed the shortfall. The pool now settles the dead worker's in-flight tasks and **rejects its pending responses at the moment of death** rather than at a timeout, removes it, emits `pool:worker:exit`, and refills toward `minSize`. **The bootstrap guard is the part worth understanding.** A worker that dies having never completed a task is a worker that failed to _load_ — a syntax error, a missing module, bad `workerOptions`. Refilling on every such death is a respawn storm: each replacement loads, throws, dies, and is replaced, at full CPU, forever. So **three consecutive load-failures stop the refill**, and the reason is logged rather than retried. A worker that had been working and then died is a _crash_, not a load failure, and resets that bound — otherwise one bad deploy would permanently shrink a healthy pool. **Known limitation, recorded rather than papered over: this is Node-only.** A browser `Worker` has no `'exit'` event, so the handler never fires there and the `drain()`-hangs symptom persists in that runtime. There is nothing to listen for; inventing a heartbeat to detect death would be a different feature with its own false-positive rate. ## Example ## Realistic Example — image thumbnail worker This example shows a common pattern: a pool of workers that produce thumbnail images from large binary blobs. The pool dispatches work, awaits per-task responses, and drains before graceful shutdown. ```javascript import ImageWorker from './image-worker.js?worker'; import { PowerPool } from 'performance-helpers/powerPool'; // Create a small pool tuned for CPU-bound thumbnailing const pool = new PowerPool(ImageWorker, { size: 2, maxSize: 4, idleTimeout: 30_000 }); // For low-load deployments avoid eager worker startup: // (example alternative: lazy startup omitted for brevity) // Helper to post a job and await the worker's response async function makeThumbnail(imageBuffer) { // workers are expected to echo back { correlationId, response } const req = { op: 'thumbnail', payload: imageBuffer }; return pool.postMessage(req, undefined, { awaitResponse: true, timeout: 10_000 }); } // Process a batch of images concurrently but with backpressure from the pool async function processImages(images) { const tasks = images.map((img) => makeThumbnail(img)); // await all thumbnails (each item may be a Promise) const thumbs = await Promise.all(tasks); console.log('generated', thumbs.length, 'thumbnails'); } // On shutdown ensure all inflight work completes async function shutdown() { // Option A: graceful shutdown — wait for in-flight work to complete then stop the pool await pool.drain(); // wait until queue empty and workers idle // `terminate()` now delegates to `shutdown()` internally, but you can call either. pool.terminate(); // Option B (new in 2.1.0): the same thing in one call. // await pool.close(); } // Example usage (async () => { const images = await loadManyImages(); // user-defined helper await processImages(images); await shutdown(); })(); ``` ## Explicit shutdown vs terminate `PowerPool` exposes two related lifecycle APIs: - `shutdown()` — performs a full stop: clears the internal reaper interval, terminates workers, clears internal queues, and rejects any pending `awaitResponse` Promises with a `PowerPoolShutdownError`. Use this when you need to ensure no background timers remain and that any callers awaiting responses are notified. - If a worker is terminated while it still has pending `awaitResponse` requests, the pool rejects those Promises immediately — with `code === 'ERR_POOL_WORKER_TERMINATED'` — and removes the associated pending state. This avoids leaked promise bookkeeping during worker teardown or replacement, and it is the only way you learn that those tasks were lost. **This includes retirements you did not ask for**: `resize()` down, idle reaping (`idleTimeout`), autoscale shrinking the fleet, and `stopThePress()` all terminate workers, and any in-flight `awaitResponse` request on a retired worker is dropped with it. If you need those responses to survive, await them before you resize or drain. - `terminate()` — delegates to `shutdown()` for consistent behavior. It is safe to call synchronously when tearing down resources; it will also reject pending Promises and clear timers. Examples: Graceful drain then explicit shutdown (preferred when you want in-flight work to finish): ```javascript await pool.drain(); await pool.shutdown(); // rejects any stray pending promises and clears timers ``` Immediate stop (rejects pending awaits): ```javascript pool.terminate(); // synchronous; delegates to shutdown internally ``` Handling shutdown rejections (when callers previously awaited a response): ```javascript try { const p = pool.postMessage({ op: 'work' }, undefined, { awaitResponse: true }); // somewhere else: pool.shutdown() or pool.terminate() may be called const resp = await p; } catch (err) { if (err?.code === 'ERR_POOL_TERMINATED' || err?.name === 'PowerPoolShutdownError') { // pool was shut down while awaiting response } else if (err?.code === 'ERR_POOL_WORKER_TERMINATED') { // the worker holding this task was retired — resize, idle reap, autoscale. // The work is lost, not delayed: retry elsewhere or surface the failure. } else { // other error } } ``` ## Worker health check `pool.healthCheck()` (alias `pool.ping()`) returns a read-only snapshot array of `{ index, pending, alive }` describing each worker. `alive` is `true` when the underlying worker object is present and not terminated. The list is empty after `terminate()`/`shutdown()`. ```js const health = pool.healthCheck(); // [{ index: 0, pending: 3, alive: true }, ...] ``` The `healthCheckIntervalMs` constructor option (default `0`, disabled) is reserved for future periodic pinging. ## Disposal `PowerPool` implements `dispose()` and `[Symbol.dispose]`, so it works with `using` / `await using` and with a DI container's teardown, like every other long-lived helper here. ```javascript { using pool = new PowerPool(WorkerScript, { size: 2 }); // ... } // dispose() runs here, calling terminate() synchronously ``` **There are timers and workers to cancel.** The pool owns a reaper interval, worker threads, and pending `awaitResponse` bookkeeping. `dispose()` calls `terminate()` synchronously, which clears the reaper, terminates every worker, and rejects any pending response Promises with `PowerPoolShutdownError`. Use `dispose()` when you are finished with the pool. `[Symbol.asyncDispose]()` is the graceful cousin: it awaits `drain()` first, letting in-flight work finish, and then calls `terminate()`. Use it when you want the pool to complete its current work before shutting down. `shutdown()` and `terminate()` are the underlying primitives. `shutdown()` performs the same hard stop as `dispose()` but returns a `Promise` that rejects with `PowerPoolShutdownError`; `terminate()` is the synchronous version. Both are safe to call multiple times. ## Batch examples ```javascript // Fire-and-forget batch (optimized path) const batch = [{ message: { task: 1 } }, { message: { task: 2 } }]; const results = pool.postMessageBatch(batch); const responseBatch = pool.postMessageBatch(batch, { awaitResponse: true, correlationIdFactory: (index, item) => `job-${item.message.a}-${index}`, }); const responses = await Promise.all(responseBatch); responses.forEach((resp, index) => { console.log('job', index, 'correlationId', resp.correlationId); }); // results: [ true, true ] — dispatched or queued // Await per-item responses (each entry returns a Promise) const r = pool.postMessageBatch([{ message: { req: 'a' } }, { message: { req: 'b' } }], { awaitResponse: true, timeout: 5000, }); // r is an array like [ Promise, Promise ] — await as needed const responses = await Promise.all(r.map((p) => (p instanceof Promise ? p : Promise.resolve(p)))); console.log('batch responses', responses); ``` ### Preparing buffers for hotspot workloads `prepareBuffers(items, { clone = false })` lets you pre-encode a batch of messages so you can avoid repeated encoding during `postMessageBatch` or `broadcast`. It returns one `{ message, transfer }` entry per item — **not** a bare `Uint8Array` — where `message` is a `Uint8Array` for a plain object and `transfer` is `[buffer]` under `clone: true`, `undefined` under `clone: false` because the cached buffer must not be detached. Each `items` entry may be a plain object, a `Uint8Array`/TypedArray, or `{ message, transfer? }`. Example — pre-encode a large shared payload and send cloned transferable buffers per worker: ```javascript // Pre-encode 100 items (clone=true makes each returned buffer safe to transfer) const prepared = pool.prepareBuffers( Array.from({ length: 100 }, () => ({ message: { big: 'payload', repeated: true } })), { clone: true } ); // prepared is an array of { message: Uint8Array, transfer: [ArrayBuffer] } const res = pool.postMessageBatch(prepared); ``` Example — prepare once and reuse cached buffer references (clone=false). WARNING: do not transfer the returned buffers when `clone:false` — they are shared cached objects. ```javascript // One payload, prepared: the array form unwrapped, because there is no singular // method. `test/docsCodeAgreement.test.js` is what stops that sentence from // drifting back — this guide previously documented a `prepareBuffer` that never // existed, and called it in a runnable example. const [cached] = pool.prepareBuffers([{ heavy: 'payload' }], { clone: false }); // `cached.message` is the shared cached buffer and `cached.transfer` is // undefined. Clone before sending, so the cache entry is not detached: pool.postMessage(cached.message.slice(), [cached.message.slice().buffer]); ``` #### What `prepareBuffers` returns, and why it is not a `Uint8Array` Under a framing codec, the default (`clone: false`) returns the item **unencoded and marked `deferred`**, and framing happens at dispatch. That is not an implementation detail you can ignore: a pre-encoded body is a bare JSON body, and posting it verbatim makes the worker read `{` — byte 123 — as a protocol version, and every batched request fails with `unsupported protocol version 123 (expected 1)`. Framing wraps the body in a fresh header per call, so a _cached_ body can never be both shared and correct on the wire. The two modes split cleanly: | Mode | Returns | When | | ------------------------ | ----------------------------------------------------------- | --------------------------------------------------------------------------------------------- | | `clone: false` (default) | `{ message, transfer: undefined, deferred: true }` | The encode is deferred to dispatch, so nothing shared is left unframed. | | `clone: true` | `{ message: , transfer: [ArrayBuffer] }` | The frame is private already, so the work really is done up front. | | `messageCodec: 'legacy'` | `{ message: , transfer: undefined }` | There is no frame, so the cached body _is_ the wire format and pre-encoding is a real saving. | `transfer` is `undefined` wherever the buffer is shared, and that is the property worth keeping: a transfer list on a shared buffer would detach the cached encode entry. A deferred item carries no buffer at all, so there is nothing that _can_ be detached. #### A shared encode buffer is now marked, so the platform refuses to transfer it **Changed in 2.0, and this is a behaviour change you may notice.** The encode cache hands the **same** `Uint8Array` to every caller encoding the same payload. Transferring it detaches the shared buffer, and the next cache hit returns a zero-length husk — the payload arrives at the worker **empty, with nothing thrown anywhere**. That was previously guarded by a doc note (the sentence above) plus a detached-buffer pre-check, and the pre-check has a gap: it catches the **second** transfer, not the first, and the first is the one that does the damage. The buffers are now marked with `markAsUntransferable()`, and **the platform enforces it**. Where Node has it (this library's floor is `>=22.12.0`; browsers do not, and there the pre-check still stands): ```js const frame = pool._encodeForTransfer({ hello: 'world' }); // the shared cached buffer structuredClone({ frame }, { transfer: [frame.buffer] }); // → DOMException: Cannot transfer object of unsupported type ``` So the failure moves **to the call site that made the mistake**, and the bytes stay intact — which is the difference between a _refused_ transfer and a _performed_ one. Concretely: - **Before:** transfer it, get no error, and every later hit of that key is empty. - **After:** transfer it, get a `DOMException` at the `postMessage`, and the cache entry is untouched. **Do not catch and ignore that error**, or you are back to the silent case with more steps. The fix is to send a copy — `.slice()` — which is what `prepareBuffers`' `clone: true` mode exists for. The error _type_ is a platform implementation detail (`DOMException` on Node 24.18, not the `DataCloneError` a first draft of this note claimed), so match on the fact that it threw, not on its class. Buffers on the **cache-bypass paths** — a payload too large to cache, for instance — are _not_ marked, because there is nothing shared to protect and refusing a transfer the caller owns would be a regression in the other direction. ### Zero-copy: forwarding raw ArrayBuffers / TypedArrays When your producer already has an `ArrayBuffer` or a `TypedArray` (for example a decoded image or a pre-serialized payload) you can avoid re-encoding and enable zero-copy transfers by passing the raw buffer directly. The pool will auto-add the underlying `ArrayBuffer` to the transfer list when no `transfer` is provided. If you want to explicitly request zero-copy semantics (forward the exact buffer without cloning), pass the `zeroCopy: true` option to `postMessage`, `postMessageBatch`, or `broadcast`. ```javascript // Send a pre-serialized Uint8Array directly (auto-transfer when transfer omitted) const buf = new Uint8Array(largePayload); pool.postMessage(buf); // pool will auto-add buf.buffer to transfer list // Explicit zero-copy (caller accepts that buffer may be neutered/transferred): pool.postMessage(buf, undefined, { zeroCopy: true }); // For batches, pass options.zeroCopy to postMessageBatch so prepared items are forwarded as-is const batch = Array.from({ length: 10 }, () => ({ message: new Uint8Array(1024) })); pool.postMessageBatch(batch, { zeroCopy: true }); ``` Notes: - `zeroCopy: true` only affects `ArrayBuffer`/TypedArray messages — plain objects cannot be forwarded zero-copy and will be encoded as before. - When using cached buffers via `prepareBuffers([...], { clone: false })`, do NOT transfer the cached buffer itself; clone it first via `slice()` if you need a transferable copy. ## Migrating to the framed protocol (breaking change in 2.0) ### What changed In 1.x, `PowerPool` encoded a plain object to a **bare `Uint8Array` of JSON** with no header, and decided what it received on the way back by _sniffing_ — "if this looks like an `ArrayBuffer`, `JSON.parse` it". That had three problems: the wire format had no version, so it could never evolve; a genuinely binary worker message was silently corrupted by `JSON.parse`; and every worker had to hand-decode the bytes. From 2.0 the pool sends a [`PowerMessageCodec`](powerMessageCodec.md) frame instead: ``` byte 0 protocol version byte 1 codec id bytes 2..5 payload length (uint32 little-endian) bytes 6.. payload ``` ### How to migrate a worker One line changes, in the worker's message handler. ```js // 1.x — bare JSON bytes import { u82o } from 'performance-helpers'; self.onmessage = (e) => handle(u82o(e.data)); // 2.0 — framed import { decodeMessage } from 'performance-helpers'; self.onmessage = (e) => handle(decodeMessage(e.data).value); ``` `decodeMessage` also accepts a raw `ArrayBuffer` or `DataView`, so a worker that receives a plain binary `ArrayBuffer` you posted yourself (not a frame) should branch: ```js const isFrame = e.data instanceof Uint8Array && e.data.length >= 6 && e.data[0] === 1; const data = isFrame ? decodeMessage(e.data).value : e.data; ``` A robust worker that must interoperate with both can simply try the frame and fall back, which is what `PowerChunker`'s inline worker does internally: ```js let data; try { data = decodeMessage(e.data).value; } catch { data = u82o(e.data); // 1.x peer } ``` > **Your worker must also _reply_ in the shape it received.** This trips people up. A > `messageCodec: 'legacy'` pool sniffs its _replies_ with `u82o`, so a framed reply is > unreadable to it and a pending `awaitResponse` promise will simply never settle. Track > which shape arrived and mirror it: > > ```js > let framed = true; > let data; > try { > data = decodeMessage(e.data).value; > } catch { > framed = false; > data = u82o(e.data); > } > > const body = { correlationId: data.correlationId, result: run(data) }; > self.postMessage(framed ? encodeMessage(body) : o2u8(body)); > ``` ### What you get in exchange - **Binary survives.** A `Uint8Array`/`ArrayBuffer` task is framed under the `raw` codec and arrives intact instead of being `JSON.parse`d into nonsense. A binary message passed **without** a `transfer` list is framed like anything else, so a worker always sees one protocol regardless of what it was handed. If you supply your own `transfer` list the pool leaves your buffer completely alone — that is the documented "I am sending this exact buffer, already in the form I want" case, and it is how you post a pre-framed shared buffer. - **A version byte.** The protocol can change again without another flag day — `decodeMessage` throws on an unknown version rather than mis-parsing. - **Self-delimiting frames.** `decodeMessage` reports `byteLength`, so one receive can carry several messages. ### Escape hatch Pass `messageCodec: 'legacy'` to restore the 1.x wire format, on a per-pool basis. This is intended to let a worker be migrated on its own schedule, not as a permanent setting. ```js const pool = new PowerPool(WorkerScript, { messageCodec: 'legacy' }); ``` | `messageCodec` | Outbound | Inbound | | ---------------------- | ----------------------------------- | ----------------------------- | | `'framed'` _(default)_ | `PowerMessageCodec` envelope | `decodeMessage` — no sniffing | | `'legacy'` | bare `Uint8Array` of JSON | `u82o` — sniffed | | `'negotiated'` | frame, or native carrier per worker | `decodeInbound` | ## Protocol negotiation (`'negotiated'`) `messageCodec: 'negotiated'` is `'framed'` plus one thing: a worker that advertises the native structured-clone carrier gets that carrier, and every other worker keeps getting the frame. It exists because the frame is lossy for a class of values. Through the shipped path, a worker using `decodeMessage` receives: | you post | the worker receives | | ------------------------- | ------------------------------------------------------- | | `new Map([['a', 1]])` | `{}` | | `new Set([1, 2])` | `{}` | | `new Date(1234567890123)` | an ISO **string** | | `10n` | the message posts unframed, then `decodeMessage` throws | | `Infinity`, `NaN` | `null` | `Date` is the sharpest: nothing fails at the boundary, and the first `.getTime()` in the worker throws somewhere unrelated, long after the `postMessage`. ### Enabling it Two steps, and they are independent on purpose — the pool can be switched on before any worker is ready, which is the whole point of a mixed fleet being a normal state. ```js // 1. the pool (any time) const pool = new PowerPool(WorkerScript, { messageCodec: 'negotiated' }); ``` ```js // 2. each worker, one line at start-up import { decodeInbound, announceCapabilities } from 'performance-helpers'; parentPort.postMessage(announceCapabilities()); ``` **The worker advertises; the pool only listens.** The pool never puts a control message on a worker's port, because a worker that has not implemented the handshake would run it as a task. The only thing negotiation asks of an un-migrated peer is silence. ### What it is not It is not a speedup. Structured clone is a tie for small objects, slower for deeply nested structure, and faster only for string-heavy payloads. A pool that posted envelopes for the speed would have been slower for the payloads a worker actually receives. The fidelity table above is the reason to adopt it, and the per-worker decision is what stops a pool adopting it where it does not pay. ### Observing it ```js pool.getStats().protocol; // { mode: 'negotiated', nativeAvailable: true, nativeWorkers: 2, // workers: [{ id: 0, codecs: ['json', 'native'] }, { id: 1, codecs: ['json'] }] } ``` A `pool:protocol` event fires when a worker's advertised capabilities change. Capability announcements are pool-internal protocol traffic: they are consumed rather than forwarded to `message` listeners, and they do not touch task accounting. ### Notes - A worker that advertises still receives **one framed message first** — the one that was in flight before its announcement landed. The ratchet is per message, not per worker lifetime. - A message containing an `ArrayBuffer` is copied before it is transferred on the native carrier, so a caller's buffer is never detached by a `postMessage`. This costs a copy; it is the price of not destroying the caller's data. - `'negotiated'` needs `structuredClone` at runtime. Without it the mode degrades to the frame rather than throwing, and `protocol.nativeAvailable` reports `false`. ### Sending a pre-encoded buffer If you pre-encode to share one buffer across many workers, frame it once and reuse the frame. Do **not** use `o2u8` on its own any more: an unframed body will fail to decode under the default protocol. ```js import { encodeMessage } from 'performance-helpers'; const shared = encodeMessage({ big: 'payload', repeated: true }); const batch = Array.from({ length: 100 }, () => ({ message: shared, transfer: [shared.buffer] })); pool.postMessageBatch(batch); ``` ## Closing the pool **New in 2.1.0.** `close()` is graceful shutdown as a single entry point. The `shutdown()` / `drain()` / `[Symbol.asyncDispose]` model is defensible and stays exactly as it is — `close()` composes it rather than replacing it. What it adds is the one call a caller reaching for "shut this down properly" actually wants, which is _both_ halves in the right order. ```js await pool.close(); // wait for in-flight work, then terminate await pool.close({ timeout: 5000 }); // wait at most 5 s, then terminate anyway await pool.close({ force: true }); // terminate now, abandoning the work ``` **Why the order matters.** `shutdown()` alone abandons in-flight work, and `drain()` alone leaves the workers running. Getting it wrong is silent — the pool looks closed and the last few tasks are gone. Three things to know: - **`force` and `timeout` are mutually exclusive.** A forced close has nothing to wait for, so passing both is a contradiction rather than a precedence rule, and it throws rather than silently preferring one. - **A drain that cannot finish does not block the close.** A timeout, an abort or too many waiters is logged and the pool terminates anyway — a close that refuses to finish because the drain timed out is worse than one that terminates. - **It resolves with the final `getStats()` snapshot**, the same shape `drain()` resolves with, so a caller can log the closing state without a second call. ## Recommendations - Use `PowerPool` when you need a small, managed pool of Workers with automatic queuing and idle termination. - Prefer sending plain objects — `PowerPool` encodes them into a framed, transferable `Uint8Array` to avoid structured-clone copies. For broadcasts, each worker receives an independently encoded transferable buffer when no transfer list is provided. - Register `error` / `messageerror` listeners to handle and log underlying Worker problems; the pool forwards these events to registered listeners. Note: Node's `worker_threads` does not emit `messageerror` natively. `PowerPool` normalizes cross-platform behavior: when binary decoding fails the pool will emit a `messageerror` event on the pool-level bus so listeners receive the event even if the underlying worker implementation lacks native `messageerror` support. The pool also still forwards the raw binary payload to `onmessage` so existing consumers receive the data. - Tune `size`, `maxSize`, `idleTimeout` and `maxTasksPerWorker` for your workload. When using many short tasks, a small pool with aggressive queuing often performs best. ## Complexity & Performance Tips - **Amortized cost:** `postMessage` and `broadcast` will try direct dispatch first and then queue. Use `postMessageBatch` to amortize per-item overhead when enqueuing many tasks. - **Encoding & transfers:** When `transfer` is omitted and you supply a plain object, `PowerPool` encodes the object to a framed, transferable `Uint8Array`. This avoids structured-clone copies but does allocate a buffer per encoded item. If you share the same large payload across many workers, `encodeMessage` it once and pass the frame in `transfer` to avoid repeated encoding. - **Batched enqueue:** `postMessageBatch` prepares each item once and uses `PowerQueue.pushMany` to enqueue remaining items in one operation, reducing O(n) push overhead. - **Awaiting responses:** `options.awaitResponse` introduces per-item Promise bookkeeping and correlation ids; for very large batches prefer fire-and-forget and implement separate result aggregation inside workers if possible. - **Tuning concurrency:** `maxTasksPerWorker` controls soft saturation per worker. Raising it increases parallelism per worker but can make latency variance higher; use `getStats()` and EWMA metrics to tune the smoothing and thresholds. - **stopThePress usage:** `stopThePress` and `stopThePressBatch` terminate inflight work and reject pending responses. By default they recreate replacement workers; pass `options.recreateWorkers = false` to skip recreation when you prefer to keep the pool reduced. ## PowerPool Examples — Advanced ```javascript // Efficient batch with pre-encoded transferable buffer (avoid per-item encoding) const shared = encodeMessage({ big: 'payload', repeated: true }); const batch = Array.from({ length: 100 }, () => ({ message: shared, transfer: [shared.buffer] })); // All workers will receive the same transferable buffer (caller responsible for reuse semantics). pool.postMessageBatch(batch); // Use stopThePress to cancel queued work and replace with urgent tasks pool.stopThePress({ command: 'flush-and-run' }, null, { recreateWorkers: true }); // Await responses for a small batch (per-item Promises) const r = pool.postMessageBatch([{ message: { req: 'a' } }, { message: { req: 'b' } }], { awaitResponse: true, timeout: 5000, }); const responses = await Promise.all(r.map((p) => (p instanceof Promise ? p : Promise.resolve(p)))); console.log('batch responses', responses); ```