commit

Author: AlessioGr

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

@@ -330,23 +330,46 @@ export const runJobs = async (args: RunJobsArgs): Promise<RunJobsResult> => {
     }
     const jobReq = isolateObjectProperty(req, 'transactionID')
 
-    const workflowConfig: WorkflowConfig =
-      job.workflowSlug && jobsConfig.workflows?.length
-        ? jobsConfig.workflows.find(({ slug }) => slug === job.workflowSlug)!
-        : {
-            slug: 'singleTask',
-            handler: async ({ job, tasks }) => {
-              await tasks[job.taskSlug as string]!('1', {
-                input: job.input,
-              })
-            },
-          }
+    let workflowConfig: undefined | WorkflowConfig = undefined
+
+    if (job.workflowSlug && jobsConfig.workflows?.length) {
+      workflowConfig = jobsConfig.workflows.find(({ slug }) => slug === job.workflowSlug)
+    } else if (job.taskSlug && jobsConfig.tasks?.length) {
+      const taskExists = jobsConfig.tasks.some(({ slug }) => slug === job.taskSlug)
+      if (taskExists) {
+        workflowConfig = {
+          slug: 'singleTask',
+          handler: async ({ job, tasks }) => {
+            await tasks[job.taskSlug as string]!('1', {
+              input: job.input,
+            })
+          },
+        }
+      }
+    }
 
     if (!workflowConfig) {
+      const slugLabel = job.taskSlug ? `task '${job.taskSlug}'` : `workflow '${job.workflowSlug}'`
+      const errorMessage = `Job references ${slugLabel}, which is not registered in payload.config.jobs. The slug may have been removed from config after the job was queued.`
+
+      if (!silent || (typeof silent === 'object' && !silent.error)) {
+        payload.logger.error({
+          msg: `Error running job ${job.workflowSlug || `Task: ${job.taskSlug}`} id: ${job.id} - ${errorMessage}`,
+        })
+      }
+
+      const updateJob = getUpdateJobFunction(job, jobReq)
+      await updateJob({
+        error: { message: errorMessage },
+        hasError: true,
+        processing: false,
+        totalTried: (job.totalTried ?? 0) + 1,
+      })
+
       return {
         id: job.id,
         result: {
-          status: 'error',
+          status: 'error-reached-max-retries',
         },
       } // Skip jobs with no workflow configuration
     }

test/queues/int.spec.ts

@@ -938,6 +938,43 @@ describe('Queues - Payload', () => {
     expect(allSimples.docs[0]?.title).toBe('from single task')
   })
 
+  describe('when a queued task slug is no longer registered in config', () => {
+    let originalTasks: typeof payload.config.jobs.tasks
+
+    beforeEach(() => {
+      originalTasks = payload.config.jobs.tasks
+    })
+
+    afterEach(() => {
+      payload.config.jobs.tasks = originalTasks
+    })
+
+    it('should permanently fail the job after one attempt instead of retrying forever', async () => {
+      payload.config.jobs.deleteJobOnComplete = false
+
+      const job = await payload.jobs.queue({
+        task: 'CreateSimple',
+        input: {
+          message: 'queued before task removal',
+        },
+      })
+
+      // Simulate a deploy that removed the 'CreateSimple' task from config
+      payload.config.jobs.tasks = originalTasks!.filter((t) => t.slug !== 'CreateSimple')
+
+      await payload.jobs.run({ silent: true })
+
+      const jobAfterRun = await payload.findByID({
+        collection: 'payload-jobs',
+        id: job.id,
+      })
+
+      expect(jobAfterRun.hasError).toBe(true)
+      expect(jobAfterRun.processing).toBe(false)
+      expect(jobAfterRun.totalTried).toBe(1)
+    })
+  })
+
   it('can queue and run via the endpoint single tasks without workflows', async () => {
     const workflowsRef = payload.config.jobs.workflows
     delete payload.config.jobs.workflows