@glitchpad/throttler

Async throttle for promise-returning functions. Concurrency cap, priority queue, pause/resume, cancel, flush, deduplication, backpressure strategies, retry with configurable strategy + jitter, and per-task metrics with latency percentiles. Zero runtime deps. RxJS-friendly.

Install

npm install @glitchpad/throttler

Quick start

import { AsyncThrottle, TAsyncThrottleFunction } from '@glitchpad/throttler';

const throttle = new AsyncThrottle({
    maxThreshold: 3,
    delayExecutions: 1000,
});

const task: TAsyncThrottleFunction<[number, number], string> = {
    args: [1, 2000],
    function: async (taskId: number, delay: number) => {
        await new Promise((r) => setTimeout(r, delay));
        return `Task ${taskId} completed`;
    },
};

throttle.addToQueue(task);

Priority queue

Higher numeric priority runs first. Default priority is 0.

throttle.addToQueue(highPriorityTask, 'high-1', 10);
throttle.addToQueue(normalTask);
throttle.addToQueue(lowPriorityTask, 'low-1', -5);

Pause / resume / stop

throttle.pause();
if (throttle.isPaused) {
    // ...
}
throttle.resume();

throttle.clearQueue();
throttle.stop();

Cancel + flush + drain

// Abort everything in flight or queued
throttle.cancelAll('user navigated away');

// Wait until the queue drains; throws if timeout exceeded
await throttle.flush(5000);

// Stop accepting new work and wait for in-flight tasks
await throttle.drain(5000);

Scheduling

// Run at a specific Date
await throttle.scheduleAt(task, new Date(Date.now() + 60_000));

// Run after N ms
await throttle.scheduleIn(task, 2500);

// Batch insertion
await throttle.addMany([taskA, taskB, taskC], 'batch-1', 0);

Backpressure

const throttle = new AsyncThrottle({
    maxThreshold: 4,
    delayExecutions: 100,
    maxQueueSize: 100,
    backpressureStrategy: 'reject', // 'reject' | 'drop-oldest' | 'block'
});

When the queue is at capacity, reject throws QueueFullError, drop-oldest evicts the head, and block is reserved for explicit handling on the caller side.

Deduplication

Use the default hash (function.name + JSON.stringify(args)) or supply your own.

const throttle = new AsyncThrottle({
    maxThreshold: 4,
    delayExecutions: 100,
    enableDeduplication: true,
    deduplicationHashFn: (task) => `${task.function.name}:${task.args[0]}`,
});

A duplicate addToQueue call resolves immediately with success: false and error: Error('Duplicate task').

Retry, strategies, jitter

Configure retry on the throttle and override per task:

const throttle = new AsyncThrottle({
    maxThreshold: 4,
    delayExecutions: 100,
    retry: {
        attempts: 5,
        delay: 500,
        strategy: 'exponential', // 'exponential' | 'linear' | 'fibonacci' | 'fixed'
        jitter: 'full',          // 'full' | 'equal' | 'decorrelated' | 'none'
        maxDelay: 30_000,
    },
});

throttle.addToQueue(task, 'override-1', 0, {
    retryAttempts: 2,
    retryDelay: 250,
    retryStrategy: 'linear',
    jitterStrategy: 'equal',
    maxRetryDelay: 5_000,
});

Per-task timeout

const throttle = new AsyncThrottle({
    maxThreshold: 4,
    delayExecutions: 100,
    defaultTimeout: 10_000,
});

throttle.addToQueue(task, 'with-timeout', 0, { timeout: 2_000 });

A timed-out task throws TimeoutError and emits taskTimeout. Cancelled tasks throw TaskCancelledError and emit taskCancelled.

AbortSignal

const controller = new AbortController();
throttle.addToQueue(task, 'abortable', 0, { signal: controller.signal });

controller.abort('component unmounted');

The throttle also exposes its global signal via throttle.signal and throttle.cancelAll(reason).

Metrics

const throttle = new AsyncThrottle({
    maxThreshold: 4,
    delayExecutions: 100,
    metricsConfig: { buckets: [10, 50, 100, 250, 500, 1000, 5000] },
});

const m = throttle.getMetrics();
// {
//   totalTasks, queued, active, succeeded, failed, cancelled, timedOut,
//   avgLatency, p50, p95, p99, latencyHistogram
// }

throttle.resetMetrics();

Events

throttle.on('result',         (r) => { /* { id, value, success, retries, duration, metadata } */ });
throttle.on('resultError',    (r) => { /* { id, error, success: false, retries, duration, metadata } */ });
throttle.on('taskTimeout',    ({ id, timeout }) => {});
throttle.on('taskCancelled',  ({ id, reason }) => {});
throttle.on('empty',          () => {});
throttle.on('add',            () => {});
throttle.on('pause',          () => {});
throttle.on('resume',         () => {});
throttle.on('stop',           () => {});
throttle.on('clear',          () => {});
throttle.on('log',            (msg) => {});

// Cleanup
throttle.off('result', handler);
throttle.removeAllListeners('result');
throttle.removeAllListeners();

API

Constructor

new AsyncThrottle({
    maxThreshold: number;
    delayExecutions: number;
    maxQueueSize?: number;
    backpressureStrategy?: 'reject' | 'drop-oldest' | 'block';
    defaultTimeout?: number;
    enableDeduplication?: boolean;
    deduplicationHashFn?: (task) => string;
    retry?: {
        attempts: number;
        delay: number;
        strategy?: 'exponential' | 'linear' | 'fibonacci' | 'fixed';
        jitter?: 'full' | 'equal' | 'decorrelated' | 'none';
        maxDelay?: number;
    };
    metricsConfig?: { buckets?: number[] };
});

Methods

• addToQueue(task, id?, priority?, options?)
• addMany(tasks, idPrefix?, priority?, options?)
• scheduleAt(task, date, id?, priority?, options?)
• scheduleIn(task, delayMs, id?, priority?, options?)
• clearQueue()
• flush(timeoutMs?)
• drain(timeoutMs?)
• cancelAll(reason?)
• pause() / resume() / stop()
• getMetrics() / resetMetrics()
• on(evt, fn) / off(evt, fn) / removeAllListeners(evt?)

Errors

• TimeoutError
• TaskCancelledError
• QueueFullError

Properties

• currentStatus — { queueSize, currentlyQueued, maxThreshold }
• isPaused
• signal — global AbortSignal

Error handling

throttle.on('resultError', ({ id, error }) => {
    console.error(`task ${id} failed`, error);
});

try {
    throttle.addToQueue(task);
} catch (err) {
    // Synchronous backpressure rejection lands here
}

Patterns

Rate-limited API calls

import { AsyncThrottle, TAsyncThrottleFunction } from '@glitchpad/throttler';

const api = new AsyncThrottle({
    maxThreshold: 5,
    delayExecutions: 1000,
    retry: { attempts: 3, delay: 500, strategy: 'exponential', jitter: 'full' },
    defaultTimeout: 8_000,
});

const fetchUser = async (id: number) => {
    const res = await fetch(`https://api.example.com/users/${id}`);
    if (!res.ok) throw new Error(`HTTP ${res.status}`);
    return res.json();
};

for (let i = 1; i <= 20; i++) {
    const task: TAsyncThrottleFunction<[number], unknown> = {
        args: [i],
        function: fetchUser,
    };
    api.addToQueue(task, `user-${i}`);
}

api.on('result',      (r) => console.log('ok', r.id, r.duration));
api.on('resultError', (r) => console.error('fail', r.id, r.error));
api.on('empty',       () => api.stop());

Stale-resource pattern

When working with database connections or any resource that may not be ready at enqueue time, close over it inside function, do not pass it through args:

const task: TAsyncThrottleFunction<[], void> = {
    args: [],
    function: async () => {
        const conn = await getConnection();
        await runQuery(conn);
    },
};

This evaluates the resource at execution time, not at enqueue time.

TypeScript

Type definitions ship in the package. .d.ts for ESM consumers and .d.cts for CommonJS consumers.

Runtime

Designed for Node.js 14+. Works in modern browsers that support AbortController and AbortSignal.any (or via polyfill). Bundle with your tool of choice.

License

MIT