@ydbjs/coordination

Distributed coordination client for YDB: semaphores, mutexes, and leader elections built on top of YDB coordination nodes.

Features

• Distributed semaphores — acquire tokens with optional data, count, and expiry
• Distributed mutexes — exclusive locking via ephemeral semaphores
• Leader elections — campaign for leadership and observe leader changes
• Automatic reconnection — sessions reconnect transparently; pending operations retry automatically
• Typed errors — SessionClosedError, SessionExpiredError, LeaseReleasedError, LeaderChangedError for reliable instanceof checks
• Session lifecycle signals — session.signal aborts when the session closes or expires
• await using support — all resources implement Symbol.asyncDispose

Installation

npm install @ydbjs/coordination

Requires Node.js >= 20.19.

Overview

CoordinationClient
  └── createNode / dropNode / describeNode / alterNode   — node management
  └── createSession()     → CoordinationSession          — one-shot, ready immediately
  └── openSession()       → AsyncIterable<Session>       — auto-reconnect loop
  └── withSession()       → Promise<T>                   — callback with cleanup

CoordinationSession
  └── mutex(name)         → Mutex                        — exclusive lock
  └── semaphore(name)     → Semaphore                    — counting semaphore
  └── election(name)      → Election                     — leader election
  └── session.signal                                     — aborts on session expiry
  └── session.sessionId                                  — current server session ID

Getting Started

Create a client

import { Driver } from '@ydbjs/core'
import { CoordinationClient } from '@ydbjs/coordination'

let driver = new Driver('grpc://localhost:2136/local')
let client = new CoordinationClient(driver)

// Create a coordination node (once, during provisioning)
await client.createNode('/local/my-app', {})

Session types

| Method | Use when | | ----------------- | ------------------------------------------------------------------- | | createSession() | One-off operation: the session is ready when the promise resolves | | openSession() | Long-running work: automatically recreates the session after expiry | | withSession() | Callback style with guaranteed cleanup |

Mutex

A mutex provides exclusive access. Under the hood it acquires all tokens of an ephemeral semaphore — no createSemaphore call needed.

Blocking lock

for await (let session of client.openSession('/local/my-app', { recoveryWindow: 15_000 }, signal)) {
  let mutex = session.mutex('job-lock')

  try {
    // Blocks until the lock is acquired.
    await using lock = await mutex.lock()

    console.log('lock acquired — doing exclusive work')
    await doWork(lock.signal)
    // lock.release() called automatically here
  } catch {
    if (session.signal.aborted) continue // session expired, retry
    throw error
  }

  break
}

Non-blocking try

await using session = await client.createSession('/local/my-app', {}, signal)
let mutex = session.mutex('job-lock')

let lock = await mutex.tryLock()
if (!lock) {
  console.log('mutex is busy — skipping')
  return
}

await using _ = lock
await doWork(lock.signal)

lock.signal aborts when the lock is released. Use session.signal to detect session death.

Semaphore

A semaphore controls access to a shared resource with a configurable token count.

Create and acquire

await using session = await client.createSession('/local/my-app', {}, signal)
let sem = session.semaphore('connections')

// Create once (idempotent — catch if already exists)
await sem.create({ limit: 10 })

// Acquire one token — blocks until available
await using lease = await sem.acquire({ count: 1 })
await doWork(lease.signal)
// lease.release() called automatically here

Ephemeral semaphore (no prior create needed)

// ephemeral: true — the server creates the semaphore automatically
// and deletes it when the last token is released
await using lease = await sem.acquire({
  count: 1,
  ephemeral: true,
  data: utf8.encode('worker-a:8080'), // optional per-token metadata
})

Non-blocking try

let lease = await sem.tryAcquire({ count: 1 })
if (!lease) {
  console.log('semaphore at capacity')
  return
}
await using _ = lease

Watch for changes

watch() yields immediately with the current state, then again on every server-side change. Reconnects automatically after session expiry.

for await (let session of client.openSession('/local/my-app', { recoveryWindow: 15_000 }, signal)) {
  let sem = session.semaphore('config')

  try {
    for await (let desc of sem.watch({ data: true })) {
      let config = JSON.parse(new TextDecoder().decode(desc.data))
      console.log('config updated:', config)
    }
  } catch {
    if (session.signal.aborted) continue
    throw error
  }

  break
}

Update semaphore data

await using session = await client.createSession('/local/my-app', {}, signal)
await session.semaphore('config').update(utf8.encode(JSON.stringify({ version: 2 })))

Election

An election is a named semaphore where exactly one session can hold the single token. The holder is the leader.

Campaign for leadership

for await (let session of client.openSession('/local/my-app', { recoveryWindow: 15_000 }, signal)) {
  let election = session.election('primary')

  try {
    // Blocks until this session wins. Attach initial leader data (e.g. endpoint).
    await using leadership = await election.campaign(utf8.encode('worker-a:8080'))

    console.log('elected — doing leader work')

    // Update leader data without re-election.
    await leadership.proclaim(utf8.encode('worker-a:9090'))

    // leadership.signal aborts when leadership is lost.
    await doLeaderWork(leadership.signal)

    // leadership.resign() called automatically here
  } catch {
    if (session.signal.aborted) continue
    throw error
  }

  break
}

Observe leader changes

for await (let session of client.openSession('/local/my-app', { recoveryWindow: 15_000 }, signal)) {
  let election = session.election('primary')

  try {
    // Yields on every leader change. state.signal aborts when the leader changes.
    for await (let state of election.observe()) {
      if (!state.data.length) {
        console.log('no leader')
        continue
      }

      let endpoint = new TextDecoder().decode(state.data)
      console.log(state.isMe ? 'i am leader:' : 'current leader:', endpoint)
    }
  } catch {
    if (session.signal.aborted) continue
    throw error
  }

  break
}

One-shot leader query

await using session = await client.createSession('/local/my-app', {}, signal)
let leader = await session.election('primary').leader()
if (leader) {
  console.log('leader:', new TextDecoder().decode(leader.data))
}

Resource management with await using

Every resource in this package implements Symbol.asyncDispose, making await using the safest and most concise way to manage lifetimes.

// Session, lock, and lease released in reverse declaration order —
// guaranteed even if an exception is thrown.
await using session = await client.createSession('/local/my-app', {}, signal)
await using _lock = await session.mutex('job').lock()
await using _lease = await session.semaphore('quota').acquire({ count: 1 })

await doWork()
// _lease.release()  ← first
// _lock.release()   ← second
// session.close()   ← last

Without await using, the equivalent requires nested try/finally blocks — one per resource. await using eliminates nesting and makes forgetting to clean up impossible.

Node management

let client = new CoordinationClient(driver)

// Create a coordination node (server-side container for sessions/semaphores)
await client.createNode('/local/my-app', {})

// Describe current node configuration
let desc = await client.describeNode('/local/my-app')

// Update node configuration
await client.alterNode('/local/my-app', { selfCheckPeriod: 1000 })

// Delete node (fails if sessions are active)
await client.dropNode('/local/my-app')

Session options

| Option | Type | Default | Description | | ---------------- | ------------- | -------- | ------------------------------------------------------------- | | recoveryWindow | number (ms) | 30_000 | How long the server preserves the session during a disconnect | | description | string | '' | Human-readable label visible in server diagnostics | | startTimeout | number (ms) | — | Timeout for the initial session handshake | | retryBackoff | number (ms) | — | Base delay between reconnect attempts |

Error classes

All error classes are exported from @ydbjs/coordination and can be checked with instanceof.

| Error | When | Found in | | ----------------------- | --------------------------------------------------------- | ----------------------------------- | | SessionClosedError | Session was closed (gracefully or destroyed) | session.signal.reason | | SessionExpiredError | Recovery window expired — server dropped the session | session.signal.reason | | LeaseReleasedError | Semaphore lease was released | lease.signal.reason | | LeaderChangedError | A new leader replaced the previous one during observe() | LeaderState.signal.reason | | ObservationEndedError | The observe() async iterator finished | LeaderState.signal.reason | | TryAcquireMissError | Non-blocking acquire found no available tokens (internal) | thrown by acquire(waitTimeout: 0) |

import { SessionExpiredError, LeaseReleasedError } from '@ydbjs/coordination'

session.signal.addEventListener('abort', () => {
  if (session.signal.reason instanceof SessionExpiredError) {
    console.log('session expired — will reconnect')
  }
})

lease.signal.addEventListener('abort', () => {
  if (lease.signal.reason instanceof LeaseReleasedError) {
    console.log('lease released normally')
  }
})

Examples

Runnable examples covering common patterns are in examples/coordination/:

| File | What it shows | | ------------------------ | ------------------------------------------------------- | | mutex.js | Exclusive locking with lock() and tryLock() | | election.js | Leader election with campaign() and observe() | | service-discovery.js | Dynamic endpoint registration with ephemeral semaphores | | shared-config.js | Real-time configuration distribution via watch() | | resource-management.js | await using vs try/finally side by side |

Documentation

• YDB Coordination Nodes
• Leader Election Recipe
• Service Discovery Recipe
• Configuration Publication Recipe

License

Apache-2.0