import { type ChildProcessWithoutNullStreams, spawn } from 'node:child_process'
export const WATCHDOG_PROTOCOL_VERSION = 1 as const
const MAX_WATCHDOG_BUFFER = 64 * 1024
const DEFAULT_WATCHDOG_EXECUTABLE = '/usr/local/bin/firehose-watchdog'
type MaybePromise<T> = T | PromiseLike<T>
export type WatchdogState = 'ready' | 'released'
export interface WatchdogStateMessage {
readonly version: typeof WATCHDOG_PROTOCOL_VERSION
readonly type: 'watchdog-state'
readonly state: WatchdogState
readonly pid: number
}
export type WatchdogCommand =
| {
readonly version: typeof WATCHDOG_PROTOCOL_VERSION
readonly type: 'hello'
readonly supervisorPid: number
}
| { readonly version: typeof WATCHDOG_PROTOCOL_VERSION; readonly type: 'release' }
export function encodeWatchdogMessage(message: WatchdogStateMessage | WatchdogCommand): string {
return `${JSON.stringify(message)}\n`
}
export function decodeWatchdogCommand(line: string): WatchdogCommand {
if (line.length > 4096) throw new Error('Watchdog command is too large')
let value: unknown
try {
value = JSON.parse(line)
} catch {
throw new Error('Watchdog command is not valid JSON')
}
if (!value || typeof value !== 'object') throw new Error('Watchdog command is not an object')
const command = value as Record<string, unknown>
if (command.version !== WATCHDOG_PROTOCOL_VERSION || (command.type !== 'hello' && command.type !== 'release')) {
throw new Error('Watchdog command has an invalid version or type')
}
if (command.type === 'hello') {
if (!Number.isSafeInteger(command.supervisorPid) || (command.supervisorPid as number) < 1) {
throw new Error('Watchdog hello has an invalid supervisor PID')
}
return {
version: WATCHDOG_PROTOCOL_VERSION,
type: 'hello',
supervisorPid: command.supervisorPid as number,
}
}
return { version: WATCHDOG_PROTOCOL_VERSION, type: 'release' }
}
export function decodeWatchdogState(line: string): WatchdogStateMessage {
if (line.length > 4096) throw new Error('Watchdog state is too large')
let value: unknown
try {
value = JSON.parse(line)
} catch {
throw new Error('Watchdog state is not valid JSON')
}
if (!value || typeof value !== 'object') throw new Error('Watchdog state is not an object')
const state = value as Record<string, unknown>
if (
state.version !== WATCHDOG_PROTOCOL_VERSION ||
state.type !== 'watchdog-state' ||
(state.state !== 'ready' && state.state !== 'released') ||
!Number.isSafeInteger(state.pid) ||
(state.pid as number) < 1
) {
throw new Error('Watchdog state has an invalid shape')
}
return {
version: WATCHDOG_PROTOCOL_VERSION,
type: 'watchdog-state',
state: state.state as WatchdogState,
pid: state.pid as number,
}
}
export type WorkerKiller = (pid: number) => void
/** The watchdog's only authority action when its supervisor pipe reaches EOF. */
export function killWorkerOnSupervisorLoss(
workerPid: number,
kill: WorkerKiller = (pid) => process.kill(pid, 'SIGKILL'),
): void {
if (!Number.isSafeInteger(workerPid) || workerPid < 1) return
try {
kill(workerPid)
} catch (error) {
if (error instanceof Error && 'code' in error && error.code === 'ESRCH') return
// A worker that cannot be killed is still unsafe. Let the watchdog exit
// nonzero; its container supervisor will restart the service.
}
}
export interface SupervisorWatchdog {
start(): Promise<void>
stop(): Promise<void>
/** Close the supervisor pipe. The watchdog treats EOF as a worker-fatal event. */
close(): MaybePromise<void>
onFailure?(handler: (error: Error) => void): void
}
export interface WatchdogProcessLike {
readonly stdin: {
write(data: string): boolean
end(): void
}
readonly stdout: {
setEncoding(encoding: string): unknown
on(event: string, listener: (...args: unknown[]) => void): unknown
}
on(event: string, listener: (...args: unknown[]) => void): unknown
}
export type WatchdogProcessSpawner = (
file: string,
args: readonly string[],
options: {
stdio: ['pipe', 'pipe', 'inherit']
env: NodeJS.ProcessEnv
},
) => WatchdogProcessLike
export interface ProcessWatchdogOptions {
readonly executable?: string
readonly workerPid: number
readonly supervisorPid?: number
readonly environment?: NodeJS.ProcessEnv
readonly spawn?: WatchdogProcessSpawner
}
/** Worker-killing watchdog process owned by the leadership supervisor. */
export class ProcessWatchdog implements SupervisorWatchdog {
private readonly executable: string
private readonly workerPid: number
private readonly supervisorPid: number
private readonly environment: NodeJS.ProcessEnv
private readonly spawnProcess: WatchdogProcessSpawner
private child: WatchdogProcessLike | null = null
private buffer = ''
private expectedStop = false
private started = false
private readyResolve: (() => void) | null = null
private readyReject: ((error: Error) => void) | null = null
private stopResolve: (() => void) | null = null
private stopReject: ((error: Error) => void) | null = null
private failureHandler: ((error: Error) => void) | null = null
constructor(options: ProcessWatchdogOptions) {
if (!Number.isSafeInteger(options.workerPid) || options.workerPid < 1)
throw new Error('Invalid watchdog worker PID')
this.executable = options.executable ?? process.env.FIREHOSE_WATCHDOG_PATH ?? DEFAULT_WATCHDOG_EXECUTABLE
this.workerPid = options.workerPid
this.supervisorPid = options.supervisorPid ?? process.pid
this.environment = options.environment ?? process.env
this.spawnProcess =
options.spawn ??
((file, args, spawnOptions) => spawn(file, args, spawnOptions) as unknown as ChildProcessWithoutNullStreams)
}
onFailure(handler: (error: Error) => void): void {
this.failureHandler = handler
}
start(): Promise<void> {
if (this.started) return Promise.resolve()
this.started = true
return new Promise<void>((resolve, reject) => {
this.readyResolve = resolve
this.readyReject = reject
let child: WatchdogProcessLike
try {
child = this.spawnProcess(
this.executable,
[`--worker-pid=${this.workerPid}`, `--supervisor-pid=${this.supervisorPid}`],
{
stdio: ['pipe', 'pipe', 'inherit'],
env: {
...this.environment,
WATCHDOG_SUPERVISOR_PID: String(this.supervisorPid),
},
},
)
} catch (error) {
this.started = false
reject(error instanceof Error ? error : new Error('Failed to start watchdog'))
return
}
this.child = child
child.stdout.setEncoding('utf8')
child.stdout.on('data', (chunk: unknown) => this.consumeOutput(String(chunk)))
child.stdout.on('error', () => this.fail(new Error('Watchdog state pipe failed')))
child.on('error', (error: unknown) => this.fail(error instanceof Error ? error : new Error('Watchdog failed')))
child.on('close', (code: unknown) => {
if (this.expectedStop) return
this.fail(new Error(`Watchdog exited unexpectedly (${String(code)})`))
})
try {
const accepted = child.stdin.write(
encodeWatchdogMessage({
version: WATCHDOG_PROTOCOL_VERSION,
type: 'hello',
supervisorPid: this.supervisorPid,
}),
)
if (!accepted) throw new Error('Watchdog command pipe backpressure')
} catch (error) {
this.fail(error instanceof Error ? error : new Error('Watchdog command pipe failed'))
reject(error instanceof Error ? error : new Error('Watchdog command pipe failed'))
}
})
}
stop(): Promise<void> {
if (!this.child || this.expectedStop) return Promise.resolve()
this.expectedStop = true
return new Promise<void>((resolve, reject) => {
this.stopResolve = resolve
this.stopReject = reject
try {
const accepted = this.child?.stdin.write(encodeWatchdogMessage({ version: 1, type: 'release' }))
if (!accepted) throw new Error('Watchdog command pipe backpressure')
} catch (error) {
reject(error instanceof Error ? error : new Error('Watchdog release failed'))
}
})
}
close(): void {
const child = this.child
if (!child) return
this.expectedStop = true
try {
child.stdin.end()
} catch {
// The supervisor is already on a fatal path. The watchdog may already
// have observed EOF and killed the worker.
}
}
private consumeOutput(chunk: string): void {
this.buffer += chunk
if (this.buffer.length > MAX_WATCHDOG_BUFFER) {
this.fail(new Error('Watchdog state exceeded limit'))
return
}
while (true) {
const newline = this.buffer.indexOf('\n')
if (newline < 0) return
const line = this.buffer.slice(0, newline).trim()
this.buffer = this.buffer.slice(newline + 1)
try {
const message = decodeWatchdogState(line)
if (message.state === 'ready') {
this.readyResolve?.()
this.readyResolve = null
this.readyReject = null
} else {
this.stopResolve?.()
this.stopResolve = null
this.stopReject = null
}
} catch {
this.fail(new Error('Watchdog sent malformed state'))
return
}
}
}
private fail(error: Error): void {
this.readyReject?.(error)
this.readyResolve = null
this.readyReject = null
this.stopReject?.(error)
this.stopResolve = null
this.stopReject = null
if (!this.expectedStop) this.failureHandler?.(error)
}
}
export function createProcessWatchdog(options: ProcessWatchdogOptions): ProcessWatchdog {
return new ProcessWatchdog(options)
}