From cc39169d2989051b8365d3483d0d5d7c87e0e6fd Mon Sep 17 00:00:00 2001 From: henrique221 Date: Wed, 9 Sep 2026 17:17:20 -0300 Subject: [PATCH 1/7] fix(workers): monitor dead-letter queues without consuming jobs --- .github/workflows/pre-merge.yml | 39 +++ docs/runbooks/worker-dead-letter-queues.md | 174 ++++++++++++ src/index.ts | 18 +- .../dead-letter-queues.integration.test.ts | 260 ++++++++++++++++++ src/lib/dead-letter-queues.test.ts | 189 +++++++++++++ src/lib/dead-letter-queues.ts | 116 ++++++++ src/lib/dead-letter-telemetry.test.ts | 47 ++++ src/lib/queue.ts | 53 ++-- src/workers/dbl-sync.worker.test.ts | 2 + src/workers/dbl-sync.worker.ts | 3 +- src/workers/ingest-bible-text.worker.test.ts | 37 ++- src/workers/ingest-bible-text.worker.ts | 5 +- src/workers/standalone-worker.ts | 16 +- 13 files changed, 905 insertions(+), 54 deletions(-) create mode 100644 docs/runbooks/worker-dead-letter-queues.md create mode 100644 src/lib/dead-letter-queues.integration.test.ts create mode 100644 src/lib/dead-letter-queues.test.ts create mode 100644 src/lib/dead-letter-queues.ts create mode 100644 src/lib/dead-letter-telemetry.test.ts diff --git a/.github/workflows/pre-merge.yml b/.github/workflows/pre-merge.yml index a2869776..9f59246c 100644 --- a/.github/workflows/pre-merge.yml +++ b/.github/workflows/pre-merge.yml @@ -38,6 +38,45 @@ jobs: - name: Build test run: npm run build + dlq-integration: + name: Dead-letter queue integration + runs-on: ubuntu-latest + timeout-minutes: 10 + if: ${{ !github.event.pull_request.draft }} + services: + postgres: + image: postgres:16-alpine + env: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: dlq-local-test + POSTGRES_DB: fluent_dlq_test + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U postgres -d fluent_dlq_test" + --health-interval 5s + --health-timeout 5s + --health-retries 10 + steps: + - name: Checkout repository + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Set up Node.js version + uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 + with: + node-version: 24.14.0 + cache: npm + + - name: Install dependencies + run: npm install --legacy-peer-deps + env: + CXXFLAGS: '-std=c++20' + + - name: Exercise retries, retention and DLQ observability + run: npm test -- --run src/lib/dead-letter-queues.integration.test.ts --maxWorkers=2 + env: + DLQ_TEST_DATABASE_URL: postgres://postgres:dlq-local-test@127.0.0.1:5432/fluent_dlq_test + docs-structure: name: Docs Structure Check runs-on: ubuntu-latest diff --git a/docs/runbooks/worker-dead-letter-queues.md b/docs/runbooks/worker-dead-letter-queues.md new file mode 100644 index 00000000..783c2d95 --- /dev/null +++ b/docs/runbooks/worker-dead-letter-queues.md @@ -0,0 +1,174 @@ +# Worker dead-letter queues + +Issue [#256](https://github.com/eten-tech-foundation/fluent-api/issues/256) follows +the retry handling added in [#212](https://github.com/eten-tech-foundation/fluent-api/pull/212). + +## Decision + +Keep dead-letter queues unconsumed and report their depth on API startup and every +60 seconds. A warning with `event=worker_dlq_depth` and `depth > 0` confirms that +retained jobs exist in a DLQ. Ordinary retry attempts remain in the source queue +and do not produce this signal. This is a backlog gauge, not an exactly-once +per-job event or a count of new failures. + +The API runs the monitor because the export WebJob refuses to boot without R2. +Monitoring therefore continues when that worker cannot start. It uses the existing +Pino/Application Insights logger, needs no new service or fluent-platform change, +and stops its timer and waits for an active sweep before stopping pg-boss. +Slow sweeps never overlap. Discovery failures and individual queue read failures +emit `worker_dlq_monitor_error`; a failure in one queue does not skip the others. + +Every API replica reports independently. Treat depth as a gauge and use the latest +sample, not a sum of samples or instances. Production Application Insights requires +the existing `APPLICATIONINSIGHTS_CONNECTION_STRING`; without it, logs are local +only. The code emits telemetry; Azure alert rules and notification destinations +still need to be configured by the environment owner. + +## Queue convention + +Use `ensureWorkerQueue(boss, name, options)` before sending or consuming jobs. +It creates `-dlq` first and updates the source's `deadLetter` setting even +when the source already exists. Export and AI retry settings stay at three retries +with 60-second exponential backoff. DBL queues keep their current retry settings +(pg-boss defaults for new queues). + +This applies to `usfm-export`, `ai-suggestions` (formerly +`ai-suggestion-trigger`), both `dbl-ingest-text` queues, and `dbl-sync` when its +optional worker is registered. This change does not enable the DBL sync worker or +add a schedule. The monitor discovers all configured dead-letter targets plus +all existing `*-dlq` queues, including orphaned legacy queues. Future queues using +the helper are included on the next sweep. + +The monitor only reads names, counts and the oldest creation time. It never fetches +jobs, acknowledges them, logs payloads, replays them, or deletes them. Each sample +has flat `event`, `queueName`, `depth`, `queuedCount`, `activeCount`, +`deferredCount`, and `oldestCreatedOn` dimensions. Depth counts every retained row, +including a row accidentally consumed by another process. Deferred jobs are also +queued, so adding those two counts would count them twice. + +The query reads the `pgboss.job` parent table (including partitions) through the +existing pg-boss connection. This deliberately avoids pg-boss **12.1.1** +`getQueueStats`: when no rows remain, that implementation can return cached +nonzero counters. A SQL aggregate without `GROUP BY` reliably reports zero. Keep +the integration test when upgrading pg-boss or changing its configured schema. + +## Retention and rollout + +New DLQ entries have at least **30 days** of retention from arrival. Longer existing +queue retention is preserved. pg-boss maintenance removes unconsumed jobs after +`keep_until`; completed/cancelled/failed jobs are removed after +`completed_on + deletion_seconds`. Both DLQ settings have a 30-day minimum. +This corrects the assumption in the issue that pg-boss keeps jobs indefinitely. + +Queue settings are copied into jobs when they are inserted. Updating the queue +does **not** rewrite existing source or DLQ rows, reset their clocks, recover +previous failures, or move old jobs to a new DLQ. In particular: + +- Existing DLQ rows retain their original deadline (normally 14 days). Export + evidence needed beyond that deadline to approved restricted storage before it + expires. Do not assume rollout grants those rows another 30 days. +- AI/DBL jobs sent before their source had `deadLetter` still have no DLQ target. + Inspect their failed source rows and per-attempt logs during rollout. +- Legacy export queues with a different immutable policy are preserved, including + completed/failed history. Startup emits `worker_queue_policy_mismatch`. Resolve + that policy through an explicit migration after reviewing and preserving all + work; startup no longer drops and recreates a queue. New export queues use + `exclusive` as before. + +There is no automated replay or application cleanup. pg-boss's existing maintenance +schedule controls when expired entries are removed. Roll back the application code +without dropping the queues; existing messages and their stored routing still need +their DLQ destinations. An older binary may resume its old queue-recreation logic, +so check legacy policy mismatches before rolling back. + +## Investigate and recover + +1. Confirm the queue and oldest timestamp from the latest depth sample. Check + `worker_dlq_monitor_error` if a queue has stopped reporting. A warning on a + nonzero backlog repeats every minute until that backlog is resolved or expires. +2. Inspect the destination's `id`, `data`, `output`, `created_on` and `keep_until` + using authorized, read-only database access. Treat payloads and error outputs + as private operational data. The DLQ has a **new job ID**: pg-boss copies payload + and failure output, not the source ID. Correlate with source jobs and worker logs; + payload equality alone is not proof of identity. +3. Fix the dependency/configuration problem and verify the worker can run. Review + the job's current business state and idempotency before replay, especially AI + requests that may already have produced external side effects. +4. Select specific source jobs for an operator-controlled retry, or explicitly + enqueue a reviewed payload if the source row is gone. Confirm completion before + resolving the corresponding retained DLQ entry. No bulk drain/purge command is + part of this runbook. Retained evidence continues to count until an operator + resolves it or its retention expires. + +## Application Insights queries + +For a backlog alert, evaluate every minute over a 10-minute window and trigger +when the result has at least one row. Use the latest value per role and queue so +multiple API replicas and repeated samples do not inflate the count: + +```kusto +traces +| where timestamp > ago(10m) +| where tostring(customDimensions.event) == "worker_dlq_depth" +| extend queueName = tostring(customDimensions.queueName), + depth = toint(customDimensions.depth), + oldestCreatedOn = todatetime(customDimensions.oldestCreatedOn) +| summarize arg_max(timestamp, *) by cloud_RoleName, queueName +| where depth > 0 +| project timestamp, cloud_RoleName, queueName, depth, oldestCreatedOn +``` + +Alert separately on monitor failures; missing telemetry must not mean an empty queue: + +```kusto +traces +| where timestamp > ago(10m) +| where tostring(customDimensions.event) == "worker_dlq_monitor_error" +| project timestamp, cloud_RoleName, cloud_RoleInstance, + queueName = tostring(customDimensions.queueName), + error = tostring(customDimensions.error) +``` + +For a missing-signal rule scoped to the API's Application Insights resource, +trigger when `samples == 0` (including when the API itself is down): + +```kusto +traces +| where timestamp > ago(10m) +| where tostring(customDimensions.event) == "worker_dlq_depth" +| summarize samples = count() +| where samples == 0 +``` + +On a workspace-scoped Logs view, use `AppTraces`, `TimeGenerated`, `Properties`, +`AppRoleName` and `AppRoleInstance` in place of the corresponding resource-scoped +names above. Bind rules to the environment's approved action group. This PR does +not create live alert resources or send notifications. +See the [Azure AppTraces table reference](https://learn.microsoft.com/en-us/azure/azure-monitor/reference/tables/apptraces) +for workspace column names. + +## Local validation + +The `Dead-letter queue integration` PR check runs against a fresh PostgreSQL 16 +service. Unit tests cover non-destructive setup, retention, current AI routing, queue +discovery, structured logs, partial failures, timer recovery and shutdown. +The opt-in PostgreSQL suite uses the real pg-boss engine and export/AI worker +handlers, replacing only their external export/storage/AI dependencies and logger. +It checks failed retries, terminal routing, recovery, payload/output preservation, +legacy rows, worker timeout, retention expiry, and a return to zero depth. + +Use a fresh, isolated PostgreSQL 16 container with a random loopback port: + +```sh +docker run --name fluent-dlq-test -e POSTGRES_PASSWORD=dlq-local-test \ + -e POSTGRES_DB=fluent_dlq_test -p 127.0.0.1::5432 -d postgres:16-alpine +docker port fluent-dlq-test 5432 +# Substitute the returned port. Never use the application's DATABASE_URL. +DLQ_TEST_DATABASE_URL=postgres://postgres:dlq-local-test@127.0.0.1:PORT/fluent_dlq_test \ + npm test -- --run src/lib/dead-letter-queues.integration.test.ts --maxWorkers=2 +``` + +The suite refuses non-loopback hosts, another database name, or a database with +existing application queues. It leaves its synthetic evidence in that disposable +database for inspection. Use a fresh test database on subsequent runs. No R2, +fluent-ai, hosted database, alerting resource or production queue is accessed. diff --git a/src/index.ts b/src/index.ts index cbaafd0a..788176f1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,8 +5,14 @@ import { reclaimOrphanedStorageObjects } from '@/domains/verse-audio/verse-audio import env from '@/env'; import { initializeAudioStorage, isAudioStorageConfigured } from '@/lib/audio-storage'; import { verifyBlobStorageOnBoot } from '@/lib/blob-storage'; +import { startDeadLetterMonitor } from '@/lib/dead-letter-queues'; import { logger } from '@/lib/logger'; -import { ensureExportQueues, initializeQueue, QUEUE_NAMES, stopQueue } from '@/lib/queue'; +import { + ensureAiSuggestionQueue, + ensureExportQueues, + initializeQueue, + stopQueue, +} from '@/lib/queue'; import app from './app'; @@ -23,13 +29,8 @@ async function startServer() { await ensureExportQueues(boss); logger.info('Ensuring AI suggestion trigger queue exists'); - await boss.createQueue(QUEUE_NAMES.AI_SUGGESTIONS, { - policy: 'exclusive', - retryLimit: 3, - retryDelay: 60, - retryBackoff: true, - expireInSeconds: 3600, - }); + await ensureAiSuggestionQueue(boss); + const stopDeadLetterMonitor = startDeadLetterMonitor(boss); logger.info('Queue ready'); @@ -66,6 +67,7 @@ async function startServer() { logger.info(`${signal} received, shutting down server`); try { if (audioReclaimInterval) clearInterval(audioReclaimInterval); + await stopDeadLetterMonitor(); server.close(() => { logger.info('HTTP server closed'); diff --git a/src/lib/dead-letter-queues.integration.test.ts b/src/lib/dead-letter-queues.integration.test.ts new file mode 100644 index 00000000..324294d8 --- /dev/null +++ b/src/lib/dead-letter-queues.integration.test.ts @@ -0,0 +1,260 @@ +import { Readable } from 'node:stream'; +import { PgBoss } from 'pg-boss'; +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; + +import { createUSFMZipStreamAsync, getProjectName } from '@/domains/usfm/usfm.service'; +import { uploadExportStream } from '@/lib/blob-storage'; +import { + DLQ_RETENTION_SECONDS, + ensureWorkerQueue, + reportDeadLetterQueues, +} from '@/lib/dead-letter-queues'; +import { logger } from '@/lib/logger'; +import { ensureAiSuggestionQueue, ensureExportQueues, QUEUE_NAMES } from '@/lib/queue'; +import { triggerAiSuggestions } from '@/lib/services/fluent-ai/fluent-ai.client'; +import { registerAiTriggerWorker } from '@/workers/ai-trigger.worker'; +import { registerUSFMExportWorker } from '@/workers/usfm-export.worker'; + +vi.mock('@/lib/logger', () => ({ logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() } })); +vi.mock('@/domains/usfm/usfm.service', () => ({ + createUSFMZipStreamAsync: vi.fn(), + getProjectName: vi.fn(), +})); +vi.mock('@/lib/blob-storage', () => ({ uploadExportStream: vi.fn() })); +vi.mock('@/lib/services/fluent-ai/fluent-ai.client', () => ({ triggerAiSuggestions: vi.fn() })); + +const connectionString = process.env.DLQ_TEST_DATABASE_URL; + +// This suite writes jobs and expires synthetic fixtures. Never use the app's +// DATABASE_URL or a shared database. See docs/runbooks/worker-dead-letter-queues.md. +describe.skipIf(!connectionString)('dead-letter queues with PostgreSQL and pg-boss 12', () => { + let boss: PgBoss; + const errors: Error[] = []; + const exportQueue = QUEUE_NAMES.USFM_EXPORT; + const exportDlq = QUEUE_NAMES.USFM_EXPORT_DLQ; + + async function rows(name: string) { + return ( + await boss + .getDb() + .executeSql('SELECT * FROM pgboss.job WHERE name = $1 ORDER BY created_on, id', [name]) + ).rows; + } + + beforeAll(async () => { + const url = new URL(connectionString!); + if ( + !['localhost', '127.0.0.1', '[::1]'].includes(url.hostname) || + url.pathname !== '/fluent_dlq_test' + ) { + throw new Error('DLQ tests require a disposable local database named fluent_dlq_test'); + } + boss = new PgBoss({ + connectionString, + schema: 'pgboss', + supervise: false, + schedule: false, + max: 2, + }); + boss.on('error', (error) => errors.push(error)); + await boss.start(); + if ((await boss.getQueues()).some((queue) => !queue.name.startsWith('__'))) { + throw new Error('DLQ tests require an empty disposable database'); + } + }); + + afterAll(async () => { + await boss?.stop({ graceful: true }); + expect(errors).toEqual([]); + }); + + it('keeps legacy source history and existing DLQ rows unchanged during setup', async () => { + await boss.createQueue(exportDlq); + await boss.createQueue(exportQueue, { policy: 'standard', retryLimit: 0 }); + const oldSourceId = await boss.send(exportQueue, { projectUnitId: 10 }); + await boss.fetch(exportQueue); + await boss.fail(exportQueue, oldSourceId!, new Error('legacy failure')); + await boss.send(exportDlq, { originalPayload: 'keep this evidence' }); + const sourceBefore = await rows(exportQueue); + const dlqBefore = await rows(exportDlq); + await ensureExportQueues(boss); + await ensureExportQueues(boss); + expect(await rows(exportQueue)).toEqual(sourceBefore); + expect(await rows(exportDlq)).toEqual(dlqBefore); + expect(await boss.getQueue(exportQueue)).toMatchObject({ + policy: 'standard', + deadLetter: exportDlq, + }); + expect(await boss.getQueue(exportDlq)).toMatchObject({ + retentionSeconds: DLQ_RETENTION_SECONDS, + }); + }); + + it('reports a retry separately from a real terminal DLQ arrival and preserves payload/output', async () => { + await ensureWorkerQueue(boss, 'retry-probe', { retryLimit: 1, retryDelay: 0 }); + const payload = { projectUnitId: 20, requestedBy: 7 }; + const id = await boss.send('retry-probe', payload); + await boss.fetch('retry-probe'); + await boss.fail('retry-probe', id!, new Error('first attempt')); + expect((await boss.getJobById('retry-probe', id!))?.state).toBe('retry'); + await reportDeadLetterQueues(boss); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ queueName: 'retry-probe-dlq', depth: 0 }), + expect.any(String) + ); + await boss.fetch('retry-probe'); + await boss.fail('retry-probe', id!, new Error('terminal attempt')); + const before = await rows('retry-probe-dlq'); + expect(before).toHaveLength(1); + expect(before[0]).toMatchObject({ + state: 'created', + data: payload, + output: { message: 'terminal attempt' }, + }); + expect(before[0].id).not.toBe(id); + expect((before[0].keep_until.getTime() - before[0].created_on.getTime()) / 1000).toBeCloseTo( + DLQ_RETENTION_SECONDS, + 0 + ); + await reportDeadLetterQueues(boss); + await reportDeadLetterQueues(boss); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + event: 'worker_dlq_depth', + queueName: 'retry-probe-dlq', + depth: 1, + }), + expect.any(String) + ); + expect(await rows('retry-probe-dlq')).toEqual(before); + }); + + it('runs the real export worker through an upload failure and exhausted retries', async () => { + await boss.updateQueue(exportQueue, { retryLimit: 1, retryDelay: 0, retryBackoff: false }); + vi.mocked(createUSFMZipStreamAsync).mockImplementation( + async () => ({ ok: true, data: { stream: Readable.from(['zip']), cleanup: vi.fn() } }) as any + ); + vi.mocked(uploadExportStream).mockRejectedValue(new Error('simulated R2 outage')); + await registerUSFMExportWorker(boss); + const payload = { projectUnitId: 30, requestedBy: 7 }; + const id = await boss.send(exportQueue, payload); + await vi.waitFor( + async () => { + expect((await boss.getJobById(exportQueue, id!))?.state).toBe('failed'); + }, + { timeout: 15_000, interval: 100 } + ); + expect(uploadExportStream).toHaveBeenCalledTimes(2); + expect(await rows(exportDlq)).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + data: payload, + state: 'created', + output: expect.objectContaining({ + message: expect.stringContaining('simulated R2 outage'), + }), + }), + ]) + ); + }, 20_000); + + it('lets a transient export failure recover without adding a DLQ entry', async () => { + const dlqBefore = await rows(exportDlq); + vi.mocked(uploadExportStream) + .mockRejectedValueOnce(new Error('temporary outage')) + .mockResolvedValue({ + filename: 'export.zip', + sizeBytes: 3, + expiresAt: new Date(Date.now() + 60_000), + }); + vi.mocked(getProjectName).mockResolvedValue({ ok: true, data: 'Test' }); + const id = await boss.send(exportQueue, { projectUnitId: 31, requestedBy: 7 }); + await vi.waitFor( + async () => { + expect((await boss.getJobById(exportQueue, id!))?.state).toBe('completed'); + }, + { timeout: 15_000, interval: 100 } + ); + expect(await rows(exportDlq)).toEqual(dlqBefore); + }, 20_000); + + it('routes each failed AI batch member to its DLQ through the real worker', async () => { + await ensureAiSuggestionQueue(boss); + await boss.updateQueue(QUEUE_NAMES.AI_SUGGESTIONS, { + retryLimit: 1, + retryDelay: 0, + retryBackoff: false, + }); + vi.mocked(triggerAiSuggestions).mockRejectedValue(new Error('simulated AI outage')); + await registerAiTriggerWorker(boss, {}); + const ids: Array = []; + for (const projectUnitId of [40, 41]) { + ids.push( + await boss.send( + QUEUE_NAMES.AI_SUGGESTIONS, + { + projectUnitId, + bibleId: 1, + bookCode: 'GEN', + chapterNumber: 1, + verseStart: 1, + verseEnd: 2, + }, + { singletonKey: `project-unit-${projectUnitId}` } + ) + ); + } + expect(ids.every(Boolean)).toBe(true); + await vi.waitFor( + async () => { + for (const id of ids) + expect((await boss.getJobById(QUEUE_NAMES.AI_SUGGESTIONS, id!))?.state).toBe('failed'); + }, + { timeout: 15_000, interval: 100 } + ); + expect(await rows('ai-suggestions-dlq')).toHaveLength(2); + await reportDeadLetterQueues(boss); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ queueName: 'ai-suggestions-dlq', depth: 2 }), + expect.any(String) + ); + }, 20_000); + + it('observes worker timeouts and pg-boss retention, including an accurate return to zero', async () => { + await ensureWorkerQueue(boss, 'expiry-probe', { retryLimit: 0, expireInSeconds: 1 }); + const id = await boss.send('expiry-probe', { fixture: 'expired worker' }); + await boss.fetch('expiry-probe'); + // Only this synthetic fixture is aged. No production or pre-existing data. + await boss + .getDb() + .executeSql( + "UPDATE pgboss.job SET started_on = now() - interval '2 seconds' WHERE name = $1 AND id = $2", + ['expiry-probe', id] + ); + await boss.supervise('expiry-probe'); + expect((await boss.getJobById('expiry-probe', id!))?.state).toBe('failed'); + expect(await rows('expiry-probe-dlq')).toHaveLength(1); + await reportDeadLetterQueues(boss); + const retained = await rows('expiry-probe-dlq'); + await boss.supervise('expiry-probe-dlq'); + expect(await rows('expiry-probe-dlq')).toEqual(retained); + await boss + .getDb() + .executeSql( + "UPDATE pgboss.job SET keep_until = now() - interval '1 second' WHERE name = $1", + ['expiry-probe-dlq'] + ); + await boss + .getDb() + .executeSql('UPDATE pgboss.queue SET maintain_on = NULL WHERE name = $1', [ + 'expiry-probe-dlq', + ]); + await boss.supervise('expiry-probe-dlq'); + expect(await rows('expiry-probe-dlq')).toHaveLength(0); + await reportDeadLetterQueues(boss); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ queueName: 'expiry-probe-dlq', depth: 0 }), + expect.any(String) + ); + }); +}); diff --git a/src/lib/dead-letter-queues.test.ts b/src/lib/dead-letter-queues.test.ts new file mode 100644 index 00000000..921bfd03 --- /dev/null +++ b/src/lib/dead-letter-queues.test.ts @@ -0,0 +1,189 @@ +import type { PgBoss } from 'pg-boss'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + DLQ_RETENTION_SECONDS, + ensureWorkerQueue, + reportDeadLetterQueues, + startDeadLetterMonitor, +} from '@/lib/dead-letter-queues'; +import { logger } from '@/lib/logger'; +import { ensureAiSuggestionQueue, ensureExportQueues } from '@/lib/queue'; + +vi.mock('@/lib/logger', () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +function fakeBoss() { + const executeSql = vi.fn().mockResolvedValue({ + rows: [{ depth: 0, queuedCount: 0, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }], + }); + const methods = { + getQueue: vi.fn().mockResolvedValue(null), + createQueue: vi.fn(), + updateQueue: vi.fn(), + deleteQueue: vi.fn(), + getQueues: vi.fn().mockResolvedValue([]), + getDb: () => ({ executeSql }), + }; + return { boss: methods as unknown as PgBoss, ...methods, executeSql }; +} + +beforeEach(() => vi.clearAllMocks()); +afterEach(() => vi.useRealTimers()); + +describe('worker queue convention', () => { + it('creates the DLQ first and converges routing on an existing source queue', async () => { + const fake = fakeBoss(); + await ensureWorkerQueue(fake.boss, 'ingestion'); + expect(fake.createQueue.mock.calls).toEqual([ + [ + 'ingestion-dlq', + { retentionSeconds: DLQ_RETENTION_SECONDS, deleteAfterSeconds: DLQ_RETENTION_SECONDS }, + ], + ['ingestion', { deadLetter: 'ingestion-dlq' }], + ]); + expect(fake.updateQueue).toHaveBeenCalledWith('ingestion', { deadLetter: 'ingestion-dlq' }); + expect(fake.deleteQueue).not.toHaveBeenCalled(); + }); + + it('preserves longer retention and never updates immutable source policy', async () => { + const fake = fakeBoss(); + fake.createQueue.mockImplementation((_name, options) => { + options.policy ??= 'standard'; + }); + fake.getQueue.mockResolvedValue({ retentionSeconds: 6_000_000, deleteAfterSeconds: 7_000_000 }); + await ensureWorkerQueue(fake.boss, 'ingestion', { policy: 'exclusive', retryLimit: 3 }); + expect(fake.updateQueue).toHaveBeenCalledWith('ingestion-dlq', { + retentionSeconds: 6_000_000, + deleteAfterSeconds: 7_000_000, + }); + expect(fake.updateQueue).toHaveBeenCalledWith('ingestion', { + retryLimit: 3, + deadLetter: 'ingestion-dlq', + }); + }); + + it('keeps legacy export queues even if only diagnostic history remains', async () => { + const fake = fakeBoss(); + fake.getQueue.mockResolvedValueOnce({ policy: 'standard' }); + await ensureExportQueues(fake.boss); + expect(fake.deleteQueue).not.toHaveBeenCalled(); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ event: 'worker_queue_policy_mismatch' }), + expect.any(String) + ); + expect(fake.updateQueue).toHaveBeenCalledWith('usfm-export', { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + expireInSeconds: 600, + deadLetter: 'usfm-export-dlq', + }); + }); + + it('adds a DLQ to the current AI queue and keeps its retry contract', async () => { + const fake = fakeBoss(); + await ensureAiSuggestionQueue(fake.boss); + expect(fake.updateQueue).toHaveBeenCalledWith('ai-suggestions', { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + expireInSeconds: 3600, + deadLetter: 'ai-suggestions-dlq', + }); + }); +}); + +describe('dead-letter monitoring', () => { + it('discovers custom, shared and orphaned DLQs without counting source retries', async () => { + const fake = fakeBoss(); + fake.getQueues.mockResolvedValue([ + { name: 'export', deadLetter: 'failures' }, + { name: 'ai', deadLetter: 'failures' }, + { name: 'old-dlq' }, + { name: 'failures' }, + ]); + fake.executeSql.mockResolvedValue({ + rows: [{ depth: 3, queuedCount: 2, activeCount: 1, deferredCount: 2, oldestCreatedOn: null }], + }); + await reportDeadLetterQueues(fake.boss); + expect(fake.executeSql.mock.calls.map((call) => call[1])).toEqual([['failures'], ['old-dlq']]); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ event: 'worker_dlq_depth', queueName: 'failures', depth: 3 }), + expect.any(String) + ); + expect(fake.updateQueue).not.toHaveBeenCalled(); + expect(fake.deleteQueue).not.toHaveBeenCalled(); + }); + + it('reports zero after the backlog clears, with flat structured dimensions', async () => { + const fake = fakeBoss(); + fake.getQueues.mockResolvedValue([{ name: 'old-dlq' }]); + await reportDeadLetterQueues(fake.boss); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ event: 'worker_dlq_depth', queueName: 'old-dlq', depth: 0 }), + 'Worker dead-letter queue is empty' + ); + }); + + it('continues after a queue read failure and reports discovery failures', async () => { + const fake = fakeBoss(); + fake.getQueues.mockResolvedValue([{ name: 'one-dlq' }, { name: 'two-dlq' }]); + fake.executeSql.mockRejectedValueOnce(new Error('database read failed')); + await reportDeadLetterQueues(fake.boss); + expect(logger.error).toHaveBeenCalledWith( + expect.objectContaining({ event: 'worker_dlq_monitor_error', queueName: 'one-dlq' }), + expect.any(String) + ); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ queueName: 'two-dlq' }), + expect.any(String) + ); + fake.getQueues.mockRejectedValue(new Error('discovery unavailable')); + await expect(reportDeadLetterQueues(fake.boss)).resolves.toBeUndefined(); + expect(logger.error).toHaveBeenCalledWith( + expect.objectContaining({ + event: 'worker_dlq_monitor_error', + error: 'discovery unavailable', + }), + expect.any(String) + ); + }); + + it('sweeps immediately, avoids overlap, waits for shutdown and stops its timer', async () => { + vi.useFakeTimers(); + const fake = fakeBoss(); + let finish!: (value: []) => void; + fake.getQueues.mockReturnValueOnce( + new Promise((resolve) => { + finish = resolve; + }) + ); + const stop = startDeadLetterMonitor(fake.boss); + expect(fake.getQueues).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(120_000); + expect(fake.getQueues).toHaveBeenCalledTimes(1); + let stopped = false; + const shutdown = stop().then(() => { + stopped = true; + }); + await Promise.resolve(); + expect(stopped).toBe(false); + finish([]); + await shutdown; + await vi.advanceTimersByTimeAsync(120_000); + expect(fake.getQueues).toHaveBeenCalledTimes(1); + }); + + it('resumes on the next interval after a failed sweep', async () => { + vi.useFakeTimers(); + const fake = fakeBoss(); + fake.getQueues.mockRejectedValueOnce(new Error('temporary outage')); + const stop = startDeadLetterMonitor(fake.boss); + await vi.advanceTimersByTimeAsync(60_000); + expect(fake.getQueues).toHaveBeenCalledTimes(2); + await stop(); + }); +}); diff --git a/src/lib/dead-letter-queues.ts b/src/lib/dead-letter-queues.ts new file mode 100644 index 00000000..84effd98 --- /dev/null +++ b/src/lib/dead-letter-queues.ts @@ -0,0 +1,116 @@ +import type { PgBoss, Queue } from 'pg-boss'; + +import { logger } from '@/lib/logger'; + +/** Time to investigate new DLQ entries before pg-boss maintenance removes them. */ +export const DLQ_RETENTION_SECONDS = 30 * 24 * 60 * 60; + +/** Create a durable diagnostic destination before enabling dead-letter routing. */ +export async function ensureWorkerQueue( + boss: PgBoss, + name: string, + options: Omit = {} +): Promise { + const deadLetter = `${name}-dlq`; + const existing = await boss.getQueue(deadLetter); + const retentionOptions = { + // Do not shorten an operator's longer retention policy. Queue updates only + // affect new jobs; existing keep_until/deletion_seconds remain untouched. + retentionSeconds: Math.max(existing?.retentionSeconds ?? 0, DLQ_RETENTION_SECONDS), + deleteAfterSeconds: Math.max(existing?.deleteAfterSeconds ?? 0, DLQ_RETENTION_SECONDS), + }; + // pg-boss mutates createQueue options (including adding an immutable policy). + // Do not pass that mutated object back into updateQueue. + await boss.createQueue(deadLetter, { ...retentionOptions }); + await boss.updateQueue(deadLetter, retentionOptions); + + await boss.createQueue(name, { ...options, deadLetter }); + // createQueue is a no-op for existing queues. Reconcile mutable settings + // without deleting any queue or modifying jobs that have already been sent. + const { policy: _policy, partition: _partition, ...mutableOptions } = options; + await boss.updateQueue(name, { ...mutableOptions, deadLetter }); +} + +/** Report retained DLQ rows without fetching, acknowledging or replaying them. */ +export async function reportDeadLetterQueues(boss: PgBoss): Promise { + try { + const queues = await boss.getQueues(); + // Discover configured targets, including custom names, and orphaned/legacy + // *-dlq queues. A worker added later is picked up on the next sweep. + const targets = new Set(); + for (const queue of queues) { + if (queue.deadLetter) targets.add(queue.deadLetter); + if (queue.name.endsWith('-dlq')) targets.add(queue.name); + } + + for (const queueName of targets) { + try { + // pg-boss 12.1.1 getQueueStats falls back to cached counters when a + // queue becomes empty. Read an aggregate without GROUP BY so a cleared + // queue always reports zero. The parent job table includes partitions. + const { rows } = await boss.getDb().executeSql( + `SELECT count(*)::int AS depth, + count(*) FILTER (WHERE state < 'active')::int AS "queuedCount", + count(*) FILTER (WHERE state = 'active')::int AS "activeCount", + count(*) FILTER (WHERE start_after > now())::int AS "deferredCount", + min(created_on) AS "oldestCreatedOn" + FROM pgboss.job WHERE name = $1`, + [queueName] + ); + const stats = rows[0]; + const properties = { + event: 'worker_dlq_depth', + queueName, + depth: stats.depth, + queuedCount: stats.queuedCount, + activeCount: stats.activeCount, + deferredCount: stats.deferredCount, + oldestCreatedOn: stats.oldestCreatedOn?.toISOString() ?? null, + }; + // Object first keeps dimensions queryable in both Pino and App Insights. + // Count rows once: deferred jobs are also queued. + if (properties.depth > 0) { + logger.warn(properties, 'Worker dead-letter queue contains jobs'); + } else { + logger.info(properties, 'Worker dead-letter queue is empty'); + } + } catch (error) { + logger.error( + { + event: 'worker_dlq_monitor_error', + queueName, + error: error instanceof Error ? error.message : String(error), + }, + 'Worker dead-letter queue inspection failed' + ); + } + } + } catch (error) { + logger.error( + { + event: 'worker_dlq_monitor_error', + error: error instanceof Error ? error.message : String(error), + }, + 'Worker dead-letter queue discovery failed' + ); + } +} + +/** Start on API boot so monitoring survives an export worker's R2 boot failure. */ +export function startDeadLetterMonitor(boss: PgBoss): () => Promise { + let running: Promise | undefined; + const sweep = () => { + if (running) return; + running = reportDeadLetterQueues(boss).finally(() => { + running = undefined; + }); + }; + sweep(); + const interval = setInterval(sweep, 60_000); + interval.unref(); + + return async () => { + clearInterval(interval); + await running; + }; +} diff --git a/src/lib/dead-letter-telemetry.test.ts b/src/lib/dead-letter-telemetry.test.ts new file mode 100644 index 00000000..c1384e07 --- /dev/null +++ b/src/lib/dead-letter-telemetry.test.ts @@ -0,0 +1,47 @@ +import type { PgBoss } from 'pg-boss'; + +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { reportDeadLetterQueues } from '@/lib/dead-letter-queues'; + +const client = vi.hoisted(() => ({ trackTrace: vi.fn() })); +vi.mock('@/env', () => ({ + default: { NODE_ENV: 'production', APPLICATIONINSIGHTS_CONNECTION_STRING: 'test-only' }, +})); +vi.mock('applicationinsights', () => ({ + default: { setup: () => ({ start: vi.fn() }), defaultClient: client }, +})); + +afterEach(() => vi.restoreAllMocks()); + +describe('dLQ Application Insights telemetry', () => { + it('sends queryable dimensions through the real production logger without job payloads', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}); + const boss = { + getQueues: async () => [{ name: 'usfm-export-dlq' }], + getDb: () => ({ + executeSql: async () => ({ + rows: [ + { depth: 1, queuedCount: 1, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }, + ], + }), + }), + } as unknown as PgBoss; + + await reportDeadLetterQueues(boss); + + expect(client.trackTrace).toHaveBeenCalledWith({ + message: 'Worker dead-letter queue contains jobs', + severity: 2, + properties: { + event: 'worker_dlq_depth', + queueName: 'usfm-export-dlq', + depth: 1, + queuedCount: 1, + activeCount: 0, + deferredCount: 0, + oldestCreatedOn: null, + }, + }); + }); +}); diff --git a/src/lib/queue.ts b/src/lib/queue.ts index d5b22c75..59e3e53c 100644 --- a/src/lib/queue.ts +++ b/src/lib/queue.ts @@ -1,5 +1,6 @@ import { PgBoss } from 'pg-boss'; +import { ensureWorkerQueue } from '@/lib/dead-letter-queues'; import { logger } from '@/lib/logger'; let boss: PgBoss | null = null; @@ -81,50 +82,42 @@ const EXPORT_QUEUE_OPTIONS = { retryDelay: 60, retryBackoff: true, expireInSeconds: 600, - deadLetter: QUEUE_NAMES.USFM_EXPORT_DLQ, } as const; /** * Creates/converges the export queues. The 'exclusive' policy backs singletonKey - * dedupe (at most one job per key in queued/active/deferred). createQueue is a - * no-op for existing queues and policy is immutable, so a queue created with an - * older policy is dropped and recreated (pre-enablement: nothing user-facing - * queues jobs yet); the remaining options are converged via updateQueue. + * dedupe (at most one job per key in queued/active/deferred). Policy is immutable; + * preserve older queues and their diagnostic history, and warn for an explicit + * migration instead of deleting a queue during API/worker startup. */ export async function ensureExportQueues(boss: PgBoss): Promise { - await boss.createQueue(QUEUE_NAMES.USFM_EXPORT_DLQ); - const existing = await boss.getQueue(QUEUE_NAMES.USFM_EXPORT); if (existing && existing.policy !== 'exclusive') { - // Policy is immutable and createQueue no-ops on existing queues, so a - // policy change needs delete+recreate. Both the API and the worker run - // this at startup, so it must stay non-destructive and race-tolerant: - // recreate only when the queue holds no work, otherwise keep serving with - // the old policy and converge on a later boot once the queue drains. - const stats = await boss.getQueueStats(QUEUE_NAMES.USFM_EXPORT); - const pendingJobs = stats.queuedCount + stats.activeCount + stats.deferredCount; - if (pendingJobs > 0) { - logger.warn('usfm-export queue policy differs but jobs are pending; skipping recreation', { + logger.warn( + { + event: 'worker_queue_policy_mismatch', + queueName: QUEUE_NAMES.USFM_EXPORT, previousPolicy: existing.policy, - pendingJobs, - }); - } else { - try { - await boss.deleteQueue(QUEUE_NAMES.USFM_EXPORT); - logger.warn('Recreated usfm-export queue with exclusive policy', { - previousPolicy: existing.policy, - }); - } catch (error) { - logger.warn('usfm-export queue recreation raced another process; continuing', { error }); - } - } + expectedPolicy: 'exclusive', + }, + 'Worker queue policy differs; preserving jobs until an explicit migration' + ); } - await boss.createQueue(QUEUE_NAMES.USFM_EXPORT, { + await ensureWorkerQueue(boss, QUEUE_NAMES.USFM_EXPORT, { policy: 'exclusive', ...EXPORT_QUEUE_OPTIONS, }); - await boss.updateQueue(QUEUE_NAMES.USFM_EXPORT, EXPORT_QUEUE_OPTIONS); +} + +export async function ensureAiSuggestionQueue(boss: PgBoss): Promise { + await ensureWorkerQueue(boss, QUEUE_NAMES.AI_SUGGESTIONS, { + policy: 'exclusive', + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + expireInSeconds: 3600, + }); } export async function getQueue(): Promise { diff --git a/src/workers/dbl-sync.worker.test.ts b/src/workers/dbl-sync.worker.test.ts index d4223f81..74ee40c6 100644 --- a/src/workers/dbl-sync.worker.test.ts +++ b/src/workers/dbl-sync.worker.test.ts @@ -26,6 +26,8 @@ describe('dblSyncWorker', () => { it('registers the on-demand worker and handles execution lifecycle', async () => { const mockBoss = { + getQueue: vi.fn().mockResolvedValue(null), + updateQueue: vi.fn(), createQueue: vi.fn().mockResolvedValue(undefined), work: vi.fn().mockResolvedValue(undefined), } as any; diff --git a/src/workers/dbl-sync.worker.ts b/src/workers/dbl-sync.worker.ts index 45781bc4..59b8fdf3 100644 --- a/src/workers/dbl-sync.worker.ts +++ b/src/workers/dbl-sync.worker.ts @@ -4,6 +4,7 @@ import { syncBiblesFromDbl } from '@/domains/bibles/sync/dbl-bible-sync'; import { syncBooksFromDbl } from '@/domains/books/sync/dbl-book-sync'; import { syncLanguagesFromDbl } from '@/domains/languages/sync/dbl-language-sync'; +import { ensureWorkerQueue } from '../lib/dead-letter-queues'; import { logger } from '../lib/logger'; /** Queue name for the DBL catalogue sync job. */ @@ -26,7 +27,7 @@ export const QUEUE_DBL_SYNC = 'dbl-sync'; */ export async function registerDblSyncWorker(boss: PgBoss) { // To trigger it manually, send a job to this queue: await boss.send(QUEUE_DBL_SYNC, {}); - await boss.createQueue(QUEUE_DBL_SYNC); + await ensureWorkerQueue(boss, QUEUE_DBL_SYNC); await boss.work(QUEUE_DBL_SYNC, { batchSize: 1 }, async (jobs) => { const job = jobs[0]; diff --git a/src/workers/ingest-bible-text.worker.test.ts b/src/workers/ingest-bible-text.worker.test.ts index 55cd8af3..a1833fd6 100644 --- a/src/workers/ingest-bible-text.worker.test.ts +++ b/src/workers/ingest-bible-text.worker.test.ts @@ -53,6 +53,8 @@ describe('dblIngestTextWorker', () => { it('registers handlers for both priority and background queues', async () => { const mockBoss = { + getQueue: vi.fn().mockResolvedValue(null), + updateQueue: vi.fn(), createQueue: vi.fn().mockResolvedValue(undefined), work: vi.fn().mockResolvedValue(undefined), } as any; @@ -73,7 +75,12 @@ describe('dblIngestTextWorker', () => { }); it('handles partial download error recovery gracefully', async () => { - const mockBoss = { createQueue: vi.fn(), work: vi.fn() } as any; + const mockBoss = { + getQueue: vi.fn().mockResolvedValue(null), + updateQueue: vi.fn(), + createQueue: vi.fn(), + work: vi.fn(), + } as any; await registerDblIngestTextWorker(mockBoss); // Extract the handler. pg-boss's WorkHandler always receives the batch as @@ -111,7 +118,12 @@ describe('dblIngestTextWorker', () => { }); it('logs a warning and skips the book instead of silently ignoring it when no matching book exists', async () => { - const mockBoss = { createQueue: vi.fn(), work: vi.fn() } as any; + const mockBoss = { + getQueue: vi.fn().mockResolvedValue(null), + updateQueue: vi.fn(), + createQueue: vi.fn(), + work: vi.fn(), + } as any; await registerDblIngestTextWorker(mockBoss); const handler = mockBoss.work.mock.calls[0][2]; const { logger } = await import('../lib/logger'); @@ -135,7 +147,12 @@ describe('dblIngestTextWorker', () => { }); it('marks a failed chapter-list fetch as a job failure so pg-boss retries, instead of silently dropping the whole book', async () => { - const mockBoss = { createQueue: vi.fn(), work: vi.fn() } as any; + const mockBoss = { + getQueue: vi.fn().mockResolvedValue(null), + updateQueue: vi.fn(), + createQueue: vi.fn(), + work: vi.fn(), + } as any; await registerDblIngestTextWorker(mockBoss); const handler = mockBoss.work.mock.calls[0][2]; @@ -180,7 +197,12 @@ describe('dblIngestTextWorker', () => { }); it('logs success only when the assignment Result is ok', async () => { - const mockBoss = { createQueue: vi.fn(), work: vi.fn() } as any; + const mockBoss = { + getQueue: vi.fn().mockResolvedValue(null), + updateQueue: vi.fn(), + createQueue: vi.fn(), + work: vi.fn(), + } as any; await registerDblIngestTextWorker(mockBoss); const handler = mockBoss.work.mock.calls[0][2]; const { logger } = await import('../lib/logger'); @@ -204,7 +226,12 @@ describe('dblIngestTextWorker', () => { }); it('does not log success and throws to trigger a retry when the assignment Result is an error', async () => { - const mockBoss = { createQueue: vi.fn(), work: vi.fn() } as any; + const mockBoss = { + getQueue: vi.fn().mockResolvedValue(null), + updateQueue: vi.fn(), + createQueue: vi.fn(), + work: vi.fn(), + } as any; await registerDblIngestTextWorker(mockBoss); const handler = mockBoss.work.mock.calls[0][2]; const { logger } = await import('../lib/logger'); diff --git a/src/workers/ingest-bible-text.worker.ts b/src/workers/ingest-bible-text.worker.ts index 0f32ac6b..68538905 100644 --- a/src/workers/ingest-bible-text.worker.ts +++ b/src/workers/ingest-bible-text.worker.ts @@ -8,6 +8,7 @@ import type { WorkerMetricsHooks } from './usfm-export.worker'; import { db } from '../db'; import { bible_texts, project_units } from '../db/schema'; import * as chapterAssignmentsService from '../domains/chapter-assignments/chapter-assignments.service'; +import { ensureWorkerQueue } from '../lib/dead-letter-queues'; import { logger } from '../lib/logger'; import { QUEUE_NAMES } from '../lib/queue'; import { dblClient } from '../lib/services/dbl/dbl.client'; @@ -225,8 +226,8 @@ export async function registerDblIngestTextWorker(boss: PgBoss, metricsHooks?: W }; // Register handler on both queues; priority queue processes first - await boss.createQueue(QUEUE_NAMES.DBL_INGEST_TEXT); - await boss.createQueue(QUEUE_NAMES.DBL_INGEST_TEXT_PRIORITY); + await ensureWorkerQueue(boss, QUEUE_NAMES.DBL_INGEST_TEXT); + await ensureWorkerQueue(boss, QUEUE_NAMES.DBL_INGEST_TEXT_PRIORITY); const workOptions = { batchSize: 1 }; await boss.work(QUEUE_NAMES.DBL_INGEST_TEXT, workOptions, handler); diff --git a/src/workers/standalone-worker.ts b/src/workers/standalone-worker.ts index 14d35e98..d4f44af3 100644 --- a/src/workers/standalone-worker.ts +++ b/src/workers/standalone-worker.ts @@ -6,7 +6,13 @@ import { isBlobStorageConfigured, } from '@/lib/blob-storage'; import { logger } from '@/lib/logger'; -import { ensureExportQueues, initializeQueue, QUEUE_NAMES, stopQueue } from '@/lib/queue'; +import { + ensureAiSuggestionQueue, + ensureExportQueues, + initializeQueue, + QUEUE_NAMES, + stopQueue, +} from '@/lib/queue'; import type { WorkerMetricsHooks } from './usfm-export.worker'; @@ -65,13 +71,7 @@ async function startWorker() { await ensureExportQueues(boss); - await boss.createQueue(QUEUE_NAMES.AI_SUGGESTIONS, { - policy: 'exclusive', - retryLimit: 3, - retryDelay: 60, - retryBackoff: true, - expireInSeconds: 3600, - }); + await ensureAiSuggestionQueue(boss); await registerUSFMExportWorker(boss, metricsHooks); await registerAiTriggerWorker(boss, metricsHooks); From aa96cd0ec695399d273a9e3c87762b070c64fd9e Mon Sep 17 00:00:00 2001 From: HenriqueCode Date: Wed, 9 Sep 2026 20:47:43 -0300 Subject: [PATCH 2/7] fix(workers): address queue setup and monitoring review Add an explicit offline policy migration that preserves retained jobs and refuses pending work. Report policy drift for both exclusive worker queues, avoid redundant creation updates, read DLQ stats concurrently, and bound monitor shutdown. Cover the migration with real PostgreSQL integration tests. Refs: #324 --- docs/runbooks/worker-dead-letter-queues.md | 52 +++++++- package.json | 1 + src/db/scripts/migrate-worker-queue-policy.ts | 22 ++++ .../dead-letter-queues.integration.test.ts | 60 +++++++++ src/lib/dead-letter-queues.test.ts | 77 ++++++++++- src/lib/dead-letter-queues.ts | 124 +++++++++++------- src/lib/exclusive-worker-queue-migration.ts | 74 +++++++++++ src/lib/queue.ts | 13 -- 8 files changed, 358 insertions(+), 65 deletions(-) create mode 100644 src/db/scripts/migrate-worker-queue-policy.ts create mode 100644 src/lib/exclusive-worker-queue-migration.ts diff --git a/docs/runbooks/worker-dead-letter-queues.md b/docs/runbooks/worker-dead-letter-queues.md index 783c2d95..38f3acf3 100644 --- a/docs/runbooks/worker-dead-letter-queues.md +++ b/docs/runbooks/worker-dead-letter-queues.md @@ -14,9 +14,12 @@ per-job event or a count of new failures. The API runs the monitor because the export WebJob refuses to boot without R2. Monitoring therefore continues when that worker cannot start. It uses the existing Pino/Application Insights logger, needs no new service or fluent-platform change, -and stops its timer and waits for an active sweep before stopping pg-boss. +and stops its timer and waits up to five seconds for an active sweep before +continuing API and pg-boss shutdown. A timeout logs +`worker_dlq_monitor_shutdown_timeout`; it does not cancel the database query. Slow sweeps never overlap. Discovery failures and individual queue read failures emit `worker_dlq_monitor_error`; a failure in one queue does not skip the others. +Queue reads run concurrently so one slow target does not delay healthy samples. Every API replica reports independently. Treat depth as a gauge and use the latest sample, not a sum of samples or instances. Production Application Insights requires @@ -69,10 +72,10 @@ previous failures, or move old jobs to a new DLQ. In particular: expires. Do not assume rollout grants those rows another 30 days. - AI/DBL jobs sent before their source had `deadLetter` still have no DLQ target. Inspect their failed source rows and per-attempt logs during rollout. -- Legacy export queues with a different immutable policy are preserved, including +- Legacy export or AI queues with a different immutable policy are preserved, including completed/failed history. Startup emits `worker_queue_policy_mismatch`. Resolve - that policy through an explicit migration after reviewing and preserving all - work; startup no longer drops and recreates a queue. New export queues use + that policy through the explicit migration below after reviewing all + work; startup no longer drops and recreates a queue. New export and AI queues use `exclusive` as before. There is no automated replay or application cleanup. pg-boss's existing maintenance @@ -81,6 +84,42 @@ without dropping the queues; existing messages and their stored routing still ne their DLQ destinations. An older binary may resume its old queue-recreation logic, so check legacy policy mismatches before rolling back. +### Migrate a legacy exclusive policy + +Run this separately from deployment only when `worker_queue_policy_mismatch` +identifies `usfm-export` or `ai-suggestions`. The script is pinned to pg-boss +**12.1.1, schema 26** and refuses another schema version or queue name. It uses +`WORKER_QUEUE_MIGRATION_DATABASE_URL` explicitly, never `.env` or the application's +`DATABASE_URL`. Use the approved environment connection and pg-boss schema owner. +Do not paste the connection string into logs or commit it. + +1. Inspect without changing queue or job data: + `npm run queue:migrate-policy -- usfm-export`. + The result contains only the policy and retained/pending counts, not payloads. +2. Pause producers, including API replicas, scheduled producers and administrative + scripts. Let queued, deferred and active jobs finish, or have an operator review + and cancel specific jobs if appropriate. Then stop **all** workers and pg-boss + maintenance processes for the maintenance window. Keep producers stopped. +3. Run `npm run queue:migrate-policy -- usfm-export --apply`. + Repeat for `ai-suggestions` if its inspection showed a mismatch. +4. Inspect again, confirm `exclusive`, then restart workers and API replicas. + Restarting clears pg-boss's cached policy. Verify a normal request completes + and the mismatch warning no longer appears. + +The apply transaction locks the queue and job tables, including partitions, and +refuses any queued, deferred, retrying or active work. Lock acquisition is limited +to five seconds and statements to thirty seconds; an error rolls back the whole +migration. The maintenance window affects all queues because the job table lock +covers their partitions. Retry only after checking the reported blocker. + +No queue or job is deleted. The migration changes the queue's policy and retained +jobs' policy metadata so a later operator retry also respects singleton dedupe. +IDs, payloads, outputs, states, retry counters, routing and original deadlines stay +unchanged; DLQ rows are untouched. Dedicated partitions receive the exclusive +index; the shared partition's existing index is checked. Re-running after success +is a no-op. Do not switch back to a non-exclusive policy as an application rollback; +older binaries already expect exclusive dedupe. + ## Investigate and recover 1. Confirm the queue and oldest timestamp from the latest depth sample. Check @@ -155,7 +194,10 @@ discovery, structured logs, partial failures, timer recovery and shutdown. The opt-in PostgreSQL suite uses the real pg-boss engine and export/AI worker handlers, replacing only their external export/storage/AI dependencies and logger. It checks failed retries, terminal routing, recovery, payload/output preservation, -legacy rows, worker timeout, retention expiry, and a return to zero depth. +legacy rows, worker timeout, retention expiry, and a return to zero depth. It also +executes the policy migration against shared and dedicated partitions, verifies +pending-work refusal and preserved history, and proves duplicate singleton keys +are rejected afterward. Use a fresh, isolated PostgreSQL 16 container with a random loopback port: diff --git a/package.json b/package.json index 9af92f44..eb7f5950 100644 --- a/package.json +++ b/package.json @@ -42,6 +42,7 @@ "db:import:languages": "npx tsx src/db/scripts/import-ethnologue-languages.ts", "db:import:language-names": "npx tsx src/db/scripts/enrich-language-names.ts", "db:migrate": "drizzle-kit migrate", + "queue:migrate-policy": "tsx src/db/scripts/migrate-worker-queue-policy.ts", "db:generate": "drizzle-kit generate --name", "db:studio": "drizzle-kit studio", "db:push": "drizzle-kit push", diff --git a/src/db/scripts/migrate-worker-queue-policy.ts b/src/db/scripts/migrate-worker-queue-policy.ts new file mode 100644 index 00000000..411fa82b --- /dev/null +++ b/src/db/scripts/migrate-worker-queue-policy.ts @@ -0,0 +1,22 @@ +import postgres from 'postgres'; + +import { migrateExclusiveWorkerQueue } from '@/lib/exclusive-worker-queue-migration'; + +const [queueName, mode, ...extra] = process.argv.slice(2); +if (!queueName || (mode && mode !== '--apply') || extra.length) { + throw new Error('Usage: npm run queue:migrate-policy -- [--apply]'); +} +// Deliberately do not load .env or fall back to the application's DATABASE_URL. +const connectionString = process.env.WORKER_QUEUE_MIGRATION_DATABASE_URL; +if (!connectionString) throw new Error('WORKER_QUEUE_MIGRATION_DATABASE_URL is required'); + +const sql = postgres(connectionString, { max: 1, connect_timeout: 5 }); +try { + const result = await migrateExclusiveWorkerQueue(sql, queueName, mode === '--apply'); + console.log(JSON.stringify({ mode: mode === '--apply' ? 'apply' : 'inspect', ...result })); +} catch (error) { + console.error(error instanceof Error ? error.message : 'Worker queue policy migration failed'); + process.exitCode = 1; +} finally { + await sql.end({ timeout: 5 }); +} diff --git a/src/lib/dead-letter-queues.integration.test.ts b/src/lib/dead-letter-queues.integration.test.ts index 324294d8..d092d5c8 100644 --- a/src/lib/dead-letter-queues.integration.test.ts +++ b/src/lib/dead-letter-queues.integration.test.ts @@ -1,5 +1,6 @@ import { Readable } from 'node:stream'; import { PgBoss } from 'pg-boss'; +import postgres from 'postgres'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { createUSFMZipStreamAsync, getProjectName } from '@/domains/usfm/usfm.service'; @@ -9,6 +10,7 @@ import { ensureWorkerQueue, reportDeadLetterQueues, } from '@/lib/dead-letter-queues'; +import { migrateExclusiveWorkerQueue } from '@/lib/exclusive-worker-queue-migration'; import { logger } from '@/lib/logger'; import { ensureAiSuggestionQueue, ensureExportQueues, QUEUE_NAMES } from '@/lib/queue'; import { triggerAiSuggestions } from '@/lib/services/fluent-ai/fluent-ai.client'; @@ -29,6 +31,7 @@ const connectionString = process.env.DLQ_TEST_DATABASE_URL; // DATABASE_URL or a shared database. See docs/runbooks/worker-dead-letter-queues.md. describe.skipIf(!connectionString)('dead-letter queues with PostgreSQL and pg-boss 12', () => { let boss: PgBoss; + let migrationSql: postgres.Sql; const errors: Error[] = []; const exportQueue = QUEUE_NAMES.USFM_EXPORT; const exportDlq = QUEUE_NAMES.USFM_EXPORT_DLQ; @@ -61,9 +64,11 @@ describe.skipIf(!connectionString)('dead-letter queues with PostgreSQL and pg-bo if ((await boss.getQueues()).some((queue) => !queue.name.startsWith('__'))) { throw new Error('DLQ tests require an empty disposable database'); } + migrationSql = postgres(connectionString!, { max: 1 }); }); afterAll(async () => { + await migrationSql?.end(); await boss?.stop({ graceful: true }); expect(errors).toEqual([]); }); @@ -90,6 +95,61 @@ describe.skipIf(!connectionString)('dead-letter queues with PostgreSQL and pg-bo }); }); + it('migrates a legacy queue without losing history and enforces singleton dedupe', async () => { + const deferredId = await boss.send( + exportQueue, + { fixture: 'pending migration' }, + { + startAfter: new Date(Date.now() + 60_000), + } + ); + const before = await rows(exportQueue); + const dlqBefore = await rows(exportDlq); + await expect(migrateExclusiveWorkerQueue(migrationSql, exportQueue)).resolves.toMatchObject({ + changed: false, + previousPolicy: 'standard', + pendingJobs: 1, + }); + expect(await rows(exportQueue)).toEqual(before); + await expect(migrateExclusiveWorkerQueue(migrationSql, exportQueue, true)).rejects.toThrow( + 'still has 1' + ); + expect((await boss.getQueue(exportQueue))?.policy).toBe('standard'); + expect(await rows(exportQueue)).toEqual(before); + + await boss.cancel(exportQueue, deferredId!); + const drained = await rows(exportQueue); + await expect( + migrateExclusiveWorkerQueue(migrationSql, exportQueue, true) + ).resolves.toMatchObject({ changed: true }); + expect(await rows(exportQueue)).toEqual( + drained.map((row) => ({ ...row, policy: 'exclusive' })) + ); + expect(await rows(exportDlq)).toEqual(dlqBefore); + expect((await boss.getQueue(exportQueue))?.policy).toBe('exclusive'); + await expect( + migrateExclusiveWorkerQueue(migrationSql, exportQueue, true) + ).resolves.toMatchObject({ changed: false }); + + const key = { singletonKey: 'migration-dedupe-proof' }; + const first = await boss.send(exportQueue, { fixture: 'first' }, key); + expect(first).toBeTruthy(); + expect(await boss.send(exportQueue, { fixture: 'duplicate' }, key)).toBeNull(); + await boss.cancel(exportQueue, first!); + }); + + it('adds the exclusive index when migrating a dedicated AI queue partition', async () => { + await boss.createQueue(QUEUE_NAMES.AI_SUGGESTIONS, { policy: 'standard', partition: true }); + await expect( + migrateExclusiveWorkerQueue(migrationSql, QUEUE_NAMES.AI_SUGGESTIONS, true) + ).resolves.toMatchObject({ changed: true }); + const options = { singletonKey: 'partition-dedupe-proof' }; + const id = await boss.send(QUEUE_NAMES.AI_SUGGESTIONS, { fixture: 'partition' }, options); + expect(id).toBeTruthy(); + expect(await boss.send(QUEUE_NAMES.AI_SUGGESTIONS, {}, options)).toBeNull(); + await boss.cancel(QUEUE_NAMES.AI_SUGGESTIONS, id!); + }); + it('reports a retry separately from a real terminal DLQ arrival and preserves payload/output', async () => { await ensureWorkerQueue(boss, 'retry-probe', { retryLimit: 1, retryDelay: 0 }); const payload = { projectUnitId: 20, requestedBy: 7 }; diff --git a/src/lib/dead-letter-queues.test.ts b/src/lib/dead-letter-queues.test.ts index 921bfd03..40283a3d 100644 --- a/src/lib/dead-letter-queues.test.ts +++ b/src/lib/dead-letter-queues.test.ts @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { DLQ_RETENTION_SECONDS, + DLQ_SHUTDOWN_TIMEOUT_MS, ensureWorkerQueue, reportDeadLetterQueues, startDeadLetterMonitor, @@ -36,18 +37,28 @@ afterEach(() => vi.useRealTimers()); describe('worker queue convention', () => { it('creates the DLQ first and converges routing on an existing source queue', async () => { const fake = fakeBoss(); + fake.getQueue.mockImplementation(async (name) => (name === 'ingestion' ? { name } : null)); await ensureWorkerQueue(fake.boss, 'ingestion'); expect(fake.createQueue.mock.calls).toEqual([ [ 'ingestion-dlq', { retentionSeconds: DLQ_RETENTION_SECONDS, deleteAfterSeconds: DLQ_RETENTION_SECONDS }, ], - ['ingestion', { deadLetter: 'ingestion-dlq' }], ]); expect(fake.updateQueue).toHaveBeenCalledWith('ingestion', { deadLetter: 'ingestion-dlq' }); expect(fake.deleteQueue).not.toHaveBeenCalled(); }); + it('creates new queues with their final settings without redundant updates', async () => { + const fake = fakeBoss(); + await ensureWorkerQueue(fake.boss, 'ingestion', { retryLimit: 3 }); + expect(fake.createQueue).toHaveBeenCalledWith('ingestion', { + retryLimit: 3, + deadLetter: 'ingestion-dlq', + }); + expect(fake.updateQueue).not.toHaveBeenCalled(); + }); + it('preserves longer retention and never updates immutable source policy', async () => { const fake = fakeBoss(); fake.createQueue.mockImplementation((_name, options) => { @@ -67,7 +78,9 @@ describe('worker queue convention', () => { it('keeps legacy export queues even if only diagnostic history remains', async () => { const fake = fakeBoss(); - fake.getQueue.mockResolvedValueOnce({ policy: 'standard' }); + fake.getQueue.mockImplementation(async (name) => + name === 'usfm-export' ? { policy: 'standard' } : null + ); await ensureExportQueues(fake.boss); expect(fake.deleteQueue).not.toHaveBeenCalled(); expect(logger.warn).toHaveBeenCalledWith( @@ -85,6 +98,9 @@ describe('worker queue convention', () => { it('adds a DLQ to the current AI queue and keeps its retry contract', async () => { const fake = fakeBoss(); + fake.getQueue.mockImplementation(async (name) => + name === 'ai-suggestions' ? { policy: 'standard' } : null + ); await ensureAiSuggestionQueue(fake.boss); expect(fake.updateQueue).toHaveBeenCalledWith('ai-suggestions', { retryLimit: 3, @@ -93,10 +109,48 @@ describe('worker queue convention', () => { expireInSeconds: 3600, deadLetter: 'ai-suggestions-dlq', }); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + event: 'worker_queue_policy_mismatch', + queueName: 'ai-suggestions', + previousPolicy: 'standard', + expectedPolicy: 'exclusive', + }), + expect.any(String) + ); + }); + + it('does not warn for an existing exclusive queue', async () => { + const fake = fakeBoss(); + fake.getQueue.mockImplementation(async (name) => + name === 'ai-suggestions' ? { policy: 'exclusive' } : null + ); + await ensureAiSuggestionQueue(fake.boss); + expect(logger.warn).not.toHaveBeenCalled(); }); }); describe('dead-letter monitoring', () => { + it('reports another queue while the first query is still pending', async () => { + const fake = fakeBoss(); + fake.getQueues.mockResolvedValue([{ name: 'slow-dlq' }, { name: 'fast-dlq' }]); + let finish!: () => void; + fake.executeSql.mockReturnValueOnce( + new Promise((resolve) => { + finish = () => resolve({ rows: [{ depth: 0 }] }); + }) + ); + const report = reportDeadLetterQueues(fake.boss); + await vi.waitFor(() => + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ queueName: 'fast-dlq', depth: 0 }), + expect.any(String) + ) + ); + finish(); + await report; + }); + it('discovers custom, shared and orphaned DLQs without counting source retries', async () => { const fake = fakeBoss(); fake.getQueues.mockResolvedValue([ @@ -186,4 +240,23 @@ describe('dead-letter monitoring', () => { expect(fake.getQueues).toHaveBeenCalledTimes(2); await stop(); }); + + it('bounds shutdown when a query never returns and never starts another sweep', async () => { + vi.useFakeTimers(); + const fake = fakeBoss(); + fake.getQueues.mockResolvedValue([{ name: 'stuck-dlq' }]); + fake.executeSql.mockReturnValue(new Promise(() => {})); + const stop = startDeadLetterMonitor(fake.boss); + await vi.advanceTimersByTimeAsync(0); + const shutdown = stop(); + await vi.advanceTimersByTimeAsync(DLQ_SHUTDOWN_TIMEOUT_MS); + await shutdown; + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ event: 'worker_dlq_monitor_shutdown_timeout' }), + expect.any(String) + ); + await vi.advanceTimersByTimeAsync(120_000); + expect(fake.getQueues).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); }); diff --git a/src/lib/dead-letter-queues.ts b/src/lib/dead-letter-queues.ts index 84effd98..83c08861 100644 --- a/src/lib/dead-letter-queues.ts +++ b/src/lib/dead-letter-queues.ts @@ -4,6 +4,7 @@ import { logger } from '@/lib/logger'; /** Time to investigate new DLQ entries before pg-boss maintenance removes them. */ export const DLQ_RETENTION_SECONDS = 30 * 24 * 60 * 60; +export const DLQ_SHUTDOWN_TIMEOUT_MS = 5_000; /** Create a durable diagnostic destination before enabling dead-letter routing. */ export async function ensureWorkerQueue( @@ -12,21 +13,35 @@ export async function ensureWorkerQueue( options: Omit = {} ): Promise { const deadLetter = `${name}-dlq`; - const existing = await boss.getQueue(deadLetter); + const [existing, source] = await Promise.all([boss.getQueue(deadLetter), boss.getQueue(name)]); const retentionOptions = { // Do not shorten an operator's longer retention policy. Queue updates only // affect new jobs; existing keep_until/deletion_seconds remain untouched. retentionSeconds: Math.max(existing?.retentionSeconds ?? 0, DLQ_RETENTION_SECONDS), deleteAfterSeconds: Math.max(existing?.deleteAfterSeconds ?? 0, DLQ_RETENTION_SECONDS), }; - // pg-boss mutates createQueue options (including adding an immutable policy). - // Do not pass that mutated object back into updateQueue. - await boss.createQueue(deadLetter, { ...retentionOptions }); - await boss.updateQueue(deadLetter, retentionOptions); + if (existing) { + await boss.updateQueue(deadLetter, retentionOptions); + } else { + await boss.createQueue(deadLetter, retentionOptions); + } - await boss.createQueue(name, { ...options, deadLetter }); - // createQueue is a no-op for existing queues. Reconcile mutable settings - // without deleting any queue or modifying jobs that have already been sent. + if (!source) { + await boss.createQueue(name, { ...options, deadLetter }); + return; + } + if (options.policy && source.policy !== options.policy) { + logger.warn( + { + event: 'worker_queue_policy_mismatch', + queueName: name, + previousPolicy: source.policy, + expectedPolicy: options.policy, + }, + 'Worker queue policy differs; run the explicit worker queue policy migration' + ); + } + // Policy and partition are immutable in pg-boss. Preserve existing jobs. const { policy: _policy, partition: _partition, ...mutableOptions } = options; await boss.updateQueue(name, { ...mutableOptions, deadLetter }); } @@ -43,48 +58,50 @@ export async function reportDeadLetterQueues(boss: PgBoss): Promise { if (queue.name.endsWith('-dlq')) targets.add(queue.name); } - for (const queueName of targets) { - try { - // pg-boss 12.1.1 getQueueStats falls back to cached counters when a - // queue becomes empty. Read an aggregate without GROUP BY so a cleared - // queue always reports zero. The parent job table includes partitions. - const { rows } = await boss.getDb().executeSql( - `SELECT count(*)::int AS depth, + await Promise.allSettled( + [...targets].map(async (queueName) => { + try { + // pg-boss 12.1.1 getQueueStats falls back to cached counters when a + // queue becomes empty. Read an aggregate without GROUP BY so a cleared + // queue always reports zero. The parent job table includes partitions. + const { rows } = await boss.getDb().executeSql( + `SELECT count(*)::int AS depth, count(*) FILTER (WHERE state < 'active')::int AS "queuedCount", count(*) FILTER (WHERE state = 'active')::int AS "activeCount", count(*) FILTER (WHERE start_after > now())::int AS "deferredCount", min(created_on) AS "oldestCreatedOn" FROM pgboss.job WHERE name = $1`, - [queueName] - ); - const stats = rows[0]; - const properties = { - event: 'worker_dlq_depth', - queueName, - depth: stats.depth, - queuedCount: stats.queuedCount, - activeCount: stats.activeCount, - deferredCount: stats.deferredCount, - oldestCreatedOn: stats.oldestCreatedOn?.toISOString() ?? null, - }; - // Object first keeps dimensions queryable in both Pino and App Insights. - // Count rows once: deferred jobs are also queued. - if (properties.depth > 0) { - logger.warn(properties, 'Worker dead-letter queue contains jobs'); - } else { - logger.info(properties, 'Worker dead-letter queue is empty'); - } - } catch (error) { - logger.error( - { - event: 'worker_dlq_monitor_error', + [queueName] + ); + const stats = rows[0]; + const properties = { + event: 'worker_dlq_depth', queueName, - error: error instanceof Error ? error.message : String(error), - }, - 'Worker dead-letter queue inspection failed' - ); - } - } + depth: stats.depth, + queuedCount: stats.queuedCount, + activeCount: stats.activeCount, + deferredCount: stats.deferredCount, + oldestCreatedOn: stats.oldestCreatedOn?.toISOString() ?? null, + }; + // Object first keeps dimensions queryable in both Pino and App Insights. + // Count rows once: deferred jobs are also queued. + if (properties.depth > 0) { + logger.warn(properties, 'Worker dead-letter queue contains jobs'); + } else { + logger.info(properties, 'Worker dead-letter queue is empty'); + } + } catch (error) { + logger.error( + { + event: 'worker_dlq_monitor_error', + queueName, + error: error instanceof Error ? error.message : String(error), + }, + 'Worker dead-letter queue inspection failed' + ); + } + }) + ); } catch (error) { logger.error( { @@ -111,6 +128,23 @@ export function startDeadLetterMonitor(boss: PgBoss): () => Promise { return async () => { clearInterval(interval); - await running; + if (!running) return; + let timeout: ReturnType | undefined; + try { + await Promise.race([ + running, + new Promise((resolve) => { + timeout = setTimeout(() => { + logger.warn( + { event: 'worker_dlq_monitor_shutdown_timeout', timeoutMs: DLQ_SHUTDOWN_TIMEOUT_MS }, + 'Continuing shutdown while a dead-letter queue inspection is still pending' + ); + resolve(); + }, DLQ_SHUTDOWN_TIMEOUT_MS); + }), + ]); + } finally { + clearTimeout(timeout); + } }; } diff --git a/src/lib/exclusive-worker-queue-migration.ts b/src/lib/exclusive-worker-queue-migration.ts new file mode 100644 index 00000000..3d346d22 --- /dev/null +++ b/src/lib/exclusive-worker-queue-migration.ts @@ -0,0 +1,74 @@ +import type postgres from 'postgres'; + +const MIGRATABLE_QUEUES = ['usfm-export', 'ai-suggestions']; + +/** Offline, operator-invoked migration for the pinned pg-boss 12.1.1 schema. */ +export async function migrateExclusiveWorkerQueue( + sql: postgres.Sql, + queueName: string, + apply = false +) { + if (!MIGRATABLE_QUEUES.includes(queueName)) { + throw new Error('Only usfm-export and ai-suggestions can be migrated'); + } + + return sql.begin(async (tx) => { + await tx`SET LOCAL lock_timeout = '5s'`; + await tx`SET LOCAL statement_timeout = '30s'`; + if (apply) { + // Includes all partitions. No sender, worker or maintenance process can + // change jobs between the pending-work check and the policy update. + await tx`LOCK TABLE pgboss.queue, pgboss.job IN ACCESS EXCLUSIVE MODE`; + } + const [version] = await tx`SELECT version FROM pgboss.version`; + if (version?.version !== 26) { + throw new Error('Migration requires the pg-boss 12.1.1 schema version 26'); + } + const [queue] = await tx` + SELECT policy, partition, table_name FROM pgboss.queue WHERE name = ${queueName} + `; + if (!queue) throw new Error(`Queue ${queueName} does not exist`); + + const [stats] = await tx` + SELECT count(*)::int AS retained, + count(*) FILTER (WHERE state <= 'active')::int AS pending + FROM pgboss.job WHERE name = ${queueName} + `; + const result = { + queueName, + previousPolicy: queue.policy as string, + retainedJobs: stats.retained as number, + pendingJobs: stats.pending as number, + changed: false, + }; + if (!apply || queue.policy === 'exclusive') return result; + if (result.pendingJobs > 0) { + throw new Error( + `${queueName} still has ${result.pendingJobs} queued, deferred or active jobs` + ); + } + + if (queue.partition) { + // Dedicated partitions only have the index for their original policy. + // Match pg-boss 12.1.1's exclusive index; identifiers are quoted by postgres. + await tx` + CREATE UNIQUE INDEX ${tx(`${queue.table_name}_i6`)} + ON ${tx(`pgboss.${queue.table_name}`)} (name, COALESCE(singleton_key, '')) + WHERE state <= 'active' AND policy = 'exclusive' + `; + } else { + // The shared partition already has every policy index in schema 26. + const [index] = await tx` + SELECT indisvalid AND indisunique AS valid FROM pg_index + WHERE indexrelid = to_regclass('pgboss.job_i6') + AND indrelid = to_regclass('pgboss.job_common') + `; + if (!index?.valid) throw new Error('The pg-boss exclusive index is missing or invalid'); + } + await tx`UPDATE pgboss.queue SET policy = 'exclusive' WHERE name = ${queueName}`; + // Keep IDs, payloads, errors, states, retry counters, routing and deadlines. + // Updating historical policy also enforces dedupe if a job is later retried. + await tx`UPDATE pgboss.job SET policy = 'exclusive' WHERE name = ${queueName}`; + return { ...result, changed: true }; + }); +} diff --git a/src/lib/queue.ts b/src/lib/queue.ts index da80ec87..76be3821 100644 --- a/src/lib/queue.ts +++ b/src/lib/queue.ts @@ -91,19 +91,6 @@ const EXPORT_QUEUE_OPTIONS = { * migration instead of deleting a queue during API/worker startup. */ export async function ensureExportQueues(boss: PgBoss): Promise { - const existing = await boss.getQueue(QUEUE_NAMES.USFM_EXPORT); - if (existing && existing.policy !== 'exclusive') { - logger.warn( - { - event: 'worker_queue_policy_mismatch', - queueName: QUEUE_NAMES.USFM_EXPORT, - previousPolicy: existing.policy, - expectedPolicy: 'exclusive', - }, - 'Worker queue policy differs; preserving jobs until an explicit migration' - ); - } - await ensureWorkerQueue(boss, QUEUE_NAMES.USFM_EXPORT, { policy: 'exclusive', ...EXPORT_QUEUE_OPTIONS, From 4faf47d001ea7d8e47433d6e6774fc7d4b5c0680 Mon Sep 17 00:00:00 2001 From: henrique221 Date: Mon, 14 Sep 2026 14:46:48 -0300 Subject: [PATCH 3/7] refactor(queues): avoid redundant DLQ updates --- src/lib/dead-letter-queues.test.ts | 93 +++++++++++++++++---- src/lib/dead-letter-queues.ts | 32 ++++++- src/lib/dead-letter-telemetry.test.ts | 19 +++-- src/lib/exclusive-worker-queue-migration.ts | 4 +- 4 files changed, 123 insertions(+), 25 deletions(-) diff --git a/src/lib/dead-letter-queues.test.ts b/src/lib/dead-letter-queues.test.ts index 40283a3d..94ce762c 100644 --- a/src/lib/dead-letter-queues.test.ts +++ b/src/lib/dead-letter-queues.test.ts @@ -17,8 +17,11 @@ vi.mock('@/lib/logger', () => ({ })); function fakeBoss() { - const executeSql = vi.fn().mockResolvedValue({ - rows: [{ depth: 0, queuedCount: 0, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }], + const executeSql = vi.fn().mockImplementation(async (query: string) => { + if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; + return { + rows: [{ depth: 0, queuedCount: 0, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }], + }; }); const methods = { getQueue: vi.fn().mockResolvedValue(null), @@ -66,16 +69,31 @@ describe('worker queue convention', () => { }); fake.getQueue.mockResolvedValue({ retentionSeconds: 6_000_000, deleteAfterSeconds: 7_000_000 }); await ensureWorkerQueue(fake.boss, 'ingestion', { policy: 'exclusive', retryLimit: 3 }); - expect(fake.updateQueue).toHaveBeenCalledWith('ingestion-dlq', { - retentionSeconds: 6_000_000, - deleteAfterSeconds: 7_000_000, - }); + expect(fake.updateQueue).not.toHaveBeenCalledWith('ingestion-dlq', expect.anything()); expect(fake.updateQueue).toHaveBeenCalledWith('ingestion', { retryLimit: 3, deadLetter: 'ingestion-dlq', }); }); + it('skips queue writes when mutable options already match', async () => { + const fake = fakeBoss(); + fake.getQueue.mockImplementation(async (name) => + name === 'ingestion-dlq' + ? { + name, + retentionSeconds: DLQ_RETENTION_SECONDS, + deleteAfterSeconds: DLQ_RETENTION_SECONDS, + } + : { name, retryLimit: 3, deadLetter: 'ingestion-dlq' } + ); + + await ensureWorkerQueue(fake.boss, 'ingestion', { retryLimit: 3 }); + + expect(fake.createQueue).not.toHaveBeenCalled(); + expect(fake.updateQueue).not.toHaveBeenCalled(); + }); + it('keeps legacy export queues even if only diagnostic history remains', async () => { const fake = fakeBoss(); fake.getQueue.mockImplementation(async (name) => @@ -135,11 +153,15 @@ describe('dead-letter monitoring', () => { const fake = fakeBoss(); fake.getQueues.mockResolvedValue([{ name: 'slow-dlq' }, { name: 'fast-dlq' }]); let finish!: () => void; - fake.executeSql.mockReturnValueOnce( - new Promise((resolve) => { - finish = () => resolve({ rows: [{ depth: 0 }] }); - }) - ); + fake.executeSql.mockImplementation(async (query: string, parameters?: unknown[]) => { + if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; + if (parameters?.[0] === 'slow-dlq') { + return new Promise((resolve) => { + finish = () => resolve({ rows: [{ depth: 0 }] }); + }); + } + return { rows: [{ depth: 0 }] }; + }); const report = reportDeadLetterQueues(fake.boss); await vi.waitFor(() => expect(logger.info).toHaveBeenCalledWith( @@ -159,11 +181,19 @@ describe('dead-letter monitoring', () => { { name: 'old-dlq' }, { name: 'failures' }, ]); - fake.executeSql.mockResolvedValue({ - rows: [{ depth: 3, queuedCount: 2, activeCount: 1, deferredCount: 2, oldestCreatedOn: null }], + fake.executeSql.mockImplementation(async (query: string) => { + if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; + return { + rows: [ + { depth: 3, queuedCount: 2, activeCount: 1, deferredCount: 2, oldestCreatedOn: null }, + ], + }; }); await reportDeadLetterQueues(fake.boss); - expect(fake.executeSql.mock.calls.map((call) => call[1])).toEqual([['failures'], ['old-dlq']]); + expect(fake.executeSql.mock.calls.slice(1).map((call) => call[1])).toEqual([ + ['failures'], + ['old-dlq'], + ]); expect(logger.warn).toHaveBeenCalledWith( expect.objectContaining({ event: 'worker_dlq_depth', queueName: 'failures', depth: 3 }), expect.any(String) @@ -185,7 +215,15 @@ describe('dead-letter monitoring', () => { it('continues after a queue read failure and reports discovery failures', async () => { const fake = fakeBoss(); fake.getQueues.mockResolvedValue([{ name: 'one-dlq' }, { name: 'two-dlq' }]); - fake.executeSql.mockRejectedValueOnce(new Error('database read failed')); + fake.executeSql.mockImplementation(async (query: string, parameters?: unknown[]) => { + if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; + if (parameters?.[0] === 'one-dlq') throw new Error('database read failed'); + return { + rows: [ + { depth: 0, queuedCount: 0, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }, + ], + }; + }); await reportDeadLetterQueues(fake.boss); expect(logger.error).toHaveBeenCalledWith( expect.objectContaining({ event: 'worker_dlq_monitor_error', queueName: 'one-dlq' }), @@ -206,6 +244,24 @@ describe('dead-letter monitoring', () => { ); }); + it('reports an explicit error and skips job queries for an unsupported schema', async () => { + const fake = fakeBoss(); + fake.executeSql.mockResolvedValue({ rows: [{ version: 25 }] }); + + await reportDeadLetterQueues(fake.boss); + + expect(fake.getQueues).not.toHaveBeenCalled(); + expect(fake.executeSql).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith( + { + event: 'worker_dlq_monitor_schema_mismatch', + expectedSchemaVersion: 26, + actualSchemaVersion: 25, + }, + 'Worker dead-letter queue monitoring requires pg-boss schema version 26' + ); + }); + it('sweeps immediately, avoids overlap, waits for shutdown and stops its timer', async () => { vi.useFakeTimers(); const fake = fakeBoss(); @@ -216,6 +272,7 @@ describe('dead-letter monitoring', () => { }) ); const stop = startDeadLetterMonitor(fake.boss); + await Promise.resolve(); expect(fake.getQueues).toHaveBeenCalledTimes(1); await vi.advanceTimersByTimeAsync(120_000); expect(fake.getQueues).toHaveBeenCalledTimes(1); @@ -245,7 +302,11 @@ describe('dead-letter monitoring', () => { vi.useFakeTimers(); const fake = fakeBoss(); fake.getQueues.mockResolvedValue([{ name: 'stuck-dlq' }]); - fake.executeSql.mockReturnValue(new Promise(() => {})); + fake.executeSql.mockImplementation((query: string) => + query.includes('pgboss.version') + ? Promise.resolve({ rows: [{ version: 26 }] }) + : new Promise(() => {}) + ); const stop = startDeadLetterMonitor(fake.boss); await vi.advanceTimersByTimeAsync(0); const shutdown = stop(); diff --git a/src/lib/dead-letter-queues.ts b/src/lib/dead-letter-queues.ts index 83c08861..dade5914 100644 --- a/src/lib/dead-letter-queues.ts +++ b/src/lib/dead-letter-queues.ts @@ -5,6 +5,12 @@ import { logger } from '@/lib/logger'; /** Time to investigate new DLQ entries before pg-boss maintenance removes them. */ export const DLQ_RETENTION_SECONDS = 30 * 24 * 60 * 60; export const DLQ_SHUTDOWN_TIMEOUT_MS = 5_000; +const PG_BOSS_SCHEMA_VERSION = 26; + +function queueMatchesOptions(queue: Queue, options: Partial): boolean { + const current = queue as unknown as Record; + return Object.entries(options).every(([key, value]) => current[key] === value); +} /** Create a durable diagnostic destination before enabling dead-letter routing. */ export async function ensureWorkerQueue( @@ -21,7 +27,9 @@ export async function ensureWorkerQueue( deleteAfterSeconds: Math.max(existing?.deleteAfterSeconds ?? 0, DLQ_RETENTION_SECONDS), }; if (existing) { - await boss.updateQueue(deadLetter, retentionOptions); + if (!queueMatchesOptions(existing, retentionOptions)) { + await boss.updateQueue(deadLetter, retentionOptions); + } } else { await boss.createQueue(deadLetter, retentionOptions); } @@ -43,12 +51,30 @@ export async function ensureWorkerQueue( } // Policy and partition are immutable in pg-boss. Preserve existing jobs. const { policy: _policy, partition: _partition, ...mutableOptions } = options; - await boss.updateQueue(name, { ...mutableOptions, deadLetter }); + const desiredOptions = { ...mutableOptions, deadLetter }; + if (!queueMatchesOptions(source, desiredOptions)) { + await boss.updateQueue(name, desiredOptions); + } } /** Report retained DLQ rows without fetching, acknowledging or replaying them. */ export async function reportDeadLetterQueues(boss: PgBoss): Promise { try { + const database = boss.getDb(); + const { rows: schemaRows } = await database.executeSql('SELECT version FROM pgboss.version'); + const schemaVersion = schemaRows[0]?.version; + if (schemaVersion !== PG_BOSS_SCHEMA_VERSION) { + logger.error( + { + event: 'worker_dlq_monitor_schema_mismatch', + expectedSchemaVersion: PG_BOSS_SCHEMA_VERSION, + actualSchemaVersion: schemaVersion ?? null, + }, + 'Worker dead-letter queue monitoring requires pg-boss schema version 26' + ); + return; + } + const queues = await boss.getQueues(); // Discover configured targets, including custom names, and orphaned/legacy // *-dlq queues. A worker added later is picked up on the next sweep. @@ -64,7 +90,7 @@ export async function reportDeadLetterQueues(boss: PgBoss): Promise { // pg-boss 12.1.1 getQueueStats falls back to cached counters when a // queue becomes empty. Read an aggregate without GROUP BY so a cleared // queue always reports zero. The parent job table includes partitions. - const { rows } = await boss.getDb().executeSql( + const { rows } = await database.executeSql( `SELECT count(*)::int AS depth, count(*) FILTER (WHERE state < 'active')::int AS "queuedCount", count(*) FILTER (WHERE state = 'active')::int AS "activeCount", diff --git a/src/lib/dead-letter-telemetry.test.ts b/src/lib/dead-letter-telemetry.test.ts index c1384e07..61878fb9 100644 --- a/src/lib/dead-letter-telemetry.test.ts +++ b/src/lib/dead-letter-telemetry.test.ts @@ -20,11 +20,20 @@ describe('dLQ Application Insights telemetry', () => { const boss = { getQueues: async () => [{ name: 'usfm-export-dlq' }], getDb: () => ({ - executeSql: async () => ({ - rows: [ - { depth: 1, queuedCount: 1, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }, - ], - }), + executeSql: async (query: string) => { + if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; + return { + rows: [ + { + depth: 1, + queuedCount: 1, + activeCount: 0, + deferredCount: 0, + oldestCreatedOn: null, + }, + ], + }; + }, }), } as unknown as PgBoss; diff --git a/src/lib/exclusive-worker-queue-migration.ts b/src/lib/exclusive-worker-queue-migration.ts index 3d346d22..3ac24678 100644 --- a/src/lib/exclusive-worker-queue-migration.ts +++ b/src/lib/exclusive-worker-queue-migration.ts @@ -1,6 +1,8 @@ import type postgres from 'postgres'; -const MIGRATABLE_QUEUES = ['usfm-export', 'ai-suggestions']; +import { QUEUE_NAMES } from '@/lib/queue'; + +const MIGRATABLE_QUEUES: string[] = [QUEUE_NAMES.USFM_EXPORT, QUEUE_NAMES.AI_SUGGESTIONS]; /** Offline, operator-invoked migration for the pinned pg-boss 12.1.1 schema. */ export async function migrateExclusiveWorkerQueue( From 425bb1ca711feb3b8d9bb624a6aa3b58ae4cd8dc Mon Sep 17 00:00:00 2001 From: henrique221 Date: Thu, 17 Sep 2026 12:17:07 -0300 Subject: [PATCH 4/7] fix(workers): close the http server before draining the dlq monitor Awaiting the monitor stop first held the HTTP listener open for up to DLQ_SHUTDOWN_TIMEOUT_MS of the orchestrator's shutdown grace period. Stop the timer, close the listener, and drain the in-flight sweep concurrently before stopping pg-boss. Refs: #324 --- docs/runbooks/worker-dead-letter-queues.md | 5 +++-- src/index.ts | 6 +++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/docs/runbooks/worker-dead-letter-queues.md b/docs/runbooks/worker-dead-letter-queues.md index 38f3acf3..adbe7f76 100644 --- a/docs/runbooks/worker-dead-letter-queues.md +++ b/docs/runbooks/worker-dead-letter-queues.md @@ -14,8 +14,9 @@ per-job event or a count of new failures. The API runs the monitor because the export WebJob refuses to boot without R2. Monitoring therefore continues when that worker cannot start. It uses the existing Pino/Application Insights logger, needs no new service or fluent-platform change, -and stops its timer and waits up to five seconds for an active sweep before -continuing API and pg-boss shutdown. A timeout logs +and stops its timer and waits up to five seconds for an active sweep. Shutdown +closes the HTTP listener first and drains that sweep concurrently, so the wait +never delays rejecting new connections, and pg-boss stops after it. A timeout logs `worker_dlq_monitor_shutdown_timeout`; it does not cancel the database query. Slow sweeps never overlap. Discovery failures and individual queue read failures emit `worker_dlq_monitor_error`; a failure in one queue does not skip the others. diff --git a/src/index.ts b/src/index.ts index 788176f1..886a83e1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -67,12 +67,16 @@ async function startServer() { logger.info(`${signal} received, shutting down server`); try { if (audioReclaimInterval) clearInterval(audioReclaimInterval); - await stopDeadLetterMonitor(); + // Stop the monitor's timer now but drain its in-flight sweep alongside + // the listener close. Awaiting it first would hold the socket open for + // up to DLQ_SHUTDOWN_TIMEOUT_MS of the orchestrator's grace period. + const monitorStopped = stopDeadLetterMonitor(); server.close(() => { logger.info('HTTP server closed'); }); + await monitorStopped; await stopQueue(); logger.info('Shutdown completed'); From 572a2af121f2aa458320fe5c2fd02143dc3ef5fb Mon Sep 17 00:00:00 2001 From: henrique221 Date: Thu, 17 Sep 2026 12:17:07 -0300 Subject: [PATCH 5/7] refactor(queues): share the pg-boss schema version and derive dlq names PG_BOSS_SCHEMA_VERSION was written out independently in the DLQ monitor and in the policy migration, so a schema bump could update one and silently disable the other. Move it to src/lib/pg-boss-schema.ts and read it from both. QUEUE_NAMES.USFM_EXPORT_DLQ was a second spelling of a name that ensureWorkerQueue already derives, and no production code used it. Drop it in favour of deadLetterQueueName(), used by the helper, the queue discovery sweep and the tests. Refs: #324 --- src/lib/dead-letter-queues.integration.test.ts | 3 ++- src/lib/dead-letter-queues.ts | 14 ++++++++++---- src/lib/exclusive-worker-queue-migration.ts | 7 +++++-- src/lib/pg-boss-schema.ts | 6 ++++++ src/lib/queue.ts | 4 ++-- 5 files changed, 25 insertions(+), 9 deletions(-) create mode 100644 src/lib/pg-boss-schema.ts diff --git a/src/lib/dead-letter-queues.integration.test.ts b/src/lib/dead-letter-queues.integration.test.ts index d092d5c8..df8a7922 100644 --- a/src/lib/dead-letter-queues.integration.test.ts +++ b/src/lib/dead-letter-queues.integration.test.ts @@ -6,6 +6,7 @@ import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { createUSFMZipStreamAsync, getProjectName } from '@/domains/usfm/usfm.service'; import { uploadExportStream } from '@/lib/blob-storage'; import { + deadLetterQueueName, DLQ_RETENTION_SECONDS, ensureWorkerQueue, reportDeadLetterQueues, @@ -34,7 +35,7 @@ describe.skipIf(!connectionString)('dead-letter queues with PostgreSQL and pg-bo let migrationSql: postgres.Sql; const errors: Error[] = []; const exportQueue = QUEUE_NAMES.USFM_EXPORT; - const exportDlq = QUEUE_NAMES.USFM_EXPORT_DLQ; + const exportDlq = deadLetterQueueName(exportQueue); async function rows(name: string) { return ( diff --git a/src/lib/dead-letter-queues.ts b/src/lib/dead-letter-queues.ts index dade5914..d88747c1 100644 --- a/src/lib/dead-letter-queues.ts +++ b/src/lib/dead-letter-queues.ts @@ -1,11 +1,17 @@ import type { PgBoss, Queue } from 'pg-boss'; import { logger } from '@/lib/logger'; +import { PG_BOSS_SCHEMA_VERSION } from '@/lib/pg-boss-schema'; /** Time to investigate new DLQ entries before pg-boss maintenance removes them. */ export const DLQ_RETENTION_SECONDS = 30 * 24 * 60 * 60; export const DLQ_SHUTDOWN_TIMEOUT_MS = 5_000; -const PG_BOSS_SCHEMA_VERSION = 26; +const DEAD_LETTER_SUFFIX = '-dlq'; + +/** The single spelling of a source queue's dead-letter destination. */ +export function deadLetterQueueName(name: string): string { + return `${name}${DEAD_LETTER_SUFFIX}`; +} function queueMatchesOptions(queue: Queue, options: Partial): boolean { const current = queue as unknown as Record; @@ -18,7 +24,7 @@ export async function ensureWorkerQueue( name: string, options: Omit = {} ): Promise { - const deadLetter = `${name}-dlq`; + const deadLetter = deadLetterQueueName(name); const [existing, source] = await Promise.all([boss.getQueue(deadLetter), boss.getQueue(name)]); const retentionOptions = { // Do not shorten an operator's longer retention policy. Queue updates only @@ -70,7 +76,7 @@ export async function reportDeadLetterQueues(boss: PgBoss): Promise { expectedSchemaVersion: PG_BOSS_SCHEMA_VERSION, actualSchemaVersion: schemaVersion ?? null, }, - 'Worker dead-letter queue monitoring requires pg-boss schema version 26' + `Worker dead-letter queue monitoring requires pg-boss schema version ${PG_BOSS_SCHEMA_VERSION}` ); return; } @@ -81,7 +87,7 @@ export async function reportDeadLetterQueues(boss: PgBoss): Promise { const targets = new Set(); for (const queue of queues) { if (queue.deadLetter) targets.add(queue.deadLetter); - if (queue.name.endsWith('-dlq')) targets.add(queue.name); + if (queue.name.endsWith(DEAD_LETTER_SUFFIX)) targets.add(queue.name); } await Promise.allSettled( diff --git a/src/lib/exclusive-worker-queue-migration.ts b/src/lib/exclusive-worker-queue-migration.ts index 3ac24678..14ba25a8 100644 --- a/src/lib/exclusive-worker-queue-migration.ts +++ b/src/lib/exclusive-worker-queue-migration.ts @@ -1,5 +1,6 @@ import type postgres from 'postgres'; +import { PG_BOSS_SCHEMA_VERSION } from '@/lib/pg-boss-schema'; import { QUEUE_NAMES } from '@/lib/queue'; const MIGRATABLE_QUEUES: string[] = [QUEUE_NAMES.USFM_EXPORT, QUEUE_NAMES.AI_SUGGESTIONS]; @@ -23,8 +24,10 @@ export async function migrateExclusiveWorkerQueue( await tx`LOCK TABLE pgboss.queue, pgboss.job IN ACCESS EXCLUSIVE MODE`; } const [version] = await tx`SELECT version FROM pgboss.version`; - if (version?.version !== 26) { - throw new Error('Migration requires the pg-boss 12.1.1 schema version 26'); + if (version?.version !== PG_BOSS_SCHEMA_VERSION) { + throw new Error( + `Migration requires the pg-boss 12.1.1 schema version ${PG_BOSS_SCHEMA_VERSION}` + ); } const [queue] = await tx` SELECT policy, partition, table_name FROM pgboss.queue WHERE name = ${queueName} diff --git a/src/lib/pg-boss-schema.ts b/src/lib/pg-boss-schema.ts new file mode 100644 index 00000000..e56d8377 --- /dev/null +++ b/src/lib/pg-boss-schema.ts @@ -0,0 +1,6 @@ +/** + * Pinned pg-boss 12.1.1 schema. The dead-letter monitor and the queue policy + * migration both read raw `pgboss.*` tables, so a schema bump has to move them + * together instead of silently disabling one of them. + */ +export const PG_BOSS_SCHEMA_VERSION = 26; diff --git a/src/lib/queue.ts b/src/lib/queue.ts index 76be3821..a79ca1ac 100644 --- a/src/lib/queue.ts +++ b/src/lib/queue.ts @@ -5,10 +5,10 @@ import { logger } from '@/lib/logger'; let boss: PgBoss | null = null; +// Dead-letter destinations are not listed here: deadLetterQueueName() derives +// them from the source name, so there is a single spelling of each. export const QUEUE_NAMES = { USFM_EXPORT: 'usfm-export', - /** Dead-letter destination for exports that exhaust their retries. */ - USFM_EXPORT_DLQ: 'usfm-export-dlq', AI_SUGGESTIONS: 'ai-suggestions', DBL_INGEST_TEXT: 'dbl-ingest-text', DBL_INGEST_TEXT_PRIORITY: 'dbl-ingest-text-priority', From 9cd10bd213af481fc9288d48cb892b97e437b78f Mon Sep 17 00:00:00 2001 From: henrique221 Date: Thu, 17 Sep 2026 12:17:07 -0300 Subject: [PATCH 6/7] fix(queues): check the queue policy before locking the pgboss tables The ACCESS EXCLUSIVE lock on pgboss.queue and pgboss.job was taken before the already-exclusive check, so an inspection or the documented no-op re-run still stalled every queue's fetch, complete and send for up to the five-second lock timeout. Read the policy first and return early, then re-read it under the lock so a concurrent migration cannot slip through. Refs: #324 --- docs/runbooks/worker-dead-letter-queues.md | 17 +++-- .../dead-letter-queues.integration.test.ts | 28 ++++++++ src/lib/exclusive-worker-queue-migration.ts | 67 +++++++++++-------- 3 files changed, 78 insertions(+), 34 deletions(-) diff --git a/docs/runbooks/worker-dead-letter-queues.md b/docs/runbooks/worker-dead-letter-queues.md index adbe7f76..f328ca17 100644 --- a/docs/runbooks/worker-dead-letter-queues.md +++ b/docs/runbooks/worker-dead-letter-queues.md @@ -107,19 +107,22 @@ Do not paste the connection string into logs or commit it. Restarting clears pg-boss's cached policy. Verify a normal request completes and the mismatch warning no longer appears. -The apply transaction locks the queue and job tables, including partitions, and -refuses any queued, deferred, retrying or active work. Lock acquisition is limited -to five seconds and statements to thirty seconds; an error rolls back the whole -migration. The maintenance window affects all queues because the job table lock -covers their partitions. Retry only after checking the reported blocker. +Both modes read the queue's current policy before locking, so an inspection and a +re-run after success take no lock at all. Only an apply that still has work to do +locks the queue and job tables, including partitions; it then re-reads the policy +under the lock and refuses any queued, deferred, retrying or active work. Lock +acquisition is limited to five seconds and statements to thirty seconds; an error +rolls back the whole migration. The maintenance window affects all queues because +the job table lock covers their partitions. Retry only after checking the reported +blocker. No queue or job is deleted. The migration changes the queue's policy and retained jobs' policy metadata so a later operator retry also respects singleton dedupe. IDs, payloads, outputs, states, retry counters, routing and original deadlines stay unchanged; DLQ rows are untouched. Dedicated partitions receive the exclusive index; the shared partition's existing index is checked. Re-running after success -is a no-op. Do not switch back to a non-exclusive policy as an application rollback; -older binaries already expect exclusive dedupe. +is a lock-free no-op. Do not switch back to a non-exclusive policy as an +application rollback; older binaries already expect exclusive dedupe. ## Investigate and recover diff --git a/src/lib/dead-letter-queues.integration.test.ts b/src/lib/dead-letter-queues.integration.test.ts index df8a7922..a7356edd 100644 --- a/src/lib/dead-letter-queues.integration.test.ts +++ b/src/lib/dead-letter-queues.integration.test.ts @@ -132,6 +132,34 @@ describe.skipIf(!connectionString)('dead-letter queues with PostgreSQL and pg-bo migrateExclusiveWorkerQueue(migrationSql, exportQueue, true) ).resolves.toMatchObject({ changed: false }); + // The no-op re-run must decide before taking ACCESS EXCLUSIVE. Another + // session holding the lock an ordinary reader takes proves it never waits: + // locking first would fail here on the five-second lock_timeout. + const reader = postgres(connectionString!, { max: 1 }); + let holding!: () => void; + const held = new Promise((resolve) => { + holding = resolve; + }); + let releaseReader!: () => void; + const released = new Promise((resolve) => { + releaseReader = resolve; + }); + const readerTransaction = reader.begin(async (tx) => { + await tx`LOCK TABLE pgboss.queue, pgboss.job IN ACCESS SHARE MODE`; + holding(); + await released; + }); + try { + await held; + await expect( + migrateExclusiveWorkerQueue(migrationSql, exportQueue, true) + ).resolves.toMatchObject({ changed: false }); + } finally { + releaseReader(); + await readerTransaction; + await reader.end(); + } + const key = { singletonKey: 'migration-dedupe-proof' }; const first = await boss.send(exportQueue, { fixture: 'first' }, key); expect(first).toBeTruthy(); diff --git a/src/lib/exclusive-worker-queue-migration.ts b/src/lib/exclusive-worker-queue-migration.ts index 14ba25a8..c9ff1b72 100644 --- a/src/lib/exclusive-worker-queue-migration.ts +++ b/src/lib/exclusive-worker-queue-migration.ts @@ -18,35 +18,48 @@ export async function migrateExclusiveWorkerQueue( return sql.begin(async (tx) => { await tx`SET LOCAL lock_timeout = '5s'`; await tx`SET LOCAL statement_timeout = '30s'`; - if (apply) { - // Includes all partitions. No sender, worker or maintenance process can - // change jobs between the pending-work check and the policy update. - await tx`LOCK TABLE pgboss.queue, pgboss.job IN ACCESS EXCLUSIVE MODE`; - } - const [version] = await tx`SELECT version FROM pgboss.version`; - if (version?.version !== PG_BOSS_SCHEMA_VERSION) { - throw new Error( - `Migration requires the pg-boss 12.1.1 schema version ${PG_BOSS_SCHEMA_VERSION}` - ); - } - const [queue] = await tx` - SELECT policy, partition, table_name FROM pgboss.queue WHERE name = ${queueName} - `; - if (!queue) throw new Error(`Queue ${queueName} does not exist`); + const readQueueState = async () => { + const [version] = await tx`SELECT version FROM pgboss.version`; + if (version?.version !== PG_BOSS_SCHEMA_VERSION) { + throw new Error( + `Migration requires the pg-boss 12.1.1 schema version ${PG_BOSS_SCHEMA_VERSION}` + ); + } + const [queue] = await tx` + SELECT policy, partition, table_name FROM pgboss.queue WHERE name = ${queueName} + `; + if (!queue) throw new Error(`Queue ${queueName} does not exist`); - const [stats] = await tx` - SELECT count(*)::int AS retained, - count(*) FILTER (WHERE state <= 'active')::int AS pending - FROM pgboss.job WHERE name = ${queueName} - `; - const result = { - queueName, - previousPolicy: queue.policy as string, - retainedJobs: stats.retained as number, - pendingJobs: stats.pending as number, - changed: false, + const [stats] = await tx` + SELECT count(*)::int AS retained, + count(*) FILTER (WHERE state <= 'active')::int AS pending + FROM pgboss.job WHERE name = ${queueName} + `; + return { + queue, + result: { + queueName, + previousPolicy: queue.policy as string, + retainedJobs: stats.retained as number, + pendingJobs: stats.pending as number, + changed: false, + }, + }; }; - if (!apply || queue.policy === 'exclusive') return result; + + // Read the policy before locking. An inspection, or a re-run after a + // successful migration, must not stall every queue's fetch/complete/send + // for the lock timeout just to discover there is nothing to do. + const preflight = await readQueueState(); + if (!apply || preflight.queue.policy === 'exclusive') return preflight.result; + + // Includes all partitions. No sender, worker or maintenance process can + // change jobs between the pending-work check and the policy update. + await tx`LOCK TABLE pgboss.queue, pgboss.job IN ACCESS EXCLUSIVE MODE`; + // The unlocked pre-check can race another migration, so decide again on the + // state this lock now protects. + const { queue, result } = await readQueueState(); + if (queue.policy === 'exclusive') return result; if (result.pendingJobs > 0) { throw new Error( `${queueName} still has ${result.pendingJobs} queued, deferred or active jobs` From f47408f01f78af57d83eaea19fe0ed86b3e587d6 Mon Sep 17 00:00:00 2001 From: henrique221 Date: Fri, 18 Sep 2026 15:33:55 -0300 Subject: [PATCH 7/7] fix(queues): address custom routing and typed mock review Preserve configured dead-letter destinations and retained diagnostics. Use typed pg-boss test helpers and restrict the DLQ integration job token. Refs: #324 --- .github/workflows/pre-merge.yml | 4 + docs/runbooks/worker-dead-letter-queues.md | 10 +- .../dead-letter-queues.integration.test.ts | 51 +++++++++- src/lib/dead-letter-queues.test.ts | 85 +++++++++-------- src/lib/dead-letter-queues.ts | 7 +- src/lib/dead-letter-telemetry.test.ts | 32 +++---- src/test/utils/test-helpers.ts | 56 +++++++++++ src/workers/dbl-sync.worker.test.ts | 18 ++-- src/workers/ingest-bible-text.worker.test.ts | 92 +++++++------------ src/workers/usfm-export.worker.test.ts | 46 ++++------ 10 files changed, 231 insertions(+), 170 deletions(-) diff --git a/.github/workflows/pre-merge.yml b/.github/workflows/pre-merge.yml index 9f59246c..bafb632f 100644 --- a/.github/workflows/pre-merge.yml +++ b/.github/workflows/pre-merge.yml @@ -40,6 +40,8 @@ jobs: dlq-integration: name: Dead-letter queue integration + permissions: + contents: read runs-on: ubuntu-latest timeout-minutes: 10 if: ${{ !github.event.pull_request.draft }} @@ -60,6 +62,8 @@ jobs: steps: - name: Checkout repository uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false - name: Set up Node.js version uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 diff --git a/docs/runbooks/worker-dead-letter-queues.md b/docs/runbooks/worker-dead-letter-queues.md index f328ca17..3b24e553 100644 --- a/docs/runbooks/worker-dead-letter-queues.md +++ b/docs/runbooks/worker-dead-letter-queues.md @@ -31,10 +31,12 @@ still need to be configured by the environment owner. ## Queue convention Use `ensureWorkerQueue(boss, name, options)` before sending or consuming jobs. -It creates `-dlq` first and updates the source's `deadLetter` setting even -when the source already exists. Export and AI retry settings stay at three retries -with 60-second exponential backoff. DBL queues keep their current retry settings -(pg-boss defaults for new queues). +It preserves an existing source's custom `deadLetter` destination so retained +jobs in that queue remain visible to the monitor. Sources without a destination +use `-dlq`. The destination is created or updated before enabling routing. +Export and AI retry settings stay at three retries with 60-second exponential +backoff. DBL queues keep their current retry settings (pg-boss defaults for new +queues). This applies to `usfm-export`, `ai-suggestions` (formerly `ai-suggestion-trigger`), both `dbl-ingest-text` queues, and `dbl-sync` when its diff --git a/src/lib/dead-letter-queues.integration.test.ts b/src/lib/dead-letter-queues.integration.test.ts index a7356edd..cad2c92f 100644 --- a/src/lib/dead-letter-queues.integration.test.ts +++ b/src/lib/dead-letter-queues.integration.test.ts @@ -3,6 +3,8 @@ import { PgBoss } from 'pg-boss'; import postgres from 'postgres'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; +import type { ExportResult } from '@/domains/usfm/usfm.types'; + import { createUSFMZipStreamAsync, getProjectName } from '@/domains/usfm/usfm.service'; import { uploadExportStream } from '@/lib/blob-storage'; import { @@ -96,6 +98,48 @@ describe.skipIf(!connectionString)('dead-letter queues with PostgreSQL and pg-bo }); }); + it('preserves custom routing and monitors retained and new failures after setup', async () => { + const source = 'custom-routing-probe'; + const destination = 'custom-failures'; + await boss.createQueue(destination, { + retentionSeconds: DLQ_RETENTION_SECONDS * 2, + deleteAfterSeconds: DLQ_RETENTION_SECONDS * 2, + }); + await boss.createQueue(source, { + policy: 'standard', + deadLetter: destination, + retryLimit: 0, + }); + await boss.send(destination, { fixture: 'retained custom failure' }); + const before = await rows(destination); + + await ensureWorkerQueue(boss, source, { policy: 'exclusive', retryLimit: 0 }); + await ensureWorkerQueue(boss, source, { policy: 'exclusive', retryLimit: 0 }); + + expect(await boss.getQueue(source)).toMatchObject({ + policy: 'standard', + deadLetter: destination, + }); + expect(await boss.getQueue(deadLetterQueueName(source))).toBeNull(); + expect(await boss.getQueue(destination)).toMatchObject({ + retentionSeconds: DLQ_RETENTION_SECONDS * 2, + deleteAfterSeconds: DLQ_RETENTION_SECONDS * 2, + }); + expect(await rows(destination)).toEqual(before); + + const id = await boss.send(source, { fixture: 'new custom failure' }); + await boss.fetch(source); + await boss.fail(source, id!, new Error('custom destination failure')); + const retained = await rows(destination); + expect(retained).toHaveLength(2); + await reportDeadLetterQueues(boss); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ event: 'worker_dlq_depth', queueName: destination, depth: 2 }), + expect.any(String) + ); + expect(await rows(destination)).toEqual(retained); + }); + it('migrates a legacy queue without losing history and enforces singleton dedupe', async () => { const deferredId = await boss.send( exportQueue, @@ -220,9 +264,10 @@ describe.skipIf(!connectionString)('dead-letter queues with PostgreSQL and pg-bo it('runs the real export worker through an upload failure and exhausted retries', async () => { await boss.updateQueue(exportQueue, { retryLimit: 1, retryDelay: 0, retryBackoff: false }); - vi.mocked(createUSFMZipStreamAsync).mockImplementation( - async () => ({ ok: true, data: { stream: Readable.from(['zip']), cleanup: vi.fn() } }) as any - ); + vi.mocked(createUSFMZipStreamAsync).mockImplementation(async () => { + const data: ExportResult = { stream: Readable.from(['zip']), cleanup: vi.fn() }; + return { ok: true, data }; + }); vi.mocked(uploadExportStream).mockRejectedValue(new Error('simulated R2 outage')); await registerUSFMExportWorker(boss); const payload = { projectUnitId: 30, requestedBy: 7 }; diff --git a/src/lib/dead-letter-queues.test.ts b/src/lib/dead-letter-queues.test.ts index 94ce762c..880ebe5c 100644 --- a/src/lib/dead-letter-queues.test.ts +++ b/src/lib/dead-letter-queues.test.ts @@ -1,5 +1,3 @@ -import type { PgBoss } from 'pg-boss'; - import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { @@ -11,36 +9,21 @@ import { } from '@/lib/dead-letter-queues'; import { logger } from '@/lib/logger'; import { ensureAiSuggestionQueue, ensureExportQueues } from '@/lib/queue'; +import { fakeBoss, queueResult } from '@/test/utils/test-helpers'; vi.mock('@/lib/logger', () => ({ logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, })); -function fakeBoss() { - const executeSql = vi.fn().mockImplementation(async (query: string) => { - if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; - return { - rows: [{ depth: 0, queuedCount: 0, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }], - }; - }); - const methods = { - getQueue: vi.fn().mockResolvedValue(null), - createQueue: vi.fn(), - updateQueue: vi.fn(), - deleteQueue: vi.fn(), - getQueues: vi.fn().mockResolvedValue([]), - getDb: () => ({ executeSql }), - }; - return { boss: methods as unknown as PgBoss, ...methods, executeSql }; -} - beforeEach(() => vi.clearAllMocks()); afterEach(() => vi.useRealTimers()); describe('worker queue convention', () => { it('creates the DLQ first and converges routing on an existing source queue', async () => { const fake = fakeBoss(); - fake.getQueue.mockImplementation(async (name) => (name === 'ingestion' ? { name } : null)); + fake.getQueue.mockImplementation(async (name) => + name === 'ingestion' ? queueResult(name) : null + ); await ensureWorkerQueue(fake.boss, 'ingestion'); expect(fake.createQueue.mock.calls).toEqual([ [ @@ -52,6 +35,29 @@ describe('worker queue convention', () => { expect(fake.deleteQueue).not.toHaveBeenCalled(); }); + it('preserves a custom destination and raises only its short retention settings', async () => { + const fake = fakeBoss(); + fake.getQueue.mockImplementation(async (name) => + name === 'ingestion' + ? queueResult(name, { deadLetter: 'custom-failures', retryLimit: 3 }) + : queueResult(name, { retentionSeconds: 60, deleteAfterSeconds: DLQ_RETENTION_SECONDS * 2 }) + ); + + await ensureWorkerQueue(fake.boss, 'ingestion', { retryLimit: 3 }); + + expect(fake.getQueue.mock.calls).toEqual([['ingestion'], ['custom-failures']]); + expect(fake.createQueue).not.toHaveBeenCalled(); + expect(fake.updateQueue.mock.calls).toEqual([ + [ + 'custom-failures', + { + retentionSeconds: DLQ_RETENTION_SECONDS, + deleteAfterSeconds: DLQ_RETENTION_SECONDS * 2, + }, + ], + ]); + }); + it('creates new queues with their final settings without redundant updates', async () => { const fake = fakeBoss(); await ensureWorkerQueue(fake.boss, 'ingestion', { retryLimit: 3 }); @@ -64,10 +70,12 @@ describe('worker queue convention', () => { it('preserves longer retention and never updates immutable source policy', async () => { const fake = fakeBoss(); - fake.createQueue.mockImplementation((_name, options) => { - options.policy ??= 'standard'; + fake.createQueue.mockImplementation(async (_name, options) => { + if (options) options.policy ??= 'standard'; }); - fake.getQueue.mockResolvedValue({ retentionSeconds: 6_000_000, deleteAfterSeconds: 7_000_000 }); + fake.getQueue.mockImplementation(async (name) => + queueResult(name, { retentionSeconds: 6_000_000, deleteAfterSeconds: 7_000_000 }) + ); await ensureWorkerQueue(fake.boss, 'ingestion', { policy: 'exclusive', retryLimit: 3 }); expect(fake.updateQueue).not.toHaveBeenCalledWith('ingestion-dlq', expect.anything()); expect(fake.updateQueue).toHaveBeenCalledWith('ingestion', { @@ -80,12 +88,11 @@ describe('worker queue convention', () => { const fake = fakeBoss(); fake.getQueue.mockImplementation(async (name) => name === 'ingestion-dlq' - ? { - name, + ? queueResult(name, { retentionSeconds: DLQ_RETENTION_SECONDS, deleteAfterSeconds: DLQ_RETENTION_SECONDS, - } - : { name, retryLimit: 3, deadLetter: 'ingestion-dlq' } + }) + : queueResult(name, { retryLimit: 3, deadLetter: 'ingestion-dlq' }) ); await ensureWorkerQueue(fake.boss, 'ingestion', { retryLimit: 3 }); @@ -97,7 +104,7 @@ describe('worker queue convention', () => { it('keeps legacy export queues even if only diagnostic history remains', async () => { const fake = fakeBoss(); fake.getQueue.mockImplementation(async (name) => - name === 'usfm-export' ? { policy: 'standard' } : null + name === 'usfm-export' ? queueResult(name, { policy: 'standard' }) : null ); await ensureExportQueues(fake.boss); expect(fake.deleteQueue).not.toHaveBeenCalled(); @@ -117,7 +124,7 @@ describe('worker queue convention', () => { it('adds a DLQ to the current AI queue and keeps its retry contract', async () => { const fake = fakeBoss(); fake.getQueue.mockImplementation(async (name) => - name === 'ai-suggestions' ? { policy: 'standard' } : null + name === 'ai-suggestions' ? queueResult(name, { policy: 'standard' }) : null ); await ensureAiSuggestionQueue(fake.boss); expect(fake.updateQueue).toHaveBeenCalledWith('ai-suggestions', { @@ -141,7 +148,7 @@ describe('worker queue convention', () => { it('does not warn for an existing exclusive queue', async () => { const fake = fakeBoss(); fake.getQueue.mockImplementation(async (name) => - name === 'ai-suggestions' ? { policy: 'exclusive' } : null + name === 'ai-suggestions' ? queueResult(name, { policy: 'exclusive' }) : null ); await ensureAiSuggestionQueue(fake.boss); expect(logger.warn).not.toHaveBeenCalled(); @@ -151,7 +158,7 @@ describe('worker queue convention', () => { describe('dead-letter monitoring', () => { it('reports another queue while the first query is still pending', async () => { const fake = fakeBoss(); - fake.getQueues.mockResolvedValue([{ name: 'slow-dlq' }, { name: 'fast-dlq' }]); + fake.getQueues.mockResolvedValue([queueResult('slow-dlq'), queueResult('fast-dlq')]); let finish!: () => void; fake.executeSql.mockImplementation(async (query: string, parameters?: unknown[]) => { if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; @@ -176,10 +183,10 @@ describe('dead-letter monitoring', () => { it('discovers custom, shared and orphaned DLQs without counting source retries', async () => { const fake = fakeBoss(); fake.getQueues.mockResolvedValue([ - { name: 'export', deadLetter: 'failures' }, - { name: 'ai', deadLetter: 'failures' }, - { name: 'old-dlq' }, - { name: 'failures' }, + queueResult('export', { deadLetter: 'failures' }), + queueResult('ai', { deadLetter: 'failures' }), + queueResult('old-dlq'), + queueResult('failures'), ]); fake.executeSql.mockImplementation(async (query: string) => { if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; @@ -204,7 +211,7 @@ describe('dead-letter monitoring', () => { it('reports zero after the backlog clears, with flat structured dimensions', async () => { const fake = fakeBoss(); - fake.getQueues.mockResolvedValue([{ name: 'old-dlq' }]); + fake.getQueues.mockResolvedValue([queueResult('old-dlq')]); await reportDeadLetterQueues(fake.boss); expect(logger.info).toHaveBeenCalledWith( expect.objectContaining({ event: 'worker_dlq_depth', queueName: 'old-dlq', depth: 0 }), @@ -214,7 +221,7 @@ describe('dead-letter monitoring', () => { it('continues after a queue read failure and reports discovery failures', async () => { const fake = fakeBoss(); - fake.getQueues.mockResolvedValue([{ name: 'one-dlq' }, { name: 'two-dlq' }]); + fake.getQueues.mockResolvedValue([queueResult('one-dlq'), queueResult('two-dlq')]); fake.executeSql.mockImplementation(async (query: string, parameters?: unknown[]) => { if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; if (parameters?.[0] === 'one-dlq') throw new Error('database read failed'); @@ -301,7 +308,7 @@ describe('dead-letter monitoring', () => { it('bounds shutdown when a query never returns and never starts another sweep', async () => { vi.useFakeTimers(); const fake = fakeBoss(); - fake.getQueues.mockResolvedValue([{ name: 'stuck-dlq' }]); + fake.getQueues.mockResolvedValue([queueResult('stuck-dlq')]); fake.executeSql.mockImplementation((query: string) => query.includes('pgboss.version') ? Promise.resolve({ rows: [{ version: 26 }] }) diff --git a/src/lib/dead-letter-queues.ts b/src/lib/dead-letter-queues.ts index d88747c1..41a632e0 100644 --- a/src/lib/dead-letter-queues.ts +++ b/src/lib/dead-letter-queues.ts @@ -8,7 +8,7 @@ export const DLQ_RETENTION_SECONDS = 30 * 24 * 60 * 60; export const DLQ_SHUTDOWN_TIMEOUT_MS = 5_000; const DEAD_LETTER_SUFFIX = '-dlq'; -/** The single spelling of a source queue's dead-letter destination. */ +/** Default dead-letter destination for a source without custom routing. */ export function deadLetterQueueName(name: string): string { return `${name}${DEAD_LETTER_SUFFIX}`; } @@ -24,8 +24,9 @@ export async function ensureWorkerQueue( name: string, options: Omit = {} ): Promise { - const deadLetter = deadLetterQueueName(name); - const [existing, source] = await Promise.all([boss.getQueue(deadLetter), boss.getQueue(name)]); + const source = await boss.getQueue(name); + const deadLetter = source?.deadLetter ?? deadLetterQueueName(name); + const existing = await boss.getQueue(deadLetter); const retentionOptions = { // Do not shorten an operator's longer retention policy. Queue updates only // affect new jobs; existing keep_until/deletion_seconds remain untouched. diff --git a/src/lib/dead-letter-telemetry.test.ts b/src/lib/dead-letter-telemetry.test.ts index 61878fb9..3db7001f 100644 --- a/src/lib/dead-letter-telemetry.test.ts +++ b/src/lib/dead-letter-telemetry.test.ts @@ -1,8 +1,7 @@ -import type { PgBoss } from 'pg-boss'; - import { afterEach, describe, expect, it, vi } from 'vitest'; import { reportDeadLetterQueues } from '@/lib/dead-letter-queues'; +import { fakeBoss, queueResult } from '@/test/utils/test-helpers'; const client = vi.hoisted(() => ({ trackTrace: vi.fn() })); vi.mock('@/env', () => ({ @@ -17,25 +16,16 @@ afterEach(() => vi.restoreAllMocks()); describe('dLQ Application Insights telemetry', () => { it('sends queryable dimensions through the real production logger without job payloads', async () => { vi.spyOn(console, 'warn').mockImplementation(() => {}); - const boss = { - getQueues: async () => [{ name: 'usfm-export-dlq' }], - getDb: () => ({ - executeSql: async (query: string) => { - if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; - return { - rows: [ - { - depth: 1, - queuedCount: 1, - activeCount: 0, - deferredCount: 0, - oldestCreatedOn: null, - }, - ], - }; - }, - }), - } as unknown as PgBoss; + const { boss, getQueues, executeSql } = fakeBoss(); + getQueues.mockResolvedValue([queueResult('usfm-export-dlq')]); + executeSql.mockImplementation(async (query) => { + if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; + return { + rows: [ + { depth: 1, queuedCount: 1, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }, + ], + }; + }); await reportDeadLetterQueues(boss); diff --git a/src/test/utils/test-helpers.ts b/src/test/utils/test-helpers.ts index 82c69b7a..51488be3 100644 --- a/src/test/utils/test-helpers.ts +++ b/src/test/utils/test-helpers.ts @@ -1,5 +1,7 @@ import type { Context } from 'hono'; +import type { Db, Job, QueueResult } from 'pg-boss'; +import { PgBoss } from 'pg-boss'; import { vi } from 'vitest'; import { ErrorCode } from '@/lib/types'; @@ -286,3 +288,57 @@ export function createResult(data: T, success: boolean = true) { }, }; } + +/** A complete pg-boss queue fixture with overridable settings. */ +export function queueResult(name: string, overrides: Partial = {}): QueueResult { + return { + name, + policy: 'standard', + deferredCount: 0, + queuedCount: 0, + activeCount: 0, + totalCount: 0, + table: 'job', + createdOn: new Date('2026-01-01T00:00:00Z'), + updatedOn: new Date('2026-01-01T00:00:00Z'), + singletonsActive: null, + ...overrides, + }; +} + +/** A complete job fixture for invoking a registered pg-boss handler. */ +export function jobResult(data: T, overrides: Partial> = {}): Job { + return { id: 'job-1', name: 'test-queue', data, expireInSeconds: 900, ...overrides }; +} + +/** + * Uses an unstarted instance so spies keep pg-boss's real method contracts. + * The stub database prevents these unit-test queues from opening a connection. + */ +export function fakeBoss() { + const executeSql = vi.fn().mockImplementation(async (query) => { + if (query.includes('pgboss.version')) return { rows: [{ version: 26 }] }; + return { + rows: [{ depth: 0, queuedCount: 0, activeCount: 0, deferredCount: 0, oldestCreatedOn: null }], + }; + }); + const boss = new PgBoss({ db: { executeSql } }); + const getQueue = vi.spyOn(boss, 'getQueue').mockResolvedValue(null); + const createQueue = vi.spyOn(boss, 'createQueue').mockResolvedValue(undefined); + const updateQueue = vi.spyOn(boss, 'updateQueue').mockResolvedValue(undefined); + const deleteQueue = vi.spyOn(boss, 'deleteQueue').mockResolvedValue(undefined); + const getQueues = vi.spyOn(boss, 'getQueues').mockResolvedValue([]); + const getDb = vi.spyOn(boss, 'getDb').mockReturnValue({ executeSql }); + const work = vi.spyOn(boss, 'work').mockResolvedValue('test-worker'); + return { + boss, + getQueue, + createQueue, + updateQueue, + deleteQueue, + getQueues, + getDb, + executeSql, + work, + }; +} diff --git a/src/workers/dbl-sync.worker.test.ts b/src/workers/dbl-sync.worker.test.ts index ac357593..822c0716 100644 --- a/src/workers/dbl-sync.worker.test.ts +++ b/src/workers/dbl-sync.worker.test.ts @@ -4,6 +4,7 @@ import * as bibleSyncModule from '@/domains/bibles/sync/dbl-bible-sync'; import * as bookSyncModule from '@/domains/books/sync/dbl-book-sync'; import * as languageSyncModule from '@/domains/languages/sync/dbl-language-sync'; import { ok } from '@/lib/types'; +import { fakeBoss, jobResult } from '@/test/utils/test-helpers'; import { registerDblSyncWorker } from './dbl-sync.worker'; @@ -26,25 +27,20 @@ describe('dblSyncWorker', () => { }); it('registers the on-demand worker and handles execution lifecycle', async () => { - const mockBoss = { - getQueue: vi.fn().mockResolvedValue(null), - updateQueue: vi.fn(), - createQueue: vi.fn().mockResolvedValue(undefined), - work: vi.fn().mockResolvedValue(undefined), - } as any; + const { boss, work } = fakeBoss(); - await registerDblSyncWorker(mockBoss); + await registerDblSyncWorker(boss); - expect(mockBoss.work).toHaveBeenCalledWith('dbl-sync', { batchSize: 1 }, expect.any(Function)); + expect(work).toHaveBeenCalledWith('dbl-sync', { batchSize: 1 }, expect.any(Function)); - const handler = mockBoss.work.mock.calls[0][2]; + const handler = work.mock.calls[0][2]; vi.mocked(languageSyncModule.syncLanguagesFromDbl).mockResolvedValueOnce(ok({} as any)); vi.mocked(bibleSyncModule.syncBiblesFromDbl).mockResolvedValueOnce(ok({} as any)); vi.mocked(bookSyncModule.syncBooksFromDbl).mockResolvedValueOnce(ok({} as any)); vi.mocked(bookSyncModule.syncAudioAvailability).mockResolvedValueOnce(ok({} as any)); - await handler([{ id: 'job-1' }]); + await handler([jobResult({}, { id: 'job-1' })]); expect(languageSyncModule.syncLanguagesFromDbl).toHaveBeenCalledTimes(1); expect(bibleSyncModule.syncBiblesFromDbl).toHaveBeenCalledTimes(1); expect(bookSyncModule.syncBooksFromDbl).toHaveBeenCalledTimes(1); @@ -54,6 +50,6 @@ describe('dblSyncWorker', () => { ok: false, error: { message: 'Sync failed' } as any, }); - await expect(handler([{ id: 'job-2' }])).rejects.toThrow('Sync failed'); + await expect(handler([jobResult({}, { id: 'job-2' })])).rejects.toThrow('Sync failed'); }); }); diff --git a/src/workers/ingest-bible-text.worker.test.ts b/src/workers/ingest-bible-text.worker.test.ts index a1833fd6..bf720ea8 100644 --- a/src/workers/ingest-bible-text.worker.test.ts +++ b/src/workers/ingest-bible-text.worker.test.ts @@ -1,5 +1,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { fakeBoss, jobResult } from '@/test/utils/test-helpers'; + import { db } from '../db'; import { registerDblIngestTextWorker } from './ingest-bible-text.worker'; @@ -52,22 +54,13 @@ describe('dblIngestTextWorker', () => { }); it('registers handlers for both priority and background queues', async () => { - const mockBoss = { - getQueue: vi.fn().mockResolvedValue(null), - updateQueue: vi.fn(), - createQueue: vi.fn().mockResolvedValue(undefined), - work: vi.fn().mockResolvedValue(undefined), - } as any; - - await registerDblIngestTextWorker(mockBoss); - - expect(mockBoss.work).toHaveBeenCalledTimes(2); - expect(mockBoss.work).toHaveBeenCalledWith( - 'dbl-ingest-text', - { batchSize: 1 }, - expect.any(Function) - ); - expect(mockBoss.work).toHaveBeenCalledWith( + const { boss, work } = fakeBoss(); + + await registerDblIngestTextWorker(boss); + + expect(work).toHaveBeenCalledTimes(2); + expect(work).toHaveBeenCalledWith('dbl-ingest-text', { batchSize: 1 }, expect.any(Function)); + expect(work).toHaveBeenCalledWith( 'dbl-ingest-text-priority', { batchSize: 1 }, expect.any(Function) @@ -75,17 +68,12 @@ describe('dblIngestTextWorker', () => { }); it('handles partial download error recovery gracefully', async () => { - const mockBoss = { - getQueue: vi.fn().mockResolvedValue(null), - updateQueue: vi.fn(), - createQueue: vi.fn(), - work: vi.fn(), - } as any; - await registerDblIngestTextWorker(mockBoss); + const { boss, work } = fakeBoss(); + await registerDblIngestTextWorker(boss); // Extract the handler. pg-boss's WorkHandler always receives the batch as // an array, even at batchSize: 1 — see the array-wrapped call below. - const handler = mockBoss.work.mock.calls[0][2]; + const handler = work.mock.calls[0][2]; vi.mocked(db.query.bibles.findFirst).mockResolvedValue({ id: 1, @@ -109,7 +97,7 @@ describe('dblIngestTextWorker', () => { }); await expect( - handler([{ data: { bibleId: 1, bookCodes: ['GEN'] }, id: 'job-1' }]) + handler([jobResult({ bibleId: 1, bookCodes: ['GEN'] }, { id: 'job-1' })]) ).rejects.toThrow(/trigger retry/); // It should have continued to chapter 2 despite the error in chapter 1 @@ -118,14 +106,9 @@ describe('dblIngestTextWorker', () => { }); it('logs a warning and skips the book instead of silently ignoring it when no matching book exists', async () => { - const mockBoss = { - getQueue: vi.fn().mockResolvedValue(null), - updateQueue: vi.fn(), - createQueue: vi.fn(), - work: vi.fn(), - } as any; - await registerDblIngestTextWorker(mockBoss); - const handler = mockBoss.work.mock.calls[0][2]; + const { boss, work } = fakeBoss(); + await registerDblIngestTextWorker(boss); + const handler = work.mock.calls[0][2]; const { logger } = await import('../lib/logger'); const warnSpy = vi.spyOn(logger, 'warn').mockImplementation(() => undefined as any); @@ -136,7 +119,7 @@ describe('dblIngestTextWorker', () => { // No book in the DB matches this code. vi.mocked(db.query.books.findFirst).mockResolvedValue(undefined); - await handler([{ data: { bibleId: 1, bookCodes: ['XYZ'] }, id: 'job-2' }]); + await handler([jobResult({ bibleId: 1, bookCodes: ['XYZ'] }, { id: 'job-2' })]); expect(warnSpy).toHaveBeenCalledWith( expect.stringContaining('XYZ'), @@ -147,14 +130,9 @@ describe('dblIngestTextWorker', () => { }); it('marks a failed chapter-list fetch as a job failure so pg-boss retries, instead of silently dropping the whole book', async () => { - const mockBoss = { - getQueue: vi.fn().mockResolvedValue(null), - updateQueue: vi.fn(), - createQueue: vi.fn(), - work: vi.fn(), - } as any; - await registerDblIngestTextWorker(mockBoss); - const handler = mockBoss.work.mock.calls[0][2]; + const { boss, work } = fakeBoss(); + await registerDblIngestTextWorker(boss); + const handler = work.mock.calls[0][2]; vi.mocked(db.query.bibles.findFirst).mockResolvedValue({ id: 1, @@ -167,7 +145,7 @@ describe('dblIngestTextWorker', () => { }); await expect( - handler([{ data: { bibleId: 1, bookCodes: ['GEN'] }, id: 'job-3' }]) + handler([jobResult({ bibleId: 1, bookCodes: ['GEN'] }, { id: 'job-3' })]) ).rejects.toThrow(/trigger retry/); // A book whose chapter list couldn't be fetched has no chapters to fetch @@ -197,14 +175,9 @@ describe('dblIngestTextWorker', () => { }); it('logs success only when the assignment Result is ok', async () => { - const mockBoss = { - getQueue: vi.fn().mockResolvedValue(null), - updateQueue: vi.fn(), - createQueue: vi.fn(), - work: vi.fn(), - } as any; - await registerDblIngestTextWorker(mockBoss); - const handler = mockBoss.work.mock.calls[0][2]; + const { boss, work } = fakeBoss(); + await registerDblIngestTextWorker(boss); + const handler = work.mock.calls[0][2]; const { logger } = await import('../lib/logger'); const infoSpy = vi.spyOn(logger, 'info').mockImplementation(() => undefined as any); @@ -217,7 +190,9 @@ describe('dblIngestTextWorker', () => { } as any); setupProjectUnitsAndBooks([42], [7]); - await handler([{ data: { bibleId: 1, bookCodes: ['GEN'], projectId: 99 }, id: 'job-4' }]); + await handler([ + jobResult({ bibleId: 1, bookCodes: ['GEN'], projectId: 99 }, { id: 'job-4' }), + ]); expect(infoSpy).toHaveBeenCalledWith( 'Created chapter assignments for project unit after text ingestion', @@ -226,14 +201,9 @@ describe('dblIngestTextWorker', () => { }); it('does not log success and throws to trigger a retry when the assignment Result is an error', async () => { - const mockBoss = { - getQueue: vi.fn().mockResolvedValue(null), - updateQueue: vi.fn(), - createQueue: vi.fn(), - work: vi.fn(), - } as any; - await registerDblIngestTextWorker(mockBoss); - const handler = mockBoss.work.mock.calls[0][2]; + const { boss, work } = fakeBoss(); + await registerDblIngestTextWorker(boss); + const handler = work.mock.calls[0][2]; const { logger } = await import('../lib/logger'); const infoSpy = vi.spyOn(logger, 'info').mockImplementation(() => undefined as any); const errorSpy = vi.spyOn(logger, 'error').mockImplementation(() => undefined as any); @@ -248,7 +218,7 @@ describe('dblIngestTextWorker', () => { setupProjectUnitsAndBooks([42], [7]); await expect( - handler([{ data: { bibleId: 1, bookCodes: ['GEN'], projectId: 99 }, id: 'job-5' }]) + handler([jobResult({ bibleId: 1, bookCodes: ['GEN'], projectId: 99 }, { id: 'job-5' })]) ).rejects.toThrow(/Failed to create chapter assignments/); expect(infoSpy).not.toHaveBeenCalledWith( diff --git a/src/workers/usfm-export.worker.test.ts b/src/workers/usfm-export.worker.test.ts index 6db8713c..08c9963a 100644 --- a/src/workers/usfm-export.worker.test.ts +++ b/src/workers/usfm-export.worker.test.ts @@ -1,10 +1,10 @@ -import type { PgBoss } from 'pg-boss'; - import { Readable } from 'node:stream'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import { createUSFMZipStreamAsync, getProjectName } from '@/domains/usfm/usfm.service'; import { uploadExportStream } from '@/lib/blob-storage'; +import { err, ErrorCode } from '@/lib/types'; +import { fakeBoss, jobResult } from '@/test/utils/test-helpers'; import { registerUSFMExportWorker } from './usfm-export.worker'; @@ -25,20 +25,10 @@ vi.mock('@/domains/usfm/usfm.service', () => ({ getProjectName: vi.fn(), })); -type WorkHandler = (jobs: { id: string; data: unknown }[]) => Promise; - -function createFakeBoss() { - const work = vi.fn().mockResolvedValue(undefined); - const boss = { work } as unknown as PgBoss; - const getHandler = (): WorkHandler => work.mock.calls[0][2] as WorkHandler; - const getOptions = (): Record => work.mock.calls[0][1]; - return { boss, work, getHandler, getOptions }; -} - -const job = { - id: 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee', - data: { projectUnitId: 1, bookIds: [1], requestedBy: 42 }, -}; +const job = jobResult( + { projectUnitId: 1, bookIds: [1], requestedBy: 42 }, + { id: 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee', name: 'usfm-export' } +); describe('registerUSFMExportWorker', () => { beforeEach(() => { @@ -46,42 +36,42 @@ describe('registerUSFMExportWorker', () => { }); it('registers with batchSize 1 so per-job failures reach pg-boss', async () => { - const { boss, getOptions } = createFakeBoss(); + const { boss, work } = fakeBoss(); await registerUSFMExportWorker(boss); - expect(getOptions()).toMatchObject({ batchSize: 1 }); + expect(work.mock.calls[0][1]).toMatchObject({ batchSize: 1 }); }); it('rejects (so pg-boss retries) when export processing fails', async () => { - const { boss, getHandler } = createFakeBoss(); + const { boss, work } = fakeBoss(); const hooks = { onJobFailure: vi.fn(), onJobSuccess: vi.fn(), onBatchEnd: vi.fn() }; await registerUSFMExportWorker(boss, hooks); - vi.mocked(createUSFMZipStreamAsync).mockResolvedValue({ ok: false } as any); + vi.mocked(createUSFMZipStreamAsync).mockResolvedValue(err(ErrorCode.INTERNAL_ERROR)); - await expect(getHandler()([job])).rejects.toThrow('No books available for export'); + await expect(work.mock.calls[0][2]([job])).rejects.toThrow('No books available for export'); expect(hooks.onJobFailure).toHaveBeenCalledTimes(1); expect(hooks.onJobSuccess).not.toHaveBeenCalled(); expect(hooks.onBatchEnd).toHaveBeenCalledTimes(1); }); it('rejects when the blob upload fails, and still runs cleanup', async () => { - const { boss, getHandler } = createFakeBoss(); + const { boss, work } = fakeBoss(); await registerUSFMExportWorker(boss); const cleanup = vi.fn(); vi.mocked(createUSFMZipStreamAsync).mockResolvedValue({ ok: true, data: { stream: Readable.from(['zip-bytes']), cleanup }, - } as any); + }); vi.mocked(uploadExportStream).mockRejectedValue(new Error('blob unavailable')); - await expect(getHandler()([job])).rejects.toThrow('blob unavailable'); + await expect(work.mock.calls[0][2]([job])).rejects.toThrow('blob unavailable'); expect(cleanup).toHaveBeenCalledTimes(1); }); it('resolves with the download result on success', async () => { - const { boss, getHandler } = createFakeBoss(); + const { boss, work } = fakeBoss(); const hooks = { onJobFailure: vi.fn(), onJobSuccess: vi.fn() }; await registerUSFMExportWorker(boss, hooks); @@ -89,15 +79,15 @@ describe('registerUSFMExportWorker', () => { vi.mocked(createUSFMZipStreamAsync).mockResolvedValue({ ok: true, data: { stream: Readable.from(['zip-bytes']), cleanup }, - } as any); + }); vi.mocked(uploadExportStream).mockResolvedValue({ filename: `export-${job.id}.zip`, sizeBytes: 9, expiresAt: new Date('2026-01-01T01:00:00Z'), }); - vi.mocked(getProjectName).mockResolvedValue({ ok: true, data: 'My Project' } as any); + vi.mocked(getProjectName).mockResolvedValue({ ok: true, data: 'My Project' }); - const result = (await getHandler()([job])) as Record; + const result = await work.mock.calls[0][2]([job]); expect(result).toMatchObject({ success: true,