feat: introduce output argument to .cancel() and .failWithoutRetry() to mark jobs as failed without implicitly retrying

.github/workflows/ci.yml

@@ -19,6 +19,7 @@ jobs:
         image: postgres
         env:
           POSTGRES_PASSWORD: postgres
+          POSTGRES_INITDB_ARGS: "-c max_connections=300"
         options: >-
           --health-cmd pg_isready
           --health-interval 10s

src/manager.js

@@ -59,6 +59,7 @@ class Manager extends EventEmitter {
     this.cancelJobsCommand = plans.cancelJobs(config.schema)
     this.resumeJobsCommand = plans.resumeJobs(config.schema)
     this.failJobsCommand = plans.failJobs(config.schema)
+    this.failJobsWithoutRetryCommand = plans.failJobsWithoutRetry(config.schema)
     this.getJobByIdCommand = plans.getJobById(config.schema)
     this.getArchivedJobByIdCommand = plans.getArchivedJobById(config.schema)
     this.subscribeCommand = plans.subscribe(config.schema)
@@ -71,6 +72,7 @@ class Manager extends EventEmitter {
       this.cancel,
       this.resume,
       this.fail,
+      this.failWithoutRetry,
       this.fetch,
       this.fetchCompleted,
       this.work,
@@ -552,10 +554,17 @@ class Manager extends EventEmitter {
     return this.mapCompletionResponse(ids, result)
   }
 
-  async cancel (id, options = {}) {
+  async failWithoutRetry (id, data, options = {}) {
+    const db = options.db || this.db
+    const ids = this.mapCompletionIdArg(id, 'fail')
+    const result = await db.executeSql(this.failJobsWithoutRetryCommand, [ids, this.mapCompletionDataArg(data)])
+    return this.mapCompletionResponse(ids, result)
+  }
+
+  async cancel (id, data, options = {}) {
     const db = options.db || this.db
     const ids = this.mapCompletionIdArg(id, 'cancel')
-    const result = await db.executeSql(this.cancelJobsCommand, [ids])
+    const result = await db.executeSql(this.cancelJobsCommand, [ids, this.mapCompletionDataArg(data)])
     return this.mapCompletionResponse(ids, result)
   }
 

src/plans.js

@@ -29,6 +29,7 @@ module.exports = {
   cancelJobs,
   resumeJobs,
   failJobs,
+  failJobsWithoutRetry,
   insertJob,
   insertJobs,
   getTime,
@@ -478,6 +479,31 @@ function failJobs (schema) {
   `
 }
 
+function failJobsWithoutRetry (schema) {
+  return `
+    WITH results AS (
+      UPDATE ${schema}.job
+      SET state = '${states.failed}'::${schema}.job_state,
+        completedOn = now(),
+        output = $2::jsonb
+      WHERE id IN (SELECT UNNEST($1::uuid[]))
+        AND state < '${states.completed}'
+      RETURNING *
+    ), completion_jobs as (
+      INSERT INTO ${schema}.job (name, data, keepUntil)
+      SELECT
+        '${COMPLETION_JOB_PREFIX}' || name,
+        ${buildJsonCompletionObject(true)},
+        ${keepUntilInheritance}
+      FROM results
+      WHERE state = '${states.failed}'
+        AND NOT name LIKE '${COMPLETION_JOB_PREFIX}%'
+        AND on_complete
+    )
+    SELECT COUNT(*) FROM results
+  `
+}
+
 function expire (schema) {
   return `
     WITH results AS (
@@ -509,7 +535,8 @@ function cancelJobs (schema) {
     with results as (
       UPDATE ${schema}.job
       SET completedOn = now(),
-        state = '${states.cancelled}'
+        state = '${states.cancelled}',
+        output = $2::jsonb
       WHERE id IN (SELECT UNNEST($1::uuid[]))
         AND state < '${states.completed}'
       RETURNING 1

test/cancelTest.js

@@ -72,11 +72,29 @@ describe('cancel', function () {
 
     const jobId = await boss.send('will_cancel', null, { startAfter: 1 })
 
-    await boss.cancel(jobId, { db })
+    await boss.cancel(jobId, null, { db })
 
     const job = await boss.getJobById(jobId)
 
     assert(job && job.state === 'cancelled')
     assert.strictEqual(called, true)
   })
+
+  it('should cancel a pending job, populating job output if provided', async function () {
+    const queue = 'cancel-data-batch'
+
+    const boss = this.test.boss = await helper.start(this.test.bossConfig)
+    await boss.send(queue)
+
+    const jobId = await boss.send('will_cancel', null, { startAfter: 1 })
+
+    const cancellationData = { msg: 'i am cancelled' }
+
+    await boss.cancel(jobId, cancellationData)
+
+    const job = await boss.getJobById(jobId)
+
+    assert(job && job.state === 'cancelled')
+    assert.strictEqual(job.output.msg, cancellationData.msg)
+  })
 })

test/failureTest.js

@@ -208,4 +208,22 @@ describe('failure', function () {
 
     assert(job)
   })
+
+  it('should accept a payload and not retry', async function () {
+    const boss = this.test.boss = await helper.start(this.test.bossConfig)
+    const queue = this.test.bossConfig.schema
+
+    const failPayload = { message: 'i am a failed job' }
+
+    const jobId = await boss.send(queue, null, { onComplete: true, retryLimit: 20 })
+
+    await boss.failWithoutRetry(jobId, failPayload)
+
+    const job = await boss.getJobById(jobId)
+
+    assert.strictEqual(job.state, 'failed')
+    assert.strictEqual(job.retrycount, 0)
+    assert.strictEqual(job.retrylimit, 20)
+    assert.strictEqual(job.output.message, failPayload.message)
+  })
 })

test/resumeTest.js

@@ -47,7 +47,7 @@ describe('cancel', function () {
       }
     }
 
-    await boss.cancel(jobId, { db })
+    await boss.cancel(jobId, null, { db })
 
     const job = await boss.getJobById(jobId, { db })
 

types.d.ts

@@ -352,6 +352,7 @@ declare class PgBoss extends EventEmitter {
   fetchCompleted<T>(name: string, batchSize: number, options: PgBoss.FetchOptions): Promise<PgBoss.Job<T>[] | null>;
 
   cancel(id: string, options?: PgBoss.ConnectionOptions): Promise<void>;
+  cancel(id: string, data: object, options?: PgBoss.ConnectionOptions): Promise<void>;
   cancel(ids: string[], options?: PgBoss.ConnectionOptions): Promise<void>;
 
   resume(id: string, options?: PgBoss.ConnectionOptions): Promise<void>;
@@ -365,6 +366,8 @@ declare class PgBoss extends EventEmitter {
   fail(id: string, data: object, options?: PgBoss.ConnectionOptions): Promise<void>;
   fail(ids: string[], options?: PgBoss.ConnectionOptions): Promise<void>;
 
+  failWithoutRetry(id: string, data: object, options?: PgBoss.ConnectionOptions): Promise<void>;
+
   getQueueSize(name: string, options?: object): Promise<number>;
   getJobById(id: string, options?: PgBoss.ConnectionOptions): Promise<PgBoss.JobWithMetadata | null>;