@polyweave/pool (1.0.8)
Installation
@polyweave:registry=https://hub.kl1.tenere.ai/api/packages/Polyweave/npm/npm install @polyweave/pool@1.0.8"@polyweave/pool": "1.0.8"About this package
@polyweave/pool
Concurrency-capped task scheduling, pauseable request queues, and round-robin requester pools.
Published as @polyweave/pool.
Synopsis
import { createSharedScheduler, createQueuedRequester, createRequesterPool } from '@polyweave/pool';
const scheduler = createSharedScheduler({ concurrency: 3 });
function makeRequest(prompt, done) {
done(null, `response to: ${prompt}`);
}
const queued = createQueuedRequester(makeRequest, scheduler);
queued('hello', (err, result) => {
console.log(result); // 'response to: hello'
});
scheduler.pauseFor(5000); // pause all execution for 5 seconds
const pool = createRequesterPool({
createRequester: (i) => (prompt, done) => done(null, `worker-${i}: ${prompt}`),
size: 4,
scheduler
});
Install
npm install @polyweave/pool
Why
Adapter requesters (LLM API calls) need concurrency gating to avoid rate limits. This module decouples scheduling from request logic: a createSharedScheduler enforces a concurrency cap, createQueuedRequester wraps any callback-style requester to run through the scheduler, and createRequesterPool distributes load round-robin across N requester instances. The scheduler also supports time-based pausing (pauseUntil/pauseFor) for backpressure or rate-limit compliance windows. Zero dependencies.
API
createSharedScheduler(options?)
Creates a concurrency-capped task scheduler with pause support.
Signature
function createSharedScheduler(options?: {
concurrency?: number
}): {
schedule: (task: (release: () => void) => void) => void,
pauseUntil: (timestampMs: number) => void,
pauseFor: (ms: number) => void,
getActive: () => number,
getQueueLength: () => number
}
Parameters
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
options.concurrency |
number |
No | 1 |
Maximum simultaneous in-flight tasks. Must be a finite integer >= 1. |
Error Conditions
| Condition | Error Type | Message |
|---|---|---|
concurrency is not finite or < 1 |
Error |
'Invalid scheduler concurrency: <value>' |
schedule(task) called with non-function |
Error |
'Scheduler task must be a function' |
Return Value
| Method | Signature | Description |
|---|---|---|
schedule |
(task: (release: () => void) => void) => void |
Enqueues a task. The task receives a release callback that MUST be called when the task completes. The scheduler auto-pumps after each release. |
pauseUntil |
(timestampMs: number) => void |
Suspends execution until the given Unix timestamp (ms). If timestampMs is not finite, this is a no-op. |
pauseFor |
(ms: number) => void |
Suspends execution for ms milliseconds. Delegates to pauseUntil(Date.now() + ms). If ms is not finite or <= 0, this is a no-op. |
getActive |
() => number |
Returns the current number of in-flight tasks. |
getQueueLength |
() => number |
Returns the number of tasks waiting in the queue. |
createQueuedRequester(requester, scheduler)
Wraps a callback-style requester to enqueue through a scheduler. Returns the requester unchanged if no scheduler is provided.
Signature
function createQueuedRequester(
requester: (prompt: any, done: (err: Error | null, result: any) => void, metadata?: any) => void,
scheduler?: { schedule: (task: (release: () => void) => void) => void } | null
): (prompt: any, done: (err: Error | null, result: any) => void, metadata?: any) => void
Parameters
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
requester |
function |
Yes | — | The underlying requester to wrap. Signature: (prompt, done, metadata?). |
scheduler |
object |
No | — | A scheduler with a schedule method (e.g., from createSharedScheduler). If scheduler is falsy or lacks schedule, the original requester is returned unmodified. |
Error Conditions
| Condition | Error Type | Message |
|---|---|---|
requester is not a function |
Error |
'Queued requester requires a function' |
Return Value
Returns a function with the same signature as requester. When called, the execution is deferred to scheduler.schedule(). The release callback is wired to the done callback so the scheduler only releases after the requester completes.
If scheduler is null/undefined or lacks a schedule method, the original requester is returned unchanged (passthrough).
createRequesterPool(options?)
Creates a round-robin pool of requesters, optionally wrapped with a scheduler for concurrency gating.
Signature
function createRequesterPool(options?: {
createRequester: (index: number) => (prompt: any, done: (err: Error | null, result: any) => void, metadata?: any) => void,
size?: number,
scheduler?: { schedule: (task: (release: () => void) => void) => void }
}): (prompt: any, done: (err: Error | null, result: any) => void, metadata?: any) => void
Parameters
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
options.createRequester |
function |
Yes | — | Factory function called once per pool slot. Receives the zero-based slot index. Must return a requester function with signature (prompt, done, metadata?). |
options.size |
number |
No | 1 |
Number of requester instances to create. Must be a finite integer >= 1. |
options.scheduler |
object |
No | undefined |
Optional scheduler for concurrency limiting. If provided, the pooled requester is wrapped via createQueuedRequester. If omitted, requests are dispatched immediately (no gating). |
Error Conditions
| Condition | Error Type | Message |
|---|---|---|
createRequester is not a function |
Error |
'Requester pool requires a createRequester function' |
size is not finite or < 1 |
Error |
'Invalid requester pool size: <value>' |
Return Value
Returns a pooled requester function with signature (prompt, done, metadata?). Each call selects the next requester in round-robin order via an incrementing cursor (cursor % requesters.length). If a scheduler was provided, the returned function is additionally wrapped by createQueuedRequester.
Tests
npm test
npm run bench
Covers:
- scheduler.test.js — Concurrency enforcement (concurrency=1 through 5), queued execution ordering, round-robin distribution across pool instances, pause delay semantics.
- pool.fuzz.test.js — Fuzz concurrency invariants: total in-flight never exceeds cap, all tasks eventually complete.
- rigor.test.js — Property-based tests on scheduler/pool contracts.
- benchmark/ —
npm run benchruns a throughput benchmark (benchmark/pool-bench.js).
Requirements
- Node.js 22 or later.
- ESM only (
"type": "module"). - Zero runtime dependencies.
Dependencies
Development Dependencies
| ID | Version |
|---|---|
| @rigor/core | * |
| fast-check | * |