refactor: clean up dead code and minor bugs in job queue system

packages/payload/src/queues/config/collection.ts

@@ -296,6 +296,5 @@ export function jobAfterRead({ config, doc }: { config: SanitizedConfig; doc: Jo
     jobLog: doc.log || [],
   })
   doc.input = doc.input || {}
-  doc.taskStatus = doc.taskStatus || {}
   return doc
 }

packages/payload/src/queues/endpoints/run.ts

@@ -68,10 +68,6 @@ export const runJobsEndpoint: Endpoint = {
       silent,
     }
 
-    if (typeof queue === 'string') {
-      runJobsArgs.queue = queue
-    }
-
     const parsedLimit = Number(limit)
     if (!isNaN(parsedLimit)) {
       runJobsArgs.limit = parsedLimit

packages/payload/src/queues/errors/calculateBackoffWaitUntil.ts

@@ -9,30 +9,22 @@ export function calculateBackoffWaitUntil({
   retriesConfig: number | RetryConfig
   totalTried: number
 }): Date {
-  let waitUntil: Date = getCurrentDate()
+  const now = getCurrentDate()
+  let waitUntil: Date = now
+
   if (typeof retriesConfig === 'object') {
     if (retriesConfig.backoff) {
       if (retriesConfig.backoff.type === 'fixed') {
         waitUntil = retriesConfig.backoff.delay
-          ? new Date(getCurrentDate().getTime() + retriesConfig.backoff.delay)
-          : getCurrentDate()
+          ? new Date(now.getTime() + retriesConfig.backoff.delay)
+          : now
       } else if (retriesConfig.backoff.type === 'exponential') {
         // 2 ^ (attempts - 1) * delay (current attempt is not included in totalTried, thus no need for -1)
         const delay = retriesConfig.backoff.delay ? retriesConfig.backoff.delay : 0
-        waitUntil = new Date(getCurrentDate().getTime() + Math.pow(2, totalTried) * delay)
+        waitUntil = new Date(now.getTime() + Math.pow(2, totalTried) * delay)
       }
     }
   }
 
-  /*
-  const differenceInMSBetweenNowAndWaitUntil = waitUntil.getTime() - getCurrentDate().getTime()
-
-  const differenceInSBetweenNowAndWaitUntil = differenceInMSBetweenNowAndWaitUntil / 1000
-  console.log('Calculated backoff', {
-    differenceInMSBetweenNowAndWaitUntil,
-    differenceInSBetweenNowAndWaitUntil,
-    retriesConfig,
-    totalTried,
-  })*/
   return waitUntil
 }

packages/payload/src/queues/operations/handleSchedules/countRunnableOrActiveJobsForQueue.ts

@@ -2,6 +2,8 @@ import type { PayloadRequest, Where } from '../../../types/index.js'
 import type { TaskType } from '../../config/types/taskTypes.js'
 import type { WorkflowTypes } from '../../config/types/workflowTypes.js'
 
+import { jobsCollectionSlug } from '../../config/collection.js'
+
 /**
  * Gets all queued jobs that can be run. This means they either:
  * - failed but do not have a definitive error => can be retried
@@ -63,7 +65,7 @@ export async function countRunnableOrActiveJobsForQueue({
   }
 
   const runnableOrActiveJobsForQueue = await req.payload.db.count({
-    collection: 'payload-jobs',
+    collection: jobsCollectionSlug,
     req,
     where: {
       and,

packages/payload/src/queues/operations/handleSchedules/defaultAfterSchedule.ts

@@ -40,7 +40,7 @@ export const defaultAfterSchedule: AfterScheduleFn = async ({ jobStats, queueabl
             },
           },
         },
-        updatedAt: new Date().toISOString(),
+        updatedAt: getCurrentDate().toISOString(),
       } as JobStats,
       req,
       returning: false,

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

@@ -303,27 +303,22 @@ export const runJobs = async (args: RunJobsArgs): Promise<RunJobsResult> => {
     }
   }
 
-  /**
-   * Just for logging purposes, we want to know how many jobs are new and how many are existing (= already been tried).
-   * This is only for logs - in the end we still want to run all jobs, regardless of whether they are new or existing.
-   */
-  const { existingJobs, newJobs } = jobs.reduce(
-    (acc, job) => {
+  if (!silent || (typeof silent === 'object' && !silent.info)) {
+    let newCount = 0
+    let retryCount = 0
+
+    for (const job of jobs) {
       if (job.totalTried > 0) {
-        acc.existingJobs.push(job)
+        retryCount++
       } else {
-        acc.newJobs.push(job)
+        newCount++
       }
-      return acc
-    },
-    { existingJobs: [] as Job[], newJobs: [] as Job[] },
-  )
+    }
 
-  if (!silent || (typeof silent === 'object' && !silent.info)) {
     payload.logger.info({
       msg: `Running ${jobs.length} jobs.`,
-      new: newJobs?.length,
-      retrying: existingJobs?.length,
+      new: newCount,
+      retrying: retryCount,
     })
   }
 

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

@@ -137,6 +137,7 @@ export const runJSONJob = async ({
     return await runJSONJob({
       job,
       req,
+      silent,
       updateJob,
       workflowConfig,
       workflowHandler,

packages/payload/src/queues/utilities/updateJob.ts

@@ -4,6 +4,7 @@ import type { Job } from '../../index.js'
 import type { PayloadRequest, Sort, Where } from '../../types/index.js'
 
 import { jobAfterRead, jobsCollectionSlug } from '../config/collection.js'
+import { getCurrentDate } from './getCurrentDate.js'
 
 type BaseArgs = {
   data: Partial<Job>
@@ -85,7 +86,7 @@ export async function updateJobs({
 
   if (typeof data.updatedAt === 'undefined') {
     // Ensure updatedAt date is always updated
-    data.updatedAt = new Date().toISOString()
+    data.updatedAt = getCurrentDate().toISOString()
   }
 
   const args: UpdateJobsArgs = id