diff --git a/src/core/minions/worker.ts b/src/core/minions/worker.ts index 81bee4b4c..beb3055ba 100644 --- a/src/core/minions/worker.ts +++ b/src/core/minions/worker.ts @@ -507,7 +507,17 @@ export class MinionWorker extends EventEmitter { try { await this.queue.promoteDelayed(); } catch (e) { - console.error('Promotion error:', e instanceof Error ? e.message : String(e)); + const msg = e instanceof Error ? e.message : String(e); + console.error('Promotion error:', msg); + // issue #1491: a retryable pool/connection loss during promotion used + // to be logged and ignored, leaving the worker in a repeated + // "Promotion error: No database connection" loop until a later path + // happened to reconnect or crash. Promotion is a standalone UPDATE + // from delayed→waiting, so after a connection failure we can safely + // rebuild the worker-owned pool before continuing to claim work. + if (isRetryableConnError(e)) { + await this.reconnectAfterConnectionError('promoteDelayed', e); + } } // Claim jobs up to concurrency limit @@ -532,13 +542,7 @@ export class MinionWorker extends EventEmitter { if (!isRetryableConnError(e)) throw e; const msg = e instanceof Error ? e.message : String(e); console.error(`[worker] claim hit a connection error; reconnecting, retry on next tick: ${msg}`); - const reconnect = (this.engine as { reconnect?: () => Promise }).reconnect; - if (reconnect) { - try { await reconnect.call(this.engine); } - catch (re) { - console.error(`[worker] reconnect after claim error failed: ${re instanceof Error ? re.message : String(re)}`); - } - } + await this.reconnectAfterConnectionError('claim', e); await new Promise(resolve => setTimeout(resolve, this.opts.pollInterval)); continue; } @@ -658,6 +662,22 @@ export class MinionWorker extends EventEmitter { this.running = false; } + /** + * Rebuild the worker-owned DB pool after a retryable connection failure. + * + * PostgresEngine exposes reconnect(); PGLite and test doubles may not. Absence + * is a no-op so non-Postgres workers preserve their legacy behavior. + */ + private async reconnectAfterConnectionError(site: string, error: unknown): Promise { + const reconnect = (this.engine as { reconnect?: (ctx?: { error?: unknown }) => Promise }).reconnect; + if (!reconnect) return; + try { + await reconnect.call(this.engine, { error }); + } catch (re) { + console.error(`[worker] reconnect after ${site} error failed: ${re instanceof Error ? re.message : String(re)}`); + } + } + /** RSS watchdog. Called from the per-job finally and the periodic timer. * Idempotent: returns early if already not running or already shut down. * When threshold is exceeded, hands off to gracefulShutdown(). */ diff --git a/test/worker-promote-reconnect.test.ts b/test/worker-promote-reconnect.test.ts new file mode 100644 index 000000000..3090bdd12 --- /dev/null +++ b/test/worker-promote-reconnect.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, test } from 'bun:test'; +import { MinionWorker } from '../src/core/minions/worker.ts'; +import type { BrainEngine } from '../src/core/engine.ts'; + +function makeEngineWithReconnect(counter: { calls: number }, events: string[]): BrainEngine & { reconnect: () => Promise } { + return { + kind: 'postgres', + reconnect: async () => { + counter.calls += 1; + events.push('reconnect'); + }, + } as unknown as BrainEngine & { reconnect: () => Promise }; +} + +describe('MinionWorker connection recovery', () => { + test('reconnects after a retryable promoteDelayed connection error before continuing the poll loop', async () => { + const reconnect = { calls: 0 }; + const events: string[] = []; + const engine = makeEngineWithReconnect(reconnect, events); + const worker = new MinionWorker(engine, { + pollInterval: 1, + stalledInterval: 60_000, + healthCheckInterval: 0, + }); + worker.register('noop', async () => ({ ok: true })); + + let promoted = false; + let claimCalls = 0; + (worker as unknown as { queue: { + ensureSchema: () => Promise; + promoteDelayed: () => Promise; + claim: () => Promise; + handleStalled: () => Promise<{ requeued: unknown[]; dead: unknown[] }>; + handleTimeouts: () => Promise; + handleWallClockTimeouts: () => Promise; + } }).queue = { + ensureSchema: async () => {}, + promoteDelayed: async () => { + if (!promoted) { + promoted = true; + throw new Error('No database connection: connect() has not been called'); + } + return []; + }, + claim: async () => { + claimCalls += 1; + events.push('claim'); + worker.stop(); + return null; + }, + handleStalled: async () => ({ requeued: [], dead: [] }), + handleTimeouts: async () => [], + handleWallClockTimeouts: async () => [], + }; + + await worker.start(); + + expect(claimCalls).toBe(1); + expect(reconnect.calls).toBe(1); + expect(events).toEqual(['reconnect', 'claim']); + }); +});