From 7809edde3c1008bd2828b01c953088a22425644b Mon Sep 17 00:00:00 2001 From: Harminder Virk Date: Wed, 7 Oct 2026 11:44:31 +0530 Subject: [PATCH 1/8] fix(sql): keep a dedup id on a single job A dispatch with .dedup() could enqueue a second job, or return the id of a job that was never stored, when it raced with a worker acquiring the job that owned the dedup id. The unique index only covered pending and delayed jobs, so a job left it once acquired. A concurrent dispatch that had read no owner then either inserted a duplicate, or lost the insert and looked for the winner among pending and delayed jobs only. Finding none, Knex returned "added" with the id of the rolled back job and Kysely threw "Unable to resolve concurrent dedup dispatch". MySQL has no partial index, so it had no constraint at all, and its SELECT ... FOR UPDATE deadlocked with the completion of the owner. A plain unique index on (queue, dedup_id) now lets a single job own a dedup id, whatever its status, on PostgreSQL, MySQL and SQLite: jobs without dedup hold NULL, which is never a duplicate. Removing the job releases the id. The dispatch no longer opens a transaction. It reads the owner and inserts when there is none; on a unique violation it reads the owner again. The payload swap, the TTL refresh and the release of an expired id are single UPDATE statements that carry the state they rely on in their WHERE clause, so a job acquired in the meantime is not replaced and an id extended in the meantime is not released. An expired id is now released whatever the status of its job, retained ones included. On MySQL, a dispatch picked as the deadlock victim among concurrent inserts of the same id is retried. addDedupColumns() creates the unique index on existing tables. Where several jobs share a dedup id, the latest keeps it and the others lose only their dedup id. The previous indexes are dropped. Without the migration, sequential dispatches are still deduplicated, but concurrent ones are not. --- README.md | 20 +++- src/drivers/knex_adapter.ts | 109 ++++++++++---------- src/drivers/kysely_adapter.ts | 113 ++++++++++----------- src/services/knex_queue_schema.ts | 90 ++++++++++++---- src/services/kysely_queue_schema.ts | 79 +++++++++----- tests/_utils/register_driver_test_suite.ts | 84 ++++++++++++++- tests/adapter.spec.ts | 48 +++++++++ tests/kysely_adapter.spec.ts | 45 +++++++- 8 files changed, 424 insertions(+), 164 deletions(-) diff --git a/README.md b/README.md index 71eafd1..b6b1cd8 100644 --- a/README.md +++ b/README.md @@ -267,7 +267,7 @@ const { jobId, deduped } = await SaveDraftJob.dispatch({ content: '...' }) - `retryJob` does not touch the dedup entry — a retried job continues to occupy the dedup slot. TTL runs on wall-clock time, so long-running retries may outlive the TTL window. Use a generous TTL or no TTL if retries must stay deduped. - Atomicity: - **Redis**: a single Lua script per dispatch performs the dedup-key lookup, state check (pending/delayed ZSCORE), payload swap, and TTL refresh atomically. - - **Knex/Kysely**: transactional `SELECT ... FOR UPDATE` + insert/update inside a transaction. On PostgreSQL and SQLite, a partial unique index makes concurrent first dispatches race-free: a savepoint catches the unique-constraint violation and returns `{ deduped: 'skipped' }` pointing at the winner. MySQL has no partial unique index, see the caveat below. + - **Knex/Kysely**: a unique index on `(queue, dedup_id)` lets a single job own a dedup id, whatever its status. Concurrent first dispatches race on that index: one inserts the job, the others read it and return `{ deduped: 'skipped' }` pointing at the winner. Payload swaps, TTL refreshes and the release of an expired id are single conditional `UPDATE` statements, so no transaction or row lock is held. Works the same on PostgreSQL, MySQL and SQLite. - **SyncAdapter**: executes inline, no dedup support. ### Caveats @@ -278,7 +278,6 @@ const { jobId, deduped } = await SaveDraftJob.dispatch({ content: '...' }) - Scheduled jobs (`.schedule()`) do not support dedup — each cron/interval fire is an independent dispatch. - With no `ttl`, dedup persists until the job is removed (completed/failed without retention). When retention keeps the record, re-dispatch stays blocked until the record is pruned. - With `ttl`, dedup expires after the window — a new job (new UUID) is created. The old job still runs. -- Knex/Kysely MySQL concurrent race: MySQL does not support partial unique indexes, so two `pushOn` calls with the same dedup id firing at the exact same instant can both succeed. Serialize at the app layer if strict guarantees are required, or use Postgres / SQLite / Redis (all of which serialize correctly via the partial unique index or Lua atomicity). ## Job History & Retention @@ -473,6 +472,23 @@ await schema.dropJobsTable() +#### Migrating SQL deduplication after an upgrade + +Jobs tables created by earlier versions have no unique index on `(queue, dedup_id)`, and dispatches racing +with a worker could enqueue a job twice. Run `addDedupColumns()` once from a migration to create it: + +```typescript +// Knex +await new KnexQueueSchemaService(connection).addDedupColumns('queue_jobs') + +// Kysely +await new KyselyQueueSchemaService(db, { dialect: 'postgres' }).addDedupColumns('queue_jobs') +``` + +The migration is idempotent. When several jobs share a dedup id, the latest one keeps it and the +others only lose their dedup id: no job is removed. The previous dedup indexes are dropped. Stop +dispatching jobs with `.dedup()` while it runs. + #### Migrating SQL schedules after an upgrade The schedules table stores its dates as epoch milliseconds (`bigint`), so they do not depend on the diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index 6b59e08..35f83bf 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -471,6 +471,12 @@ export class KnexAdapter implements Adapter { }) } + /** + * A single job owns a dedup slot, whatever its status: the unique + * (queue, dedup_id) index rejects a second owner. No lock is held between + * the statements. Each write carries the state it relies on in its WHERE + * clause, and a dispatch that loses the slot looks at its owner again. + */ async #pushWithDedup( queue: string, jobData: JobData, @@ -479,7 +485,13 @@ export class KnexAdapter implements Adapter { const dedup = jobData.dedup! try { - return await this.#pushWithDedupTxn(queue, jobData, insertRow, dedup) + while (true) { + const now = Date.now() + const result = + (await this.#resolveExistingDedup(queue, jobData, dedup, now)) ?? + (await this.#insertDedup(queue, jobData.id, insertRow, dedup, now)) + if (result) return result + } } catch (err) { if (this.#isMissingDedupColumn(err)) { throw new Error( @@ -503,37 +515,25 @@ export class KnexAdapter implements Adapter { ) } - #pushWithDedupTxn( - queue: string, - jobData: JobData, - insertRow: Record, - dedup: NonNullable - ): Promise { - return this.#connection.transaction(async (trx) => { - const now = Date.now() - const existingResult = await this.#resolveExistingDedup(trx, queue, jobData, dedup, now) - if (existingResult) return existingResult - - return this.#insertDedup(trx, queue, jobData.id, insertRow, dedup, now) - }) - } - + /** + * Returns null when the slot has no owner, or when its owner expired. + */ async #resolveExistingDedup( - trx: Knex.Transaction, queue: string, jobData: JobData, dedup: NonNullable, now: number ): Promise { - const existing = await trx(this.#jobsTable) + const existing = await this.#connection(this.#jobsTable) .where('queue', queue) .where('dedup_id', dedup.id) .orderBy('dedup_at', 'desc') - .forUpdate() .first() if (!existing) return null + const owner = () => + this.#connection(this.#jobsTable).where({ id: existing.id, queue, dedup_id: dedup.id }) const dedupAt = existing.dedup_at != null ? Number(existing.dedup_at) : null const dedupTtl = existing.dedup_ttl != null ? Number(existing.dedup_ttl) : null const withinTtl = dedupTtl === null || (dedupAt !== null && now - dedupAt < dedupTtl) @@ -550,65 +550,53 @@ export class KnexAdapter implements Adapter { if (dedup.extend && dedupTtl) { updates.dedup_at = now } - await trx(this.#jobsTable).where({ id: existing.id, queue }).update(updates) - return { outcome: 'replaced' as DedupOutcome, jobId: existing.id as string } + // Not replaced when a worker acquired the job in the meantime + const replaced = await owner().whereIn('status', ['pending', 'delayed']).update(updates) + if (replaced > 0) { + return { outcome: 'replaced' as DedupOutcome, jobId: existing.id as string } + } } if (dedup.extend && dedupTtl) { - await trx(this.#jobsTable).where({ id: existing.id, queue }).update({ dedup_at: now }) + await owner().update({ dedup_at: now }) return { outcome: 'extended' as DedupOutcome, jobId: existing.id as string } } return { outcome: 'skipped' as DedupOutcome, jobId: existing.id as string } } - // Release the expired dedup slot. The old job keeps running to completion. - const status = existing.status as JobStatus - if (status === 'pending' || status === 'delayed' || status === 'active') { - await trx(this.#jobsTable) - .where({ id: existing.id, queue }) - .update({ dedup_id: null, dedup_at: null, dedup_ttl: null }) - } + // Release the expired dedup slot, unless it was extended in the meantime. + // The old job keeps running to completion. + await owner() + .where('dedup_at', existing.dedup_at) + .update({ dedup_id: null, dedup_at: null, dedup_ttl: null }) return null } + /** + * Returns null when a concurrent dispatch took the slot first. + */ async #insertDedup( - trx: Knex.Transaction, queue: string, jobId: string, insertRow: Record, dedup: NonNullable, now: number - ): Promise { - let raceLost = false + ): Promise { try { - await trx.transaction(async (sp) => { - await sp(this.#jobsTable).insert({ - ...insertRow, - dedup_id: dedup.id, - dedup_at: now, - dedup_ttl: dedup.ttl ?? null, - }) + await this.#connection(this.#jobsTable).insert({ + ...insertRow, + dedup_id: dedup.id, + dedup_at: now, + dedup_ttl: dedup.ttl ?? null, }) } catch (err) { - if (this.#isUniqueViolation(err)) { - raceLost = true - } else { - throw err - } - } - - if (raceLost) { - const winner = await trx(this.#jobsTable) - .where('queue', queue) - .where('dedup_id', dedup.id) - .whereIn('status', ['pending', 'delayed']) - .orderBy('dedup_at', 'desc') - .first() - if (winner) { - return { outcome: 'skipped' as DedupOutcome, jobId: winner.id as string } - } + if (this.#isDeadlock(err)) return null + if (!this.#isUniqueViolation(err)) throw err + // The job id itself is the duplicate + if (await this.getJob(jobId, queue)) throw err + return null } return { outcome: 'added' as DedupOutcome, jobId } @@ -620,10 +608,19 @@ export class KnexAdapter implements Adapter { return ( e.code === '23505' || e.code === 'SQLITE_CONSTRAINT_UNIQUE' || + e.code === 'ER_DUP_ENTRY' || /UNIQUE constraint/i.test(e.message ?? '') ) } + /** + * InnoDB picks a victim among dispatches inserting the same dedup id right + * after its owner was removed. + */ + #isDeadlock(err: unknown): boolean { + return (err as { code?: string } | null)?.code === 'ER_LOCK_DEADLOCK' + } + async pushMany(jobs: JobData[]): Promise { return this.pushManyOn('default', jobs) } diff --git a/src/drivers/kysely_adapter.ts b/src/drivers/kysely_adapter.ts index e808c93..8406c4b 100644 --- a/src/drivers/kysely_adapter.ts +++ b/src/drivers/kysely_adapter.ts @@ -465,21 +465,28 @@ export class KyselyAdapter implements Adapter { await this.#jobs(this.#connection).insertInto(this.#jobsTable).values(row).execute() } + /** + * A single job owns a dedup slot, whatever its status: the unique + * (queue, dedup_id) index rejects a second owner. No lock is held between + * the statements. Each write carries the state it relies on in its WHERE + * clause, and a dispatch that loses the slot looks at its owner again. + */ async #pushWithDedup( connection: Kysely, queue: string, jobData: JobData, insertRow: Partial & Pick ): Promise { + const dedup = jobData.dedup! + try { - return await this.#withTransaction(connection, async (trx) => { + while (true) { const now = Date.now() - const dedup = jobData.dedup! - const existingResult = await this.#resolveExistingDedup(trx, queue, jobData, dedup, now) - if (existingResult) return existingResult - - return this.#insertDedup(trx, queue, jobData.id, insertRow, dedup, now) - }) + const result = + (await this.#resolveExistingDedup(connection, queue, jobData, dedup, now)) ?? + (await this.#insertDedup(connection, queue, jobData.id, insertRow, dedup, now)) + if (result) return result + } } catch (error) { if (this.#isMissingDedupColumn(error)) { throw new Error( @@ -491,26 +498,32 @@ export class KyselyAdapter implements Adapter { } } + /** + * Returns null when the slot has no owner, or when its owner expired. + */ async #resolveExistingDedup( - trx: Transaction, + connection: Kysely, queue: string, jobData: JobData, dedup: NonNullable, now: number ): Promise { - let existingQuery = this.#jobs(trx) + const existing = await this.#jobs(connection) .selectFrom(this.#jobsTable) .selectAll() .where('queue', '=', queue) .where('dedup_id', '=', dedup.id) .orderBy('dedup_at', 'desc') .limit(1) + .executeTakeFirst() - if (this.#supportsSkipLocked()) existingQuery = existingQuery.forUpdate() - - const existing = await existingQuery.executeTakeFirst() if (!existing) return null + const owner = this.#jobs(connection) + .updateTable(this.#jobsTable) + .where('id', '=', existing.id) + .where('queue', '=', queue) + .where('dedup_id', '=', dedup.id) const dedupAt = existing.dedup_at == null ? null : Number(existing.dedup_at) const dedupTtl = existing.dedup_ttl == null ? null : Number(existing.dedup_ttl) const withinTtl = dedupTtl === null || (dedupAt !== null && now - dedupAt < dedupTtl) @@ -525,57 +538,45 @@ export class KyselyAdapter implements Adapter { } if (dedup.extend && dedupTtl) updates.dedup_at = now - await this.#jobs(trx) - .updateTable(this.#jobsTable) + // Not replaced when a worker acquired the job in the meantime + const replaced = await owner .set(updates) - .where('id', '=', existing.id) - .where('queue', '=', queue) - .execute() - return { outcome: 'replaced', jobId: existing.id } + .where('status', 'in', ['pending', 'delayed']) + .executeTakeFirst() + if (replaced.numUpdatedRows > 0n) return { outcome: 'replaced', jobId: existing.id } } if (dedup.extend && dedupTtl) { - await this.#jobs(trx) - .updateTable(this.#jobsTable) - .set({ dedup_at: now }) - .where('id', '=', existing.id) - .where('queue', '=', queue) - .execute() + await owner.set({ dedup_at: now }).execute() return { outcome: 'extended', jobId: existing.id } } return { outcome: 'skipped', jobId: existing.id } } - if ( - existing.status === 'pending' || - existing.status === 'delayed' || - existing.status === 'active' - ) { - await this.#jobs(trx) - .updateTable(this.#jobsTable) - .set({ dedup_id: null, dedup_at: null, dedup_ttl: null }) - .where('id', '=', existing.id) - .where('queue', '=', queue) - .execute() - } + // Release the expired dedup slot, unless it was extended in the meantime. + // The old job keeps running to completion. + await owner + .set({ dedup_id: null, dedup_at: null, dedup_ttl: null }) + .where('dedup_at', existing.dedup_at == null ? 'is' : '=', existing.dedup_at ?? null) + .execute() return null } + /** + * Returns null when a concurrent dispatch took the slot first. + */ async #insertDedup( - trx: Transaction, + connection: Kysely, queue: string, jobId: string, insertRow: Partial & Pick, dedup: NonNullable, now: number - ): Promise { - const savepoint = `queue_dedup_${randomUUID().replaceAll('-', '')}` - await sql`savepoint ${sql.id(savepoint)}`.execute(trx) - + ): Promise { try { - await this.#jobs(trx) + await this.#jobs(connection) .insertInto(this.#jobsTable) .values({ ...insertRow, @@ -584,25 +585,15 @@ export class KyselyAdapter implements Adapter { dedup_ttl: dedup.ttl ?? null, }) .execute() - await sql`release savepoint ${sql.id(savepoint)}`.execute(trx) - return { outcome: 'added', jobId } } catch (error) { - await sql`rollback to savepoint ${sql.id(savepoint)}`.execute(trx) - await sql`release savepoint ${sql.id(savepoint)}`.execute(trx) + if (this.#isDeadlock(error)) return null if (!this.#isUniqueViolation(error)) throw error + // The job id itself is the duplicate + if (await this.getJob(jobId, queue)) throw error + return null } - const winner = await this.#jobs(trx) - .selectFrom(this.#jobsTable) - .select('id') - .where('queue', '=', queue) - .where('dedup_id', '=', dedup.id) - .where('status', 'in', ['pending', 'delayed']) - .orderBy('dedup_at', 'desc') - .executeTakeFirst() - - if (!winner) throw new Error(`Unable to resolve concurrent dedup dispatch for "${dedup.id}"`) - return { outcome: 'skipped', jobId: winner.id } + return { outcome: 'added', jobId } } async pushMany(jobs: JobData[]): Promise { @@ -962,6 +953,14 @@ export class KyselyAdapter implements Adapter { ) } + /** + * InnoDB picks a victim among dispatches inserting the same dedup id right + * after its owner was removed. + */ + #isDeadlock(error: unknown): boolean { + return (error as { code?: string } | null)?.code === 'ER_LOCK_DEADLOCK' + } + #isMissingDedupColumn(error: unknown): boolean { if (!error || typeof error !== 'object') return false const message = (error as { message?: string }).message ?? '' diff --git a/src/services/knex_queue_schema.ts b/src/services/knex_queue_schema.ts index 5a69c33..00e0fd2 100644 --- a/src/services/knex_queue_schema.ts +++ b/src/services/knex_queue_schema.ts @@ -51,20 +51,22 @@ export class KnexQueueSchemaService { table.index(['queue', 'status', 'score']) table.index(['queue', 'status', 'execute_at']) table.index(['queue', 'status', 'finished_at']) - table.index(['queue', 'dedup_id']) extend?.(table) }) - await this.#createDedupActiveUniqueIndex(tableName) + await this.#createDedupUniqueIndex(tableName) } /** * Idempotent migration: adds dedup columns (dedup_id, dedup_at, dedup_ttl) - * and a (queue, dedup_id) index to an existing jobs table. + * and a unique (queue, dedup_id) index to an existing jobs table. * - * Safe to run multiple times. Uses hasColumn checks so it won't fail on re-runs. - * For large Postgres tables, consider pausing workers during the run. + * Safe to run multiple times. Run it again when upgrading from a version + * that created a non-unique index: jobs sharing a dedup id keep running, and + * only the latest one keeps the dedup slot. + * + * Stop dispatching jobs with .dedup() during the run. */ async addDedupColumns(tableName: string = 'queue_jobs'): Promise { const hasDedupId = await this.#connection.schema.hasColumn(tableName, 'dedup_id') @@ -79,30 +81,74 @@ export class KnexQueueSchemaService { }) } - if (!hasDedupId) { - await this.#connection.schema.alterTable(tableName, (table) => { - table.index(['queue', 'dedup_id']) - }) + // The derived table is what allows MySQL to read the table it updates. + await this.#connection.raw( + `UPDATE ?? SET dedup_id = NULL, dedup_at = NULL, dedup_ttl = NULL + WHERE (id, queue) IN ( + SELECT id, queue FROM ( + SELECT DISTINCT older.id, older.queue + FROM ?? AS older + JOIN ?? AS newer ON newer.queue = older.queue AND newer.dedup_id = older.dedup_id + WHERE newer.dedup_at > older.dedup_at + OR (newer.dedup_at = older.dedup_at AND newer.id > older.id) + ) AS stale + )`, + [tableName, tableName, tableName] + ) + + await this.#createDedupUniqueIndex(tableName) + await this.#dropLegacyDedupIndexes(tableName) + } + + /** + * Unique index on (queue, dedup_id): a single job owns a dedup slot, whatever + * its status. Jobs without dedup hold NULL, which is never a duplicate. + */ + async #createDedupUniqueIndex(tableName: string): Promise { + const indexName = `${tableName}_dedup_uidx` + + if (this.#dialect() !== 'mysql') { + await this.#connection.raw('CREATE UNIQUE INDEX IF NOT EXISTS ?? ON ?? (??, ??)', [ + indexName, + tableName, + 'queue', + 'dedup_id', + ]) + return } - await this.#createDedupActiveUniqueIndex(tableName) + // MySQL has no CREATE INDEX IF NOT EXISTS + try { + await this.#connection.raw('CREATE UNIQUE INDEX ?? ON ?? (??, ??)', [ + indexName, + tableName, + 'queue', + 'dedup_id', + ]) + } catch (error) { + if ((error as { code?: string }).code !== 'ER_DUP_KEYNAME') throw error + } } /** - * Partial unique index on (queue, dedup_id) for active dedup slots. - * Prevents two concurrent inserts with the same dedup_id from both succeeding. - * Only PG and SQLite support partial unique indexes; MySQL is skipped. + * Drops the indexes the unique index replaces: the non-unique one, and the + * partial unique one PostgreSQL and SQLite had on pending and delayed jobs. */ - async #createDedupActiveUniqueIndex(tableName: string): Promise { - const client = this.#connection.client.config.client - if (client !== 'pg' && client !== 'better-sqlite3' && client !== 'sqlite3') return + async #dropLegacyDedupIndexes(tableName: string): Promise { + if (this.#dialect() !== 'mysql') { + await this.#connection.raw('DROP INDEX IF EXISTS ??', [`${tableName}_queue_dedup_id_index`]) + await this.#connection.raw('DROP INDEX IF EXISTS ??', [`${tableName}_dedup_active_uidx`]) + return + } - const indexName = `${tableName}_dedup_active_uidx` - await this.#connection.raw( - `CREATE UNIQUE INDEX IF NOT EXISTS ?? ON ?? ("queue", "dedup_id") ` + - `WHERE "dedup_id" IS NOT NULL AND "status" IN ('pending', 'delayed')`, - [indexName, tableName] - ) + try { + await this.#connection.raw('DROP INDEX ?? ON ??', [ + `${tableName}_queue_dedup_id_index`, + tableName, + ]) + } catch (error) { + if ((error as { code?: string }).code !== 'ER_CANT_DROP_FIELD_OR_KEY') throw error + } } /** diff --git a/src/services/kysely_queue_schema.ts b/src/services/kysely_queue_schema.ts index d93b952..75e41b2 100644 --- a/src/services/kysely_queue_schema.ts +++ b/src/services/kysely_queue_schema.ts @@ -57,7 +57,7 @@ export class KyselyQueueSchemaService { .execute() await this.#createJobsIndexes(tableName) - await this.#createDedupActiveUniqueIndex(tableName) + await this.#createDedupUniqueIndex(tableName) } async addDedupColumns(tableName: string = 'queue_jobs'): Promise { @@ -84,21 +84,23 @@ export class KyselyQueueSchemaService { await this.#connection.schema.alterTable(tableName).addColumn('dedup_ttl', 'bigint').execute() } - const index = this.#connection.schema - .createIndex(`${tableName}_queue_dedup_idx`) - .on(tableName) - .columns(['queue', 'dedup_id']) + // The derived table is what allows MySQL to read the table it updates. + await sql` + update ${sql.table(tableName)} set dedup_id = null, dedup_at = null, dedup_ttl = null + where (id, queue) in ( + select id, queue from ( + select distinct older.id, older.queue + from ${sql.table(tableName)} as older + join ${sql.table(tableName)} as newer + on newer.queue = older.queue and newer.dedup_id = older.dedup_id + where newer.dedup_at > older.dedup_at + or (newer.dedup_at = older.dedup_at and newer.id > older.id) + ) as stale + ) + `.execute(this.#connection) - if (this.#dialect === 'mysql') { - try { - await index.execute() - } catch (error) { - if (!this.#isDuplicateIndexError(error)) throw error - } - } else { - await index.ifNotExists().execute() - } - await this.#createDedupActiveUniqueIndex(tableName) + await this.#createDedupUniqueIndex(tableName) + await this.#dropLegacyDedupIndexes(tableName) } async createSchedulesTable(tableName: string = 'queue_schedules'): Promise { @@ -506,7 +508,6 @@ export class KyselyQueueSchemaService { ['status_score', ['queue', 'status', 'score']], ['status_execute', ['queue', 'status', 'execute_at']], ['status_finished', ['queue', 'status', 'finished_at']], - ['queue_dedup', ['queue', 'dedup_id']], ] for (const [suffix, columns] of indexes) { @@ -518,17 +519,47 @@ export class KyselyQueueSchemaService { } } - async #createDedupActiveUniqueIndex(tableName: string): Promise { - if (this.#dialect === 'mysql') return - - await this.#connection.schema - .createIndex(`${tableName}_dedup_active_uidx`) - .ifNotExists() + /** + * Unique index on (queue, dedup_id): a single job owns a dedup slot, whatever + * its status. Jobs without dedup hold NULL, which is never a duplicate. + */ + async #createDedupUniqueIndex(tableName: string): Promise { + const index = this.#connection.schema + .createIndex(`${tableName}_dedup_uidx`) .unique() .on(tableName) .columns(['queue', 'dedup_id']) - .where(sql`dedup_id is not null and status in ('pending', 'delayed')`) - .execute() + + // MySQL has no CREATE INDEX IF NOT EXISTS + if (this.#dialect === 'mysql') { + try { + await index.execute() + } catch (error) { + if (!this.#isDuplicateIndexError(error)) throw error + } + } else { + await index.ifNotExists().execute() + } + } + + /** + * Drops the indexes the unique index replaces: the non-unique one, and the + * partial unique one PostgreSQL and SQLite had on pending and delayed jobs. + */ + async #dropLegacyDedupIndexes(tableName: string): Promise { + const legacyIndex = this.#connection.schema.dropIndex(`${tableName}_queue_dedup_idx`) + + if (this.#dialect === 'mysql') { + try { + await legacyIndex.on(tableName).execute() + } catch (error) { + if ((error as { code?: string }).code !== 'ER_CANT_DROP_FIELD_OR_KEY') throw error + } + return + } + + await legacyIndex.ifExists().execute() + await this.#connection.schema.dropIndex(`${tableName}_dedup_active_uidx`).ifExists().execute() } /** LONGTEXT on MySQL, whose TEXT stops at 64 KB; TEXT elsewhere. */ diff --git a/tests/_utils/register_driver_test_suite.ts b/tests/_utils/register_driver_test_suite.ts index 8bfdb4b..a803271 100644 --- a/tests/_utils/register_driver_test_suite.ts +++ b/tests/_utils/register_driver_test_suite.ts @@ -1,5 +1,5 @@ import { test as JapaTest } from '@japa/runner' -import type { AcquiredJob, Adapter, JobLease } from '../../src/contracts/adapter.js' +import type { AcquiredJob, Adapter, JobLease, PushResult } from '../../src/contracts/adapter.js' interface DriverTestSuiteOptions { test: typeof JapaTest @@ -12,7 +12,7 @@ interface DriverTestSuiteOptions { supportsConcurrency?: boolean /** * Whether concurrent dispatches have an atomic dedup constraint. - * MySQL has no partial unique indexes, matching the documented Knex limitation. + * Some in-memory adapters do not serialize concurrent deduplicated dispatches. * @default true */ supportsAtomicDedup?: boolean @@ -3089,6 +3089,86 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) { }) if (options.supportsAtomicDedup !== false) { + test('dedup: concurrent dispatches stay deduplicated while jobs are acquired', async ({ + assert, + }) => { + const adapter = await options.createAdapter() + adapter.setWorkerId('worker-1') + + let dispatching = true + const acquiring = (async () => { + while (dispatching) await adapter.popFrom('concurrent-dedup-queue') + })() + + const results = await Promise.all( + Array.from({ length: 4 }, async (_, dispatcher) => { + const dispatches = [] + for (let index = 0; index < 500; index++) { + dispatches.push( + await adapter.pushOn('concurrent-dedup-queue', { + id: `${dispatcher}-${index}`, + name: 'TestJob', + payload: {}, + attempts: 0, + dedup: { id: String(index % 200), ttl: 60_000 }, + }) + ) + } + return dispatches + }) + ).finally(() => (dispatching = false)) + await acquiring + + const dispatches = results.flat().map((result) => result as PushResult) + assert.lengthOf( + dispatches.filter(({ outcome }) => outcome === 'added'), + 200 + ) + + // Every dispatch points at a job that was stored + for (const jobId of new Set(dispatches.map((result) => result.jobId))) { + assert.isNotNull(await adapter.getJob(jobId, 'concurrent-dedup-queue')) + } + }).timeout(20_000) + + test('dedup: concurrent dispatches do not fail while jobs are completed', async ({ + assert, + }) => { + const adapter = await options.createAdapter() + adapter.setWorkerId('worker-1') + + let dispatching = true + let completed = 0 + const workers = Array.from({ length: 2 }, async () => { + while (dispatching || (await adapter.sizeOf('concurrent-dedup-queue')) > 0) { + const job = await adapter.popFrom('concurrent-dedup-queue') + if (job && (await adapter.completeJob(job, 'concurrent-dedup-queue'))) completed++ + } + }) + + const results = await Promise.all( + Array.from({ length: 4 }, async (_, dispatcher) => { + const dispatches = [] + for (let index = 0; index < 200; index++) { + dispatches.push( + await adapter.pushOn('concurrent-dedup-queue', { + id: `${dispatcher}-${index}`, + name: 'TestJob', + payload: {}, + attempts: 0, + dedup: { id: String(index % 3) }, + }) + ) + } + return dispatches + }) + ).finally(() => (dispatching = false)) + await Promise.all(workers) + + const added = results.flat().filter((result) => (result as PushResult).outcome === 'added') + assert.equal(completed, added.length) + }).timeout(20_000) + test('dedup: concurrent pushOn with same id - only one wins, rest skipped', async ({ assert, }) => { diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 8b27877..e72f37d 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -3,6 +3,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import Knex from 'knex' import { test } from '@japa/runner' +import type { Assert } from '@japa/assert' import { Redis } from 'ioredis' import { CronExpressionParser } from 'cron-parser' import { MemoryAdapter } from './_mocks/memory_adapter.js' @@ -1970,6 +1971,45 @@ test.group('Adapter | Redis', (group) => { }) }) +/** + * A jobs table as the previous versions left it: no unique index, and expired + * owners that still hold their dedup id. + */ +async function assertDedupMigration( + assert: Assert, + connection: ReturnType, + tableName: string +) { + const schemaService = new KnexQueueSchemaService(connection) + const job = (id: string, dedupId: string, dedupAt: number) => ({ + id, + queue: 'default', + status: 'completed', + data: '{}', + dedup_id: dedupId, + dedup_at: dedupAt, + dedup_ttl: 10, + }) + + await connection.raw('DROP INDEX ??', [`${tableName}_dedup_uidx`]) + await connection.schema.alterTable(tableName, (table) => table.index(['queue', 'dedup_id'])) + await connection(tableName).insert([ + job('older', 'shared', 1_000), + job('newer', 'shared', 2_000), + job('alone', 'alone', 1_000), + ]) + + await schemaService.addDedupColumns(tableName) + await schemaService.addDedupColumns(tableName) + + assert.deepEqual(await connection(tableName).select('id', 'dedup_id').orderBy('id'), [ + { id: 'alone', dedup_id: 'alone' }, + { id: 'newer', dedup_id: 'shared' }, + { id: 'older', dedup_id: null }, + ]) + await assert.rejects(() => connection(tableName).insert(job('duplicate', 'shared', 3_000))) +} + test.group('Adapter | Knex (SQLite)', (group) => { let connection: ReturnType let adapter: KnexAdapter @@ -2003,6 +2043,10 @@ test.group('Adapter | Knex (SQLite)', (group) => { }, }) + test('addDedupColumns should leave a dedup slot to its latest job', async ({ assert }) => { + await assertDedupMigration(assert, connection, 'queue_jobs') + }) + test('listSchedules should execute a single SQL query in Knex adapter', async ({ assert }) => { const knexAdapter = new KnexAdapter({ connection }) @@ -2110,6 +2154,10 @@ test.group('Adapter | Knex (PostgreSQL)', (group) => { }, }) + test('addDedupColumns should leave a dedup slot to its latest job', async ({ assert }) => { + await assertDedupMigration(assert, connection, tableName) + }) + test('listSchedules should execute a single SQL query in Knex PostgreSQL adapter', async ({ assert, }) => { diff --git a/tests/kysely_adapter.spec.ts b/tests/kysely_adapter.spec.ts index 51f827b..df1674f 100644 --- a/tests/kysely_adapter.spec.ts +++ b/tests/kysely_adapter.spec.ts @@ -10,6 +10,7 @@ import { KyselyAdapter, KyselyQueueSchemaService, type QueueDatabase, + type QueueJobTable, } from '../src/drivers/kysely_adapter.js' import { registerDriverTestSuite } from './_utils/register_driver_test_suite.js' @@ -174,7 +175,6 @@ test.group('Adapter | Kysely (MySQL)', (group) => { registerDriverTestSuite({ test, - supportsAtomicDedup: false, createAdapter: () => { adapter = new KyselyAdapter({ connection, @@ -190,6 +190,49 @@ test.group('Adapter | Kysely (MySQL)', (group) => { await schema.addDedupColumns(tableName) await schema.addDedupColumns(tableName) }) + + test('addDedupColumns should leave a dedup slot to its latest job', async ({ assert }) => { + const jobs = connection.withTables>() + const job = (id: string, dedupId: string, dedupAt: number) => ({ + id, + queue: 'default', + status: 'completed' as const, + data: '{}', + dedup_id: dedupId, + dedup_at: dedupAt, + dedup_ttl: 10, + }) + + // A jobs table as the previous versions left it: no unique index, and + // expired owners that still hold their dedup id. + await connection.schema.dropIndex(`${tableName}_dedup_uidx`).on(tableName).execute() + await connection.schema + .createIndex(`${tableName}_queue_dedup_idx`) + .on(tableName) + .columns(['queue', 'dedup_id']) + .execute() + await jobs + .insertInto(tableName) + .values([job('older', 'shared', 1_000), job('newer', 'shared', 2_000)]) + .execute() + + await schema.addDedupColumns(tableName) + await schema.addDedupColumns(tableName) + + assert.deepEqual( + await jobs.selectFrom(tableName).select(['id', 'dedup_id']).orderBy('id').execute(), + [ + { id: 'newer', dedup_id: 'shared' }, + { id: 'older', dedup_id: null }, + ] + ) + await assert.rejects(() => + jobs + .insertInto(tableName) + .values(job('duplicate', 'shared', 3_000)) + .execute() + ) + }) }) test.group('KyselyAdapter | SQLite file, two connections', () => { From c7b6c920f38c1405b5d6b4169f95c8ca4f8882df Mon Sep 17 00:00:00 2001 From: Harminder Virk Date: Wed, 7 Oct 2026 12:12:54 +0530 Subject: [PATCH 2/8] fix(sql): lock the owner of a dedup id during a dispatch The dispatch read the owner of a dedup id without a lock, and guarded each update with the state it had read. A change between the read and the update could still be lost: a replace rewrites the whole data column, so a job retried in that window got its attempts counter set back. The dispatch runs in a transaction again, and locks the owner with SELECT ... FOR UPDATE before it skips, extends, replaces or releases it. The updates are plain again. The owner is found through the dedup index without a lock, then locked through its primary key, as a worker locks a job. Locking it through the dedup index takes the index entry then the row, while the removal of a completed job takes the row then the index entry, and the two deadlocked on MySQL. FOR UPDATE cannot lock an owner that does not exist yet, so concurrent first dispatches still race on the unique index. The one that loses the insert runs its transaction again and locks the winner. The same applies to a dispatch MySQL picks as a deadlock victim among concurrent inserts of the same id. --- README.md | 2 +- src/drivers/knex_adapter.ts | 111 +++++++++++++-------------- src/drivers/kysely_adapter.ts | 140 ++++++++++++++++------------------ 3 files changed, 120 insertions(+), 133 deletions(-) diff --git a/README.md b/README.md index b6b1cd8..856408f 100644 --- a/README.md +++ b/README.md @@ -267,7 +267,7 @@ const { jobId, deduped } = await SaveDraftJob.dispatch({ content: '...' }) - `retryJob` does not touch the dedup entry — a retried job continues to occupy the dedup slot. TTL runs on wall-clock time, so long-running retries may outlive the TTL window. Use a generous TTL or no TTL if retries must stay deduped. - Atomicity: - **Redis**: a single Lua script per dispatch performs the dedup-key lookup, state check (pending/delayed ZSCORE), payload swap, and TTL refresh atomically. - - **Knex/Kysely**: a unique index on `(queue, dedup_id)` lets a single job own a dedup id, whatever its status. Concurrent first dispatches race on that index: one inserts the job, the others read it and return `{ deduped: 'skipped' }` pointing at the winner. Payload swaps, TTL refreshes and the release of an expired id are single conditional `UPDATE` statements, so no transaction or row lock is held. Works the same on PostgreSQL, MySQL and SQLite. + - **Knex/Kysely**: a unique index on `(queue, dedup_id)` lets a single job own a dedup id, whatever its status. Each dispatch runs in a transaction: an existing owner is locked with `SELECT ... FOR UPDATE`, then skipped, extended or replaced under that lock. When there is no owner to lock, concurrent first dispatches race on the unique index: one inserts the job, the others run again, lock the winner and return `{ deduped: 'skipped' }` pointing at it. Works the same on PostgreSQL, MySQL and SQLite. - **SyncAdapter**: executes inline, no dedup support. ### Caveats diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index 35f83bf..c3bdb95 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -471,12 +471,6 @@ export class KnexAdapter implements Adapter { }) } - /** - * A single job owns a dedup slot, whatever its status: the unique - * (queue, dedup_id) index rejects a second owner. No lock is held between - * the statements. Each write carries the state it relies on in its WHERE - * clause, and a dispatch that loses the slot looks at its owner again. - */ async #pushWithDedup( queue: string, jobData: JobData, @@ -485,13 +479,7 @@ export class KnexAdapter implements Adapter { const dedup = jobData.dedup! try { - while (true) { - const now = Date.now() - const result = - (await this.#resolveExistingDedup(queue, jobData, dedup, now)) ?? - (await this.#insertDedup(queue, jobData.id, insertRow, dedup, now)) - if (result) return result - } + return await this.#pushWithDedupTxn(queue, jobData, insertRow, dedup) } catch (err) { if (this.#isMissingDedupColumn(err)) { throw new Error( @@ -516,24 +504,62 @@ export class KnexAdapter implements Adapter { } /** - * Returns null when the slot has no owner, or when its owner expired. + * A single job owns a dedup slot, whatever its status: the unique + * (queue, dedup_id) index rejects a second owner. An existing owner is + * locked for the transaction. When there is none to lock and a concurrent + * dispatch inserts first, the transaction runs again and locks that owner. */ + async #pushWithDedupTxn( + queue: string, + jobData: JobData, + insertRow: Record, + dedup: NonNullable + ): Promise { + while (true) { + try { + return await this.#connection.transaction(async (trx) => { + const now = Date.now() + const existingResult = await this.#resolveExistingDedup(trx, queue, jobData, dedup, now) + if (existingResult) return existingResult + + await trx(this.#jobsTable).insert({ + ...insertRow, + dedup_id: dedup.id, + dedup_at: now, + dedup_ttl: dedup.ttl ?? null, + }) + return { outcome: 'added' as DedupOutcome, jobId: jobData.id } + }) + } catch (err) { + if (!this.#isUniqueViolation(err) && !this.#isDeadlock(err)) throw err + // The job id itself is the duplicate + if (await this.getJob(jobData.id, queue)) throw err + } + } + } + async #resolveExistingDedup( + trx: Knex.Transaction, queue: string, jobData: JobData, dedup: NonNullable, now: number ): Promise { - const existing = await this.#connection(this.#jobsTable) + const owner = await trx(this.#jobsTable) .where('queue', queue) .where('dedup_id', dedup.id) .orderBy('dedup_at', 'desc') - .first() + .first('id') + + if (!owner) return null + + // The owner is locked through its primary key, as workers lock a job. + // Locking it through the dedup index deadlocks with them on MySQL. + const existing = await trx(this.#jobsTable).where({ id: owner.id, queue }).forUpdate().first() - if (!existing) return null + // Removed or released in the meantime: the insert tells whether the slot is free + if (existing?.dedup_id !== dedup.id) return null - const owner = () => - this.#connection(this.#jobsTable).where({ id: existing.id, queue, dedup_id: dedup.id }) const dedupAt = existing.dedup_at != null ? Number(existing.dedup_at) : null const dedupTtl = existing.dedup_ttl != null ? Number(existing.dedup_ttl) : null const withinTtl = dedupTtl === null || (dedupAt !== null && now - dedupAt < dedupTtl) @@ -550,58 +576,26 @@ export class KnexAdapter implements Adapter { if (dedup.extend && dedupTtl) { updates.dedup_at = now } - // Not replaced when a worker acquired the job in the meantime - const replaced = await owner().whereIn('status', ['pending', 'delayed']).update(updates) - if (replaced > 0) { - return { outcome: 'replaced' as DedupOutcome, jobId: existing.id as string } - } + await trx(this.#jobsTable).where({ id: existing.id, queue }).update(updates) + return { outcome: 'replaced' as DedupOutcome, jobId: existing.id as string } } if (dedup.extend && dedupTtl) { - await owner().update({ dedup_at: now }) + await trx(this.#jobsTable).where({ id: existing.id, queue }).update({ dedup_at: now }) return { outcome: 'extended' as DedupOutcome, jobId: existing.id as string } } return { outcome: 'skipped' as DedupOutcome, jobId: existing.id as string } } - // Release the expired dedup slot, unless it was extended in the meantime. - // The old job keeps running to completion. - await owner() - .where('dedup_at', existing.dedup_at) + // Release the expired dedup slot. The old job keeps running to completion. + await trx(this.#jobsTable) + .where({ id: existing.id, queue }) .update({ dedup_id: null, dedup_at: null, dedup_ttl: null }) return null } - /** - * Returns null when a concurrent dispatch took the slot first. - */ - async #insertDedup( - queue: string, - jobId: string, - insertRow: Record, - dedup: NonNullable, - now: number - ): Promise { - try { - await this.#connection(this.#jobsTable).insert({ - ...insertRow, - dedup_id: dedup.id, - dedup_at: now, - dedup_ttl: dedup.ttl ?? null, - }) - } catch (err) { - if (this.#isDeadlock(err)) return null - if (!this.#isUniqueViolation(err)) throw err - // The job id itself is the duplicate - if (await this.getJob(jobId, queue)) throw err - return null - } - - return { outcome: 'added' as DedupOutcome, jobId } - } - #isUniqueViolation(err: unknown): boolean { if (!err || typeof err !== 'object') return false const e = err as { code?: string; message?: string } @@ -614,8 +608,7 @@ export class KnexAdapter implements Adapter { } /** - * InnoDB picks a victim among dispatches inserting the same dedup id right - * after its owner was removed. + * InnoDB picks a victim among dispatches inserting the same dedup id. */ #isDeadlock(err: unknown): boolean { return (err as { code?: string } | null)?.code === 'ER_LOCK_DEADLOCK' diff --git a/src/drivers/kysely_adapter.ts b/src/drivers/kysely_adapter.ts index 8406c4b..80d4d92 100644 --- a/src/drivers/kysely_adapter.ts +++ b/src/drivers/kysely_adapter.ts @@ -467,9 +467,9 @@ export class KyselyAdapter implements Adapter { /** * A single job owns a dedup slot, whatever its status: the unique - * (queue, dedup_id) index rejects a second owner. No lock is held between - * the statements. Each write carries the state it relies on in its WHERE - * clause, and a dispatch that loses the slot looks at its owner again. + * (queue, dedup_id) index rejects a second owner. An existing owner is + * locked for the transaction. When there is none to lock and a concurrent + * dispatch inserts first, the transaction runs again and locks that owner. */ async #pushWithDedup( connection: Kysely, @@ -479,51 +479,71 @@ export class KyselyAdapter implements Adapter { ): Promise { const dedup = jobData.dedup! - try { - while (true) { - const now = Date.now() - const result = - (await this.#resolveExistingDedup(connection, queue, jobData, dedup, now)) ?? - (await this.#insertDedup(connection, queue, jobData.id, insertRow, dedup, now)) - if (result) return result - } - } catch (error) { - if (this.#isMissingDedupColumn(error)) { - throw new Error( - `Dedup columns missing on "${this.#jobsTable}". Run KyselyQueueSchemaService.addDedupColumns() before dispatching jobs with .dedup().`, - { cause: error } - ) + while (true) { + try { + return await this.#withTransaction(connection, async (trx) => { + const now = Date.now() + const existingResult = await this.#resolveExistingDedup(trx, queue, jobData, dedup, now) + if (existingResult) return existingResult + + await this.#jobs(trx) + .insertInto(this.#jobsTable) + .values({ + ...insertRow, + dedup_id: dedup.id, + dedup_at: now, + dedup_ttl: dedup.ttl ?? null, + }) + .execute() + return { outcome: 'added', jobId: jobData.id } + }) + } catch (error) { + if (this.#isMissingDedupColumn(error)) { + throw new Error( + `Dedup columns missing on "${this.#jobsTable}". Run KyselyQueueSchemaService.addDedupColumns() before dispatching jobs with .dedup().`, + { cause: error } + ) + } + if (!this.#isUniqueViolation(error) && !this.#isDeadlock(error)) throw error + // The job id itself is the duplicate + if (await this.getJob(jobData.id, queue)) throw error } - throw error } } - /** - * Returns null when the slot has no owner, or when its owner expired. - */ async #resolveExistingDedup( - connection: Kysely, + trx: Transaction, queue: string, jobData: JobData, dedup: NonNullable, now: number ): Promise { - const existing = await this.#jobs(connection) + const owner = await this.#jobs(trx) .selectFrom(this.#jobsTable) - .selectAll() + .select('id') .where('queue', '=', queue) .where('dedup_id', '=', dedup.id) .orderBy('dedup_at', 'desc') .limit(1) .executeTakeFirst() - if (!existing) return null + if (!owner) return null - const owner = this.#jobs(connection) - .updateTable(this.#jobsTable) - .where('id', '=', existing.id) + // The owner is locked through its primary key, as workers lock a job. + // Locking it through the dedup index deadlocks with them on MySQL. + let existingQuery = this.#jobs(trx) + .selectFrom(this.#jobsTable) + .selectAll() + .where('id', '=', owner.id) .where('queue', '=', queue) - .where('dedup_id', '=', dedup.id) + + if (this.#supportsSkipLocked()) existingQuery = existingQuery.forUpdate() + + const existing = await existingQuery.executeTakeFirst() + + // Removed or released in the meantime: the insert tells whether the slot is free + if (existing?.dedup_id !== dedup.id) return null + const dedupAt = existing.dedup_at == null ? null : Number(existing.dedup_at) const dedupTtl = existing.dedup_ttl == null ? null : Number(existing.dedup_ttl) const withinTtl = dedupTtl === null || (dedupAt !== null && now - dedupAt < dedupTtl) @@ -538,64 +558,39 @@ export class KyselyAdapter implements Adapter { } if (dedup.extend && dedupTtl) updates.dedup_at = now - // Not replaced when a worker acquired the job in the meantime - const replaced = await owner + await this.#jobs(trx) + .updateTable(this.#jobsTable) .set(updates) - .where('status', 'in', ['pending', 'delayed']) - .executeTakeFirst() - if (replaced.numUpdatedRows > 0n) return { outcome: 'replaced', jobId: existing.id } + .where('id', '=', existing.id) + .where('queue', '=', queue) + .execute() + return { outcome: 'replaced', jobId: existing.id } } if (dedup.extend && dedupTtl) { - await owner.set({ dedup_at: now }).execute() + await this.#jobs(trx) + .updateTable(this.#jobsTable) + .set({ dedup_at: now }) + .where('id', '=', existing.id) + .where('queue', '=', queue) + .execute() return { outcome: 'extended', jobId: existing.id } } return { outcome: 'skipped', jobId: existing.id } } - // Release the expired dedup slot, unless it was extended in the meantime. - // The old job keeps running to completion. - await owner + // Release the expired dedup slot. The old job keeps running to completion. + await this.#jobs(trx) + .updateTable(this.#jobsTable) .set({ dedup_id: null, dedup_at: null, dedup_ttl: null }) - .where('dedup_at', existing.dedup_at == null ? 'is' : '=', existing.dedup_at ?? null) + .where('id', '=', existing.id) + .where('queue', '=', queue) .execute() return null } - /** - * Returns null when a concurrent dispatch took the slot first. - */ - async #insertDedup( - connection: Kysely, - queue: string, - jobId: string, - insertRow: Partial & Pick, - dedup: NonNullable, - now: number - ): Promise { - try { - await this.#jobs(connection) - .insertInto(this.#jobsTable) - .values({ - ...insertRow, - dedup_id: dedup.id, - dedup_at: now, - dedup_ttl: dedup.ttl ?? null, - }) - .execute() - } catch (error) { - if (this.#isDeadlock(error)) return null - if (!this.#isUniqueViolation(error)) throw error - // The job id itself is the duplicate - if (await this.getJob(jobId, queue)) throw error - return null - } - - return { outcome: 'added', jobId } - } - async pushMany(jobs: JobData[]): Promise { return this.pushManyOn('default', jobs) } @@ -954,8 +949,7 @@ export class KyselyAdapter implements Adapter { } /** - * InnoDB picks a victim among dispatches inserting the same dedup id right - * after its owner was removed. + * InnoDB picks a victim among dispatches inserting the same dedup id. */ #isDeadlock(error: unknown): boolean { return (error as { code?: string } | null)?.code === 'ER_LOCK_DEADLOCK' From a9c43a8238e35c2f1bb9269c51bf8859c9fccd76 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 07:12:34 +0000 Subject: [PATCH 3/8] fix(sql): respect dedup column collation Case-insensitive SQL lookups can match a dedup key whose spelling differs from the input. Comparing that stored key with the input rejected the owner and retried a conflicting insert forever. Compare the locked key with the value read by the initial lookup instead, keeping the primary-key locking query unchanged. Cover both adapters with case-insensitive SQLite dispatch tests. --- src/drivers/knex_adapter.ts | 7 ++++--- src/drivers/kysely_adapter.ts | 7 ++++--- tests/adapter.spec.ts | 29 +++++++++++++++++++++++++++++ tests/kysely_adapter.spec.ts | 31 ++++++++++++++++++++++++++++++- 4 files changed, 67 insertions(+), 7 deletions(-) diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index c3bdb95..f158a91 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -549,7 +549,7 @@ export class KnexAdapter implements Adapter { .where('queue', queue) .where('dedup_id', dedup.id) .orderBy('dedup_at', 'desc') - .first('id') + .first('id', 'dedup_id') if (!owner) return null @@ -557,8 +557,9 @@ export class KnexAdapter implements Adapter { // Locking it through the dedup index deadlocks with them on MySQL. const existing = await trx(this.#jobsTable).where({ id: owner.id, queue }).forUpdate().first() - // Removed or released in the meantime: the insert tells whether the slot is free - if (existing?.dedup_id !== dedup.id) return null + // Compare stored values: SQL collation may equate different input strings. + // Removed or released in the meantime: the insert tells whether the slot is free. + if (!existing || existing.dedup_id !== owner.dedup_id) return null const dedupAt = existing.dedup_at != null ? Number(existing.dedup_at) : null const dedupTtl = existing.dedup_ttl != null ? Number(existing.dedup_ttl) : null diff --git a/src/drivers/kysely_adapter.ts b/src/drivers/kysely_adapter.ts index 80d4d92..4c1f23a 100644 --- a/src/drivers/kysely_adapter.ts +++ b/src/drivers/kysely_adapter.ts @@ -520,7 +520,7 @@ export class KyselyAdapter implements Adapter { ): Promise { const owner = await this.#jobs(trx) .selectFrom(this.#jobsTable) - .select('id') + .select(['id', 'dedup_id']) .where('queue', '=', queue) .where('dedup_id', '=', dedup.id) .orderBy('dedup_at', 'desc') @@ -541,8 +541,9 @@ export class KyselyAdapter implements Adapter { const existing = await existingQuery.executeTakeFirst() - // Removed or released in the meantime: the insert tells whether the slot is free - if (existing?.dedup_id !== dedup.id) return null + // Compare stored values: SQL collation may equate different input strings. + // Removed or released in the meantime: the insert tells whether the slot is free. + if (!existing || existing.dedup_id !== owner.dedup_id) return null const dedupAt = existing.dedup_at == null ? null : Number(existing.dedup_at) const dedupTtl = existing.dedup_ttl == null ? null : Number(existing.dedup_ttl) diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index e72f37d..f9da185 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -2047,6 +2047,35 @@ test.group('Adapter | Knex (SQLite)', (group) => { await assertDedupMigration(assert, connection, 'queue_jobs') }) + test('dedup should respect a case-insensitive column collation', async ({ assert }) => { + await connection.raw('DROP INDEX queue_jobs_dedup_uidx') + await connection.raw('ALTER TABLE queue_jobs DROP COLUMN dedup_id') + await connection.raw('ALTER TABLE queue_jobs ADD COLUMN dedup_id TEXT COLLATE NOCASE') + await connection.raw( + 'CREATE UNIQUE INDEX queue_jobs_dedup_uidx ON queue_jobs (queue, dedup_id)' + ) + adapter = new KnexAdapter({ connection }) + + const owner = await adapter.pushOn('default', { + id: 'owner', + name: 'TestJob', + payload: {}, + attempts: 0, + dedup: { id: 'Order-A' }, + }) + const duplicate = await adapter.pushOn('default', { + id: 'duplicate', + name: 'TestJob', + payload: {}, + attempts: 0, + dedup: { id: 'order-a' }, + }) + + assert.deepEqual(owner, { outcome: 'added', jobId: 'owner' }) + assert.deepEqual(duplicate, { outcome: 'skipped', jobId: 'owner' }) + assert.equal(await adapter.sizeOf('default'), 1) + }) + test('listSchedules should execute a single SQL query in Knex adapter', async ({ assert }) => { const knexAdapter = new KnexAdapter({ connection }) diff --git a/tests/kysely_adapter.spec.ts b/tests/kysely_adapter.spec.ts index df1674f..e48436b 100644 --- a/tests/kysely_adapter.spec.ts +++ b/tests/kysely_adapter.spec.ts @@ -2,7 +2,7 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import Database from 'better-sqlite3' -import { Kysely, MysqlDialect, PostgresDialect, SqliteDialect } from 'kysely' +import { Kysely, MysqlDialect, PostgresDialect, SqliteDialect, sql } from 'kysely' import { createPool } from 'mysql2' import { Pool } from 'pg' import { test } from '@japa/runner' @@ -41,6 +41,35 @@ test.group('Adapter | Kysely (SQLite)', (group) => { }, }) + test('dedup should respect a case-insensitive column collation', async ({ assert }) => { + await sql`DROP INDEX queue_jobs_dedup_uidx`.execute(connection) + await sql`ALTER TABLE queue_jobs DROP COLUMN dedup_id`.execute(connection) + await sql`ALTER TABLE queue_jobs ADD COLUMN dedup_id TEXT COLLATE NOCASE`.execute(connection) + await sql`CREATE UNIQUE INDEX queue_jobs_dedup_uidx ON queue_jobs (queue, dedup_id)`.execute( + connection + ) + adapter = new KyselyAdapter({ connection, dialect: 'sqlite' }) + + const owner = await adapter.pushOn('default', { + id: 'owner', + name: 'TestJob', + payload: {}, + attempts: 0, + dedup: { id: 'Order-A' }, + }) + const duplicate = await adapter.pushOn('default', { + id: 'duplicate', + name: 'TestJob', + payload: {}, + attempts: 0, + dedup: { id: 'order-a' }, + }) + + assert.deepEqual(owner, { outcome: 'added', jobId: 'owner' }) + assert.deepEqual(duplicate, { outcome: 'skipped', jobId: 'owner' }) + assert.equal(await adapter.sizeOf('default'), 1) + }) + test('concurrent workers should not acquire the same job', async ({ assert }) => { const firstWorker = new KyselyAdapter({ connection, dialect: 'sqlite' }) const secondWorker = new KyselyAdapter({ connection, dialect: 'sqlite' }) From 3557dfa8ed822464a4e1709933dbdce025e2d6a2 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 07:21:53 +0000 Subject: [PATCH 4/8] fix(sql): bound the retries of a dedup dispatch A dispatch with dedup ran its transaction again after any unique violation or deadlock, with no limit. Its only exit was finding the owner of the dedup id, so a violation that a retry never resolves, such as the collation mismatch fixed in the previous commit or a future unique index on the jobs table, made the dispatch loop forever. The transaction now runs at most DEDUP_DISPATCH_ATTEMPTS times, then the dispatch fails with the last error as its cause. Losing the insert to a concurrent dispatch takes one extra pass, so five leaves room for an owner that changes in between. Also cover the guard that rethrows when the job id itself is the duplicate: the test expects the primary key violation, not the error of the bounded retries. --- src/constants.ts | 7 +++++++ src/drivers/knex_adapter.ts | 10 ++++++++-- src/drivers/kysely_adapter.ts | 10 ++++++++-- tests/adapter.spec.ts | 14 ++++++++++++++ tests/kysely_adapter.spec.ts | 14 ++++++++++++++ 5 files changed, 51 insertions(+), 4 deletions(-) diff --git a/src/constants.ts b/src/constants.ts index c92b373..1967c52 100644 --- a/src/constants.ts +++ b/src/constants.ts @@ -42,3 +42,10 @@ export const UNKNOWN_JOB_RETRY_DELAY = 30_000 * Default delay before retrying after an error */ export const DEFAULT_ERROR_RETRY_DELAY = '5s' + +/** + * Number of times a dispatch with dedup runs its transaction before it fails. + * A dispatch runs again when it loses the insert of a dedup id to a concurrent + * one, so a second pass is enough unless the owner keeps changing. + */ +export const DEDUP_DISPATCH_ATTEMPTS = 5 diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index f158a91..3ad69cd 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -18,7 +18,7 @@ import type { ScheduleData, ScheduleListOptions, } from '../types/main.js' -import { DEFAULT_PRIORITY } from '../constants.js' +import { DEDUP_DISPATCH_ATTEMPTS, DEFAULT_PRIORITY } from '../constants.js' import { calculateScore, createLeaseToken, @@ -515,7 +515,7 @@ export class KnexAdapter implements Adapter { insertRow: Record, dedup: NonNullable ): Promise { - while (true) { + for (let attempt = 1; ; attempt++) { try { return await this.#connection.transaction(async (trx) => { const now = Date.now() @@ -534,6 +534,12 @@ export class KnexAdapter implements Adapter { if (!this.#isUniqueViolation(err) && !this.#isDeadlock(err)) throw err // The job id itself is the duplicate if (await this.getJob(jobData.id, queue)) throw err + if (attempt === DEDUP_DISPATCH_ATTEMPTS) { + throw new Error( + `Unable to dispatch job "${jobData.id}" with dedup id "${dedup.id}" after ${attempt} attempts`, + { cause: err } + ) + } } } } diff --git a/src/drivers/kysely_adapter.ts b/src/drivers/kysely_adapter.ts index 4c1f23a..d034684 100644 --- a/src/drivers/kysely_adapter.ts +++ b/src/drivers/kysely_adapter.ts @@ -16,7 +16,7 @@ import type { ScheduleData, ScheduleListOptions, } from '../types/main.js' -import { DEFAULT_PRIORITY } from '../constants.js' +import { DEDUP_DISPATCH_ATTEMPTS, DEFAULT_PRIORITY } from '../constants.js' import { calculateScore, createLeaseToken, @@ -479,7 +479,7 @@ export class KyselyAdapter implements Adapter { ): Promise { const dedup = jobData.dedup! - while (true) { + for (let attempt = 1; ; attempt++) { try { return await this.#withTransaction(connection, async (trx) => { const now = Date.now() @@ -507,6 +507,12 @@ export class KyselyAdapter implements Adapter { if (!this.#isUniqueViolation(error) && !this.#isDeadlock(error)) throw error // The job id itself is the duplicate if (await this.getJob(jobData.id, queue)) throw error + if (attempt === DEDUP_DISPATCH_ATTEMPTS) { + throw new Error( + `Unable to dispatch job "${jobData.id}" with dedup id "${dedup.id}" after ${attempt} attempts`, + { cause: error } + ) + } } } } diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index f9da185..254c7b1 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -2047,6 +2047,20 @@ test.group('Adapter | Knex (SQLite)', (group) => { await assertDedupMigration(assert, connection, 'queue_jobs') }) + test('dedup should reject a job id that is already stored', async ({ assert }) => { + adapter = new KnexAdapter({ connection }) + const job = { id: 'same-id', name: 'TestJob', payload: {}, attempts: 0 } + + await adapter.pushOn('default', { ...job, dedup: { id: 'first' } }) + // The primary key violation itself, not a retry of the dispatch + const error = await adapter.pushOn('default', { ...job, dedup: { id: 'second' } }).then( + () => null, + (rejection: { code?: string }) => rejection + ) + assert.equal(error?.code, 'SQLITE_CONSTRAINT_PRIMARYKEY') + assert.equal(await adapter.sizeOf('default'), 1) + }) + test('dedup should respect a case-insensitive column collation', async ({ assert }) => { await connection.raw('DROP INDEX queue_jobs_dedup_uidx') await connection.raw('ALTER TABLE queue_jobs DROP COLUMN dedup_id') diff --git a/tests/kysely_adapter.spec.ts b/tests/kysely_adapter.spec.ts index e48436b..96e53a0 100644 --- a/tests/kysely_adapter.spec.ts +++ b/tests/kysely_adapter.spec.ts @@ -41,6 +41,20 @@ test.group('Adapter | Kysely (SQLite)', (group) => { }, }) + test('dedup should reject a job id that is already stored', async ({ assert }) => { + adapter = new KyselyAdapter({ connection, dialect: 'sqlite' }) + const job = { id: 'same-id', name: 'TestJob', payload: {}, attempts: 0 } + + await adapter.pushOn('default', { ...job, dedup: { id: 'first' } }) + // The primary key violation itself, not a retry of the dispatch + const error = await adapter.pushOn('default', { ...job, dedup: { id: 'second' } }).then( + () => null, + (rejection: { code?: string }) => rejection + ) + assert.equal(error?.code, 'SQLITE_CONSTRAINT_PRIMARYKEY') + assert.equal(await adapter.sizeOf('default'), 1) + }) + test('dedup should respect a case-insensitive column collation', async ({ assert }) => { await sql`DROP INDEX queue_jobs_dedup_uidx`.execute(connection) await sql`ALTER TABLE queue_jobs DROP COLUMN dedup_id`.execute(connection) From e9c8cbb272a4e2b7837ca6358d53d072473845d7 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 07:21:54 +0000 Subject: [PATCH 5/8] docs: describe the outcome of a dispatch that loses a dedup race A dispatch that loses the insert of a dedup id runs again and treats the winner as an existing owner, so replace and extend apply to it. The README said it always returned skipped. --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 856408f..2853358 100644 --- a/README.md +++ b/README.md @@ -267,7 +267,7 @@ const { jobId, deduped } = await SaveDraftJob.dispatch({ content: '...' }) - `retryJob` does not touch the dedup entry — a retried job continues to occupy the dedup slot. TTL runs on wall-clock time, so long-running retries may outlive the TTL window. Use a generous TTL or no TTL if retries must stay deduped. - Atomicity: - **Redis**: a single Lua script per dispatch performs the dedup-key lookup, state check (pending/delayed ZSCORE), payload swap, and TTL refresh atomically. - - **Knex/Kysely**: a unique index on `(queue, dedup_id)` lets a single job own a dedup id, whatever its status. Each dispatch runs in a transaction: an existing owner is locked with `SELECT ... FOR UPDATE`, then skipped, extended or replaced under that lock. When there is no owner to lock, concurrent first dispatches race on the unique index: one inserts the job, the others run again, lock the winner and return `{ deduped: 'skipped' }` pointing at it. Works the same on PostgreSQL, MySQL and SQLite. + - **Knex/Kysely**: a unique index on `(queue, dedup_id)` lets a single job own a dedup id, whatever its status. Each dispatch runs in a transaction: an existing owner is locked with `SELECT ... FOR UPDATE`, then skipped, extended or replaced under that lock. When there is no owner to lock, concurrent first dispatches race on the unique index: one inserts the job, the others run again and lock the winner, which they then skip, extend or replace like any existing owner. Works the same on PostgreSQL, MySQL and SQLite. - **SyncAdapter**: executes inline, no dedup support. ### Caveats From 7e4a40c15724785d71fc0b14dd8521ecd8ec6dc8 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 07:21:54 +0000 Subject: [PATCH 6/8] test: ignore popFrom deadlocks in the concurrent dedup tests On MySQL, the workers of these tests can be picked as deadlock victims by concurrent acquisitions, a popFrom issue unrelated to deduplication, which failed the Kysely MySQL group about one run in ten. A deadlocked poll now counts as an empty one, so the tests only check dedup. --- tests/_utils/register_driver_test_suite.ts | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/tests/_utils/register_driver_test_suite.ts b/tests/_utils/register_driver_test_suite.ts index a803271..9d6702c 100644 --- a/tests/_utils/register_driver_test_suite.ts +++ b/tests/_utils/register_driver_test_suite.ts @@ -18,6 +18,20 @@ interface DriverTestSuiteOptions { supportsAtomicDedup?: boolean } +/** + * Acquire a job, treating a MySQL deadlock victim as an empty poll. popFrom + * can deadlock with concurrent acquisitions on MySQL, which these dedup tests + * do not cover. + */ +async function popIgnoringDeadlock(adapter: Adapter, queue: string): Promise { + try { + return await adapter.popFrom(queue) + } catch (error) { + if ((error as { code?: string } | null)?.code === 'ER_LOCK_DEADLOCK') return null + throw error + } +} + /** A lease on a job that was never acquired. */ function unleased(id: string): JobLease { return { id, leaseToken: 'worker-1:never-acquired' } @@ -3097,7 +3111,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) { let dispatching = true const acquiring = (async () => { - while (dispatching) await adapter.popFrom('concurrent-dedup-queue') + while (dispatching) await popIgnoringDeadlock(adapter, 'concurrent-dedup-queue') })() const results = await Promise.all( @@ -3141,7 +3155,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) { let completed = 0 const workers = Array.from({ length: 2 }, async () => { while (dispatching || (await adapter.sizeOf('concurrent-dedup-queue')) > 0) { - const job = await adapter.popFrom('concurrent-dedup-queue') + const job = await popIgnoringDeadlock(adapter, 'concurrent-dedup-queue') if (job && (await adapter.completeJob(job, 'concurrent-dedup-queue'))) completed++ } }) From 998f64fcb85cec37126e606b74290addc5a51a74 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 07:38:17 +0000 Subject: [PATCH 7/8] fix(sql): let a dedup dispatch run again until it resolves Bounding the retries to five attempts made dispatches fail under contention on MySQL: with workers completing the owners of a few dedup ids while several dispatchers insert them, a dispatch can lose the insert more than five times in a row, each time to a dispatch that succeeded. The CI run of the concurrent dedup test failed that way. Each extra pass follows a change made by another transaction, so the loop no longer has a bound. The test of a duplicate job id still expects the primary key violation. --- src/constants.ts | 7 ------- src/drivers/knex_adapter.ts | 10 ++-------- src/drivers/kysely_adapter.ts | 10 ++-------- tests/adapter.spec.ts | 2 +- tests/kysely_adapter.spec.ts | 2 +- 5 files changed, 6 insertions(+), 25 deletions(-) diff --git a/src/constants.ts b/src/constants.ts index 1967c52..c92b373 100644 --- a/src/constants.ts +++ b/src/constants.ts @@ -42,10 +42,3 @@ export const UNKNOWN_JOB_RETRY_DELAY = 30_000 * Default delay before retrying after an error */ export const DEFAULT_ERROR_RETRY_DELAY = '5s' - -/** - * Number of times a dispatch with dedup runs its transaction before it fails. - * A dispatch runs again when it loses the insert of a dedup id to a concurrent - * one, so a second pass is enough unless the owner keeps changing. - */ -export const DEDUP_DISPATCH_ATTEMPTS = 5 diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index 3ad69cd..f158a91 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -18,7 +18,7 @@ import type { ScheduleData, ScheduleListOptions, } from '../types/main.js' -import { DEDUP_DISPATCH_ATTEMPTS, DEFAULT_PRIORITY } from '../constants.js' +import { DEFAULT_PRIORITY } from '../constants.js' import { calculateScore, createLeaseToken, @@ -515,7 +515,7 @@ export class KnexAdapter implements Adapter { insertRow: Record, dedup: NonNullable ): Promise { - for (let attempt = 1; ; attempt++) { + while (true) { try { return await this.#connection.transaction(async (trx) => { const now = Date.now() @@ -534,12 +534,6 @@ export class KnexAdapter implements Adapter { if (!this.#isUniqueViolation(err) && !this.#isDeadlock(err)) throw err // The job id itself is the duplicate if (await this.getJob(jobData.id, queue)) throw err - if (attempt === DEDUP_DISPATCH_ATTEMPTS) { - throw new Error( - `Unable to dispatch job "${jobData.id}" with dedup id "${dedup.id}" after ${attempt} attempts`, - { cause: err } - ) - } } } } diff --git a/src/drivers/kysely_adapter.ts b/src/drivers/kysely_adapter.ts index d034684..4c1f23a 100644 --- a/src/drivers/kysely_adapter.ts +++ b/src/drivers/kysely_adapter.ts @@ -16,7 +16,7 @@ import type { ScheduleData, ScheduleListOptions, } from '../types/main.js' -import { DEDUP_DISPATCH_ATTEMPTS, DEFAULT_PRIORITY } from '../constants.js' +import { DEFAULT_PRIORITY } from '../constants.js' import { calculateScore, createLeaseToken, @@ -479,7 +479,7 @@ export class KyselyAdapter implements Adapter { ): Promise { const dedup = jobData.dedup! - for (let attempt = 1; ; attempt++) { + while (true) { try { return await this.#withTransaction(connection, async (trx) => { const now = Date.now() @@ -507,12 +507,6 @@ export class KyselyAdapter implements Adapter { if (!this.#isUniqueViolation(error) && !this.#isDeadlock(error)) throw error // The job id itself is the duplicate if (await this.getJob(jobData.id, queue)) throw error - if (attempt === DEDUP_DISPATCH_ATTEMPTS) { - throw new Error( - `Unable to dispatch job "${jobData.id}" with dedup id "${dedup.id}" after ${attempt} attempts`, - { cause: error } - ) - } } } } diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 254c7b1..8fae70a 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -2052,7 +2052,7 @@ test.group('Adapter | Knex (SQLite)', (group) => { const job = { id: 'same-id', name: 'TestJob', payload: {}, attempts: 0 } await adapter.pushOn('default', { ...job, dedup: { id: 'first' } }) - // The primary key violation itself, not a retry of the dispatch + // The primary key violation itself, rather than a dispatch that runs again const error = await adapter.pushOn('default', { ...job, dedup: { id: 'second' } }).then( () => null, (rejection: { code?: string }) => rejection diff --git a/tests/kysely_adapter.spec.ts b/tests/kysely_adapter.spec.ts index 96e53a0..621beaf 100644 --- a/tests/kysely_adapter.spec.ts +++ b/tests/kysely_adapter.spec.ts @@ -46,7 +46,7 @@ test.group('Adapter | Kysely (SQLite)', (group) => { const job = { id: 'same-id', name: 'TestJob', payload: {}, attempts: 0 } await adapter.pushOn('default', { ...job, dedup: { id: 'first' } }) - // The primary key violation itself, not a retry of the dispatch + // The primary key violation itself, rather than a dispatch that runs again const error = await adapter.pushOn('default', { ...job, dedup: { id: 'second' } }).then( () => null, (rejection: { code?: string }) => rejection From 8c9c46096b39539abf3b308637967d083f823dfa Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 07:39:44 +0000 Subject: [PATCH 8/8] Revert "test: ignore popFrom deadlocks in the concurrent dedup tests" This reverts commit 7e4a40c15724785d71fc0b14dd8521ecd8ec6dc8. A popFrom deadlock on MySQL is a defect: the worker reports an error and waits before polling again. Treating it as an empty poll hid it from the only tests that run into it. It is fixed separately. --- tests/_utils/register_driver_test_suite.ts | 18 ++---------------- 1 file changed, 2 insertions(+), 16 deletions(-) diff --git a/tests/_utils/register_driver_test_suite.ts b/tests/_utils/register_driver_test_suite.ts index 9d6702c..a803271 100644 --- a/tests/_utils/register_driver_test_suite.ts +++ b/tests/_utils/register_driver_test_suite.ts @@ -18,20 +18,6 @@ interface DriverTestSuiteOptions { supportsAtomicDedup?: boolean } -/** - * Acquire a job, treating a MySQL deadlock victim as an empty poll. popFrom - * can deadlock with concurrent acquisitions on MySQL, which these dedup tests - * do not cover. - */ -async function popIgnoringDeadlock(adapter: Adapter, queue: string): Promise { - try { - return await adapter.popFrom(queue) - } catch (error) { - if ((error as { code?: string } | null)?.code === 'ER_LOCK_DEADLOCK') return null - throw error - } -} - /** A lease on a job that was never acquired. */ function unleased(id: string): JobLease { return { id, leaseToken: 'worker-1:never-acquired' } @@ -3111,7 +3097,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) { let dispatching = true const acquiring = (async () => { - while (dispatching) await popIgnoringDeadlock(adapter, 'concurrent-dedup-queue') + while (dispatching) await adapter.popFrom('concurrent-dedup-queue') })() const results = await Promise.all( @@ -3155,7 +3141,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) { let completed = 0 const workers = Array.from({ length: 2 }, async () => { while (dispatching || (await adapter.sizeOf('concurrent-dedup-queue')) > 0) { - const job = await popIgnoringDeadlock(adapter, 'concurrent-dedup-queue') + const job = await adapter.popFrom('concurrent-dedup-queue') if (job && (await adapter.completeJob(job, 'concurrent-dedup-queue'))) completed++ } })