DLQ redrive

Author: timgit

README.md

@@ -47,7 +47,7 @@ This will likely cater the most to teams already familiar with the simplicity of
 * Job dependency workflow orchestration
 * Cron scheduling, job deferral
 * Queue storage policies to support a variety of rate limiting, debouncing, and concurrency use cases
-* Priority queues, dead letter queues, automatic retries with exponential backoff
+* Priority queues, dead letter queues with redrive, automatic retries with exponential backoff
 * Pub/sub API for fan-out queue relationships
 * SQL support for non-Node.js runtimes for most operations
 * Serverless function compatible

docs/api/jobs.md

@@ -319,10 +319,20 @@ Returns an array of jobs from a queue
       blocking: boolean,
       deadLetter: string,
       policy: string,
-      output: object
+      output: object,
+      sourceName: string | null,
+      sourceId: string | null,
+      sourceCreatedOn: Date | null,
+      sourceRetryCount: number | null
     }
     ```
 
+When a job is moved into a dead letter queue, the `source*` fields record where it
+came from: `sourceName` is the queue it originally failed on, `sourceId` is the id
+of the original job, `sourceCreatedOn` is the original job's creation time (so its
+true age survives the move), and `sourceRetryCount` is how many retries it consumed
+before being dead-lettered. These are `null` for jobs that were not dead-lettered.
+
 
 **Notes**
 
@@ -354,6 +364,37 @@ Deletes a job by id.
 
 Deletes a set of jobs by id.
 
+### `redrive(name, options)`
+
+Moves jobs out of a dead letter queue and re-creates them as fresh jobs on their original source queue. `name` is the
+dead letter queue to drain. Returns the number of jobs moved.
+
+Each job is routed back to the queue it originally failed on (its `sourceName`),
+so a single dead letter queue that collects from many source queues fans back out
+correctly. Re-created jobs get a new id, a reset retry count, cleared output, and
+the destination queue's current retry, retention, and policy configuration. Only
+jobs that are not currently being processed (still in the `created`/`retry` state)
+are moved.
+
+`options`:
+
+- `destination` — override queue to move all matched jobs into, instead of each
+  job's original source queue. Required to redrive jobs that have no recorded
+  source queue (e.g. jobs dead-lettered before this feature existed); such jobs are
+  left in place otherwise.
+- `sourceName` — only redrive jobs that originated from this source queue.
+- `limit` — maximum number of jobs to move in this call, oldest first (default
+  `1000`). Loop or schedule repeated calls to drain large dead letter queues at a
+  controlled rate.
+
+```js
+// drain a dead letter queue back to its source queues, 500 at a time
+let moved
+do {
+  moved = await boss.redrive('email-dlq', { limit: 500 })
+} while (moved > 0)
+```
+
 ### `deleteQueuedJobs(name)`
 
 Deletes all queued jobs in a queue.

docs/api/queues.md

@@ -41,7 +41,7 @@ Allowed policy values:
 
 * **deadLetter**, string
 
-  When a job fails after all retries, if the queue has a `deadLetter` property, the job's payload will be copied into that queue, copying the same retention and retry configuration as the original job.
+  When a job fails after all retries, if the queue has a `deadLetter` property, the job's payload will be copied into that queue, copying the same retention and retry configuration as the original job. The dead-lettered job also records where it came from via the `sourceName`, `sourceId`, `sourceCreatedOn`, and `sourceRetryCount` fields, which power [`redrive()`](jobs#redrivename-options) for moving jobs back to their source queue.
 
 * **warningQueueSize**, int
 

docs/index.md

@@ -59,7 +59,7 @@ This will likely cater the most to teams already familiar with the simplicity of
 * Job dependency workflow orchestration
 * Cron scheduling, job deferral
 * Queue storage policies to support a variety of rate limiting, debouncing, and concurrency use cases
-* Priority queues, dead letter queues, automatic retries with exponential backoff
+* Priority queues, dead letter queues with redrive, automatic retries with exponential backoff
 * Pub/sub API for fan-out queue relationships
 * SQL support for non-Node.js runtimes for most operations
 * Serverless function compatible

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "pg-boss",
-  "version": "12.22.0",
+  "version": "12.23.0",
   "description": "Queueing jobs in Postgres from Node.js like a boss",
   "type": "module",
   "main": "./dist/index.js",
@@ -61,7 +61,7 @@
     "docs:readme": "node ./scripts/sync-readme.js"
   },
   "pgboss": {
-    "schema": 33
+    "schema": 34
   },
   "repository": {
     "type": "git",

package.json

@@ -301,6 +301,10 @@ export class PgBoss extends EventEmitter<types.PgBossEventMap> {
     return this.#manager.deleteJob(name, id, options)
   }
 
+  redrive (name: string, options?: types.RedriveOptions): Promise<number> {
+    return this.#manager.redrive(name, options)
+  }
+
   deleteQueuedJobs (name: string): Promise<void> {
     return this.#manager.deleteQueuedJobs(name)
   }

src/index.ts

@@ -1405,7 +1405,7 @@ class Manager extends EventEmitter implements types.EventsMixin {
 
         // Insert to dead letter queue if failed and has dead_letter configured
         if (job.dead_letter) {
-          await tx.executeSql(dlqSql, [job.dead_letter, job.data, jobOutput])
+          await tx.executeSql(dlqSql, [job.dead_letter, job.data, jobOutput, job.name, job.id, job.created_on, job.retry_count])
         }
       }
 
@@ -1425,6 +1425,28 @@ class Manager extends EventEmitter implements types.EventsMixin {
     return this.mapCommandResponse(ids, result)
   }
 
+  async redrive (name: string, options: types.RedriveOptions = {}): Promise<number> {
+    Attorney.assertQueueName(name)
+
+    const { destination, sourceName, limit = 1000 } = options
+
+    if (destination !== undefined) {
+      Attorney.assertQueueName(destination)
+    }
+
+    if (sourceName !== undefined) {
+      Attorney.assertQueueName(sourceName)
+    }
+
+    assert(Number.isInteger(limit) && limit >= 1, 'limit must be an integer >= 1')
+
+    const db = this.assertDb(options)
+    const { table } = await this.getQueueCache(name)
+    const sql = plans.redriveJobs(this.config.schema, table)
+    const result = await db.executeSql(sql, [name, destination ?? null, sourceName ?? null, limit])
+    return result.rows[0].moved as number
+  }
+
   async cancel (name: string, id: string | string[], options: types.ConnectionOptions = {}) {
     Attorney.assertQueueName(name)
     const db = this.assertDb(options)

src/manager.ts

@@ -1005,6 +1005,26 @@ function getAll (schema: string): types.Migration[] {
         `SELECT ${schema}.job_table_run($cmd$DROP INDEX IF EXISTS ${schema}.job_i9$cmd$)`,
         `ALTER TABLE ${schema}.version DROP COLUMN flow_on`
       ]
+    },
+    {
+      release: '12.23.0',
+      version: 34,
+      previous: 33,
+      // Dead-letter source provenance. Plain columns on the partitioned parent cascade to
+      // job_common (DEFAULT partition) and every existing/future partition, so no job_table_run
+      // fan-out or createQueueFn bump is needed (queue-creation/index logic is unchanged).
+      install: [
+        `ALTER TABLE ${schema}.job ADD COLUMN IF NOT EXISTS source_name text`,
+        `ALTER TABLE ${schema}.job ADD COLUMN IF NOT EXISTS source_id uuid`,
+        `ALTER TABLE ${schema}.job ADD COLUMN IF NOT EXISTS source_created_on timestamp with time zone`,
+        `ALTER TABLE ${schema}.job ADD COLUMN IF NOT EXISTS source_retry_count int`
+      ],
+      uninstall: [
+        `ALTER TABLE ${schema}.job DROP COLUMN source_name`,
+        `ALTER TABLE ${schema}.job DROP COLUMN source_id`,
+        `ALTER TABLE ${schema}.job DROP COLUMN source_created_on`,
+        `ALTER TABLE ${schema}.job DROP COLUMN source_retry_count`
+      ]
     }
   ]
 }

src/migrationStore.ts

@@ -366,7 +366,11 @@ function createTableJob (schema: string, noPartitioning = false) {
       heartbeat_seconds int,
       blocked boolean not null default false,
       blocking boolean not null default false,
-      pending_dependencies int not null default 0
+      pending_dependencies int not null default 0,
+      source_name text,
+      source_id uuid,
+      source_created_on timestamp with time zone,
+      source_retry_count int
     ) ${partitionClause}
   `
 }
@@ -394,7 +398,11 @@ const JOB_COLUMNS_ALL = `${JOB_COLUMNS_MIN},
   blocked,
   blocking,
   pending_dependencies as "pendingDependencies",
-  output
+  output,
+  source_name as "sourceName",
+  source_id as "sourceId",
+  source_created_on as "sourceCreatedOn",
+  source_retry_count as "sourceRetryCount"
 `
 
 function createTableJobCommon (schema: string) {
@@ -1691,16 +1699,21 @@ function failJobsBody (schema: string, table: string, where: string, output: str
       SELECT * FROM failed_jobs
     ),
     dlq_jobs as (
-      INSERT INTO ${schema}.job (name, data, output, retry_limit, retry_backoff, retry_delay, keep_until, deletion_seconds)
+      INSERT INTO ${schema}.job (name, data, output, retry_limit, retry_backoff, retry_delay, keep_until, deletion_seconds,
+        source_name, source_id, source_created_on, source_retry_count)
       SELECT
         r.dead_letter,
-        data,
-        output,
+        r.data,
+        r.output,
         q.retry_limit,
         q.retry_backoff,
         q.retry_delay,
         now() + q.retention_seconds * interval '1s',
-        q.deletion_seconds
+        q.deletion_seconds,
+        r.name,
+        r.id,
+        r.created_on,
+        r.retry_count
       FROM results r
         JOIN ${schema}.queue q ON q.name = r.dead_letter
       WHERE state = '${JOB_STATES.failed}'
@@ -1911,12 +1924,52 @@ export function insertRetryJob (schema: string, table: string): string {
 
 export function insertDeadLetterJob (schema: string): string {
   return `
-    INSERT INTO ${schema}.job (name, data, output, retry_limit, retry_backoff, retry_delay, keep_until, deletion_seconds)
-    SELECT $1, $2, $3, q.retry_limit, q.retry_backoff, q.retry_delay, now() + q.retention_seconds * interval '1s', q.deletion_seconds
+    INSERT INTO ${schema}.job (name, data, output, retry_limit, retry_backoff, retry_delay, keep_until, deletion_seconds,
+      source_name, source_id, source_created_on, source_retry_count)
+    SELECT $1, $2, $3, q.retry_limit, q.retry_backoff, q.retry_delay, now() + q.retention_seconds * interval '1s', q.deletion_seconds,
+      $4, $5, $6, $7
     FROM ${schema}.queue q WHERE q.name = $1
   `
 }
 
+// Dead-letter redrive. Moves un-started jobs out of a dead-letter queue and
+// re-creates them as fresh jobs on their original source queue (or $2 destination override),
+// oldest-first, capped at $4. The JOIN in `candidates` only matches jobs whose destination queue
+// exists, so legacy/orphaned jobs (NULL source_name, no override) are never deleted — they stay
+// in the DLQ rather than being lost. Re-created jobs get a new id, `created` state, retry_count 0,
+// cleared output, NULL source_*, and the destination queue's current retry/retention/policy config.
+export function redriveJobs (schema: string, table: string): string {
+  return `
+    WITH candidates AS (
+      SELECT j.id
+      FROM ${schema}.${table} j
+      JOIN ${schema}.queue q ON q.name = COALESCE($2, j.source_name)
+      WHERE j.name = $1
+        AND j.state < '${JOB_STATES.active}'
+        AND ($3::text IS NULL OR j.source_name = $3)
+      ORDER BY j.created_on
+      LIMIT $4
+      FOR UPDATE OF j SKIP LOCKED
+    ),
+    moved AS (
+      DELETE FROM ${schema}.${table}
+      WHERE id IN (SELECT id FROM candidates)
+      RETURNING *
+    ),
+    ins AS (
+      INSERT INTO ${schema}.job
+        (name, data, priority, retry_limit, retry_backoff, retry_delay, retry_delay_max,
+         expire_seconds, keep_until, deletion_seconds, policy)
+      SELECT COALESCE($2, m.source_name), m.data, m.priority, q.retry_limit, q.retry_backoff,
+        q.retry_delay, q.retry_delay_max, q.expire_seconds,
+        now() + q.retention_seconds * interval '1s', q.deletion_seconds, q.policy
+      FROM moved m JOIN ${schema}.queue q ON q.name = COALESCE($2, m.source_name)
+      RETURNING 1
+    )
+    SELECT count(*)::int AS moved FROM ins
+  `
+}
+
 export function deletion (schema: string, table: string, queues: string[], noAdvisoryLocks?: boolean): string {
   const sql = `
     DELETE FROM ${schema}.${table}