@studnicky/throttle
Generic async operation throttle with sliding window concurrency control.
Install
bash
pnpm add @studnicky/throttleUsage
Build a Throttle instance with the constructor or factory, then pass any operation to execute. The instance tracks stats and enforces the concurrency limit:
ts
const throttle = Throttle.create({ 'concurrencyLimit': 3 });
const results = await Promise.all(
[0, 1, 2, 3, 4, 5].map((i) => {
const result = throttle.execute(async () => {
await setTimeout(1);
return i;
});
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:
ts
const throttle = Throttle.create({ 'concurrencyLimit': 5 });
// Submit 4 operations before draining
const pending = Promise.all(
[1, 2, 3, 4].map((i) => {
const result = throttle.execute(async () => {
await setTimeout(1);
return i * 10;
});
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
Builder
Throttle.builder().withConcurrencyLimit(3).build() constructs the throttle through the fluent builder. Press Execute to submit 6 operations through the concurrencyLimit-3 throttle; all 6 complete in order, and stats confirm 6 total executed with the configured limit.
Builder — fluent throttle construction
/** builder-throttle — construct a Throttle via the fluent builder API. Run: npx tsx packages/throttle/examples/builder-throttle.ts */
import assert from 'node:assert/strict';
import { setTimeout } from 'node:timers/promises';
import { Throttle } from '../src/index.js';
import { BuilderThrottleFixtures } from './fixtures/builderThrottleFixtures.js';
// #region usage
// Build a Throttle with a concurrency limit via the fluent builder.
const throttle = Throttle.builder()
.withConcurrencyLimit(BuilderThrottleFixtures.concurrencyLimit)
.build();
console.log('throttle built via builder:', throttle);
// Submit operations through the throttle; each awaits a tick then returns its index.
const results = await (async (): Promise<(number | undefined)[]> => {
const promises: Promise<number | undefined>[] = [];
for (let i = 0; i < BuilderThrottleFixtures.operationCount; i += 1) {
const result = throttle.execute<number>(async () => {
await setTimeout(BuilderThrottleFixtures.tickMs);
return i;
});
promises.push(result);
}
return await Promise.all(promises);
})();
console.log('results:', results);
console.log('stats:', throttle.getStats());
// #endregion usage
assert.equal(results.length, BuilderThrottleFixtures.operationCount, 'results length must match operationCount');
for (let i = 0; i < BuilderThrottleFixtures.operationCount; i++) {
assert.equal(results[i], i, `result[${i}] must equal ${i}`);
}
assert.equal(throttle.getStats().concurrencyLimit, BuilderThrottleFixtures.concurrencyLimit, 'concurrencyLimit must match fixture');
assert.equal(throttle.getStats().totalExecuted, BuilderThrottleFixtures.operationCount, 'totalExecuted must match operationCount');
assert.equal(throttle.getStats().activeCount, 0, 'activeCount must be 0');
assert.equal(throttle.getStats().queuedCount, 0, 'queuedCount must be 0');
console.log('builder-throttle: all assertions passed');
Output
Press Execute to run this example against the real library.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.
Observed throttle — lifecycle hook trace
/** observedThrottle — subclass Throttle and override hooks to emit trace logs; saturate the window so callers queue and then drain. Run: npx tsx examples/observedThrottle.ts */
import assert from 'node:assert/strict';
import { setTimeout } from 'node:timers/promises';
import type { ThrottleStateType } from '../src/types/ThrottleStateType.js';
import { Throttle } from '../src/index.js';
// #region usage
class TracingThrottle extends Throttle {
static override create(config?: Parameters<typeof Throttle.create>[0]): TracingThrottle {
return new TracingThrottle(config);
}
public constructor(config?: Parameters<typeof Throttle.create>[0]) {
super(config);
}
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: ThrottleStateType, from: ThrottleStateType): 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>[] = [];
for (let i = 0; i < 4; i++) {
ops.push(throttle.execute(async () => {
await setTimeout(5);
return i;
}));
}
// 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());
// #endregion usage
// ── Assertions ────────────────────────────────────────────────────────────────
// 4 ops submitted, concurrencyLimit=2 → 2 immediate acquires, 2 contended/queued
assert.equal(throttle.acquireEvents.length, 2, 'Expected 2 immediate acquire events');
assert.equal(throttle.contendedEvents.length, 2, 'Expected 2 contention events (ops 2 and 3 hit a saturated window)');
assert.equal(throttle.acquireWaitEvents.length, 2, 'Expected 2 enqueue events');
// Queue depths at enqueue time: first waiter sees depth 1, second sees depth 2
assert.equal(throttle.acquireWaitEvents[0]!.queuedCount, 1, 'First waiter: queue depth 1');
assert.equal(throttle.acquireWaitEvents[1]!.queuedCount, 2, 'Second waiter: queue depth 2');
// 2 queued ops each get promoted via a window slide
assert.equal(throttle.windowSlideEvents.length, 2, 'Expected 2 window-slide events (one per dequeued waiter)');
// 4 ops complete → 4 total releases (including the mid-queue releases)
assert.ok(throttle.releaseEvents.length >= 4, 'Expected at least 4 release events');
// drain was called once → drainStart fires once
assert.equal(throttle.drainStartEvents.length, 1, 'Expected 1 drainStart event');
// drain finished with real ops in flight → drainComplete fires once
assert.equal(throttle.drainCompleteEvents.length, 1, 'Expected 1 drainComplete event');
assert.equal(throttle.drainCompleteEvents[0]!.totalExecuted, 4, 'totalExecuted is 4 at drain complete');
// All 4 ops completed
assert.equal(throttle.getStats().totalExecuted, 4, 'All 4 ops completed');
console.log('observedThrottle: all assertions passed');
Output
Press Execute to run this example against the real library.Abort support
typescript
const throttle = Throttle.create({ concurrencyLimit: 3 });
// Queued operations resolve with undefined, active ones continue silently
await throttle.abort();Builder
typescript
import { Throttle } from '@studnicky/throttle';
const throttle = Throttle.builder()
.withConcurrencyLimit(8)
.build();Subpath exports
| Subpath | Contents |
|---|---|
@studnicky/throttle | Throttle, ThrottleBuilder, errors, interfaces, type guards |
@studnicky/throttle/constants | Default configuration constants |
@studnicky/throttle/errors | ConfigurationError, ThrottleAbortedError, ThrottleDrainingError |
@studnicky/throttle/interfaces | Interface types |
@studnicky/throttle/types | AdaptiveConfigInputType |
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: ThrottleStateType, from: ThrottleStateType |
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 |
ts
class TracingThrottle extends Throttle {
static override create(config?: Parameters<typeof Throttle.create>[0]): TracingThrottle {
return new TracingThrottle(config);
}
public constructor(config?: Parameters<typeof Throttle.create>[0]) {
super(config);
}
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: ThrottleStateType, from: ThrottleStateType): 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>[] = [];
for (let i = 0; i < 4; i++) {
ops.push(throttle.execute(async () => {
await setTimeout(5);
return i;
}));
}
// 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.