@zukhruf/mutex

A mutex for Node.js with interchangeable lock stores. You select who shares the locks: one object, the threads of one process, a parent process and its children, or all processes on one host. The code that uses the mutex stays the same. Each lease has a fencing token, so a resource can refuse the late writes of a holder that lost its key.

The words in these documents have one meaning each. See the glossary in CONTEXT.md.

The problem

Two requests ask for the last item at the same time. Each request reads the stock, waits for the database, and then writes the stock. Both requests read 1, so both sell the item.

if (stock > 0) {
  // both requests see 1
  await saveReservation(); // the other request runs here
  stock -= 1; // both requests write
}

A mutex lets one holder at a time run this code for a key:

import { MemoryStore, Mutex } from '@zukhruf/mutex';

const mutex = new Mutex(new MemoryStore());

const reserved = await mutex.acquire('product:42', async () => {
  if (stock === 0) return false;
  await saveReservation();
  stock -= 1;
  return true;
});

The full program is in the recipe Stop two requests from selling the last item.

Select a lock store

First find who writes to the resource. Then select the lock store with that reach.

| Lock store | Reach | Order | A holder stops | Needs | | ---------------------------------------------------------------- | ----------------------- | --------------------------------------- | --------------------------------------------- | --------------------------- | | MemoryStore | One object | First come, first served | The locks stop with the process | Nothing | | ThreadStore | One process (threads) | First come, first served | Released when the worker exits | adopt(worker) | | IpcStore | Parent and its children | First come, first served | Released in approximately 2 ms | fork() and adopt(child) | | TicketQueueFileStore | One host | First come, first served | Released after a waiter finds that it stopped | A shared directory | | LockFileStore | One host | No order | Released after a waiter finds that it stopped | A shared directory | | SqliteStore | One host | First come, first served in one process | Released by the kernel | A shared directory | | SocketStore | One host | First come, first served | Released in approximately 2 ms | A shared directory |

If you are not sure:

• One process: use MemoryStore.
• Several processes on one host: use SqliteStore.

Acquire modes

A key is exclusive for every caller. The acquire mode decides only what one caller does while the key is busy: wait (the default), or skip.

import { Modes } from '@zukhruf/mutex';

const report = mutex.key('report:daily', { mode: Modes.skipIfBusy() });

const tick = await report.run(buildReport); // a cron tick skips if a report runs
if (!tick.acquired) return;

const fresh = await report.run(buildReport, { mode: Modes.wait() }); // an admin waits, then runs
await mutex.acquire('product:42', reserve, {
  mode: Modes.skipIfBusy({ waitAtMost: 500 }),
});

A mode that can skip returns { acquired: true, value } | { acquired: false }, and TypeScript makes you check acquired.

To stop a wait, give a signal to the call: mutex.acquire(key, task, { signal }). When the signal aborts, the call rejects with signal.reason, and the task does not run. See Acquire modes.

Holder check

A status view must show if a task runs, but it must not make the key busy. isHeld tells if a key has a holder now, and it never acquires the key:

const running = await report.isHeld(); // or: await mutex.isHeld('report:daily')

A caller that skips if busy is never refused because of a holder check. The holder can change before the answer arrives, so show the answer, but do not acquire a key because of it: an acquire mode decides that. See ADR 0015.

Fencing tokens

A holder can lose its key and not know it, for example when its process freezes. Each lease has a fencing token that increases with each grant. Send the token with each write, and let the resource refuse lower tokens:

await mutex.acquire('product:42', async (lease) => {
  await database.run(
    `UPDATE stock SET quantity = quantity - 1, fence = ?
		 WHERE product = ? AND fence <= ? AND quantity > 0`,
    [lease.token.value, 'product:42', lease.token.value],
  );
});

Sometimes the lock store sees the loss, for example after a failover of SocketStore. Then the signal of the lease aborts with LeaseLostError, and the task can stop before its next write. When the task ends, the call rejects with LeaseLostError:

await mutex.acquire('product:42', async (lease) => {
  for (const order of orders) {
    lease.signal.throwIfAborted();
    await ship(order, lease.token);
  }
});

The signal is a warning. A frozen holder can write before its lock store sees the loss, so the fencing token is still the protection.

The lease is a FencedLease of @zukhruf/fencing. Import FencingToken and the token sources from @zukhruf/fencing. LeaseLostError is from @zukhruf/lease, and this package also exports it, because its calls reject with it.

See Fencing tokens and the recipe Protect a database from stale holders.

Documentation

Concepts

• Reach: who can share a lock, and how to select it.
• Acquire modes: wait or skip while a key is busy, and cancel a wait.
• Fencing tokens: how a resource refuses a stale holder.
• Leader election: how SocketStore selects its coordinator, with SqliteElection of @zukhruf/election.
• Failure modes: what each lock store does when something stops.

Lock stores: one page for each lock store, with What, Why, When, When not, How it works, Acquire modes, Holder check, Failure modes, Options, and Evidence. The SocketStore page also has Two package versions in one directory. See the table above.

Recipes: one use case each, with a full program that you can run.

1. Stop two requests from selling the last item
2. Several app instances on one host
3. A worker pool that you start
4. Worker threads that share a lock
5. Protect a database from stale holders
6. Survive a crashed holder
7. Run a job in only one process, with @zukhruf/election
8. Write your own lock store
9. Skip a job that is already running
10. Compute a value once and share it

Decisions: the architecture decision records tell why the design is as it is.

Use it

This project is an experiment.

npm install @zukhruf/mutex

import { Mutex, SqliteStore } from '@zukhruf/mutex';

Development

This package is part of the zukhruf workspace. Run the commands from the workspace root.

npm install
npx nx run mutex:test        # builds, then runs the tests in src/
npx nx run mutex:test-latest-release   # checks src/ against the latest release on npm
npx nx run mutex:typecheck   # formats, lints, then type checks
npx nx run mutex:build       # compiles src/ to dist/

The tests run from src/, not from dist/: they start workers and child processes from .ts files. A *.release.test.ts file checks this source against the latest release, which it downloads from npm. Its result can change when a new version is released, with no change here, so it has its own target, which nx never caches, and CI runs it.

src/
  mutex/             Mutex, Key, acquire modes, LockHandle, LockStore
  lock-stores/       one folder for each lock store
    remote/          the coordinator and client that ThreadStore, IpcStore and SocketStore share
  shared/            a record check (key file names, file writes and file locks are in @zukhruf/fs)
  testing/           test helpers and the matrix of lock stores (not published)
docs/
  concepts/  stores/  recipes/  adr/

The tests run each lock store through the same scenarios: single process, many threads, many processes, a killed holder, and a failover. Mutation tests broke the mechanisms on purpose, and a test found each break. A test that breaks something stops the run at once (--test-force-exit), so a broken release fails in seconds and does not hang.