fix: ensure scheduling by default only handles default queue, add allQueues config to autoRun (#13395)

By default, `payload.jobs.run` only runs jobs from the `default` queue
(since #12799). It exposes an
`allQueues` property to run jobs from all queues.

For handling schedules (`payload.jobs.handleSchedules` and
`config.jobs.autoRun`), this behaves differently - jobs are run from all
queues by default, and no `allQueues` property exists.

This PR adds an `allQueues` property to scheduling, as well as changes
the default behavior to only handle schedules for the `default` queue.
That way, the behavior of running and scheduling jobs matches.

---
- To see the specific tasks where the Asana app for GitHub is being
used, see below:
  - https://app.asana.com/0/0/1210982048221260

Author: AlessioGr

packages/payload/src/fields/baseFields/baseIDField.ts

@@ -14,6 +14,7 @@ export const baseIDField: TextField = {
   defaultValue: () => new ObjectId().toHexString(),
   hooks: {
     beforeChange: [({ value }) => value || new ObjectId().toHexString()],
+    // ID field values for arrays and blocks need to be unique when duplicating, as on postgres they are stored on the same table as primary keys.
     beforeDuplicate: [() => new ObjectId().toHexString()],
   },
   label: 'ID',

packages/payload/src/fields/hooks/beforeDuplicate/promise.ts

@@ -63,7 +63,8 @@ export const promise = async <T>({
     let fieldData = siblingDoc?.[field.name!]
     const fieldIsLocalized = localization && fieldShouldBeLocalized({ field, parentIsLocalized })
 
-    // Run field beforeDuplicate hooks
+    // Run field beforeDuplicate hooks.
+    // These hooks are responsible for resetting the `id` field values of array and block rows. See `baseIDField`.
     if (Array.isArray(field.hooks?.beforeDuplicate)) {
       if (fieldIsLocalized) {
         const localeData: JsonObject = {}

packages/payload/src/index.ts

@@ -873,6 +873,7 @@ export class BasePayload {
               this.config.jobs.scheduling
             ) {
               await this.jobs.handleSchedules({
+                allQueues: cronConfig.allQueues,
                 queue: cronConfig.queue,
               })
             }
@@ -891,6 +892,7 @@ export class BasePayload {
             }
 
             await this.jobs.run({
+              allQueues: cronConfig.allQueues,
               limit: cronConfig.limit ?? DEFAULT_LIMIT,
               queue: cronConfig.queue,
               silent: cronConfig.silent,

packages/payload/src/queues/config/types/index.ts

@@ -7,6 +7,13 @@ import type { TaskConfig } from './taskTypes.js'
 import type { WorkflowConfig } from './workflowTypes.js'
 
 export type AutorunCronConfig = {
+  /**
+   * If you want to autoRUn jobs from all queues, set this to true.
+   * If you set this to true, the `queue` property will be ignored.
+   *
+   * @default false
+   */
+  allQueues?: boolean
   /**
    * The cron schedule for the job.
    * @default '* * * * *' (every minute).
@@ -43,6 +50,8 @@ export type AutorunCronConfig = {
   limit?: number
   /**
    * The queue name for the job.
+   *
+   * @default 'default'
    */
   queue?: string
   /**

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

@@ -45,11 +45,18 @@ export const handleSchedulesJobsEndpoint: Endpoint = {
       )
     }
 
-    const { queue } = req.query as {
+    const { allQueues, queue } = req.query as {
+      allQueues?: 'false' | 'true'
       queue?: string
     }
 
-    const { errored, queued, skipped } = await handleSchedules({ queue, req })
+    const runAllQueues = allQueues && !(typeof allQueues === 'string' && allQueues === 'false')
+
+    const { errored, queued, skipped } = await handleSchedules({
+      allQueues: runAllQueues,
+      queue,
+      req,
+    })
 
     return Response.json(
       {

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

@@ -56,7 +56,7 @@ export const runJobsEndpoint: Endpoint = {
 
     if (shouldHandleSchedules && jobsConfig.scheduling) {
       // If should handle schedules and schedules are defined
-      await req.payload.jobs.handleSchedules({ queue: runAllQueues ? undefined : queue, req })
+      await req.payload.jobs.handleSchedules({ allQueues: runAllQueues, queue, req })
     }
 
     const runJobsArgs: RunJobsArgs = {

packages/payload/src/queues/localAPI.ts

@@ -22,20 +22,28 @@ export type RunJobsSilent =
   | boolean
 export const getJobsLocalAPI = (payload: Payload) => ({
   handleSchedules: async (args?: {
+    /**
+     * If you want to schedule jobs from all queues, set this to true.
+     * If you set this to true, the `queue` property will be ignored.
+     *
+     * @default false
+     */
+    allQueues?: boolean
     // By default, schedule all queues - only scheduling jobs scheduled to be added to the `default` queue would not make sense
     // here, as you'd usually specify a different queue than `default` here, especially if this is used in combination with autorun.
     // The `queue` property for setting up schedules is required, and not optional.
     /**
      * If you want to only schedule jobs that are set to schedule in a specific queue, set this to the queue name.
      *
-     * @default all jobs for all queues will be scheduled.
+     * @default jobs from the `default` queue will be executed.
      */
     queue?: string
     req?: PayloadRequest
   }): Promise<HandleSchedulesResult> => {
     const newReq: PayloadRequest = args?.req ?? (await createLocalReq({}, payload))
 
     return await handleSchedules({
+      allQueues: args?.allQueues,
       queue: args?.queue,
       req: newReq,
     })

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

@@ -23,17 +23,26 @@ export type HandleSchedulesResult = {
  * after they are scheduled
  */
 export async function handleSchedules({
-  queue,
+  allQueues = false,
+  queue: _queue,
   req,
 }: {
+  /**
+   * If you want to schedule jobs from all queues, set this to true.
+   * If you set this to true, the `queue` property will be ignored.
+   *
+   * @default false
+   */
+  allQueues?: boolean
   /**
    * If you want to only schedule jobs that are set to schedule in a specific queue, set this to the queue name.
    *
-   * @default all jobs for all queues will be scheduled.
+   * @default jobs from the `default` queue will be executed.
    */
   queue?: string
   req: PayloadRequest
 }): Promise<HandleSchedulesResult> {
+  const queue = _queue ?? 'default'
   const jobsConfig = req.payload.config.jobs
   const queuesWithSchedules = getQueuesWithSchedules({
     jobsConfig,
@@ -53,7 +62,7 @@ export async function handleSchedules({
   // Need to know when that particular job was last scheduled in that particular queue
 
   for (const [queueName, { schedules }] of Object.entries(queuesWithSchedules)) {
-    if (queue && queueName !== queue) {
+    if (!allQueues && queueName !== queue) {
       // If a queue is specified, only schedule jobs for that queue
       continue
     }

test/queues/schedules.int.spec.ts

@@ -69,7 +69,28 @@ describe('Queues - scheduling, without automatic scheduling handling', () => {
 
   it('can auto-schedule through local API and autorun jobs', async () => {
     // Do not call payload.jobs.queue() - the `EverySecond` task should be scheduled here
-    await payload.jobs.handleSchedules()
+    await payload.jobs.handleSchedules({ queue: 'autorunSecond' })
+
+    // Do not call payload.jobs.run{silent: true})
+
+    await waitUntilAutorunIsDone({
+      payload,
+      queue: 'autorunSecond',
+      onlyScheduled: true,
+    })
+
+    const allSimples = await payload.find({
+      collection: 'simple',
+      limit: 100,
+    })
+
+    expect(allSimples.totalDocs).toBe(1)
+    expect(allSimples?.docs?.[0]?.title).toBe('This task runs every second')
+  })
+
+  it('can auto-schedule through local API and autorun jobs when passing allQueues', async () => {
+    // Do not call payload.jobs.queue() - the `EverySecond` task should be scheduled here
+    await payload.jobs.handleSchedules({ queue: 'autorunSecond', allQueues: true })
 
     // Do not call payload.jobs.run{silent: true})
 
@@ -88,9 +109,29 @@ describe('Queues - scheduling, without automatic scheduling handling', () => {
     expect(allSimples?.docs?.[0]?.title).toBe('This task runs every second')
   })
 
+  it('should not auto-schedule through local API and autorun jobs when not passing queue and schedule is not set on the default queue', async () => {
+    // Do not call payload.jobs.queue() - the `EverySecond` task should be scheduled here
+    await payload.jobs.handleSchedules()
+
+    // Do not call payload.jobs.run{silent: true})
+
+    await waitUntilAutorunIsDone({
+      payload,
+      queue: 'autorunSecond',
+      onlyScheduled: true,
+    })
+
+    const allSimples = await payload.find({
+      collection: 'simple',
+      limit: 100,
+    })
+
+    expect(allSimples.totalDocs).toBe(0)
+  })
+
   it('can auto-schedule through handleSchedules REST API and autorun jobs', async () => {
     // Do not call payload.jobs.queue() - the `EverySecond` task should be scheduled here
-    await restClient.GET('/payload-jobs/handle-schedules', {
+    await restClient.GET('/payload-jobs/handle-schedules?queue=autorunSecond', {
       headers: {
         Authorization: `JWT ${token}`,
       },
@@ -115,7 +156,7 @@ describe('Queues - scheduling, without automatic scheduling handling', () => {
 
   it('can auto-schedule through run REST API and autorun jobs', async () => {
     // Do not call payload.jobs.queue() - the `EverySecond` task should be scheduled here
-    await restClient.GET('/payload-jobs/run?silent=true', {
+    await restClient.GET('/payload-jobs/run?silent=true&allQueues=true', {
       headers: {
         Authorization: `JWT ${token}`,
       },
@@ -161,7 +202,7 @@ describe('Queues - scheduling, without automatic scheduling handling', () => {
   it('ensure scheduler does not schedule more jobs than needed if executed sequentially', async () => {
     await withoutAutoRun(async () => {
       for (let i = 0; i < 3; i++) {
-        await payload.jobs.handleSchedules()
+        await payload.jobs.handleSchedules({ allQueues: true })
       }
     })
 
@@ -192,7 +233,7 @@ describe('Queues - scheduling, without automatic scheduling handling', () => {
         })
       }
       for (let i = 0; i < 3; i++) {
-        await payload.jobs.handleSchedules()
+        await payload.jobs.handleSchedules({ allQueues: true })
       }
     })
 
@@ -271,8 +312,8 @@ describe('Queues - scheduling, without automatic scheduling handling', () => {
     for (let i = 0; i < 3; i++) {
       await withoutAutoRun(async () => {
         // Call it twice to test that it only schedules one
-        await payload.jobs.handleSchedules()
-        await payload.jobs.handleSchedules()
+        await payload.jobs.handleSchedules({ allQueues: true })
+        await payload.jobs.handleSchedules({ allQueues: true })
       })
       // Advance time to satisfy the waitUntil of newly scheduled jobs
       timeTravel(20)