fix(sql): lock SQLite for writing before reading in claims · boringnode/queue@fb65271 · GitHub
Skip to content

Commit fb65271

Browse files
RomainLanzRomainLanz
committed
fix(sql): lock SQLite for writing before reading in claims
Knex and Kysely start SQLite transactions with a deferred BEGIN. The claim, delayed-job promotion, stalled recovery, schedule claim, and dedup transactions read before they write: when another process wrote in between, SQLite failed them at once with SQLITE_BUSY, without applying the busy timeout. A single worker next to a dispatcher process lost a cycle and paused 5 s each time, and two processes deduplicating the same id got a rejection instead of `skipped`. These transactions now start with a no-op UPDATE on SQLite, which takes the write lock up front like BEGIN IMMEDIATE, so they wait for the busy timeout. Neither query builder lets us change its BEGIN statement. The README documents running SQLite with several processes.
1 parent 71d2fa1 commit fb65271

7 files changed

Lines changed: 195 additions & 19 deletions

File tree

‎.changelog/sqlite-write-lock.md‎

Lines changed: 14 additions & 0 deletions

‎README.md‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -512,6 +512,22 @@ which avoids deadlocks between workers under REPEATABLE READ. With `binlog_forma
512512
rejects writes in READ COMMITTED, so jobs cannot be acquired. `ROW`, the MySQL 8 default, and
513513
`MIXED`, the MariaDB default, both work.
514514

515+
#### SQLite with several processes
516+
517+
SQLite lets one connection write at a time. The Knex and Kysely adapters take the write lock at the
518+
start of each transaction that reads then writes, as `BEGIN IMMEDIATE` does, so a worker and
519+
dispatchers in other processes wait for each other up to the busy timeout (5 seconds by default with
520+
`better-sqlite3`) instead of failing with `SQLITE_BUSY`.
521+
522+
- Use a database file shared by the processes, and enable WAL (`PRAGMA journal_mode = WAL`) so
523+
reads do not wait for writes.
524+
- Open one connection to the file per process. `better-sqlite3` waits for the lock synchronously:
525+
a second connection in the same process holding the lock cannot release it while the first one
526+
waits.
527+
- Every write is serialized, so a busy timeout shorter than your longest write transaction still
528+
fails, and throughput stays bounded. For several workers or a high dispatch rate, use PostgreSQL
529+
or MySQL.
530+
515531
#### Migrating SQL schedules after an upgrade
516532

517533
The schedules table stores its dates as epoch milliseconds (`bigint`), so they do not depend on the

‎src/drivers/knex_adapter.ts‎

Lines changed: 29 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -153,7 +153,7 @@ export class KnexAdapter implements Adapter {
153153
await this.#processDelayedJobs(queue, now)
154154

155155
// Use a transaction to atomically pop a job
156-
return this.#claimTransaction(async (trx) => {
156+
return this.#claimTransaction(this.#jobsTable, async (trx) => {
157157
// Build the query for highest priority job (lowest score)
158158
let query = trx(this.#jobsTable)
159159
.where('queue', queue)
@@ -227,9 +227,30 @@ export class KnexAdapter implements Adapter {
227227
* wait on each other and InnoDB aborts one with ER_LOCK_DEADLOCK. READ
228228
* COMMITTED only locks the claimed rows, as PostgreSQL does by default.
229229
*/
230-
#claimTransaction<T>(callback: (trx: Knex.Transaction) => Promise<T>): Promise<T> {
231-
if (!this.#isMysql()) return this.#connection.transaction(callback)
232-
return this.#connection.transaction(callback, { isolationLevel: 'read committed' })
230+
#claimTransaction<T>(table: string, callback: (trx: Knex.Transaction) => Promise<T>): Promise<T> {
231+
if (!this.#isMysql()) return this.#writeTransaction(table, callback)
232+
return this.#writeTransaction(table, callback, { isolationLevel: 'read committed' })
233+
}
234+
235+
/**
236+
* Transaction that reads rows of `table`, then writes.
237+
*
238+
* Knex starts a SQLite transaction with a deferred BEGIN, which takes the
239+
* write lock at the first write. When another connection, such as a
240+
* dispatcher process, writes between the read and that write, SQLite fails
241+
* the transaction with SQLITE_BUSY at once, without waiting for the busy
242+
* timeout. Writing first takes the write lock up front, as BEGIN IMMEDIATE
243+
* does: the transaction waits for the busy timeout instead.
244+
*/
245+
#writeTransaction<T>(
246+
table: string,
247+
callback: (trx: Knex.Transaction) => Promise<T>,
248+
config?: Knex.TransactionConfig
249+
): Promise<T> {
250+
return this.#connection.transaction(async (trx) => {
251+
if (this.#isSqlite()) await trx.raw('update ?? set id = id where 0', [table])
252+
return callback(trx)
253+
}, config)
233254
}
234255

235256
#isSqlite(): boolean {
@@ -239,7 +260,7 @@ export class KnexAdapter implements Adapter {
239260

240261
async #processDelayedJobs(queue: string, now: number): Promise<void> {
241262
// Use a transaction with row locking to prevent race conditions
242-
await this.#claimTransaction(async (trx) => {
263+
await this.#claimTransaction(this.#jobsTable, async (trx) => {
243264
let query = trx(this.#jobsTable)
244265
.where('queue', queue)
245266
.where('status', 'delayed')
@@ -540,7 +561,7 @@ export class KnexAdapter implements Adapter {
540561
): Promise<PushResult> {
541562
while (true) {
542563
try {
543-
return await this.#connection.transaction(async (trx) => {
564+
return await this.#writeTransaction(this.#jobsTable, async (trx) => {
544565
const now = Date.now()
545566
const existingResult = await this.#resolveExistingDedup(trx, queue, jobData, dedup, now)
546567
if (existingResult) return existingResult
@@ -685,7 +706,7 @@ export class KnexAdapter implements Adapter {
685706
const stalledCutoff = now - stalledThreshold
686707

687708
// Use a transaction with row locking to prevent race conditions
688-
return this.#claimTransaction(async (trx) => {
709+
return this.#claimTransaction(this.#jobsTable, async (trx) => {
689710
let recovered = 0
690711
const exceeded: AcquiredJob[] = []
691712

@@ -884,7 +905,7 @@ export class KnexAdapter implements Adapter {
884905

885906
const now = Date.now()
886907

887-
return this.#claimTransaction(async (trx) => {
908+
return this.#claimTransaction(this.#schedulesTable, async (trx) => {
888909
// Find one due schedule with row locking
889910
let query = trx(this.#schedulesTable)
890911
.where('status', 'active')

‎src/drivers/kysely_adapter.ts‎

Lines changed: 24 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -178,7 +178,7 @@ export class KyselyAdapter<DB = QueueDatabase> implements Adapter {
178178

179179
await this.#processDelayedJobs(this.#connection, queue, now)
180180

181-
return this.#withClaimTransaction(this.#connection, async (trx) => {
181+
return this.#withClaimTransaction(this.#connection, this.#jobsTable, async (trx) => {
182182
let query = this.#jobs(trx)
183183
.selectFrom(this.#jobsTable)
184184
.selectAll()
@@ -214,7 +214,7 @@ export class KyselyAdapter<DB = QueueDatabase> implements Adapter {
214214
}
215215

216216
async #processDelayedJobs(connection: Kysely<DB>, queue: string, now: number): Promise<void> {
217-
await this.#withClaimTransaction(connection, async (trx) => {
217+
await this.#withClaimTransaction(connection, this.#jobsTable, async (trx) => {
218218
let query = this.#jobs(trx)
219219
.selectFrom(this.#jobsTable)
220220
.select(['id', 'data'])
@@ -483,7 +483,7 @@ export class KyselyAdapter<DB = QueueDatabase> implements Adapter {
483483

484484
while (true) {
485485
try {
486-
return await this.#withTransaction(connection, async (trx) => {
486+
return await this.#withTransaction(connection, this.#jobsTable, async (trx) => {
487487
const now = Date.now()
488488
const existingResult = await this.#resolveExistingDedup(trx, queue, jobData, dedup, now)
489489
if (existingResult) return existingResult
@@ -642,7 +642,7 @@ export class KyselyAdapter<DB = QueueDatabase> implements Adapter {
642642
): Promise<StalledJobsRecovery> {
643643
const now = Date.now()
644644

645-
return this.#withClaimTransaction(this.#connection, async (trx) => {
645+
return this.#withClaimTransaction(this.#connection, this.#jobsTable, async (trx) => {
646646
let query = this.#jobs(trx)
647647
.selectFrom(this.#jobsTable)
648648
.select(['id', 'data'])
@@ -846,7 +846,7 @@ export class KyselyAdapter<DB = QueueDatabase> implements Adapter {
846846

847847
const now = Date.now()
848848

849-
return this.#withClaimTransaction(this.#connection, async (trx) => {
849+
return this.#withClaimTransaction(this.#connection, this.#schedulesTable, async (trx) => {
850850
let query = this.#schedules(trx)
851851
.selectFrom(this.#schedulesTable)
852852
.selectAll()
@@ -902,11 +902,27 @@ export class KyselyAdapter<DB = QueueDatabase> implements Adapter {
902902
})
903903
}
904904

905+
/**
906+
* Transaction that reads rows of `table`, then writes.
907+
*
908+
* Kysely starts a SQLite transaction with a deferred BEGIN, which takes the
909+
* write lock at the first write. When another connection, such as a
910+
* dispatcher process, writes between the read and that write, SQLite fails
911+
* the transaction with SQLITE_BUSY at once, without waiting for the busy
912+
* timeout. Writing first takes the write lock up front, as BEGIN IMMEDIATE
913+
* does: the transaction waits for the busy timeout instead.
914+
*/
905915
#withTransaction<T>(
906916
connection: Kysely<DB>,
917+
table: string,
907918
callback: (trx: Transaction<DB>) => Promise<T>
908919
): Promise<T> {
909-
return connection.transaction().execute(callback)
920+
return connection.transaction().execute(async (trx) => {
921+
if (this.#dialect === 'sqlite') {
922+
await sql`update ${sql.table(table)} set id = id where 0`.execute(trx)
923+
}
924+
return callback(trx)
925+
})
910926
}
911927

912928
/**
@@ -922,9 +938,10 @@ export class KyselyAdapter<DB = QueueDatabase> implements Adapter {
922938
*/
923939
#withClaimTransaction<T>(
924940
connection: Kysely<DB>,
941+
table: string,
925942
callback: (trx: Transaction<DB>) => Promise<T>
926943
): Promise<T> {
927-
if (this.#dialect !== 'mysql') return this.#withTransaction(connection, callback)
944+
if (this.#dialect !== 'mysql') return this.#withTransaction(connection, table, callback)
928945
return connection.transaction().setIsolationLevel('read committed').execute(callback)
929946
}
930947

‎tests/_utils/sqlite_write_contender.ts‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
import { createRequire } from 'node:module'
2+
import { Worker } from 'node:worker_threads'
3+
4+
const contenderSource = `
5+
const { workerData, parentPort } = require('node:worker_threads')
6+
const Database = require(workerData.betterSqlite3)
7+
const db = new Database(workerData.filename, { timeout: 5000 })
8+
db.exec('create table if not exists write_contender (id integer primary key)')
9+
const stop = new Int32Array(workerData.stop)
10+
const sleep = (ms) => Atomics.wait(stop, 0, 0, ms)
11+
parentPort.postMessage('ready')
12+
while (Atomics.load(stop, 0) === 0) {
13+
db.exec('begin immediate')
14+
db.exec('insert into write_contender default values')
15+
sleep(2)
16+
db.exec('commit')
17+
sleep(6)
18+
}
19+
db.close()
20+
`
21+
22+
/**
23+
* Write to a SQLite file from another thread, as another process would: a
24+
* loop of short write transactions that keep the write lock a quarter of the time.
25+
* Stop it before closing the file.
26+
*/
27+
export async function startSqliteWriteContender(filename: string) {
28+
const stop = new SharedArrayBuffer(4)
29+
const worker = new Worker(contenderSource, {
30+
eval: true,
31+
workerData: {
32+
filename,
33+
stop,
34+
betterSqlite3: createRequire(import.meta.url).resolve('better-sqlite3'),
35+
},
36+
})
37+
38+
await new Promise<void>((resolve, reject) => {
39+
worker.once('message', () => resolve())
40+
worker.once('error', reject)
41+
})
42+
43+
return async () => {
44+
Atomics.store(new Int32Array(stop), 0, 1)
45+
Atomics.notify(new Int32Array(stop), 0)
46+
await new Promise((resolve) => worker.once('exit', resolve))
47+
}
48+
}

‎tests/adapter.spec.ts‎

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import { withKnexQuerySpy } from './_utils/with_knex_query_spy.js'
1717
import { pauseKnexDeletes } from './_utils/pause_deletes.js'
1818
import { registerRetentionRaceSuite } from './_utils/register_retention_race_suite.js'
1919
import { MemoryLogger } from './_mocks/memory_logger.js'
20+
import { startSqliteWriteContender } from './_utils/sqlite_write_contender.js'
2021

2122
const KEY_PREFIX = 'boringnode::queue::test::'
2223

@@ -2660,10 +2661,11 @@ test.group('Adapter | Knex (MySQL)', (group) => {
26602661
test.group('Adapter | Knex (SQLite file, two connections)', () => {
26612662
const connectTwice = async (cleanup: (handler: () => Promise<void>) => void) => {
26622663
const directory = await mkdtemp(join(tmpdir(), 'queue-sqlite-'))
2664+
const filename = join(directory, 'queue.sqlite')
26632665
const connect = () =>
26642666
Knex({
26652667
client: 'better-sqlite3',
2666-
connection: { filename: join(directory, 'queue.sqlite') },
2668+
connection: { filename },
26672669
useNullAsDefault: true,
26682670
pool: { afterCreate: (db: any, done: any) => done(null, db.pragma('journal_mode = WAL')) },
26692671
})
@@ -2674,9 +2676,37 @@ test.group('Adapter | Knex (SQLite file, two connections)', () => {
26742676
await second.destroy()
26752677
await rm(directory, { recursive: true, force: true })
26762678
})
2677-
return [first, second] as const
2679+
return Object.assign([first, second] as const, { filename })
26782680
}
26792681

2682+
test('popFrom and dedup pushes wait for a write from another process', async ({
2683+
assert,
2684+
cleanup,
2685+
}) => {
2686+
const connections = await connectTwice(cleanup)
2687+
await new KnexQueueSchemaService(connections[0]).createJobsTable()
2688+
const adapter = new KnexAdapter({ connection: connections[0] })
2689+
adapter.setWorkerId('worker-1')
2690+
2691+
const stopContender = await startSqliteWriteContender(connections.filename)
2692+
const outcomes: string[] = []
2693+
try {
2694+
for (let i = 0; i < 20; i++) {
2695+
const job = { id: `job-${i}`, name: 'TestJob', payload: {}, attempts: 0 }
2696+
for (let dispatch = 0; dispatch < 2; dispatch++) {
2697+
const result = await adapter.pushOn('test-queue', { ...job, dedup: { id: `dedup-${i}` } })
2698+
outcomes.push(result?.outcome ?? 'none')
2699+
}
2700+
await adapter.pushLaterOn('test-queue', { ...job, id: `delayed-${i}` }, 0)
2701+
assert.isNotNull(await adapter.popFrom('test-queue'))
2702+
}
2703+
} finally {
2704+
await stopContender()
2705+
}
2706+
2707+
assert.deepEqual(outcomes, Array.from({ length: 20 }, () => ['added', 'skipped']).flat())
2708+
})
2709+
26802710
registerRetentionRaceSuite({
26812711
test,
26822712
createAdapters: async (beforeDelete, cleanup) => {

‎tests/kysely_adapter.spec.ts‎

Lines changed: 32 additions & 2 deletions

0 commit comments

Comments
 (0)