@studnicky/mutex

Key-based async mutex for preventing race conditions in concurrent operations.

Install

bash

pnpm add @studnicky/mutex

Usage

Acquire a lock on a named key with runExclusive. Different keys run concurrently; the same key serializes:

import { Mutex } from '../src/index.js';

const mutex = Mutex.create<string>();

// Different keys run concurrently — track completion order
const completionOrder: string[] = [];

class KeyedMutexDemo {
  static async runParallelKeys(): Promise<void> {
    // Two concurrent runExclusive calls on DIFFERENT keys should both start immediately
    await Promise.all([
      mutex.runExclusive('keyA', () => {
        completionOrder.push('keyA');
      }),
      mutex.runExclusive('keyB', () => {
        completionOrder.push('keyB');
      })
    ]);

    console.log('Completion order (parallel keys):', completionOrder);
  }

  static async runSerialSameKey(): Promise<void> {
    // Concurrent calls on the SAME key must serialize
    let counter = 0;
    const results: number[] = [];

    await Promise.all([
      mutex.runExclusive('shared', () => {
        const snapshot = counter;
        counter++;
        results.push(snapshot);
      }),
      mutex.runExclusive('shared', () => {
        const snapshot = counter;
        counter++;
        results.push(snapshot);
      }),
      mutex.runExclusive('shared', () => {
        const snapshot = counter;
        counter++;
        results.push(snapshot);
      })
    ]);

    console.log('Serialized counter:', counter);
    console.log('Serialized results:', results.sort((a, b) => { return a - b; }));
  }

  static showStats(): void {
    const stats = mutex.getStats();
    console.log('Stats:', stats);
  }
}

await KeyedMutexDemo.runParallelKeys();
await KeyedMutexDemo.runSerialSameKey();
KeyedMutexDemo.showStats();

Manual acquire/release

Use acquire() when you need explicit try/finally control, or acquireDisposable() for a releaseable handle:

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

Try it

The builder demo constructs a Mutex via Mutex.builder().withTimeout().withMaxQueueSize().build(). Watch the stats output — maxQueueSize and timeout reflect the values set on the builder, and totalExecuted increments with every runExclusive call.

Mutex builder

/** builderMutex — constructs a Mutex via Mutex.builder()...build() and exercises key-based locking. Run: npx tsx examples/builderMutex.ts */

import assert from 'node:assert/strict';

// #region usage
import { Mutex } from '../src/index.js';

// Build a Mutex with a 5 000 ms acquisition timeout and a max queue depth of 50
const mutex = Mutex.builder<string>()
  .withTimeout(5_000)
  .withMaxQueueSize(50)
  .build();

console.log('Mutex built:', mutex.getStats());

// Different keys run concurrently — both complete without queuing
const order: string[] = [];
await Promise.all([
  mutex.runExclusive('orders', () => { order.push('orders'); }),
  mutex.runExclusive('payments', () => { order.push('payments'); })
]);
console.log('Parallel keys completed:', order.sort());

// Same key serializes — counter increments are race-free
let counter = 0;
await Promise.all([
  mutex.runExclusive('account', () => { counter += 1; }),
  mutex.runExclusive('account', () => { counter += 1; }),
  mutex.runExclusive('account', () => { counter += 1; })
]);
console.log('Serialized counter:', counter);

const stats = mutex.getStats();
console.log('Final stats:', stats);
// #endregion usage

assert.equal(order.length, 2, 'Both parallel keys completed');
assert.equal(counter, 3, 'Three serial increments without data race');
assert.equal(stats.activeLocksCount, 0, 'No active locks after all operations');

console.log('builderMutex: all assertions passed');

Output

Press Execute to run this example against the real library.

The hooks demo subclasses Mutex and overrides eight protected lifecycle methods. Observe the trace: beforeAcquire fires for every caller regardless of contention; onContended fires only for the queued waiter; onAcquireWait fires only after the waiter acquires through the queue; and onQueueDrain fires once the key's queue empties.

Mutex lifecycle hooks

/** observedMutex — trace all Mutex lifecycle hooks in a two-caller contention race. Run: npx tsx examples/observedMutex.ts */

import assert from 'node:assert/strict';

// #region usage
import type { AcquireWaitEventEntity } from '../src/entities/AcquireWaitEventEntity.js';
import type { QueueDrainEventEntity } from '../src/entities/QueueDrainEventEntity.js';
import type { ReleaseEventEntity } from '../src/entities/ReleaseEventEntity.js';

import { Mutex } from '../src/index.js';

class TracingMutex extends Mutex<string> {
  readonly acquireWaitEvents: AcquireWaitEventEntity.Type[] = [];
  readonly queueDrainEvents: QueueDrainEventEntity.Type[] = [];
  readonly releaseEvents: ReleaseEventEntity.Type[] = [];

  protected override beforeAcquire(key: string): void {
    console.log(`[mutex] beforeAcquire key=${key}`);
  }

  protected override afterAcquire(key: string, waitTimeMs: number): void {
    console.log(`[mutex] afterAcquire key=${key} waitTimeMs=${waitTimeMs}`);
  }

  protected override onAcquireWait(key: string, waitTimeMs: number): void {
    console.log(`[mutex] onAcquireWait key=${key} waitTimeMs=${waitTimeMs}`);
    this.acquireWaitEvents.push({ 'key': key, 'waitTimeMs': waitTimeMs });
  }

  protected override onContended(key: string, queueSize: number): void {
    console.log(`[mutex] onContended key=${key} queueSize=${queueSize}`);
  }

  protected override onRelease(key: string): void {
    console.log(`[mutex] onRelease key=${key}`);
    this.releaseEvents.push({ 'key': key });
  }

  protected override beforeRelease(key: string, holdTimeMs: number): void {
    console.log(`[mutex] beforeRelease key=${key} holdTimeMs=${holdTimeMs}`);
  }

  protected override afterRelease(key: string): void {
    console.log(`[mutex] afterRelease key=${key}`);
  }

  protected override onQueueDrain(key: string): void {
    console.log(`[mutex] onQueueDrain key=${key}`);
    this.queueDrainEvents.push({ 'key': key });
  }

  protected override onTimeout(key: string, timeoutMs: number): void {
    console.log(`[mutex] onTimeout key=${key} timeoutMs=${timeoutMs}`);
  }
}

// Runs the contention scenario end-to-end so the module ends up with a single
// top-level `mutex` binding instead of one per intermediate step.
class MutexDemoRunner {
  static async run(mutex: TracingMutex): Promise<void> {
    // Caller 1 grabs the lock
    const release1 = await mutex.acquire('account-42');

    // Caller 2 enqueues (runs concurrently, not awaited yet)
    const pending2 = mutex.acquire('account-42');

    // Simulate holder doing work
    await new Promise<void>((resolve) => { setTimeout(resolve, 10); });

    // Release — hands lock to caller 2
    release1();

    // Caller 2 now gets the lock
    const release2 = await pending2;

    release2();

    // Caller 3 — different key, immediate acquisition, no contention
    const accountBalances: number[] = [];
    await mutex.runExclusive('account-99', () => {
      accountBalances.push(42);
    });
  }
}

const mutex = TracingMutex.create<string>();
await MutexDemoRunner.run(mutex);
// #endregion usage

assert.strictEqual(mutex.acquireWaitEvents.length, 1, `Expected 1 acquireWait event, got ${mutex.acquireWaitEvents.length}`);
assert.strictEqual(mutex.releaseEvents.length, 3, `Expected 3 release events, got ${mutex.releaseEvents.length}`);
assert.strictEqual(mutex.queueDrainEvents.length, 1, `Expected 1 queueDrain event, got ${mutex.queueDrainEvents.length}`);

console.log('observedMutex: all assertions passed');

Output

Press Execute to run this example against the real library.

Subpath exports

Subpath	Contents
`@studnicky/mutex`	`Mutex`, `MutexBuilder`, errors, interfaces
`@studnicky/mutex/constants`	Default configuration constants
`@studnicky/mutex/errors`	`ConfigurationError`, `LockTimeoutError`, `QueueSizeExceededError`
`@studnicky/mutex/interfaces`	`MutexInterface`, `MutexConfigInterface`, `MutexStatsInterface`, `MutexObservabilityInterface`

Observability hooks

Subclass Mutex and override any protected hook to inject trace logging, metrics, or side-effects at the exact stage where they are needed. Hooks should stay fast and non-blocking; observer-hook failures are contained by the base class so lock acquisition and release semantics still win.

Hook	When it fires	Args
`beforeAcquire(key)`	Before any acquisition attempt (immediate or queued)	`key: K`
`afterAcquire(key, waitTimeMs)`	After lock is granted (both immediate and queued paths)	`key: K`, `waitTimeMs: number`
`onAcquireWait(key, waitTimeMs)`	After a queued waiter finally acquires the lock (never fires for immediate grants)	`key: K`, `waitTimeMs: number`
`onContended(key, queueSize)`	When a caller finds the lock held and enqueues itself	`key: K`, `queueSize: number` (depth before enqueue)
`onRelease(key)`	On every lock release by its holder, before any handoff or drop	`key: K`
`beforeRelease(key, holdTimeMs)`	Before a lock is released, with hold time	`key: K`, `holdTimeMs: number`
`afterRelease(key)`	After the lock is dropped completely (no waiters remained)	`key: K`
`onQueueDrain(key)`	When the last waiter for a key leaves the queue (by acquiring or timing out)	`key: K`
`onTimeout(key, timeoutMs)`	When a queued acquisition exceeds the configured timeout	`key: K`, `timeoutMs: number`
`onEnterKey(key, to, from)`	On every per-key FSM state transition	`key: K`, `to: MutexKeyStateType`, `from: MutexKeyStateType`

import type { AcquireWaitEventEntity } from '../src/entities/AcquireWaitEventEntity.js';
import type { QueueDrainEventEntity } from '../src/entities/QueueDrainEventEntity.js';
import type { ReleaseEventEntity } from '../src/entities/ReleaseEventEntity.js';

import { Mutex } from '../src/index.js';

class TracingMutex extends Mutex<string> {
  readonly acquireWaitEvents: AcquireWaitEventEntity.Type[] = [];
  readonly queueDrainEvents: QueueDrainEventEntity.Type[] = [];
  readonly releaseEvents: ReleaseEventEntity.Type[] = [];

  protected override beforeAcquire(key: string): void {
    console.log(`[mutex] beforeAcquire key=${key}`);
  }

  protected override afterAcquire(key: string, waitTimeMs: number): void {
    console.log(`[mutex] afterAcquire key=${key} waitTimeMs=${waitTimeMs}`);
  }

  protected override onAcquireWait(key: string, waitTimeMs: number): void {
    console.log(`[mutex] onAcquireWait key=${key} waitTimeMs=${waitTimeMs}`);
    this.acquireWaitEvents.push({ 'key': key, 'waitTimeMs': waitTimeMs });
  }

  protected override onContended(key: string, queueSize: number): void {
    console.log(`[mutex] onContended key=${key} queueSize=${queueSize}`);
  }

  protected override onRelease(key: string): void {
    console.log(`[mutex] onRelease key=${key}`);
    this.releaseEvents.push({ 'key': key });
  }

  protected override beforeRelease(key: string, holdTimeMs: number): void {
    console.log(`[mutex] beforeRelease key=${key} holdTimeMs=${holdTimeMs}`);
  }

  protected override afterRelease(key: string): void {
    console.log(`[mutex] afterRelease key=${key}`);
  }

  protected override onQueueDrain(key: string): void {
    console.log(`[mutex] onQueueDrain key=${key}`);
    this.queueDrainEvents.push({ 'key': key });
  }

  protected override onTimeout(key: string, timeoutMs: number): void {
    console.log(`[mutex] onTimeout key=${key} timeoutMs=${timeoutMs}`);
  }
}

// Runs the contention scenario end-to-end so the module ends up with a single
// top-level `mutex` binding instead of one per intermediate step.
class MutexDemoRunner {
  static async run(mutex: TracingMutex): Promise<void> {
    // Caller 1 grabs the lock
    const release1 = await mutex.acquire('account-42');

    // Caller 2 enqueues (runs concurrently, not awaited yet)
    const pending2 = mutex.acquire('account-42');

    // Simulate holder doing work
    await new Promise<void>((resolve) => { setTimeout(resolve, 10); });

    // Release — hands lock to caller 2
    release1();

    // Caller 2 now gets the lock
    const release2 = await pending2;

    release2();

    // Caller 3 — different key, immediate acquisition, no contention
    const accountBalances: number[] = [];
    await mutex.runExclusive('account-99', () => {
      accountBalances.push(42);
    });
  }
}

const mutex = TracingMutex.create<string>();
await MutexDemoRunner.run(mutex);

The base class never calls any logger or metrics library. All hooks are no-ops by default.

Source on GitHub

@studnicky/mutex ​

Install ​

Usage ​

Manual acquire/release ​

Try it ​

Subpath exports ​

Observability hooks ​