Parallel Task support in workflows

[6TELOIV]

It would be nice if workflows could kick off parallel tasks nicely:

async ({tasks, runParallelTasks}) => {
  await runParallelTasks([tasks.doSomething('doSomething'), tasks.fooBar('fooBar')]);
}

Current (incomplete) Solutions

Currently, if I do something like this:

async ({tasks}) => {
  await Promise.all([tasks.doSomething('doSomething'), tasks.fooBar('fooBar')]);
}

and doSomething fails before fooBar finishes, then fooBar is retried while the first run is still going, and the outcome of the first run of fooBar is ignored.

This is due to the nature of Promise.all and how it re-throws the first promise that rejects, and how Payload's workflow handles errors, depending on type.

• TaskError will handle the task's retries, and will handle the workflow's retries if the task still has more retries.
• other errors will only handle workflow's retries, and no tasks.

Alternatively, you could avoid the issue of the workflow finishing too quickly using Promise.allSettled, but then you have an issue when multiple sub-tasks fail, and there is no appropriate error to throw that payload will handle correctly to update the tasks and workflow.

Proposed Implementation

Implementing a parallel function for workflows could look something like the following:

1. Implement an AggregateTaskError type which can recieve multiple task failures
2. Change handleTaskError function to handleAggregateTaskError instead
   • The important change is that it will check all the tasks for retries before updating the workflow retries once. If you called handleTaskError many times, it would update the workflow many times.
   • You could also split this behavior into multiple, but it seems more straighforward to unify the behavior, and convert single TaskErrors in to an AggregateTaskError with only one error. As it's only called in one place (runJob) changing it's behavior doesn't seem too complicated.
3. Create parallelTasks function, which would look something like the following (types might be wrong, but functionality is there):

   type RunParralelTasksFn = <TTaskSlug extends keyof TypedJobs['tasks']>(
     tasks: Promise<TaskOutput<TTaskSlug>>[]
   ) => Promise<TaskOutput<TTaskSlug>[]>
   export const runParallelTasks: RunParralelTasksFn = async (tasks) => {
     const results = await Promise.allSettled(tasks);
     const rejected = results.filter(result => result.status === 'rejected');
     if (rejected.length > 0) {
       const taskErrors: TaskError[] = []
       for (const result of rejected) {
         if (result.reason instanceof TaskError) {
           taskErrors.push(result.reason);
         } else {
           // rethrow non-task errors (job cancellations, unknown errors)
           throw result.reason;
         }
       }
       throw new AggregateTaskError(taskErrors);
     }
     return results.map(result => result.
   }

4. Update runJob to pass runParrallelTasks function, and update error handling to use handleAggregateTaksError and convert single TaskErrors into aggregate, i.e. handleAggregateTaksError(new AggregateTaskError([error]))

[6TELOIV]

Not sure if I need to explain the merrits of parallel tasks to anyone, but for example, compare it to Payloads CI github action: multiple long-running jobs can run at the same time.

For Payload, this is useful when there are multiple independent network and/or CPU bound tasks that need to run as part of one cohesive job.

In my case, I am creating social media posts to multiple outlets, sending firebase cloud messages to notify a mobile app, and sending email. It should only send to each outlet once, and it should send them all at approximately the same time (i.e, one failing won't stop the others). Additionally, each should retry on failure (external APIs may experience temporary 5XX errors)