@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()
})

Soft-Target Semantics

Scheduler concurrency values and Limiter rates are backpressure targets, not hard correctness or safety boundaries. Do not use them as the only protection for a resource that must never exceed an exact quota.

• In shared Scheduler mode, admission reads RUNNING and then increments it atomically. Multiple workers can pass the same stale check, so shared.running may briefly exceed max; the race excursion is bounded by the contenders admitted in that window and settles as they release. Disposing an instance also returns its slots before its abandoned in-flight work necessarily finishes, another intentional source of temporary overshoot.
• Scheduler admission is not a mutex, work-ownership claim, or deduplication primitive. Separate workers may therefore perform logically equivalent work during a race (for example, two cache fills). Use it where duplicate execution is harmless or produces the same observable result. Protect non-idempotent side effects with an external transactional lock, single-flight/deduplication layer, or another ownership mechanism.
• Per-priority values are admission targets. The fairness lottery deliberately bypasses them to prevent starvation, so a priority may run above its target, up to the overall admission target on a serialized dispatcher (plus shared-mode race drift).
• A shared Limiter can briefly overdraft or overfill because its check/debit and refill updates are separate atomic operations. Its starvation guard also deliberately force-admits stalled work into debt. Repayment pulls later throughput back toward tokensPerSecond, but no arbitrary time window is guaranteed to match that rate exactly. If an extreme admission race would exceed the signed Int32 debt range, the balance saturates at its minimum instead of wrapping into positive credit; this deliberately favors conservative backpressure over exact debt accounting.

Use external transactional coordination when crossing a concurrency/rate boundary or executing the same logical operation twice would be unsafe. Use this package's counters, queues, debt, and stall duration to detect sustained pressure.

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 normally admitted on the fast path only when total running tasks are below that priority's target; otherwise the task queues. max is the overall soft admission target.

// Target 100 concurrent tasks total. Background work (low/lowest) normally
// targets a small fraction so interactive (normal/high) traffic is less likely
// to queue behind a flood of batch jobs.
const scheduler = new Scheduler({
  concurrency: { max: 100, low: 20, lowest: 5 },
})

Unspecified priorities inherit the target from the priority just below them — targets 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 configured value is clamped to max.

Starvation prevention

Per-priority targets act as backpressure: when running tasks reach a priority's target, new tasks at that priority normally queue. If higher-priority tasks keep arriving, that queue could be starved indefinitely. To prevent that, the dispatch loop's fairness lottery bypasses per-priority targets when it picks a queued priority, so a fully pressured queue still drains slowly. The serialized drain stops admitting at the observed overall target; shared contenders can still produce the bounded race overshoot described above. When the lottery picks an empty tier, its share goes to the highest-priority backlogged queue, so unused tiers never inflate lower-priority throughput.

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

const sharedState = Scheduler.makeSharedState(100) // shared soft target=100
const scheduler = new Scheduler(sharedState, {
  concurrency: { low: 20, lowest: 5 }, // per-instance priority targets
})

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 target 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 shared max is a soft admission target across workers. A worker that observes the counter at or above the target queues locally and waits (via Atomics.waitAsync); contenders that observed capacity at the same time may still increment past it. 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.

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.

Configured concurrency/rate values are references for pressure thresholds, not invariants. Transient shared.running > concurrency, a priority running beyond its target, or Limiter token debt/overfill can be expected. Alert or back off on sustained queueing, stall duration, debt, and counter rates—not on a single target crossing.

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

Field	Kind	Meaning
running	gauge	tasks executing right now (this worker)
concurrency	const	configured overall target (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 the cumulative number of formerly-deferred items dispatched from queues, and total.deferred - total.completed is the whole-scheduler live backlog. Fast-path work never enters a queue, so it is not included in completed.

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, or that the limiter is in token debt. Scheduler's queues[p].completed only counts work dispatched from a queue; fast-path work never enters those counters. A Scheduler controller must therefore use a monotonic application completion counter (incremented whenever scheduled work settles) for its full decision.

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

// Effective Scheduler targets in stats.queues order (lowest -> highest). Derive
// this once from the SAME constructor configuration; do not seed it from
// scheduler.concurrency, which is only the shared-state max in shared mode.
// This example matches { max: 100, low: 20, lowest: 5 }.
const schedulerLimits = [5, 5, 20, 100, 100, 100, 100]

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, applicationCompleted) {
  // A queue with an infinite effective target cannot be slot-stalled.
  // In shared mode use the global running gauge; the local gauge can be zero
  // while another worker holds every shared slot.
  const running = s.shared?.running ?? s.running
  const someNow = s.queues.some(
    (q, i) => q.count > 0 && schedulerLimits[i] !== Infinity && running >= schedulerLimits[i],
  )
  // Scheduler queue counters omit fast-path work. The caller supplies a monotonic
  // application counter that advances whenever any scheduled operation settles.
  return { kind: 'scheduler', someNow, completed: applicationCompleted, 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 {
    kind: 'limiter',
    someNow: frontStalled || l.tokens < 0,
    completed: sumCompleted(l.queues),
    forced: l.forced,
    inDebt: l.tokens < 0,
  }
}

function makePressureController() {
  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
  let lastSample = performance.now()

  return {
    get shed() {
      return shed
    },
    get paused() {
      return paused
    },
    update(v) {
      // 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 =
          v.kind === 'limiter'
            ? v.someNow && !progressed && (v.inDebt || v.forced - prev.forced > 0)
            : v.someNow && !progressed
      }

      // Actual elapsed time keeps the EWMA time-constant honest under a jittery loop.
      const now = performance.now()
      const dt = Math.max(0, now - lastSample)
      lastSample = now
      const a = 1 - Math.exp(-dt / 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
    },
  }
}

// Keep one controller (EWMA, counter baseline, clock and latches) per monitored
// Scheduler or Limiter instance. Never alternate multiple resources through one.
const schedulerPressure = makePressureController()
let schedulerCompleted = 0

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

// Producers consult the latched flags (cheap; no per-task stats read):
function submit(task, priority = 'low') {
  if (schedulerPressure.paused) return // hard stop: nothing new
  if (schedulerPressure.shed && priority === 'low') return // discretionary work: first to go
  // This counter belongs to the application, so it includes both queued and
  // fast-path work without adding accounting to Scheduler's hot path.
  return scheduler.run(task, priority).finally(() => {
    schedulerCompleted += 1
  })
}

The scheduler sampler needs the effective target for each queue because a backlog can be delayed by a per-priority target even while the overall scheduler still has room. Those targets are deliberately not duplicated in stats: derive schedulerLimits once from the same constructor configuration, propagating unspecified priorities and clamping every entry to the effective overall/local max (the example above resolves to [5, 5, 20, 100, 100, 100, 100]). In shared mode, compare against the global shared.running gauge but still apply the instance's known local/per-priority targets. The fairness lottery can occasionally dispatch through a priority target; sustained backlog at that target still represents pressure, while the application completion counter keeps it from reading as full when either queued or fast-path work continues to advance. Create an independent controller for every monitored instance: sharing prev, EWMA state, timestamps, or latches across schedulers/limiters mixes unrelated counter domains and produces false progress or false stalls. Aggregate their resulting signals only after each instance has been sampled by its own controller.

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 sampleLimiter(limiter.stats) to a separate controller's update(); never reuse the Scheduler controller. (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-instance and the limiter does not expose a global waiter gauge; aggregate the per-worker signals in your metrics backend rather than treating one worker as process-wide.

Tracing

Both classes accept an optional trace writer. Its shape is exported as TraceWriter:

type TraceWriter = {
  write: ((doc: object, op: string) => void) | null
}

Omitting trace resolves the per-thread writer installed by @nxtedition/trace lazily for each operation. Passing null disables that default for the instance. A writer may also switch its write field between a function and null at runtime. Writers must not call back into the scheduler or limiter being traced; emission occurs inside queue/dispatch bookkeeping. Exceptions from a writer are contained and surfaced as process warnings.

Scheduler operations include scheduler:task, scheduler:queue, scheduler:dispatch, scheduler:park, scheduler:unpark, scheduler:error, and scheduler:dispose. The optional traceTaskThresholdMs suppresses successful scheduler:task documents only when both queue wait and execution duration are below the threshold; errors and disposal rejections always emit. Limiter operations include limiter:queue, limiter:dispatch, limiter:backlog, limiter:stall, limiter:force, and limiter:error.

API

new Scheduler(opts?) / new Scheduler(sharedState, options?)

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

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

• opts.trace — an optional TraceWriter; null explicitly disables the per-thread default.

• opts.traceTaskThresholdMs — a non-negative finite duration threshold for successful scheduler:task trace documents. A successful task is suppressed only when both its queue wait and execution duration are below the threshold; errors and disposals always emit (default: 0, which emits every traced task).

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

• In the shared form, the second argument also accepts { trace, traceTaskThresholdMs }.

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

Execute fn within the scheduler. Returns a promise that resolves with the return value of fn. Callable .then accessors are captured once and assimilated with native promise semantics; non-callable .then accessors may be read again when the returned promise resolves. Dropped rejected tasks remain visible to the host's unhandled-rejection reporting. 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 this instance observed an available slot and fn ran synchronously, or false if the task was queued (also false if the scheduler has been disposed). false records that the call took the queued path; it does not guarantee the callback is still pending when acquire() returns. In shared mode a remote release can race the enqueue, allowing the newly queued callback to dispatch synchronously before the call returns while its result remains false. Another worker may also observe the same room and return true; success is admission bookkeeping, not an exclusive ownership token. 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. Returns true only if this instance observes room at the given priority and nothing is queued at that priority (mirroring limiter.tryConsume); otherwise returns false without queuing. With shared state, racing workers may all return true from the same capacity observation, so this is a soft admission result rather than a lock or deduplication claim. Every successful tryAcquire must still be paired with release().

scheduler.release(): void

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

scheduler.concurrency

The configured overall concurrency target (Infinity when unlimited). In shared mode this reports the buffer's shared max; a lower per-instance options.concurrency.max is used for admission but is not reflected here. The live count may transiently exceed either target.

scheduler.stats

{ running, concurrency, pending, total, queues, shared } — locally running tasks, the configured target, 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.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 target is soft by design).

parsePriority(value): number

Parse a string or number into a normalized priority value. The return type is exported as NumberPriority (-3 | -2 | -1 | 0 | 1 | 2 | 3); Priority additionally includes the seven named string forms.

Limiter

Limiter is a token-bucket rate limiter that targets how many tokens per second are processed — typically bytes, but any integer unit works. Shared races and starvation force-throughs make this a long-run backpressure target rather than an exact per-window quota. 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'

// Target 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)

The shared rate is a soft cap, not an airtight instantaneous ceiling. To keep the cross-worker admission path cheap, each worker reads availability and then performs one atomic debit instead of retrying a compare-and-exchange reservation. Workers racing on the same balance can therefore all be admitted, briefly driving the bucket into debt; earned refills repay that debt before new work fits. Separate refill-clock and token updates may likewise cause timing-dependent under- or overshoot, while elapsed windows are still claimed once and the long-run rate remains the governing intent. The signed Int32 token balance is accounting state, not a hard-cap mechanism.

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 applies an additional rate target on top of its parent. A consume of N tokens through the child normally waits for credit in both the child's own bucket and the parent chain's bucket(s), and the N tokens are debited from every level. The canonical shape is a shared parent targeting the aggregate rate while each worker targets 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 — target 4 MB/s on THIS worker, ~16 MB/s long-run 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 later traffic is delayed and each level trends back toward its configured rate; this is not an exact guarantee for every time window. 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.trace — an optional TraceWriter; null explicitly disables the per-thread default.
• opts may also be a SharedArrayBuffer created by Limiter.makeSharedState(tokensPerSecond). In that case, an optional second argument { trace } configures tracing for this instance.

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 Int32 debit would otherwise add tokens 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 when that debit remains representable as a signed Int32; otherwise it stays queued until refills create enough representable room. Subsequent refills repay debt and suppress later throughput, pulling the average back toward the target rather than guaranteeing any exact time window (the trade-off is a transient burst, at most one stalled item per queue per second). Subject to that representability guard, the same mechanism bounds the wait for a request larger than tokensPerSecond (the configured bucket capacity, which the normal refill path cannot cover): 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. A shared soft-cap overfill may occasionally make such a request fit earlier, but liveness never depends on that race. 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 observed right now and nothing is queued; otherwise returns false without queuing. With shared state, concurrent workers can observe the same tokens and all return true under the soft-cap behavior described above.

limiter.tokensPerSecond

The configured rate target (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 applies an additional rate target on top of this one (see Child Limiters). opts accepts { tokensPerSecond, trace? }. When trace is omitted, the child inherits its parent's setting; an explicit null disables tracing for the child. A consume normally waits until 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 a cross-worker rate target.

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. bench.js is a repository-only harness and is not included in the published package; from a repository checkout, run node --expose-gc packages/scheduler/bench.js to reproduce.

Scheduler hot paths (single thread)

Operation	Throughput
acquire+release (unlimited, no queue)	~120 Mops/s
tryAcquire fast path	~155 Mops/s
acquire+release (SharedArrayBuffer)	~36 Mops/s
queue drain (concurrency=1, batch)	~31 Mops/s
run() (sync body, Promise per task)	~3.6 Mops/s

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