@studnicky/concurrency
FIFO admission queues, keyed async channels, keyed permit pools, counting semaphores, keyed mutexes, concurrent-call coalescing, async iterable combinators, and finite batch processing.
What it is
@studnicky/concurrency is a set of composable async coordination primitives: queues, channels, permits, keyed locks, single-flight work, async-iterable combinators, batches, and platform-native file locks. It supplies admission and ordering mechanics, not a bookstore job or workflow product.
What it is for
Use it when Northstar Books needs bounded catalogue indexing, per-title mutual exclusion, duplicate-read coalescing, ordered work admission, or finite back-office processing. Consumers own domain keys, retry policy, persistence, and delivery semantics.
Northstar Books examples
- FIFO queue admission maps catalogue-change intake to bounded ordered admission and proves backpressure remains visible to the consumer.
- Batch fixed-window processing maps a finite publisher-price import to fixed-size processing and proves ordered results can be collected.
- Keyed mutex acquisition maps concurrent updates to one ISBN to per-key mutual exclusion while unrelated titles continue.
- Web Locks API mutual exclusion maps a browser-held book-edit draft to the native Web Locks API and proves browser coordination stays runtime-specific.
- Channel and Semaphore maps stock-refresh work to buffered delivery and a shared execution limit.
- Keyed semaphore maps independently limited publisher feeds to per-key permit pools.
- AsyncIter merge / filter / enrich maps catalogue streams to merge, filter, and enrichment combinators without Node streams.
Install
bash
pnpm add @studnicky/concurrencyRuntime imports
Node.js:
typescript
import { Batch, KeyedSemaphore, Mutex, Semaphore } from "@studnicky/concurrency/node";Browser:
typescript
import { Batch, KeyedSemaphore, Mutex, Semaphore } from "@studnicky/concurrency/browser";Requires @studnicky:registry=https://npm.pkg.github.com in .npmrc.
Runtime APIs have identical @studnicky/concurrency/node and @studnicky/concurrency/browser entry points. Choose the entry point for the consumer runtime; both export the same API.
Usage
Channel and Semaphore
Northstar's stock-refresh worker needs two different guarantees at once: deliveries should queue up safely even before anyone's listening, and only so many refresh jobs should run at the same time so the supplier API doesn't get hammered. Channel solves the first — it buffers published items per key so a late subscriber still receives everything in order — while Semaphore solves the second, capping concurrent work with a counting permit. The example below publishes numbers into a channel before anyone subscribes, keeps two independent keys from crossing streams, and then runs four competing tasks through a two-permit semaphore while watching activeCount and available move as permits are acquired, resized with setPermits(), and released.
ts
import { Channel, Semaphore } from '../src/index.js';
class ChannelSemaphoreDemo {
static async runChannel(): Promise<void> {
const channel = Channel.create<number>();
// Publish items before subscribing — they buffer in the channel
await channel.publish('nums', 1);
await channel.publish('nums', 2);
await channel.publish('nums', 3);
await channel.close();
const received = await Array.fromAsync(channel.subscribe('nums'));
console.log('Channel received:', received);
assert.deepEqual(received, [1, 2, 3]);
}
static async runChannelMultiKey(): Promise<void> {
const channel = Channel.create<string>();
// Two independent keys — no cross-talk
await channel.publish('a', 'alpha');
await channel.publish('b', 'beta');
await channel.publish('a', 'apple');
await channel.close();
const fromA = await Array.fromAsync(channel.subscribe('a'));
const fromB = await Array.fromAsync(channel.subscribe('b'));
console.log('Channel key=a:', fromA);
console.log('Channel key=b:', fromB);
assert.deepEqual(fromA, ['alpha', 'apple']);
assert.deepEqual(fromB, ['beta']);
}
static async runChannelConcurrent(): Promise<void> {
const channel = Channel.create<number>();
const received: number[] = [];
// Start subscriber before publishing — it will wait for items
const consuming = (async () => {
for await (const item of channel.subscribe('live')) {
received.push(item);
}
})();
// Publish asynchronously, then close
await Promise.resolve();
await channel.publish('live', 10);
await channel.publish('live', 20);
await channel.close();
await consuming;
console.log('Channel concurrent received:', received);
assert.deepEqual(received, [10, 20]);
}
static async runSemaphoreWithPermit(): Promise<void> {
const sem = Semaphore.create({ 'permits': 2 });
console.log('Semaphore permits:', sem.permits, 'available:', sem.available);
let concurrent = 0;
let maximumConcurrent = 0;
const runTask = async (): Promise<void> => {
concurrent += 1;
maximumConcurrent = Math.max(maximumConcurrent, concurrent);
// Yield to the event loop so tasks interleave
await new Promise<void>((resolve) => { setImmediate(resolve); });
concurrent -= 1;
};
// Launch 4 tasks against a semaphore of 2
await Promise.all([
sem.withPermit(runTask),
sem.withPermit(runTask),
sem.withPermit(runTask),
sem.withPermit(runTask)
]);
console.log('Semaphore maximum concurrent:', maximumConcurrent, '(≤ 2), available after:', sem.available);
assert.ok(maximumConcurrent <= 2, `maximum concurrent ${maximumConcurrent} exceeded permits`);
}
static async runSemaphoreCapacity(): Promise<void> {
const sem = Semaphore.create({ 'permits': 1 });
const releaseFirst = await sem.acquire();
const waiting = sem.acquire();
await Promise.resolve();
assert.equal(sem.queuedCount, 1);
await sem.setPermits(2);
const releaseSecond = await waiting;
assert.equal(sem.activeCount, 2);
await releaseFirst();
await releaseSecond();
await sem.waitForIdle();
assert.equal(sem.available, 2);
}
static async runSemaphoreAcquireRelease(): Promise<void> {
const sem = Semaphore.create({ 'permits': 1 });
const release = await sem.acquire();
const afterAcquire = sem.available;
console.log('Semaphore available after acquire:', afterAcquire);
await release();
const afterRelease = sem.available;
console.log('Semaphore available after release:', afterRelease);
assert.equal(afterAcquire, 0);
assert.equal(afterRelease, 1);
}
}A semaphore accepts an optional queue cap. Pass signal to acquire() or withPermit() to cancel a wait or apply a deadline with AbortSignal.timeout(...); Node and browser consumers use the same API.
FIFO queue
When Northstar's catalogue-change feed fires faster than the indexer can keep up, the fix isn't to drop updates — it's to make the backlog visible and let the producer feel the slowdown. BusQueue is exactly that: a generic FIFO admission primitive that preserves delivery order, applies backpressure once a configured highWaterMark is reached, and accepts an AbortSignal for cancellation. The example enqueues three catalogue changes against a queue capped at two in flight, drains it, and confirms every item arrives in the order it was admitted.
typescript
import { BusQueue } from "@studnicky/concurrency/queue/node";ts
import { BusQueue } from '../src/queue/index.js';
const delivered: string[] = [];
const queue = BusQueue.create<string>({
'handler': (item) => {
delivered.push(item);
const result = Promise.resolve();
return result;
},
'highWaterMark': 2
});
await Promise.all([queue.enqueue('first'), queue.enqueue('second'), queue.enqueue('third')]);
await queue.drain();
console.log('Delivered FIFO:', delivered);Loading example…
Batch finite work
Northstar's nightly publisher-price import has a hard ceiling on how many requests can run at once, but it still needs every result accounted for in the original order. Batch processes a finite list in fixed-size windows — here, five prices doubled two at a time — yielding each window's results as it completes. The demo walks through process() collecting ordered, fulfilled batches and shows how the final flattened array still lines up with the input, regardless of which window a given price landed in.
ts
import { Batch } from '../src/index.js';
class NumberItem {
static double(n: number): Promise<number> {
const doubled = n * 2;
const result = Promise.resolve(doubled);
return result;
}
}
const allResults: number[] = [];
let batchIndex = 0;
for await (const batch of Batch.create<number>(2).process([1, 2, 3, 4, 5], NumberItem.double)) {
console.log('Batch', batchIndex, ':', batch);
allResults.push(...batch);
batchIndex++;
}
console.log('All results:', allResults);Loading example…
Compose bounded dispatch
Compose Semaphore, EventBus, and a scheduler directly when an application needs bounded execution, lifecycle events, and delayed work. The application owns its topic map and publication policy.
Mutex
Two warehouse clerks editing the same ISBN's stock count at once is a recipe for a lost update — but blocking every unrelated title while they sort it out would be worse. Mutex serializes work per key: one holder at a time for 'resource', FIFO-ordered waiters, and every other key free to proceed untouched. The example below acquires the lock manually with a try/finally release, watches a second acquirer queue up behind it, and then shows the same guarantee through acquireDisposable()'s await using-friendly handle.
ts
import { Mutex } from '../src/index.js';
const mutex = Mutex.create<string>();
class AcquireReleaseDemo {
static async runManualRelease(): Promise<void> {
const release = await mutex.acquire('resource');
// Queue a second waiter without awaiting — it will sit in the queue
let secondAcquired = false;
const secondPromise = mutex.acquire('resource').then((rel) => {
secondAcquired = true;
rel();
});
// The second waiter is in the queue
console.log('Queue size while locked:', mutex.queueSize('resource'));
try {
// Critical section — 'resource' is exclusively held here
console.log('Is locked:', mutex.isLocked('resource'));
} finally {
release();
}
// Wait for the queued waiter to complete
await secondPromise;
console.log('Second acquired:', secondAcquired);
console.log('Is locked after all released:', mutex.isLocked('resource'));
console.log('Queue size after completion:', mutex.queueSize('resource'));
}
static async runDisposable(): Promise<void> {
// acquireDisposable — manual release via .release()
const lock = await mutex.acquireDisposable('disposable');
console.log('Lock key:', lock.key);
console.log('Is locked (disposable):', mutex.isLocked('disposable'));
lock.release();
console.log('Is locked after release:', mutex.isLocked('disposable'));
}
}
await AcquireReleaseDemo.runManualRelease();
await AcquireReleaseDemo.runDisposable();Loading example…
File lock
A bookseller editing a draft catalogue entry needs exclusive access to that one file, on whatever platform they're running — a Node import script or a browser tab. FileLock gives Node.js this guarantee through an atomic rename into an owner-qualified lock path, so only one process ever holds the file at a time. The example below acquires the lock, reads the original content, writes an update, reads it back, and releases — all inside a try/finally so the lock can't leak even if something goes wrong in between.
typescript
import { FileLock } from "@studnicky/concurrency/file-lock/node";
import { WebLock } from "@studnicky/concurrency/file-lock/browser";ts
import { FileLock } from '../src/file-lock/node/index.js';
const lock = await FileLock.create({ 'path': filePath });
let original: string;
let updated: string;
try {
original = lock.read();
console.log(`Original content: ${original}`);
lock.write('updated content');
updated = lock.read();
console.log(`Updated content: ${updated}`);
} finally {
lock.release();
}
console.log(`File exists after release: ${String(existsSync(filePath))}`);Loading example…
FileLock coordinates participants that use the same path and lock protocol; it does not replace a database or distributed-locking policy. A held lock exposes synchronous read(), write(), and idempotent release(). FileLock.create() also supports using disposal.
KeyedSemaphore
Northstar pulls stock updates from several publisher feeds in parallel, but each publisher's own API can only take one request at a time — hammer it from two directions and you'll get throttled. KeyedSemaphore hands out a completely independent permit pool per key, so account:a and account:b can both be active at once while a second request for account:a queues up and waits its turn. The example proves the isolation: two keys acquire simultaneously, a third request for an already-busy key is forced to wait, and once everything releases, every key's bookkeeping drops back to zero.
ts
import { KeyedSemaphore } from '../src/index.js';
class KeyedSemaphoreDemo {
static async run(): Promise<void> {
const semaphore = KeyedSemaphore.create<string>({ 'maximumQueueSize': 1, 'permits': 1 });
const releaseAccountA = await semaphore.acquire('account:a');
const releaseAccountB = await semaphore.acquire('account:b');
const waitingAccountA = semaphore.acquire('account:a');
assert.equal(semaphore.activeCount(), 2);
assert.equal(semaphore.queuedCount('account:a'), 1);
await releaseAccountA();
const releaseQueuedAccountA = await waitingAccountA;
await releaseQueuedAccountA();
await releaseAccountB();
await semaphore.waitForIdle();
assert.equal(semaphore.keyCount, 0);
}
}Coalesce: deduplicate concurrent calls by key
When three parts of the storefront ask for the same book's price at the same moment, there's no reason to hit the pricing API three times — the first caller's answer is good enough for all of them. Coalesce deduplicates concurrent calls by key: whoever arrives first starts the factory, and every other caller for that key simply waits on the same promise instead of starting their own. The example confirms the sharing with a call counter that stays at one for concurrent callers, checks isInflight() before, during, and after the call, and then shows that calls which don't overlap in time each pay for their own factory invocation.
ts
import { Coalesce } from '../src/index.js';
class CoalesceDemo {
static async runSharedInFlight(): Promise<void> {
const coalesce = Coalesce.create<string>();
let callCount = 0;
const factory = (): Promise<string> => {
callCount += 1;
return new Promise<string>((resolve) => {
setImmediate(() => { resolve('data'); });
});
};
// Both calls start before factory resolves — they share one in-flight promise
const [a, b] = await Promise.all([
coalesce.run('key', factory),
coalesce.run('key', factory)
]);
console.log('Coalesce shared result a:', a, 'b:', b, 'callCount:', callCount);
assert.equal(a, 'data');
assert.equal(b, 'data');
assert.equal(callCount, 1);
}
static async runIsInflight(): Promise<void> {
const coalesce = Coalesce.create<number>();
const completion = Promise.withResolvers<number>();
const factoryStarted = Promise.withResolvers<void>();
const factory = (): Promise<number> => {
factoryStarted.resolve();
return completion.promise;
};
// Not in-flight before we start
const beforeStart = coalesce.isInflight('item');
const promise = coalesce.run('item', factory);
await factoryStarted.promise;
// In-flight while factory is pending
const duringInflight = coalesce.isInflight('item');
completion.resolve(42);
const result = await promise;
// No longer in-flight once resolved
const afterResolve = coalesce.isInflight('item');
console.log('isInflight — before:', beforeStart, 'during:', duringInflight, 'after:', afterResolve, 'result:', result);
assert.equal(result, 42);
}
static async runDistinctKeys(): Promise<void> {
const coalesce = Coalesce.create<string>();
const callCounts = new Map<string, number>([['a', 0], ['b', 0]]);
const factory = (key: string) => {return (): Promise<string> => {
const nextCount = (callCounts.get(key) ?? 0) + 1;
callCounts.set(key, nextCount);
const result = Promise.resolve(`result-${key}`);
return result;
};};
const [a, b] = await Promise.all([
coalesce.run('a', factory('a')),
coalesce.run('b', factory('b'))
]);
console.log('Distinct keys — a:', a, 'b:', b, 'callCounts:', callCounts);
assert.equal(a, 'result-a');
assert.equal(b, 'result-b');
assert.equal(callCounts.get('a'), 1);
assert.equal(callCounts.get('b'), 1);
}
static async runSequentialCallsEachInvokeFactory(): Promise<void> {
const coalesce = Coalesce.create<number>();
let callCount = 0;
const factory = (): Promise<number> => {
callCount += 1;
const result = Promise.resolve(callCount);
return result;
};
// Sequential calls — no in-flight overlap — each invokes factory
const first = await coalesce.run('seq', factory);
const second = await coalesce.run('seq', factory);
console.log('Sequential — first:', first, 'second:', second, 'callCount:', callCount);
assert.equal(first, 1);
assert.equal(second, 2);
assert.equal(callCount, 2);
}
}Coalesce reserves and publishes the shared completion before it awaits onCoalesceStart or invokes the factory. Reentrant or concurrent callers for that key therefore join the same promise during either stage. A rejection from onCoalesceStart or the factory rejects that shared promise for the leader and every joiner, and the entry is removed after settlement.
Timeout: CoalesceTimeoutError
A timeout option caps how long an individual caller waits on the shared in-flight promise. When a caller's timeout elapses, only that caller's run() rejects with CoalesceTimeoutError — the in-flight entry is left untouched for other callers still waiting on it:
typescript
import { Coalesce, CoalesceTimeoutError } from "@studnicky/concurrency/node";
const coalesce = Coalesce.create<Response>({ timeout: 5000 });
try {
await coalesce.run("user:42", () => fetch("/api/user/42"));
} catch (error) {
if (error instanceof CoalesceTimeoutError) {
console.log(error.key); // 'user:42'
console.log(error.timeoutMs); // 5000
}
}AsyncIter: merge, filter, enrich
Northstar's catalogue has several independent streams of change events — new titles, price updates, inventory ticks — and combining them without pulling in Node's stream machinery keeps the logic portable to the browser too. AsyncIter works directly against plain async function* generators: merge interleaves multiple sources in arrival order, filter keeps only the values a predicate accepts (sync or async), and enrich left-joins extra data onto matching items while passing the rest through untouched. The example chains all three — merging two numeric ranges, filtering down to multiples of three, and tagging only the larger survivors with a tier label — to show the combinators composing into one pipeline.
ts
import { AsyncIter } from '../src/index.js';
// Native async sources — `async function*` is an AsyncIterable in browsers and
// Node alike, so AsyncIter's combinators consume them directly (no node:stream).
// Each awaits a microtask before yielding to model a genuinely asynchronous
// producer.
class AsyncSources {
static async *range(start: number, end: number): AsyncGenerator<number> {
for (let i = start; i <= end; i += 1) {
await Promise.resolve();
yield i;
}
}
static async *words(items: string[]): AsyncGenerator<string> {
const itemLength = items.length;
for (let index = 0; index < itemLength; index += 1) {
const item = items[index];
await Promise.resolve();
if (item !== undefined) { yield item; }
}
}
static async *itemsOf<T>(items: T[]): AsyncGenerator<T> {
const itemLength = items.length;
for (let index = 0; index < itemLength; index += 1) {
const item = items[index];
await Promise.resolve();
if (item !== undefined) { yield item; }
}
}
}
class AsyncIterDemo {
static async runMerge(): Promise<void> {
// Merge two non-overlapping ranges
const merged = AsyncIter.merge(
AsyncSources.range(1, 3),
AsyncSources.range(10, 12)
);
const results = await Array.fromAsync(merged);
// All 6 values present — order is arrival-based (FIFO within each source), so check membership
assert.equal(results.length, 6);
const expected = [1, 2, 3, 10, 11, 12];
const expectedLength = expected.length;
const resultsSet = new Set(results);
for (let i = 0; i < expectedLength; i += 1) {
assert.ok(resultsSet.has(expected[i]!), `Missing value ${expected[i]}`);
}
// Within each source, relative order is preserved
assert.ok(results.indexOf(1) < results.indexOf(2));
assert.ok(results.indexOf(2) < results.indexOf(3));
assert.ok(results.indexOf(10) < results.indexOf(11));
assert.ok(results.indexOf(11) < results.indexOf(12));
console.log('merge results:', results);
}
static async runMergeEmpty(): Promise<void> {
const results = await Array.fromAsync(AsyncIter.merge<number>());
assert.equal(results.length, 0);
console.log('merge empty: []');
}
static async runFilter(): Promise<void> {
const evens = AsyncIter.filter(AsyncSources.range(1, 10), (number) => {
const result = number % 2 === 0;
return result;
});
const results = await Array.fromAsync(evens);
assert.deepEqual(results, [2, 4, 6, 8, 10]);
console.log('filter evens:', results);
}
static async runFilterAsync(): Promise<void> {
// Predicate may return a Promise
const longWords = AsyncIter.filter(
AsyncSources.words(['hi', 'hello', 'hey', 'greetings']),
async (w) => { const result = await Promise.resolve(w.length > 3); return result; }
);
const results = await Array.fromAsync(longWords);
assert.deepEqual(results, ['hello', 'greetings']);
console.log('filter long words:', results);
}
static async runEnrich(): Promise<void> {
const items = AsyncSources.itemsOf<ItemEntity.Type>([{ 'id': 1 }, { 'id': 2 }, { 'id': 3 }]);
// Only even ids get enrichment; odd ids pass through unchanged
const enriched = AsyncIter.enrich<ItemEntity.Type, EnrichmentEntity.Type, EnrichedEntity.Type>(
items,
(item) => { const result = Promise.resolve(item.id % 2 === 0 ? { 'label': `even-${item.id}` } : null); return result; },
(item, enrichment) => {
const result: EnrichedEntity.Type = { 'id': item.id, 'label': enrichment.label };
return result;
}
);
const results = await Array.fromAsync(enriched);
assert.equal(results.length, 3);
// id:1 — unenriched, passes through as-is
assert.deepEqual(results[0], { 'id': 1 });
// id:2 — enriched
assert.deepEqual(results[1], { 'id': 2, 'label': 'even-2' });
// id:3 — unenriched
assert.deepEqual(results[2], { 'id': 3 });
console.log('enrich results:', results);
}
static async runMergeFilterEnrichPipeline(): Promise<void> {
// Compose all three combinators in a pipeline
const merged = AsyncIter.merge<number>(
AsyncSources.range(1, 5),
AsyncSources.range(6, 10)
);
const filtered = AsyncIter.filter<number>(merged, (number) => {
const result = number % 3 === 0;
return result;
});
const enriched = AsyncIter.enrich<number, { 'tier': string }, { 'n': number; 'tier': string }>(
filtered,
(n) => { const result = Promise.resolve(n > 5 ? { 'tier': 'high' as const } : null); return result; },
(number, enrichment) => {
const result: { 'n': number; 'tier': string } = { 'n': number, 'tier': enrichment.tier };
return result;
}
);
const results = await Array.fromAsync(enriched);
// Multiples of 3 in 1..10: 3, 6, 9
// 3 → unenriched (≤5)
// 6 → enriched { n:6, tier:'high' }
// 9 → enriched { n:9, tier:'high' }
// merge() yields in arrival order across concurrent sources — sort for stable assertions
const sorted = [...results].toSorted((a, b) => {
const nA = typeof a === 'number' ? a : a.n;
const nB = typeof b === 'number' ? b : b.n;
const result = nA - nB;
return result;
});
assert.equal(sorted.length, 3);
assert.deepEqual(sorted[0], 3);
assert.deepEqual(sorted[1], { 'n': 6, 'tier': 'high' });
assert.deepEqual(sorted[2], { 'n': 9, 'tier': 'high' });
console.log('merge+filter+enrich pipeline:', sorted);
}
}Observability hooks
Watching what a Semaphore, Channel, or Coalesce instance is actually doing shouldn't require rewriting its logic — each class exposes protected hook methods built for exactly that, fired at every meaningful lifecycle event. Subclass any of them, override the hooks that matter, and the base class keeps working exactly as before while your overrides collect whatever trace or metric Northstar needs.
Semaphore hooks
| Hook | When it fires | Args |
|---|---|---|
onAcquire | Permit granted immediately | permitsBefore: number |
onAcquireWait | Caller queued (no permit available) | — |
onContended | New waiter added to queue | queueLength: number |
onRelease | Permit returned to pool (no waiting callers) | permitsAfter: number |
onReleaseDelegated | Permit handed to queued waiter | — |
Channel hooks
| Hook | When it fires | Args |
|---|---|---|
onEnqueue | Item added to buffer | key: string, item: T |
onDequeue | Item removed from buffer by subscriber | key: string, item: T |
onPublishDropped | Publish attempted on closed channel | key: string, item: T |
onClose | Channel closes (all keys) | — |
onOverflow | Configured highWaterMark is reached after an item is staged | key: string, depth: number |
Batch hooks
| Hook | When it fires | Args |
|---|---|---|
onBatchStart | A non-empty operation starts. | total: number |
onConcurrencySaturated | A fixed-size window is full. | — |
onItemStart | An item operation starts. | index: number |
onItemSuccess | An item operation fulfills. | index: number, result: TResult |
onItemError | An item operation rejects. | index: number, error: BaseError |
onItemSettled | An item operation settles. | index: number |
onBatchComplete | All items settle successfully or through processSettled(). | stats: BatchStatsEntity.Type |
Mutex hooks
| Hook | When it fires | Args |
|---|---|---|
beforeAcquire | Before each acquisition attempt. | key: K |
afterAcquire | A lock is granted. | key: K, waitTimeMs: number |
onContended | An acquisition queues behind a holder. | key: K, queueSize: number |
onAcquireWait | A queued acquisition is granted. | key: K, waitTimeMs: number |
beforeRelease | Before a holder releases. | key: K, holdTimeMs: number |
afterRelease | After release processing completes. | key: K |
onRelease | Every holder release completes. | key: K |
onQueueDrain | The final waiter leaves a key queue. | key: K |
onTimeout | A queued acquisition reaches its deadline. | key: K, timeoutMs: number |
Coalesce hooks
| Hook | When it fires | Args |
|---|---|---|
onCoalesceStart | After the shared completion is reserved and before the leader invokes the factory | key: string |
onCoalesceJoin | Caller joined an in-flight call | key: string |
onCoalesceSettled | In-flight promise settled | key: string, success: boolean |
onTimeout | One caller exceeds its configured wait timeout without disturbing the shared in-flight call | key: string, timeoutMs: number |
ts
import { Channel, Coalesce, Semaphore } from '../src/index.js';
class ObservedSemaphore extends Semaphore {
readonly acquireEvents: number[] = [];
readonly acquireWaitEvents: number[] = [];
readonly contendedEvents: number[] = [];
readonly releaseEvents: number[] = [];
readonly releaseDelegatedEvents: number[] = [];
constructor(permits: number) { super({ 'permits': permits }); }
protected override onAcquire(permitsBefore: number): void {
this.acquireEvents.push(permitsBefore);
console.log(`[concurrency:semaphore] acquire permitsBefore=${permitsBefore}`);
}
protected override onAcquireWait(): void {
this.acquireWaitEvents.push(1);
console.log('[concurrency:semaphore] acquireWait');
}
protected override onContended(queueLength: number): void {
this.contendedEvents.push(queueLength);
console.log(`[concurrency:semaphore] contended queueLength=${queueLength}`);
}
protected override onRelease(permitsAfter: number): void {
this.releaseEvents.push(permitsAfter);
console.log(`[concurrency:semaphore] release permitsAfter=${permitsAfter}`);
}
protected override onReleaseDelegated(): void {
this.releaseDelegatedEvents.push(1);
console.log('[concurrency:semaphore] releaseDelegated');
}
}
class ObservedChannel<T> extends Channel<T> {
readonly dequeueEvents: { 'item': T; 'key': string }[] = [];
readonly droppedEvents: { 'item': T; 'key': string }[] = [];
readonly enqueueEvents: { 'item': T; 'key': string }[] = [];
closeCount = 0;
constructor() { super(); }
protected override onDequeue(key: string, item: T): void {
this.dequeueEvents.push({ 'item': item, 'key': key });
console.log(`[concurrency:channel] dequeue key=${key} item=${JSON.stringify(item)}`);
}
protected override onEnqueue(key: string, item: T): void {
this.enqueueEvents.push({ 'item': item, 'key': key });
console.log(`[concurrency:channel] enqueue key=${key} item=${JSON.stringify(item)}`);
}
protected override onPublishDropped(key: string, item: T): void {
this.droppedEvents.push({ 'item': item, 'key': key });
console.log(`[concurrency:channel] publishDropped key=${key} item=${JSON.stringify(item)}`);
}
protected override onClose(): void {
this.closeCount += 1;
console.log('[concurrency:channel] close');
}
}
class ObservedCoalesce<T> extends Coalesce<T> {
readonly joinEvents: string[] = [];
readonly settledEvents: { 'key': string; 'success': boolean }[] = [];
readonly startEvents: string[] = [];
constructor() { super(); }
protected override onCoalesceJoin(key: string): void {
this.joinEvents.push(key);
console.log(`[concurrency:coalesce] join key=${key}`);
}
protected override onCoalesceSettled(key: string, success: boolean): void {
this.settledEvents.push({ 'key': key, 'success': success });
console.log(`[concurrency:coalesce] settled key=${key} success=${success}`);
}
protected override onCoalesceStart(key: string): void {
this.startEvents.push(key);
console.log(`[concurrency:coalesce] start key=${key}`);
}
}
class ObservedConcurrencyDemo {
private static resolveImmediatelyFactory(): Promise<string> {
return new Promise<string>((resolve) => {
setImmediate(() => { resolve('v'); });
});
}
static async runSemaphore(): Promise<ObservedSemaphore> {
console.log('\n=== Semaphore ===');
const sem = new ObservedSemaphore(1);
// First acquire: immediate grant
const r1 = await sem.acquire();
// Second acquire: no permits, must wait
const pendingR2 = sem.acquire();
// Give microtasks a chance to run the queued waiter registration
await Promise.resolve();
// Release first permit: delegates to queued waiter
await r1();
// Wait for the second acquire to resolve
const r2 = await pendingR2;
// Release the second permit: returns to pool
await r2();
return sem;
}
static async runChannel(): Promise<ObservedChannel<string>> {
console.log('\n=== Channel ===');
const ch: ObservedChannel<string> = new ObservedChannel<string>();
await ch.publish('events', 'hello');
await ch.publish('events', 'world');
const collected: string[] = [];
const gen: AsyncGenerator<string> = ch.subscribe('events');
for await (const item of gen) {
collected.push(item);
if (collected.length >= 2) { break; }
}
await ch.close();
console.log('Channel received:', collected);
return ch;
}
static async runCoalesce(): Promise<ObservedCoalesce<string>> {
console.log('\n=== Coalesce ===');
const coalesce: ObservedCoalesce<string> = new ObservedCoalesce<string>();
// 3 concurrent calls for same key: 1 leader + 2 joiners
await Promise.all([
coalesce.run('batch', ObservedConcurrencyDemo.resolveImmediatelyFactory),
coalesce.run('batch', ObservedConcurrencyDemo.resolveImmediatelyFactory),
coalesce.run('batch', ObservedConcurrencyDemo.resolveImmediatelyFactory)
]);
// Sequential call: new leader
await coalesce.run('batch', ObservedConcurrencyDemo.resolveImmediatelyFactory);
return coalesce;
}
}The base class never calls any logger or metrics library. All hooks are no-ops by default.
Try it
This demo builds a Semaphore and a Channel directly with create(...) and puts both through their paces live. Watch the semaphore cap four competing tasks down to two running at once no matter how they race to start, and watch the channel hand back everything it buffered in exactly the order it was published, even across multiple keys and a concurrent subscriber.
Loading example…
Loading example…
This demo feeds two independent async function* ranges — no Node.js streams involved — straight into AsyncIter.merge, AsyncIter.filter, and AsyncIter.enrich. Watch the merge interleave values from both ranges as they arrive, the filter narrow a 1–10 range down to just the evens, and the final composed pipeline keep only the multiples of three while tagging anything above five with a tier: 'high' enrichment.
Loading example…
Public entrypoints
| Import path | Use it when |
|---|---|
@studnicky/concurrency/node | A server or worker needs general runtime coordination primitives. |
@studnicky/concurrency/browser | A browser bundle needs the same general primitives; it is the runtime alternative to /node. |
@studnicky/concurrency/entities | A consumer needs shared coordination schemas as contracts. |
@studnicky/concurrency/interfaces | A consumer needs shared coordination interfaces as contracts. |
@studnicky/concurrency/queue/node | A server needs FIFO admission for catalogue work. |
@studnicky/concurrency/queue/browser | A browser needs the same queue primitive; it is the runtime alternative to /queue/node. |
@studnicky/concurrency/queue/entities | A consumer needs queue state and option schemas as contracts. |
@studnicky/concurrency/queue/interfaces | A consumer needs queue construction contracts. |
@studnicky/concurrency/mutex | An ISBN or order key needs FIFO mutual exclusion. |
@studnicky/concurrency/file-lock/node | A Node process needs file-based exclusive coordination. |
@studnicky/concurrency/file-lock/browser | A browser needs native Web Locks; it is the runtime alternative to /file-lock/node. |
@studnicky/concurrency/file-lock/entities | A consumer needs file-lock schemas as contracts. |
@studnicky/concurrency/file-lock/interfaces | A consumer needs file-lock ports as contracts. |
@studnicky/concurrency/batch | A finite import needs bounded windowed processing. |
@studnicky/concurrency/throttle/node | A server needs adaptive or fixed concurrency limiting with drain and abort control. |
@studnicky/concurrency/throttle/browser | A browser needs the same throttle primitive; it is the runtime alternative to /throttle/node. |
@studnicky/concurrency/throttle/entities | A consumer needs throttle configuration and stats schemas as contracts. |
@studnicky/concurrency/throttle/interfaces | A consumer needs throttle construction and event contracts. |
Exports
| Symbol | Purpose | Import path |
|---|---|---|
AsyncIter | Static combinators for async iterables. | @studnicky/concurrency/node |
Batch | Fixed-window and immediate-refill processing for finite inputs. | @studnicky/concurrency/batch@studnicky/concurrency/node |
BatchError | Signals invalid Batch construction. | @studnicky/concurrency/node |
BusQueue | Generic FIFO admission and ordered delivery with backpressure. | @studnicky/concurrency/queue/node |
BusQueueConfigError | Signals invalid queue construction. | @studnicky/concurrency/queue/node |
BusQueueCreateOptionsInterface | Queue construction contract. | @studnicky/concurrency/queue/interfaces |
BusQueueAbortedStateEntity | Terminal queue lifecycle state. | @studnicky/concurrency/queue/entities |
BusQueueAbortEventEntity | Queue abort lifecycle event. | @studnicky/concurrency/queue/entities |
BusQueueAbortingStateEntity | Queue abort-in-progress lifecycle state. | @studnicky/concurrency/queue/entities |
BusQueueCreateOptionsEntity | Schema-backed queue construction options. | @studnicky/concurrency/queue/entities |
BusQueueDrainingStateEntity | Queue drain lifecycle state. | @studnicky/concurrency/queue/entities |
BusQueueLoopFinishedEventEntity | Queue loop-finished lifecycle event. | @studnicky/concurrency/queue/entities |
BusQueueOpenStateEntity | Queue open lifecycle state. | @studnicky/concurrency/queue/entities |
BusQueueOptionsEntity | Schema-backed serializable queue options. | @studnicky/concurrency/queue/entities |
BusQueueReleaseForAbortEffectEntity | Queue abort release effect. | @studnicky/concurrency/queue/entities |
BusQueueStartLoopEventEntity | Queue loop-start lifecycle event. | @studnicky/concurrency/queue/entities |
Channel | String-keyed fan-in async-generator inbox. | @studnicky/concurrency/node |
ChannelError | Base error for channel operations. | @studnicky/concurrency/node |
Coalesce | Deduplicates concurrent calls by key. | @studnicky/concurrency/node |
CoalesceTimeoutError | Signals a caller timeout while a coalesced operation remains in flight. | @studnicky/concurrency/node |
CoalesceWaitCompletedError | Abort reason for a caller's cancelled timeout timer once its wait on the shared in-flight promise finishes. | @studnicky/concurrency/node |
ConcurrencyError | Base error for the package. | @studnicky/concurrency/node |
LockTimeoutError | Signals a queued mutex acquisition deadline. | @studnicky/concurrency/node |
Mutex | FIFO keyed mutual exclusion with queue caps, deadlines, disposable handles, and lifecycle hooks. | @studnicky/concurrency/mutex@studnicky/concurrency/node |
MutexError | Base error for mutex operations. | @studnicky/concurrency/node |
FileLock | Atomic-rename, single-writer lock for a Node.js file. | @studnicky/concurrency/file-lock/node |
FileLockConfigError | Signals invalid file-lock configuration. | @studnicky/concurrency/file-lock/node |
FileLockContentionError | Signals an unsuccessful atomic rename acquisition. | @studnicky/concurrency/file-lock/node |
FileLockError | Base error for file-lock operations. | @studnicky/concurrency/file-lock/node |
FileLockFileSystemError | Signals an underlying filesystem failure. | @studnicky/concurrency/file-lock/node |
FileLockInspection | Inspects owner-qualified lock paths. | @studnicky/concurrency/file-lock/node |
FileLockLivenessError | Signals an owner-liveness query failure. | @studnicky/concurrency/file-lock/node |
FileLockWebLockError | Signals a native Web Locks API request failure. | @studnicky/concurrency/file-lock/browser |
FileLockRecovery | Restores an explicitly inspected lock path. | @studnicky/concurrency/file-lock/node |
FileLockRecoveryConflictError | Signals recovery blocked by an occupied original path. | @studnicky/concurrency/file-lock/node |
FileLockTimeoutError | Signals acquisition past its deadline. | @studnicky/concurrency/file-lock/node |
FileRenameLock | Atomic rename acquisition and release primitive. | @studnicky/concurrency/file-lock/node |
NodeOwnerLiveness | Checks a Node process owner's liveness. | @studnicky/concurrency/file-lock/node |
WebLock | Native browser Web Locks API adapter. | @studnicky/concurrency/file-lock/browser |
FileLockAcquiredEventEntity | File-lock acquisition transition event. | @studnicky/concurrency/file-lock/entities |
FileLockInspectionEntity | Inspected owner-qualified lock-path record. | @studnicky/concurrency/file-lock/entities |
FileLockOptionsEntity | Validates Node file-lock options. | @studnicky/concurrency/file-lock/entities |
FileLockPathStateEntity | Canonical original and held lock paths. | @studnicky/concurrency/file-lock/entities |
FileLockReleasedEventEntity | File-lock release transition event. | @studnicky/concurrency/file-lock/entities |
FileLockStateEntity | File-lock lifecycle state. | @studnicky/concurrency/file-lock/entities |
WebLockOptionsEntity | Validates browser Web Lock options. | @studnicky/concurrency/file-lock/entities |
FileLockCreateOptionsInterface | Node file-lock construction contract. | @studnicky/concurrency/file-lock/interfaces |
FileLockInspectionOptionsInterface | Lock inspection contract. | @studnicky/concurrency/file-lock/interfaces |
FileLockRecoveryOptionsInterface | Explicit recovery contract. | @studnicky/concurrency/file-lock/interfaces |
FileRenameLockCreateOptionsInterface | Atomic rename-lock construction contract. | @studnicky/concurrency/file-lock/interfaces |
LockInterface | Shared idempotent release contract. | @studnicky/concurrency/file-lock/interfaces |
OwnerLivenessInterface | Owner liveness abstraction. | @studnicky/concurrency/file-lock/interfaces |
OwnerTokenInterface | Owner token abstraction. | @studnicky/concurrency/file-lock/interfaces |
WebLockCreateOptionsInterface | Native Web Lock construction contract. | @studnicky/concurrency/file-lock/interfaces |
WebLockManagerInterface | Native Web Locks manager dependency contract. | @studnicky/concurrency/file-lock/interfaces |
QueueSizeExceededError | Signals that a keyed mutex queue has reached its configured capacity. | @studnicky/concurrency/node |
Semaphore | Counting permit gate for asynchronous work. | @studnicky/concurrency/node |
KeyedSemaphore | Independent per-key permit gates. | @studnicky/concurrency/node |
SemaphoreQueueFullError | Signals that a bounded semaphore queue is full. | @studnicky/concurrency/node |
SemaphoreError | Base error for semaphore operations. | @studnicky/concurrency/node |
Entities
@studnicky/concurrency/entities exports all schema-backed option, state, transition, and event declarations. Each entity namespace provides Schema, Type, and validate.
typescript
import {
BatchStatsEntity,
ChannelKeyStateEntity,
SemaphoreOptionsEntity,
} from "@studnicky/concurrency/entities";Interfaces
@studnicky/concurrency/interfaces exports runtime-only contracts that compose platform values rather than serializable schema fields.
typescript
import type { SemaphoreAcquireOptionsInterface } from "@studnicky/concurrency/interfaces";Batch<TResult>
| Member | Signature | Description |
|---|---|---|
create | static create<TResult>(maximumConcurrent?: number) => Batch<TResult> | Constructs a Batch with a positive integer window size. |
process | <T>(items, operation) => AsyncGenerator<TResult[]> | Processes fixed windows and yields each fulfilled result window. |
processSettled | <T>(items, operation) => AsyncGenerator<PromiseSettledResult<TResult>[]> | Processes fixed windows and yields every settlement. |
processContinuous | <T>(items, operation) => Promise<TResult[]> | Immediately refills capacity and rejects on the first failure. |
processContinuousSettled | <T>(items, operation) => Promise<PromiseSettledResult<TResult>[]> | Immediately refills capacity and returns every settlement. |
Channel<T>
| Member | Signature | Description |
|---|---|---|
create | static create<T>(options?: ChannelOptionsEntity.InputType) => Channel<T> | Constructs a channel from optional configuration |
publish | (key: string, item: T) => Promise<void> | Sends an item to key and completes its admission hooks |
subscribe | (key: string) => AsyncGenerator<T> | Yields items published to key |
close | () => Promise<void> | Closes all channels; subscribers stop after draining |
Semaphore
| Member | Signature | Description |
|---|---|---|
create | static create(options: SemaphoreOptionsEntity.InputType) => Semaphore | Constructs a semaphore with the required permit count |
acquire | (options?: SemaphoreAcquireOptionsInterface) => Promise<() => Promise<void>> | Waits for a permit; signal cancels or bounds the wait |
withPermit | <T>(callback: () => Promise<T>) => Promise<T> | Acquires, runs callback, releases |
setPermits | (permits: number) => Promise<void> | Changes capacity; work already holding permits continues |
waitForIdle | () => Promise<void> | Resolves when no work is active or queued |
activeCount | number | Current holders |
maximumQueueSize | number | Waiting capacity; 0 means unlimited |
available | number | Current available permit count; may be negative while a reduced capacity drains |
queuedCount | number | Current waiting acquirers |
permits | number | Total configured permits |
Mutex<K>
| Member | Signature | Description |
|---|---|---|
create | static create<K>(options?: MutexCreateOptionsInterface) => Mutex<K> | Constructs a keyed mutex. |
acquire | (key: K) => Promise<() => void> | Waits for the key and returns an idempotent release function. |
acquireDisposable | (key: K) => Promise<MutexLockInterface> | Returns an await using compatible lock handle. |
runExclusive | (key: K, callback) => Promise<unknown> | Runs one callback at a time for the key. |
queueSize, isLocked, size | (key?: K) => number | boolean | Inspects admission and active locks. |
completeQueue | () => Promise<void> | Resolves when all held and queued work drains. |
clear | () => void | Rejects queued acquirers and clears active state. |
KeyedSemaphore<K>
| Member | Signature | Description |
|---|---|---|
create | static create<K>(options: SemaphoreOptionsEntity.InputType) => KeyedSemaphore<K> | Constructs one permit policy per key |
acquire | (key: K, options?: SemaphoreAcquireOptionsInterface) => Promise<() => Promise<void>> | Acquires a permit for key |
withPermit | <T>(key: K, callback: () => Promise<T>, options?: SemaphoreAcquireOptionsInterface) => Promise<T> | Runs work within key’s permit limit |
waitForIdle | (key?: K) => Promise<void> | Waits for one key or all keys to drain |
activeCount, queuedCount | (key?: K) => number | Reads per-key or aggregate load |
Coalesce<T>
| Member | Signature | Description |
|---|---|---|
create | static create<T>(options?: CoalesceOptionsEntity.InputType) => Coalesce<T> | Constructs a coalescer from optional timeout configuration |
run | (key: string, factory: () => Promise<T>) => Promise<T> | Shares in-flight promise for key |
isInflight | (key: string) => boolean | True if a promise for key is pending |
AsyncIter
| Member | Signature | Description |
|---|---|---|
merge | <T>(...sources) => AsyncGenerator<T> | FIFO merge of N async iterables |
filter | <T>(source, predicate) => AsyncGenerator<T> | Yields items matching sync/async predicate |
enrich | <T, E, R>(source, lookup, merge) => AsyncGenerator<T | R> | Left-join enrichment per item |