From eb7520f6bb36e1af786ac0820c17699512a23163 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 08:18:11 +0000 Subject: [PATCH 1/4] fix(knex): prune completed jobs by count on MySQL Retention by count deleted the jobs whose id was not in a subquery ordered by finished_at with a LIMIT. MySQL and MariaDB reject LIMIT in an IN subquery, so completeJob() and failJob() threw on MySQL as soon as removeOnComplete or removeOnFail set a count. The Knex adapter now reads the ids to keep first and deletes the others, as the Kysely adapter does. The driver test suite now runs on Knex with MySQL, which had no test group and missed this error. --- src/drivers/knex_adapter.ts | 8 +++++-- tests/adapter.spec.ts | 43 +++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index 6b59e08..b13d4b2 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -361,7 +361,8 @@ export class KnexAdapter implements Adapter { } if (maxCount > 0) { - const toKeep = this.#connection(this.#jobsTable) + // MySQL rejects LIMIT in a NOT IN subquery: read the ids to keep first. + const toKeep = await this.#connection(this.#jobsTable) .where('queue', queue) .where('status', status) .orderBy('finished_at', 'desc') @@ -371,7 +372,10 @@ export class KnexAdapter implements Adapter { await this.#connection(this.#jobsTable) .where('queue', queue) .where('status', status) - .whereNotIn('id', toKeep) + .whereNotIn( + 'id', + toKeep.map((row) => row.id) + ) .delete() } } diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 8b27877..149216a 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -2363,6 +2363,49 @@ test.group('Adapter | Knex (PostgreSQL)', (group) => { }) }) +test.group('Adapter | Knex (MySQL)', (group) => { + let connection: ReturnType + let adapter: KnexAdapter + let schemaService: KnexQueueSchemaService + const tableName = 'knex_mysql_queue_jobs_test' + const schedulesTableName = 'knex_mysql_queue_schedules_test' + + group.each.setup(async () => { + connection = Knex({ + client: 'mysql2', + connection: { + host: process.env.MYSQL_HOST || 'localhost', + port: Number.parseInt(process.env.MYSQL_PORT || '3307', 10), + user: process.env.MYSQL_USER || 'root', + password: process.env.MYSQL_PASSWORD || 'mysql', + database: process.env.MYSQL_DATABASE || 'queue_test', + }, + }) + schemaService = new KnexQueueSchemaService(connection) + + await schemaService.dropJobsTable(tableName) + await schemaService.dropSchedulesTable(schedulesTableName) + await schemaService.createJobsTable(tableName) + await schemaService.createSchedulesTable(schedulesTableName) + + return async () => { + await adapter?.destroy() + await schemaService.dropJobsTable(tableName) + await schemaService.dropSchedulesTable(schedulesTableName) + await connection.destroy() + } + }) + + registerDriverTestSuite({ + test, + supportsAtomicDedup: false, + createAdapter: () => { + adapter = new KnexAdapter({ connection, tableName, schedulesTableName }) + return adapter + }, + }) +}) + test.group('Adapter | Knex (SQLite file, two connections)', () => { test('concurrent upserts from two connections all succeed', async ({ assert, cleanup }) => { const directory = await mkdtemp(join(tmpdir(), 'queue-sqlite-')) From 8a38c618958008ce142c5bd9676dbbef936c0873 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 08:18:12 +0000 Subject: [PATCH 2/4] fix(sql): claim jobs under READ COMMITTED on MySQL On MySQL, popFrom() failed with ER_LOCK_DEADLOCK when several workers acquired jobs while others were dispatched, with or without dedup. The Worker then waited 5 s before polling again. Under REPEATABLE READ, the MySQL default, SELECT ... FOR UPDATE SKIP LOCKED skips locked rows but keeps a lock on the gap before every index entry it scans. The claim UPDATE moves the row to another status, so InnoDB inserts its new entry in the status index, often into a gap that a concurrent claim holds. With the Kysely schema, status is a varchar: the new 'active' entry lands in the gap before the first 'pending' one, which every popFrom() locks, and two workers wait on each other. With the Knex schema, an enum, two promotions of delayed jobs deadlock on the execute_at and finished_at indexes the same way. The claims of both SQL adapters (popFrom(), the promotion of delayed jobs, recoverStalledJobs() and claimDueSchedule()) now run under READ COMMITTED on MySQL, which locks the claimed rows only, as PostgreSQL does by default. A retry on ER_LOCK_DEADLOCK would have kept the conflict and only hidden it. Other transactions keep the default level. Related https://github.com/boringnode/queue/pull/31 --- src/drivers/knex_adapter.ts | 24 ++++++++-- src/drivers/kysely_adapter.ts | 27 ++++++++++-- tests/_utils/register_driver_test_suite.ts | 51 ++++++++++++++++++++++ 3 files changed, 94 insertions(+), 8 deletions(-) diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index b13d4b2..8c62cbd 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -153,7 +153,7 @@ export class KnexAdapter implements Adapter { await this.#processDelayedJobs(queue, now) // Use a transaction to atomically pop a job - return this.#connection.transaction(async (trx) => { + return this.#claimTransaction(async (trx) => { // Build the query for highest priority job (lowest score) let query = trx(this.#jobsTable) .where('queue', queue) @@ -216,6 +216,22 @@ export class KnexAdapter implements Adapter { return client === 'mysql' || client === 'mysql2' || client === 'mariadb' } + /** + * Transaction for a SELECT ... FOR UPDATE SKIP LOCKED claim followed by an + * UPDATE of the claimed rows. + * + * Under the MySQL default, REPEATABLE READ, the locking read also locks the + * gap before every index entry it scans, skipped rows included. The claim + * UPDATE then moves the row to another status, which inserts its new index + * entry into a gap that a concurrent claim holds: two workers acquiring jobs + * wait on each other and InnoDB aborts one with ER_LOCK_DEADLOCK. READ + * COMMITTED only locks the claimed rows, as PostgreSQL does by default. + */ + #claimTransaction(callback: (trx: Knex.Transaction) => Promise): Promise { + if (!this.#isMysql()) return this.#connection.transaction(callback) + return this.#connection.transaction(callback, { isolationLevel: 'read committed' }) + } + #isSqlite(): boolean { const client = String(this.#connection.client.config.client) return client.includes('sqlite') @@ -223,7 +239,7 @@ export class KnexAdapter implements Adapter { async #processDelayedJobs(queue: string, now: number): Promise { // Use a transaction with row locking to prevent race conditions - await this.#connection.transaction(async (trx) => { + await this.#claimTransaction(async (trx) => { let query = trx(this.#jobsTable) .where('queue', queue) .where('status', 'delayed') @@ -675,7 +691,7 @@ export class KnexAdapter implements Adapter { const stalledCutoff = now - stalledThreshold // Use a transaction with row locking to prevent race conditions - return this.#connection.transaction(async (trx) => { + return this.#claimTransaction(async (trx) => { let recovered = 0 const exceeded: AcquiredJob[] = [] @@ -874,7 +890,7 @@ export class KnexAdapter implements Adapter { const now = Date.now() - return this.#connection.transaction(async (trx) => { + return this.#claimTransaction(async (trx) => { // Find one due schedule with row locking let query = trx(this.#schedulesTable) .where('status', 'active') diff --git a/src/drivers/kysely_adapter.ts b/src/drivers/kysely_adapter.ts index e808c93..e0d8ec4 100644 --- a/src/drivers/kysely_adapter.ts +++ b/src/drivers/kysely_adapter.ts @@ -178,7 +178,7 @@ export class KyselyAdapter implements Adapter { await this.#processDelayedJobs(this.#connection, queue, now) - return this.#withTransaction(this.#connection, async (trx) => { + return this.#withClaimTransaction(this.#connection, async (trx) => { let query = this.#jobs(trx) .selectFrom(this.#jobsTable) .selectAll() @@ -214,7 +214,7 @@ export class KyselyAdapter implements Adapter { } async #processDelayedJobs(connection: Kysely, queue: string, now: number): Promise { - await this.#withTransaction(connection, async (trx) => { + await this.#withClaimTransaction(connection, async (trx) => { let query = this.#jobs(trx) .selectFrom(this.#jobsTable) .select(['id', 'data']) @@ -653,7 +653,7 @@ export class KyselyAdapter implements Adapter { ): Promise { const now = Date.now() - return this.#withTransaction(this.#connection, async (trx) => { + return this.#withClaimTransaction(this.#connection, async (trx) => { let query = this.#jobs(trx) .selectFrom(this.#jobsTable) .select(['id', 'data']) @@ -857,7 +857,7 @@ export class KyselyAdapter implements Adapter { const now = Date.now() - return this.#withTransaction(this.#connection, async (trx) => { + return this.#withClaimTransaction(this.#connection, async (trx) => { let query = this.#schedules(trx) .selectFrom(this.#schedulesTable) .selectAll() @@ -920,6 +920,25 @@ export class KyselyAdapter implements Adapter { return connection.transaction().execute(callback) } + /** + * Transaction for a SELECT ... FOR UPDATE SKIP LOCKED claim followed by an + * UPDATE of the claimed rows. + * + * Under the MySQL default, REPEATABLE READ, the locking read also locks the + * gap before every index entry it scans, skipped rows included. The claim + * UPDATE then moves the row to another status, which inserts its new index + * entry into a gap that a concurrent claim holds: two workers acquiring jobs + * wait on each other and InnoDB aborts one with ER_LOCK_DEADLOCK. READ + * COMMITTED only locks the claimed rows, as PostgreSQL does by default. + */ + #withClaimTransaction( + connection: Kysely, + callback: (trx: Transaction) => Promise + ): Promise { + if (this.#dialect !== 'mysql') return this.#withTransaction(connection, callback) + return connection.transaction().setIsolationLevel('read committed').execute(callback) + } + #jobs(connection: InternalKyselyConnection): Kysely { return connection as unknown as Kysely } diff --git a/tests/_utils/register_driver_test_suite.ts b/tests/_utils/register_driver_test_suite.ts index 8bfdb4b..c59de9e 100644 --- a/tests/_utils/register_driver_test_suite.ts +++ b/tests/_utils/register_driver_test_suite.ts @@ -1483,6 +1483,57 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) { assert.notEqual(job1!.id, job2!.id, 'Workers should acquire different jobs') }) + // Under REPEATABLE READ, this load made the MySQL claims deadlock on every run. + test('concurrent popFrom does not fail while jobs are dispatched', async ({ assert }) => { + const dispatchers = 6 + const jobsPerDispatcher = 400 + const total = dispatchers * jobsPerDispatcher + const acquired: string[] = [] + let stopped = false + + // Any error, a deadlock included, stops every task and fails the test. + const stopOnError = (task: Promise) => + task.catch((error) => { + stopped = true + throw error + }) + + const results = await Promise.allSettled( + [ + ...Array.from({ length: dispatchers }, async (_, dispatcher) => { + const adapter = await options.createAdapter() + for (let index = 0; index < jobsPerDispatcher && !stopped; index++) { + const job = { + id: `${dispatcher}-${index}`, + name: 'TestJob', + payload: {}, + attempts: 0, + } + if (index % 2) await adapter.pushLaterOn('test-queue', job, 1) + else await adapter.pushOn('test-queue', job) + } + }), + ...Array.from({ length: 4 }, async (_, worker) => { + const adapter = await options.createAdapter() + adapter.setWorkerId(`worker-${worker}`) + while (!stopped && acquired.length < total) { + const job = await adapter.popFrom('test-queue') + if (!job) continue + acquired.push(job.id) + await adapter.completeJob(job, 'test-queue') + } + }), + ].map(stopOnError) + ) + + const failure = results.find( + (result): result is PromiseRejectedResult => result.status === 'rejected' + ) + if (failure) throw failure.reason + assert.equal(new Set(acquired).size, total) + assert.lengthOf(acquired, total) + }).timeout(30_000) + test('a worker cannot finalize a job acquired again by another worker', async ({ assert }) => { const adapter1 = await options.createAdapter() const adapter2 = await options.createAdapter() From 959574ef92277d7ba64f430886b2d2b41bd73fc3 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 08:20:16 +0000 Subject: [PATCH 3/4] docs: note that MySQL claims need a row or mixed binlog The SQL adapters now acquire jobs and schedules in READ COMMITTED transactions on MySQL. InnoDB rejects writes at that level when binlog_format is STATEMENT, so document which formats work. --- README.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/README.md b/README.md index 71eafd1..fd37923 100644 --- a/README.md +++ b/README.md @@ -473,6 +473,13 @@ await schema.dropJobsTable() +#### MySQL binary log format + +On MySQL, the Knex and Kysely adapters acquire jobs and schedules in READ COMMITTED transactions, +which avoids deadlocks between workers under REPEATABLE READ. With `binlog_format=STATEMENT`, InnoDB +rejects writes in READ COMMITTED, so jobs cannot be acquired. `ROW`, the MySQL 8 default, and +`MIXED`, the MariaDB default, both work. + #### Migrating SQL schedules after an upgrade The schedules table stores its dates as epoch milliseconds (`bigint`), so they do not depend on the From 14547439d665f4cc949d55dea85b82ff19569f66 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Mon, 12 Oct 2026 08:37:05 +0000 Subject: [PATCH 4/4] fix(sql): prune jobs by count in a single statement Retention by count read the ids to keep, then deleted the other jobs in a second statement, outside a transaction. Two workers could prune between these steps: with { count: 1 }, A finished its job and read [A], B finished a more recent job and pruned A, then the DELETE of A removed B and no job remained. Kysely had pruned in two steps before; the Knex adapter did so since eb7520f, which removed a LIMIT subquery that MySQL rejects. Both adapters now delete in one statement and select the ids to keep through a derived table, which MySQL accepts with a LIMIT. A test pauses the first worker before its DELETE, on two connections, and checks that the latest job remains, for completeJob() and failJob() on SQLite and MySQL. --- src/drivers/knex_adapter.ts | 23 +++--- src/drivers/kysely_adapter.ts | 28 ++++---- tests/_utils/pause_deletes.ts | 45 ++++++++++++ tests/_utils/register_retention_race_suite.ts | 68 ++++++++++++++++++ tests/adapter.spec.ts | 59 ++++++++++++--- tests/kysely_adapter.spec.ts | 72 +++++++++++++++---- 6 files changed, 247 insertions(+), 48 deletions(-) create mode 100644 tests/_utils/pause_deletes.ts create mode 100644 tests/_utils/register_retention_race_suite.ts diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index 8c62cbd..2e8ceaf 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -377,21 +377,24 @@ export class KnexAdapter implements Adapter { } if (maxCount > 0) { - // MySQL rejects LIMIT in a NOT IN subquery: read the ids to keep first. - const toKeep = await this.#connection(this.#jobsTable) - .where('queue', queue) - .where('status', status) - .orderBy('finished_at', 'desc') - .limit(maxCount) + // One statement, so that a concurrent pruning cannot delete a job kept here. + // MySQL rejects LIMIT in an IN subquery, but accepts it in a derived table. + const toKeep = this.#connection + .from( + this.#connection(this.#jobsTable) + .where('queue', queue) + .where('status', status) + .orderBy('finished_at', 'desc') + .limit(maxCount) + .select('id') + .as('keep') + ) .select('id') await this.#connection(this.#jobsTable) .where('queue', queue) .where('status', status) - .whereNotIn( - 'id', - toKeep.map((row) => row.id) - ) + .whereNotIn('id', toKeep) .delete() } } diff --git a/src/drivers/kysely_adapter.ts b/src/drivers/kysely_adapter.ts index e0d8ec4..6be8ccc 100644 --- a/src/drivers/kysely_adapter.ts +++ b/src/drivers/kysely_adapter.ts @@ -360,23 +360,25 @@ export class KyselyAdapter implements Adapter { } if (maxCount > 0) { - const toKeep = await this.#jobs(connection) - .selectFrom(this.#jobsTable) - .select('id') - .where('queue', '=', queue) - .where('status', '=', status) - .orderBy('finished_at', 'desc') - .limit(maxCount) - .execute() - + // One statement, so that a concurrent pruning cannot delete a job kept here. + // MySQL rejects LIMIT in an IN subquery, but accepts it in a derived table. await this.#jobs(connection) .deleteFrom(this.#jobsTable) .where('queue', '=', queue) .where('status', '=', status) - .where( - 'id', - 'not in', - toKeep.map((row) => row.id) + .where('id', 'not in', (expression) => + expression + .selectFrom((subquery) => + subquery + .selectFrom(this.#jobsTable) + .select('id') + .where('queue', '=', queue) + .where('status', '=', status) + .orderBy('finished_at', 'desc') + .limit(maxCount) + .as('keep') + ) + .select('keep.id') ) .execute() } diff --git a/tests/_utils/pause_deletes.ts b/tests/_utils/pause_deletes.ts new file mode 100644 index 0000000..64a827d --- /dev/null +++ b/tests/_utils/pause_deletes.ts @@ -0,0 +1,45 @@ +import type { Knex } from 'knex' +import type { DatabaseConnection, Dialect } from 'kysely' + +type BeforeDelete = () => Promise + +const isDelete = (sql: string) => /^\s*delete\b/i.test(sql) + +/** Makes a Knex instance await `beforeDelete` before it sends each DELETE. */ +export function pauseKnexDeletes(connection: Knex, beforeDelete: BeforeDelete): Knex { + const client = connection.client as { _query(connection: unknown, query: { sql: string }): any } + const query = client._query.bind(client) + client._query = async (raw, obj) => { + if (isDelete(obj.sql)) await beforeDelete() + return query(raw, obj) + } + return connection +} + +/** Wraps a Kysely dialect so that it awaits `beforeDelete` before it sends each DELETE. */ +export function pauseKyselyDeletes(dialect: Dialect, beforeDelete: BeforeDelete): Dialect { + return { + createAdapter: () => dialect.createAdapter(), + createIntrospector: (db) => dialect.createIntrospector(db), + createQueryCompiler: () => dialect.createQueryCompiler(), + createDriver: () => { + const driver = dialect.createDriver() + const paused = new WeakSet() + const acquireConnection = driver.acquireConnection.bind(driver) + + driver.acquireConnection = async () => { + const connection = await acquireConnection() + if (paused.has(connection)) return connection + + paused.add(connection) + const executeQuery = connection.executeQuery.bind(connection) + connection.executeQuery = async (query, options) => { + if (isDelete(query.sql)) await beforeDelete() + return executeQuery(query, options) + } + return connection + } + return driver + }, + } +} diff --git a/tests/_utils/register_retention_race_suite.ts b/tests/_utils/register_retention_race_suite.ts new file mode 100644 index 0000000..3e68420 --- /dev/null +++ b/tests/_utils/register_retention_race_suite.ts @@ -0,0 +1,68 @@ +import { setTimeout } from 'node:timers/promises' +import { test as JapaTest } from '@japa/runner' +import type { AcquiredJob, Adapter } from '../../src/contracts/adapter.js' + +interface RetentionRaceSuiteOptions { + test: typeof JapaTest + /** + * Two adapters on separate connections to the same jobs table. The first one + * awaits `beforeDelete` before it sends each DELETE. + */ + createAdapters: ( + beforeDelete: () => Promise, + cleanup: (handler: () => Promise) => void + ) => Promise<[Adapter, Adapter]> +} + +const finalizers = { + completeJob: (adapter: Adapter, job: AcquiredJob) => + adapter.completeJob(job, 'retention-race', { count: 1 }), + failJob: (adapter: Adapter, job: AcquiredJob) => + adapter.failJob(job, 'retention-race', new Error('boom'), { count: 1 }), +} + +export function registerRetentionRaceSuite(options: RetentionRaceSuiteOptions) { + for (const [name, finalize] of Object.entries(finalizers)) { + options + .test( + `${name} keeps the latest job when another worker prunes meanwhile`, + async ({ assert, cleanup }) => { + let beforeDelete: (() => Promise) | undefined + const [first, second] = await options.createAdapters( + async () => beforeDelete?.(), + cleanup + ) + first.setWorkerId('worker-1') + second.setWorkerId('worker-2') + + for (const id of ['job-a', 'job-b']) { + await first.pushOn('retention-race', { id, name: 'TestJob', payload: {}, attempts: 0 }) + } + const jobA = (await first.popFrom('retention-race'))! + const jobB = (await second.popFrom('retention-race'))! + + // A finalizes its job and stops before the DELETE of its pruning. + const reached = Promise.withResolvers() + const resumed = Promise.withResolvers() + beforeDelete = async () => { + beforeDelete = undefined + reached.resolve() + await resumed.promise + } + const finalizingA = finalize(first, jobA) + await reached.promise + + // B finalizes a more recent job and prunes the job of A. + await setTimeout(5) + await finalize(second, jobB) + + resumed.resolve() + await finalizingA + + assert.isNull(await second.getJob(jobA.id, 'retention-race')) + assert.isNotNull(await second.getJob(jobB.id, 'retention-race')) + } + ) + .timeout(10_000) + } +} diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 149216a..c3dea3c 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -2,6 +2,7 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import Knex from 'knex' +import type { Knex as KnexTypes } from 'knex' import { test } from '@japa/runner' import { Redis } from 'ioredis' import { CronExpressionParser } from 'cron-parser' @@ -12,6 +13,8 @@ import { KnexQueueSchemaService } from '../src/services/knex_queue_schema.js' import { registerDriverTestSuite } from './_utils/register_driver_test_suite.js' import { withRedisWriteSpy } from './_utils/with_redis_write_spy.js' import { withKnexQuerySpy } from './_utils/with_knex_query_spy.js' +import { pauseKnexDeletes } from './_utils/pause_deletes.js' +import { registerRetentionRaceSuite } from './_utils/register_retention_race_suite.js' const KEY_PREFIX = 'boringnode::queue::test::' @@ -2369,18 +2372,19 @@ test.group('Adapter | Knex (MySQL)', (group) => { let schemaService: KnexQueueSchemaService const tableName = 'knex_mysql_queue_jobs_test' const schedulesTableName = 'knex_mysql_queue_schedules_test' + const config: KnexTypes.Config = { + client: 'mysql2', + connection: { + host: process.env.MYSQL_HOST || 'localhost', + port: Number.parseInt(process.env.MYSQL_PORT || '3307', 10), + user: process.env.MYSQL_USER || 'root', + password: process.env.MYSQL_PASSWORD || 'mysql', + database: process.env.MYSQL_DATABASE || 'queue_test', + }, + } group.each.setup(async () => { - connection = Knex({ - client: 'mysql2', - connection: { - host: process.env.MYSQL_HOST || 'localhost', - port: Number.parseInt(process.env.MYSQL_PORT || '3307', 10), - user: process.env.MYSQL_USER || 'root', - password: process.env.MYSQL_PASSWORD || 'mysql', - database: process.env.MYSQL_DATABASE || 'queue_test', - }, - }) + connection = Knex(config) schemaService = new KnexQueueSchemaService(connection) await schemaService.dropJobsTable(tableName) @@ -2404,10 +2408,26 @@ test.group('Adapter | Knex (MySQL)', (group) => { return adapter }, }) + + registerRetentionRaceSuite({ + test, + createAdapters: async (beforeDelete, cleanup) => { + const first = pauseKnexDeletes(Knex(config), beforeDelete) + const second = Knex(config) + cleanup(async () => { + await first.destroy() + await second.destroy() + }) + return [ + new KnexAdapter({ connection: first, tableName }), + new KnexAdapter({ connection: second, tableName }), + ] + }, + }) }) test.group('Adapter | Knex (SQLite file, two connections)', () => { - test('concurrent upserts from two connections all succeed', async ({ assert, cleanup }) => { + const connectTwice = async (cleanup: (handler: () => Promise) => void) => { const directory = await mkdtemp(join(tmpdir(), 'queue-sqlite-')) const connect = () => Knex({ @@ -2423,6 +2443,23 @@ test.group('Adapter | Knex (SQLite file, two connections)', () => { await second.destroy() await rm(directory, { recursive: true, force: true }) }) + return [first, second] as const + } + + registerRetentionRaceSuite({ + test, + createAdapters: async (beforeDelete, cleanup) => { + const [first, second] = await connectTwice(cleanup) + await new KnexQueueSchemaService(first).createJobsTable() + return [ + new KnexAdapter({ connection: pauseKnexDeletes(first, beforeDelete) }), + new KnexAdapter({ connection: second }), + ] + }, + }) + + test('concurrent upserts from two connections all succeed', async ({ assert, cleanup }) => { + const [first, second] = await connectTwice(cleanup) const schedulesTableName = 'queue_schedules' await new KnexQueueSchemaService(first).createSchedulesTable(schedulesTableName) diff --git a/tests/kysely_adapter.spec.ts b/tests/kysely_adapter.spec.ts index 51f827b..c4a89e7 100644 --- a/tests/kysely_adapter.spec.ts +++ b/tests/kysely_adapter.spec.ts @@ -12,6 +12,8 @@ import { type QueueDatabase, } from '../src/drivers/kysely_adapter.js' import { registerDriverTestSuite } from './_utils/register_driver_test_suite.js' +import { pauseKyselyDeletes } from './_utils/pause_deletes.js' +import { registerRetentionRaceSuite } from './_utils/register_retention_race_suite.js' test.group('Adapter | Kysely (SQLite)', (group) => { let connection: Kysely @@ -145,18 +147,19 @@ test.group('Adapter | Kysely (MySQL)', (group) => { let adapter: KyselyAdapter let schema: KyselyQueueSchemaService - group.each.setup(async () => { - connection = new Kysely({ - dialect: new MysqlDialect({ - pool: createPool({ - host: process.env.MYSQL_HOST || 'localhost', - port: Number.parseInt(process.env.MYSQL_PORT || '3307', 10), - user: process.env.MYSQL_USER || 'root', - password: process.env.MYSQL_PASSWORD || 'mysql', - database: process.env.MYSQL_DATABASE || 'queue_test', - }), + const createDialect = () => + new MysqlDialect({ + pool: createPool({ + host: process.env.MYSQL_HOST || 'localhost', + port: Number.parseInt(process.env.MYSQL_PORT || '3307', 10), + user: process.env.MYSQL_USER || 'root', + password: process.env.MYSQL_PASSWORD || 'mysql', + database: process.env.MYSQL_DATABASE || 'queue_test', }), }) + + group.each.setup(async () => { + connection = new Kysely({ dialect: createDialect() }) schema = new KyselyQueueSchemaService(connection, { dialect: 'mysql' }) await schema.dropJobsTable(tableName) @@ -190,23 +193,64 @@ test.group('Adapter | Kysely (MySQL)', (group) => { await schema.addDedupColumns(tableName) await schema.addDedupColumns(tableName) }) + + registerRetentionRaceSuite({ + test, + createAdapters: async (beforeDelete, cleanup) => { + const first = new Kysely({ + dialect: pauseKyselyDeletes(createDialect(), beforeDelete), + }) + const second = new Kysely({ dialect: createDialect() }) + cleanup(async () => { + await first.destroy() + await second.destroy() + }) + return [ + new KyselyAdapter({ connection: first, dialect: 'mysql', tableName }), + new KyselyAdapter({ connection: second, dialect: 'mysql', tableName }), + ] + }, + }) }) test.group('KyselyAdapter | SQLite file, two connections', () => { - test('concurrent upserts from two connections all succeed', async ({ assert, cleanup }) => { + const connectTwice = async ( + cleanup: (handler: () => Promise) => void, + beforeDelete?: () => Promise + ) => { const directory = await mkdtemp(join(tmpdir(), 'queue-sqlite-')) - const connect = () => { + const connect = (pause?: () => Promise) => { const database = new Database(join(directory, 'queue.sqlite')) database.pragma('journal_mode = WAL') - return new Kysely({ dialect: new SqliteDialect({ database }) }) + const dialect = new SqliteDialect({ database }) + return new Kysely({ + dialect: pause ? pauseKyselyDeletes(dialect, pause) : dialect, + }) } - const first = connect() + const first = connect(beforeDelete) const second = connect() cleanup(async () => { await first.destroy() await second.destroy() await rm(directory, { recursive: true, force: true }) }) + return [first, second] as const + } + + registerRetentionRaceSuite({ + test, + createAdapters: async (beforeDelete, cleanup) => { + const [first, second] = await connectTwice(cleanup, beforeDelete) + await new KyselyQueueSchemaService(first, { dialect: 'sqlite' }).createJobsTable() + return [ + new KyselyAdapter({ connection: first, dialect: 'sqlite' }), + new KyselyAdapter({ connection: second, dialect: 'sqlite' }), + ] + }, + }) + + test('concurrent upserts from two connections all succeed', async ({ assert, cleanup }) => { + const [first, second] = await connectTwice(cleanup) const schedulesTableName = 'queue_schedules' await new KyselyQueueSchemaService(first, { dialect: 'sqlite' }).createSchedulesTable(