diff --git a/docs/jobs-queue/jobs.mdx b/docs/jobs-queue/jobs.mdx index f103ffac973..c1959165fe0 100644 --- a/docs/jobs-queue/jobs.mdx +++ b/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. diff --git a/docs/jobs-queue/queues.mdx b/docs/jobs-queue/queues.mdx index 51298e05d38..555f3e0e13f 100644 --- a/docs/jobs-queue/queues.mdx +++ b/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 diff --git a/docs/jobs-queue/workflows.mdx b/docs/jobs-queue/workflows.mdx index ef80d2b3454..5aaece069a1 100644 --- a/docs/jobs-queue/workflows.mdx +++ b/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. diff --git a/docs/migration-guide/v4.mdx b/docs/migration-guide/v4.mdx index 0ce0742509a..847e0c13dc0 100644 --- a/docs/migration-guide/v4.mdx +++ b/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. diff --git a/packages/db-d1-sqlite/src/exports/migration-utils.ts b/packages/db-d1-sqlite/src/exports/migration-utils.ts index 56f26820e39..44509aa7b71 100644 --- a/packages/db-d1-sqlite/src/exports/migration-utils.ts +++ b/packages/db-d1-sqlite/src/exports/migration-utils.ts @@ -1 +1 @@ -export { getBlocksToJsonMigrator } from '@payloadcms/drizzle' +export { getBlocksToJsonMigrator, migrateJobsProcessingLease } from '@payloadcms/drizzle' diff --git a/packages/db-d1-sqlite/src/predefinedMigrations/jobs-processing-lease-v4.ts b/packages/db-d1-sqlite/src/predefinedMigrations/jobs-processing-lease-v4.ts new file mode 100644 index 00000000000..f423adfea62 --- /dev/null +++ b/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', + }, +) diff --git a/packages/db-postgres/src/exports/migration-utils.ts b/packages/db-postgres/src/exports/migration-utils.ts index 5cc7c1543f3..494a98e5ab2 100644 --- a/packages/db-postgres/src/exports/migration-utils.ts +++ b/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' diff --git a/packages/db-postgres/src/predefinedMigrations/jobs-processing-lease-v4.ts b/packages/db-postgres/src/predefinedMigrations/jobs-processing-lease-v4.ts new file mode 100644 index 00000000000..7501f46601a --- /dev/null +++ b/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', + }, +) diff --git a/packages/db-sqlite/src/exports/migration-utils.ts b/packages/db-sqlite/src/exports/migration-utils.ts index 70068fb26d8..62ea472d637 100644 --- a/packages/db-sqlite/src/exports/migration-utils.ts +++ b/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' diff --git a/packages/db-sqlite/src/predefinedMigrations/jobs-processing-lease-v4.ts b/packages/db-sqlite/src/predefinedMigrations/jobs-processing-lease-v4.ts new file mode 100644 index 00000000000..0f9c948f379 --- /dev/null +++ b/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', + }, +) diff --git a/packages/db-vercel-postgres/src/exports/migration-utils.ts b/packages/db-vercel-postgres/src/exports/migration-utils.ts index f5b74e7fa2b..7f1545c6609 100644 --- a/packages/db-vercel-postgres/src/exports/migration-utils.ts +++ b/packages/db-vercel-postgres/src/exports/migration-utils.ts @@ -1 +1,2 @@ +export { migrateJobsProcessingLease } from '@payloadcms/drizzle' export { migratePostgresV2toV3 } from '@payloadcms/drizzle/postgres' diff --git a/packages/db-vercel-postgres/src/predefinedMigrations/jobs-processing-lease-v4.ts b/packages/db-vercel-postgres/src/predefinedMigrations/jobs-processing-lease-v4.ts new file mode 100644 index 00000000000..e945a2d08da --- /dev/null +++ b/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', + }, +) diff --git a/packages/drizzle/src/index.ts b/packages/drizzle/src/index.ts index ceac40303d9..b24459efd9e 100644 --- a/packages/drizzle/src/index.ts +++ b/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' diff --git a/packages/drizzle/src/updateJobs.ts b/packages/drizzle/src/updateJobs.ts index 5093739145a..c97d5260cf5 100644 --- a/packages/drizzle/src/updateJobs.ts +++ b/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): 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) { diff --git a/packages/drizzle/src/utilities/jobsProcessingLeaseMigration.spec.ts b/packages/drizzle/src/utilities/jobsProcessingLeaseMigration.spec.ts new file mode 100644 index 00000000000..4b553b4da6a --- /dev/null +++ b/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'") + }) +}) diff --git a/packages/drizzle/src/utilities/jobsProcessingLeaseMigration.ts b/packages/drizzle/src/utilities/jobsProcessingLeaseMigration.ts new file mode 100644 index 00000000000..4203cebec85 --- /dev/null +++ b/packages/drizzle/src/utilities/jobsProcessingLeaseMigration.ts @@ -0,0 +1,129 @@ +import type { SQL } from 'drizzle-orm' +import type { DynamicMigrationTemplate } from 'payload' + +import { writeFileSync } from 'fs' +import toSnakeCase from 'to-snake-case' + +import type { DrizzleAdapter } from '../types.js' + +type MigrationArgs = { + direction: 'down' | 'up' + newIndexName: string + oldIndexName: string + schemaName?: string + sql: { raw: (statement: string) => SQL } + tableName: string +} & ( + | { + db: { execute: (statement: SQL) => unknown } + dialect: 'postgres' + } + | { + db: { run: (statement: SQL) => unknown } + dialect: 'sqlite' + } +) + +const quoteIdentifier = (identifier: string): string => `"${identifier.replaceAll('"', '""')}"` + +const getDefaultIndexName = (name: string): string => { + const suffix = '_idx' + + return `${name.slice(0, 60 - suffix.length)}${suffix}` +} + +export async function migrateJobsProcessingLease(args: MigrationArgs): Promise { + const { direction, newIndexName, oldIndexName, schemaName = 'public', sql, tableName } = args + const quotedTableName = quoteIdentifier(tableName) + const qualifiedTableName = `${quoteIdentifier(schemaName)}.${quotedTableName}` + const upPostgres = [ + `ALTER TABLE ${qualifiedTableName} ADD COLUMN "processing_until" timestamp(3) with time zone`, + `ALTER TABLE ${qualifiedTableName} ADD COLUMN "processing_token" varchar`, + `UPDATE ${qualifiedTableName} SET "processing_until" = '1970-01-01T00:00:00.000Z' WHERE "processing" = true`, + `DROP INDEX IF EXISTS ${quoteIdentifier(schemaName)}.${quoteIdentifier(oldIndexName)}`, + `ALTER TABLE ${qualifiedTableName} DROP COLUMN "processing"`, + `CREATE INDEX ${quoteIdentifier(newIndexName)} ON ${qualifiedTableName} USING btree ("processing_until")`, + ] + const downPostgres = [ + `ALTER TABLE ${qualifiedTableName} ADD COLUMN "processing" boolean DEFAULT false`, + `UPDATE ${qualifiedTableName} SET "processing" = true WHERE "processing_until" IS NOT NULL`, + `DROP INDEX IF EXISTS ${quoteIdentifier(schemaName)}.${quoteIdentifier(newIndexName)}`, + `ALTER TABLE ${qualifiedTableName} DROP COLUMN "processing_token"`, + `ALTER TABLE ${qualifiedTableName} DROP COLUMN "processing_until"`, + `CREATE INDEX ${quoteIdentifier(oldIndexName)} ON ${qualifiedTableName} USING btree ("processing")`, + ] + const upSQLite = [ + `ALTER TABLE ${quotedTableName} ADD COLUMN "processing_until" text`, + `ALTER TABLE ${quotedTableName} ADD COLUMN "processing_token" text`, + `UPDATE ${quotedTableName} SET "processing_until" = '1970-01-01T00:00:00.000Z' WHERE "processing" = true`, + `DROP INDEX IF EXISTS ${quoteIdentifier(oldIndexName)}`, + `ALTER TABLE ${quotedTableName} DROP COLUMN "processing"`, + `CREATE INDEX ${quoteIdentifier(newIndexName)} ON ${quotedTableName} ("processing_until")`, + ] + const downSQLite = [ + `ALTER TABLE ${quotedTableName} ADD COLUMN "processing" integer DEFAULT false`, + `UPDATE ${quotedTableName} SET "processing" = true WHERE "processing_until" IS NOT NULL`, + `DROP INDEX IF EXISTS ${quoteIdentifier(newIndexName)}`, + `ALTER TABLE ${quotedTableName} DROP COLUMN "processing_token"`, + `ALTER TABLE ${quotedTableName} DROP COLUMN "processing_until"`, + `CREATE INDEX ${quoteIdentifier(oldIndexName)} ON ${quotedTableName} ("processing")`, + ] + + if (args.dialect === 'postgres') { + const statements = direction === 'up' ? upPostgres : downPostgres + + for (const statement of statements) { + await args.db.execute(sql.raw(statement)) + } + } else { + const statements = direction === 'up' ? upSQLite : downSQLite + + for (const statement of statements) { + await args.db.run(sql.raw(statement)) + } + } +} + +export const buildDynamicPredefinedJobsProcessingLeaseMigration = ({ + dialect, + packageName, +}: { + dialect: 'postgres' | 'sqlite' + packageName: string +}): DynamicMigrationTemplate => { + return async ({ filePath, payload }) => { + const adapter = payload.db as DrizzleAdapter + const tableName = adapter.tableNameMap.get(toSnakeCase('payload-jobs')) + + if (!tableName) { + throw new Error('Could not find the payload-jobs database table') + } + + const newIndexName = + Object.values(adapter.rawTables[tableName]?.indexes ?? {}).find( + (index) => index.on === 'processingUntil', + )?.name ?? getDefaultIndexName(`${tableName}_processing_until`) + const oldIndexName = getDefaultIndexName(`${tableName}_processing`) + const drizzleSnapshot = await adapter.requireDrizzleKit().generateDrizzleJson(adapter.schema) + const sharedArgs = ` + db, + dialect: ${JSON.stringify(dialect)}, + newIndexName: ${JSON.stringify(newIndexName)}, + oldIndexName: ${JSON.stringify(oldIndexName)}, + schemaName: ${JSON.stringify(adapter.schemaName)}, + sql, + tableName: ${JSON.stringify(tableName)},` + + writeFileSync(`${filePath}.json`, JSON.stringify(drizzleSnapshot, null, 2)) + + return { + downSQL: `await migrateJobsProcessingLease({${sharedArgs} + direction: 'down', + })`, + imports: `import { migrateJobsProcessingLease } from '${packageName}/migration-utils'`, + upSQL: `await migrateJobsProcessingLease({${sharedArgs} + direction: 'up', + })`, + } + } +} diff --git a/packages/payload/src/config/defaults.ts b/packages/payload/src/config/defaults.ts index ca90af675e2..bf7da9824a7 100644 --- a/packages/payload/src/config/defaults.ts +++ b/packages/payload/src/config/defaults.ts @@ -65,6 +65,10 @@ export const addDefaultsToConfig = (config: Config): Config => { run: defaultAccess, ...(config.jobs?.access || {}), }, + processingLease: { + duration: config.jobs?.processingLease?.duration ?? 20 * 60 * 1000, + safetyBuffer: config.jobs?.processingLease?.safetyBuffer ?? 30 * 1000, + }, } as JobsConfig config.localization = config.localization ?? false config.maxDepth = config.maxDepth ?? 10 diff --git a/packages/payload/src/database/types.ts b/packages/payload/src/database/types.ts index f4c45542073..3d9d7199cfe 100644 --- a/packages/payload/src/database/types.ts +++ b/packages/payload/src/database/types.ts @@ -633,7 +633,7 @@ export type UpdateJobsArgs = { /** * Updates jobs matching the provided `where` condition. * - * Job claims must only update jobs that are still not processing. + * Job claims must only update jobs whose lease is missing or expired. * When a `processingToken` is provided, only jobs updated with that token may be returned. */ export type UpdateJobs = (args: UpdateJobsArgs) => Promise diff --git a/packages/payload/src/queues/config/collection.ts b/packages/payload/src/queues/config/collection.ts index 496b7082bfc..c951ff8ab97 100644 --- a/packages/payload/src/queues/config/collection.ts +++ b/packages/payload/src/queues/config/collection.ts @@ -226,21 +226,17 @@ export const getDefaultJobsCollection: (jobsConfig: SanitizedConfig['jobs']) => index: true, }, { - name: 'processing', - type: 'checkbox', + name: 'processingUntil', + type: 'date', admin: { + date: { pickerAppearance: 'dayAndTime' }, position: 'sidebar', }, - defaultValue: false, index: true, }, { name: 'processingToken', type: 'text', - admin: { - hidden: true, - readOnly: true, - }, }, // Only add concurrencyKey field if concurrency control is enabled ...(jobsConfig.enableConcurrencyControl @@ -274,8 +270,8 @@ export const getDefaultJobsCollection: (jobsConfig: SanitizedConfig['jobs']) => beforeChange: [ ({ data, originalDoc }) => { if (originalDoc?.error?.cancelled) { - data.processing = false data.processingToken = null + data.processingUntil = null data.hasError = true delete data.completedAt delete data.waitUntil diff --git a/packages/payload/src/queues/config/types/index.ts b/packages/payload/src/queues/config/types/index.ts index 816ba6d25a1..e38798f43d1 100644 --- a/packages/payload/src/queues/config/types/index.ts +++ b/packages/payload/src/queues/config/types/index.ts @@ -81,12 +81,36 @@ export type CancelJobAccessArgs = { export type CancelJobAccess = (args: CancelJobAccessArgs) => MaybePromise export type QueueJobAccess = (args: QueueJobAccessArgs) => MaybePromise +export type ProcessingLeaseConfig = { + /** + * How long a job remains claimed without a successful lease renewal. + * This is not a maximum job runtime; active jobs renew their lease automatically. + * + * @default 1200000 (20 minutes) + */ + duration?: number + /** + * Minimum time that must remain on a lease before a worker **starts** an update. This gives the + * update time to finish before another worker can claim the job. + * + * Set this to `0` if your database adapter applies the ownership check and update atomically. + * If the database update itself is not atomic, increase this value, as another worker + * may claim the job before the update finishes. + * + * It must be less than `duration`. + * + * @default 30000 (30 seconds) + */ + safetyBuffer?: number +} + export type SanitizedJobsConfig = { /** * If set to `true`, the job system is enabled and a payload-jobs collection exists. * This property is automatically set during sanitization. */ enabled?: boolean + processingLease: Required /** * If set to `true`, at least one task or workflow has scheduling enabled. * This property is automatically set during sanitization. @@ -97,7 +121,7 @@ export type SanitizedJobsConfig = { * This property is automatically set during sanitization. */ stats?: boolean -} & JobsConfig +} & Omit export type JobsConfig = { /** * Specify access control to determine who can interact with jobs. @@ -152,6 +176,7 @@ export type JobsConfig = { * a new collection. */ jobsCollectionOverrides?: (args: { defaultJobsCollection: CollectionConfig }) => CollectionConfig + processingLease?: ProcessingLeaseConfig /** * Adjust the job processing order using a Payload sort string. This can be set globally or per queue. * diff --git a/packages/payload/src/queues/config/types/workflowTypes.ts b/packages/payload/src/queues/config/types/workflowTypes.ts index dd50c1bd43f..9d83901ea2e 100644 --- a/packages/payload/src/queues/config/types/workflowTypes.ts +++ b/packages/payload/src/queues/config/types/workflowTypes.ts @@ -66,8 +66,8 @@ export type BaseJob< */ scheduled?: boolean } - processing?: boolean processingToken?: null | string + processingUntil?: null | string queue?: string taskSlug?: null | TaskType taskStatus: JobTaskStatus diff --git a/packages/payload/src/queues/errors/handleTaskError.ts b/packages/payload/src/queues/errors/handleTaskError.ts index a6c310a93a9..1c8a53f7929 100644 --- a/packages/payload/src/queues/errors/handleTaskError.ts +++ b/packages/payload/src/queues/errors/handleTaskError.ts @@ -115,7 +115,7 @@ export async function handleTaskError({ log: { $push: taskLogToPush, } as any, - processing: false, + processingUntil: null, totalTried: (job.totalTried ?? 0) + 1, waitUntil: job.waitUntil, }) @@ -177,7 +177,7 @@ export async function handleTaskError({ log: { $push: taskLogToPush, } as any, - processing: false, + processingUntil: null, totalTried: (job.totalTried ?? 0) + 1, waitUntil: job.waitUntil, }) diff --git a/packages/payload/src/queues/errors/handleWorkflowError.ts b/packages/payload/src/queues/errors/handleWorkflowError.ts index bab08abadd2..a02f98708ad 100644 --- a/packages/payload/src/queues/errors/handleWorkflowError.ts +++ b/packages/payload/src/queues/errors/handleWorkflowError.ts @@ -87,7 +87,7 @@ export async function handleWorkflowError({ await updateJob({ error: errorJSON, hasError: hasFinalError, // If reached max retries => final error. If hasError is true this job will not be retried - processing: false, + processingUntil: null, totalTried: (job.totalTried ?? 0) + 1, waitUntil: job.waitUntil, }) diff --git a/packages/payload/src/queues/errors/index.ts b/packages/payload/src/queues/errors/index.ts index 5622b221636..55df4405659 100644 --- a/packages/payload/src/queues/errors/index.ts +++ b/packages/payload/src/queues/errors/index.ts @@ -39,6 +39,12 @@ export class WorkflowError extends Error { } } +/** + * Stops the current job run after this worker loses ownership. + * The job remains recoverable by another worker. + */ +export class JobRunAbortedError extends Error {} + /** * Throw this error from within a task or workflow handler to cancel the job. * Unlike failing a job (e.g. by throwing any other error), a cancelled job will not be retried. diff --git a/packages/payload/src/queues/localAPI.ts b/packages/payload/src/queues/localAPI.ts index fe390cbc9dd..5d053848eab 100644 --- a/packages/payload/src/queues/localAPI.ts +++ b/packages/payload/src/queues/localAPI.ts @@ -199,7 +199,7 @@ export const getJobsLocalAPI = (payload: Payload) => ({ where: { and: [ { concurrencyKey: { equals: concurrencyKey } }, - { processing: { equals: false } }, + { processingUntil: { exists: false } }, { completedAt: { exists: false } }, ], }, @@ -378,7 +378,7 @@ export const getJobsLocalAPI = (payload: Payload) => ({ cancelled: true, }, hasError: true, - processing: false, + processingUntil: null, waitUntil: null, }, req, @@ -422,7 +422,7 @@ export const getJobsLocalAPI = (payload: Payload) => ({ cancelled: true, }, hasError: true, - processing: false, + processingUntil: null, waitUntil: null, }, req, diff --git a/packages/payload/src/queues/operations/runJobs/heartbeat.ts b/packages/payload/src/queues/operations/runJobs/heartbeat.ts new file mode 100644 index 00000000000..57a28bd08d5 --- /dev/null +++ b/packages/payload/src/queues/operations/runJobs/heartbeat.ts @@ -0,0 +1,76 @@ +import type { PayloadRequest } from '../../../types/index.js' +import type { RunJobsSilent } from '../../localAPI.js' + +import { getCurrentDate } from '../../utilities/getCurrentDate.js' +import { updateJobs } from '../../utilities/updateJob.js' + +type ProcessingLeaseHeartbeatArgs = { + jobIDs: (number | string)[] + processingLeaseDuration: number + processingLeaseSafetyBuffer: number + processingToken: string + req: PayloadRequest + silent: RunJobsSilent +} + +export function startProcessingLeaseHeartbeat(args: ProcessingLeaseHeartbeatArgs): () => void { + const { processingLeaseDuration } = args + const heartbeatInterval = Math.max(Math.floor(processingLeaseDuration / 3), 1) + let isStopped = false + let timeout: ReturnType | undefined + + const scheduleHeartbeat = () => { + if (isStopped) { + return + } + + timeout = setTimeout(() => { + void sendHeartbeat(args).finally(scheduleHeartbeat) + }, heartbeatInterval) + timeout.unref?.() + } + + scheduleHeartbeat() + + return () => { + isStopped = true + if (timeout) { + clearTimeout(timeout) + } + } +} + +async function sendHeartbeat({ + jobIDs, + processingLeaseDuration, + processingLeaseSafetyBuffer, + processingToken, + req, + silent, +}: ProcessingLeaseHeartbeatArgs): Promise { + const now = getCurrentDate() + const minimumProcessingUntil = new Date(now.getTime() + processingLeaseSafetyBuffer).toISOString() + const processingUntil = new Date(now.getTime() + processingLeaseDuration).toISOString() + + try { + await updateJobs({ + data: { processingUntil }, + req, + returning: false, + where: { + and: [ + { id: { in: jobIDs } }, + { processingToken: { equals: processingToken } }, + { processingUntil: { greater_than: minimumProcessingUntil } }, + ], + }, + }) + } catch (error) { + if (!silent || (typeof silent === 'object' && !silent.error)) { + req.payload.logger.error({ + err: error, + msg: 'Failed to renew job processing leases', + }) + } + } +} diff --git a/packages/payload/src/queues/operations/runJobs/index.ts b/packages/payload/src/queues/operations/runJobs/index.ts index cadac3f5472..2483d5f4db3 100644 --- a/packages/payload/src/queues/operations/runJobs/index.ts +++ b/packages/payload/src/queues/operations/runJobs/index.ts @@ -10,9 +10,10 @@ import type { RunJobResult } from './runJob/index.js' import { Forbidden } from '../../../errors/Forbidden.js' import { isolateObjectProperty } from '../../../utilities/isolateObjectProperty.js' import { jobsCollectionSlug } from '../../config/collection.js' -import { JobCancelledError } from '../../errors/index.js' +import { JobCancelledError, JobRunAbortedError } from '../../errors/index.js' import { getCurrentDate } from '../../utilities/getCurrentDate.js' -import { updateJob, updateJobs } from '../../utilities/updateJob.js' +import { updateJobs } from '../../utilities/updateJob.js' +import { startProcessingLeaseHeartbeat } from './heartbeat.js' import { getUpdateJobFunction } from './runJob/getUpdateJobFunction.js' import { importHandlerPath } from './runJob/importHandlerPath.js' import { runJob } from './runJob/index.js' @@ -111,6 +112,11 @@ export const runJobs = async (args: RunJobsArgs): Promise => { throw new Forbidden(req.t) } } + const now = getCurrentDate() + const { duration: processingLeaseDuration, safetyBuffer: processingLeaseSafetyBuffer } = + jobsConfig.processingLease + const nowISOString = now.toISOString() + const processingUntil = new Date(now.getTime() + processingLeaseDuration).toISOString() const processingToken = uuid() const and: Where[] = [ { @@ -124,9 +130,10 @@ export const runJobs = async (args: RunJobsArgs): Promise => { }, }, { - processing: { - equals: false, - }, + or: [ + { processingUntil: { exists: false } }, + { processingUntil: { less_than_equal: nowISOString } }, + ], }, { or: [ @@ -137,7 +144,7 @@ export const runJobs = async (args: RunJobsArgs): Promise => { }, { waitUntil: { - less_than: getCurrentDate().toISOString(), + less_than: nowISOString, }, }, ], @@ -173,7 +180,10 @@ export const runJobs = async (args: RunJobsArgs): Promise => { concurrencyKey: true, }, where: { - and: [{ processing: { equals: true } }, { concurrencyKey: { exists: true } }], + and: [ + { processingUntil: { greater_than: nowISOString } }, + { concurrencyKey: { exists: true } }, + ], }, }) @@ -207,8 +217,8 @@ export const runJobs = async (args: RunJobsArgs): Promise => { if (id) { const updatedDocs = await updateJobs({ data: { - processing: true, processingToken, + processingUntil, }, limit: 1, req, @@ -241,8 +251,8 @@ export const runJobs = async (args: RunJobsArgs): Promise => { } const updatedDocs = await updateJobs({ data: { - processing: true, processingToken, + processingUntil, }, limit, req, @@ -289,11 +299,21 @@ export const runJobs = async (args: RunJobsArgs): Promise => { if (jobsToRelease.length > 0) { const releaseIds = jobsToRelease.map((job) => job.id) await updateJobs({ - data: { processing: false }, + data: { processingUntil: null }, req, returning: false, where: { - and: [{ id: { in: releaseIds } }, { processingToken: { equals: processingToken } }], + and: [ + { id: { in: releaseIds } }, + { processingToken: { equals: processingToken } }, + { + processingUntil: { + greater_than: new Date( + getCurrentDate().getTime() + processingLeaseSafetyBuffer, + ).toISOString(), + }, + }, + ], }, }) } @@ -332,10 +352,13 @@ export const runJobs = async (args: RunJobsArgs): Promise => { const runSingleJob = async ( job: Job, - ): Promise<{ - id: number | string - result: RunJobResult - }> => { + ): Promise< + | { + id: number | string + result: RunJobResult + } + | null + > => { if (!job.workflowSlug && !job.taskSlug) { throw new Error('Job must have either a workflowSlug or a taskSlug') } @@ -359,33 +382,33 @@ export const runJobs = async (args: RunJobsArgs): Promise => { } } - if (!workflowConfig) { - // Permanently fail jobs whose task/workflow slug is no longer registered in config — they can never complete. - const errorMessage = `${job.taskSlug ? `Task '${job.taskSlug}'` : `Workflow '${job.workflowSlug}'`} is not registered in payload.config.jobs.` + try { + if (!workflowConfig) { + // Permanently fail jobs whose task/workflow slug is no longer registered in config — they can never complete. + const errorMessage = `${job.taskSlug ? `Task '${job.taskSlug}'` : `Workflow '${job.workflowSlug}'`} is not registered in payload.config.jobs.` - if (!silent || (typeof silent === 'object' && !silent.error)) { - payload.logger.error({ - msg: `Error running job ${job.workflowSlug || `Task: ${job.taskSlug}`} id: ${job.id} - ${errorMessage}`, - }) - } + if (!silent || (typeof silent === 'object' && !silent.error)) { + payload.logger.error({ + msg: `Error running job ${job.workflowSlug || `Task: ${job.taskSlug}`} id: ${job.id} - ${errorMessage}`, + }) + } - const updateJob = getUpdateJobFunction(job, jobReq) - await updateJob({ - error: { message: errorMessage }, - hasError: true, - processing: false, - totalTried: (job.totalTried ?? 0) + 1, - }) + const updateJob = getUpdateJobFunction(job, jobReq) + await updateJob({ + error: { message: errorMessage }, + hasError: true, + processingUntil: null, + totalTried: (job.totalTried ?? 0) + 1, + }) - return { - id: job.id, - result: { - status: 'error-reached-max-retries', - }, + return { + id: job.id, + result: { + status: 'error-reached-max-retries', + }, + } } - } - try { const updateJob = getUpdateJobFunction(job, jobReq) // the runner will either be passed to the config @@ -412,7 +435,7 @@ export const runJobs = async (args: RunJobsArgs): Promise => { error: errorMessage, }, hasError: true, - processing: false, + processingUntil: null, }) return { @@ -456,30 +479,30 @@ export const runJobs = async (args: RunJobsArgs): Promise => { return { id: job.id, result } } } catch (error) { + if (error instanceof JobRunAbortedError) { + return null + } if (error instanceof JobCancelledError) { if ( !(job.error as Record | undefined)?.cancelled || !job.hasError || - job.processing || + job.processingUntil || job.completedAt || job.waitUntil ) { // When using the local API to cancel jobs, the local API will update the job data for us to ensure the job is cancelled. // But when throwing a JobCancelledError within a task or workflow handler, we are responsible for updating the job data ourselves. + // Use the lease-protected getUpdateJobFunction so an old worker cannot cancel the job after another worker claims it. + const updateJob = getUpdateJobFunction(job, jobReq) await updateJob({ - id: job.id, - data: { - completedAt: null, - error: { - cancelled: true, - message: error.message, - }, - hasError: true, - processing: false, - waitUntil: null, + completedAt: null, + error: { + cancelled: true, + message: error.message, }, - req, - returning: false, + hasError: true, + processingUntil: null, + waitUntil: null, }) } @@ -494,20 +517,30 @@ export const runJobs = async (args: RunJobsArgs): Promise => { } } + const stopHeartbeat = startProcessingLeaseHeartbeat({ + jobIDs: jobs.map((job) => job.id), + processingLeaseDuration, + processingLeaseSafetyBuffer, + processingToken, + req, + silent, + }) + let resultsArray: { id: number | string; result: RunJobResult }[] = [] - if (sequential) { - for (const job of jobs) { - const result = await runSingleJob(job) - if (result) { - resultsArray.push(result) + try { + if (sequential) { + for (const job of jobs) { + const result = await runSingleJob(job) + if (result) { + resultsArray.push(result) + } } + } else { + const jobPromises = jobs.map(runSingleJob) + resultsArray = (await Promise.all(jobPromises)).filter((result) => result !== null) } - } else { - const jobPromises = jobs.map(runSingleJob) - resultsArray = (await Promise.all(jobPromises)) as { - id: number | string - result: RunJobResult - }[] + } finally { + stopHeartbeat() } if (jobsConfig.deleteJobOnComplete && successfullyCompletedJobs.length) { diff --git a/packages/payload/src/queues/operations/runJobs/runJSONJob/index.ts b/packages/payload/src/queues/operations/runJobs/runJSONJob/index.ts index c307880fd61..3f8af32c962 100644 --- a/packages/payload/src/queues/operations/runJobs/runJSONJob/index.ts +++ b/packages/payload/src/queues/operations/runJobs/runJSONJob/index.ts @@ -7,7 +7,11 @@ import type { UpdateJobFunction } from '../runJob/getUpdateJobFunction.js' import type { JobRunStatus } from '../runJob/index.js' import { handleWorkflowError } from '../../../errors/handleWorkflowError.js' -import { WorkflowError } from '../../../errors/index.js' +import { + JobCancelledError, + JobRunAbortedError, + WorkflowError, +} from '../../../errors/index.js' import { getCurrentDate } from '../../../utilities/getCurrentDate.js' import { getRunTaskFunction } from '../runJob/getRunTaskFunction.js' @@ -79,6 +83,10 @@ export const runJSONJob = async ({ }), ) } catch (error) { + if (error instanceof JobCancelledError || error instanceof JobRunAbortedError) { + throw error + } + const { hasFinalError } = await handleWorkflowError({ error: error instanceof WorkflowError @@ -125,7 +133,7 @@ export const runJSONJob = async ({ if (workflowCompleted) { await updateJob({ completedAt: getCurrentDate().toISOString(), - processing: false, + processingUntil: null, totalTried: (job.totalTried ?? 0) + 1, }) @@ -133,7 +141,7 @@ export const runJSONJob = async ({ status: 'success', } } else { - // Retry the job - no need to bump processing or totalTried as this does not count as a retry. A condition of a different task might have just opened up! + // Retry the job - no need to bump processingUntil or totalTried as this does not count as a retry. A condition of a different task might have just opened up! return await runJSONJob({ job, req, diff --git a/packages/payload/src/queues/operations/runJobs/runJob/getRunTaskFunction.ts b/packages/payload/src/queues/operations/runJobs/runJob/getRunTaskFunction.ts index a3b56212d8e..20f5440e275 100644 --- a/packages/payload/src/queues/operations/runJobs/runJob/getRunTaskFunction.ts +++ b/packages/payload/src/queues/operations/runJobs/runJob/getRunTaskFunction.ts @@ -20,7 +20,7 @@ import type { } from '../../../config/types/workflowTypes.js' import type { UpdateJobFunction } from './getUpdateJobFunction.js' -import { JobCancelledError, TaskError } from '../../../errors/index.js' +import { JobCancelledError, JobRunAbortedError, TaskError } from '../../../errors/index.js' import { getCurrentDate } from '../../../utilities/getCurrentDate.js' import { getTaskHandlerFromConfig } from './importHandlerPath.js' @@ -147,8 +147,8 @@ export const getRunTaskFunction = ( }) )?.output } catch (err: any) { - if (err instanceof JobCancelledError) { - // Re-throw JobCancelledError to be handled by the top-level error handler + if (err instanceof JobCancelledError || err instanceof JobRunAbortedError) { + // Job run aborts are handled by the top-level runner. throw err } throw new TaskError({ diff --git a/packages/payload/src/queues/operations/runJobs/runJob/getUpdateJobFunction.ts b/packages/payload/src/queues/operations/runJobs/runJob/getUpdateJobFunction.ts index d4980246ce9..c59c6065537 100644 --- a/packages/payload/src/queues/operations/runJobs/runJob/getUpdateJobFunction.ts +++ b/packages/payload/src/queues/operations/runJobs/runJob/getUpdateJobFunction.ts @@ -1,26 +1,49 @@ import type { Job } from '../../../../index.js' import type { PayloadRequest } from '../../../../types/index.js' -import { JobCancelledError } from '../../../errors/index.js' -import { updateJob } from '../../../utilities/updateJob.js' +import { JobCancelledError, JobRunAbortedError } from '../../../errors/index.js' +import { getCurrentDate } from '../../../utilities/getCurrentDate.js' +import { updateJobs } from '../../../utilities/updateJob.js' export type UpdateJobFunction = (jobData: Partial) => Promise /** * Helper for updating a job that does the following, additionally to updating the job: * - Merges incoming data from the updated job into the original job object - * - Handles job cancellation by throwing a `JobCancelledError` if the job was cancelled. + * - Throws a `JobCancelledError` if another process cancelled the job. + * - Stops updates after this worker loses the job's lease. */ export function getUpdateJobFunction(job: Job, req: PayloadRequest): UpdateJobFunction { return async (jobData) => { - const updatedJob = await updateJob({ - id: job.id, + const processingToken = job.processingToken + + if (!processingToken) { + throw new JobRunAbortedError(`Job ${job.id} can no longer be updated by this runner`) + } + + const minimumProcessingUntil = new Date( + getCurrentDate().getTime() + req.payload.config.jobs.processingLease.safetyBuffer, + ).toISOString() + const updatedJobs = await updateJobs({ data: jobData, + limit: 1, req, + returning: true, + where: { + and: [ + { id: { equals: job.id } }, + { processingToken: { equals: processingToken } }, + // Do not update this job if the lease expired. + // Append safety buffer, as, depending on the db adapter, the update itself + // may not be atomic, and another worker may have claimed the job before this update finishes. + { processingUntil: { greater_than: minimumProcessingUntil } }, + ], + }, }) + const updatedJob = updatedJobs?.[0] if (!updatedJob) { - return job + throw new JobRunAbortedError(`Job ${job.id} can no longer be updated by this runner`) } // Update job object like this to modify the original object - that way, incoming changes (e.g. taskStatus field that will be re-generated through the hook) will be reflected in the calling function @@ -38,7 +61,10 @@ export function getUpdateJobFunction(job: Job, req: PayloadRequest): UpdateJobFu } } - if ((updatedJob?.error as Record)?.cancelled) { + if ( + (updatedJob.error as Record)?.cancelled && + !(jobData.error as Record | undefined)?.cancelled + ) { throw new JobCancelledError(`Job ${job.id} was cancelled`) } diff --git a/packages/payload/src/queues/operations/runJobs/runJob/index.ts b/packages/payload/src/queues/operations/runJobs/runJob/index.ts index 2195c5e023c..3f8bbd2c79f 100644 --- a/packages/payload/src/queues/operations/runJobs/runJob/index.ts +++ b/packages/payload/src/queues/operations/runJobs/runJob/index.ts @@ -6,7 +6,12 @@ import type { UpdateJobFunction } from './getUpdateJobFunction.js' import { handleTaskError } from '../../../errors/handleTaskError.js' import { handleWorkflowError } from '../../../errors/handleWorkflowError.js' -import { JobCancelledError, TaskError, WorkflowError } from '../../../errors/index.js' +import { + JobCancelledError, + JobRunAbortedError, + TaskError, + WorkflowError, +} from '../../../errors/index.js' import { getCurrentDate } from '../../../utilities/getCurrentDate.js' import { getRunTaskFunction } from './getRunTaskFunction.js' @@ -50,8 +55,8 @@ export const runJob = async ({ tasks: getRunTaskFunction(job, workflowConfig, req, false, updateJob), }) } catch (error) { - if (error instanceof JobCancelledError) { - throw error // Job cancellation is handled in a top-level error handler, as higher up code may themselves throw this error + if (error instanceof JobCancelledError || error instanceof JobRunAbortedError) { + throw error // Job run aborts are handled by the top-level runner. } if (error instanceof TaskError) { const { hasFinalError } = await handleTaskError({ @@ -94,7 +99,7 @@ export const runJob = async ({ // Job log modifications are already updated at the end of the runTask function. await updateJob({ completedAt: getCurrentDate().toISOString(), - processing: false, + processingUntil: null, totalTried: (job.totalTried ?? 0) + 1, }) diff --git a/packages/payload/src/queues/utilities/updateJob.ts b/packages/payload/src/queues/utilities/updateJob.ts index 548f96bca21..4e8da873814 100644 --- a/packages/payload/src/queues/utilities/updateJob.ts +++ b/packages/payload/src/queues/utilities/updateJob.ts @@ -62,7 +62,7 @@ export async function updateJobs({ : undefined, } - if (data.processing === false) { + if (data.processingUntil === null) { data.processingToken = null } diff --git a/packages/payload/src/versions/deleteScheduledPublishJobs.ts b/packages/payload/src/versions/deleteScheduledPublishJobs.ts index 6ce4199f8c2..1267a5153bf 100644 --- a/packages/payload/src/versions/deleteScheduledPublishJobs.ts +++ b/packages/payload/src/versions/deleteScheduledPublishJobs.ts @@ -29,8 +29,8 @@ export const deleteScheduledPublishJobs = async ({ }, }, { - processing: { - equals: false, + processingUntil: { + exists: false, }, }, { diff --git a/test/queues/config.crashWorker.ts b/test/queues/config.crashWorker.ts new file mode 100644 index 00000000000..11507ab3e45 --- /dev/null +++ b/test/queues/config.crashWorker.ts @@ -0,0 +1,22 @@ +import { sqliteAdapter } from '@payloadcms/db-sqlite' + +import { buildConfigWithDefaults } from '../buildConfigWithDefaults.js' +import { getConfig } from './getConfig.js' + +const idType = + process.env.PAYLOAD_DATABASE === 'sqlite-uuid' + ? 'uuid' + : process.env.PAYLOAD_DATABASE === 'sqlite-uuidv7' + ? 'uuidv7' + : undefined + +export default buildConfigWithDefaults({ + ...getConfig(), + db: sqliteAdapter({ + client: { + url: process.env.SQLITE_URL || process.env.DATABASE_URL || 'file:./payload.db', + }, + ...(idType ? { idType } : { autoIncrement: true }), + push: false, + }), +}) diff --git a/test/queues/crashWorker.ts b/test/queues/crashWorker.ts new file mode 100644 index 00000000000..eb8e60bc0b9 --- /dev/null +++ b/test/queues/crashWorker.ts @@ -0,0 +1,29 @@ +import { _internal_jobSystemGlobals, getPayload } from 'payload' + +import config from './config.crashWorker.js' + +const jobID = JSON.parse(process.argv[2] ?? 'null') as number | string | null + +if (jobID === null) { + throw new Error('A job ID is required') +} + +_internal_jobSystemGlobals.shouldAutoRun = false +_internal_jobSystemGlobals.shouldAutoSchedule = false + +const payload = await getPayload({ config, cron: false }) + +payload.config.jobs.processingLease.duration = Number(process.env.CRASH_WORKER_LEASE_DURATION) +payload.config.jobs.processingLease.safetyBuffer = Number( + process.env.CRASH_WORKER_LEASE_SAFETY_BUFFER, +) +payload.config.jobs.deleteJobOnComplete = false + +try { + await payload.jobs.runByID({ + id: jobID, + silent: true, + }) +} finally { + await payload.destroy() +} diff --git a/test/queues/int.spec.ts b/test/queues/int.spec.ts index a01099c10a0..9586bf92e12 100644 --- a/test/queues/int.spec.ts +++ b/test/queues/int.spec.ts @@ -1,3 +1,4 @@ +import { spawn } from 'node:child_process' import path from 'path' import { _internal_jobSystemGlobals, @@ -10,10 +11,11 @@ import { } from 'payload' import { wait } from 'payload/shared' import { fileURLToPath } from 'url' -import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest' +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect } from 'vitest' import type { NextRESTClient } from '../__helpers/shared/NextRESTClient.js' +import { it } from '../__helpers/int/vitest.js' import { initPayloadInt } from '../__helpers/shared/initPayloadInt.js' import { devUser } from '../credentials.js' import { clearAndSeedEverything } from './seed.js' @@ -25,6 +27,10 @@ const dirname = path.dirname(filename) describe('Queues - Payload', () => { let payload: Payload + let processingLeaseDefaults: { + duration: number + safetyBuffer: number + } let restClient: NextRESTClient let token: string let user: User @@ -32,6 +38,7 @@ describe('Queues - Payload', () => { beforeAll(async () => { process.env.SEED_IN_CONFIG_ONINIT = 'false' // Makes it so the payload config onInit seed is not run. Otherwise, the seed would be run unnecessarily twice for the initial test run - once for beforeEach and once for onInit ;({ payload, restClient } = await initPayloadInt(dirname)) + processingLeaseDefaults = { ...payload.config.jobs.processingLease } }) afterAll(async () => { @@ -48,6 +55,7 @@ describe('Queues - Payload', () => { afterEach(() => { _internal_resetJobSystemGlobals() + Object.assign(payload.config.jobs.processingLease, processingLeaseDefaults) }) beforeEach(async () => { @@ -973,7 +981,7 @@ describe('Queues - Payload', () => { const after = await payload.findByID({ collection: 'payload-jobs', id, disableErrors: true }) expect(after?.id).toBe(id) - expect(after?.processing).toBe(false) + expect(after?.processingUntil).toBeFalsy() expect(after?.processingToken).toBeFalsy() }) @@ -993,7 +1001,7 @@ describe('Queues - Payload', () => { const after = await payload.findByID({ collection: 'payload-jobs', id, disableErrors: true }) expect(after?.id).toBe(id) - expect(after?.processing).toBe(false) + expect(after?.processingUntil).toBeFalsy() expect(after?.processingToken).toBeFalsy() }) @@ -1048,7 +1056,7 @@ describe('Queues - Payload', () => { }) expect(jobAfterRun.hasError).toBe(true) - expect(jobAfterRun.processing).toBe(false) + expect(jobAfterRun.processingUntil).toBeFalsy() expect(jobAfterRun.totalTried).toBe(1) }) @@ -1072,7 +1080,7 @@ describe('Queues - Payload', () => { }) expect(jobAfterRun.hasError).toBe(true) - expect(jobAfterRun.processing).toBe(false) + expect(jobAfterRun.processingUntil).toBeFalsy() expect(jobAfterRun.totalTried).toBe(1) }) }) @@ -1093,7 +1101,7 @@ describe('Queues - Payload', () => { }) expect(jobAfterRun.hasError).toBe(true) - expect(jobAfterRun.processing).toBe(false) + expect(jobAfterRun.processingUntil).toBeFalsy() expect(jobAfterRun.totalTried).toBe(1) }) @@ -1122,7 +1130,321 @@ describe('Queues - Payload', () => { // so the loop naturally bounds at exactly 2 attempts. expect(jobAfterRun.totalTried).toBe(2) expect(jobAfterRun.hasError).toBe(true) - expect(jobAfterRun.processing).toBe(false) + expect(jobAfterRun.processingUntil).toBeFalsy() + }) + + describe('worker recovery', () => { + // The child process can share the file-backed SQLite test database. + it( + 'should recover a job after its worker process is killed', + { db: (type) => type.startsWith('sqlite') }, + async () => { + _internal_jobSystemGlobals.shouldAutoRun = false + payload.config.jobs.deleteJobOnComplete = false + + const postTitle = 'created after a worker process crash' + const job = await payload.jobs.queue({ + input: { + postTitle, + }, + workflow: 'longRunning', + }) + + const worker = spawn( + process.execPath, + [ + '--no-deprecation', + '--import', + '@swc-node/register/esm-register', + path.resolve(dirname, 'crashWorker.ts'), + JSON.stringify(job.id), + ], + { + cwd: path.resolve(dirname, '../..'), + env: { + ...process.env, + CRASH_WORKER_LEASE_DURATION: '1000', + CRASH_WORKER_LEASE_SAFETY_BUFFER: '100', + PAYLOAD_DROP_DATABASE: 'false', + SEED_IN_CONFIG_ONINIT: 'false', + }, + stdio: ['ignore', 'ignore', 'pipe'], + }, + ) + + let workerErrorOutput = '' + + worker.stderr.on('data', (data) => { + workerErrorOutput += data.toString() + }) + + const workerExit = new Promise<{ + code: null | number + signal: NodeJS.Signals | null + }>((resolve) => { + worker.once('exit', (code, signal) => { + resolve({ code, signal }) + }) + }) + + try { + let processingUntil: null | string | undefined + + for (let attempt = 0; attempt < 200 && !processingUntil; attempt += 1) { + if (worker.exitCode !== null || worker.signalCode !== null) { + throw new Error(`Worker exited before claiming the job.\n${workerErrorOutput}`) + } + + try { + const currentJob = await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + + processingUntil = currentJob.processingUntil + } catch (error) { + const cause = error instanceof Error ? error.cause : undefined + const isDatabaseLocked = + (error instanceof Error && error.message.includes('database is locked')) || + (cause instanceof Error && cause.message.includes('database is locked')) + + if (!isDatabaseLocked) { + throw error + } + } + + await wait(50) + } + + expect(processingUntil).toBeTruthy() + expect(worker.kill('SIGKILL')).toBe(true) + expect((await workerExit).signal).toBe('SIGKILL') + + const millisecondsUntilRecovery = new Date(processingUntil!).getTime() - Date.now() + 100 + await wait(Math.max(millisecondsUntilRecovery, 0)) + + const replacementWorkerResult = await payload.jobs.runByID({ + id: job.id, + silent: true, + }) + const completedJob = await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + const createdPosts = await payload.find({ + collection: 'posts', + where: { title: { equals: postTitle } }, + }) + + expect(replacementWorkerResult.jobStatus?.[job.id]?.status).toBe('success') + expect(createdPosts.totalDocs).toBe(1) + expect(completedJob.completedAt).toBeTruthy() + expect(completedJob.hasError).toBe(false) + expect(completedJob.totalTried).toBe(1) + } finally { + if (worker.exitCode === null && worker.signalCode === null) { + worker.kill('SIGKILL') + await workerExit + } + } + }, + ) + + it('should let only one replacement worker recover a job after its original worker stops responding', async () => { + _internal_jobSystemGlobals.shouldAutoRun = false + payload.config.jobs.deleteJobOnComplete = false + // Short lease (shorter than 500ms task delay in longRunning workflow) to simulate lease expiry + payload.config.jobs.processingLease.duration = 300 + payload.config.jobs.processingLease.safetyBuffer = 250 + + const postTitle = 'created by the replacement worker' + const job = await payload.jobs.queue({ + input: { postTitle }, + queue: 'worker-recovery', + workflow: 'longRunning', + }) + const originalWorker = payload.jobs.run({ + queue: 'worker-recovery', + silent: true, + }) + + // Wait until the first worker starts, then let its short lease expire as if it stopped. + await expect + .poll( + async () => + ( + await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + ).processingToken, + ) + .toBeTruthy() + await wait(350) + // Give the replacement worker the normal lease so it can finish the long-running job. + Object.assign(payload.config.jobs.processingLease, processingLeaseDefaults) + + const replacementWorkers = await Promise.all( + Array.from({ length: 10 }, () => + payload.jobs.run({ + limit: 1, + queue: 'worker-recovery', + silent: true, + }), + ), + ) + const originalWorkerResult = await originalWorker + + const completedJob = await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + const createdPosts = await payload.find({ + collection: 'posts', + where: { title: { equals: postTitle } }, + }) + const workersThatRecoveredTheJob = replacementWorkers.filter( + (result) => result.jobStatus?.[job.id], + ) + + expect(originalWorkerResult.jobStatus?.[job.id]).toBeUndefined() + expect(workersThatRecoveredTheJob).toHaveLength(1) + expect(workersThatRecoveredTheJob[0]?.jobStatus?.[job.id]?.status).toBe('success') + expect(createdPosts.totalDocs).toBe(1) + expect(completedJob.completedAt).toBeTruthy() + expect(completedJob.hasError).toBe(false) + expect(completedJob.totalTried).toBe(1) + }) + + it('should not let another worker pick up a healthy long-running job', async () => { + _internal_jobSystemGlobals.shouldAutoRun = false + expect(payload.config.jobs.processingLease.duration).toBe(20 * 60 * 1000) + expect(payload.config.jobs.processingLease.safetyBuffer).toBe(30_000) + payload.config.jobs.deleteJobOnComplete = false + payload.config.jobs.processingLease.duration = 600 + payload.config.jobs.processingLease.safetyBuffer = 50 + + const postTitle = 'created by the healthy worker' + const job = await payload.jobs.queue({ + input: { postTitle }, + workflow: 'longRunning', + }) + const firstWorker = payload.jobs.run({ silent: true }) + + await expect + .poll( + async () => + ( + await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + ).processingToken, + ) + .toBeTruthy() + // Wait longer than the original lease. Heartbeats must keep the first worker active. + await wait(900) + + const secondWorkerResult = await payload.jobs.run({ silent: true }) + const firstWorkerResult = await firstWorker + const completedJob = await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + const createdPosts = await payload.find({ + collection: 'posts', + where: { title: { equals: postTitle } }, + }) + Object.assign(payload.config.jobs.processingLease, processingLeaseDefaults) + + expect(secondWorkerResult.jobStatus?.[job.id]).toBeUndefined() + expect(firstWorkerResult.jobStatus?.[job.id]?.status).toBe('success') + expect(createdPosts.totalDocs).toBe(1) + expect(completedJob.completedAt).toBeTruthy() + expect(completedJob.totalTried).toBe(1) + }) + + it('should recover an interrupted job without consuming a retry', async () => { + _internal_jobSystemGlobals.shouldAutoRun = false + payload.config.jobs.deleteJobOnComplete = false + payload.config.jobs.processingLease.duration = 300 + payload.config.jobs.processingLease.safetyBuffer = 250 + + const postTitle = 'created after retry-free recovery' + const job = await payload.jobs.queue({ + input: { postTitle }, + workflow: 'longRunning', + }) + + const unresponsiveWorkerResult = await payload.jobs.run({ silent: true }) + // Give the replacement worker the normal lease so it can finish the long-running job. + Object.assign(payload.config.jobs.processingLease, processingLeaseDefaults) + const replacementWorkerResult = await payload.jobs.run({ silent: true }) + + const completedJob = await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + const createdPosts = await payload.find({ + collection: 'posts', + where: { title: { equals: postTitle } }, + }) + + expect(unresponsiveWorkerResult.jobStatus?.[job.id]).toBeUndefined() + expect(replacementWorkerResult.jobStatus?.[job.id]?.status).toBe('success') + expect(createdPosts.totalDocs).toBe(1) + expect(completedJob.completedAt).toBeTruthy() + expect(completedJob.hasError).toBe(false) + expect(completedJob.totalTried).toBe(1) + }) + + it('should ignore task results from a timed-out worker after another worker recovers the job', async () => { + _internal_jobSystemGlobals.shouldAutoRun = false + payload.config.jobs.deleteJobOnComplete = false + payload.config.jobs.processingLease.duration = 300 + payload.config.jobs.processingLease.safetyBuffer = 250 + + const postTitle = 'created after ignoring stale task results' + const job = await payload.jobs.queue({ + input: { postTitle }, + workflow: 'longRunning', + }) + const timedOutWorker = payload.jobs.run({ silent: true }) + + await expect + .poll( + async () => + ( + await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + ).processingToken, + ) + .toBeTruthy() + await wait(350) + // Give the replacement worker the normal lease so it can finish the long-running job. + Object.assign(payload.config.jobs.processingLease, processingLeaseDefaults) + + const replacementWorkerResult = await payload.jobs.run({ silent: true }) + const timedOutWorkerResult = await timedOutWorker + const completedJob = await payload.findByID({ + id: job.id, + collection: 'payload-jobs', + }) + const createdPosts = await payload.find({ + collection: 'posts', + where: { title: { equals: postTitle } }, + }) + + expect(timedOutWorkerResult.jobStatus?.[job.id]).toBeUndefined() + expect(replacementWorkerResult.jobStatus?.[job.id]?.status).toBe('success') + expect(createdPosts.totalDocs).toBe(1) + expect(completedJob.log).toHaveLength(4) + expect(new Set(completedJob.log?.map(({ taskID }) => taskID)).size).toBe(4) + expect(completedJob.completedAt).toBeTruthy() + expect(completedJob.hasError).toBe(false) + }) }) it('can queue and run via the endpoint single tasks without workflows', async () => { @@ -1179,7 +1501,7 @@ describe('Queues - Payload', () => { input: { message: 'Number ' + i + ' Update 1', }, - processing: true, + processingUntil: new Date(Date.now() + 60_000).toISOString(), taskSlug: 'CreateSimple', }, }) @@ -1201,7 +1523,7 @@ describe('Queues - Payload', () => { input: { message: 'Number ' + i + ' Update 2', }, - processing: true, + processingUntil: new Date(Date.now() + 60_000).toISOString(), taskSlug: 'CreateSimple', }, }) @@ -1222,7 +1544,7 @@ describe('Queues - Payload', () => { input: { message: 'Number ' + i + ' Update 3', }, - processing: true, + processingUntil: new Date(Date.now() + 60_000).toISOString(), taskSlug: 'CreateSimple', }, }) @@ -1241,7 +1563,7 @@ describe('Queues - Payload', () => { input: { message: 'Number ' + i + ' Update 4', }, - processing: true, + processingUntil: new Date(Date.now() + 60_000).toISOString(), taskSlug: 'CreateSimple', }, }) @@ -1671,7 +1993,8 @@ describe('Queues - Payload', () => { id: job.id, depth: 0, }) - expect(jobAfterRunProcessing.processing).toBe(true) + expect(jobAfterRunProcessing.processingUntil).toBeTruthy() + expect(jobAfterRunProcessing.processingToken).toBeTruthy() // Should be in processing - cancel job await payload.jobs.cancelByID({ @@ -1694,7 +2017,8 @@ describe('Queues - Payload', () => { expect(jobAfterRun.hasError).toBe(true) // @ts-expect-error error is not typed expect(jobAfterRun.error?.cancelled).toBe(true) - expect(jobAfterRun.processing).toBe(false) + expect(jobAfterRun.processingToken).toBeFalsy() + expect(jobAfterRun.processingUntil).toBeFalsy() // Ensure job is not retried const runResponse = await payload.jobs.run({ silent: true }) @@ -1737,7 +2061,8 @@ describe('Queues - Payload', () => { expect(jobAfterRun.hasError).toBe(true) // @ts-expect-error error is not typed expect(jobAfterRun.error?.cancelled).toBe(true) - expect(jobAfterRun.processing).toBe(false) + expect(jobAfterRun.processingToken).toBeFalsy() + expect(jobAfterRun.processingUntil).toBeFalsy() }) it('ensure jobs can cancel themselves by throwing a JobCancelledError in workflow handler', async () => { @@ -1819,7 +2144,7 @@ describe('Queues - Payload', () => { expect(jobAfterRun.hasError).toBe(true) // @ts-expect-error error is not typed expect(jobAfterRun.error?.cancelled).toBe(true) - expect(jobAfterRun.processing).toBe(false) + expect(jobAfterRun.processingUntil).toBeFalsy() // Run again to ensure the job is not retried const runResponse2 = await payload.jobs.run({ silent: true }) @@ -1918,7 +2243,7 @@ describe('Queues - Payload', () => { expect(jobAfterRun.hasError).toBe(true) // @ts-expect-error error is not typed expect(jobAfterRun.error?.cancelled).toBe(true) - expect(jobAfterRun.processing).toBe(false) + expect(jobAfterRun.processingUntil).toBeFalsy() // Run again to ensure the job is not retried const runResponse2 = await payload.jobs.run({ silent: true }) @@ -2153,7 +2478,7 @@ describe('Queues - Payload', () => { expect(job1After.completedAt).toBeDefined() // Second job should still be pending (not yet run due to exclusive lock) expect(job2After.completedAt).toBeFalsy() - expect(job2After.processing).toBe(false) + expect(job2After.processingUntil).toBeFalsy() // Run jobs again - now the second job should complete await payload.jobs.run({ silent: true, limit: 10 }) @@ -2482,7 +2807,7 @@ describe('Queues - Payload', () => { }) expect(pendingJobStatus.completedAt).toBeFalsy() - expect(pendingJobStatus.processing).toBe(false) + expect(pendingJobStatus.processingUntil).toBeFalsy() // Wait for original job to complete await runPromise @@ -2611,7 +2936,7 @@ describe('Queues - Payload', () => { }) expect(runningJobAfter).not.toBeNull() - expect(runningJobAfter.processing).toBe(true) + expect(runningJobAfter.processingUntil).toBeTruthy() // The new job should exist const newJobAfter = await payload.findByID({ @@ -2747,12 +3072,12 @@ describe('Queues - Payload', () => { }) // Running job should still exist and be processing - expect(runningJobCheck.processing).toBe(true) + expect(runningJobCheck.processingUntil).toBeTruthy() // Middle job should be deleted (superseded) expect(middleJobCheck).toBeNull() // Latest job should exist and be pending expect(latestJobCheck).not.toBeNull() - expect(latestJobCheck.processing).toBe(false) + expect(latestJobCheck.processingUntil).toBeFalsy() // Wait for running job to complete await runPromise diff --git a/test/queues/payload-types.ts b/test/queues/payload-types.ts index 5c730cf1f72..ceb4e8c8342 100644 --- a/test/queues/payload-types.ts +++ b/test/queues/payload-types.ts @@ -407,7 +407,7 @@ export interface PayloadJob { | null; queue?: string | null; waitUntil?: string | null; - processing?: boolean | null; + processingUntil?: string | null; processingToken?: string | null; /** * Used for concurrency control. Jobs with the same key are subject to exclusive/supersedes rules. @@ -556,7 +556,7 @@ export interface PayloadJobsSelect { taskSlug?: T; queue?: T; waitUntil?: T; - processing?: T; + processingUntil?: T; processingToken?: T; concurrencyKey?: T; updatedAt?: T; @@ -896,7 +896,9 @@ export interface WorkflowSubTaskFails { * via the `definition` "WorkflowLongRunning". */ export interface WorkflowLongRunning { - input?: unknown; + input: { + postTitle?: string | null; + }; } /** * This interface was referenced by `Config`'s JSON-Schema @@ -1107,4 +1109,4 @@ export interface LexicalRichText { declare module 'payload' { // @ts-ignore export interface GeneratedTypes extends Config {} -} \ No newline at end of file +} diff --git a/test/queues/postgres-logs.int.spec.ts b/test/queues/postgres-logs.int.spec.ts index 827efcf769d..ec8bdd14e96 100644 --- a/test/queues/postgres-logs.int.spec.ts +++ b/test/queues/postgres-logs.int.spec.ts @@ -52,7 +52,7 @@ describePostgres('queues - postgres logs', () => { jobStatus: { '1': { status: 'success' } }, remainingJobsFromQueried: 0, }) - expect(consoleCount).toHaveBeenCalledTimes(14) // Should be 14 sql calls if the optimizations are used. If not, this would be 23 calls + expect(consoleCount).toHaveBeenCalledTimes(16) consoleCount.mockRestore() }) }) diff --git a/test/queues/workflows/longRunning.ts b/test/queues/workflows/longRunning.ts index 8d3f1d0a8e0..13b6e6016f3 100644 --- a/test/queues/workflows/longRunning.ts +++ b/test/queues/workflows/longRunning.ts @@ -5,10 +5,15 @@ import type { WorkflowConfig } from 'payload' */ export const longRunningWorkflow: WorkflowConfig<'longRunning'> = { slug: 'longRunning', - inputSchema: [], + inputSchema: [ + { + name: 'postTitle', + type: 'text', + }, + ], // Set to 4, to test that this job is not retried despite the workflow level retries being set to 4 retries: 4, - handler: async ({ inlineTask }) => { + handler: async ({ inlineTask, job, req }) => { for (let i = 0; i < 4; i += 1) { await inlineTask(String(i), { task: async () => { @@ -20,5 +25,15 @@ export const longRunningWorkflow: WorkflowConfig<'longRunning'> = { }, }) } + + if (job.input.postTitle) { + await req.payload.create({ + collection: 'posts', + data: { + title: job.input.postTitle, + }, + req, + }) + } }, }