Stream Throttle

b.streamThrottle.create({ bytesPerSec, burstBytes }) returns a token bucket that hands out transform() instances; every transform consumes from the same shared bucket. Operators wiring bulk-transfer daemons (object-storage fan-out, log shippers, replication readers) compose a single throttle and apply it to every concurrent transfer — N parallel transforms share the bytesPerSec budget rather than each getting their own.

Algorithm:

- Bucket holds up to burstBytes tokens (default = bytesPerSec, i.e. one second of headroom). Tokens refill at bytesPerSec bytes per second, capped at burstBytes. Refill is computed lazily on every chunk write so there is no per-throttle timer. - On each chunk, the transform asks the bucket for the chunk's byte count. If enough tokens are available, the chunk passes immediately and the tokens are decremented. If not, the transform sleeps for ceil((bytes - tokens) / bytesPerSec * 1000) ms and then retries — the chunk is forwarded as-is once the debt is paid.

Composes with:

- node:stream.pipeline(src, throttle.transform(), dst) — the transform is a regular stream.Transform, so backpressure flows in both directions without operator wiring. - b.appShutdown — the throttle has no background timer; once every transform finishes its _transform, the bucket is garbage-collected with the surrounding daemon.

Refusal posture:

- bytesPerSec <= 0 / non-finite throws stream-throttle/bad-rate. - burstBytes < bytesPerSec throws stream-throttle/bad-burst (smaller burst than refill rate would stall on a single full-rate chunk forever). - Chunks larger than burstBytes would never fit in the bucket; transform({ allowOversize: true }) opts into splitting them across multiple wait windows. Default refuses with a typed error so operators catch this at config time.

RFC + reference:

- [RFC 2697 srTCM](https://www.rfc-editor.org/rfc/rfc2697.html) — single-rate three-color marker, the canonical token-bucket shape this primitive implements (single PIR + CBS, no committed burst tier). - [Wikipedia: Token bucket](https://en.wikipedia.org/wiki/Token_bucket).

b.streamThrottle.create(opts) #

{
    bytesPerSec:  number,    // refill rate (bytes per second; required, > 0)
    burstBytes:   number,    // bucket capacity (default = bytesPerSec)

  `transform(tOpts)` opts:
    allowOversize: boolean,  // permit chunks larger than burstBytes (default false)
    maxWaitMs:     number,   // per-chunk wait ceiling — when set, any
                             //   computed wait > maxWaitMs refuses the chunk
                             //   with `stream-throttle/wait-exceeds-max`
                             //   instead of silently pinning the pipeline.
}

Create a shared token bucket. Returns { transform(opts?), state() }. transform(tOpts?) returns a stream.Transform that consumes from the shared bucket; multiple transforms returned from the same bucket share the rate budget. state() returns { bytesPerSec, burstBytes, tokens, lastRefillMs } for observation.

Refill resilience: _refill clamps elapsed-since-last-refill to the "empty-to-full" duration (burstBytes / bytesPerSec seconds) so an NTP clock step or VM resume can't credit hours of pent-up tokens into the bucket in a single call.

var throttle = b.streamThrottle.create({ bytesPerSec: 5 * 1024 * 1024 });
await new Promise(function (resolve, reject) {
  require("node:stream").pipeline(src, throttle.transform(), dst,
    function (e) { return e ? reject(e) : resolve(); });
});

Last updated 2026-08-08T16:39:49.652Z by seeder.