@kysera/infra

Infrastructure utilities for Kysera - health monitoring, resilience patterns, and graceful shutdown.

Installation

npm install @kysera/infra kysely

Overview

Dependencies: @kysera/core (peer: kysely >=0.29.0)

:::info Package Type This is a utility package providing infrastructure and resilience features. It's not part of the Repository/DAL pattern - it works with Kysely instances directly. :::

Module Exports

Everything is available from the package root, plus four subpath exports for targeted imports:

import { checkDatabaseHealth, CircuitBreaker } from '@kysera/infra' // Package root
import { performHealthCheck, getMetrics } from '@kysera/infra/health' // Health checks & metrics
import { withRetry, CircuitBreakerError } from '@kysera/infra/resilience' // Retry & circuit breaker
import { createMetricsPool } from '@kysera/infra/pool' // Pool metrics
import { registerShutdownHandlers } from '@kysera/infra/shutdown' // Graceful shutdown

note

CircuitBreakerError is exported only from the @kysera/infra/resilience subpath, not from the package root.

Key Features

• Health Monitoring - Database connectivity checks with latency tracking
• Retry Logic - Automatic retries with exponential backoff
• Transaction Retry - Re-run whole transactions on serialization failures and deadlocks
• Circuit Breaker - Prevent cascading failures
• Graceful Shutdown - Clean database connection termination
• Pool Metrics - Connection pool monitoring (pg, mysql2, better-sqlite3, tarn; other pools report static placeholders flagged detected: false)

Quick Start

import {
  checkDatabaseHealth,
  HealthMonitor,
  withRetry,
  CircuitBreaker,
  registerShutdownHandlers,
  createMetricsPool
} from '@kysera/infra'

// Create metrics-enabled pool
const metricsPool = createMetricsPool(pgPool)

// Health monitoring
const monitor = new HealthMonitor(db, { pool: metricsPool, intervalMs: 30000 })
monitor.start(result => {
  if (result.status !== 'healthy') console.warn('Health issue:', result)
})

// Resilience patterns
const breaker = new CircuitBreaker(5, 60000)
const users = await breaker.execute(() => withRetry(() => db.selectFrom('users').execute()))

// Graceful shutdown
registerShutdownHandlers(db, {
  timeout: 10000,
  onShutdown: async () => monitor.stop()
})

Health Monitoring

Basic Health Check

import { checkDatabaseHealth } from '@kysera/infra'

const result = await checkDatabaseHealth(db)
console.log(result.status) // 'healthy' | 'degraded' | 'unhealthy'

Health Status Levels:

• healthy - Latency < 100ms
• degraded - Latency 100-500ms
• unhealthy - Latency > 500ms or connection failed

With Pool Metrics

import { checkDatabaseHealth, createMetricsPool } from '@kysera/infra'

const metricsPool = createMetricsPool(pgPool)
const result = await checkDatabaseHealth(db, metricsPool)

console.log(result.metrics?.poolMetrics)
// { totalConnections: 10, activeConnections: 2, idleConnections: 8, waitingRequests: 0 }

Extended Health Check

performHealthCheck wraps checkDatabaseHealth with an options object and verbose mode:

import { performHealthCheck } from '@kysera/infra'

const result = await performHealthCheck(db, {
  pool: metricsPool,
  verbose: true // Adds databaseVersion to metrics
})

interface HealthCheckOptions {
  pool?: MetricsPool // Connection pool for metrics extraction
  verbose?: boolean // Include verbose information
  logger?: KyseraLogger // Custom logger
}

Continuous Monitoring

import { HealthMonitor } from '@kysera/infra'

const monitor = new HealthMonitor(db, {
  pool: metricsPool,
  intervalMs: 30000
})

monitor.start(result => {
  if (result.status !== 'healthy') {
    // Send alert, log to monitoring system
  }
})

monitor.getLastCheck() // Get last result
await monitor.checkNow() // Immediate check
monitor.isRunning() // true while started
monitor.stop()
monitor.destroy() // Alias for stop() with explicit destruction semantics

HealthMonitor implements Disposable, so it works with explicit resource management (using):

{
  using monitor = new HealthMonitor(db, { intervalMs: 30000 })
  monitor.start()
} // Automatically stopped when scope exits (Symbol.dispose calls stop())

Database Metrics

getMetrics aggregates real query statistics from a database wrapped with withDebug() from @kysera/debug:

import { withDebug } from '@kysera/debug'
import { getMetrics, hasDatabaseMetrics } from '@kysera/infra'

const debugDb = withDebug(db, { maxMetrics: 1000 })
await debugDb.selectFrom('users').selectAll().execute()

const metrics = getMetrics(debugDb, {
  slowQueryThreshold: 100,
  pool: metricsPool
})
console.log(metrics.queries?.avgDuration) // Real average from tracked queries
console.log(metrics.recommendations) // Performance recommendations

warning

getMetrics works only on a withDebug-wrapped database - it throws an error otherwise. Use the hasDatabaseMetrics(db) type guard to check whether a database instance tracks metrics.

interface GetMetricsOptions {
  period?: string // Time period label (informational, default: '1h')
  pool?: MetricsPool // Optional pool for connection metrics
  slowQueryThreshold?: number // Slow query threshold in ms (default: 100)
}

interface MetricsResult {
  period: string
  timestamp: string // ISO timestamp of collection
  connections?: { total: number; active: number; idle: number; max: number }
  queries?: {
    total: number
    avgDuration: number
    minDuration: number
    maxDuration: number
    p95Duration: number
    p99Duration: number
    slowCount: number
  }
  recommendations?: string[]
}

Resilience Patterns

Retry with Exponential Backoff

import { withRetry, isTransientError } from '@kysera/infra'

const result = await withRetry(() => db.selectFrom('users').execute(), {
  maxAttempts: 5,
  delayMs: 500,
  backoff: true,
  onRetry: (attempt, error) => console.log(`Retry ${attempt}:`, error),
  shouldRetry: isTransientError
})

Recognized Transient Errors:

• Network: ECONNREFUSED, ETIMEDOUT, ECONNRESET, EPIPE
• PostgreSQL: 57P03, 08006, 40001, 40P01 (deadlock)
• MySQL: ER_LOCK_DEADLOCK, ER_LOCK_WAIT_TIMEOUT
• SQLite: SQLITE_BUSY, SQLITE_LOCKED

Reusable Retry Wrapper

createRetryWrapper wraps a function once so every call retries with the same options:

import { createRetryWrapper } from '@kysera/infra'

const fetchUsers = async () => db.selectFrom('users').selectAll().execute()
const fetchUsersWithRetry = createRetryWrapper(fetchUsers, { maxAttempts: 3 })

const users = await fetchUsersWithRetry() // Retries automatically

function createRetryWrapper<TArgs extends unknown[], TResult>(
  fn: (...args: TArgs) => Promise<TResult>,
  options?: RetryOptions
): (...args: TArgs) => Promise<TResult>

Transaction Retry (Serialization Failures & Deadlocks)

Under SERIALIZABLE/REPEATABLE READ isolation — or whenever two transactions deadlock — the database aborts one of them and expects the application to re-run it. withTransactionRetry re-runs the whole callback in a fresh transaction per attempt, so aborted attempts leave no partial writes and retries see current data:

import { withTransactionRetry } from '@kysera/infra'

const receipt = await withTransactionRetry(
  db, // Kysely instance or KyseraExecutor — plugins stay active
  async trx => {
    const from = await trx
      .selectFrom('accounts')
      .where('id', '=', fromId)
      .select('balance')
      .executeTakeFirstOrThrow()
    if (from.balance < amount) throw new Error('insufficient funds')
    await trx.updateTable('accounts').where('id', '=', fromId)
      .set(eb => ({ balance: eb('balance', '-', amount) })).execute()
    await trx.updateTable('accounts').where('id', '=', toId)
      .set(eb => ({ balance: eb('balance', '+', amount) })).execute()
    return { fromId, toId, amount }
  },
  { maxAttempts: 5, onRetry: attempt => metrics.increment('txn_retry') }
)

Retried error codes (via the default shouldRetry: isSerializationError):

Dialect	Codes
PostgreSQL	40001 (serialization_failure), 40P01 (deadlock_detected)
MySQL	ER_LOCK_DEADLOCK/1213, ER_LOCK_WAIT_TIMEOUT/1205
MSSQL	error number 1205 (deadlock victim)
SQLite	SQLITE_BUSY (including SQLITE_BUSY_* variants)

The cause chain of wrapped errors is followed (up to 5 levels). isSerializationError is deliberately narrower than isTransientError: connection drops are excluded because a transaction interrupted mid-commit may already have committed — re-running it is not automatically safe.

Defaults differ from withRetry where it matters for conflicts: delayMs: 100, maxDelayMs: 2000 (still exponential backoff + jitter), maxAttempts: 3.

Two rules for the callback:

1. It must be safe to re-run. Keep external side effects (queue publishes, HTTP calls) outside the callback, or make them idempotent.
2. Set isolation inside the callback if you need it — issue SET TRANSACTION ISOLATION LEVEL SERIALIZABLE (or REPEATABLE READ) as the first statement; each retry re-applies it because each attempt is a brand-new transaction.

Works with both Kysely<DB> and KyseraExecutor<DB> — it only calls db.transaction().execute(fn), so a plugin-wrapped executor keeps its plugins active inside every attempt.

Circuit Breaker

import { CircuitBreaker } from '@kysera/infra'
import { CircuitBreakerError } from '@kysera/infra/resilience'

// Constructor signature 1: Simple parameters
const breaker1 = new CircuitBreaker(5, 60000) // threshold, resetTimeMs

// Constructor signature 2: Options object
const breaker2 = new CircuitBreaker({
  threshold: 5,
  resetTimeMs: 60000,
  onStateChange: (newState, oldState) => console.log(`${oldState} -> ${newState}`)
})

try {
  const result = await breaker.execute(() => db.selectFrom('users').execute())
} catch (error) {
  if (error instanceof CircuitBreakerError) {
    // Service unavailable (circuit open or testing recovery)
  }
}

// Check circuit state (synchronous reads)
if (breaker.isOpen()) {
  console.log('Circuit is open - service unavailable')
}
if (breaker.isClosed()) {
  console.log('Circuit is closed - operating normally')
}

breaker.getState() // { state: 'open', failures: 5, lastFailureTime: ..., isTestingHalfOpen: false }
await breaker.reset() // Reset to closed (async)
await breaker.forceOpen() // Force open for maintenance (async)

Circuit States:

• closed - Normal operation
• open - Too many failures, requests fail immediately
• half-open - Testing recovery, allows one request

Concurrency Contract:

• execute(), reset(), and forceOpen() serialize state transitions through an internal mutex - that's why they're async
• getState(), isOpen(), and isClosed() are synchronous snapshot reads (JavaScript is single-threaded, so reads are atomic and need no mutex)
• Safe to use across multiple concurrent requests

CircuitBreakerError:

execute() rejects with CircuitBreakerError when the circuit is open ('Circuit breaker is open') or while a half-open test request is already in flight ('Circuit breaker is testing recovery'). It extends DatabaseError from @kysera/core and is exported only from the @kysera/infra/resilience subpath.

Combined Resilience

const result = await breaker.execute(() =>
  withRetry(() => db.selectFrom('users').execute(), { maxAttempts: 3 })
)

Connection Pool Metrics

import { createMetricsPool, isMetricsPool } from '@kysera/infra'

// PostgreSQL
const metricsPool = createMetricsPool(pgPool)
const metrics = metricsPool.getMetrics()
// { total: 10, idle: 8, active: 2, waiting: 0 }

// Type guard
if (isMetricsPool(pool)) {
  const metrics = pool.getMetrics()
}

Pool type is detected once at creation. Recognized pools: pg (PostgreSQL), mysql2, better-sqlite3, and tarn.js (the pool kysely's MssqlDialect and knex use — detected via its public numUsed()/numFree()/numPendingAcquires() counters; tarn has destroy() instead of end(), so cast it to DatabasePool when passing it in).

Every result carries a detected flag so dashboards can tell truth from stub:

• Recognized pools → real counters with detected: true
• Any other pool type → static placeholders { total: 10, idle: 0, active: 0, waiting: 0, detected: false } — treat these as "unknown", not as real utilization

Note for MSSQL: kysely's MssqlDialect constructs its tarn pool internally and does not expose it, so in a typical kysely MSSQL setup there is no pool object to hand to createMetricsPool — you get tarn metrics only when you construct or otherwise hold the tarn pool yourself.

Graceful Shutdown

Automatic Signal Handlers

import { registerShutdownHandlers } from '@kysera/infra'

registerShutdownHandlers(db, {
  signals: ['SIGTERM', 'SIGINT'],
  timeout: 30000,
  onShutdown: async () => {
    await flushCache()
    monitor.stop()
  }
})

Manual Shutdown

import { gracefulShutdown, shutdownDatabase } from '@kysera/infra'

// With cleanup
await gracefulShutdown(db, {
  timeout: 10000,
  onShutdown: async () => console.log('Cleanup...')
})

// Simple
await shutdownDatabase(db)

Shutdown Controller

import { createShutdownController } from '@kysera/infra'

const shutdown = createShutdownController(db, { timeout: 10000 })
shutdown.registerSignals()

if (!shutdown.isShuttingDown()) {
  await shutdown.execute()
}

Drain Semantics

What gracefulShutdown actually does: run your onShutdown hook, then call db.destroy(), racing the whole thing against timeout. Draining behavior therefore comes from the driver's destroy, not from Kysera:

• In-flight queries: drained, not killed. With the pg driver, destroy() is pool.end(), which waits for checked-out connections to be returned — a query already running completes normally and shutdown resolves after it finishes (pinned by an integration test against live PostgreSQL). mysql2's pool.end() behaves the same way.
• New work: NOT gated. Nothing stops other code from issuing queries while shutdown is in progress; queries that miss the window fail with "driver has already been destroyed". isShuttingDown() on the controller is a flag you can check, not a gate.
• On timeout: rejected, not cancelled. If draining takes longer than timeout, gracefulShutdown rejects — but db.destroy() keeps running in the background (a promise race cannot cancel it).

The pattern for a clean drain is: stop intake → await in-flight work → shut down the pool:

process.on('SIGTERM', async () => {
  server.close() // 1. stop intake (no new HTTP requests / job pulls)
  await jobQueue.onIdle() // 2. let in-flight application work finish
  await gracefulShutdown(db, { timeout: 30000 }) // 3. drain & close the pool
  process.exit(0)
})

API Reference

Health Types

type HealthStatus = 'healthy' | 'degraded' | 'unhealthy'

interface HealthCheckResult {
  status: HealthStatus
  checks: HealthCheck[]
  errors?: string[]
  metrics?: HealthMetrics
  timestamp: Date
}

interface HealthCheckOptions {
  pool?: MetricsPool
  verbose?: boolean
  logger?: KyseraLogger
}

interface HealthMonitorOptions {
  pool?: MetricsPool
  intervalMs?: number // Default: 30000
  logger?: KyseraLogger
}

See Database Metrics for GetMetricsOptions, MetricsResult, and the hasDatabaseMetrics type guard.

Resilience Types

interface RetryOptions {
  maxAttempts?: number // Default: 3
  delayMs?: number // Default: 1000
  maxDelayMs?: number // Default: 30000 (caps exponential backoff)
  backoff?: boolean // Default: true
  jitterFactor?: number // Default: 0.25 (prevents thundering herd)
  shouldRetry?: (error: unknown) => boolean
  onRetry?: (attempt: number, error: unknown) => void
}

// Same shape as RetryOptions; withTransactionRetry defaults differ:
// shouldRetry → isSerializationError, delayMs → 100, maxDelayMs → 2000
type TransactionRetryOptions = RetryOptions

// Anything that can start transactions: Kysely<DB>, KyseraExecutor<DB>,
// or a structural test double. TRX is what the callback receives
// (Transaction<DB> for both Kysely and KyseraExecutor).
interface RetryableTransactionSource<TRX> {
  transaction(): {
    execute<T>(callback: (trx: TRX) => Promise<T>): Promise<T>
  }
}

function withTransactionRetry<TRX, T>(
  db: RetryableTransactionSource<TRX>,
  fn: (trx: TRX) => Promise<T>,
  options?: TransactionRetryOptions
): Promise<T>

// True for pg 40001/40P01, mysql 1213/1205, mssql 1205, SQLITE_BUSY*
// (follows the error `cause` chain up to 5 levels)
function isSerializationError(error: unknown): boolean

type CircuitState = 'closed' | 'open' | 'half-open'

interface CircuitBreakerOptions {
  threshold?: number // Default: 5
  resetTimeMs?: number // Default: 60000
  onStateChange?: (newState: CircuitState, previousState: CircuitState) => void
}

interface CircuitBreakerState {
  state: CircuitState
  failures: number
  lastFailureTime: number | undefined
  isTestingHalfOpen: boolean // True while a half-open test request is in flight
}

CircuitBreaker Class

class CircuitBreaker {
  // Constructor signatures
  constructor(threshold?: number, resetTimeMs?: number)
  constructor(options?: CircuitBreakerOptions)

  // Execute a function with circuit breaker protection
  execute<T>(fn: () => Promise<T>): Promise<T>

  // Synchronous state reads (atomic snapshots, no mutex needed)
  getState(): CircuitBreakerState
  isOpen(): boolean // Check if circuit is open
  isClosed(): boolean // Check if circuit is closed

  // Async state transitions (serialized through internal mutex)
  reset(): Promise<void> // Reset to closed state
  forceOpen(): Promise<void> // Force circuit open
}

// Thrown when execute() rejects a request (circuit open or half-open test in flight).
// Exported only from '@kysera/infra/resilience'.
class CircuitBreakerError extends DatabaseError {
  name: 'CircuitBreakerError'
}

Pool Types

interface PoolMetrics {
  total: number
  idle: number
  active: number
  waiting: number
  // Always set by createMetricsPool: true for recognized pools
  // (pg, mysql2, better-sqlite3, tarn), false for placeholder numbers.
  // Optional only for backward compatibility with pre-0.9.x literals.
  detected?: boolean
}

interface MetricsPool extends DatabasePool {
  getMetrics(): PoolMetrics
}

Shutdown Types

interface ShutdownOptions {
  timeout?: number // Default: 30000
  onShutdown?: () => void | Promise<void>
  logger?: KyseraLogger
}

interface RegisterShutdownOptions extends ShutdownOptions {
  signals?: NodeJS.Signals[] // Default: ['SIGTERM', 'SIGINT']
}

Best Practices

1. Always use health monitoring in production
2. Combine retry and circuit breaker for maximum resilience
3. Set appropriate timeouts matching your SLA
4. Monitor pool metrics for connection exhaustion
5. Register shutdown handlers early to prevent leaks