fix(sql): stop MySQL deadlocks when workers acquire jobs

README.md

@@ -473,6 +473,13 @@ await schema.dropJobsTable()
 
 </details>
 
+#### 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

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,14 +216,30 @@ 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<T>(callback: (trx: Knex.Transaction) => Promise<T>): Promise<T> {
+    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')
   }
 
   async #processDelayedJobs(queue: string, now: number): Promise<void> {
     // 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')

src/drivers/kysely_adapter.ts

@@ -178,7 +178,7 @@ export class KyselyAdapter<DB = QueueDatabase> 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<DB = QueueDatabase> implements Adapter {
   }
 
   async #processDelayedJobs(connection: Kysely<DB>, queue: string, now: number): Promise<void> {
-    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<DB = QueueDatabase> 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<DB = QueueDatabase> implements Adapter {
   ): Promise<StalledJobsRecovery> {
     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<DB = QueueDatabase> 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<DB = QueueDatabase> 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<T>(
+    connection: Kysely<DB>,
+    callback: (trx: Transaction<DB>) => Promise<T>
+  ): Promise<T> {
+    if (this.#dialect !== 'mysql') return this.#withTransaction(connection, callback)
+    return connection.transaction().setIsolationLevel('read committed').execute(callback)
+  }
+
   #jobs(connection: InternalKyselyConnection<DB>): Kysely<JobsDatabase> {
     return connection as unknown as Kysely<JobsDatabase>
   }

tests/_utils/pause_deletes.ts

@@ -0,0 +1,45 @@
+import type { Knex } from 'knex'
+import type { DatabaseConnection, Dialect } from 'kysely'
+
+type BeforeDelete = () => Promise<void>
+
+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<DatabaseConnection>()
+      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
+    },
+  }
+}

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<void>) =>
+        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()

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<void>,
+    cleanup: (handler: () => Promise<void>) => 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<void>) | 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<void>()
+          const resumed = Promise.withResolvers<void>()
+          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)
+  }
+}