Merge pull request #827 from timgit/dlq-redrive

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

@@ -273,6 +273,33 @@ export default function JobDetail ({ loaderData }: Route.ComponentProps) {
                 </div>
               )}
 
+              {(job.sourceName || job.sourceId) && (
+                <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-x-6 gap-y-4">
+                  {job.sourceName && (
+                    <div>
+                      <dt className="pgb-eyebrow">Source Queue</dt>
+                      <dd className="mt-1 text-sm text-[var(--text-primary)]">
+                        <DbLink
+                          to={`/queues/${job.sourceName}`}
+                          className="font-mono text-xs text-primary-600 hover:text-primary-700 dark:text-primary-400 dark:hover:text-primary-300"
+                        >
+                          {job.sourceName}
+                        </DbLink>
+                      </dd>
+                    </div>
+                  )}
+                  {job.sourceId && (
+                    <ConfigItem label="Source Job ID" value={job.sourceId} mono />
+                  )}
+                  {job.sourceRetryCount !== null && job.sourceRetryCount !== undefined && (
+                    <ConfigItem label="Source Retry Count" value={job.sourceRetryCount} mono />
+                  )}
+                  {job.sourceCreatedOn && (
+                    <ConfigItem label="Source Created" value={formatDate(new Date(job.sourceCreatedOn))} mono />
+                  )}
+                </div>
+              )}
+
               <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-5 gap-x-6 gap-y-4">
                 <ConfigItem
                   label="Created"

packages/dashboard/app/routes/queues.$name.jobs.$jobId.tsx

@@ -757,6 +757,39 @@ describe('Job Queries', () => {
       expect(job!.priority).toBe(5)
       expect(job!.data).toEqual({ foo: 'bar' })
     })
+
+    it('returns null source-tracking fields for a job that was not dead-lettered', async () => {
+      await createTestQueue('test-queue')
+      const jobId = await sendTestJob('test-queue', { foo: 'bar' })
+
+      const job = await getJobById(ctx.connectionString, ctx.schema, 'test-queue', jobId!)
+
+      expect(job!.sourceName).toBeNull()
+      expect(job!.sourceId).toBeNull()
+      expect(job!.sourceCreatedOn).toBeNull()
+      expect(job!.sourceRetryCount).toBeNull()
+    })
+
+    it('exposes source-tracking fields for a dead-lettered job', async () => {
+      await createTestQueue('test-dlq')
+      await createTestQueue('test-queue', { deadLetter: 'test-dlq', retryLimit: 0 })
+
+      const jobId = await sendTestJob('test-queue', { foo: 'bar' })
+
+      // fail() routes the job to the dead letter queue synchronously
+      await fetchTestJob('test-queue')
+      await failTestJob('test-queue', jobId!)
+
+      // The dead-lettered copy is a new job in the DLQ with its own id
+      const moved = await fetchTestJob('test-dlq')
+      const dlqJob = await getJobById(ctx.connectionString, ctx.schema, 'test-dlq', moved!.id)
+
+      expect(dlqJob).not.toBeNull()
+      expect(dlqJob!.sourceName).toBe('test-queue')
+      expect(dlqJob!.sourceId).toBe(jobId)
+      expect(dlqJob!.sourceCreatedOn).toBeTruthy()
+      expect(dlqJob!.sourceRetryCount).toBe(0)
+    })
   })
 })
 

packages/dashboard/tests/server/queries.test.ts

@@ -93,6 +93,12 @@ export const findJobsOptionsSchema = z.object({
   queued: z.boolean().optional(),
 }) satisfies z.ZodType<types.HttpFindJobsOptions>
 
+export const redriveOptionsSchema = z.object({
+  destination: queueNameSchema.optional(),
+  sourceName: queueNameSchema.optional(),
+  limit: z.number().optional(),
+}) satisfies z.ZodType<types.HttpRedriveOptions>
+
 export const insertOptionsSchema = z.object({
   returnId: z.boolean().optional(),
 }) satisfies z.ZodType<types.HttpInsertOptions>
@@ -162,6 +168,10 @@ export const jobWithMetadataSchema = jobSchemaBase.extend({
   policy: z.string(),
   deadLetter: z.string(),
   output: jsonRecordSchema,
+  sourceName: z.string().nullable(),
+  sourceId: z.string().nullable(),
+  sourceCreatedOn: z.iso.datetime().nullable().transform((val) => val ? new Date(val) : null),
+  sourceRetryCount: z.number().nullable(),
 }) satisfies z.ZodType<types.HttpJobWithMetadata>
 
 export const commandResponseSchema = z.object({
@@ -380,6 +390,16 @@ export const deleteJobResponseSchema: z.ZodType<types.HttpDeleteJobResponse> = z
   result: commandResponseSchema
 })
 
+export const redriveRequestSchema: z.ZodType<types.HttpRedriveRequest> = z.object({
+  name: queueNameSchema,
+  options: redriveOptionsSchema.optional()
+})
+
+export const redriveResponseSchema: z.ZodType<types.HttpRedriveResponse> = z.object({
+  ok: z.literal(true),
+  result: z.number()
+})
+
 export const deleteQueuedJobsRequestSchema: z.ZodType<types.HttpDeleteQueuedJobsRequest> = z.object({
   name: queueNameSchema
 })

packages/proxy/src/contracts.ts

@@ -35,6 +35,8 @@ import {
   isInstalledResponseSchema,
   publishRequestSchema,
   publishResponseSchema,
+  redriveRequestSchema,
+  redriveResponseSchema,
   resumeRequestSchema,
   resumeResponseSchema,
   retryRequestSchema,
@@ -177,6 +179,7 @@ export const postMethods: RouteEntry[] = [
   post('jobs', 'resume', resumeRequestSchema, resumeResponseSchema, (body) => [body.name, body.id]),
   post('jobs', 'retry', retryRequestSchema, retryResponseSchema, (body) => [body.name, body.id]),
   post('jobs', 'deleteJob', deleteJobRequestSchema, deleteJobResponseSchema, (body) => [body.name, body.id]),
+  post('jobs', 'redrive', redriveRequestSchema, redriveResponseSchema, (body) => withOptionalOptions([body.name], body.options)),
   post('jobs', 'deleteQueuedJobs', deleteQueuedJobsRequestSchema, deleteQueuedJobsResponseSchema, (body) => [body.name]),
   post('jobs', 'deleteStoredJobs', deleteStoredJobsRequestSchema, deleteStoredJobsResponseSchema, (body) => [body.name]),
   post('jobs', 'deleteAllJobs', deleteAllJobsRequestSchema, deleteAllJobsResponseSchema, (body) => (body.name ? [body.name] : [])),