diff --git a/test/worker-lock-renewal-e2e.serial.test.ts b/test/worker-lock-renewal-e2e.serial.test.ts new file mode 100644 index 000000000..0d31600fc --- /dev/null +++ b/test/worker-lock-renewal-e2e.serial.test.ts @@ -0,0 +1,168 @@ +/** + * v0.41.26.1 — gold-standard E2E regression for the lock-renewal + * cathedral wave (gap H from the post-ship coverage audit). + * + * Single test, single describe. The bun:test serial runner has an + * unresolved interaction with PGLite when multiple MinionWorker-driven + * tests share a file (the second test's `queue.add` hangs indefinitely + * even with cleanly reset state between tests). Isolating H into its + * own file sidesteps the issue without touching production code. + * + * Companion tests for gaps A, B, C, D, E, F, G live in: + * - test/worker-lock-renewal.test.ts (pure-function state machine, 18 cases) + * - test/worker-lock-renewal-shape.test.ts (source-shape behavioral pins, this wave) + * - test/audit/lock-renewal-audit.test.ts (audit module, 11 cases) + * - test/audit/redact-connection-info.test.ts (privacy redactor, 15 cases) + * + * What this test pins: + * + * - With renewLock throws happening continuously, the worker process + * MUST NOT crash via unhandledRejection (the v0.41.22.1 bug). + * - The handler observes abort.signal.aborted = true (proves the + * time-based abort fires through the wiring). + * - The audit JSONL contains both `failure` (per-throw) and `gave_up` + * (time-based deadline trip) events. + * + * Hermetic: real PGLite, no DATABASE_URL, no docker, no live PgBouncer. + * Engine wrap on executeRaw injects the simulated connection drop on + * renewLock-shaped SQL only; everything else passes through. + */ + +import { describe, test, expect, beforeAll, afterAll } from 'bun:test'; +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { MinionWorker } from '../src/core/minions/worker.ts'; +import { MinionQueue } from '../src/core/minions/queue.ts'; +import { readRecentLockRenewalEvents } from '../src/core/audit/lock-renewal-audit.ts'; + +// Module-level audit dir + env mutation. withEnv's restore semantics +// race with the worker's setInterval callbacks (audit writes can land +// AFTER withEnv exits, under the wrong env). Persistent module-level +// is the deterministic fix. +const auditDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lr-e2e-')); +const PRIOR_AUDIT_DIR = process.env.GBRAIN_AUDIT_DIR; +process.env.GBRAIN_AUDIT_DIR = auditDir; + +let engine: PGLiteEngine; +let queue: MinionQueue; +let originalExecuteRaw: PGLiteEngine['executeRaw']; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({ database_url: '' }); + await engine.initSchema(); + queue = new MinionQueue(engine); + originalExecuteRaw = engine.executeRaw.bind(engine); +}); + +afterAll(async () => { + await engine.disconnect(); + try { fs.rmSync(auditDir, { recursive: true, force: true }); } catch { /* */ } + if (PRIOR_AUDIT_DIR === undefined) { + delete process.env.GBRAIN_AUDIT_DIR; + } else { + process.env.GBRAIN_AUDIT_DIR = PRIOR_AUDIT_DIR; + } +}); + +describe('H: gold-standard regression — worker survives renewLock throws', () => { + test('the v0.41.22.1 production crash class no longer crashes the worker', async () => { + await engine.executeRaw('DELETE FROM minion_jobs'); + await queue.add('long-runner', {}); + + // Wrap executeRaw to inject renewLock failures. The renewLock SQL + // shape (`UPDATE minion_jobs SET lock_until = now() + ...`) is narrow + // enough to skip claim / completeJob / failJob / etc. + let throwsRemaining = 50; + let renewLockCallCount = 0; + (engine as { executeRaw: PGLiteEngine['executeRaw'] }).executeRaw = async ( + sql: string, + ...args: unknown[] + ) => { + const isRenewLock = sql.includes('SET lock_until = now()') && sql.includes('lock_token'); + if (isRenewLock) { + renewLockCallCount++; + if (throwsRemaining > 0) { + throwsRemaining--; + throw new Error('simulated PgBouncer connection drop'); + } + } + return originalExecuteRaw(sql, ...(args as Parameters)); + }; + + // Short lockDuration → 50ms timer interval, abort deadline at + // lockDuration - safetyMargin = 100 - 16 = 84ms. Sustained throws + // should trip the deadline within ~150ms. + const worker = new MinionWorker(engine, { + concurrency: 1, + pollInterval: 25, + lockDuration: 100, + }); + + let handlerEntered = false; + let handlerAbortObserved = false; + let abortReason: string | null = null; + worker.register('long-runner', async (ctx) => { + handlerEntered = true; + const start = Date.now(); + while (!ctx.signal.aborted && Date.now() - start < 4000) { + await new Promise((r) => setTimeout(r, 10)); + } + handlerAbortObserved = ctx.signal.aborted; + if (ctx.signal.aborted) { + abortReason = ctx.signal.reason instanceof Error + ? ctx.signal.reason.message + : String(ctx.signal.reason); + } + }); + + // The headline regression check: install a process-level + // unhandledRejection listener. Pre-v0.41.26.1, a renewLock throw + // inside `setInterval(async ...)` would propagate here and the + // daemon would exit code 1. + let unhandledRejectionFired: unknown = null; + const rejectionListener = (reason: unknown) => { + unhandledRejectionFired = reason; + }; + process.on('unhandledRejection', rejectionListener); + + const p = worker.start(); + try { + // Fixed sleep — handler enters within ~50ms, abort fires by + // ~200ms; 2s gives plenty of margin AND lets audit events + // accumulate before we read them back. + await new Promise((r) => setTimeout(r, 2000)); + + // Headline assertion: worker process didn't die. + expect(unhandledRejectionFired).toBe(null); + + // Handler wiring assertions: launchJob plumbed the abort + // signal through correctly. + expect(handlerEntered).toBe(true); + expect(handlerAbortObserved).toBe(true); + expect(abortReason).toBe('lock-renewal-failed'); + + // renewLock was actually called multiple times (sanity check + // that the fault injection fired). + expect(renewLockCallCount).toBeGreaterThan(0); + + // Audit JSONL has the expected event shapes. `failure` (per-throw) + // and `gave_up` (time-based deadline tripped) MUST both appear. + const audit = readRecentLockRenewalEvents(48); + expect(audit.events.length).toBeGreaterThan(0); + const failures = audit.events.filter((e) => e.outcome === 'failure'); + const gaveUp = audit.events.filter((e) => e.outcome === 'gave_up'); + expect(failures.length).toBeGreaterThan(0); + expect(gaveUp.length).toBeGreaterThan(0); + // The error message survived through the redactor + truncator. + expect(gaveUp[0].error_message_summary).toMatch(/simulated PgBouncer/); + } finally { + process.off('unhandledRejection', rejectionListener); + (engine as { executeRaw: PGLiteEngine['executeRaw'] }).executeRaw = originalExecuteRaw; + worker.stop(); + await Promise.race([p, new Promise((r) => setTimeout(r, 2000))]); + } + }, 30_000); +}); diff --git a/test/worker-lock-renewal-shape.test.ts b/test/worker-lock-renewal-shape.test.ts new file mode 100644 index 000000000..b2094e0b5 --- /dev/null +++ b/test/worker-lock-renewal-shape.test.ts @@ -0,0 +1,258 @@ +/** + * v0.41.26.1 — source-shape behavioral pins for the lock-renewal + * cathedral wave. + * + * These tests grep the actual worker.ts source for the patterns the + * locked decisions promised. They're not behavioral in the + * "run-the-code" sense — they're structural-regression guards that + * fail loud if a future refactor strips a load-bearing piece. + * + * Why source-shape pins instead of runtime integration tests: + * bun:test's serial runner has an unresolved interaction with PGLite + * + multiple MinionWorker-driven tests in the same file (the second + * test's queue.add hangs indefinitely). The headline regression (gap + * H) lives in its own single-test file + * (test/worker-lock-renewal-e2e.serial.test.ts); A, B, C, E, G are + * pinned here via source-shape grep. + * + * These pins are bug-pattern-specific, not implementation-specific — + * a future refactor that changes the SHAPE of how each guarantee is + * provided (e.g., AbortController.signal events → a different + * notification mechanism) would need to update both the source AND + * this test, which is the right level of friction. A refactor that + * accidentally REMOVES the guarantee would fail this test loudly. + * + * Companion files: + * - test/worker-lock-renewal.test.ts — pure state machine (18 unit tests) + * - test/worker-lock-renewal-e2e.serial.test.ts — H gold-standard E2E (1 test) + * - test/audit/lock-renewal-audit.test.ts — audit primitive (11 tests) + * - test/audit/redact-connection-info.test.ts — privacy redactor (15 tests) + * - test/audit/batch-retry-redaction.test.ts — sibling privacy backfill (3 tests) + * - test/scripts/check-worker-lock-renewal-shape.test.ts — CI guard meta (5 tests) + * + * Coverage map: + * - A. launchJob wires runLockRenewalTick → pinned here + the CI guard + * - B. executeJob skip-failJob on infra abort → pinned here + * - C. .catch() on stored executeJob.finally promise → pinned here + * - D. INFRASTRUCTURE_ABORT_REASONS export → pinned here + pure unit tests + * - E. universal grace-evict listener → pinned here + * - F. logExecuteJobRejected end-to-end → folded into C pin (call site) + * - G. tickInFlight re-entrancy guard → pinned here + * - H. gold-standard regression → behavioral E2E (sibling file) + */ + +import { describe, test, expect } from 'bun:test'; +import * as fs from 'fs'; +import * as path from 'path'; +import { INFRASTRUCTURE_ABORT_REASONS } from '../src/core/minions/worker.ts'; + +const REPO_ROOT = path.resolve(import.meta.dir, '..'); +const WORKER_PATH = path.join(REPO_ROOT, 'src/core/minions/worker.ts'); + +// Read once at module load — failure to find the file is a strong signal +// the test file was moved without updating the path. +let workerSource: string; +try { + workerSource = fs.readFileSync(WORKER_PATH, 'utf8'); +} catch (err) { + throw new Error(`Cannot read ${WORKER_PATH}: ${(err as Error).message}`); +} + +// Helper: scope text to a single function body. Looks for `private launchJob(` +// (or whatever signature) and returns everything from that line to the next +// top-level `}` (heuristic — works for the project's bracing style). +function extractFunctionBody(source: string, signatureMarker: string): string { + const startIdx = source.indexOf(signatureMarker); + if (startIdx === -1) { + throw new Error(`Marker not found: ${signatureMarker}`); + } + // Find the opening brace of the function body. + const braceIdx = source.indexOf('{', startIdx); + if (braceIdx === -1) { + throw new Error(`No opening brace after marker: ${signatureMarker}`); + } + // Walk forward counting braces. + let depth = 1; + let i = braceIdx + 1; + while (i < source.length && depth > 0) { + const ch = source[i]; + if (ch === '{') depth++; + else if (ch === '}') depth--; + i++; + if (depth === 0) break; + } + return source.slice(braceIdx, i); +} + +// ============================================================================= +// D — INFRASTRUCTURE_ABORT_REASONS export contract (also covered in pure tests) +// ============================================================================= + +describe('D: INFRASTRUCTURE_ABORT_REASONS export contract', () => { + test('exported Set contains exactly lock-renewal-failed + lock-lost', () => { + // Named-constant regression: any change to this set is a deliberate + // two-line edit (the constant + this test). + expect(INFRASTRUCTURE_ABORT_REASONS).toBeInstanceOf(Set); + expect(INFRASTRUCTURE_ABORT_REASONS.size).toBe(2); + expect(INFRASTRUCTURE_ABORT_REASONS.has('lock-renewal-failed')).toBe(true); + expect(INFRASTRUCTURE_ABORT_REASONS.has('lock-lost')).toBe(true); + }); + + test('source exports the constant (so executeJob.catch can import it)', () => { + // Without the `export const` shape, executeJob's infrastructure- + // abort guard can't reach the set; the import would fail at compile + // time but pinning the export site here documents the contract. + expect(workerSource).toMatch(/export const INFRASTRUCTURE_ABORT_REASONS/); + }); +}); + +// ============================================================================= +// A — launchJob wires runLockRenewalTick +// ============================================================================= + +describe('A: launchJob wires the pure tick function', () => { + let launchJobBody: string; + test('extracts launchJob function body for further assertions', () => { + launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + expect(launchJobBody.length).toBeGreaterThan(0); + }); + + test('launchJob calls runLockRenewalTick (the extracted pure function)', () => { + expect(launchJobBody).toMatch(/runLockRenewalTick\s*\(/); + }); + + test('launchJob constructs the LockRenewalState via the documented helper', () => { + // resolveLockRenewalKnobs reads the env knobs (D2). If it disappears + // from launchJob, operators can't tune via env vars. + expect(launchJobBody).toMatch(/resolveLockRenewalKnobs\s*\(/); + }); + + test('launchJob uses the lockRenewalAudit sink (not a fake / inline)', () => { + expect(launchJobBody).toMatch(/lockRenewalAudit/); + }); +}); + +// ============================================================================= +// G — tickInFlight re-entrancy guard at the worker layer +// ============================================================================= + +describe('G: tickInFlight re-entrancy guard', () => { + test('launchJob declares the tickInFlight flag', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + expect(launchJobBody).toMatch(/let\s+tickInFlight\s*=\s*false/); + }); + + test('the setInterval callback checks tickInFlight and bails on re-entry', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + // The pattern is `if (tickInFlight) return;` — minor whitespace + // variation tolerated. + expect(launchJobBody).toMatch(/if\s*\(\s*tickInFlight\s*\)\s*return/); + }); + + test('the setInterval callback sets tickInFlight=true before scheduling work', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + expect(launchJobBody).toMatch(/tickInFlight\s*=\s*true/); + }); + + test('the post-tick finally clears tickInFlight back to false', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + expect(launchJobBody).toMatch(/tickInFlight\s*=\s*false/); + }); +}); + +// ============================================================================= +// C + F — .catch() on stored executeJob.finally promise + +// logExecuteJobRejected end-to-end via the catch +// ============================================================================= + +describe('C + F: .catch() on stored executeJob promise + logExecuteJobRejected', () => { + test('the stored executeJob promise has a .catch() handler', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + // The pattern is `.executeJob(...).finally(...).catch(...)` — the + // catch closes the SECOND unhandledRejection vector codex caught. + // Match any `.catch(` after `.finally(` within launchJob's body. + const finallyIdx = launchJobBody.indexOf('.finally('); + expect(finallyIdx).toBeGreaterThan(-1); + // Search for .catch( after the .finally( + const tail = launchJobBody.slice(finallyIdx); + expect(tail).toMatch(/\.catch\s*\(/); + }); + + test('the .catch() handler calls lockRenewalAudit.logExecuteJobRejected', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + expect(launchJobBody).toMatch(/logExecuteJobRejected\s*\(/); + }); + + test('the .catch() handler also logs to stderr (operator visibility)', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + // The implementation uses console.error for the human-visible trail. + // Pin the call so a future refactor that drops it leaves the operator + // with audit JSONL only (which is harder to grep live during an + // incident). + expect(launchJobBody).toMatch(/console\.error[^)]*executeJob unhandled/); + }); +}); + +// ============================================================================= +// E — Universal grace-evict listener fires on any abort reason +// ============================================================================= + +describe('E: universal grace-evict listener (D8b)', () => { + test('launchJob registers an abort.signal.addEventListener for grace-evict', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + // The pre-v0.41.26.1 form lived inside `if (job.timeout_ms != null)`. + // Now it's at launchJob top level and listens to the abort signal + // directly. Pin the listener registration. + expect(launchJobBody).toMatch(/abort\.signal\.addEventListener\s*\(\s*['"]abort['"]/); + }); + + test('the grace-evict path consults INFRASTRUCTURE_ABORT_REASONS before failJob', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + // The infrastructure-reason guard ensures lock-renewal aborts don't + // burn job attempts even if the handler is still wedged at the 30s + // force-evict deadline. + expect(launchJobBody).toMatch(/INFRASTRUCTURE_ABORT_REASONS/); + }); + + test('the 30s grace timer fires for any abort, not just timeout_ms', () => { + const launchJobBody = extractFunctionBody(workerSource, 'private launchJob('); + // The 30_000 literal must appear OUTSIDE the `if (job.timeout_ms != null)` + // branch. Check the addEventListener block contains the 30_000 literal. + // Find the addEventListener and look in its function body. + const listenerIdx = launchJobBody.indexOf("abort.signal.addEventListener('abort'"); + expect(listenerIdx).toBeGreaterThan(-1); + // Grab ~1500 chars after the listener to capture its body. + const listenerWindow = launchJobBody.slice(listenerIdx, listenerIdx + 1500); + expect(listenerWindow).toMatch(/30_000|30000/); + }); +}); + +// ============================================================================= +// B — executeJob skips failJob on infrastructure aborts +// ============================================================================= + +describe('B: executeJob skip-failJob on infrastructure abort (D8a)', () => { + test('executeJob detects the infrastructure abort reason and returns early', () => { + const executeJobBody = extractFunctionBody(workerSource, 'private async executeJob('); + // The skip-failJob branch checks abort.signal.reason against + // INFRASTRUCTURE_ABORT_REASONS and returns BEFORE the failJob call + // path. Pin both the constant reference AND the early-return shape. + expect(executeJobBody).toMatch(/INFRASTRUCTURE_ABORT_REASONS\.has/); + // The return-early shape: after the INFRASTRUCTURE_ABORT_REASONS + // check there must be a `return;` to skip the rest of catch. + // Pin the structural shape by locating the check + finding `return;` + // within ~500 chars after. + const idx = executeJobBody.indexOf('INFRASTRUCTURE_ABORT_REASONS.has'); + expect(idx).toBeGreaterThan(-1); + const window = executeJobBody.slice(idx, idx + 500); + expect(window).toMatch(/return\s*;/); + }); + + test('executeJob still calls failJob for non-infrastructure errors', () => { + const executeJobBody = extractFunctionBody(workerSource, 'private async executeJob('); + // Regression guard for the negative side: a future refactor that + // accidentally removes failJob entirely would make handler defects + // silently disappear. Pin that failJob remains in the catch path. + expect(executeJobBody).toMatch(/this\.queue\.failJob\s*\(/); + }); +});