feat: add exclusive concurrency controls for workflows and tasks (#15177)

Adds exclusive concurrency controls to prevent race conditions when
multiple jobs operate on the same resource. Jobs with the same
concurrency key will not run in parallel.

## Key Changes

- Added `concurrency` option to workflows and tasks for defining
resource keys
- Added `enableConcurrencyControl` feature flag (defaults to `false`,
will be `true` in 4.0)
- Added indexed `concurrencyKey` field to jobs collection when
`enableConcurrencyControl` is set to `true`

## Usage Example

```typescript
export default buildConfig({
  jobs: {
    enableConcurrencyControl: true,
    workflows: [{
      slug: 'syncDocument',
      concurrency: ({ input }) => `sync:${input.documentId}`,
      handler: async ({ job }) => {
        // Only one job per documentId runs at a time
      }
    }]
  }
})
```

Author: AlessioGr

.gitignore

@@ -46,6 +46,7 @@ test/media
 *payloadtests.db-journal
 *payloadtests.db-shm
 *payloadtests.db-wal
+*payload.db
 /versions
 no-restrict-file-*
 

docs/jobs-queue/tasks.mdx

@@ -32,6 +32,7 @@ Simply add a task to the `jobs.tasks` array in your Payload config. A task consi
 | `onFail`        | Function to be executed if the task fails.                                                                                                                                                                                                                                                                                                                                                                                                       |
 | `onSuccess`     | Function to be executed if the task succeeds.                                                                                                                                                                                                                                                                                                                                                                                                    |
 | `retries`       | Specify the number of times that this step should be retried if it fails. If this is undefined, the task will either inherit the retries from the workflow or have no retries. If this is 0, the task will not be retried. By default, this is undefined.                                                                                                                                                                                        |
+| `concurrency`   | Control how jobs with the same concurrency key are handled. Jobs with the same key will run exclusively (one at a time). Requires `jobs.enableConcurrencyControl: true` to be set. See [Concurrency Controls](/docs/jobs-queue/workflows#concurrency-controls) for details.                                                                                                                                                                      |
 
 The logic for the Task is defined in the `handler` - which can be defined as a function, or a path to a function. The `handler` will run once a worker picks up a Job that includes this task.
 

docs/jobs-queue/workflows.mdx

@@ -56,6 +56,7 @@ To define a JS-based workflow, simply add a workflow to the `jobs.workflows` arr
 | `label`         | Define a human-friendly label for this workflow.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
 | `queue`         | Optionally, define the queue name that this workflow should be tied to. Defaults to "default".                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
 | `retries`       | You can define `retries` on the workflow level, which will enforce that the workflow can only fail up to that number of retries. If a task does not have retries specified, it will inherit the retry count as specified on the workflow. You can specify `0` as `workflow` retries, which will disregard all `task` retry specifications and fail the entire workflow on any task failure. You can leave `workflow` retries as undefined, in which case, the workflow will respect what each task dictates as their own retry count. By default this is undefined, meaning workflows retries are defined by their tasks |
+| `concurrency`   | Control how jobs with the same concurrency key are handled. Jobs with the same key will run exclusively (one at a time). Requires `jobs.enableConcurrencyControl: true` to be set. See [Concurrency Controls](#concurrency-controls) below for details.                                                                                                                                                                                                                                                                                                                                                                  |
 
 Example:
 
@@ -341,3 +342,148 @@ handler: async ({ input, req }) => {
   }
 }
 ```
+
+### Concurrency Controls
+
+When multiple jobs operate on the same resource, race conditions can occur. For example, if a user creates a document and then quickly updates it, two jobs might be queued that both try to process the same document simultaneously, leading to unexpected results.
+
+The `concurrency` option allows you to prevent this by ensuring that jobs with the same "key" run exclusively (one at a time).
+
+<Banner type="warning">
+  **Important:** To use concurrency controls, you must first enable them in your
+  Payload config by setting `jobs.enableConcurrencyControl: true`. This adds an
+  indexed `concurrencyKey` field to your jobs collection schema and may require
+  a database migration depending on your database adapter.
+</Banner>
+
+#### Enabling Concurrency Controls
+
+First, enable the feature in your Payload config:
+
+```ts
+export default buildConfig({
+  jobs: {
+    enableConcurrencyControl: true,
+    // ... your tasks and workflows
+  },
+})
+```
+
+Then add the `concurrency` option to your workflow configuration:
+
+```ts
+export default buildConfig({
+  jobs: {
+    workflows: [
+      {
+        slug: 'syncDocument',
+        inputSchema: [{ name: 'documentId', type: 'text', required: true }],
+        // Jobs with the same concurrency key run one at a time
+        concurrency: ({ input }) => `sync:${input.documentId}`,
+        handler: async ({ job, inlineTask }) => {
+          await inlineTask('fetch-and-update', {
+            task: async ({ req }) => {
+              // This runs exclusively - no other job for the same
+              // documentId can run at the same time
+              const doc = await req.payload.findByID({
+                collection: 'posts',
+                id: job.input.documentId,
+              })
+
+              await req.payload.update({
+                collection: 'posts',
+                id: job.input.documentId,
+                data: { syncedAt: new Date().toISOString() },
+              })
+
+              return { output: { synced: true } }
+            },
+          })
+        },
+      },
+    ],
+  },
+})
+```
+
+#### How It Works
+
+When you define a `concurrency` key:
+
+1. **When queuing:** The concurrency key is computed from the job's input and stored on the job document.
+
+2. **When running:** The job runner enforces exclusive execution through two mechanisms:
+
+   - It first checks which concurrency keys are currently being processed and excludes pending jobs with those keys from the query
+   - If multiple pending jobs with the same key are picked up in the same batch, only the first one (by creation order) runs - the others are released back to `processing: false` and will be picked up on subsequent runs
+
+3. **Result:** Jobs with the same concurrency key are guaranteed to run sequentially, never in parallel. All jobs are preserved and will eventually complete - they just wait their turn.
+
+#### Concurrency Configuration Options
+
+The `concurrency` option accepts either a function (shorthand) or an object with more options:
+
+**Shorthand (function only):**
+
+```ts
+// Exclusive concurrency is enabled by default
+concurrency: ({ input }) => `my-key:${input.resourceId}`
+```
+
+**Full configuration:**
+
+```ts
+concurrency: {
+  // Function that returns a key to group related jobs
+  // The queue name is provided to allow for queue-specific keys if needed
+  key: ({ input, queue }) => `my-key:${input.resourceId}`,
+
+  // Only one job with this key can run at a time
+  // @default true
+  exclusive: true,
+}
+```
+
+#### Use Cases
+
+**1. Document processing (prevent race conditions):**
+
+```ts
+concurrency: ({ input }) => `process:${input.documentId}`
+```
+
+Multiple updates to the same document will be processed one at a time, in the order they were queued.
+
+**2. Per-user operations:**
+
+```ts
+concurrency: ({ input }) => `user:${input.userId}`
+```
+
+Ensures only one job per user runs at a time, preventing conflicts when a user triggers multiple actions quickly.
+
+**3. External API calls (respect rate limits):**
+
+```ts
+concurrency: ({ input }) => `api:${input.resourceId}`
+```
+
+Prevents parallel API calls for the same resource, useful when external services don't handle concurrent updates well.
+
+**4. Queue-specific concurrency:**
+
+```ts
+concurrency: ({ input, queue }) => `${queue}:process:${input.documentId}`
+```
+
+Include the queue name in the key to allow the same resource to be processed concurrently in different queues (e.g., `emails` queue vs `default` queue).
+
+#### Important Considerations
+
+- **Key uniqueness:** The concurrency key should uniquely identify the resource being operated on. Include all relevant identifiers (collection slug, document ID, locale, etc.).
+
+- **Global by default:** By default, concurrency is global across all queues. A job with key `sync:doc1` in the `default` queue will block a job with the same key in the `emails` queue. Include the queue name in your key if you want queue-specific concurrency.
+
+- **No concurrency key = no restrictions:** Jobs without a concurrency configuration run in parallel as before.
+
+- **Pending jobs wait:** Jobs that can't run due to concurrency constraints remain in the queue with `processing: false` and will be picked up on subsequent runs.

packages/payload/src/index.ts

@@ -1712,6 +1712,7 @@ export type {
 } from './queues/config/types/taskTypes.js'
 export type {
   BaseJob,
+  ConcurrencyConfig,
   JobLog,
   JobTaskStatus,
   RunningJob,

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

@@ -18,6 +18,15 @@ export const getDefaultJobsCollection: (jobsConfig: SanitizedConfig['jobs']) =>
   if (jobsConfig.workflows?.length) {
     jobsConfig.workflows.forEach((workflow) => {
       workflowSlugs.add(workflow.slug)
+
+      // Validate concurrency config requires enableConcurrencyControl flag
+      if (workflow.concurrency && !jobsConfig.enableConcurrencyControl) {
+        throw new Error(
+          `Workflow "${workflow.slug}" uses concurrency controls but "jobs.enableConcurrencyControl" is not enabled. ` +
+            `Set "jobs.enableConcurrencyControl: true" in your Payload config to use concurrency controls. ` +
+            `Note: This adds a new indexed field to the jobs collection schema and may require a database migration.`,
+        )
+      }
     })
   }
 
@@ -28,6 +37,16 @@ export const getDefaultJobsCollection: (jobsConfig: SanitizedConfig['jobs']) =>
           `Task slug "${task.slug}" is already used by a workflow. No tasks are allowed to have the same slug as a workflow.`,
         )
       }
+
+      // Validate concurrency config requires enableConcurrencyControl flag
+      if (task.concurrency && !jobsConfig.enableConcurrencyControl) {
+        throw new Error(
+          `Task "${task.slug}" uses concurrency controls but "jobs.enableConcurrencyControl" is not enabled. ` +
+            `Set "jobs.enableConcurrencyControl: true" in your Payload config to use concurrency controls. ` +
+            `Note: This adds a new indexed field to the jobs collection schema and may require a database migration.`,
+        )
+      }
+
       taskSlugs.add(task.slug)
     })
   }
@@ -215,6 +234,22 @@ export const getDefaultJobsCollection: (jobsConfig: SanitizedConfig['jobs']) =>
         defaultValue: false,
         index: true,
       },
+      // Only add concurrencyKey field if concurrency control is enabled
+      ...(jobsConfig.enableConcurrencyControl
+        ? [
+            {
+              name: 'concurrencyKey',
+              type: 'text',
+              admin: {
+                description:
+                  'Used for concurrency control. Jobs with the same key are subject to exclusive/supersedes rules.',
+                position: 'sidebar',
+                readOnly: true,
+              },
+              index: true,
+            } as Field,
+          ]
+        : []),
     ],
     hooks: {
       afterRead: [

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

@@ -146,6 +146,18 @@ export type JobsConfig = {
    * @deprecated - this will be removed in 4.0
    */
   depth?: number
+  /**
+   * Enable concurrency controls for workflows and tasks.
+   * When enabled, adds a `concurrencyKey` field to the jobs collection schema.
+   * This allows workflows and tasks to use the `concurrency` option to prevent race conditions.
+   *
+   * **Important:** Enabling this may require a database migration depending on your database adapter,
+   * as it adds a new indexed field to the jobs collection schema.
+   *
+   * @default false
+   * @todo In 4.0, this will default to `true`.
+   */
+  enableConcurrencyControl?: boolean
   /**
    * Override any settings on the default Jobs collection. Accepts the default collection and allows you to return
    * a new collection.

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

@@ -7,7 +7,7 @@ import type {
   TypedJobs,
 } from '../../../index.js'
 import type { ScheduleConfig } from './index.js'
-import type { SingleTaskStatus } from './workflowTypes.js'
+import type { ConcurrencyConfig, SingleTaskStatus } from './workflowTypes.js'
 
 export type TaskInputOutput = {
   input: object
@@ -218,6 +218,19 @@ export type RetryConfig = {
 export type TaskConfig<
   TTaskSlugOrInputOutput extends keyof TypedJobs['tasks'] | TaskInputOutput = TaskType,
 > = {
+  /**
+   * Job concurrency controls for preventing race conditions.
+   *
+   * Can be an object with full options, or a shorthand function that just returns the key
+   * (in which case exclusive defaults to true).
+   */
+  concurrency?: ConcurrencyConfig<
+    TTaskSlugOrInputOutput extends keyof TypedJobs['tasks']
+      ? TypedJobs['tasks'][TTaskSlugOrInputOutput]['input']
+      : TTaskSlugOrInputOutput extends TaskInputOutput
+        ? TTaskSlugOrInputOutput['input']
+        : object
+  >
   /**
    * The function that should be responsible for running the job.
    * You can either pass a string-based path to the job function file, or the job function itself.

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

@@ -45,6 +45,10 @@ export type BaseJob<
   TWorkflowSlugOrInput extends false | keyof TypedJobs['workflows'] | object = false,
 > = {
   completedAt?: null | string
+  /**
+   * Used for concurrency control. Jobs with the same key are subject to exclusive/supersedes rules.
+   */
+  concurrencyKey?: null | string
   createdAt: string
   error?: unknown
   hasError?: boolean
@@ -126,9 +130,44 @@ export type JobTaskStatus = {
   }
 }
 
+/**
+ * Concurrency configuration for workflows and tasks.
+ * Controls how jobs with the same concurrency key are handled.
+ */
+export type ConcurrencyConfig<TInput = object> =
+  | ((args: { input: TInput; queue: string }) => string)
+  // Shorthand: key function only, exclusive defaults to true
+  | {
+      /**
+       * Only one job with this key can run at a time.
+       * Other jobs with the same key remain queued until the running job completes.
+       * @default true
+       */
+      exclusive?: boolean
+      /**
+       * Function that returns a key to group related jobs.
+       * Jobs with the same key are subject to concurrency rules.
+       * The queue name is provided to allow for queue-specific concurrency keys if needed.
+       */
+      key: (args: { input: TInput; queue: string }) => string
+    }
+
 export type WorkflowConfig<
   TWorkflowSlugOrInput extends false | keyof TypedJobs['workflows'] | object = false,
 > = {
+  /**
+   * Job concurrency controls for preventing race conditions.
+   *
+   * Can be an object with full options, or a shorthand function that just returns the key
+   * (in which case exclusive defaults to true).
+   */
+  concurrency?: ConcurrencyConfig<
+    TWorkflowSlugOrInput extends false
+      ? object
+      : TWorkflowSlugOrInput extends keyof TypedJobs['workflows']
+        ? TypedJobs['workflows'][TWorkflowSlugOrInput]['input']
+        : TWorkflowSlugOrInput
+  >
   /**
    * You can either pass a string-based path to the workflow function file, or the workflow function itself.
    *