/**
* PowerPoolShutdownError
*
* Error thrown when the `PowerPool` is shut down and pending tasks are rejected.
*
* Carries `code === 'ERR_POOL_TERMINATED'`, the same code the pool uses for the
* synchronous throw from a dispatch method on a shut-down pool. Both mean the
* same thing — the pool is finished, so is the work — and `guides/errors.md`
* tells callers to branch on `err.code`. Without it, a caller awaiting a
* response at shutdown got an error with no code and fell through the
* documented `switch` to `default`, which is the case most likely to be hit:
* shutting down is exactly when pending promises are still outstanding.
* `name` is unchanged, so `err.name === 'PowerPoolShutdownError'` keeps
* working.
*
* @class PowerPoolShutdownError
* @extends {Error}
* @public
*/
export class PowerPoolShutdownError extends Error {
constructor(message?: string);
code: string;
}
/**
* @typedef {import('./jsdoc-types.js').WorkerObj} WorkerObj
*/
/**
* PostMessage and pending-response typedefs are defined centrally to avoid
* duplication across multiple helper modules. Import aliases are used here
* so typedoc and editors can resolve the shape while keeping local docs
* concise.
* @typedef {import('./jsdoc-types.js').PostMessageOptions} PostMessageOptions
*/
/**
* @typedef {import('./jsdoc-types.js').PendingResponseEntry} PendingResponseEntry
*/
/**
* @typedef {import('./jsdoc-types.js').PowerPoolOptions} PowerPoolOptions
*/
/**
* Manager for a pool of web workers.
*
* @example
* import MinionWorker from './worker.js?worker&inline'
* const pool = new PowerPool(MinionWorker, { size: 4, idleTimeout: 30000 });
* pool.onmessage = (e) => { logger.log(e.data); };
* pool.postMessage({ payload: {} });
*
* @example
Explicit Resource Management (Node >= 22.12 / modern browsers)
* // `Symbol.dispose` runs synchronously, so `using` hard-stops the pool at
* // scope exit. Prefer `await using` to drain in-flight work first.
* import MinionWorker from './worker.js?worker&inline'
* {
* using pool = new PowerPool(MinionWorker, { size: 4 });
* await pool.postMessage({ payload: {} }, undefined, { awaitResponse: true });
* } // pool.shutdown() is called here, synchronously
*
* // To drain pending tasks before terminating:
* {
* await using pool = new PowerPool(MinionWorker, { size: 4 });
* await pool.postMessage({ payload: {} }, undefined, { awaitResponse: true });
* } // pool[Symbol.asyncDispose]() drains then terminates
*/
/**
* PowerPool
*
* Manager for a pool of worker-like objects providing task dispatch, queuing,
* autoscaling, and lifecycle management. See constructor docs for options.
*
* @class PowerPool
* @public
*/
export class PowerPool {
/**
* Create a PowerPool.
*
* @param {Function|string} workerSource - A Worker constructor, a worker factory, or a relative path string. If the provided function is not constructable, it is invoked directly; if a string path is provided, the pool attempts to resolve it via `new URL(path, import.meta.url)` before falling back to a plain `Worker(path)`.
* @param {PowerPoolOptions=} options
* @param {number} [options.size] - Initial number of workers to create.
* @param {number} [options.minSize=1] - Minimum number of workers to keep alive.
* @param {number} [options.maxSize] - Maximum number of workers allowed in the pool. The pool coerces this value to be at least `minSize`.
* @param {Object} [options.workerOptions] - Options forwarded to the Worker constructor when using a string path.
* @param {number} [options.maxTasksPerWorker=Infinity] - Soft capacity per worker before considering it busy.
* @param {number} [options.idleTimeout=60000] - Milliseconds after which idle workers (beyond `minSize`) will be terminated.
* @param {boolean} [options.taskQueue=true] - Whether to queue tasks when all workers are busy.
* @param {'enqueue'|'drop-oldest'|'drop-newest'|'reject'} [options.queuePolicy='enqueue'] - Queue overflow behavior when the pool is saturated.
* @param {boolean} [options.lazy=true] - If true, defer creating workers up to `size` until demand; only `minSize` workers are created at construction.
* @param {number} [options.slowTaskThreshold=Infinity] - Task duration (ms) above which a completed task is counted as "slow". When finite, `stats().performance.percentSlowTasks` reports the exact percentage of tasks exceeding this threshold. Defaults to `Infinity` (disabled; `percentSlowTasks` stays `0`).
* @param {number|'auto'} [options.maxQueueLength=Infinity] - Hard cap on queued tasks. With the default `queuePolicy: 'enqueue'` and no cap, a saturated pool grows its queue until the process runs out of memory; a cap makes the overflow observable instead - the incoming task is refused (`false`, or `ERR_POOL_QUEUE_FULL` when awaiting a response). `drop-oldest` still evicts the oldest to make room. Pass `'auto'` to resolve the cap to `maxSize ** 2` — a starting point for the common case of an unbounded queue, not a default. The resolved number is reported by `getStats().maxQueueLength`, and a later `resize()` does not re-derive it.
* @param {number} [options.maxDrainWaiters=100] - Cap on concurrent `drain()` waits, so a caller that drains in a loop cannot accumulate unbounded `idle` listeners. (Internally `DEFAULT_MAX_DRAIN_WAITERS`; not exported.)
*/
constructor(workerSource: Function | string, options?: PowerPoolOptions | undefined, ...args: any[]);
minSize: number;
maxSize: number;
idleTimeout: number;
taskQueueEnabled: boolean;
/**
* Number of `drain()` calls currently *waiting* for idle. Each one holds an
* `idle` listener, so this is the bound that keeps a caller draining in a
* loop from accumulating listeners without limit. See
* `DEFAULT_MAX_DRAIN_WAITERS`.
* @type {number}
*/
/** @type {number|null} */
/** @type {WorkerObj[]} */
workers: WorkerObj[];
queue: PowerQueue | PowerPriorityQueue;
/** number of currently active (dispatched) tasks across all workers */
/** whether the pool is considered idle (no active tasks and empty queue) */
/** whether queued dispatch is paused */
/**
* Wire protocol for object messages.
*
* - `'framed'` (**default since 2.0**) posts a `PowerMessageCodec`
* envelope: `[version][codec][length][payload]`. Workers read
* `decodeMessage(e.data).value` instead of `u82o(e.data)`, binary frames
* survive intact, and the version byte lets the protocol evolve without
* another flag day.
* - `'legacy'` restores the 1.x behaviour: a bare `Uint8Array` of JSON,
* sniffed on the way back in. Provided so a worker can be migrated on its
* own schedule. See the migration note in guides/powerPool.md.
* - `'negotiated'` starts out identical to `'framed'` and upgrades **per
* worker**: a worker that advertises the native carrier with
* `announceCapabilities()` is sent the structured-clone carrier instead,
* which preserves `Map`, `Set`, `Date`, `BigInt` and cycles that the JSON
* frame silently destroys. Workers that do not advertise keep getting the
* frame, so a pool can be switched on before any worker is ready.
*
* @type {'framed'|'legacy'|'negotiated'}
*/
/**
* Whether this runtime can structured-clone at all. Checked once here so
* `'negotiated'` on a runtime without `structuredClone` degrades to the
* framed path rather than throwing per message.
*/
/**
* Terminal flag. Set by `shutdown()` / `terminate()`; once true the pool
* refuses to dispatch, enqueue or grow, so a late `postMessage()` cannot
* resurrect it (which previously created a worker with no reaper
* interval, pinning the Node.js process).
* @type {boolean}
*/
/** @type {string | number | null} */
_metrics: {
unregister: () => boolean;
name: string;
} | null;
/**
* Log debug information about swallowed errors when debug logging is enabled.
* @param {*} err - The swallowed error, or falsy when the call is informational.
* @param {string} [msg]
* @returns {void}
* @private
*/
/** Ensure the reaper interval exists; recreate it if missing. @private */
_createPendingResponsePromise(correlationId: any, options: any): {
pendingPromise: Promise;
correlationKey: any;
};
/**
* Resolve a deferred message into the carrier this specific worker expects.
*
* This is where protocol negotiation is decided, and the reason it has to be
* here rather than in `_prepareForTransfer` is that the target worker is not
* known when a message is prepared: a queued item can be handed to any worker
* by either drain site. Deciding at dispatch is what makes the decision
* *per worker*, which is the whole point — a pool with one upgraded worker
* and nine legacy ones is the state this feature exists for.
*
* Non-deferred items pass straight through, so `'framed'`, `'legacy'` and
* every binary message keep their existing path untouched.
*
* @private
* @param {import('./jsdoc-types.js').WorkerObj} obj - Target worker, if known.
* @param {{message: *, transfer: (TransferList|undefined), deferred?: boolean}} prepared
* @returns {{message: *, transfer: (TransferList|undefined)}}
*/
/**
* Wrap a deferred message in the native envelope, or return `null` when this
* runtime cannot do it.
*
* The transfer list is the subtle part. A transfer list names buffers inside
* the object being posted, and `postMessage` **detaches** them — so naming
* the caller's buffers would detach the caller's data, which is a data-loss
* bug that looks like a successful post. When the payload carries no binary
* there is nothing to name, and the envelope is posted with no list and no
* copy: the platform clones it, which is the whole reason for choosing this
* carrier. When it does carry binary, a private copy is made first and *its*
* buffers are transferred.
*
* @private
* @param {PreparedItem} prepared - Already framed or marked `deferred`.
* @returns {{message: object, transfer: (TransferList|undefined)}|null}
*/
/**
* Record what a worker says it can decode, and announce the change.
*
* The rules here are deliberately conservative, because a worker that
* over-claims costs a broken message and a worker that under-claims costs
* nothing but speed:
*
* - `native` is only granted when the runtime can structured-clone. A worker
* that advertises it on a runtime that cannot is talking about a carrier
* that does not exist here, and honouring that would throw per message.
* - `json` is always retained. It is what the pool sends until this
* announcement lands, and a worker that claims `native` alone has not said
* it can read the messages it has already been sent.
*
* @private
* @param {import('./jsdoc-types.js').WorkerObj} obj
* @param {{codecs: string[]}} announcement
*/
/**
* Post a prepared message to a specific worker object and update bookkeeping.
* Returns the `pendingPromise` when `wantResponse` is true, otherwise `true` on success.
* On failure, rejects/cleans up the pending response when applicable and
* returns `pendingPromise` (when awaiting) or `false`.
* @param {import('./jsdoc-types.js').WorkerObj} obj
* @param {{message: *, transfer: (TransferList|undefined)}} prepared
* @param {number} startTime
* @param {boolean} wantResponse
* @param {string|undefined} correlationKey
* @param {Promise} pendingPromise
* @param {AbortSignal} [signal] - Per-task abort signal (PISC-003), carried
* through to the in-flight policy in `_dispatchToWorker`.
* @returns {Promise|boolean}
* @private
*/
/**
* Report a failed `postMessage` and clean up the pending response for it.
*
* Single choke point for the failure half of `_postToWorkerObj`, so the two
* posting paths (direct to the underlying worker, through the wrapper) cannot
* drift on what the caller is told. The important property is that a
* post that failed *after* possibly detaching buffers is still reported with
* its **original** error - never with a downstream `DataCloneError` from a
* retry against the same buffers.
*
* @param {any} err - The error thrown by `postMessage`.
* @param {boolean} wantResponse - Whether the caller is awaiting a response.
* @param {string|undefined} correlationKey - Pending-response key to reject.
* @param {Promise} pendingPromise - The caller's pending promise.
* @param {{scope: string}} [info] - Debug-log scope.
* @returns {Promise|boolean} `pendingPromise` when awaiting a response,
* `false` otherwise - matching `postMessage`'s contract.
* @private
*/
/**
* Post one prepared item to one worker, and account for it.
*
* The single dispatch choke point. There were eight `postMessage` call sites
* before this, each with its own copy of the post-and-account sequence, and
* they drifted in four different directions:
*
* - the single-worker batch fast path skipped the deferred encode, so every
* batched request reached the worker as raw JSON and failed at the first line
* (`POOL-001` — `unsupported protocol version 123`, since `{` is 123);
* - the inline drain in `worker.onmessage` never recorded which worker took the
* task, so a queued `awaitResponse` promise could never be settled
* (`POOL-002`);
* - `stopThePress` enqueued the caller's object with no preparation at all
* (`POOL-003`);
* - only the direct path reported a failed post through `_failPost`, and only
* the batch loops and `broadcast` had their own idle-state policy.
*
* All three defects were *the same defect*. A property is checkable only when
* there is one place that owns it, so a fourth copy of a sequence is a fourth
* place to forget a step.
*
* @private
* @param {*} workerObj - A `WorkerObj`. Left untyped to match the other
* private dispatch helpers here; annotating it stricter than what the worker
* records actually are would add an error at every call site for no gain.
* @param {PreparedItem} prepared - Must be prepared. A `deferred` item is
* framed here, which is why every route can hand one over unchanged.
* @param {{correlationId?: (string|undefined), startTime?: (number|undefined), signal?: (AbortSignal|null)}=} options
* `correlationId` for a task awaiting a response, so the pending entry can be
* tied to this worker; `startTime` when the caller already took a timestamp
* and wants every record in a batch or a broadcast to share it.
* @returns {number} The `startTime` used, for a caller that has not taken one.
*/
_recordQueueWait(wait: any): void;
/**
* Report a batched post that could not be dispatched.
*
* `postMessageBatch` answers with a per-item boolean, so a failure is a
* `false` in an array and — before this — nothing else. Two of the three
* failure sites caught the error and dropped it on the floor: no log, no
* `pool:error` event, no count. A caller seeing `false` could not tell a
* dispatch failure from a busy worker, and an operator had no way to know a
* batch had silently lost half its items. The third site already logged and
* emitted, which is the pattern; this gives all three one, and adds the count.
*
* Every step is individually guarded because this runs **inside a `catch`**:
* a logger or an event bus that throws would replace the original failure with
* its own, and the item would be reported as neither failed nor sent.
*
* @param {any} err - The error thrown while dispatching.
* @param {string} scope - Which site failed, for the log and the event.
* @returns {false} So a call site can assign it straight to its result slot.
* @private
*/
/**
* Report a failed `postMessage` and clean up the pending response for it.
* @param {any} err - The error thrown by `postMessage`.
* @param {boolean} wantResponse - Whether the caller is awaiting a response.
* @param {string|undefined} correlationKey - Pending-response key to reject.
* @param {Promise} pendingPromise - The caller's pending promise.
* @param {{scope: string}} [info] - Debug-log scope.
* @returns {Promise|boolean} `pendingPromise` when awaiting a response,
* `false` otherwise - matching `postMessage`'s contract.
* @private
*/
/**
* Attempt to grow the pool by adding a worker and dispatching the message.
* Preserves the same pending-response cleanup semantics as inline logic.
* @param {*} message
* @param {TransferList} [transfer]
* @param {import('./jsdoc-types.js').PostMessageOptions} [options]
* @param {number} startTime
* @param {boolean} [wantResponse]
* @param {string} [correlationId]
* @param {number} [timeout]
* @returns {Promise|boolean}
* @private
*/
/**
* Resolve and validate the per-item correlation ids for a batch that expects
* responses.
*
* Two failure modes are checked, and both are checked *before* any dispatch:
*
* 1. A duplicate **within** the batch. A `correlationIdFactory` is caller
* code; a constant or a sloppy one returns the same id twice, and the
* second `postMessage` under a live key rejects the first waiter and
* takes the key over. The caller would hold a batch where one promise is
* already rejected and the last result resolves for all of them.
* 2. A collision with an id that is **already in flight** from an earlier
* call. Same mechanism, same outcome, and it is silently the caller's
* fault because the factory is a global naming scheme, not a local one.
*
* @param {{message:*,transfer?:Transferable[]}[]} items - The batch items.
* @param {Function} factory - The caller's `correlationIdFactory`.
* @returns {string[]} One validated id per item, positionally aligned.
* @throws {Error} `ERR_POOL_DUPLICATE_CORRELATION_ID` on the first collision.
* @private
*/
/**
* Reserve room for `count` more tasks in the task queue, honouring
* `maxQueueLength` and the configured `queuePolicy`.
*
* Without a cap this is a no-op and returns `count`, which is what keeps the
* pre-2.0 behaviour intact: `queuePolicy: 'enqueue'` grew the queue until the
* process ran out of memory, with no event and no error. The cap exists to
* make that overflow *observable*, so the interesting decision is **which
* task gets refused**:
*
* - `drop-oldest` keeps its documented meaning: evict the oldest to make room
* for the newest, so the queue length never exceeds the cap.
* - every other policy refuses the **incoming** task. That is the honest
* reading of a caller-provided bound - the newest arrival is the one that
* did not fit, and refusing it keeps the work already accepted.
*
* @param {number} count - How many tasks are about to be queued.
* @returns {number} How many of them may be queued. The remainder must be
* refused by the caller.
* @private
*/
/**
* Enqueue or reject a prepared message according to the configured queue policy.
* Returns `pendingPromise`/`true`/`false` to match `postMessage` semantics.
*
* @param {PreparedItem} prepared - Already prepared by `_prepareForTransfer`
* or `prepareBuffers`. Taking the item whole is deliberate: a signature of
* `(message, transfer)` would let a caller hand in an unprepared object,
* which is the shape of the POOL-003 defect.
* @param {import('./jsdoc-types.js').PostMessageOptions} [options] - The caller's
* post options, read here for `signal` (PISC-003).
* @private
*/
/**
* Single choke point for retiring a worker from the pool.
*
* Every removal path - `shutdown()`, `removeWorker()`, `resize()`,
* `_autoScaleTick()`, `_reapIdleWorkers()` and `_resetPoolForStopThePress()`
* - routes through here so the active-task accounting, the terminated-worker
* statistics and the underlying-worker mapping can no longer drift apart
* between paths (the idle reaper previously skipped the statistics
* entirely, inflating `getStats().performance.timePerTask` over time).
*
* Note: this deliberately does **not** remove the entry from `this.workers`;
* callers own the array bookkeeping (some paths pop, some swap-remove, the
* shutdown path clears the whole list).
*
* @param {WorkerObj|null} workerObj - Worker entry to retire.
* @param {string} [reason] - Why the worker is being removed, used for
* debug logging only.
* @returns {number|null} The retired worker id, or `null` when nothing was
* retired.
* @private
*/
/**
* Reject every pending response that was dispatched to a worker that is being
* retired. The response for those tasks lives in that worker and is never
* coming, so the promise is settled here rather than on its timeout.
*
* Scans the pending map rather than keeping a per-worker index, because the
* number of outstanding responses is small and bounded by
* `awaitResponseTimeout`, while the index would have to be maintained on the
* response path too — and a per-worker index that is updated in one place and
* not another silently under-rejects, which is the failure this exists to
* prevent.
*
* @param {number|string|null} workerId - The worker being retired.
* @param {string} [reason] - Why, used in the error message only.
* @returns {number} How many pending responses were rejected.
* @private
*/
/**
* Record which worker a pending response was dispatched to, so retirement
* can find it. Stamped after the post succeeds: a response that never left
* the pool is not owed anything by the worker it was aimed at.
*
* @param {string|number|null|undefined} correlationKey - The pending key.
* @param {number|string|null} workerId - The worker that received the task.
* @returns {void}
* @private
*/
/**
* Throws when the pool has been shut down. Called from every public entry
* point that would otherwise dispatch, enqueue or grow workers.
* @private
*/
/**
* Clear lifecycle timer intervals used by the pool.
* @private
*/
/**
* Shutdown the pool: clear timers, reject pending responses, terminate workers,
* and clear internal queues. This is a full stop that prevents background
* timers from keeping the process alive.
*
* Shutdown is **final**: the pool refuses every subsequent `postMessage()`,
* `postMessageBatch()`, `addWorker()` and `resize()` with an
* `ERR_POOL_TERMINATED` error rather than silently recreating workers.
* Create a new `PowerPool` to start again.
*
* @returns {void}
*/
/**
* Graceful shutdown as a single entry point: wait for in-flight work, then
* terminate. **New in 2.1.0** (PISC-011).
*
* The `shutdown()` / `drain()` / `[Symbol.asyncDispose]` model is defensible and
* stays exactly as it is — this is a composition of it, not a replacement. 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. Getting the order
* wrong is the failure this prevents: `shutdown()` alone abandons in-flight work,
* and `drain()` alone leaves the workers running.
*
* ```js
* await pool.close(); // wait for work, then terminate
* await pool.close({ timeout: 5000 }); // wait at most 5 s, then terminate
* await pool.close({ force: true }); // terminate now, abandon work
* ```
*
* `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.
*
* @param {{force?: boolean, timeout?: number, signal?: AbortSignal}} [options]
* @returns {Promise