@studnicky/throttle
Generic async operation throttle with sliding window concurrency control.
Install
pnpm add @studnicky/throttleUsage
Create a Throttle instance with Throttle.create(config), then pass any operation to execute. Configuration is supplied once at construction and is not exposed through setters. The instance tracks stats and enforces the concurrency limit:
const throttle = Throttle.create({ 'concurrencyLimit': 3 });
const results = await Promise.all([
throttle.execute(async () => { await setTimeout(1); const result = 0; return result; }),
throttle.execute(async () => { await setTimeout(1); const result = 1; return result; }),
throttle.execute(async () => { await setTimeout(1); const result = 2; return result; }),
throttle.execute(async () => { await setTimeout(1); const result = 3; return result; }),
throttle.execute(async () => { await setTimeout(1); const result = 4; return result; }),
throttle.execute(async () => { await setTimeout(1); const result = 5; return result; })
]);
const stats = throttle.getStats();
console.log('basicThrottle stats:', stats);
console.log('results:', results);Drain
Call drain() to stop accepting new work and wait for all active and queued operations to finish gracefully:
const throttle = Throttle.create({ 'concurrencyLimit': 5 });
// Submit 4 operations before draining
const pending = Promise.all([
throttle.execute(async () => { await setTimeout(1); const result = 10; return result; }),
throttle.execute(async () => { await setTimeout(1); const result = 20; return result; }),
throttle.execute(async () => { await setTimeout(1); const result = 30; return result; }),
throttle.execute(async () => { await setTimeout(1); const result = 40; return result; })
]);
// drain() stops accepting new work and waits for all queued/active ops to finish
await throttle.drain();
// All submitted operations should have resolved by now
const results = await pending;
const stats = throttle.getStats();
console.log('drainThrottle stats:', stats);
console.log('results:', results);Try it
Lifecycle hooks
TracingThrottle subclasses Throttle and overrides eight hooks: onAcquire, onContended, onAcquireWait, onWindowSlide, onRelease, onDrainStart, onDrainComplete, and the FSM transition hook onEnter. With concurrencyLimit=2 and 4 ops submitted, watch the first two acquire immediately, the second two contend and queue, then window-slide events as slots free up. A drain() call then drains the throttle gracefully.
Abort support
const throttle = Throttle.create({ concurrencyLimit: 3 });
// Queued operations resolve with undefined, active ones continue silently
await throttle.abort();Public API
Import the runtime API from @studnicky/throttle. Defaults and scheduling constants are implementation details.
Throttle.create(config) validates and copies the supplied configuration into instance-owned state. Adaptive concurrency may adjust the instance's effective limit without mutating the caller's config object.
Use ThrottleConfigEntity.validate(candidate) at an untrusted configuration boundary. It validates against ThrottleConfigEntity.Schema, throws for invalid input, and narrows a valid candidate to ThrottleConfigEntity.Type.
getStats() returns ThrottleStatsEntity.Type. Use the entity-subpath compiled validator at trust boundaries:
import { Throttle } from '@studnicky/throttle';
import { ThrottleStatsEntity } from '@studnicky/throttle/entities';
const throttle = Throttle.create({ concurrencyLimit: 3 });
const stats = throttle.getStats();
if (!ThrottleStatsEntity.validate(stats)) {
throw new Error('invalid throttle statistics');
}Entities
@studnicky/throttle/entities exports every schema namespace in src/entities, including configuration, statistics, abort results, and lifecycle state, event, and effect values.
import { ThrottleConfigEntity } from '@studnicky/throttle/entities';Interfaces
@studnicky/throttle/interfaces exports the ThrottleInterface contract for consumers that accept or implement a throttle abstraction.
import type { ThrottleInterface } from '@studnicky/throttle/interfaces';Exports
| Symbol | Purpose | Import path |
|---|---|---|
Throttle | Creates and runs a sliding-window concurrency throttle. | @studnicky/throttle |
ThrottleInterface | Defines the consumer-facing throttle contract. | @studnicky/throttle |
ThrottleAbortedError | Represents an aborted throttle operation. | @studnicky/throttle |
ThrottleDrainingError | Represents work rejected while the throttle drains. | @studnicky/throttle |
Observability hooks
Subclass Throttle and override any of the protected hooks below to add logging, metrics, or tracing without coupling the throttle core to any observability library.
| Hook | When it fires | Args |
|---|---|---|
onEnter(to, from) | Every FSM state transition | to: ThrottleStateEntity.Type, from: ThrottleStateEntity.Type |
onAcquire(activeCount, queuedCount) | A slot is granted immediately (window not full) | activeCount: number, queuedCount: number |
onContended(activeCount, queuedCount) | A caller arrives at a saturated window and is about to queue | activeCount: number, queuedCount: number |
onAcquireWait(queuedCount) | A caller has been pushed onto the queue; queue depth after enqueue | queuedCount: number |
onWindowSlide(activeCount, queuedCount) | A queued caller is dequeued and granted a slot; fires before its promise resolves | activeCount: number, queuedCount: number |
onRelease(activeCount, totalExecuted) | A concurrency slot is freed after an operation completes | activeCount: number, totalExecuted: number |
onDrainStart(activeCount, queuedCount) | drain() is called and draining mode begins | activeCount: number, queuedCount: number |
onDrainComplete(totalExecuted) | All operations finish and the throttle transitions draining → idle | totalExecuted: number |
onAbortStart(cancelledCount) | abort() executes and is about to cancel operations | cancelledCount: number |
onAdaptiveAdjust(previousLimit, newLimit) | Adaptive concurrency changes the concurrency limit | previousLimit: number, newLimit: number |
onReject(reason) | An operation's async function throws or rejects | reason: unknown |
class TracingThrottle extends Throttle {
readonly acquireEvents: { 'activeCount': number; 'queuedCount': number }[] = [];
readonly contendedEvents: { 'activeCount': number; 'queuedCount': number }[] = [];
readonly acquireWaitEvents: { 'queuedCount': number }[] = [];
readonly windowSlideEvents: { 'activeCount': number; 'queuedCount': number }[] = [];
readonly releaseEvents: { 'activeCount': number; 'totalExecuted': number }[] = [];
readonly drainStartEvents: { 'activeCount': number; 'queuedCount': number }[] = [];
readonly drainCompleteEvents: { 'totalExecuted': number }[] = [];
protected override onEnter(to: ThrottleStateEntity.Type, from: ThrottleStateEntity.Type): void {
console.log(`[throttle] fsm transition: ${from} → ${to}`);
}
protected override onAcquire(activeCount: number, queuedCount: number): void {
console.log(`[throttle] onAcquire activeCount=${activeCount} queuedCount=${queuedCount}`);
this.acquireEvents.push({ 'activeCount': activeCount, 'queuedCount': queuedCount });
}
protected override onContended(activeCount: number, queuedCount: number): void {
console.log(`[throttle] onContended (window saturated) activeCount=${activeCount} queuedCount=${queuedCount}`);
this.contendedEvents.push({ 'activeCount': activeCount, 'queuedCount': queuedCount });
}
protected override onAcquireWait(queuedCount: number): void {
console.log(`[throttle] onAcquireWait (caller queued) queuedCount=${queuedCount}`);
this.acquireWaitEvents.push({ 'queuedCount': queuedCount });
}
protected override onWindowSlide(activeCount: number, queuedCount: number): void {
console.log(`[throttle] onWindowSlide (slot freed, waiter promoted) activeCount=${activeCount} queuedCount=${queuedCount}`);
this.windowSlideEvents.push({ 'activeCount': activeCount, 'queuedCount': queuedCount });
}
protected override onRelease(activeCount: number, totalExecuted: number): void {
console.log(`[throttle] onRelease activeCount=${activeCount} totalExecuted=${totalExecuted}`);
this.releaseEvents.push({ 'activeCount': activeCount, 'totalExecuted': totalExecuted });
}
protected override onDrainStart(activeCount: number, queuedCount: number): void {
console.log(`[throttle] onDrainStart activeCount=${activeCount} queuedCount=${queuedCount}`);
this.drainStartEvents.push({ 'activeCount': activeCount, 'queuedCount': queuedCount });
}
protected override onDrainComplete(totalExecuted: number): void {
console.log(`[throttle] onDrainComplete totalExecuted=${totalExecuted}`);
this.drainCompleteEvents.push({ 'totalExecuted': totalExecuted });
}
}
// Concurrency limit of 2; we submit 4 ops so 2 queue immediately.
const throttle = TracingThrottle.create({ 'concurrencyLimit': 2 });
// Submit 4 ops. Ops 0-1 acquire immediately; ops 2-3 contend and wait.
const ops: Promise<number | undefined>[] = [
throttle.execute(async () => { await setTimeout(5); const result = 0; return result; }),
throttle.execute(async () => { await setTimeout(5); const result = 1; return result; }),
throttle.execute(async () => { await setTimeout(5); const result = 2; return result; }),
throttle.execute(async () => { await setTimeout(5); const result = 3; return result; })
];
// Initiate a graceful drain — no new ops accepted; wait for the 4 to finish.
const drainPromise = throttle.drain();
await Promise.all([...ops, drainPromise]);
console.log('acquireEvents:', throttle.acquireEvents);
console.log('contendedEvents:', throttle.contendedEvents);
console.log('acquireWaitEvents:', throttle.acquireWaitEvents);
console.log('windowSlideEvents:', throttle.windowSlideEvents);
console.log('releaseEvents:', throttle.releaseEvents);
console.log('drainStartEvents:', throttle.drainStartEvents);
console.log('drainCompleteEvents:', throttle.drainCompleteEvents);
console.log('stats:', throttle.getStats());The base class never calls any logger or metrics library. All hooks are no-ops by default.