Control the queue

Every valve keeps a local queue. This guide covers the options that decide which job starts next, how long it may wait, and what happens on shutdown.

PriorityCopy Anchor Link

Higher integers start first. The default is 0. Equal priorities keep call order.

// A user is waiting on this one.
await crm.run(() => fetchContact(id), { priority: 10 });

// Background sync can wait.
await crm.run(() => syncContact(id), { priority: -1 });

Priority only orders this process's queue. It does not affect other workers.

CostCopy Anchor Link

cost is how many units a job spends from every rule. The default is 1. Use it when some endpoints count more against the quota than others.

await search.run(() => bulkSearch(query), { cost: 5 });

Under GCRA, cost cannot exceed the rule's burst. Under a fixed window, it cannot exceed limit. Otherwise run rejects with COST_EXCEEDS_LIMIT.

Bound the queueCopy Anchor Link

const crm = createValve({
  id: 'crm:tenant-9',
  rate: { limit: 10, interval: '1s' },
  maxQueue: 1000,
  maxWait: '5s',
});

• maxQueue limits jobs waiting in this process, including jobs whose store call is in flight. Running jobs do not count. When the queue is full, run rejects with QUEUE_FULL.
• maxWait limits how long a job may wait before starting. An expired job rejects with MAX_WAIT_EXCEEDED. It does not limit how long the job runs. Override it per job with run(fn, { maxWait }). maxWait: 0 runs a job only if capacity is free on its first admission attempt. With an asynchronous store such as Redis, zero-wait jobs queued behind an in-flight store call wait for the valve to send another batch. They reject with MAX_WAIT_EXCEEDED without an attempt if the valve does not send one (after a rate denial, a pause, a store outage, or with local concurrency full) or if higher-priority work submitted meanwhile takes their place.

Both turn an overloaded queue into fast, explicit errors that you can handle, instead of unbounded memory and latency.

Cancel a jobCopy Anchor Link

Pass an AbortSignal. If the job is still queued, it rejects with ABORTED, and the error's cause is signal.reason. If it is already running, valve passes the signal to your function and lets it decide.

const controller = new AbortController();

const job = crm.run(({ signal }) => fetch(url, { signal }), {
  signal: controller.signal,
});

controller.abort(new Error('request ended'));

To bound total time, waiting plus running, use AbortSignal.timeout:

await crm.run(({ signal }) => fetch(url, { signal }), {
  signal: AbortSignal.timeout(10_000),
});

Shut downCopy Anchor Link

close stops new work. New run calls reject with CLOSED. Calling close twice returns the same promise.

With global concurrency, close also waits up to leaseTtl for an admission already sent to the store and releases any leases it grants. A wrapFetch request that receives a rate limit response while closing returns that response, or retries when draining.

process.on('SIGTERM', async () => {
  await crm.close({ drain: true });
  await redis.quit();
});

close never closes a store client you supplied. Close your clients after the valves.

Drain during a store outageCopy Anchor Link

With onStoreError: 'wait', draining keeps retrying queued jobs while Redis is offline. The default maxWait is infinite, so close({ drain: true }) can remain pending indefinitely. Set a finite maxWait on the valve or each job to reject queued work with MAX_WAIT_EXCEEDED, or abort queued jobs with a caller-owned signal when your shutdown deadline arrives.

Choose close() if queued work should be rejected immediately on shutdown. The first close call fixes the policy; calling close() after starting a drain returns the same promise and does not cancel the queue.

Every close policy still waits for running callbacks to settle. A queue wait deadline does not limit running work, and lease loss only signals cancellation. Pass the job's signal to operations that support it and give running work its own timeout. Keep the process supervisor's termination deadline as a final bound for callbacks that never settle.

Wait without closingCopy Anchor Link

idle() resolves when nothing is queued or running in this process. Use it to wait for a batch to finish while keeping the valve open.

for (const id of ids) crm.run(() => sync(id)).catch(console.error);
await crm.idle();

Explicit resource managementCopy Anchor Link

Valves are AsyncDisposable. In runtimes that support await using, the valve closes when the scope exits.

await using crm = createValve({
  id: 'crm:script',
  rate: { limit: 5, interval: '1s' },
});