Share a concurrency limit

Some providers limit concurrent requests, not just request rate. A numeric concurrency only limits one process. The object form with scope: 'global' limits every process that shares the valve id.

Before you beginCopy Anchor Link

Set up a shared Redis store as described in Share a quota across workers.

Configure a global limitCopy Anchor Link

import { Redis } from 'ioredis';
import { createValve } from 'valvejs';
import { redisStore } from 'valvejs/redis';

const client = new Redis(process.env.REDIS_URL!);
const subscriber = client.duplicate();
const store = redisStore({ client, subscriber });

const reports = createValve({
  id: 'reports-api:key-7',
  rate: { limit: 20, interval: '1s' },
  concurrency: { limit: 8, scope: 'global', leaseTtl: '30s' },
  store,
});

await reports.run(({ signal }) =>
  fetch('https://reports.example.com/v1/export', { signal }),
);

Use the same limit and leaseTtl in every worker that shares the id. Redis and a shared memoryStore() reject mismatched rate rules or global concurrency settings with INVALID_OPTIONS. To change them, follow Changing a shared quota. The memory store forgets an id's settings once its rate and lease state expire.

How leases workCopy Anchor Link

1. A job acquires a lease in the same atomic step that spends its rate budget.
2. While the job runs, valve renews the lease every leaseTtl / 3.
3. When the job settles, valve releases the lease.
4. If a worker crashes, its leases expire after leaseTtl.

leaseTtl defaults to '30s', with a minimum of 30ms.

Honor the abort signalCopy Anchor Link

If renewal fails or a lease expires, valve aborts the job's signal. After the callback settles, the job rejects with LEASE_LOST. Pass the signal to your HTTP client so the request stops when ownership is lost. wrapFetch does this for you.

Global concurrency rejects onStoreError: 'open', because running without the store would bypass lease ownership.

Wake queues with a subscriberCopy Anchor Link

Without a subscriber, a process notices shared changes on its next store call, and waiters for a global slot poll at most every 50ms. With a subscriber, pause, resume, and slot release publish a notification. Waiting valves recheck the store after a short random delay, so one release does not send every waiter to the store at once. A valve that keeps losing slots to others waits up to 50ms before it rechecks. Waits caused by rate rules follow the store's retry hint in both cases.

• The subscriber must be a separate connection. Passing the command client is rejected.
• With node-redis, call connect() on the duplicate before passing it in.
• Notifications are hints. Redis remains the source of truth. Global-slot waiters still recheck at least once a second, so a missed message delays a wake by about a second at most.

// node-redis
const subscriber = await client.duplicate().connect();
const store = redisStore({ client, subscriber });

Shut down in orderCopy Anchor Link

valve.close() detaches the valve's listener. store.close() removes the store's subscriptions. Neither closes the connections you supplied.

await reports.close({ drain: true });
await store.close();
await subscriber.quit();
await client.quit();

A shared memoryStore() instance supports global concurrency and notifications too, for valves in the same process.