feat: add jobs processing lease migrations

docs/jobs-queue/jobs.mdx

@@ -209,7 +209,8 @@ Each job document contains:
   completedAt: '2024-01-15...',  // When job completed (null if pending)
   hasError: false,              // True if job failed
   totalTried: 1,                // Number of attempts
-  processing: false,            // True if currently running
+  processingUntil: null,        // Future date while a worker owns the job
+  processingToken: null,        // Internal token for the current owner
   taskStatus: {                 // Status of each task (for workflows)
     sendEmail: {
       '1': {
@@ -227,6 +228,8 @@ Each job document contains:
 }
 ```
 
+Payload increments `totalTried` after each completed attempt. While the handler runs, Payload automatically renews `processingUntil`. The claim also gets a `processingToken`, and worker updates only start while that token still matches and enough lease time remains for the configured safety buffer. If the worker disappears, the lease expires and a later jobs runner call can claim the job again. A crashed worker does not increment `totalTried` or consume a handler retry.
+
 #### Access Control
 
 By default, Payload's job operations bypass access control when used from the Local API. You can enable access control by passing `overrideAccess: false` to any job operation.

docs/jobs-queue/queues.mdx

@@ -273,6 +273,36 @@ const results = await payload.jobs.runByID({
 })
 ```
 
+## Processing Leases
+
+Running jobs are owned through a renewable `processingUntil` lease and an internal `processingToken`. Payload renews the lease while the worker is alive. Every worker update must match the token and start while enough lease time remains for the configured safety buffer.
+
+If the worker is killed, its lease eventually expires and a later jobs runner call can claim the job with a new token. A worker crash does not consume the job's retry configuration; retries and backoff still apply normally when a handler reports an error.
+
+By default, Payload only starts a worker update when more than 30 seconds remain on its lease. This safety buffer gives the update time to finish before the lease expires and another worker can claim the job. If too little time remains, the update is rejected and the job becomes claimable after the lease expires.
+
+The default lease duration is 20 minutes and the default safety buffer is 30 seconds. You can change both globally:
+
+```ts
+export default buildConfig({
+  jobs: {
+    processingLease: {
+      duration: 5 * 60 * 1000,
+      safetyBuffer: 60 * 1000,
+    },
+    tasks: [
+      // your tasks here
+    ],
+  },
+})
+```
+
+`processingLease.safetyBuffer` must be non-negative and less than `processingLease.duration`. Set it to `0` only when your database adapter checks ownership and performs the update atomically, as the MongoDB adapter does.
+
+The buffer should be longer than a normal job database update. It is a safety window for adapters that read and write separately, not a task timeout. Healthy long-running jobs keep renewing their leases and can run longer than `processingLease.duration`.
+
+Expiry does not start a new process by itself, so serverless deployments must continue calling the jobs runner from a cron or another callback system.
+
 ## Processing Order
 
 By default, jobs are processed first in, first out (FIFO). This means that the first job added to the queue will be the first one processed. However, you can also configure the order in which jobs are processed.
@@ -626,7 +656,9 @@ jobsCollectionOverrides: ({ defaultJobsCollection }) => ({
 
 Look for jobs with:
 
-- `processing: true` but stuck → Worker may have crashed
+- `processingUntil` in the future → A worker currently owns the job
+- `processingUntil` in the past → The worker may have crashed; the next runner call can claim it
+- `processingToken` set → Internal identity of the worker that owns the current lease
 - `hasError: true` → Check the `log` field for errors
 - `completedAt: null` → Job hasn't run yet
 

docs/jobs-queue/workflows.mdx

@@ -415,7 +415,7 @@ When you define a `concurrency` key:
 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
+   - 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 have their processing lease cleared 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.
 
@@ -527,4 +527,4 @@ 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.
+- **Pending jobs wait:** Jobs that can't run due to concurrency constraints remain in the queue without a `processingUntil` lease or `processingToken` and will be picked up on subsequent runs.

docs/migration-guide/v4.mdx

@@ -861,6 +861,22 @@ Payload now uses a unique `processingToken` when claiming jobs so multiple worke
 
 The default `updateJobs` implementation has been removed because it could not safely claim jobs across multiple workers. If you maintain a custom database adapter, you must now provide `updateJobs`. It must only claim jobs that still match the provided `where` condition and, when a `processingToken` is provided, return only the jobs claimed with that token. The official Payload database adapters already implement this behavior.
 
+### Jobs: crashed workers are recovered with processing leases
+
+The permanent `processing` flag on `payload-jobs` has been replaced by a renewable `processingUntil` lease. If a worker exits, a later jobs runner call can claim the job after its lease expires. Worker crashes do not consume the job's handler retry configuration.
+
+Worker updates use a 30-second safety buffer by default, so an update only starts when enough lease time remains for it to finish before another claim. Configure the lease with `jobs.processingLease.duration` and `jobs.processingLease.safetyBuffer`.
+
+Existing SQL databases need a migration that replaces `processing` with the nullable, indexed `processingUntil` column. MongoDB does not require a migration.
+
+One predefined migration handles both Jobs Queue changes. Generate it with the package for your database adapter:
+
+```sh
+pnpm payload migrate:create --file @payloadcms/db-postgres/jobs-processing-lease-v4
+```
+
+Replace `@payloadcms/db-postgres` with `@payloadcms/db-vercel-postgres`, `@payloadcms/db-sqlite`, or `@payloadcms/db-d1-sqlite` when using one of those adapters. The migration keeps custom table, schema, and index names. Any old job with `processing: true` receives an expired lease so a jobs runner can recover it.
+
 ### Jobs: the `state` field on task return values has been removed ([#16416](https://github.com/payloadcms/payload/pull/16416))
 
 Job task handlers and `inlineTask` callbacks can no longer use the `state` field to signal success or failure. Both shapes were deprecated in v3 and are now removed from tasks and inline tasks.

packages/db-d1-sqlite/src/exports/migration-utils.ts

@@ -1 +1 @@
-export { getBlocksToJsonMigrator } from '@payloadcms/drizzle'
+export { getBlocksToJsonMigrator, migrateJobsProcessingLease } from '@payloadcms/drizzle'

packages/db-d1-sqlite/src/predefinedMigrations/jobs-processing-lease-v4.ts

@@ -0,0 +1,10 @@
+import type { DynamicMigrationTemplate } from 'payload'
+
+import { buildDynamicPredefinedJobsProcessingLeaseMigration } from '@payloadcms/drizzle'
+
+export const dynamic: DynamicMigrationTemplate = buildDynamicPredefinedJobsProcessingLeaseMigration(
+  {
+    dialect: 'sqlite',
+    packageName: '@payloadcms/db-d1-sqlite',
+  },
+)

packages/db-postgres/src/exports/migration-utils.ts

@@ -1,3 +1,3 @@
 export { migrateLocalizeStatus } from '../predefinedMigrations/migrateLocalizeStatus.js'
-export { getBlocksToJsonMigrator } from '@payloadcms/drizzle'
+export { getBlocksToJsonMigrator, migrateJobsProcessingLease } from '@payloadcms/drizzle'
 export { migratePostgresLocalizeStatus, migratePostgresV2toV3 } from '@payloadcms/drizzle/postgres'

packages/db-postgres/src/predefinedMigrations/jobs-processing-lease-v4.ts

@@ -0,0 +1,10 @@
+import type { DynamicMigrationTemplate } from 'payload'
+
+import { buildDynamicPredefinedJobsProcessingLeaseMigration } from '@payloadcms/drizzle'
+
+export const dynamic: DynamicMigrationTemplate = buildDynamicPredefinedJobsProcessingLeaseMigration(
+  {
+    dialect: 'postgres',
+    packageName: '@payloadcms/db-postgres',
+  },
+)

packages/db-sqlite/src/exports/migration-utils.ts

@@ -1,3 +1,3 @@
 export { migrateLocalizeStatus } from '../predefinedMigrations/migrateLocalizeStatus.js'
-export { getBlocksToJsonMigrator } from '@payloadcms/drizzle'
+export { getBlocksToJsonMigrator, migrateJobsProcessingLease } from '@payloadcms/drizzle'
 export { migrateSqliteLocalizeStatus } from '@payloadcms/drizzle/sqlite'

packages/db-sqlite/src/predefinedMigrations/jobs-processing-lease-v4.ts

@@ -0,0 +1,10 @@
+import type { DynamicMigrationTemplate } from 'payload'
+
+import { buildDynamicPredefinedJobsProcessingLeaseMigration } from '@payloadcms/drizzle'
+
+export const dynamic: DynamicMigrationTemplate = buildDynamicPredefinedJobsProcessingLeaseMigration(
+  {
+    dialect: 'sqlite',
+    packageName: '@payloadcms/db-sqlite',
+  },
+)

packages/db-vercel-postgres/src/exports/migration-utils.ts

@@ -1 +1,2 @@
+export { migrateJobsProcessingLease } from '@payloadcms/drizzle'
 export { migratePostgresV2toV3 } from '@payloadcms/drizzle/postgres'

packages/db-vercel-postgres/src/predefinedMigrations/jobs-processing-lease-v4.ts

@@ -0,0 +1,10 @@
+import type { DynamicMigrationTemplate } from 'payload'
+
+import { buildDynamicPredefinedJobsProcessingLeaseMigration } from '@payloadcms/drizzle'
+
+export const dynamic: DynamicMigrationTemplate = buildDynamicPredefinedJobsProcessingLeaseMigration(
+  {
+    dialect: 'postgres',
+    packageName: '@payloadcms/db-vercel-postgres',
+  },
+)

packages/drizzle/src/index.ts

@@ -90,6 +90,10 @@ export { createSchemaGenerator } from './utilities/createSchemaGenerator.js'
 export { executeSchemaHooks } from './utilities/executeSchemaHooks.js'
 export { extendDrizzleTable } from './utilities/extendDrizzleTable.js'
 export { hasLocalesTable } from './utilities/hasLocalesTable.js'
+export {
+  buildDynamicPredefinedJobsProcessingLeaseMigration,
+  migrateJobsProcessingLease,
+} from './utilities/jobsProcessingLeaseMigration.js'
 export { pushDevSchema } from './utilities/pushDevSchema.js'
 export { validateExistingBlockIsIdentical } from './utilities/validateExistingBlockIsIdentical.js'
 import { findMigrationDir as payloadFindMigrationDir } from 'payload'

packages/drizzle/src/updateJobs.ts

@@ -1,7 +1,7 @@
 import type { LibSQLDatabase } from 'drizzle-orm/libsql'
 import type { UpdateJobs, Where } from 'payload'
 
-import { and, eq, inArray } from 'drizzle-orm'
+import { and, inArray, isNull, lte, or } from 'drizzle-orm'
 import toSnakeCase from 'to-snake-case'
 
 import type { DrizzleAdapter } from './types.js'
@@ -15,10 +15,10 @@ import { getTransaction } from './utilities/getTransaction.js'
 import { markWrite } from './utilities/readAfterWrite.js'
 
 const isInitialJobClaim = (data: Record<string, unknown>): boolean =>
-  data.processing === true &&
   typeof data.processingToken === 'string' &&
+  typeof data.processingUntil === 'string' &&
   Object.keys(data).every(
-    (field) => field === 'processing' || field === 'processingToken' || field === 'updatedAt',
+    (field) => field === 'processingToken' || field === 'processingUntil' || field === 'updatedAt',
   )
 
 export const updateJobs: UpdateJobs = async function updateMany(
@@ -81,8 +81,12 @@ export const updateJobs: UpdateJobs = async function updateMany(
   const db = getPrimaryDb(this, await getTransaction(this, req))
 
   if (isInitialJobClaim(data)) {
+    // Atomically claim jobs so multiple workers cannot pick up the same job.
+    // We don't need FOR UPDATE SKIP LOCKED here because of the compare-and-swap-style update using processingUntil and processingToken.
+    // The first update makes the where clause fail for other workers, so they cannot claim the same job.
     const _db = db as LibSQLDatabase
     const table = this.tables[tableName]
+    const now = new Date().toISOString()
     const { row } = transformForWrite({
       adapter: this,
       data,
@@ -99,7 +103,12 @@ export const updateJobs: UpdateJobs = async function updateMany(
     const claimedRows = await _db
       .update(table)
       .set(row)
-      .where(and(inArray(table.id, jobIDs), eq(table.processing, false)))
+      .where(
+        and(
+          inArray(table.id, jobIDs),
+          or(isNull(table.processingUntil), lte(table.processingUntil, now)),
+        ),
+      )
       .returning({ id: table.id })
 
     if (claimedRows.length) {

packages/drizzle/src/utilities/jobsProcessingLeaseMigration.spec.ts

@@ -0,0 +1,145 @@
+import type { Client } from '@libsql/client'
+import type { Payload } from 'payload'
+
+import { createClient } from '@libsql/client'
+import { sql } from 'drizzle-orm'
+import { drizzle } from 'drizzle-orm/libsql'
+import { existsSync, mkdtempSync, rmSync } from 'fs'
+import { tmpdir } from 'os'
+import path from 'path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+import type { DrizzleAdapter } from '../types.js'
+
+import {
+  buildDynamicPredefinedJobsProcessingLeaseMigration,
+  migrateJobsProcessingLease,
+} from './jobsProcessingLeaseMigration.js'
+
+describe('jobs processing lease predefined migration', () => {
+  let client: Client | undefined
+  const temporaryDirectories: string[] = []
+
+  afterEach(() => {
+    client?.close()
+    client = undefined
+
+    for (const directory of temporaryDirectories) {
+      rmSync(directory, { force: true, recursive: true })
+    }
+    temporaryDirectories.length = 0
+  })
+
+  it('should run the SQLite migration up and down', async () => {
+    client = createClient({ url: ':memory:' })
+    const db = drizzle(client)
+    await client.execute(
+      'CREATE TABLE "custom_jobs" ("id" integer primary key, "processing" integer DEFAULT false)',
+    )
+    await client.execute(
+      'CREATE INDEX "custom_jobs_processing_idx" ON "custom_jobs" ("processing")',
+    )
+    await client.execute(
+      'INSERT INTO "custom_jobs" ("id", "processing") VALUES (1, true), (2, false)',
+    )
+
+    await migrateJobsProcessingLease({
+      db,
+      dialect: 'sqlite',
+      direction: 'up',
+      newIndexName: 'custom_jobs_processing_until_idx',
+      oldIndexName: 'custom_jobs_processing_idx',
+      sql,
+      tableName: 'custom_jobs',
+    })
+
+    const migratedJobs = await client.execute(
+      'SELECT "id", "processing_until", "processing_token" FROM "custom_jobs" ORDER BY "id"',
+    )
+    expect(migratedJobs.rows).toEqual([
+      { id: 1, processing_token: null, processing_until: '1970-01-01T00:00:00.000Z' },
+      { id: 2, processing_token: null, processing_until: null },
+    ])
+
+    await migrateJobsProcessingLease({
+      db,
+      dialect: 'sqlite',
+      direction: 'down',
+      newIndexName: 'custom_jobs_processing_until_idx',
+      oldIndexName: 'custom_jobs_processing_idx',
+      sql,
+      tableName: 'custom_jobs',
+    })
+
+    const restoredJobs = await client.execute(
+      'SELECT "id", "processing" FROM "custom_jobs" ORDER BY "id"',
+    )
+    expect(restoredJobs.rows).toEqual([
+      { id: 1, processing: 1 },
+      { id: 2, processing: 0 },
+    ])
+  })
+
+  it('should run PostgreSQL statements with custom names', async () => {
+    const execute = vi.fn()
+    const raw = vi.fn(sql.raw)
+
+    await migrateJobsProcessingLease({
+      db: { execute },
+      dialect: 'postgres',
+      direction: 'up',
+      newIndexName: 'custom_jobs_processing_until_idx',
+      oldIndexName: 'custom_jobs_processing_idx',
+      schemaName: 'custom_schema',
+      sql: { raw },
+      tableName: 'custom_jobs',
+    })
+
+    expect(raw.mock.calls.map(([statement]) => statement)).toContain(
+      'UPDATE "custom_schema"."custom_jobs" SET "processing_until" = \'1970-01-01T00:00:00.000Z\' WHERE "processing" = true',
+    )
+    expect(raw.mock.calls.map(([statement]) => statement)).toContain(
+      'DROP INDEX IF EXISTS "custom_schema"."custom_jobs_processing_idx"',
+    )
+    expect(execute).toHaveBeenCalledTimes(6)
+  })
+
+  it('should generate a migration using the adapter names', async () => {
+    const temporaryDirectory = mkdtempSync(path.join(tmpdir(), 'payload-jobs-migration-'))
+    temporaryDirectories.push(temporaryDirectory)
+    const filePath = path.join(temporaryDirectory, 'migration')
+    const dynamic = buildDynamicPredefinedJobsProcessingLeaseMigration({
+      dialect: 'postgres',
+      packageName: '@payloadcms/db-postgres',
+    })
+    const migration = await dynamic({
+      filePath,
+      payload: {
+        db: {
+          rawTables: {
+            custom_jobs: {
+              indexes: {
+                processingUntil: {
+                  name: 'custom_jobs_processing_until_idx',
+                  on: 'processingUntil',
+                },
+              },
+            },
+          },
+          requireDrizzleKit: () => ({
+            generateDrizzleJson: vi.fn().mockResolvedValue({ version: 'test-snapshot' }),
+          }),
+          schema: {},
+          schemaName: 'custom_schema',
+          tableNameMap: new Map([['payload_jobs', 'custom_jobs']]),
+        } as unknown as DrizzleAdapter,
+      } as Payload,
+    })
+
+    expect(existsSync(`${filePath}.json`)).toBe(true)
+    expect(migration.imports).toContain('@payloadcms/db-postgres/migration-utils')
+    expect(migration.upSQL).toContain("direction: 'up'")
+    expect(migration.upSQL).toContain('custom_jobs_processing_until_idx')
+    expect(migration.downSQL).toContain("direction: 'down'")
+  })
+})