diff --git a/deploy/self-hosting/docker-compose.yml b/deploy/self-hosting/docker-compose.yml index abbc953f6..f3f90c92b 100644 --- a/deploy/self-hosting/docker-compose.yml +++ b/deploy/self-hosting/docker-compose.yml @@ -345,6 +345,13 @@ services: FLUXER_API_WORKER_ENABLE_CRON_SCHEDULER: "true" FLUXER_API_WORKER_ENABLE_VOICE_RECONCILIATION: "true" FLUXER_POSTGRES_MAX_CONNECTIONS: "25" + healthcheck: + test: ["CMD", "node", "-e", "const age=Date.now()-require('node:fs').statSync('/tmp/fluxer-worker-heartbeat').mtimeMs;if(age>30000){console.error('worker heartbeat is '+Math.round(age)+'ms old');process.exit(1)}"] + interval: 10s + timeout: 5s + retries: 3 + start_period: 90s + start_interval: 1s depends_on: postgres: {condition: service_healthy} valkey: {condition: service_healthy} diff --git a/fluxer_api/src/api/database/TransientDatabaseError.ts b/fluxer_api/src/api/database/TransientDatabaseError.ts new file mode 100644 index 000000000..a55124cef --- /dev/null +++ b/fluxer_api/src/api/database/TransientDatabaseError.ts @@ -0,0 +1,99 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +const TRANSIENT_POSTGRES_SQLSTATES: ReadonlySet = new Set([ + '08000', + '08001', + '08003', + '08004', + '08006', + '08007', + '57P01', + '57P02', + '57P03', +]); + +const TRANSIENT_SOCKET_CODES: ReadonlySet = new Set([ + 'EAI_AGAIN', + 'ECONNABORTED', + 'ECONNREFUSED', + 'ECONNRESET', + 'EHOSTUNREACH', + 'ENETDOWN', + 'ENETUNREACH', + 'ENOTFOUND', + 'EPIPE', + 'ETIMEDOUT', +]); + +const TRANSIENT_DRIVER_MESSAGES: ReadonlySet = new Set([ + 'Client has encountered a connection error and is not queryable', + 'Client was closed and is not queryable', + 'Connection terminated', + 'Connection terminated due to connection timeout', + 'Connection terminated unexpectedly', + 'timeout exceeded when trying to connect', +]); + +const MAX_ERROR_CHAIN_DEPTH = 8; + +type ErrorNode = Record; + +function readString(value: unknown): string | null { + return typeof value === 'string' ? value : null; +} + +function collectErrorChain(error: unknown): Array { + const nodes: Array = []; + const queue: Array = [error]; + const seen = new Set(); + while (queue.length > 0 && nodes.length < MAX_ERROR_CHAIN_DEPTH) { + const current = queue.shift(); + if (typeof current !== 'object' || current === null || seen.has(current)) { + continue; + } + seen.add(current); + const node = current as ErrorNode; + nodes.push(node); + if ('cause' in node) { + queue.push(node['cause']); + } + const aggregated = node['errors']; + if (Array.isArray(aggregated)) { + for (const nested of aggregated) { + queue.push(nested); + } + } + } + return nodes; +} + +function carriesPostgresClient(node: ErrorNode): boolean { + const client = node['client']; + return typeof client === 'object' && client !== null; +} + +function hasTransientSqlState(node: ErrorNode): boolean { + const code = readString(node['code']); + return code !== null && TRANSIENT_POSTGRES_SQLSTATES.has(code); +} + +function hasTransientSocketCode(node: ErrorNode): boolean { + const code = readString(node['code']); + return code !== null && TRANSIENT_SOCKET_CODES.has(code); +} + +function hasTransientDriverMessage(node: ErrorNode): boolean { + const message = readString(node['message']); + return message !== null && TRANSIENT_DRIVER_MESSAGES.has(message); +} + +export function isTransientDatabaseError(error: unknown): boolean { + const nodes = collectErrorChain(error); + if (nodes.some(hasTransientSqlState)) { + return true; + } + if (nodes.some(carriesPostgresClient) && nodes.some(hasTransientSocketCode)) { + return true; + } + return nodes.some(hasTransientDriverMessage); +} diff --git a/fluxer_api/src/api/worker/CronScheduler.ts b/fluxer_api/src/api/worker/CronScheduler.ts index 261334741..d47b1d924 100644 --- a/fluxer_api/src/api/worker/CronScheduler.ts +++ b/fluxer_api/src/api/worker/CronScheduler.ts @@ -3,6 +3,7 @@ import type {LoggerInterface} from '@fluxer/logger/src/LoggerInterface'; import type {IKVProvider} from '@pkgs/kv_client/src/IKVProvider'; import type {WorkerJobPayload} from '@pkgs/worker/src/contracts/WorkerTypes'; +import {WORKER_CRON_STALE_AFTER_MS, type WorkerHeartbeat, type WorkerHeartbeatSignal} from './WorkerHeartbeat'; import type {WorkerTaskName} from './WorkerLaneConfig'; import type {WorkerService} from './WorkerService'; @@ -92,14 +93,22 @@ export class CronScheduler { private readonly workerService: WorkerService; private readonly logger: LoggerInterface; private readonly kvClient: IKVProvider | null; + private readonly heartbeat: WorkerHeartbeat | null; private readonly definitions: Map = new Map(); private intervalId: NodeJS.Timeout | null = null; private lastTickSecond: number | null = null; + private heartbeatSignal: WorkerHeartbeatSignal | null = null; - constructor(workerService: WorkerService, logger: LoggerInterface, kvClient: IKVProvider | null = null) { + constructor( + workerService: WorkerService, + logger: LoggerInterface, + kvClient: IKVProvider | null = null, + heartbeat: WorkerHeartbeat | null = null, + ) { this.workerService = workerService; this.logger = logger; this.kvClient = kvClient; + this.heartbeat = heartbeat; } upsert( @@ -123,6 +132,7 @@ export class CronScheduler { if (this.intervalId !== null) { return; } + this.heartbeatSignal = this.heartbeat?.register('cron', WORKER_CRON_STALE_AFTER_MS) ?? null; this.intervalId = setInterval(() => { this.tick().catch((error) => { this.logger.error({err: error}, 'Cron scheduler tick failed'); @@ -136,10 +146,20 @@ export class CronScheduler { clearInterval(this.intervalId); this.intervalId = null; } + this.heartbeatSignal?.release(); + this.heartbeatSignal = null; this.lastTickSecond = null; } private async tick(): Promise { + try { + await this.runDueDefinitions(); + } finally { + this.heartbeatSignal?.report(); + } + } + + private async runDueDefinitions(): Promise { const nowSeconds = Math.floor(Date.now() / 1000); const previousTickSecond = this.lastTickSecond; this.lastTickSecond = nowSeconds; diff --git a/fluxer_api/src/api/worker/WorkerHeartbeat.ts b/fluxer_api/src/api/worker/WorkerHeartbeat.ts new file mode 100644 index 000000000..3b13b54d8 --- /dev/null +++ b/fluxer_api/src/api/worker/WorkerHeartbeat.ts @@ -0,0 +1,135 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {writeFileSync} from 'node:fs'; +import type {ILogger} from '../ILogger'; + +export const WORKER_HEARTBEAT_PATH = '/tmp/fluxer-worker-heartbeat'; +export const WORKER_HEARTBEAT_WRITE_INTERVAL_MS = 5000; +export const WORKER_HEARTBEAT_MAX_AGE_MS = 30000; +export const WORKER_LANE_HEARTBEAT_INTERVAL_MS = 5000; +export const WORKER_LANE_STALE_AFTER_MS = 30000; +export const WORKER_CRON_STALE_AFTER_MS = 90000; + +export interface WorkerHeartbeatSignal { + report(): void; + release(): void; +} + +interface WorkerHeartbeatOptions { + logger: Pick; + path?: string; + intervalMs?: number; + now?: () => number; + write?: (path: string, contents: string) => void; +} + +interface WorkerHeartbeatComponent { + staleAfterMs: number; + lastReportedAt: number; +} + +export class WorkerHeartbeat { + private readonly logger: Pick; + private readonly path: string; + private readonly intervalMs: number; + private readonly now: () => number; + private readonly write: (path: string, contents: string) => void; + private readonly components = new Map(); + private intervalId: NodeJS.Timeout | null = null; + private stalled = false; + + constructor(options: WorkerHeartbeatOptions) { + this.logger = options.logger; + this.path = options.path ?? WORKER_HEARTBEAT_PATH; + this.intervalMs = options.intervalMs ?? WORKER_HEARTBEAT_WRITE_INTERVAL_MS; + this.now = options.now ?? Date.now; + this.write = options.write ?? ((path, contents) => writeFileSync(path, contents)); + } + + getPath(): string { + return this.path; + } + + register(name: string, staleAfterMs: number): WorkerHeartbeatSignal { + const component: WorkerHeartbeatComponent = {staleAfterMs, lastReportedAt: this.now()}; + this.components.set(name, component); + return { + report: () => { + component.lastReportedAt = this.now(); + }, + release: () => { + if (this.components.get(name) === component) { + this.components.delete(name); + } + }, + }; + } + + stalledComponents(): Array { + const at = this.now(); + const stalled: Array = []; + for (const [name, component] of this.components) { + if (at - component.lastReportedAt > component.staleAfterMs) { + stalled.push(name); + } + } + return stalled; + } + + writeOnce(): boolean { + const stalled = this.stalledComponents(); + if (stalled.length > 0) { + if (!this.stalled) { + this.stalled = true; + this.logger.error( + {components: stalled, path: this.path}, + 'Worker heartbeat stalled, the container will report unhealthy', + ); + } + return false; + } + try { + this.write(this.path, this.snapshot()); + } catch (error) { + this.logger.error({err: error, path: this.path}, 'Failed to write the worker heartbeat file'); + return false; + } + if (this.stalled) { + this.stalled = false; + this.logger.info({path: this.path}, 'Worker heartbeat recovered'); + } + return true; + } + + start(): void { + if (this.intervalId !== null) { + return; + } + this.writeOnce(); + this.intervalId = setInterval(() => { + this.writeOnce(); + }, this.intervalMs); + this.logger.info( + {path: this.path, intervalMs: this.intervalMs, components: [...this.components.keys()]}, + 'Worker heartbeat started', + ); + } + + stop(): void { + if (this.intervalId !== null) { + clearInterval(this.intervalId); + this.intervalId = null; + } + } + + private snapshot(): string { + const at = this.now(); + return JSON.stringify({ + at: new Date(at).toISOString(), + components: [...this.components].map(([name, component]) => ({ + name, + ageMs: at - component.lastReportedAt, + })), + }); + } +} diff --git a/fluxer_api/src/api/worker/WorkerMain.ts b/fluxer_api/src/api/worker/WorkerMain.ts index 2173623e3..f6fba54dd 100644 --- a/fluxer_api/src/api/worker/WorkerMain.ts +++ b/fluxer_api/src/api/worker/WorkerMain.ts @@ -23,12 +23,14 @@ import {CronScheduler} from './CronScheduler'; import {JetStreamWorkerQueue} from './JetStreamWorkerQueue'; import {clearWorkerDependencies, setWorkerDependencies} from './WorkerContext'; import {initializeWorkerDependencies, shutdownWorkerDependencies, type WorkerDependencies} from './WorkerDependencies'; +import {WorkerHeartbeat} from './WorkerHeartbeat'; import { resolveCronSchedulerEnabled, resolveWorkerLanes, validateLaneCompleteness, type WorkerLaneDefinition, } from './WorkerLaneConfig'; +import {createWorkerProcessErrorHandler} from './WorkerProcessErrorHandler'; import {WorkerQueueOverflowError} from './WorkerQueueOverflowError'; import {WorkerRunner} from './WorkerRunner'; import {WorkerService} from './WorkerService'; @@ -88,6 +90,7 @@ export async function startWorkerMain(): Promise { let snowflakeService: ISnowflakeService | null = null; let dependencies: WorkerDependencies | null = null; let cron: CronScheduler | null = null; + const heartbeat = new WorkerHeartbeat({logger: Logger}); const runners: Array = []; let searchInitialized = false; let shuttingDown = false; @@ -106,6 +109,7 @@ export async function startWorkerMain(): Promise { } shuttingDown = true; Logger.info('Shutting down worker backend...'); + await cleanupStep('heartbeat', () => heartbeat.stop()); await cleanupStep('cron', () => cron?.stop()); await cleanupStep('runners', async () => { await Promise.all(runners.map((runner) => runner.stop())); @@ -228,7 +232,7 @@ export async function startWorkerMain(): Promise { } } } - cron = new CronScheduler(workerService, Logger, dependencies.kvClient); + cron = new CronScheduler(workerService, Logger, dependencies.kvClient, heartbeat); registerCronJobs(cron); for (const lane of activeWorkerLanes) { const laneTasks: Record = {}; @@ -248,6 +252,7 @@ export async function startWorkerMain(): Promise { concurrency: lane.concurrency, maxDeliver: lane.maxDeliver, ackWaitMs: lane.ackWaitMs, + heartbeat, }); runners.push(runner); } @@ -278,18 +283,20 @@ export async function startWorkerMain(): Promise { {lanes: activeWorkerLanes.map((l) => `${l.name}(${l.concurrency})`).join(', ')}, 'Worker runners started', ); + heartbeat.start(); setupGracefulShutdown(shutdown, {logger: Logger, timeoutMs: 30000}); - process.on('uncaughtException', async (error) => { - Logger.error({err: error}, 'Uncaught Exception'); - setTimeout(() => process.exit(1), ms('5 seconds')).unref(); - await shutdown(); - process.exit(1); + const handleProcessError = createWorkerProcessErrorHandler({ + logger: Logger, + shutdown, + exit: (code) => { + process.exit(code); + }, }); - process.on('unhandledRejection', async (reason: unknown) => { - Logger.error({err: reason}, 'Unhandled Rejection at Promise'); - setTimeout(() => process.exit(1), ms('5 seconds')).unref(); - await shutdown(); - process.exit(1); + process.on('uncaughtException', (error) => { + void handleProcessError('uncaughtException', error); + }); + process.on('unhandledRejection', (reason: unknown) => { + void handleProcessError('unhandledRejection', reason); }); } catch (error: unknown) { Logger.error({err: error}, 'Failed to start worker backend'); diff --git a/fluxer_api/src/api/worker/WorkerProcessErrorHandler.ts b/fluxer_api/src/api/worker/WorkerProcessErrorHandler.ts new file mode 100644 index 000000000..e64837d8a --- /dev/null +++ b/fluxer_api/src/api/worker/WorkerProcessErrorHandler.ts @@ -0,0 +1,44 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {ms} from 'itty-time'; +import {isTransientDatabaseError} from '../database/TransientDatabaseError'; +import type {ILogger} from '../ILogger'; + +export type WorkerProcessErrorSource = 'uncaughtException' | 'unhandledRejection'; + +export type WorkerProcessErrorHandler = (source: WorkerProcessErrorSource, error: unknown) => Promise; + +interface WorkerProcessErrorHandlerOptions { + logger: Pick; + shutdown: () => Promise; + exit: (code: number) => void; + forceExitDelayMs?: number; +} + +const FATAL_MESSAGE: Record = { + uncaughtException: 'Uncaught Exception', + unhandledRejection: 'Unhandled Rejection at Promise', +}; + +export function createWorkerProcessErrorHandler(options: WorkerProcessErrorHandlerOptions): WorkerProcessErrorHandler { + const forceExitDelayMs = options.forceExitDelayMs ?? ms('5 seconds'); + return async (source, error) => { + if (isTransientDatabaseError(error)) { + options.logger.warn( + {err: error, source}, + 'Transient database connection error reached the worker process, keeping the worker running', + ); + return; + } + options.logger.error({err: error, source}, FATAL_MESSAGE[source]); + const forceExit = setTimeout(() => options.exit(1), forceExitDelayMs); + forceExit.unref(); + try { + await options.shutdown(); + } catch (shutdownError) { + options.logger.error({err: shutdownError, source}, 'Worker shutdown failed while handling a fatal error'); + } + clearTimeout(forceExit); + options.exit(1); + }; +} diff --git a/fluxer_api/src/api/worker/WorkerRunner.ts b/fluxer_api/src/api/worker/WorkerRunner.ts index a0e25addf..ea8076044 100644 --- a/fluxer_api/src/api/worker/WorkerRunner.ts +++ b/fluxer_api/src/api/worker/WorkerRunner.ts @@ -8,6 +8,12 @@ import type {IJobLedgerRepository} from '../jobs/IJobLedgerRepository'; import {Logger} from '../Logger'; import {getWorkerService} from '../middleware/ServiceRegistry'; import {isJsonRecord, parseJsonRecord} from '../utils/JsonBoundaryUtils'; +import { + WORKER_LANE_HEARTBEAT_INTERVAL_MS, + WORKER_LANE_STALE_AFTER_MS, + type WorkerHeartbeat, + type WorkerHeartbeatSignal, +} from './WorkerHeartbeat'; const MAX_DLQ_PUBLISH_ATTEMPTS = 3; const MIN_ACK_HEARTBEAT_MS = 1000; @@ -54,6 +60,7 @@ interface WorkerRunnerOptions { concurrency?: number; maxDeliver?: number; ackWaitMs?: number; + heartbeat?: WorkerHeartbeat; } export class WorkerRunner { @@ -68,6 +75,9 @@ export class WorkerRunner { private readonly ackWaitMs: number; private readonly workerService: IWorkerService; private readonly ledger: IJobLedgerRepository; + private readonly heartbeat: WorkerHeartbeat | null; + private heartbeatSignal: WorkerHeartbeatSignal | null = null; + private heartbeatTimer: ReturnType | null = null; private running = false; private consumerMessages: ConsumerMessages | null = null; private processingLoop: Promise | null = null; @@ -85,6 +95,7 @@ export class WorkerRunner { this.ackWaitMs = options.ackWaitMs ?? 60000; this.workerService = getWorkerService(); this.ledger = options.ledger; + this.heartbeat = options.heartbeat ?? null; } async start(): Promise { @@ -94,8 +105,11 @@ export class WorkerRunner { } this.running = true; Logger.info({workerId: this.workerId, lane: this.laneName, concurrency: this.concurrency}, 'Worker starting'); + this.startHeartbeat(); this.consumerMessages = await this.openConsumerMessages(); - this.processingLoop = this.consumeUntilStopped(this.consumerMessages); + this.processingLoop = this.consumeUntilStopped(this.consumerMessages).finally(() => { + this.stopHeartbeatTicker(); + }); } async stop(): Promise { @@ -111,9 +125,29 @@ export class WorkerRunner { await this.processingLoop; this.processingLoop = null; } + this.stopHeartbeatTicker(); + this.heartbeatSignal?.release(); + this.heartbeatSignal = null; Logger.info({workerId: this.workerId}, 'Worker stopped'); } + private startHeartbeat(): void { + if (this.heartbeat === null) { + return; + } + this.heartbeatSignal = this.heartbeat.register(`lane:${this.laneName}`, WORKER_LANE_STALE_AFTER_MS); + this.heartbeatTimer = setInterval(() => { + this.heartbeatSignal?.report(); + }, WORKER_LANE_HEARTBEAT_INTERVAL_MS); + } + + private stopHeartbeatTicker(): void { + if (this.heartbeatTimer !== null) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + } + private async openConsumerMessages(): Promise { const js = this.queue.getConnectionManager().getJetStreamClient(); const consumer = await js.consumers.get(this.queue.getStreamName(), this.consumerName); diff --git a/fluxer_api/src/api/worker/tests/WorkerHeartbeat.test.ts b/fluxer_api/src/api/worker/tests/WorkerHeartbeat.test.ts new file mode 100644 index 000000000..1bff88fa8 --- /dev/null +++ b/fluxer_api/src/api/worker/tests/WorkerHeartbeat.test.ts @@ -0,0 +1,246 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import type {LoggerInterface} from '@fluxer/logger/src/LoggerInterface'; +import type {IKVProvider} from '@pkgs/kv_client/src/IKVProvider'; +import type {ConsumerMessages, JsMsg} from 'nats'; +import {afterEach, beforeAll, describe, expect, it, vi} from 'vitest'; +import type {IJobLedgerRepository} from '../../jobs/IJobLedgerRepository'; +import {setInjectedWorkerService} from '../../middleware/ServiceRegistry'; +import {NoopWorkerService} from '../../test/NoopWorkerService'; +import {CronScheduler} from '../CronScheduler'; +import { + WORKER_CRON_STALE_AFTER_MS, + WORKER_HEARTBEAT_WRITE_INTERVAL_MS, + WORKER_LANE_STALE_AFTER_MS, + WorkerHeartbeat, +} from '../WorkerHeartbeat'; +import {WorkerRunner} from '../WorkerRunner'; +import type {WorkerService} from '../WorkerService'; + +const HEARTBEAT_PATH = '/tmp/fluxer-worker-heartbeat-test'; +const TASK_TYPE = 'processInactivityDeletions'; + +function createHeartbeat(): { + heartbeat: WorkerHeartbeat; + write: ReturnType; + logger: {info: ReturnType; error: ReturnType}; +} { + const logger = {info: vi.fn(), error: vi.fn()}; + const write = vi.fn(); + const heartbeat = new WorkerHeartbeat({logger, path: HEARTBEAT_PATH, write}); + return {heartbeat, write, logger}; +} + +class FakeConsumerMessages { + private notify: (() => void) | null = null; + private closed = false; + + async close(): Promise { + this.closed = true; + const notify = this.notify; + this.notify = null; + notify?.(); + } + + async *[Symbol.asyncIterator](): AsyncGenerator { + while (!this.closed) { + await new Promise((resolve) => { + this.notify = resolve; + }); + } + } +} + +function createRunner(messages: FakeConsumerMessages, heartbeat: WorkerHeartbeat): WorkerRunner { + return new WorkerRunner({ + tasks: {[TASK_TYPE]: async () => {}}, + queue: { + getConnectionManager: () => ({ + getJetStreamClient: () => ({ + consumers: { + get: async () => ({ + consume: async () => messages as unknown as ConsumerMessages, + }), + }, + }), + }), + getStreamName: () => 'JOBS', + publishToDlq: vi.fn(), + }, + consumerName: 'workers_batch', + laneName: 'batch', + ledger: {} as IJobLedgerRepository, + concurrency: 12, + maxDeliver: 25, + ackWaitMs: 120000, + heartbeat, + }); +} + +function createCronLogger(): LoggerInterface { + const logger = { + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + child: () => logger, + }; + return logger as unknown as LoggerInterface; +} + +function createScheduler(heartbeat: WorkerHeartbeat): CronScheduler { + const workerService = {addJob: vi.fn().mockResolvedValue(1n)} as unknown as WorkerService; + const kvClient = {setnx: vi.fn().mockResolvedValue(true)} as unknown as IKVProvider; + return new CronScheduler(workerService, createCronLogger(), kvClient, heartbeat); +} + +describe('Worker heartbeat', () => { + beforeAll(() => { + setInjectedWorkerService(new NoopWorkerService()); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('keeps rewriting the file while every component reports', async () => { + vi.useFakeTimers(); + const {heartbeat, write} = createHeartbeat(); + const lane = heartbeat.register('lane:realtime', WORKER_LANE_STALE_AFTER_MS); + + heartbeat.start(); + expect(write).toHaveBeenCalledTimes(1); + write.mockClear(); + for (let elapsed = 0; elapsed < WORKER_LANE_STALE_AFTER_MS * 3; elapsed += WORKER_HEARTBEAT_WRITE_INTERVAL_MS) { + lane.report(); + await vi.advanceTimersByTimeAsync(WORKER_HEARTBEAT_WRITE_INTERVAL_MS); + } + heartbeat.stop(); + + expect(write).toHaveBeenCalledTimes((WORKER_LANE_STALE_AFTER_MS * 3) / WORKER_HEARTBEAT_WRITE_INTERVAL_MS); + expect(write.mock.calls[0]?.[0]).toBe(HEARTBEAT_PATH); + expect(heartbeat.stalledComponents()).toEqual([]); + }); + + it('stops rewriting the file once a component goes stale', async () => { + vi.useFakeTimers(); + const {heartbeat, write, logger} = createHeartbeat(); + heartbeat.register('lane:realtime', WORKER_LANE_STALE_AFTER_MS); + + heartbeat.start(); + await vi.advanceTimersByTimeAsync(WORKER_LANE_STALE_AFTER_MS); + const writesBeforeStall = write.mock.calls.length; + await vi.advanceTimersByTimeAsync(WORKER_LANE_STALE_AFTER_MS * 3); + heartbeat.stop(); + + expect(write).toHaveBeenCalledTimes(writesBeforeStall); + expect(heartbeat.stalledComponents()).toEqual(['lane:realtime']); + expect(logger.error).toHaveBeenCalledTimes(1); + expect(logger.error.mock.calls[0]?.[1]).toBe('Worker heartbeat stalled, the container will report unhealthy'); + }); + + it('resumes rewriting the file when a stalled component reports again', async () => { + vi.useFakeTimers(); + const {heartbeat, write, logger} = createHeartbeat(); + const lane = heartbeat.register('lane:realtime', WORKER_LANE_STALE_AFTER_MS); + + heartbeat.start(); + await vi.advanceTimersByTimeAsync(WORKER_LANE_STALE_AFTER_MS * 3); + const writesBeforeRecovery = write.mock.calls.length; + lane.report(); + await vi.advanceTimersByTimeAsync(WORKER_HEARTBEAT_WRITE_INTERVAL_MS); + heartbeat.stop(); + + expect(write).toHaveBeenCalledTimes(writesBeforeRecovery + 1); + expect(heartbeat.stalledComponents()).toEqual([]); + expect(logger.info).toHaveBeenCalledWith({path: HEARTBEAT_PATH}, 'Worker heartbeat recovered'); + }); + + it('ignores a released component', async () => { + vi.useFakeTimers(); + const {heartbeat, write} = createHeartbeat(); + const lane = heartbeat.register('lane:realtime', WORKER_LANE_STALE_AFTER_MS); + + heartbeat.start(); + lane.release(); + write.mockClear(); + await vi.advanceTimersByTimeAsync(WORKER_LANE_STALE_AFTER_MS * 3); + heartbeat.stop(); + + expect(write).toHaveBeenCalledTimes((WORKER_LANE_STALE_AFTER_MS * 3) / WORKER_HEARTBEAT_WRITE_INTERVAL_MS); + expect(heartbeat.stalledComponents()).toEqual([]); + }); + + it('reports a lane for as long as the runner is running and releases it on stop', async () => { + vi.useFakeTimers(); + const {heartbeat, write} = createHeartbeat(); + const messages = new FakeConsumerMessages(); + const runner = createRunner(messages, heartbeat); + + heartbeat.start(); + await runner.start(); + write.mockClear(); + await vi.advanceTimersByTimeAsync(WORKER_LANE_STALE_AFTER_MS * 3); + + expect(heartbeat.stalledComponents()).toEqual([]); + expect(write).toHaveBeenCalledTimes((WORKER_LANE_STALE_AFTER_MS * 3) / WORKER_HEARTBEAT_WRITE_INTERVAL_MS); + + await runner.stop(); + await vi.advanceTimersByTimeAsync(WORKER_LANE_STALE_AFTER_MS * 3); + heartbeat.stop(); + + expect(heartbeat.stalledComponents()).toEqual([]); + }); + + it('stalls when a running lane stops ticking, which is what a frozen worker looks like', async () => { + vi.useFakeTimers(); + const {heartbeat, write} = createHeartbeat(); + const messages = new FakeConsumerMessages(); + const runner = createRunner(messages, heartbeat); + + heartbeat.start(); + await runner.start(); + vi.clearAllTimers(); + write.mockClear(); + await vi.advanceTimersByTimeAsync(WORKER_LANE_STALE_AFTER_MS * 3); + + expect(write).not.toHaveBeenCalled(); + expect(heartbeat.stalledComponents()).toEqual(['lane:batch']); + expect(heartbeat.writeOnce()).toBe(false); + + await runner.stop(); + heartbeat.stop(); + }); + + it('reports the cron scheduler on every tick and releases it on stop', async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-01-01T00:00:00.000Z')); + const {heartbeat} = createHeartbeat(); + const scheduler = createScheduler(heartbeat); + scheduler.upsert('flushUserActivityBuffer', 'flushUserActivityBuffer', {}, '*/10 * * * * *', {ledger: false}); + + scheduler.start(); + await vi.advanceTimersByTimeAsync(WORKER_CRON_STALE_AFTER_MS * 2); + expect(heartbeat.stalledComponents()).toEqual([]); + + scheduler.stop(); + await vi.advanceTimersByTimeAsync(WORKER_CRON_STALE_AFTER_MS * 2); + expect(heartbeat.stalledComponents()).toEqual([]); + }); + + it('stalls when the cron scheduler stops ticking while it is still registered', async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-01-01T00:00:00.000Z')); + const {heartbeat} = createHeartbeat(); + const scheduler = createScheduler(heartbeat); + scheduler.upsert('flushUserActivityBuffer', 'flushUserActivityBuffer', {}, '*/10 * * * * *', {ledger: false}); + + scheduler.start(); + await vi.advanceTimersByTimeAsync(1000); + vi.clearAllTimers(); + await vi.advanceTimersByTimeAsync(WORKER_CRON_STALE_AFTER_MS * 2); + + expect(heartbeat.stalledComponents()).toEqual(['cron']); + }); +}); diff --git a/fluxer_api/src/api/worker/tests/WorkerProcessErrorHandler.test.ts b/fluxer_api/src/api/worker/tests/WorkerProcessErrorHandler.test.ts new file mode 100644 index 000000000..48f2eb159 --- /dev/null +++ b/fluxer_api/src/api/worker/tests/WorkerProcessErrorHandler.test.ts @@ -0,0 +1,174 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {describe, expect, it, vi} from 'vitest'; +import {createWorkerProcessErrorHandler, type WorkerProcessErrorSource} from '../WorkerProcessErrorHandler'; + +function createHarness(overrides: {shutdown?: () => Promise; forceExitDelayMs?: number} = {}) { + const logger = {error: vi.fn(), warn: vi.fn()}; + const exit = vi.fn(); + const shutdown = vi.fn(overrides.shutdown ?? (async () => {})); + const handle = createWorkerProcessErrorHandler({ + logger, + shutdown, + exit, + forceExitDelayMs: overrides.forceExitDelayMs ?? 5, + }); + return {logger, exit, shutdown, handle}; +} + +function pooledClientError(fields: Record): Error { + const error = new Error(String(fields['message'] ?? 'pooled client error')); + Object.assign(error, {client: {}}, fields); + return error; +} + +function adminShutdownError(): Error { + return pooledClientError({ + message: 'terminating connection due to administrator command', + code: '57P01', + severity: 'FATAL', + routine: 'ProcessInterrupts', + length: 116, + name: 'error', + }); +} + +describe('Worker process error handler', () => { + it('keeps the worker running when Postgres terminates a pooled connection', async () => { + const {logger, exit, shutdown, handle} = createHarness(); + + await handle('uncaughtException', adminShutdownError()); + + expect(shutdown).not.toHaveBeenCalled(); + expect(exit).not.toHaveBeenCalled(); + expect(logger.error).not.toHaveBeenCalled(); + expect(logger.warn).toHaveBeenCalledTimes(1); + const [context, message] = logger.warn.mock.calls[0]!; + expect(message).toBe('Transient database connection error reached the worker process, keeping the worker running'); + expect(context).toMatchObject({source: 'uncaughtException'}); + }); + + it.each([ + ['57P01 admin_shutdown', '57P01'], + ['57P02 crash_shutdown', '57P02'], + ['57P03 cannot_connect_now', '57P03'], + ['08006 connection_failure', '08006'], + ['08003 connection_does_not_exist', '08003'], + ])('survives %s', async (_label, code) => { + const {exit, shutdown, handle} = createHarness(); + + await handle('uncaughtException', pooledClientError({message: 'connection lost', code})); + + expect(shutdown).not.toHaveBeenCalled(); + expect(exit).not.toHaveBeenCalled(); + }); + + it.each([ + ['uncaughtException' as WorkerProcessErrorSource], + ['unhandledRejection' as WorkerProcessErrorSource], + ])('survives a transient error arriving as %s', async (source) => { + const {exit, shutdown, handle} = createHarness(); + + await handle(source, adminShutdownError()); + + expect(shutdown).not.toHaveBeenCalled(); + expect(exit).not.toHaveBeenCalled(); + }); + + it('survives a socket error raised by a pooled Postgres client', async () => { + const {exit, shutdown, handle} = createHarness(); + + await handle('uncaughtException', pooledClientError({message: 'read ECONNRESET', code: 'ECONNRESET'})); + + expect(shutdown).not.toHaveBeenCalled(); + expect(exit).not.toHaveBeenCalled(); + }); + + it('survives a transient error wrapped in a cause chain', async () => { + const {exit, shutdown, handle} = createHarness(); + const wrapped = new Error('Connection terminated due to connection timeout', {cause: adminShutdownError()}); + + await handle('uncaughtException', wrapped); + + expect(shutdown).not.toHaveBeenCalled(); + expect(exit).not.toHaveBeenCalled(); + }); + + it('survives the pg driver telling us the client is no longer queryable', async () => { + const {exit, shutdown, handle} = createHarness(); + + await handle('uncaughtException', new Error('Client has encountered a connection error and is not queryable')); + + expect(shutdown).not.toHaveBeenCalled(); + expect(exit).not.toHaveBeenCalled(); + }); + + it('still tears the worker down on a programming error', async () => { + const {logger, exit, shutdown, handle} = createHarness(); + const bug = new TypeError('cannot read properties of undefined'); + + await handle('uncaughtException', bug); + + expect(logger.warn).not.toHaveBeenCalled(); + expect(logger.error).toHaveBeenCalledWith({err: bug, source: 'uncaughtException'}, 'Uncaught Exception'); + expect(shutdown).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(1); + }); + + it('still tears the worker down on a Postgres error that is not connection level', async () => { + const {exit, shutdown, handle} = createHarness(); + + await handle( + 'uncaughtException', + pooledClientError({message: 'duplicate key value violates unique constraint', code: '23505'}), + ); + + expect(shutdown).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(1); + }); + + it('does not treat a socket error from a non-Postgres source as transient', async () => { + const {exit, shutdown, handle} = createHarness(); + const socketError = Object.assign(new Error('read ECONNRESET'), {code: 'ECONNRESET'}); + + await handle('uncaughtException', socketError); + + expect(shutdown).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(1); + }); + + it('labels an unhandled rejection distinctly when it is fatal', async () => { + const {logger, exit, shutdown, handle} = createHarness(); + const bug = new Error('boom'); + + await handle('unhandledRejection', bug); + + expect(logger.error).toHaveBeenCalledWith( + {err: bug, source: 'unhandledRejection'}, + 'Unhandled Rejection at Promise', + ); + expect(shutdown).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(1); + }); + + it('force exits when shutdown hangs on a fatal error', async () => { + const {exit, handle} = createHarness({shutdown: () => new Promise(() => {}), forceExitDelayMs: 5}); + + void handle('uncaughtException', new Error('boom')); + + await vi.waitFor(() => expect(exit).toHaveBeenCalledWith(1)); + }); + + it('exits even when shutdown itself throws', async () => { + const {logger, exit, handle} = createHarness({ + shutdown: async () => { + throw new Error('shutdown failed'); + }, + }); + + await handle('uncaughtException', new Error('boom')); + + expect(exit).toHaveBeenCalledWith(1); + expect(logger.error).toHaveBeenCalledTimes(2); + }); +});