import { createHash, randomUUID } from 'node:crypto'
import Redis from 'ioredis'
import postgres from 'postgres'
import type { SupervisorConfig } from './supervisor-config'
import type { SupervisorCommand, SupervisorState, SupervisorStateMessage } from './supervisor-protocol'
import { SUPERVISOR_PROTOCOL_VERSION } from './supervisor-protocol'
import { createProcessWatchdog, type SupervisorWatchdog } from './watchdog'
/** All candidates acquire authority in this order: Redis lease, then Postgres lock. */
export const SUPERVISOR_LOCK_ORDER = ['redis-lease', 'postgres-primary-advisory-lock'] as const
export const ACQUIRE_LEASE_SCRIPT = `
if redis.call('set', KEYS[1], ARGV[1], 'NX', 'PX', tonumber(ARGV[2])) then
return redis.call('incr', KEYS[2])
end
return 0
`
export const RENEW_LEASE_SCRIPT = `
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('set', KEYS[1], ARGV[1], 'XX', 'PX', tonumber(ARGV[2]))
end
return false
`
export const RELEASE_LEASE_SCRIPT = `
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
end
return 0
`
type MaybePromise<T> = T | PromiseLike<T>
export interface RedisLease {
acquire(token: string, ttlMs: number, deadlineAt: number): MaybePromise<number | null>
renew(token: string, ttlMs: number, deadlineAt: number): MaybePromise<boolean>
release(token: string, deadlineAt: number): MaybePromise<boolean>
close(): MaybePromise<void>
}
export interface PrimaryAdvisorySession {
/** Verify primary/read-write state, then take the session-level advisory lock. */
acquire(lockId: bigint, deadlineAt: number): MaybePromise<boolean>
/** Verify the same backend session, role, primary state, and held lock. */
verify(lockId: bigint, deadlineAt: number): MaybePromise<boolean>
release(lockId: bigint, deadlineAt: number): MaybePromise<void>
/** Drop an uncertain session so a partially acquired lock cannot survive. */
reset(): MaybePromise<void>
close(): MaybePromise<void>
}
export interface SupervisorDependencies {
readonly redis: RedisLease
readonly postgres: PrimaryAdvisorySession
readonly now?: () => number
readonly sleep?: (milliseconds: number) => Promise<void>
readonly killParent?: (pid: number) => void
readonly sendState?: (message: SupervisorStateMessage) => MaybePromise<void>
}
export interface LeadershipSupervisorOptions {
readonly parentPid: number
readonly redisLeaseKey: string
readonly redisEpochKey: string
readonly advisoryLockId: bigint
readonly leaseTtlMs: number
readonly renewIntervalMs: number
readonly pollIntervalMs: number
readonly commandTimeoutMs: number
readonly instanceId?: string
readonly watchdog?: SupervisorWatchdog
readonly dependencies: SupervisorDependencies
}
export interface SupervisorRunResult {
readonly exitCode: 0 | 1
readonly state: SupervisorState
readonly epoch?: number
}
export class SupervisorDeadlineError extends Error {
constructor(operation: string) {
super(`Supervisor ${operation} exceeded its deadline`)
this.name = 'SupervisorDeadlineError'
}
}
export class SupervisorInvariantError extends Error {
constructor(reason: string) {
super(`Supervisor invariant failed: ${reason}`)
this.name = 'SupervisorInvariantError'
}
}
function defaultNow(): number {
return typeof performance !== 'undefined' ? performance.now() : Date.now()
}
function defaultSleep(milliseconds: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, milliseconds))
}
function defaultKillParent(pid: number): void {
try {
process.kill(pid, 'SIGKILL')
} catch (error) {
// ESRCH means the parent has already gone away. Do not prevent authority
// cleanup; any other error is still represented by the fatal exit state.
if (error instanceof Error && 'code' in error && error.code === 'ESRCH') return
throw error
}
}
function errorKind(error: unknown): string {
return error instanceof Error && error.name ? error.name : 'UnknownError'
}
/** Bound an adapter operation even when a dependency ignores its deadline argument. */
async function withDeadline<T>(
operation: string,
deadlineAt: number,
now: () => number,
work: () => MaybePromise<T>,
): Promise<T> {
const remaining = deadlineAt - now()
if (remaining <= 0) throw new SupervisorDeadlineError(operation)
let timer: ReturnType<typeof setTimeout> | undefined
const timeout = new Promise<never>((_, reject) => {
timer = setTimeout(() => reject(new SupervisorDeadlineError(operation)), remaining)
})
try {
const result = await Promise.race([Promise.resolve().then(work), timeout])
if (now() > deadlineAt) throw new SupervisorDeadlineError(operation)
return result
} finally {
if (timer) clearTimeout(timer)
}
}
function boundedEpoch(value: unknown): number | null {
const epoch = typeof value === 'string' ? Number(value) : value
if (!Number.isSafeInteger(epoch) || (epoch as number) < 1) return null
return epoch as number
}
/**
* Dependency-injected supervisor state machine.
*
* The state machine deliberately does not run in the firehose worker. Its fatal path kills the
* parent before releasing either authority, so a blocked parent callback cannot
* continue firehose writes after another instance takes over.
*/
export class LeadershipSupervisor {
private readonly parentPid: number
private readonly redisLeaseKey: string
private readonly redisEpochKey: string
private readonly advisoryLockId: bigint
private readonly leaseTtlMs: number
private readonly renewIntervalMs: number
private readonly pollIntervalMs: number
private readonly commandTimeoutMs: number
private readonly instanceId: string
private readonly dependencies: SupervisorDependencies
private readonly now: () => number
private readonly sleep: (milliseconds: number) => Promise<void>
private readonly killParent: (pid: number) => void
private readonly sendState: (message: SupervisorStateMessage) => MaybePromise<void>
private readonly watchdog: SupervisorWatchdog | undefined
private state: SupervisorState = 'standby'
private epoch: number | undefined
private lastEpoch = 0
private hasRedisLease = false
private hasPostgresLock = false
private releaseRequested = false
private parentHandshakeReceived = false
private terminal = false
private fatalInFlight: Promise<void> | null = null
private wakeResolver: (() => void) | null = null
private watchdogStarted = false
private announcedStandby = false
constructor(options: LeadershipSupervisorOptions) {
if (!Number.isSafeInteger(options.parentPid) || options.parentPid < 1) {
throw new SupervisorInvariantError('invalid parent PID')
}
if (options.leaseTtlMs <= 0 || options.renewIntervalMs <= 0 || options.pollIntervalMs <= 0) {
throw new SupervisorInvariantError('invalid supervisor intervals')
}
if (options.commandTimeoutMs <= 0 || options.commandTimeoutMs >= options.leaseTtlMs) {
throw new SupervisorInvariantError('command deadline must be below lease TTL')
}
this.parentPid = options.parentPid
this.redisLeaseKey = options.redisLeaseKey
this.redisEpochKey = options.redisEpochKey
this.advisoryLockId = options.advisoryLockId
this.leaseTtlMs = options.leaseTtlMs
this.renewIntervalMs = options.renewIntervalMs
this.pollIntervalMs = options.pollIntervalMs
this.commandTimeoutMs = options.commandTimeoutMs
this.instanceId = options.instanceId ?? randomUUID()
this.dependencies = options.dependencies
this.now = options.dependencies.now ?? defaultNow
this.sleep = options.dependencies.sleep ?? defaultSleep
this.killParent = options.dependencies.killParent ?? defaultKillParent
this.sendState = options.dependencies.sendState ?? (() => undefined)
this.watchdog = options.watchdog
this.watchdog?.onFailure?.(() => {
void this.fail('watchdog process exited or lost its pipe')
})
}
get currentState(): SupervisorState {
return this.state
}
get currentEpoch(): number | undefined {
return this.epoch
}
get ownsAuthority(): boolean {
return this.hasRedisLease && this.hasPostgresLock
}
get id(): string {
return this.instanceId
}
/** Receive a validated parent command. It never executes worker callbacks. */
async handleCommand(command: SupervisorCommand): Promise<void> {
if (this.terminal) return
if (command.type === 'hello') {
if (command.parentPid !== this.parentPid) {
await this.fail('parent PID changed')
return
}
this.parentHandshakeReceived = true
return
}
if (!this.parentHandshakeReceived) {
await this.fail('parent release arrived before handshake')
return
}
this.releaseRequested = true
this.wake()
}
/** Execute one independent supervisor tick. Useful for deterministic tests. */
async tick(): Promise<void> {
if (this.terminal) return
if (this.releaseRequested) {
await this.releaseVoluntarily()
return
}
if (this.ownsAuthority) {
await this.renewAuthority()
return
}
await this.tryAcquire()
}
/** Run until a voluntary release or a fatal authority/IPC failure. */
async run(): Promise<SupervisorRunResult> {
if (this.watchdog) {
try {
this.watchdogStarted = true
const deadline = this.now() + this.commandTimeoutMs
await withDeadline('Watchdog start', deadline, this.now, () => this.watchdog?.start())
} catch (error) {
await this.fail(`watchdog startup failed (${errorKind(error)})`)
}
}
if (this.terminal) {
return {
exitCode: 1,
state: this.state,
...(this.epoch === undefined ? {} : { epoch: this.epoch }),
}
}
try {
await this.announce('standby')
} catch (error) {
await this.fail(`initial state announcement failed (${errorKind(error)})`)
}
while (!this.terminal) {
try {
await this.tick()
} catch (error) {
await this.fail(`supervisor tick failed (${errorKind(error)})`)
}
if (this.terminal) break
try {
await this.wait(this.ownsAuthority ? this.renewIntervalMs : this.pollIntervalMs)
} catch (error) {
await this.fail(`supervisor timer failed (${errorKind(error)})`)
}
}
if (this.fatalInFlight) await this.fatalInFlight
return {
exitCode: this.state === 'released' ? 0 : 1,
state: this.state,
...(this.epoch === undefined ? {} : { epoch: this.epoch }),
}
}
/** Fail closed for protocol, timer, or adapter invariants detected by the CLI. */
async failClosed(reason: string): Promise<void> {
await this.fail(reason)
}
private async tryAcquire(): Promise<void> {
const leaseDeadline = this.now() + this.commandTimeoutMs
let epochValue: number | null
try {
epochValue = await withDeadline('Redis acquire', leaseDeadline, this.now, () =>
this.dependencies.redis.acquire(this.instanceId, this.leaseTtlMs, leaseDeadline),
)
} catch (error) {
if (error instanceof SupervisorInvariantError) {
await this.fail(error.message)
return
}
// No authority was obtained. Keep the worker a healthy standby while Redis
// recovers; a lease command failure cannot permit worker startup.
await this.announceStandby('redis-unavailable')
return
}
if (epochValue === null) {
await this.announceStandby()
return
}
if (epochValue === undefined) {
await this.fail('Redis returned no acquisition result')
return
}
const epoch = boundedEpoch(epochValue)
if (epoch === null) {
await this.fail('Redis returned an invalid epoch')
return
}
this.hasRedisLease = true
let acquired = false
try {
const postgresDeadline = this.now() + this.commandTimeoutMs
const acquisitionResult = await withDeadline('Postgres acquire', postgresDeadline, this.now, () =>
this.dependencies.postgres.acquire(this.advisoryLockId, postgresDeadline),
)
if (typeof acquisitionResult !== 'boolean')
throw new SupervisorInvariantError('Postgres returned an invalid acquisition result')
acquired = acquisitionResult
} catch {
await this.discardPartialAuthority()
await this.announceStandby('postgres-unavailable')
return
}
if (!acquired) {
await this.discardPartialAuthority()
await this.announceStandby('postgres-not-primary-or-lock-held')
return
}
this.hasPostgresLock = true
if (epoch <= this.lastEpoch) {
await this.fail('Redis epoch did not increase')
return
}
this.lastEpoch = epoch
this.epoch = epoch
await this.announce('acquired')
}
private async renewAuthority(): Promise<void> {
if (!this.ownsAuthority || this.epoch === undefined) {
await this.fail('authority flags are inconsistent')
return
}
const leaseDeadline = this.now() + this.commandTimeoutMs
let renewed = false
try {
renewed = await withDeadline('Redis renew', leaseDeadline, this.now, () =>
this.dependencies.redis.renew(this.instanceId, this.leaseTtlMs, leaseDeadline),
)
} catch {
await this.fail('Redis lease deadline or connection failure')
return
}
if (!renewed) {
await this.fail('Redis lease lost')
return
}
const postgresDeadline = this.now() + this.commandTimeoutMs
let valid = false
try {
valid = await withDeadline('Postgres verify', postgresDeadline, this.now, () =>
this.dependencies.postgres.verify(this.advisoryLockId, postgresDeadline),
)
} catch {
await this.fail('Postgres session deadline or connection failure')
return
}
if (!valid) await this.fail('Postgres primary, role, session, or advisory lock lost')
}
private async discardPartialAuthority(): Promise<void> {
// If the PG command timed out after acquiring, reset first so a hidden
// session-level lock cannot outlive the Redis lease we are about to drop.
try {
await withDeadline('Postgres reset', this.now() + this.commandTimeoutMs, this.now, () =>
this.dependencies.postgres.reset(),
)
} catch {
// reset() is a best effort before no authority has been advertised.
}
this.hasPostgresLock = false
if (this.hasRedisLease) {
try {
await withDeadline('Redis release after failed acquire', this.now() + this.commandTimeoutMs, this.now, () =>
this.dependencies.redis.release(this.instanceId, this.now() + this.commandTimeoutMs),
)
} catch {
// The lease will expire. No worker has started, so this is still a
// safe standby outcome.
}
this.hasRedisLease = false
}
}
private async releaseVoluntarily(): Promise<void> {
if (this.terminal) return
// Confirm both authorities immediately before the reverse-order release.
// This turns a lease that expired between ticks into the fatal kill path,
// rather than discovering it only after dropping the PG lock.
if (this.ownsAuthority) {
await this.renewAuthority()
if (this.terminal) return
}
this.terminal = true
if (this.ownsAuthority) {
try {
await this.announce('releasing')
await this.releaseAuthority()
await this.closeDependencies(true)
this.state = 'released'
await this.announce('released')
} catch (error) {
// The parent has already drained. Still fail closed if authority
// cannot be proven released.
this.terminal = false
await this.fail(`graceful release failed (${errorKind(error)})`)
}
return
}
try {
// A standby has no worker activity, but its watchdog still needs an
// explicit release acknowledgement. Closing the pipe would treat normal
// shutdown as supervisor loss and SIGKILL the parent.
await this.closeDependencies(true)
this.state = 'released'
await this.announce('released')
} catch {
this.terminal = false
await this.fail('standby close failed')
}
}
private async releaseAuthority(): Promise<void> {
let firstError: unknown
if (this.hasPostgresLock) {
try {
const deadline = this.now() + this.commandTimeoutMs
await withDeadline('Postgres release', deadline, this.now, () =>
this.dependencies.postgres.release(this.advisoryLockId, deadline),
)
} catch (error) {
firstError = error
} finally {
this.hasPostgresLock = false
}
}
if (this.hasRedisLease) {
try {
const deadline = this.now() + this.commandTimeoutMs
const released = await withDeadline('Redis release', deadline, this.now, () =>
this.dependencies.redis.release(this.instanceId, deadline),
)
if (released === false) throw new SupervisorInvariantError('Redis release was not confirmed')
} catch (error) {
firstError ??= error
} finally {
this.hasRedisLease = false
}
}
if (firstError) throw firstError
}
private async closeDependencies(graceful = false): Promise<void> {
await Promise.all([this.dependencies.redis.close(), this.dependencies.postgres.close()])
if (!this.watchdog || !this.watchdogStarted) return
if (graceful) {
const deadline = this.now() + this.commandTimeoutMs
await withDeadline('Watchdog release', deadline, this.now, () => this.watchdog?.stop())
this.watchdogStarted = false
return
}
this.watchdogStarted = false
// EOF is intentional here: the watchdog kills the worker if the fatal kill above
// was interrupted or raced an OS process failure.
await this.watchdog.close()
}
private async fail(reason: string): Promise<void> {
if (this.fatalInFlight) return await this.fatalInFlight
this.wake()
this.fatalInFlight = (async () => {
this.terminal = true
this.state = 'fatal'
try {
await this.announce('fatal', reason)
} catch {
// A broken output pipe is itself an IPC loss. Killing the worker below is
// still mandatory even when the diagnostic state cannot be written.
}
// This order is a safety invariant: kill the worker before releasing either
// authority, regardless of whether the kill syscall reports ESRCH.
try {
this.killParent(this.parentPid)
} catch {
// Continue cleanup and return a nonzero result from the CLI.
}
try {
await this.releaseAuthority()
} catch {
// A lost authority is intentionally not allowed to block supervisor exit.
}
try {
await this.closeDependencies()
} catch {
// Exit remains nonzero.
}
})()
await this.fatalInFlight
}
private async announce(state: SupervisorState, reason?: string): Promise<void> {
this.state = state
if (state === 'standby') this.announcedStandby = true
const message: SupervisorStateMessage = {
version: SUPERVISOR_PROTOCOL_VERSION,
type: 'state',
state,
pid: process.pid,
...(this.epoch === undefined ? {} : { epoch: this.epoch }),
...(reason ? { reason } : {}),
}
await this.sendState(message)
}
private async announceStandby(reason?: string): Promise<void> {
if (this.state === 'standby' && this.announcedStandby && !reason) return
await this.announce('standby', reason)
}
private wake(): void {
const resolve = this.wakeResolver
this.wakeResolver = null
resolve?.()
}
private async wait(milliseconds: number): Promise<void> {
if (this.terminal || this.releaseRequested) return
const wake = new Promise<void>((resolve) => {
this.wakeResolver = resolve
})
try {
await Promise.race([this.sleep(milliseconds), wake])
} finally {
this.wakeResolver = null
}
}
}
/** Construct production adapters. The supervisor binary is the only caller. */
export function createRedisLease(
config: Pick<SupervisorConfig, 'redisUrl' | 'redisLeaseKey' | 'redisEpochKey' | 'commandTimeoutMs'>,
): RedisLease {
const client = new Redis(config.redisUrl, {
lazyConnect: true,
maxRetriesPerRequest: 0,
enableReadyCheck: true,
enableOfflineQueue: false,
connectTimeout: config.commandTimeoutMs,
})
return {
acquire: async (token, ttlMs) => {
const result = await client.eval(
ACQUIRE_LEASE_SCRIPT,
2,
config.redisLeaseKey,
config.redisEpochKey,
token,
String(ttlMs),
)
if (result === 0 || result === '0' || result === null) return null
const epoch = boundedEpoch(result)
if (epoch === null) throw new SupervisorInvariantError('Redis returned an invalid epoch')
return epoch
},
renew: async (token, ttlMs) => {
const result = await client.eval(RENEW_LEASE_SCRIPT, 1, config.redisLeaseKey, token, String(ttlMs))
return result === 'OK'
},
release: async (token) => {
const result = await client.eval(RELEASE_LEASE_SCRIPT, 1, config.redisLeaseKey, token)
return result === 1 || result === '1'
},
close: () => {
client.disconnect()
},
}
}
export function createPrimaryAdvisorySession(
config: Pick<SupervisorConfig, 'databaseUrl' | 'commandTimeoutMs'>,
): PrimaryAdvisorySession {
let client = postgres(config.databaseUrl, {
max: 1,
idle_timeout: 0,
max_lifetime: null,
connect_timeout: Math.max(1, Math.ceil(config.commandTimeoutMs / 1000)),
})
let closed = false
let backendPid: number | undefined
let roleName: string | undefined
let held = false
return {
acquire: async (lockId) => {
const primary = await client<
Array<{
in_recovery: boolean
transaction_read_only: string
backend_pid: number
role_name: string
role_valid: boolean
}>
>`
SELECT pg_is_in_recovery() AS in_recovery,
current_setting('transaction_read_only') AS transaction_read_only,
pg_backend_pid() AS backend_pid,
current_user AS role_name,
has_database_privilege(current_user, current_database(), 'CONNECT') AS role_valid
`
const row = primary[0]
if (!row || row.in_recovery || row.transaction_read_only !== 'off' || !row.role_valid) return false
const lockResult = await client.unsafe<Array<{ locked: boolean }>>(
`SELECT pg_try_advisory_lock(${lockId.toString()}::bigint) AS locked`,
)
if (!lockResult[0]?.locked) return false
backendPid = Number(row.backend_pid)
roleName = row.role_name
held = true
return true
},
verify: async (lockId) => {
if (!held || backendPid === undefined || roleName === undefined) return false
const lockNumber = lockId.toString()
const rows = await client.unsafe<
Array<{
in_recovery: boolean
transaction_read_only: string
backend_pid: number
role_name: string
role_valid: boolean
lock_held: boolean
}>
>(
`SELECT pg_is_in_recovery() AS in_recovery,
current_setting('transaction_read_only') AS transaction_read_only,
pg_backend_pid() AS backend_pid,
current_user AS role_name,
has_database_privilege(current_user, current_database(), 'CONNECT') AS role_valid,
EXISTS (
SELECT 1 FROM pg_locks
WHERE pid = pg_backend_pid()
AND locktype = 'advisory'
AND granted
-- bigint advisory locks use objsubid=1; objsubid=2 is
-- reserved for the two-int-key API.
AND objsubid = 1
AND ((classid::bigint << 32) | objid::bigint) = ${lockNumber}::bigint
) AS lock_held`,
)
const row = rows[0]
return Boolean(
row &&
!row.in_recovery &&
row.transaction_read_only === 'off' &&
Number(row.backend_pid) === backendPid &&
row.role_name === roleName &&
row.role_valid &&
row.lock_held,
)
},
release: async (lockId) => {
if (!held) return
const result = await client.unsafe<Array<{ unlocked: boolean }>>(
`SELECT pg_advisory_unlock(${lockId.toString()}::bigint) AS unlocked`,
)
if (!result[0]?.unlocked) throw new Error('Postgres advisory unlock was not confirmed')
held = false
},
reset: async () => {
held = false
backendPid = undefined
roleName = undefined
await client.end({ timeout: 0 })
if (!closed) {
client = postgres(config.databaseUrl, {
max: 1,
idle_timeout: 0,
max_lifetime: null,
connect_timeout: Math.max(1, Math.ceil(config.commandTimeoutMs / 1000)),
})
}
},
close: async () => {
closed = true
await client.end({ timeout: config.commandTimeoutMs / 1000 })
},
}
}
export function createProductionSupervisor(
config: SupervisorConfig,
sendState: (message: SupervisorStateMessage) => MaybePromise<void>,
): LeadershipSupervisor {
const watchdog = createProcessWatchdog({
workerPid: config.parentPid,
executable: config.watchdogPath,
})
return new LeadershipSupervisor({
parentPid: config.parentPid,
redisLeaseKey: config.redisLeaseKey,
redisEpochKey: config.redisEpochKey,
advisoryLockId: config.advisoryLockId,
leaseTtlMs: config.leaseTtlMs,
renewIntervalMs: config.renewIntervalMs,
pollIntervalMs: config.pollIntervalMs,
commandTimeoutMs: config.commandTimeoutMs,
watchdog,
dependencies: {
redis: createRedisLease(config),
postgres: createPrimaryAdvisorySession(config),
sendState,
},
})
}
export function advisoryLockIdForName(name: string): bigint {
const digest = createHash('sha256').update(name).digest('hex')
return BigInt(`0x${digest.slice(0, 16)}`) & 0x7fffffffffffffffn
}