fix(minions): reconnect worker after promote connection loss (#2025)

Recover the worker-owned Postgres pool when promoteDelayed escapes a retryable connection error, preventing the repeated Promotion error: No database connection loop from issue #1491.\n\nAdds a regression test proving reconnect happens before the worker continues to claim work.

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<void> }).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<void> {
+    const reconnect = (this.engine as { reconnect?: (ctx?: { error?: unknown }) => Promise<void> }).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(). */

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<void> } {
+  return {
+    kind: 'postgres',
+    reconnect: async () => {
+      counter.calls += 1;
+      events.push('reconnect');
+    },
+  } as unknown as BrainEngine & { reconnect: () => Promise<void> };
+}
+
+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<void>;
+      promoteDelayed: () => Promise<unknown[]>;
+      claim: () => Promise<null>;
+      handleStalled: () => Promise<{ requeued: unknown[]; dead: unknown[] }>;
+      handleTimeouts: () => Promise<unknown[]>;
+      handleWallClockTimeouts: () => Promise<unknown[]>;
+    } }).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']);
+  });
+});