isotropic-throttle

npm version License

A sliding-window rate limiter that makes asynchronous code wait until the appropriate time to continue.

Why Use This?

• Rate Limiting: Cap how many operations run within any rolling window of time
• Promise Based: await a single method and continue when a slot is available
• Sliding Window: Old activity expires continuously rather than resetting on fixed boundaries
• Fair Ordering: Queued waits resolve in the order they were requested
• Cancelable: Abort a pending wait with an AbortSignal, a timeout, or the promise's own cancel() method
• Synchronous Option: tryWait() consumes a slot immediately or reports that none is free, without queueing
• External Accounting: mark() records consumption that happened outside the throttle so local and remote limits stay in sync
• Clean Shutdown: Disposable via using, releasing its timer and settling any pending waits
• Minimal Dependencies: Builds on isotropic-error, isotropic-later, and isotropic-make

Installation

npm install isotropic-throttle

How It Works

A throttle allows at most count operations to proceed during any duration-millisecond window. Each time a slot is consumed the current time is recorded. When a new wait() would exceed count within the window, it is queued and resolves only once enough time has passed for the oldest recorded timestamp to fall outside the window.

The window slides: timestamps expire individually as time passes, so a steady stream of work settles into a smooth rate rather than running in bursts at fixed reset points. Timing is measured with a monotonic clock, so the throttle is unaffected by changes to the system clock.

Usage

import _throttle from 'isotropic-throttle';

{
    // Allow at most 5 operations per second.
    const throttle = _throttle({
        count: 5,
        duration: 1000
    });

    // Somewhere in an async function:
    await throttle.wait();

    // Fewer than 5 operations have run in the last second, so this runs
    // right away. Once the limit is reached, wait() resolves only after the
    // window clears.
    doRateLimitedWork();
}

Limiting a Stream of Work

import _throttle from 'isotropic-throttle';

const _processAll = async items => {
    const throttle = _throttle({
        count: 10,
        duration: 1000
    });

    // Launch everything at once; the throttle paces the actual work to no
    // more than 10 starts per second.
    await Promise.all(items.map(async item => {
        await throttle.wait();

        return _process(item);
    }));
};

Canceling a Wait

A pending wait() can be canceled three ways: an AbortSignal passed to wait(), a timeout, or the cancel() method on the returned promise. A canceled wait rejects rather than resolving, and does not consume a slot.

import _throttle from 'isotropic-throttle';

{
    const throttle = _throttle({
        count: 1,
        duration: 1000
    });

    // Give up if a slot does not open within 500ms.
    try {
        await throttle.wait({
            timeout: 500
        });

        doRateLimitedWork();
    } catch (error) {
        // error.name === 'TimeoutError'
    }
}

import _throttle from 'isotropic-throttle';

{
    const controller = new AbortController(),
        throttle = _throttle({
            count: 1,
            duration: 1000
        });

    const promise = throttle.wait({
        signal: controller.signal
    });

    // Later, abort the wait from anywhere that holds the controller.
    controller.abort();

    try {
        await promise;
    } catch (error) {
        // error.name === 'AbortError'
    }
}

Trying Without Waiting

tryWait() consumes a slot and returns true if one is available right now, or returns false without queueing if the throttle is at its limit.

import _throttle from 'isotropic-throttle';

{
    const throttle = _throttle({
        count: 5,
        duration: 1000
    });

    if (throttle.tryWait()) {
        doRateLimitedWork();
    } else {
        // At the limit; skip this one rather than waiting.
    }
}

Recording External Activity

wait() records its own slot when it resolves. mark() records a slot that was consumed somewhere else, without waiting. Its intended use is to track usage that you are not the one consuming.

Consider a websocket with a strict limit of 50 messages per minute covering both directions. Outbound messages can wait for a slot before being sent, but inbound messages arrive on their own schedule and must simply be counted:

import _throttle from 'isotropic-throttle';

{
    const throttle = _throttle({
        count: 50,
        duration: 60000
    });

    socket.addEventListener('message', () => {
        // An inbound message just used part of the shared budget. Count it so
        // outbound waits stay within the real limit. mark() returns the
        // throttle, so calls can be chained.
        throttle.mark();
    });

    const send = async message => {
        await throttle.wait();

        socket.send(message);
    };
}

Note: wait() already records a slot for the operation it permits. Do not also call mark() for that same operation, or it will be counted twice. Use mark() only for activity that did not go through wait().

Releasing a Throttle

A throttle that has queued waits keeps an internal timer running. Disposing the throttle releases that timer and rejects any pending waits with a DisposedError. The using declaration disposes automatically at the end of the scope:

import _throttle from 'isotropic-throttle';

{
    using throttle = _throttle({
        count: 5,
        duration: 1000
    });

    await throttle.wait();

    doRateLimitedWork();

    // throttle is disposed automatically here.
}

API

throttle(options)

Creates a throttle instance. Invalid options throw an error whose name is 'ArgumentError'.

Parameters

• options (Object, optional): Configuration object.
  • count (Number, optional): The maximum number of slots that may be consumed within a window. Must be a positive integer. Defaults to 1.
  • duration (Number, optional): The length of the sliding window in milliseconds. Must be a non-negative number. Defaults to 1000.
  • durationBuffer (Number, optional): A small amount of time, in milliseconds, added to each computed wake-up delay. Defaults to 13.

Returns

• (Object): A throttle instance.

Why durationBuffer?

A remote service may count or time your activity slightly differently than you do. If a limit is documented as 100 requests per second and you use that allotment exactly, small differences in measurement can push you over the remote service's count and trigger a penalty. durationBuffer is a conservative undershoot: it makes each retry wait a little longer than the bare minimum, so the throttle stays comfortably inside the real limit. It also serves as the minimum delay between retries, smoothing out timer imprecision near the window boundary.

throttle.wait(options)

Waits until a slot is available within the current window.

Parameters

• options (Object, optional): Configuration object.
  • signal (AbortSignal, optional): Cancels the wait if the signal is or becomes aborted.
  • timeout (Number | Temporal.Duration, optional): Cancels the wait if a slot does not become available within this amount of time. A number is interpreted as milliseconds.

Returns

• (Promise): A promise that resolves with the throttle instance once a slot is available, recording a slot at that moment. If the throttle is under its limit, the promise resolves immediately; otherwise it resolves once the window has cleared enough capacity. The promise has an additional cancel method (see below). If the wait is canceled by signal, by timeout, or by cancel() the promise rejects instead of resolving and no slot is consumed. Rejection reasons are isotropic-error instances whose name is one of:
  • 'AbortError' - an AbortSignal aborted the wait.
  • 'CanceledError' - cancel() was called with no reason.
  • 'DisposedError' - the throttle was disposed while the wait was pending.
  • 'TimeoutError' - the timeout elapsed first.

promise.cancel(options)

Cancels a pending wait. Available on the promise returned by wait(). Once the wait has settled (resolved or rejected), this is a no-op; it is also a no-op on a wait that resolved immediately.

Parameters

• options (Object, optional): Configuration object.
  • reason (Error, optional): the wait rejects with that error. If omitted, the wait rejects with a CanceledError.
  • signal (AbortSignal, optional): if already aborted, the wait rejects with an AbortError; otherwise the signal is registered so the wait is canceled if and when it aborts. A signal may be supplied to wait(), to cancel(), or to both.

throttle.tryWait()

Attempts to consume a slot synchronously, without waiting or queueing.

Returns

• (Boolean): true if a slot was available and has been recorded; false if the throttle is at its limit or already has waits queued.

throttle.mark()

Records that a slot was consumed at the current time, without waiting, and prunes expired timestamps. Use this to account for activity that happened outside of wait() so the throttle's view of recent activity stays accurate.

Returns

• (Object): The throttle instance, for chaining.

throttle.ref()

References the throttle's internal timer so that a pending wait keeps the Node.js event loop alive. This is the default behavior.

Returns

• (Object): The throttle instance, for chaining.

throttle.unref()

Unreferences the throttle's internal timer so that a pending wait will not, on its own, keep the Node.js event loop alive.

Returns

• (Object): The throttle instance, for chaining.

throttle.hasRef()

Returns whether the throttle's internal timer is referenced.

Returns

• (Boolean): The throttle's ref status.

throttle.cancelAll(options)

Cancels all pending waits.

Parameters

• options (Object, optional): Configuration object.
  • reason (Error, optional): the waits reject with that error. If omitted, the waits reject with a CanceledError.
  • signal (AbortSignal, optional): if already aborted, the waits reject with an AbortError; otherwise the signal is registered so the waits are canceled if and when it aborts.

Examples

Throttling Outbound API Requests

import _throttle from 'isotropic-throttle';

const _createClient = () => {
    // Stay under a documented limit of 20 requests per second.
    const throttle = _throttle({
        count: 20,
        duration: 1000
    });

    return {
        async request (url) {
            await throttle.wait();

            return fetch(url);
        }
    };
};

{
    const client = _createClient();

    // These resolve as fast as the rate limit allows, in request order.
    const responses = await Promise.all([
        'https://api.example.com/a',
        'https://api.example.com/b',
        'https://api.example.com/c'
    ].map(url => client.request(url)));
}

Giving Up Under Load

import _throttle from 'isotropic-throttle';

const _createClient = () => {
    const throttle = _throttle({
        count: 20,
        duration: 1000
    });

    return {
        async request (url, {
            signal
        } = {}) {
            // Abandon the request if it is still waiting for a slot when the
            // caller's signal aborts, or after two seconds, whichever is first.
            await throttle.wait({
                signal,
                timeout: 2000
            });

            return fetch(url, {
                signal
            });
        }
    };
};

Smoothing High-Frequency Events

import _throttle from 'isotropic-throttle';

{
    const throttle = _throttle({
        count: 1,
        duration: 200
    });

    window.addEventListener('resize', async () => {
        await throttle.wait();

        // Runs at most once every 200ms, but the latest resize is never lost
        // because waits resolve in order.
        recomputeLayout();
    });
}

Keeping a Local Throttle in Sync With a Shared Limit

import _throttle from 'isotropic-throttle';

const _createSocketClient = socket => {
    // A single budget of 60 messages per minute covers both directions.
    const throttle = _throttle({
        count: 60,
        duration: 60000
    });

    // Inbound messages are not initiated here, but they still consume the
    // shared budget, so record each one.
    socket.addEventListener('message', () => {
        throttle.mark();
    });

    return {
        async send (message) {
            // Outbound messages wait their turn within whatever capacity the
            // inbound traffic has left.
            await throttle.wait();

            socket.send(message);
        }
    };
};

Contributing

Please refer to CONTRIBUTING.md for contribution guidelines.

Issues

If you encounter any issues, please file them at https://github.com/ibi-group/isotropic-throttle/issues