@nxtedition/scheduler

@nxtedition/scheduler

A high-performance, priority-based task scheduler for Node.js with support for concurrency limiting, byte-rate throttling, and multi-worker coordination via SharedArrayBuffer.

Install

npm install @nxtedition/scheduler

Usage

Basic

import { Scheduler } from '@nxtedition/scheduler'

const scheduler = new Scheduler({ concurrency: 4 })

const result = await scheduler.run(async () => {
  const res = await fetch('https://example.com')
  return res.json()
})

Priority

Seven priority levels are available:

Name	Value
lowest	-3
lower	-2
low	-1
normal	0
high	1
higher	2
highest	3

// Using string priority
await scheduler.run(() => importantWork(), 'high')

// Using static constants
await scheduler.run(() => backgroundWork(), Scheduler.LOW)

Per-Priority Concurrency

Concurrency can be configured per priority to reserve capacity for higher-priority work. A task at priority p is admitted on the fast path only when total running tasks are below that priority's cap; otherwise the task queues. The overall max is always a hard ceiling.

// At most 100 concurrent tasks total. Background work (low/lowest) is held
// to a small fraction so interactive (normal/high) traffic isn't queued behind
// a flood of batch jobs.
const scheduler = new Scheduler({
  concurrency: { max: 100, low: 20, lowest: 5 },
})

Unspecified priorities inherit the cap from the priority just below them — caps propagate downward, not upward. Priorities above the topmost explicit fall back to max, and an unspecified lowest seeds from the bottommost explicit. The example above resolves to lowest=lower=5, low=20, normal..highest=100. Each per-priority value is itself capped at max.

Starvation prevention

Per-priority caps act as backpressure: when running tasks exceed a priority's cap, new tasks at that priority queue. If higher-priority tasks keep arriving, the capped queue could be starved indefinitely. To prevent that, the dispatch loop's fairness lottery (already used to give lower priorities a turn under uniform concurrency) bypasses per-priority caps when it picks a queued priority — so a fully-capped queue still drains slowly. The overall max is still respected. When the lottery picks an empty tier, its share goes to the highest-priority backlogged queue (within caps), so unused tiers never inflate lower-priority throughput.

In SharedArrayBuffer mode, the second constructor argument carries per-priority limits (the buffer carries max across workers):

const sharedState = Scheduler.makeSharedState(100) // global max=100
const scheduler = new Scheduler(sharedState, {
  concurrency: { low: 20, lowest: 5 }, // per-worker caps
})

Low-Level API

For more control, use acquire / release directly:

scheduler.acquire(
  async (opaque) => {
    try {
      await doWork(opaque)
    } finally {
      scheduler.release()
    }
  },
  Scheduler.NORMAL,
  opaqueData,
)

Error contract: if fn throws synchronously, the scheduler releases the slot itself (the error propagates to the acquire() caller on the fast path, and surfaces as a deferred uncaught exception when thrown from a queued dispatch). Do not also call release() on the synchronous-throw path — a try/finally around synchronous work would double-release and free a slot belonging to some other in-flight task. With an async callback as above the finally is correct: a rejecting async function does not throw synchronously, so the slot is yours to release exactly once.

Multi-Worker Coordination

Share a concurrency limit across worker threads using SharedArrayBuffer:

// Main thread
import { Scheduler } from '@nxtedition/scheduler'
import { Worker } from 'node:worker_threads'

const sharedState = Scheduler.makeSharedState(8)
const worker = new Worker('./worker.js', { workerData: sharedState })

// Worker thread
import { Scheduler } from '@nxtedition/scheduler'
import { workerData } from 'node:worker_threads'

const scheduler = new Scheduler(workerData)
await scheduler.run(() => work())

The global max is a hard cap across all workers — a worker that finds the global counter at the limit queues its task locally and waits (via Atomics.waitAsync on the shared counter). When another worker releases a slot, Atomics.notify wakes the waiter, which retries dispatch.

Atomics.waitAsync does not keep the Node event loop alive. If a worker is fully idle (running === 0) with tasks queued waiting for capacity, you need something else holding the loop open — usually trivial in real applications (HTTP servers, intervals, etc.), but standalone scripts may need a setInterval(() => {}, …) until their work is done.

Child Schedulers (hierarchical limits)

scheduler.child(opts?) creates a child that enforces an additional concurrency limit on top of its parent. A task dispatched through a child runs only when it is admitted by both the child's own per-priority cap and the parent's overall max. This composes the classic "global pool of N, M per worker" shape: a shared parent caps total concurrency across the whole process while each worker uses a local child to cap its own share.

// main.js — one shared global pool of 16
const sharedState = Scheduler.makeSharedState(16)
for (let i = 0; i < 4; i++) new Worker('./worker.js', { workerData: sharedState })

// worker.js — at most 4 concurrent reads on THIS worker, ≤16 across the whole process
const parent = new Scheduler(workerData)
const reads = parent.child({ concurrency: 4 })
const data = await reads.run(() => fs.readFile(path), 'high')

• opts mirrors the constructor ({ concurrency }), so children get their own per-priority caps too: parent.child({ concurrency: { max: 4, low: 1 } }). Omit it for an unlimited local cap (only the parent constrains).
• The child has its own independent priority lottery and starvation prevention; the parent's overall max is the global gate — a soft limit when the parent/root is shared (the same bounded read-then-increment overshoot across workers as a plain shared Scheduler; see the dispose note above). The parent's own per-priority caps constrain only tasks submitted directly to the parent — child tasks see the parent solely as an overall max.
• Children may be nested arbitrarily; the tightest cap in the chain wins, and every level's max is enforced.
• A child is a local (non-shared) scheduler that delegates all global accounting — and, when the root is shared, all cross-thread waiting — up the parent chain; only the root touches the SharedArrayBuffer.
• child[Symbol.dispose]() returns the child's in-flight slots to the parent, rejects its queued run() promises, and unregisters it. Disposing a parent disposes its descendants. Children of a long-lived shared root live until process exit unless disposed — dispose short-lived children explicitly.

UV Thread Pool Scheduling

A practical use case is coordinating access to the libuv thread pool (UV_THREADPOOL_SIZE) across multiple worker threads. For example, several HTTP file-serving workers can share a scheduler so that file-system operations (which consume UV thread pool slots) are prioritized and throttled globally:

// main.js
import { Scheduler } from '@nxtedition/scheduler'
import { Worker } from 'node:worker_threads'

const UV_THREADPOOL_SIZE = parseInt(process.env.UV_THREADPOOL_SIZE || '4', 10)
const sharedState = Scheduler.makeSharedState(UV_THREADPOOL_SIZE)

for (let i = 0; i < 4; i++) {
  new Worker('./http-worker.js', { workerData: { sharedState } })
}

// http-worker.js
import { Scheduler, parsePriority } from '@nxtedition/scheduler'
import { workerData } from 'node:worker_threads'
import fs from 'node:fs/promises'

const scheduler = new Scheduler(workerData.sharedState)

async function handleRequest(req, res) {
  // Derive priority from a request header, e.g. "X-Priority: high"
  const priority = parsePriority(req.headers['x-priority'] || 'normal')

  const data = await scheduler.run(() => fs.readFile(filePath), priority)
  res.end(data)
}

This ensures that high-priority requests get file-system access first, while low-priority background work (thumbnails, transcoding, etc.) yields thread pool capacity without starving entirely — thanks to the built-in starvation prevention.

Monitoring

stats mixes gauges (instantaneous; rise and fall) with counters (cumulative; monotonically increasing). Read gauges for a "right now" snapshot; difference counters across a window (Δcounter / Δt) for rates. A counter can't miss a short burst that filled and drained between two samples, the way a gauge can.

const { running, concurrency, pending, queues } = scheduler.stats

Field	Kind	Meaning
running	gauge	tasks executing right now (this worker)
concurrency	const	configured overall ceiling (Infinity if unlimited)
pending	gauge	tasks queued, waiting for a slot
queues[p].count	gauge	items currently queued at priority p (index = priority + 3)
queues[p].deferred	counter	items ever enqueued at p (couldn't run immediately)
queues[p].completed	counter	items ever dispatched out of p's queue
total	—	{ count, deferred, completed } summed across all priorities
shared.running	gauge	global running across all workers (shared mode only)
shared.waiters	gauge	workers parked waiting for a slot (shared mode only)

While the scheduler is live, deferred - completed === count at every priority. total is a convenience roll-up of the per-priority queues array so callers needn't sum it themselves: total.count === pending, total.completed is overall throughput, and total.deferred - total.completed is the whole-scheduler live backlog.

The Limiter exposes the same per-priority fields and the same total roll-up, plus tokens (gauge — current balance, negative while in token debt) and tokensPerSecond (const), forced (counter — anti-starvation force-throughs into debt; increments only when the limiter is overloaded or an item exceeds the bucket capacity), and queues[p].stall (gauge — consecutive ~100ms ticks a priority's front has been blocked on tokens). stall is per-priority only — total does not aggregate it (a sum across priorities isn't meaningful; take the max over queues for a single "most-stalled" figure).

Pressure & backoff: utilization is the wrong trigger

A naive backoff steers on utilization — running / concurrency — or on pending > 0. But utilization is the wrong question. Linux PSI (Pressure Stall Information; /proc/pressure/{cpu,memory,io}, by Johannes Weiner — the signal behind systemd-oomd and Facebook's oomd/senpai) exists precisely to make this distinction.

PSI's core insight: a resource can be 100% utilized with ~0 pressure. If every slot is busy but work keeps completing and nothing is waiting, that is the healthy operating point — exactly where you must NOT throttle. Pressure is not how busy you are; it is the percentage of wall-clock time that runnable work is stalled, unable to make progress. A saturated scheduler draining its queue at full speed has near-zero pressure; pressure rises only when tasks are delayed.

So keep the USE method for observability — utilization stays a first-class dashboard gauge — but move the backoff decision onto a pressure signal. PSI defines two levels, both of which map cleanly onto this library:

Axis	Question	Role
Utilization (USE)	How full is the resource? (running/concurrency)	Dashboards, capacity planning — never the trigger
Pressure some (PSI)	Is at least one task delayed?	Primary, sensitive backoff — shed discretionary
Pressure full (PSI)	Is nothing making progress?	Second, harder tier — pause the producer

PSI reports each level as decaying averages over 10s/60s/300s (EWMAs, the loadavg shape) plus a monotonic total=<µs> accumulator you diff for exact stall time over any window — immune to the aliasing a gauge suffers when a burst fills and drains between two samples. Controllers like oomd then act on a threshold held for a sustained duration, a time-based hysteresis.

Borrow the idea, not the kernel machinery (the kernel integrates stall time at task state-transitions; sampling stats on a loop is a coarse approximation). The library bakes in no timer; you already poll stats (health check, metrics scrape), so reconstruct the signal there: each tick, decide one boolean per level — is the resource stalled right now? — and feed it through an EWMA. some requires a backlog and no free capacity to absorb it, so a busy-but-draining scheduler (pending returns to 0) scores 0 — the false positive the naive running/concurrency recipe fires on is eliminated by construction. full additionally requires zero forward progress this window (Δcompleted === 0), or that the limiter is in token debt.

Two independent defenses give the "sensitive enough to catch real overload, not so twitchy it flaps and kills throughput" property: (1) the predicate filters healthy saturation before it enters the signal; (2) the EWMA window plus a Schmitt-trigger dead-band (engage high, release low) is the oomd-style sustained-duration requirement — a one-tick blip can't trip backoff, a brief recovery can't prematurely release it. The two-tier some→shed / full→pause split then lets you respond proportionally.

// Reconstruct a PSI-style pressure signal from Scheduler/Limiter `stats` on a
// loop you ALREADY run. The library bakes in no timer; we add none to the hot
// path. Each tick yields one "stalled right now?" bool per level, smoothed into
// loadavg-shaped EWMAs. some -> shed discretionary work; full -> pause producer.

const SAMPLE_MS = 200
const TAU_MS = 10_000 // PSI avg10 analogue: the smoothing/desensitizing window
const STALL_TICKS = 10 // limiter's own ~1s force-through threshold (10 x ~100ms)

// Schmitt-trigger dead-bands (engage high, release low) = time-based hysteresis.
const SOME_HI = 0.5,
  SOME_LO = 0.2 // shed discretionary work
const FULL_HI = 0.3,
  FULL_LO = 0.1 // pause the producer

let some = 0,
  full = 0 // EWMA state (fraction of recent time stalled)
let shed = false,
  paused = false
let prev = null // { completed, forced } for monotonic-counter diffs

const sumCompleted = (queues) => {
  let s = 0
  for (const q of queues) s += q.completed
  return s
}

// One sample -> instantaneous "stalled?" per resource. -----------------------
function sampleScheduler(s) {
  // Infinity concurrency has no slot to stall on: never pressured, no NaN.
  const saturated = s.concurrency !== Infinity && s.running >= s.concurrency
  return { someNow: s.pending > 0 && saturated, completed: sumCompleted(s.queues), forced: 0 }
}

function sampleLimiter(l) {
  // q.stall = consecutive ~100ms ticks a priority's front sat blocked on tokens
  // (real elapsed time, traffic-independent). >= STALL_TICKS == ~1s delayed.
  const frontStalled = l.queues.some((q) => q.count > 0 && q.stall >= STALL_TICKS)
  // tokens < 0 = debt: an anti-starvation force-through is being repaid, so
  // other work is held up -> a genuine stall.
  return {
    someNow: frontStalled || l.tokens < 0,
    completed: sumCompleted(l.queues),
    forced: l.forced,
  }
}

function update(stats) {
  const isLimiter = 'tokens' in stats
  const v = isLimiter ? sampleLimiter(stats) : sampleScheduler(stats)

  // FULL needs the windowed COUNTER diff, not a gauge: a standing stall with
  // ZERO completions this window means nobody advanced -> all non-idle work is
  // simultaneously stalled. For the limiter, debt or a fresh force-through
  // corroborates it (so healthy back-to-back draining can't read as full).
  let fullNow = false
  if (prev) {
    const progressed = v.completed - prev.completed > 0
    fullNow = isLimiter
      ? v.someNow && !progressed && (stats.tokens < 0 || v.forced - prev.forced > 0)
      : v.someNow && !progressed
  }

  // dt-aware EWMA gain keeps the time-constant honest under a jittery loop.
  const a = 1 - Math.exp(-SAMPLE_MS / TAU_MS)
  some += a * ((v.someNow ? 1 : 0) - some)
  full += a * ((fullNow ? 1 : 0) - full)
  prev = { completed: v.completed, forced: v.forced }

  // Hysteresis: engage high, release low.
  if (!shed && some > SOME_HI) shed = true
  else if (shed && some < SOME_LO) shed = false
  if (!paused && full > FULL_HI) paused = true
  else if (paused && full < FULL_LO) paused = false
}

// The loop is YOURS; just .unref() so the sampler can't keep the process alive.
setInterval(() => update(scheduler.stats), SAMPLE_MS).unref()

// Producers consult the latched flags (cheap; no per-task stats read):
function submit(task, priority = 'low') {
  if (paused) return // hard stop: nothing new
  if (shed && priority === 'low') return // discretionary work: first to go
  return scheduler.run(task, priority)
}

Limiter-side pressure. The limiter stalls on tokens (a rate resource), not slots, and surfaces the most directly PSI-shaped field in the API: queues[p].stall — the count of consecutive ~100ms ticks a priority's front has sat blocked on tokens. Because it ticks on the real-time refill clock independent of traffic volume, it is a measured wall-clock stall duration, and stall >= STALL_TICKS (10, ≈1s) is the very threshold the engine uses to force-admit a starved item — reusing it keeps your signal in lockstep with the engine's own notion of "stuck." Map it like the scheduler: some = any priority with count > 0 && stall >= STALL_TICKS, or tokens < 0 (debt). For full, also require zero completions this window and (tokens < 0 or Δforced > 0). Treat forced as a counter — diff it (Δforced > 0 means the anti-starvation valve fired this interval); never gate on its raw cumulative value. Pass limiter.stats through the same update() — the 'tokens' in stats probe routes it to the limiter sampler. (If you pipe data through limiter.stream(), byte backpressure is already applied for you via the stream's highWaterMark.) In shared mode, forced/stall are per-worker; OR-in shared.waiters > 0 for a cross-worker some and aggregate in your metrics backend rather than trusting one worker.

API

new Scheduler(opts, options?)

• opts.concurrency — concurrency configuration (default: Infinity). Either a number (the overall max) or an object with max and per-priority caps:

  type ConcurrencyOptions =
    | number
    | {
        max?: number
        highest?: number
        higher?: number
        high?: number
        normal?: number
        low?: number
        lower?: number
        lowest?: number
      }
  

• opts may also be a SharedArrayBuffer created by Scheduler.makeSharedState(). In that case, an optional second argument { concurrency } may carry per-priority caps (the buffer's max is always the global ceiling).

scheduler.run(fn, priority?, opaque?): Promise<T>

Execute fn within the scheduler. Returns a promise that resolves with the return value of fn (thenables are assimilated via Promise.resolve). After the scheduler has been disposed, run() rejects with Error('Scheduler is disposed') instead of hanging; with an overall concurrency of 0 (which could never dispatch anything) it rejects with Error('Scheduler concurrency is 0').

scheduler.acquire(fn, priority?, opaque?): boolean

Low-level task acquisition. Returns true if a slot was available and fn ran synchronously, or false if the task was queued (also false if the scheduler has been disposed). The fast path additionally requires that nothing is queued at the same priority, preserving FIFO order within a priority; a fresh task can still start ahead of work queued at a different priority while a freed slot is in flight (a deliberate latency trade-off, bounded to a microtask in shared mode). You must call scheduler.release() when fn completes — except when fn throws synchronously, in which case the scheduler has already released the slot (see the Low-Level API example).

scheduler.tryAcquire(priority?): boolean

Non-blocking. Takes a slot and returns true only if one is free at the given priority and nothing is queued at that priority (mirroring limiter.tryConsume); otherwise returns false without queuing. A successful tryAcquire must be paired with release().

scheduler.release(): void

Signal task completion. Dequeues the next pending task if concurrency allows.

scheduler.concurrency

The overall concurrency ceiling (Infinity when unlimited). In shared mode this is the buffer's global max; a per-instance options.concurrency.max is enforced but not reflected here.

scheduler.stats

{ running, concurrency, pending, total, queues, shared } — locally running tasks, the configured ceiling, locally queued tasks, a total roll-up of { count, deferred, completed } summed across all priorities, a per-priority array of { count, deferred, completed } (current depth gauge + cumulative enqueued/dispatched counters), and (shared mode only) the global { running, waiters } gauges. See Monitoring for the gauge-vs-counter distinction and a backoff recipe. Allocates fresh objects on every read; poll accordingly.

scheduler.child(opts?): Scheduler

Create a child scheduler that enforces an additional concurrency limit on top of this one (see Child Schedulers). opts mirrors the constructor ({ concurrency }, a number or per-priority object); omit it for an unlimited local cap. A task runs only when admitted by both the child's cap and the parent's overall max. Returns a fully-featured Scheduler (run / acquire / tryAcquire / release / stats / Symbol.dispose). Throws if the parent is disposed.

Scheduler.makeSharedState(concurrency): SharedArrayBuffer

Create shared state for cross-worker scheduling. concurrency must be Infinity or a non-negative integer that fits in a signed 32-bit integer (≤ 2,147,483,647) — the counter is stored as Int32.

scheduler[Symbol.dispose]()

Disposes the scheduler: releases any global slots it still holds (shared mode), rejects queued run() promises with Error('Scheduler is disposed'), drops the reference to a pending shared-mode Atomics.waitAsync, and makes all further calls inert (acquire / tryAcquire return false; run rejects). Use when abandoning a shared-mode Scheduler with tasks still queued so the instance and its parked wait can be garbage-collected (it is otherwise kept alive until process exit so its held slots can be released). Works with using. Note that slots held by still-in-flight tasks are returned to the shared pool immediately — until those tasks actually finish, other workers can transiently admit more than max (the limit is soft by design). Disposing a scheduler also disposes its children; disposing a child returns its in-flight slots to the parent and unregisters it.

parsePriority(value): number

Parse a string or number into a normalized priority value.

Limiter

Limiter is a token-bucket rate limiter that controls how many tokens per second are processed — typically bytes, but any integer unit works. It shares the same priority system as Scheduler.

Previously exported as Throttle. The Throttle alias is retained for backward compatibility but is deprecated — prefer Limiter.

Tokens refill automatically: the constructor starts an internal timer, and refills are also triggered lazily from consume() / tryConsume(). There is no manual refill step. While work is queued the timer holds the event loop open (so a script never exits with queued callbacks silently dropped); an idle limiter does not keep the process alive.

Basic

import { Limiter } from '@nxtedition/scheduler'

// Allow 1 MB/s (tokens are bytes here)
const limiter = new Limiter({ tokensPerSecond: 1_000_000 })

// Consume `chunk.length` tokens at normal priority, running the callback when available
limiter.consume(() => send(chunk), chunk.length)

Streaming

limiter.stream() returns a Node.js Transform stream that enforces backpressure — it won't pass a chunk through until enough tokens are available. Each chunk consumes chunk.length tokens:

import { createReadStream, createWriteStream } from 'node:fs'
import { pipeline } from 'node:stream/promises'

const limiter = new Limiter({ tokensPerSecond: 1_000_000 }) // 1 MB/s

await pipeline(createReadStream('input.mp4'), limiter.stream(), createWriteStream('output.mp4'))

Priority

Both consume() and stream() accept a priority. Higher-priority work drains first when tokens are available:

// Low-priority stream (yields to higher-priority work)
const stream = limiter.stream('low')

Multi-Worker Coordination

Share a token bucket across worker threads using SharedArrayBuffer, exactly like Scheduler:

// Main thread
import { Limiter } from '@nxtedition/scheduler'
import { Worker } from 'node:worker_threads'

const sharedState = Limiter.makeSharedState(1_000_000) // 1 MB/s shared across all workers
new Worker('./worker.js', { workerData: sharedState })

// Worker thread
import { Limiter } from '@nxtedition/scheduler'
import { workerData } from 'node:worker_threads'

const limiter = new Limiter(workerData)
limiter.consume(() => send(chunk), chunk.length)

Low-Level API

const ran = limiter.consume(
  () => {
    sendPacket(data)
  },
  data.byteLength,
  'normal',
)

consume returns true if the callback ran immediately (tokens were available and nothing was queued ahead of it — the call refills the bucket from elapsed time first, so an idle or freshly created limiter answers true when the credit covers the request), or false if it was queued. Note the boundary is not airtight: a false (queued) task can still run synchronously within the same call when the drain the call triggers reaches it immediately.

Child Limiters (hierarchical rates)

limiter.child(opts) creates a child token bucket that enforces an additional rate on top of its parent. A consume of N tokens through the child runs only when N tokens are available in both the child's own bucket and the parent chain's bucket(s), and the N tokens are debited from every level — exactly mirroring Scheduler.child(), for byte-rate instead of concurrency. The canonical shape is a shared parent capping the aggregate rate while each worker caps its own share:

// main.js — 16 MB/s shared across the whole process
const sharedState = Limiter.makeSharedState(16_000_000)
for (let i = 0; i < 4; i++) new Worker('./worker.js', { workerData: sharedState })

// worker.js — at most 4 MB/s on THIS worker, ≤16 MB/s across the process
const parent = new Limiter(workerData)
const out = parent.child({ tokensPerSecond: 4_000_000 })
out.consume(() => send(chunk), chunk.length, 'high')

Priorities, stream(), and the ~1s anti-starvation force-through all work as usual and apply per level (a forced item is charged into debt on the whole chain, so each level's long-run rate still holds). The child has its own refill timer; the parent needs no knowledge of its children. Children may be nested.

API

new Limiter(opts)

• opts.tokensPerSecond — tokens (e.g. bytes) per second. Required; must be a positive integer that fits in a signed 32-bit integer (1 … 2,147,483,647), since tokens are tracked as Int32. 0, negative, non-integer, NaN, Infinity, and values above 2,147,483,647 all throw.
• opts may also be a SharedArrayBuffer created by Limiter.makeSharedState(tokensPerSecond).

limiter.consume(fn, bytes, priority?, opaque?): boolean

Consume bytes tokens, running fn(opaque) when they are available. Returns true if fn ran immediately, false if it was queued. opaque is passed through to fn on both paths.

bytes must be a non-negative integer no larger than 2,147,483,647 (the signed Int32 max — tokens are tracked as Int32); negative, non-integer, NaN, Infinity, and out-of-range values throw (a negative or wrapping value would otherwise add tokens via Atomics.sub and corrupt the shared bucket). A bytes of 0 consumes no tokens and never waits on the bucket. It runs immediately, returning true, only when nothing is queued at all (pending === 0); otherwise consume returns false, but once it reaches the front of its priority queue it drains regardless of the token balance — even while the bucket is empty or in debt. Queued behind token-starved work at the same priority it waits its turn like any other task (FIFO head-of-line; a different, unblocked priority queue can drain it sooner).

Starvation guard: a queued task larger than the instantaneous token balance could otherwise starve under sustained traffic — smaller tasks would keep spending each refill's credit first. Any queue whose front item makes no progress for ~1 second is force-admitted into token debt; the negative balance is repaid by subsequent refills before anything else drains, so the long-run rate still holds (the trade-off is a transient burst, at most one stalled item per queue per second). The same mechanism bounds the wait for a request larger than tokensPerSecond (the bucket's capacity, which it could never fit): it is admitted once the bucket fills or the ~1s stall guard fires, whichever comes first — i.e. it is throttled, not dropped or stalled forever. In shared mode the guard is per-instance best-effort: other workers may keep consuming the shared bucket.

limiter.tryConsume(bytes): boolean

Non-blocking. Consumes bytes and returns true only if enough tokens are available right now and nothing is queued; otherwise returns false without queuing.

limiter.tokensPerSecond

The configured rate (and bucket capacity).

limiter.stats

{ tokens, tokensPerSecond, pending, forced, total, queues } — current token balance (negative while in debt), the configured rate, queued tasks, the cumulative count of anti-starvation force-throughs, a total roll-up of { count, deferred, completed } summed across all priorities, and a per-priority array of { count, deferred, completed, stall }. See Monitoring for what each field means and how to use them for backoff. Allocates fresh objects on every read.

limiter.stream(priority?, options?): Transform

Returns a Transform stream that rate-limits data passing through it. Each chunk consumes chunk.length tokens; a chunk with no non-negative-integer length (e.g. in objectMode) is treated as weightless (0 tokens) — it never waits on tokens, though it still preserves the limiter's queue ordering (if other work is already queued, it drains in turn rather than jumping ahead). options is forwarded to the underlying Transform (e.g. objectMode, highWaterMark). Destroying the stream while a chunk is still queued in the limiter does not push into the dead stream.

limiter.child(opts): Limiter

Create a child limiter that enforces an additional rate on top of this one (see Child Limiters). opts mirrors the constructor ({ tokensPerSecond }). A consume runs only when its tokens fit both the child's bucket and the parent chain's, and is debited from both. Returns a fully-featured Limiter (consume / tryConsume / stream / stats).

Limiter.makeSharedState(tokensPerSecond): SharedArrayBuffer

Create shared state for cross-worker rate limiting.

Performance

Microbenchmarks from bench.js on an AMD EPYC (linux-x64, Node 26). Mops/s = millions of ops/sec; higher is better. Numbers are indicative and machine-dependent — run node --expose-gc bench.js to reproduce.

Scheduler hot paths (single thread)

Operation	Throughput
acquire+release (unlimited, no queue)	~110 Mops/s
tryAcquire fast path	~150 Mops/s
acquire+release (SharedArrayBuffer)	~38 Mops/s
queue drain (concurrency=1, batch)	~37 Mops/s
run() (sync body, Promise per task)	~4 Mops/s

child() overhead (single thread)

A child adds a local per-priority gate plus one parent-chain acquire/release per task:

Operation	Throughput
acquire+release (parent only)	~71 Mops/s
acquire+release (through 1 child)	~36 Mops/s
acquire+release (through 2 nested children)	~24 Mops/s
child queue drain (cap=1, batch)	~20 Mops/s

With vs without per-worker child() (8 workers, shared global cap=8)

The overhead that matters in practice is end-to-end, where real per-task work dominates the extra bookkeeping. With realistic ~50µs tasks a per-worker child() costs only ~4%:

Workload (8 workers)	Plain	With child(4)	Overhead
0µs tasks (pure scheduler overhead)	~1.45 Mops/s	~1.38 Mops/s	~5%
50µs tasks (realistic)	~132 Kops/s	~127 Kops/s	~4%

An earlier revision sharded the global RUNNING/WAITERS counters across cache lines to cut contention. Benchmarking on the x64 target showed it was a net regression (~25% slower at 32 workers): RUNNING is read on every acquire, so aggregating shards costs more coherency traffic than the single-line write contention it avoids. It was reverted to a single counter.

License

MIT