Skip to content

Commit a404106

Browse files
isimisiRomainLanz
andauthored
perf!(redis): use sorted set index for O(log N) schedule claiming (#14)
* perf(redis): use sorted set index for O(log N) schedule claiming * fix(redis): auto-backfill due index on first schedule claim Existing users upgrading will have schedules in the legacy format (hashes + SET) but not in the new ZSET. Run backfillDueIndex() once per worker process on the first claimDueSchedule() call so schedules keep firing without manual intervention. * perf(redis): update claim schedule script to use sorted set index * feat(adapter): add migrate() to interface, validate hash in claim script Replace auto-backfill with an explicit migrate() lifecycle method on the Adapter interface. Remove #ensureDueIndex() and #dueIndexReady from RedisAdapter — users call migrate() once after upgrading to populate the schedules::due ZSET from pre-existing data. The Lua claim script now validates the hash's next_run_at before claiming, repairing stale ZSET scores on sight. This keeps the hash canonical and the ZSET as a derived index. * fix(redis): harden due schedule index migration Keep paused and exhausted schedules out of the derived due index, rebuild it idempotently during migration, and restore the Adapter contract after merging current main. Document the breaking Redis migration and cover schedule lifecycle consistency across adapters. * chore: remove released changelog entries * docs: add Redis schedule index changelog * fix(redis): make schedule index updates atomic Move schedule upserts, updates, and due-index backfills into Lua so canonical hashes and the derived ZSET cannot diverge under concurrent writes. Add regression coverage for resume, upsert, and migration interleavings. * fix(redis): guard cron claim finalization Finalize cron schedules only when the claimed canonical state is still current, preventing concurrent pause or delete operations from being undone. Repair malformed due scores during claiming so they cannot block valid schedules. * fix(redis): reject superseded cron finalization * style(redis): clarify claim script nesting * docs: note Redis schedule race safeguards * fix(redis): preserve schedules across concurrent updates * fix(redis): guard cron claim ownership * fix(redis): preserve cron next run during concurrent pause --------- Co-authored-by: Romain Lanz <romain.lanz@pm.me>
1 parent 52b38f0 commit a404106

16 files changed

Lines changed: 1145 additions & 184 deletions

‎.changelog/adapter-aware-workers-and-schedules.md‎

Lines changed: 0 additions & 68 deletions
This file was deleted.

‎.changelog/consistent-job-runtime.md‎

Lines changed: 0 additions & 30 deletions
This file was deleted.

‎.changelog/hot-reloading-jobs.md‎

Lines changed: 0 additions & 44 deletions
This file was deleted.
Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
# Indexed Redis Schedule Claims
2+
3+
## Performance Improvement
4+
5+
The Redis adapter now maintains a `schedules::due` sorted-set index scored by each schedule's
6+
`next_run_at`. Claiming the next due schedule uses this index instead of scanning every stored
7+
schedule, so polling no longer grows linearly with the total schedule count.
8+
9+
Schedule hashes remain the source of truth. Creating, updating, pausing, resuming, deleting, and
10+
claiming schedules maintain the derived index, while claiming repairs stale entries when the hash
11+
and index disagree.
12+
13+
Schedule hash and index writes are atomic, and index consistency is preserved across concurrent
14+
lifecycle changes. Cron finalization now rejects stale calculations when a schedule is deleted or
15+
reconfigured while its next run is being calculated. Paused schedules retain their next run but stay
16+
out of the due index until resumed. Claiming also discards malformed due scores instead of allowing
17+
one corrupt entry to block later schedules.
18+
19+
## Upgrade Notes
20+
21+
This change requires an explicit migration for existing Redis schedules. The `Adapter` contract now
22+
includes an idempotent `migrate()` method; built-in adapters without migrations implement it as a
23+
no-op, and custom adapters must implement it as well.
24+
25+
Run the migration once during deployment, before starting workers or any process that creates or
26+
updates schedules:
27+
28+
```typescript
29+
await QueueManager.init(config)
30+
await QueueManager.use('redis').migrate()
31+
```
32+
33+
Existing Redis schedules will not fire through the new index until the migration has completed.
34+
The migration scans all schedules and should remain an explicit deployment step rather than run in
35+
the worker polling loop.

‎README.md‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -334,6 +334,26 @@ const connection = new Redis({ host: 'localhost' })
334334
const adapter = redis(connection)
335335
```
336336

337+
#### Migrating Redis schedules after an upgrade
338+
339+
The Redis adapter uses a `schedules::due` sorted-set index to find due schedules. When upgrading
340+
from a version that predates this index, run the adapter migration once during deployment, before
341+
starting any workers:
342+
343+
```typescript
344+
import { QueueManager, Worker } from '@boringnode/queue'
345+
346+
await QueueManager.init(config)
347+
await QueueManager.use('redis').migrate()
348+
349+
const worker = new Worker(config)
350+
await worker.start(['default'])
351+
```
352+
353+
The migration is idempotent and rebuilds the derived index from the canonical schedule hashes.
354+
Existing Redis schedules will not fire through the new index until it has run. Do not run the
355+
`O(number of schedules)` migration from the worker polling loop.
356+
337357
### Knex (PostgreSQL, MySQL, SQLite)
338358

339359
```typescript

‎src/contracts/adapter.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -223,6 +223,16 @@ export interface Adapter {
223223
*/
224224
destroy(): Promise<void>
225225

226+
/**
227+
* Run adapter-specific migrations needed after a major version upgrade.
228+
*
229+
* This method is idempotent — it is always safe to call multiple times.
230+
* Adapters that have no pending migrations return immediately.
231+
*
232+
* Call this once during your deployment process before starting workers.
233+
*/
234+
migrate(): Promise<void>
235+
226236
/**
227237
* Create or update a schedule.
228238
*

‎src/drivers/fake_adapter.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -404,6 +404,10 @@ export class FakeAdapter implements Adapter {
404404
return Promise.resolve()
405405
}
406406

407+
migrate(): Promise<void> {
408+
return Promise.resolve()
409+
}
410+
407411
async upsertSchedule(config: ScheduleConfig): Promise<string> {
408412
const id = config.id ?? randomUUID()
409413
const existing = this.#schedules.get(id)

‎src/drivers/knex_adapter.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,8 @@ export class KnexAdapter implements Adapter {
8888
}
8989
}
9090

91+
async migrate(): Promise<void> {}
92+
9193
async pop(): Promise<AcquiredJob | null> {
9294
return this.popFrom('default')
9395
}

‎src/drivers/kysely_adapter.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,8 @@ export class KyselyAdapter<DB = QueueDatabase> implements Adapter {
122122
// The Kysely instance is always owned by the application.
123123
}
124124

125+
async migrate(): Promise<void> {}
126+
125127
async pop(): Promise<AcquiredJob | null> {
126128
return this.popFrom('default')
127129
}

0 commit comments

Comments
 (0)