add test

test/queues/getConfig.ts

@@ -19,6 +19,7 @@ import { UpdatePostStep2Task } from './tasks/UpdatePostStep2Task.js'
 import { UpdatePostTask } from './tasks/UpdatePostTask.js'
 import { externalWorkflow } from './workflows/externalWorkflow.js'
 import { failsImmediatelyWorkflow } from './workflows/failsImmediately.js'
+import { fastParallelTaskWorkflow } from './workflows/fastParallelTaskWokflow.js'
 import { inlineTaskTestWorkflow } from './workflows/inlineTaskTest.js'
 import { inlineTaskTestDelayedWorkflow } from './workflows/inlineTaskTestDelayed.js'
 import { longRunningWorkflow } from './workflows/longRunning.js'
@@ -164,6 +165,7 @@ export const getConfig: () => Partial<Config> = () => ({
       subTaskFailsWorkflow,
       longRunningWorkflow,
       parallelTaskWorkflow,
+      fastParallelTaskWorkflow,
     ],
   },
   editor: lexicalEditor(),

test/queues/int.spec.ts

@@ -1424,6 +1424,8 @@ describe('Queues', () => {
       id: job.id,
     })
 
+    // error can be defined while hasError is true, as hasError: true is only set if the job cannot retry anymore.
+    expect(jobAfterRun.error).toBeNull()
     expect(jobAfterRun.hasError).toBe(false)
     expect(jobAfterRun.log?.length).toBe(amount)
 
@@ -1444,6 +1446,30 @@ describe('Queues', () => {
     }
   })
 
+  it('can reliably run workflows with parallel tasks that complete immediately', async () => {
+    const amount = 2
+    payload.config.jobs.deleteJobOnComplete = false
+
+    const job = await payload.jobs.queue({
+      workflow: 'fastParallelTask',
+      input: {
+        amount,
+      },
+    })
+
+    await payload.jobs.run({ silent: false })
+
+    const jobAfterRun = await payload.findByID({
+      collection: 'payload-jobs',
+      id: job.id,
+    })
+
+    // error can be defined while hasError is true, as hasError: true is only set if the job cannot retry anymore.
+    expect(jobAfterRun.error).toBeNull()
+    expect(jobAfterRun.hasError).toBe(false)
+    expect(jobAfterRun.log?.length).toBe(amount)
+  })
+
   it('can create and autorun jobs', async () => {
     await payload.jobs.queue({
       workflow: 'inlineTaskTest',

test/queues/payload-types.ts

@@ -132,6 +132,7 @@ export interface Config {
       subTaskFails: WorkflowSubTaskFails;
       longRunning: WorkflowLongRunning;
       parallelTask: WorkflowParallelTask;
+      fastParallelTask: WorkflowFastParallelTask;
     };
   };
 }
@@ -332,6 +333,7 @@ export interface PayloadJob {
         | 'subTaskFails'
         | 'longRunning'
         | 'parallelTask'
+        | 'fastParallelTask'
       )
     | null;
   taskSlug?:
@@ -821,6 +823,15 @@ export interface WorkflowParallelTask {
     amount: number;
   };
 }
+/**
+ * This interface was referenced by `Config`'s JSON-Schema
+ * via the `definition` "WorkflowFastParallelTask".
+ */
+export interface WorkflowFastParallelTask {
+  input: {
+    amount: number;
+  };
+}
 /**
  * This interface was referenced by `Config`'s JSON-Schema
  * via the `definition` "auth".

test/queues/workflows/fastParallelTaskWokflow.ts

@@ -0,0 +1,34 @@
+import type { WorkflowConfig } from 'payload'
+
+export const fastParallelTaskWorkflow: WorkflowConfig<'fastParallelTask'> = {
+  slug: 'fastParallelTask',
+  inputSchema: [
+    {
+      name: 'amount',
+      type: 'number',
+      required: true,
+    },
+  ],
+  handler: async ({ job, inlineTask }) => {
+    const taskFunctions = []
+    for (let i = 0; i < job.input.amount; i++) {
+      const idx = i + 1
+      taskFunctions.push(async () => {
+        return await inlineTask(`fast parallel task ${idx}`, {
+          input: {
+            test: idx,
+          },
+          task: () => {
+            return {
+              output: {
+                taskID: idx.toString(),
+              },
+            }
+          },
+        })
+      })
+    }
+
+    await Promise.all(taskFunctions.map((f) => f()))
+  },
+}