feat: supersedes option to job queue concurrency controls (#15179)

Merge first: #15177

Adds `supersedes` option to concurrency controls allowing newer jobs to
automatically delete older pending jobs with the same concurrency key.
This enables "last queued wins" behavior for scenarios where only the
latest state matters.

## Usage

```typescript
concurrency: {
  key: ({ input }) => `generate:${input.documentId}`,
  exclusive: true,
  supersedes: true,  // Newer jobs delete older pending ones (not yet completed and did not start processing yet)
}
```

**Example scenario:** User rapidly edits a document triggering 5
regeneration jobs. Only the last job runs - intermediate pending jobs
are automatically deleted.

## Important

Only **pending** jobs are deleted. Running jobs complete normally and
the new job waits.

Author: AlessioGr

docs/jobs-queue/workflows.mdx

@@ -426,7 +426,7 @@ The `concurrency` option accepts either a function (shorthand) or an object with
 **Shorthand (function only):**
 
 ```ts
-// Exclusive concurrency is enabled by default
+// Exclusive defaults to true, supersedes defaults to false
 concurrency: ({ input }) => `my-key:${input.resourceId}`
 ```
 
@@ -441,42 +441,83 @@ concurrency: {
   // Only one job with this key can run at a time
   // @default true
   exclusive: true,
+
+  // Delete older pending jobs when a new job is queued
+  // @default false
+  supersedes: false,
 }
 ```
 
-#### Use Cases
+#### Common Patterns
 
-**1. Document processing (prevent race conditions):**
+**1. Exclusive only (preserve all jobs):**
 
 ```ts
-concurrency: ({ input }) => `process:${input.documentId}`
+concurrency: {
+  key: ({ input }) => `process:${input.documentId}`,
+  exclusive: true,
+  supersedes: false, // All jobs run, just not in parallel
+}
 ```
 
-Multiple updates to the same document will be processed one at a time, in the order they were queued.
+Use when every job represents unique work that must complete (e.g., processing distinct versions of a document).
 
-**2. Per-user operations:**
+**2. Exclusive + Supersedes (last queued wins):**
 
 ```ts
-concurrency: ({ input }) => `user:${input.userId}`
+concurrency: {
+  key: ({ input }) => `generate:${input.documentId}`,
+  exclusive: true,
+  supersedes: true, // Only latest job runs
+}
 ```
 
-Ensures only one job per user runs at a time, preventing conflicts when a user triggers multiple actions quickly.
+Use when only the latest state matters (e.g., regenerating embeddings after rapid edits - intermediate states can be skipped).
 
-**3. External API calls (respect rate limits):**
+**3. Queue-specific concurrency:**
 
 ```ts
-concurrency: ({ input }) => `api:${input.resourceId}`
+concurrency: {
+  key: ({ input, queue }) => `${queue}:sync:${input.resourceId}`,
+}
 ```
 
-Prevents parallel API calls for the same resource, useful when external services don't handle concurrent updates well.
+Include the queue name to allow the same resource to be processed concurrently in different queues.
+
+#### Supersedes Behavior
 
-**4. Queue-specific concurrency:**
+When `supersedes: true` is set, newly queued jobs will automatically delete older pending (not yet running) jobs with the same concurrency key:
+
+**Example scenario:**
+
+```
+1. Job A queued → pending
+2. Job B queued → A deleted, B pending
+3. Job C queued → B deleted, C pending
+4. Only Job C runs
+```
+
+**Configuration:**
 
 ```ts
-concurrency: ({ input, queue }) => `${queue}:process:${input.documentId}`
+concurrency: {
+  key: ({ input }) => `generate:${input.documentId}`,
+  exclusive: true,  // Still enforced
+  supersedes: true, // Delete older pending jobs
+}
 ```
 
-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).
+**When to use:**
+
+- Data regeneration (embeddings, thumbnails) after rapid edits
+- Report generation where only the latest parameters matter
+- Any scenario where intermediate pending jobs are made obsolete by newer ones
+
+**Important notes:**
+
+- Only **pending** jobs (not yet running) are deleted
+- If a job is already **running**, it completes normally and the new job waits
+- Without `exclusive: true`, supersedes still deletes pending jobs but won't prevent parallel execution
 
 #### Important Considerations
 

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

@@ -136,7 +136,7 @@ export type JobTaskStatus = {
  */
 export type ConcurrencyConfig<TInput = object> =
   | ((args: { input: TInput; queue: string }) => string)
-  // Shorthand: key function only, exclusive defaults to true
+  // Shorthand: key function only, exclusive defaults to true, supersedes defaults to false
   | {
       /**
        * Only one job with this key can run at a time.
@@ -150,6 +150,13 @@ export type ConcurrencyConfig<TInput = object> =
        * The queue name is provided to allow for queue-specific concurrency keys if needed.
        */
       key: (args: { input: TInput; queue: string }) => string
+      /**
+       * When a new job is queued, delete older pending (not yet running) jobs with the same key.
+       * Already-running jobs are not affected.
+       * Useful when only the latest state matters (e.g., regenerating data after multiple rapid edits).
+       * @default false
+       */
+      supersedes?: boolean
     }
 
 export type WorkflowConfig<

packages/payload/src/queues/localAPI.ts

@@ -161,6 +161,7 @@ export const getJobsLocalAPI = (payload: Payload) => ({
     // Compute concurrency key from workflow or task config (only if feature is enabled)
     if (payload.config.jobs?.enableConcurrencyControl) {
       let concurrencyKey: null | string = null
+      let supersedes = false
       const queueName = queue || 'default'
 
       if (args.workflow) {
@@ -171,6 +172,7 @@ export const getJobsLocalAPI = (payload: Payload) => ({
             concurrencyKey = concurrencyConfig({ input: args.input, queue: queueName })
           } else {
             concurrencyKey = concurrencyConfig.key({ input: args.input, queue: queueName })
+            supersedes = concurrencyConfig.supersedes ?? false
           }
         }
       } else if (args.task) {
@@ -181,12 +183,43 @@ export const getJobsLocalAPI = (payload: Payload) => ({
             concurrencyKey = concurrencyConfig({ input: args.input, queue: queueName })
           } else {
             concurrencyKey = concurrencyConfig.key({ input: args.input, queue: queueName })
+            supersedes = concurrencyConfig.supersedes ?? false
           }
         }
       }
 
       if (concurrencyKey) {
         data.concurrencyKey = concurrencyKey
+
+        // If supersedes is enabled, delete older pending jobs with the same key
+        if (supersedes) {
+          if (payload.config.jobs.runHooks) {
+            await payload.delete({
+              collection: jobsCollectionSlug,
+              depth: 0,
+              disableTransaction: true,
+              where: {
+                and: [
+                  { concurrencyKey: { equals: concurrencyKey } },
+                  { processing: { equals: false } },
+                  { completedAt: { exists: false } },
+                ],
+              },
+            })
+          } else {
+            await payload.db.deleteMany({
+              collection: jobsCollectionSlug,
+              req,
+              where: {
+                and: [
+                  { concurrencyKey: { equals: concurrencyKey } },
+                  { processing: { equals: false } },
+                  { completedAt: { exists: false } },
+                ],
+              },
+            })
+          }
+        }
       }
     }
 

payload.db

@@ -37,6 +37,7 @@ import { retriesWorkflowLevelTestWorkflow } from './workflows/retriesWorkflowLev
 import { selfCancelWorkflow } from './workflows/selfCancel.js'
 import { subTaskWorkflow } from './workflows/subTask.js'
 import { subTaskFailsWorkflow } from './workflows/subTaskFails.js'
+import { supersedesConcurrencyWorkflow } from './workflows/supersedesConcurrency.js'
 import { updatePostWorkflow } from './workflows/updatePost.js'
 import { updatePostJSONWorkflow } from './workflows/updatePostJSON.js'
 import { workflowAndTasksRetriesUndefinedWorkflow } from './workflows/workflowAndTasksRetriesUndefined.js'
@@ -177,6 +178,7 @@ export const getConfig: () => Partial<Config> = () => ({
       exclusiveConcurrencyWorkflow,
       noConcurrencyWorkflow,
       queueSpecificConcurrencyWorkflow,
+      supersedesConcurrencyWorkflow,
     ],
   },
   editor: lexicalEditor(),