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 diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index 6b59e08..2e8ceaf 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') @@ -361,11 +377,18 @@ export class KnexAdapter implements Adapter { } if (maxCount > 0) { - const toKeep = 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) @@ -671,7 +694,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[] = [] @@ -870,7 +893,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..6be8ccc 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']) @@ -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() } @@ -653,7 +655,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 +859,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 +922,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/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_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() 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 8b27877..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::' @@ -2363,8 +2366,68 @@ 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' + 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(config) + 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 + }, + }) + + 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({ @@ -2380,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(