diff --git a/.github/workflows/pre-merge.yml b/.github/workflows/pre-merge.yml index a2869776..bafb632f 100644 --- a/.github/workflows/pre-merge.yml +++ b/.github/workflows/pre-merge.yml @@ -38,6 +38,49 @@ jobs: - name: Build test run: npm run build + dlq-integration: + name: Dead-letter queue integration + permissions: + contents: read + 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 + with: + persist-credentials: false + + - 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..3b24e553 --- /dev/null +++ b/docs/runbooks/worker-dead-letter-queues.md @@ -0,0 +1,222 @@ +# 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 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. +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 +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 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 +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 or AI queues with a different immutable policy are preserved, including + completed/failed history. Startup emits `worker_queue_policy_mismatch`. Resolve + 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 +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. + +### 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. + +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 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 + +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. 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: + +```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/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/index.ts b/src/index.ts index cbaafd0a..886a83e1 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,11 +67,16 @@ async function startServer() { logger.info(`${signal} received, shutting down server`); try { if (audioReclaimInterval) clearInterval(audioReclaimInterval); + // 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'); 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..cad2c92f --- /dev/null +++ b/src/lib/dead-letter-queues.integration.test.ts @@ -0,0 +1,394 @@ +import { Readable } from 'node:stream'; +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 { + deadLetterQueueName, + DLQ_RETENTION_SECONDS, + 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'; +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; + let migrationSql: postgres.Sql; + const errors: Error[] = []; + const exportQueue = QUEUE_NAMES.USFM_EXPORT; + const exportDlq = deadLetterQueueName(exportQueue); + + 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'); + } + migrationSql = postgres(connectionString!, { max: 1 }); + }); + + afterAll(async () => { + await migrationSql?.end(); + 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('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, + { 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 }); + + // 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(); + 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 }; + 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 () => { + 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 }; + 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..880ebe5c --- /dev/null +++ b/src/lib/dead-letter-queues.test.ts @@ -0,0 +1,330 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + DLQ_RETENTION_SECONDS, + DLQ_SHUTDOWN_TIMEOUT_MS, + ensureWorkerQueue, + reportDeadLetterQueues, + startDeadLetterMonitor, +} 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() }, +})); + +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' ? queueResult(name) : null + ); + await ensureWorkerQueue(fake.boss, 'ingestion'); + expect(fake.createQueue.mock.calls).toEqual([ + [ + 'ingestion-dlq', + { retentionSeconds: DLQ_RETENTION_SECONDS, deleteAfterSeconds: DLQ_RETENTION_SECONDS }, + ], + ]); + expect(fake.updateQueue).toHaveBeenCalledWith('ingestion', { deadLetter: 'ingestion-dlq' }); + 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 }); + 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(async (_name, options) => { + if (options) options.policy ??= 'standard'; + }); + 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', { + 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' + ? queueResult(name, { + retentionSeconds: DLQ_RETENTION_SECONDS, + deleteAfterSeconds: DLQ_RETENTION_SECONDS, + }) + : queueResult(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) => + name === 'usfm-export' ? queueResult(name, { policy: 'standard' }) : null + ); + 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(); + fake.getQueue.mockImplementation(async (name) => + name === 'ai-suggestions' ? queueResult(name, { policy: 'standard' }) : null + ); + await ensureAiSuggestionQueue(fake.boss); + expect(fake.updateQueue).toHaveBeenCalledWith('ai-suggestions', { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + 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' ? queueResult(name, { 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([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 }] }; + 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( + 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([ + 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 }] }; + return { + rows: [ + { depth: 3, queuedCount: 2, activeCount: 1, deferredCount: 2, oldestCreatedOn: null }, + ], + }; + }); + await reportDeadLetterQueues(fake.boss); + 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) + ); + 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([queueResult('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([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'); + 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' }), + 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('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(); + let finish!: (value: []) => void; + fake.getQueues.mockReturnValueOnce( + new Promise((resolve) => { + finish = resolve; + }) + ); + const stop = startDeadLetterMonitor(fake.boss); + await Promise.resolve(); + 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(); + }); + + it('bounds shutdown when a query never returns and never starts another sweep', async () => { + vi.useFakeTimers(); + const fake = fakeBoss(); + fake.getQueues.mockResolvedValue([queueResult('stuck-dlq')]); + 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(); + 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 new file mode 100644 index 00000000..41a632e0 --- /dev/null +++ b/src/lib/dead-letter-queues.ts @@ -0,0 +1,183 @@ +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 DEAD_LETTER_SUFFIX = '-dlq'; + +/** Default dead-letter destination for a source without custom routing. */ +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; + return Object.entries(options).every(([key, value]) => current[key] === value); +} + +/** Create a durable diagnostic destination before enabling dead-letter routing. */ +export async function ensureWorkerQueue( + boss: PgBoss, + name: string, + options: Omit = {} +): Promise { + 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. + retentionSeconds: Math.max(existing?.retentionSeconds ?? 0, DLQ_RETENTION_SECONDS), + deleteAfterSeconds: Math.max(existing?.deleteAfterSeconds ?? 0, DLQ_RETENTION_SECONDS), + }; + if (existing) { + if (!queueMatchesOptions(existing, retentionOptions)) { + await boss.updateQueue(deadLetter, retentionOptions); + } + } else { + await boss.createQueue(deadLetter, retentionOptions); + } + + 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; + 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 ${PG_BOSS_SCHEMA_VERSION}` + ); + 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. + const targets = new Set(); + for (const queue of queues) { + if (queue.deadLetter) targets.add(queue.deadLetter); + if (queue.name.endsWith(DEAD_LETTER_SUFFIX)) targets.add(queue.name); + } + + 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 database.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); + 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/dead-letter-telemetry.test.ts b/src/lib/dead-letter-telemetry.test.ts new file mode 100644 index 00000000..3db7001f --- /dev/null +++ b/src/lib/dead-letter-telemetry.test.ts @@ -0,0 +1,46 @@ +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', () => ({ + 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, 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); + + 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/exclusive-worker-queue-migration.ts b/src/lib/exclusive-worker-queue-migration.ts new file mode 100644 index 00000000..c9ff1b72 --- /dev/null +++ b/src/lib/exclusive-worker-queue-migration.ts @@ -0,0 +1,92 @@ +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]; + +/** 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'`; + 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} + `; + return { + queue, + result: { + queueName, + previousPolicy: queue.policy as string, + retainedJobs: stats.retained as number, + pendingJobs: stats.pending as number, + changed: false, + }, + }; + }; + + // 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` + ); + } + + 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/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 76dddbcf..a79ca1ac 100644 --- a/src/lib/queue.ts +++ b/src/lib/queue.ts @@ -1,13 +1,14 @@ import { PgBoss } from 'pg-boss'; +import { ensureWorkerQueue } from '@/lib/dead-letter-queues'; 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', @@ -81,50 +82,29 @@ 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', { - 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 }); - } - } - } - - 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/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 3f9114c0..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,23 +27,20 @@ describe('dblSyncWorker', () => { }); it('registers the on-demand worker and handles execution lifecycle', async () => { - const mockBoss = { - 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); @@ -52,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/dbl-sync.worker.ts b/src/workers/dbl-sync.worker.ts index 0b691b41..dfea1afe 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 { syncAudioAvailability, 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..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,20 +54,13 @@ describe('dblIngestTextWorker', () => { }); it('registers handlers for both priority and background queues', async () => { - const mockBoss = { - createQueue: vi.fn().mockResolvedValue(undefined), - work: vi.fn().mockResolvedValue(undefined), - } as any; + const { boss, work } = fakeBoss(); - await registerDblIngestTextWorker(mockBoss); + await registerDblIngestTextWorker(boss); - expect(mockBoss.work).toHaveBeenCalledTimes(2); - expect(mockBoss.work).toHaveBeenCalledWith( - 'dbl-ingest-text', - { batchSize: 1 }, - expect.any(Function) - ); - expect(mockBoss.work).toHaveBeenCalledWith( + 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) @@ -73,12 +68,12 @@ describe('dblIngestTextWorker', () => { }); it('handles partial download error recovery gracefully', async () => { - const mockBoss = { 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, @@ -102,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 @@ -111,9 +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 = { 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); @@ -124,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'), @@ -135,9 +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 = { 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, @@ -150,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 @@ -180,9 +175,9 @@ describe('dblIngestTextWorker', () => { }); it('logs success only when the assignment Result is ok', async () => { - const mockBoss = { 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); @@ -195,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', @@ -204,9 +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 = { 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); @@ -221,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/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); 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,