@studnicky/concurrency

Keyed async channels, counting semaphores, concurrent-call coalescing, and async iterable combinators.

Install

bash

pnpm add @studnicky/concurrency

Requires @studnicky:registry=https://npm.pkg.github.com in .npmrc.

The package declares runtime dependencies on @studnicky/circular-buffer, @studnicky/errors, @studnicky/json, ajv, and json-schema-to-ts.

Usage

Channel and Semaphore

Channel provides keyed producer/consumer buffering; Semaphore gates concurrent access to a shared resource with a counting permit:

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 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);
  }
}

Coalesce: deduplicate concurrent calls by key

All concurrent callers for the same key share a single in-flight promise; sequential callers each invoke the factory independently:

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';

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

Compose async iterables with FIFO merge, sync/async predicate filter, and left-join enrichment:

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

Each class exposes protected hook methods you can override in a subclass to observe internal lifecycle events without modifying the class logic.

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`

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`

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

The channel-and-semaphore demo constructs both primitives directly with create(...). Watch the Semaphore limit concurrent executions to two at a time across four competing tasks, then watch the Channel deliver buffered items in publish order.

Loading example…

The async-iter demo uses native async function* generators as sources — no Node.js streams — and passes them through AsyncIter.merge, AsyncIter.filter, and AsyncIter.enrich. Watch the merged output interleave values from two independent ranges, the filter keep only even numbers, and the final composed pipeline emit only the multiples-of-three with a tier enrichment applied to values above five.

Loading example…

Exports

Symbol	Purpose	Import path
`AsyncIter`	Static combinators for async iterables.	`@studnicky/concurrency`
`Channel`	String-keyed fan-in async-generator inbox.	`@studnicky/concurrency`
`ChannelError`	Base error for channel operations.	`@studnicky/concurrency`
`Coalesce`	Deduplicates concurrent calls by key.	`@studnicky/concurrency`
`CoalesceTimeoutError`	Signals a caller timeout while a coalesced operation remains in flight.	`@studnicky/concurrency`
`ConcurrencyError`	Base error for the package.	`@studnicky/concurrency`
`Semaphore`	Counting permit gate for asynchronous work.	`@studnicky/concurrency`
`SemaphoreError`	Base error for semaphore operations.	`@studnicky/concurrency`

Entities

@studnicky/concurrency/entities exports all schema-backed option, state, transition, and event declarations. Each entity namespace provides Schema, Type, and validate.

typescript

import { SemaphoreOptionsEntity } from '@studnicky/concurrency/entities';

Interfaces

@studnicky/concurrency/interfaces exports the type-only state and transition contracts used by the coordination state machines.

typescript

import type { ChannelKeyStateInterface } from '@studnicky/concurrency/interfaces';

`Channel<T>`

Member	Signature	Description
`create`	`static create<T>(options?: ChannelOptionsEntity.Type) => 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.Type) => Semaphore`	Constructs a semaphore with the required permit count
`acquire`	`() => Promise<() => Promise<void>>`	Waits for a permit; returns an asynchronous release function
`withPermit`	`<T>(callback: () => Promise<T>) => Promise<T>`	Acquires, runs callback, releases
`available`	`number`	Current available permit count
`permits`	`number`	Total configured permits

`Coalesce<T>`

Member	Signature	Description
`create`	`static create<T>(options?: CoalesceOptionsEntity.Type) => 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

Source on GitHub

@studnicky/concurrency ​

Install ​

Usage ​

Channel and Semaphore ​

Coalesce: deduplicate concurrent calls by key ​

Timeout: CoalesceTimeoutError ​

AsyncIter: merge, filter, enrich ​

Observability hooks ​

Semaphore hooks ​

Channel hooks ​

Coalesce hooks ​

Try it ​

Exports ​

Entities ​

Interfaces ​

Channel<T> ​

Semaphore ​

Coalesce<T> ​

AsyncIter ​

@studnicky/concurrency

Install

Usage

Channel and Semaphore

Coalesce: deduplicate concurrent calls by key

Timeout: `CoalesceTimeoutError`

AsyncIter: merge, filter, enrich

Observability hooks

Semaphore hooks

Channel hooks

Coalesce hooks

Try it

Exports

Entities

Interfaces

`Channel<T>`

`Semaphore`

`Coalesce<T>`

`AsyncIter`