fix(api): survive transient database errors in the worker (#2326)

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}

fluxer_api/src/api/database/TransientDatabaseError.ts

@@ -0,0 +1,99 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+const TRANSIENT_POSTGRES_SQLSTATES: ReadonlySet<string> = new Set([
+	'08000',
+	'08001',
+	'08003',
+	'08004',
+	'08006',
+	'08007',
+	'57P01',
+	'57P02',
+	'57P03',
+]);
+
+const TRANSIENT_SOCKET_CODES: ReadonlySet<string> = new Set([
+	'EAI_AGAIN',
+	'ECONNABORTED',
+	'ECONNREFUSED',
+	'ECONNRESET',
+	'EHOSTUNREACH',
+	'ENETDOWN',
+	'ENETUNREACH',
+	'ENOTFOUND',
+	'EPIPE',
+	'ETIMEDOUT',
+]);
+
+const TRANSIENT_DRIVER_MESSAGES: ReadonlySet<string> = 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<string, unknown>;
+
+function readString(value: unknown): string | null {
+	return typeof value === 'string' ? value : null;
+}
+
+function collectErrorChain(error: unknown): Array<ErrorNode> {
+	const nodes: Array<ErrorNode> = [];
+	const queue: Array<unknown> = [error];
+	const seen = new Set<unknown>();
+	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);
+}

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<string, CronDefinition> = 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<void> {
+		try {
+			await this.runDueDefinitions();
+		} finally {
+			this.heartbeatSignal?.report();
+		}
+	}
+
+	private async runDueDefinitions(): Promise<void> {
 		const nowSeconds = Math.floor(Date.now() / 1000);
 		const previousTickSecond = this.lastTickSecond;
 		this.lastTickSecond = nowSeconds;

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<ILogger, 'info' | 'error'>;
+	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<ILogger, 'info' | 'error'>;
+	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<string, WorkerHeartbeatComponent>();
+	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<string> {
+		const at = this.now();
+		const stalled: Array<string> = [];
+		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,
+			})),
+		});
+	}
+}

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<void> {
 	let snowflakeService: ISnowflakeService | null = null;
 	let dependencies: WorkerDependencies | null = null;
 	let cron: CronScheduler | null = null;
+	const heartbeat = new WorkerHeartbeat({logger: Logger});
 	const runners: Array<WorkerRunner> = [];
 	let searchInitialized = false;
 	let shuttingDown = false;
@@ -106,6 +109,7 @@ export async function startWorkerMain(): Promise<void> {
 		}
 		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<void> {
 				}
 			}
 		}
-		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<string, WorkerTaskHandler> = {};
@@ -248,6 +252,7 @@ export async function startWorkerMain(): Promise<void> {
 				concurrency: lane.concurrency,
 				maxDeliver: lane.maxDeliver,
 				ackWaitMs: lane.ackWaitMs,
+				heartbeat,
 			});
 			runners.push(runner);
 		}
@@ -278,18 +283,20 @@ export async function startWorkerMain(): Promise<void> {
 			{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');

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<void>;
+
+interface WorkerProcessErrorHandlerOptions {
+	logger: Pick<ILogger, 'error' | 'warn'>;
+	shutdown: () => Promise<void>;
+	exit: (code: number) => void;
+	forceExitDelayMs?: number;
+}
+
+const FATAL_MESSAGE: Record<WorkerProcessErrorSource, string> = {
+	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);
+	};
+}

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<typeof setInterval> | null = null;
 	private running = false;
 	private consumerMessages: ConsumerMessages | null = null;
 	private processingLoop: Promise<void> | 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<void> {
@@ -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<void> {
@@ -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<ConsumerMessages> {
 		const js = this.queue.getConnectionManager().getJetStreamClient();
 		const consumer = await js.consumers.get(this.queue.getStreamName(), this.consumerName);

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<typeof vi.fn>;
+	logger: {info: ReturnType<typeof vi.fn>; error: ReturnType<typeof vi.fn>};
+} {
+	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<void> {
+		this.closed = true;
+		const notify = this.notify;
+		this.notify = null;
+		notify?.();
+	}
+
+	async *[Symbol.asyncIterator](): AsyncGenerator<JsMsg> {
+		while (!this.closed) {
+			await new Promise<void>((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']);
+	});
+});

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<void>; 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<string, unknown>): 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<void>(() => {}), 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);
+	});
+});