Add debug logging to @ydbjs/coordination

- Add 'coordination' category to @ydbjs/debug loggers
- Log session lifecycle: connecting, ready, reconnecting, expiry, close
- Log all semaphore operations with meaningful context
- Log retry when a request is interrupted by reconnect
- Add Semaphore, Mutex, Election scopes for targeted DEBUG filtering
- Use human-readable messages describing what is happening, not what code runs

Author: polRk

packages/coordination/src/client.ts

@@ -1,10 +1,13 @@
 import type { Driver } from '@ydbjs/core'
+import { loggers } from '@ydbjs/debug'
 
 import { CoordinationNodeRuntime } from './runtime/node-runtime.js'
 import { getSessionRuntime } from './internal/session-runtime.js'
 import type { CoordinationNodeConfig, CoordinationNodeDescription } from './runtime/node-runtime.js'
 import { CoordinationSession } from './session.js'
 
+let dbg = loggers.coordination.extend('client')
+
 export interface SessionOptions {
 	description?: string
 	recoveryWindow?: number
@@ -46,12 +49,15 @@ export class CoordinationClient {
 			throw signal.reason
 		}
 
+		dbg.log('creating session on %s', path)
 		let session = new CoordinationSession(this.#driver, { path, ...options }, signal)
 
 		try {
 			await getSessionRuntime(session).waitReady(signal)
+			dbg.log('session ready on %s (id=%s)', path, session.sessionId)
 			return session
 		} catch (error) {
+			dbg.log('failed to open session on %s: %O', path, error)
 			session.destroy(error)
 			throw error
 		}
@@ -62,6 +68,7 @@ export class CoordinationClient {
 		options?: SessionOptions,
 		signal?: AbortSignal
 	): AsyncIterable<CoordinationSession> {
+		dbg.log('opening persistent session on %s', path)
 		for (;;) {
 			if (signal?.aborted) {
 				return
@@ -76,6 +83,8 @@ export class CoordinationClient {
 			if (!shouldOpenNext) {
 				return
 			}
+
+			dbg.log('session expired on %s, reopening', path)
 		}
 	}
 

packages/coordination/src/election.ts

@@ -1,7 +1,10 @@
+import { loggers } from '@ydbjs/debug'
 import { getSessionRuntime } from './internal/session-runtime.js'
 import { Lease, Semaphore } from './semaphore.js'
 import type { CoordinationSession } from './session.js'
 
+let dbg = loggers.coordination.extend('election')
+
 export interface LeaderInfo {
 	data: Uint8Array
 }
@@ -33,6 +36,7 @@ export class Leadership implements AsyncDisposable {
 	async proclaim(data: Uint8Array, signal?: AbortSignal): Promise<void> {
 		throwIfAborted(signal)
 
+		dbg.log('proclaiming leadership on %s (%d bytes)', this.#name, data.byteLength)
 		return getSessionRuntime(this.#session).updateSemaphore(this.#name, data, signal)
 	}
 
@@ -41,6 +45,7 @@ export class Leadership implements AsyncDisposable {
 			return
 		}
 
+		dbg.log('resigning from leadership on %s', this.#name)
 		this.#resigned = true
 		await this.#lease.release(signal)
 	}
@@ -64,13 +69,16 @@ export class Election {
 	}
 
 	async campaign(data: Uint8Array, signal?: AbortSignal): Promise<Leadership> {
+		dbg.log('campaigning for leadership on %s', this.#name)
 		let semaphore = new Semaphore(this.#session, this.#name)
 		let lease = await semaphore.acquire({ count: 1, data }, signal)
 
+		dbg.log('became leader on %s', this.#name)
 		return new Leadership(lease, this.#name, this.#session)
 	}
 
 	async *observe(signal?: AbortSignal): AsyncIterable<LeaderState> {
+		dbg.log('observing leadership changes on %s', this.#name)
 		let previousLeader: { sessionId: bigint; orderId: bigint } | null = null
 		// Tracks the AbortController for the currently live LeaderState so we can
 		// abort it as soon as the leader changes, before yielding the next state.
@@ -98,6 +106,7 @@ export class Election {
 				currentController = new AbortController()
 
 				if (!owner) {
+					dbg.log('no leader on %s', this.#name)
 					yield {
 						data: emptyBytes,
 						isMe: false,
@@ -106,30 +115,40 @@ export class Election {
 					continue
 				}
 
+				let isMe =
+					this.#session.sessionId !== null && owner.sessionId === this.#session.sessionId
+				dbg.log(
+					'leader changed on %s (sessionId=%s, isMe=%s)',
+					this.#name,
+					owner.sessionId,
+					isMe
+				)
 				yield {
 					data: owner.data,
-					isMe:
-						this.#session.sessionId !== null &&
-						owner.sessionId === this.#session.sessionId,
+					isMe,
 					signal: currentController.signal,
 				}
 			}
 		} finally {
 			// Ensure the last yielded state's signal is aborted when iteration ends
 			// so consumers relying on it for cancellation are always unblocked.
+			dbg.log('stopped observing %s', this.#name)
 			currentController?.abort(new Error('Election observation ended'))
 		}
 	}
 
 	async leader(signal?: AbortSignal): Promise<LeaderInfo | null> {
+		dbg.log('reading current leader on %s', this.#name)
 		let semaphore = new Semaphore(this.#session, this.#name)
 		let description = await semaphore.describe({ owners: true }, signal)
 		let owner = description.owners?.[0]
 
 		if (!owner) {
+			dbg.log('no current leader on %s', this.#name)
 			return null
 		}
 
+		dbg.log('current leader on %s is session %s', this.#name, owner.sessionId)
 		return {
 			data: owner.data,
 		}

packages/coordination/src/mutex.ts

@@ -1,9 +1,12 @@
 import type { CoordinationSession } from './session.js'
+import { loggers } from '@ydbjs/debug'
 import { getSessionRuntime } from './internal/session-runtime.js'
 import { isTryAcquireMiss } from './internal/try-acquire.js'
 import { Lease } from './semaphore.js'
 import type { SessionRuntime } from './runtime/session-runtime.js'
 
+let dbg = loggers.coordination.extend('mutex')
+
 // Ephemeral semaphores in YDB have a server-hardcoded limit of MAX_UINT64.
 // Mutex exclusivity is achieved by acquiring all tokens at once — no other
 // session can acquire even a single token while they are all held.
@@ -29,26 +32,31 @@ export class Mutex {
 	}
 
 	async lock(signal?: AbortSignal): Promise<Lock> {
+		dbg.log('waiting to acquire lock on %s', this.#name)
 		let lease = await this.#runtime.acquireSemaphore(
 			this.#name,
 			{ count: mutexCapacity, ephemeral: true },
 			signal
 		)
 
+		dbg.log('lock acquired on %s', this.#name)
 		return new Lock(this.#name, lease)
 	}
 
 	async tryLock(signal?: AbortSignal): Promise<Lock | null> {
+		dbg.log('trying to acquire lock on %s without waiting', this.#name)
 		try {
 			let lease = await this.#runtime.acquireSemaphore(
 				this.#name,
 				{ count: mutexCapacity, ephemeral: true, waitTimeout: 0 },
 				signal
 			)
 
+			dbg.log('lock acquired on %s', this.#name)
 			return new Lock(this.#name, lease)
 		} catch (error) {
 			if (isTryAcquireMiss(error)) {
+				dbg.log('%s is already locked, skipping', this.#name)
 				return null
 			}
 

packages/coordination/src/runtime/node-runtime.ts

@@ -16,9 +16,12 @@ import {
 import { StatusIds_StatusCode } from '@ydbjs/api/operation'
 import type { Entry } from '@ydbjs/api/scheme'
 import type { Driver } from '@ydbjs/core'
+import { loggers } from '@ydbjs/debug'
 import { YDBError } from '@ydbjs/error'
 import type { Client } from 'nice-grpc'
 
+let dbg = loggers.coordination.extend('node')
+
 export interface CoordinationNodeConfig {
 	selfCheckPeriod?: number
 	sessionGracePeriod?: number
@@ -41,6 +44,7 @@ export class CoordinationNodeRuntime {
 	}
 
 	async describe(path: string, signal?: AbortSignal): Promise<CoordinationNodeDescription> {
+		dbg.log('reading configuration of coordination node at %s', path)
 		let request = create(DescribeNodeRequestSchema, { path })
 		let response = signal
 			? await this.#client.describeNode(request, { signal })
@@ -72,6 +76,7 @@ export class CoordinationNodeRuntime {
 	}
 
 	async create(path: string, cfg: CoordinationNodeConfig, signal?: AbortSignal): Promise<void> {
+		dbg.log('creating coordination node at %s', path)
 		let request = create(CreateNodeRequestSchema, { path, config: toNodeConfigMessage(cfg) })
 		let response = signal
 			? await this.#client.createNode(request, { signal })
@@ -88,6 +93,7 @@ export class CoordinationNodeRuntime {
 	}
 
 	async alter(path: string, cfg: CoordinationNodeConfig, signal?: AbortSignal): Promise<void> {
+		dbg.log('updating configuration of coordination node at %s', path)
 		let request = create(AlterNodeRequestSchema, { path, config: toNodeConfigMessage(cfg) })
 		let response = signal
 			? await this.#client.alterNode(request, { signal })
@@ -104,6 +110,7 @@ export class CoordinationNodeRuntime {
 	}
 
 	async drop(path: string, signal?: AbortSignal): Promise<void> {
+		dbg.log('dropping coordination node at %s', path)
 		let request = create(DropNodeRequestSchema, { path })
 		let response = signal
 			? await this.#client.dropNode(request, { signal })