fix: ensure deleteJobOnComplete property for jobs works (#9283)

Ensures that the `deleteJobOnComplete` (which is `true` by default)
property works properly

Author: r1tsuu

packages/payload/src/queues/operations/runJobs/index.ts

@@ -130,6 +130,9 @@ export const runJobs = async ({
   if (jobsQuery?.docs?.length) {
     req.payload.logger.info(`Running ${jobsQuery.docs.length} jobs.`)
   }
+  const jobsToDelete: (number | string)[] | undefined = req.payload.config.jobs.deleteJobOnComplete
+    ? []
+    : undefined
 
   const jobPromises = jobsQuery.docs.map(async (job) => {
     if (!job.workflowSlug && !job.taskSlug) {
@@ -191,6 +194,11 @@ export const runJobs = async ({
         workflowConfig,
         workflowHandler,
       })
+
+      if (result.status !== 'error' && jobsToDelete) {
+        jobsToDelete.push(job.id)
+      }
+
       return { id: job.id, result }
     } else {
       const result = await runJSONJob({
@@ -200,11 +208,32 @@ export const runJobs = async ({
         workflowConfig,
         workflowHandler,
       })
+
+      if (result.status !== 'error' && jobsToDelete) {
+        jobsToDelete.push(job.id)
+      }
+
       return { id: job.id, result }
     }
   })
 
   const resultsArray = await Promise.all(jobPromises)
+
+  if (jobsToDelete && jobsToDelete.length > 0) {
+    try {
+      await req.payload.delete({
+        collection: 'payload-jobs',
+        req,
+        where: { id: { in: jobsToDelete } },
+      })
+    } catch (err) {
+      req.payload.logger.error({
+        err,
+        msg: `failed to delete jobs ${jobsToDelete.join(', ')} on complete`,
+      })
+    }
+  }
+
   const resultsObject: RunJobsResult['jobStatus'] = resultsArray.reduce((acc, cur) => {
     if (cur !== null) {
       // Check if there's a valid result to include

test/queues/int.spec.ts

@@ -151,6 +151,7 @@ describe('Queues', () => {
   })
 
   it('ensure job retrying works', async () => {
+    payload.config.jobs.deleteJobOnComplete = false
     const job = await payload.jobs.queue({
       workflow: 'retriesTest',
       queue: 'default',
@@ -183,9 +184,11 @@ describe('Queues', () => {
 
     // @ts-expect-error amountRetried is new arbitrary data and not in the type
     expect(jobAfterRun.input.amountRetried).toBe(3)
+    payload.config.jobs.deleteJobOnComplete = true
   })
 
   it('ensure workflow-level retries are respected', async () => {
+    payload.config.jobs.deleteJobOnComplete = false
     const job = await payload.jobs.queue({
       workflow: 'retriesWorkflowLevelTest',
       input: {
@@ -217,6 +220,8 @@ describe('Queues', () => {
 
     // @ts-expect-error amountRetried is new arbitrary data and not in the type
     expect(jobAfterRun.input.amountRetried).toBe(2)
+
+    payload.config.jobs.deleteJobOnComplete = true
   })
 
   /*
@@ -256,6 +261,7 @@ describe('Queues', () => {
   })*/
 
   it('ensure backoff strategy of task is respected', async () => {
+    payload.config.jobs.deleteJobOnComplete = false
     const job = await payload.jobs.queue({
       workflow: 'retriesBackoffTest',
       input: {
@@ -338,6 +344,8 @@ describe('Queues', () => {
     expect(durations[1]).toBeGreaterThan(600)
     expect(durations[2]).toBeGreaterThan(1200)
     expect(durations[3]).toBeGreaterThan(2400)
+
+    payload.config.jobs.deleteJobOnComplete = true
   })
 
   it('can create new inline jobs', async () => {
@@ -359,6 +367,43 @@ describe('Queues', () => {
     expect(allSimples.docs[0].title).toBe('hello!')
   })
 
+  it('should respect deleteJobOnComplete true default configuration', async () => {
+    const { id } = await payload.jobs.queue({
+      workflow: 'inlineTaskTest',
+      input: {
+        message: 'hello!',
+      },
+    })
+
+    const before = await payload.findByID({ collection: 'payload-jobs', id, disableErrors: true })
+    expect(before.id).toBe(id)
+
+    await payload.jobs.run()
+
+    const after = await payload.findByID({ collection: 'payload-jobs', id, disableErrors: true })
+    expect(after).toBeNull()
+  })
+
+  it('should respect deleteJobOnComplete false configuration', async () => {
+    payload.config.jobs.deleteJobOnComplete = false
+    const { id } = await payload.jobs.queue({
+      workflow: 'inlineTaskTest',
+      input: {
+        message: 'hello!',
+      },
+    })
+
+    const before = await payload.findByID({ collection: 'payload-jobs', id, disableErrors: true })
+    expect(before.id).toBe(id)
+
+    await payload.jobs.run()
+
+    const after = await payload.findByID({ collection: 'payload-jobs', id, disableErrors: true })
+    expect(after.id).toBe(id)
+
+    payload.config.jobs.deleteJobOnComplete = true
+  })
+
   it('can queue single tasks', async () => {
     await payload.jobs.queue({
       task: 'CreateSimple',
@@ -513,6 +558,7 @@ describe('Queues', () => {
   })
 
   it('can queue single tasks 500 times', async () => {
+    payload.config.jobs.deleteJobOnComplete = false
     for (let i = 0; i < 500; i++) {
       await payload.jobs.queue({
         task: 'CreateSimple',
@@ -534,6 +580,7 @@ describe('Queues', () => {
     expect(allSimples.totalDocs).toBe(500) // Default limit: 10
     expect(allSimples.docs[0].title).toBe('from single task')
     expect(allSimples.docs[490].title).toBe('from single task')
+    payload.config.jobs.deleteJobOnComplete = true
   })
 
   it('ensure default jobs run limit of 10 works', async () => {