From 812af117dc3c00ea7afaf0a0cac07bf0479a3725 Mon Sep 17 00:00:00 2001 From: Isaac <91521821+isimisi@users.noreply.github.com> Date: Fri, 15 May 2026 10:41:30 +0200 Subject: [PATCH 01/14] perf(redis): use sorted set index for O(log N) schedule claiming --- src/drivers/redis_adapter.ts | 214 ++++++++++++++++++++++++----------- 1 file changed, 148 insertions(+), 66 deletions(-) diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 27fe11d..12e6ba8 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -16,6 +16,7 @@ import { resolveRetention } from '../utils.js' const redisKey = 'jobs' const schedulesKey = 'schedules' const schedulesIndexKey = 'schedules::index' +const schedulesDueKey = 'schedules::due' type RedisConfig = Redis | RedisOptions /** @@ -352,76 +353,98 @@ const GET_JOB_SCRIPT = ` ` /** - * Lua script for atomically claiming a due schedule. - * Iterates the schedule index server-side and claims the first due schedule. - * Returns the schedule data if claimed, nil otherwise. + * Lua script for atomically claiming a due schedule using a sorted set index. + * + * Uses ZRANGEBYSCORE on schedules::due (scored by next_run_at) for O(log N) + * lookup instead of scanning all schedule hashes via SMEMBERS. + * + * Stale entries (paused, exhausted, deleted) are cleaned from the ZSET on + * sight so subsequent calls skip them. + * + * KEYS[1] = schedules::due (the ZSET) + * KEYS[2] = schedule key prefix (e.g. "schedules::") + * ARGV[1] = now (epoch milliseconds) */ const CLAIM_SCHEDULE_SCRIPT = ` - local schedules_index_key = KEYS[1] - local schedule_key_prefix = KEYS[2] + local due_key = KEYS[1] + local prefix = KEYS[2] local now = tonumber(ARGV[1]) - local ids = redis.call('SMEMBERS', schedules_index_key) + while true do + local candidates = redis.call('ZRANGEBYSCORE', due_key, '-inf', tostring(now), 'LIMIT', 0, 1) - for i = 1, #ids do - local schedule_key = schedule_key_prefix .. ids[i] + if #candidates == 0 then + return nil + end + + local id = candidates[1] + local schedule_key = prefix .. id -- Get schedule data local data = redis.call('HGETALL', schedule_key) - if #data > 0 then + + -- Deleted schedule still in ZSET + if #data == 0 then + redis.call('ZREM', due_key, id) + else -- Convert HGETALL result to table local schedule = {} for j = 1, #data, 2 do schedule[data[j]] = data[j + 1] end - -- Check if schedule is due - if schedule.status == 'active' then - local next_run_at = tonumber(schedule.next_run_at) - - if next_run_at and next_run_at <= now then - local run_count = tonumber(schedule.run_count or '0') - local run_limit = schedule.run_limit and tonumber(schedule.run_limit) or nil - local to_date = schedule.to_date and tonumber(schedule.to_date) or nil - - -- Check limits - if not (run_limit and run_count >= run_limit) and not (to_date and now > to_date) then - -- This schedule is claimable - atomically update it - local new_run_count = run_count + 1 - - -- Calculate new next_run_at (simple interval-based for now) - -- Complex cron calculation happens in the caller - local new_next_run_at = '' - local every_ms = schedule.every_ms and tonumber(schedule.every_ms) or nil - if every_ms then - new_next_run_at = tostring(now + every_ms) - end - - -- Check if we've hit the limit after this run - if run_limit and new_run_count >= run_limit then - new_next_run_at = '' - end - - -- Check if past end date - if to_date and new_next_run_at ~= '' and tonumber(new_next_run_at) > to_date then - new_next_run_at = '' - end - - -- Update the schedule atomically - redis.call('HSET', schedule_key, - 'next_run_at', new_next_run_at, - 'last_run_at', tostring(now), - 'run_count', tostring(new_run_count)) - - -- Return the schedule data (before update) as JSON - return cjson.encode(schedule) + -- Check if schedule is active + if schedule.status ~= 'active' then + redis.call('ZREM', due_key, id) + else + local run_count = tonumber(schedule.run_count or '0') + local run_limit = schedule.run_limit and tonumber(schedule.run_limit) or nil + local to_date = schedule.to_date and tonumber(schedule.to_date) or nil + + -- Check limits + if (run_limit and run_count >= run_limit) or (to_date and now > to_date) then + redis.call('ZREM', due_key, id) + else + -- This schedule is claimable - atomically update it + local new_run_count = run_count + 1 + + -- Calculate new next_run_at (simple interval-based for now) + -- Complex cron calculation happens in the caller + local new_next_run_at = '' + local every_ms = schedule.every_ms and tonumber(schedule.every_ms) or nil + if every_ms then + new_next_run_at = tostring(now + every_ms) + end + + -- Check if we've hit the limit after this run + if run_limit and new_run_count >= run_limit then + new_next_run_at = '' + end + + -- Check if past end date + if to_date and new_next_run_at ~= '' and tonumber(new_next_run_at) > to_date then + new_next_run_at = '' end + + -- Update the schedule atomically + redis.call('HSET', schedule_key, + 'next_run_at', new_next_run_at, + 'last_run_at', tostring(now), + 'run_count', tostring(new_run_count)) + + -- Update or remove from ZSET + if new_next_run_at ~= '' then + redis.call('ZADD', due_key, tonumber(new_next_run_at), id) + else + redis.call('ZREM', due_key, id) + end + + -- Return the schedule data (before update) as JSON + return cjson.encode(schedule) end end end end - - return nil ` /** @@ -700,10 +723,11 @@ export class RedisAdapter implements Adapter { const id = config.id ?? randomUUID() const now = Date.now() const scheduleKey = `${schedulesKey}::${id}` - const [existingRunCount, existingCreatedAt] = await this.#connection.hmget( + const [existingRunCount, existingCreatedAt, existingNextRunAt] = await this.#connection.hmget( scheduleKey, 'run_count', - 'created_at' + 'created_at', + 'next_run_at' ) const scheduleData: Record = { @@ -722,13 +746,17 @@ export class RedisAdapter implements Adapter { if (config.to !== undefined) scheduleData.to_date = config.to.getTime().toString() if (config.limit !== undefined) scheduleData.run_limit = config.limit.toString() - // Upsert schedule and clear stale optional fields from previous config. - await this.#connection + const multi = this.#connection .multi() .hdel(scheduleKey, 'cron_expression', 'every_ms', 'from_date', 'to_date', 'run_limit') .hset(scheduleKey, scheduleData) .sadd(schedulesIndexKey, id) - .exec() + + if (existingNextRunAt) { + multi.zadd(schedulesDueKey, Number.parseInt(existingNextRunAt, 10), id) + } + + await multi.exec() return id } @@ -804,14 +832,34 @@ export class RedisAdapter implements Adapter { } if (updates.runCount !== undefined) data.run_count = updates.runCount.toString() - if (Object.keys(data).length > 0) { - await this.#connection.hset(scheduleKey, data) + if (Object.keys(data).length === 0) return + + const multi = this.#connection.multi().hset(scheduleKey, data) + + if (updates.nextRunAt) { + multi.zadd(schedulesDueKey, updates.nextRunAt.getTime(), id) + } else if (updates.nextRunAt === null || updates.status === 'paused') { + multi.zrem(schedulesDueKey, id) + } + + if (updates.status === 'active' && updates.nextRunAt === undefined) { + const existing = await this.#connection.hget(scheduleKey, 'next_run_at') + if (existing) { + multi.zadd(schedulesDueKey, Number.parseInt(existing, 10), id) + } } + + await multi.exec() } async deleteSchedule(id: string): Promise { const scheduleKey = `${schedulesKey}::${id}` - await this.#connection.multi().del(scheduleKey).srem(schedulesIndexKey, id).exec() + await this.#connection + .multi() + .del(scheduleKey) + .srem(schedulesIndexKey, id) + .zrem(schedulesDueKey, id) + .exec() } async claimDueSchedule(): Promise { @@ -819,7 +867,7 @@ export class RedisAdapter implements Adapter { const result = await this.#connection.eval( CLAIM_SCHEDULE_SCRIPT, 2, - schedulesIndexKey, + schedulesDueKey, `${schedulesKey}::`, now.toString() ) @@ -841,7 +889,6 @@ export class RedisAdapter implements Adapter { }) const nextRun = cron.next().toDate().getTime() - // Check limits before updating const runCount = Number.parseInt(data.run_count || '0', 10) + 1 const runLimit = data.run_limit ? Number.parseInt(data.run_limit, 10) : null const toDate = data.to_date ? Number.parseInt(data.to_date, 10) : null @@ -854,16 +901,51 @@ export class RedisAdapter implements Adapter { newNextRunAt = '' } - await this.#connection.hset( - `${schedulesKey}::${data.id}`, - 'next_run_at', - newNextRunAt.toString() - ) + const scheduleKey = `${schedulesKey}::${data.id}` + const multi = this.#connection + .multi() + .hset(scheduleKey, 'next_run_at', newNextRunAt.toString()) + + if (typeof newNextRunAt === 'number') { + multi.zadd(schedulesDueKey, newNextRunAt, data.id) + } else { + multi.zrem(schedulesDueKey, data.id) + } + + await multi.exec() } return this.#hashToScheduleData(data) } + async backfillDueIndex(): Promise { + const ids = await this.#connection.smembers(schedulesIndexKey) + if (ids.length === 0) return 0 + + const pipeline = this.#connection.pipeline() + for (const id of ids) { + pipeline.hmget(`${schedulesKey}::${id}`, 'next_run_at', 'status') + } + const results = await pipeline.exec() + if (!results) return 0 + + const addPipeline = this.#connection.pipeline() + let count = 0 + + for (let i = 0; i < ids.length; i++) { + const [err, values] = results[i] + if (err || !values) continue + const [nextRunAt, status] = values as [string | null, string | null] + if (nextRunAt && status === 'active') { + addPipeline.zadd(schedulesDueKey, Number.parseInt(nextRunAt, 10), ids[i]) + count++ + } + } + + if (count > 0) await addPipeline.exec() + return count + } + #hashToScheduleData(data: Record): ScheduleData { return { id: data.id, From dd967dba82cf88a689295a6e1567da958e62fd16 Mon Sep 17 00:00:00 2001 From: Isaac <91521821+isimisi@users.noreply.github.com> Date: Tue, 2 Jun 2026 13:53:43 +0200 Subject: [PATCH 02/14] 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. --- src/drivers/redis_adapter.ts | 9 +++++++++ tests/adapter.spec.ts | 3 +++ 2 files changed, 12 insertions(+) diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 12e6ba8..509ce5c 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -480,6 +480,7 @@ export class RedisAdapter implements Adapter { readonly #connection: Redis readonly #ownsConnection: boolean #workerId: string = '' + #dueIndexReady = false constructor(connection: Redis, ownsConnection: boolean = false) { this.#connection = connection @@ -862,7 +863,15 @@ export class RedisAdapter implements Adapter { .exec() } + async #ensureDueIndex(): Promise { + if (this.#dueIndexReady) return + await this.backfillDueIndex() + this.#dueIndexReady = true + } + async claimDueSchedule(): Promise { + await this.#ensureDueIndex() + const now = Date.now() const result = await this.#connection.eval( CLAIM_SCHEDULE_SCRIPT, diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index a276625..869dc75 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -99,6 +99,9 @@ test.group('Adapter | Redis', (group) => { await adapter.updateSchedule(id, { nextRunAt: futureRunAt }) } + // Warm the due-index backfill so it doesn't count against the spy + await adapter.claimDueSchedule() + const { result: claimed, writes } = await withRedisWriteSpy({ connection, run: () => adapter.claimDueSchedule(), From f3c85cd18b8e37575da976cba3d6d2f484454cdf Mon Sep 17 00:00:00 2001 From: Isaac <91521821+isimisi@users.noreply.github.com> Date: Wed, 3 Jun 2026 12:29:12 +0200 Subject: [PATCH 03/14] perf(redis): update claim schedule script to use sorted set index --- src/drivers/redis_scripts.ts | 140 ++++++++++++++++++++--------------- 1 file changed, 81 insertions(+), 59 deletions(-) diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index e0ff6dd..2f8cedf 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -1,4 +1,4 @@ -import { REDIS_DEDUP_LUA, REDIS_JOB_STORAGE_LUA } from './redis_job_storage.js' +import { REDIS_DEDUP_LUA, REDIS_JOB_STORAGE_LUA } from './redis_job_storage.js'; /** * Lua script for pushing a job to the queue. @@ -18,7 +18,7 @@ ${REDIS_JOB_STORAGE_LUA} redis.call('ZADD', pending_key, score, job_id) return 1 -` +`; /** * Lua script for pushing a dedup job. @@ -80,7 +80,7 @@ ${REDIS_DEDUP_LUA} redis.call('PEXPIRE', dedup_key, ttl) end return {'added', job_id} -` +`; /** * Lua script for pushing a delayed job. @@ -100,7 +100,7 @@ ${REDIS_JOB_STORAGE_LUA} redis.call('ZADD', delayed_key, execute_at, job_id) return 1 -` +`; /** * Lua script for atomic job acquisition. @@ -158,7 +158,7 @@ ${REDIS_JOB_STORAGE_LUA} return encode_job_result(job_data, overlay_key, job_id, { acquiredAt = now }) -` +`; /** * Lua script for removing a job completely (no history). @@ -193,7 +193,7 @@ ${REDIS_JOB_STORAGE_LUA} delete_job_data(data_key, overlay_key, job_id) return 1 -` +`; /** * Lua script for finalizing a job in history. @@ -277,7 +277,7 @@ ${REDIS_JOB_STORAGE_LUA} end return 1 -` +`; /** * Lua script for retrying a job. @@ -330,7 +330,7 @@ ${REDIS_JOB_STORAGE_LUA} end return 1 -` +`; /** * Lua script for recovering stalled jobs. @@ -399,7 +399,7 @@ ${REDIS_JOB_STORAGE_LUA} end return recovered -` +`; /** * Lua script for getting a job record with its status. @@ -458,77 +458,99 @@ ${REDIS_JOB_STORAGE_LUA} finishedAt = finished_at, error = error_msg }) -` +`; /** - * Lua script for atomically claiming a due schedule. - * Iterates the schedule index server-side and claims the first due schedule. - * Returns the schedule data if claimed, nil otherwise. + * Lua script for atomically claiming a due schedule using a sorted set index. + * + * Uses ZRANGEBYSCORE on schedules::due (scored by next_run_at) for O(log N) + * lookup instead of scanning all schedule hashes via SMEMBERS. + * + * Stale entries (paused, exhausted, deleted) are cleaned from the ZSET on + * sight so subsequent calls skip them. + * + * KEYS[1] = schedules::due (the ZSET) + * KEYS[2] = schedule key prefix (e.g. "schedules::") + * ARGV[1] = now (epoch milliseconds) */ export const CLAIM_SCHEDULE_SCRIPT = ` - local schedules_index_key = KEYS[1] - local schedule_key_prefix = KEYS[2] + local due_key = KEYS[1] + local prefix = KEYS[2] local now = tonumber(ARGV[1]) - local ids = redis.call('SMEMBERS', schedules_index_key) + while true do + local candidates = redis.call('ZRANGEBYSCORE', due_key, '-inf', tostring(now), 'LIMIT', 0, 1) + + if #candidates == 0 then + return nil + end - for i = 1, #ids do - local schedule_key = schedule_key_prefix .. ids[i] + local id = candidates[1] + local schedule_key = prefix .. id -- Get schedule data local data = redis.call('HGETALL', schedule_key) - if #data > 0 then + + -- Deleted schedule still in ZSET + if #data == 0 then + redis.call('ZREM', due_key, id) + else -- Convert HGETALL result to table local schedule = {} for j = 1, #data, 2 do schedule[data[j]] = data[j + 1] end - -- Check if schedule is due - if schedule.status == 'active' then - local next_run_at = tonumber(schedule.next_run_at) - - if next_run_at and next_run_at <= now then - local run_count = tonumber(schedule.run_count or '0') - local run_limit = schedule.run_limit and tonumber(schedule.run_limit) or nil - local to_date = schedule.to_date and tonumber(schedule.to_date) or nil - - -- Check limits - if not (run_limit and run_count >= run_limit) and not (to_date and now > to_date) then - -- This schedule is claimable - atomically update it - local new_run_count = run_count + 1 - - -- Calculate new next_run_at (simple interval-based for now) - -- Complex cron calculation happens in the caller - local new_next_run_at = '' - local every_ms = schedule.every_ms and tonumber(schedule.every_ms) or nil - if every_ms then - new_next_run_at = tostring(now + every_ms) - end - - -- Check if we've hit the limit after this run - if run_limit and new_run_count >= run_limit then - new_next_run_at = '' - end + -- Check if schedule is active + if schedule.status ~= 'active' then + redis.call('ZREM', due_key, id) + else + local run_count = tonumber(schedule.run_count or '0') + local run_limit = schedule.run_limit and tonumber(schedule.run_limit) or nil + local to_date = schedule.to_date and tonumber(schedule.to_date) or nil + + -- Check limits + if (run_limit and run_count >= run_limit) or (to_date and now > to_date) then + redis.call('ZREM', due_key, id) + else + -- This schedule is claimable - atomically update it + local new_run_count = run_count + 1 + + -- Calculate new next_run_at (simple interval-based for now) + -- Complex cron calculation happens in the caller + local new_next_run_at = '' + local every_ms = schedule.every_ms and tonumber(schedule.every_ms) or nil + if every_ms then + new_next_run_at = tostring(now + every_ms) + end - -- Check if past end date - if to_date and new_next_run_at ~= '' and tonumber(new_next_run_at) > to_date then - new_next_run_at = '' - end + -- Check if we've hit the limit after this run + if run_limit and new_run_count >= run_limit then + new_next_run_at = '' + end - -- Update the schedule atomically - redis.call('HSET', schedule_key, - 'next_run_at', new_next_run_at, - 'last_run_at', tostring(now), - 'run_count', tostring(new_run_count)) + -- Check if past end date + if to_date and new_next_run_at ~= '' and tonumber(new_next_run_at) > to_date then + new_next_run_at = '' + end - -- Return the schedule data (before update) as JSON - return cjson.encode(schedule) + -- Update the schedule atomically + redis.call('HSET', schedule_key, + 'next_run_at', new_next_run_at, + 'last_run_at', tostring(now), + 'run_count', tostring(new_run_count)) + + -- Update or remove from ZSET + if new_next_run_at ~= '' then + redis.call('ZADD', due_key, tonumber(new_next_run_at), id) + else + redis.call('ZREM', due_key, id) end + + -- Return the schedule data (before update) as JSON + return cjson.encode(schedule) end end end end - - return nil -` +`; From 408b18e3bc6ebffbdba2b6596a42666050c91455 Mon Sep 17 00:00:00 2001 From: Isaac <91521821+isimisi@users.noreply.github.com> Date: Mon, 8 Jun 2026 10:07:00 +0200 Subject: [PATCH 04/14] feat(adapter): add migrate() to interface, validate hash in claim script MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/contracts/adapter.ts | 10 + src/drivers/fake_adapter.ts | 4 + src/drivers/knex_adapter.ts | 2 + src/drivers/redis_adapter.ts | 1312 +++++++++++++++----------------- src/drivers/redis_scripts.ts | 31 +- src/drivers/sync_adapter.ts | 4 + tests/_mocks/memory_adapter.ts | 4 + tests/adapter.spec.ts | 89 ++- 8 files changed, 753 insertions(+), 703 deletions(-) diff --git a/src/contracts/adapter.ts b/src/contracts/adapter.ts index 8bd8e12..c7a7fdd 100644 --- a/src/contracts/adapter.ts +++ b/src/contracts/adapter.ts @@ -205,6 +205,16 @@ export interface Adapter { */ destroy(): Promise + /** + * Run adapter-specific migrations needed after a major version upgrade. + * + * This method is idempotent — it is always safe to call multiple times. + * Adapters that have no pending migrations return immediately. + * + * Call this once during your deployment process before starting workers. + */ + migrate(): Promise + /** * Create or update a schedule. * diff --git a/src/drivers/fake_adapter.ts b/src/drivers/fake_adapter.ts index 3671cc4..1b98825 100644 --- a/src/drivers/fake_adapter.ts +++ b/src/drivers/fake_adapter.ts @@ -385,6 +385,10 @@ export class FakeAdapter implements Adapter { return Promise.resolve() } + migrate(): Promise { + return Promise.resolve() + } + async upsertSchedule(config: ScheduleConfig): Promise { const id = config.id ?? randomUUID() const existing = this.#schedules.get(id) diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index adeb610..7c43b03 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -88,6 +88,8 @@ export class KnexAdapter implements Adapter { } } + async migrate(): Promise {} + async pop(): Promise { return this.popFrom('default') } diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 8f384a6..178bb2c 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -1,40 +1,37 @@ -import { randomUUID } from 'node:crypto'; -import { Redis, type RedisOptions } from 'ioredis'; -import { DEFAULT_PRIORITY } from '../constants.js'; -import { calculateScore } from '../utils.js'; -import type { Adapter, AcquiredJob, PushResult } from '../contracts/adapter.js'; -import type { DedupOutcome } from '../types/main.js'; +import { randomUUID } from 'node:crypto' +import { Redis, type RedisOptions } from 'ioredis' +import { DEFAULT_PRIORITY } from '../constants.js' +import { calculateScore } from '../utils.js' +import type { Adapter, AcquiredJob, PushResult } from '../contracts/adapter.js' +import type { DedupOutcome } from '../types/main.js' import type { - JobData, - JobRecord, - JobRetention, - ScheduleConfig, - ScheduleData, - ScheduleListOptions, -} from '../types/main.js'; -import { resolveRetention } from '../utils.js'; + JobData, + JobRecord, + JobRetention, + ScheduleConfig, + ScheduleData, + ScheduleListOptions, +} from '../types/main.js' +import { resolveRetention } from '../utils.js' +import { encodeRedisJobPayloadOverlay, hydrateRedisJob } from './redis_job_storage.js' import { - encodeRedisJobPayloadOverlay, - hydrateRedisJob, -} from './redis_job_storage.js'; -import { - ACQUIRE_JOB_SCRIPT, - CLAIM_SCHEDULE_SCRIPT, - FINALIZE_JOB_SCRIPT, - GET_JOB_SCRIPT, - PUSH_DEDUP_JOB_SCRIPT, - PUSH_DELAYED_JOB_SCRIPT, - PUSH_JOB_SCRIPT, - RECOVER_STALLED_JOBS_SCRIPT, - REMOVE_JOB_SCRIPT, - RETRY_JOB_SCRIPT, -} from './redis_scripts.js'; - -const redisKey = 'jobs'; -const schedulesKey = 'schedules'; -const schedulesIndexKey = 'schedules::index'; -const schedulesDueKey = 'schedules::due'; -type RedisConfig = Redis | RedisOptions; + ACQUIRE_JOB_SCRIPT, + CLAIM_SCHEDULE_SCRIPT, + FINALIZE_JOB_SCRIPT, + GET_JOB_SCRIPT, + PUSH_DEDUP_JOB_SCRIPT, + PUSH_DELAYED_JOB_SCRIPT, + PUSH_JOB_SCRIPT, + RECOVER_STALLED_JOBS_SCRIPT, + REMOVE_JOB_SCRIPT, + RETRY_JOB_SCRIPT, +} from './redis_scripts.js' + +const redisKey = 'jobs' +const schedulesKey = 'schedules' +const schedulesIndexKey = 'schedules::index' +const schedulesDueKey = 'schedules::due' +type RedisConfig = Redis | RedisOptions /** * Create a new Redis adapter factory. @@ -47,676 +44,613 @@ type RedisConfig = Redis | RedisOptions; * managing the connection lifecycle. */ export function redis(config?: RedisConfig) { - return () => { - if (config instanceof Redis) { - return new RedisAdapter(config, false); - } - - const options: RedisOptions = { - host: 'localhost', - port: 6379, - keyPrefix: 'boringnode::queue::', - db: 0, - ...config, - }; - - const connection = new Redis(options); - return new RedisAdapter(connection, true); - }; + return () => { + if (config instanceof Redis) { + return new RedisAdapter(config, false) + } + + const options: RedisOptions = { + host: 'localhost', + port: 6379, + keyPrefix: 'boringnode::queue::', + db: 0, + ...config, + } + + const connection = new Redis(options) + return new RedisAdapter(connection, true) + } } export class RedisAdapter implements Adapter { - readonly #connection: Redis; - readonly #ownsConnection: boolean; - #workerId: string = ''; - #dueIndexReady = false; - - constructor(connection: Redis, ownsConnection: boolean = false) { - this.#connection = connection; - this.#ownsConnection = ownsConnection; - } - - #getKeys(queue: string) { - return { - data: `${redisKey}::${queue}::data`, - pending: `${redisKey}::${queue}::pending`, - delayed: `${redisKey}::${queue}::delayed`, - active: `${redisKey}::${queue}::active`, - overlay: `${redisKey}::${queue}::metadata`, - completed: `${redisKey}::${queue}::completed`, - completedIndex: `${redisKey}::${queue}::completed::index`, - failed: `${redisKey}::${queue}::failed`, - failedIndex: `${redisKey}::${queue}::failed::index`, - }; - } - - #getDedupKey(queue: string, dedupId: string): string { - return `${this.#getDedupPrefix(queue)}${dedupId}`; - } - - #getDedupPrefix(queue: string): string { - return `${redisKey}::${queue}::dedup::`; - } - - setWorkerId(workerId: string): void { - this.#workerId = workerId; - } - - async destroy(): Promise { - if (this.#ownsConnection) { - await this.#connection.quit(); - } - } - - pop(): Promise { - return this.popFrom('default'); - } - - async popFrom(queue: string): Promise { - const keys = this.#getKeys(queue); - const now = Date.now(); - - const result = await this.#connection.eval( - ACQUIRE_JOB_SCRIPT, - 5, - keys.data, - keys.pending, - keys.active, - keys.delayed, - keys.overlay, - this.#workerId, - now.toString(), - ); - - if (!result) { - return null; - } - - const { data, overlay, acquiredAt } = JSON.parse(result as string) as { - data: string; - overlay?: string; - acquiredAt: number; - }; - - return { ...hydrateRedisJob(data, overlay), acquiredAt }; - } - - async completeJob( - jobId: string, - queue: string, - removeOnComplete?: JobRetention, - ): Promise { - const keys = this.#getKeys(queue); - const dedupPrefix = this.#getDedupPrefix(queue); - const { keep, maxAge, maxCount } = resolveRetention(removeOnComplete); - - if (!keep) { - await this.#connection.eval( - REMOVE_JOB_SCRIPT, - 3, - keys.data, - keys.active, - keys.overlay, - jobId, - dedupPrefix, - ); - return; - } - - await this.#connection.eval( - FINALIZE_JOB_SCRIPT, - 5, - keys.data, - keys.active, - keys.completed, - keys.completedIndex, - keys.overlay, - jobId, - Date.now().toString(), - maxAge.toString(), - maxCount.toString(), - '', - dedupPrefix, - ); - } - - async failJob( - jobId: string, - queue: string, - error?: Error, - removeOnFail?: JobRetention, - ): Promise { - const keys = this.#getKeys(queue); - const dedupPrefix = this.#getDedupPrefix(queue); - const { keep, maxAge, maxCount } = resolveRetention(removeOnFail); - - if (!keep) { - await this.#connection.eval( - REMOVE_JOB_SCRIPT, - 3, - keys.data, - keys.active, - keys.overlay, - jobId, - dedupPrefix, - ); - return; - } - - await this.#connection.eval( - FINALIZE_JOB_SCRIPT, - 5, - keys.data, - keys.active, - keys.failed, - keys.failedIndex, - keys.overlay, - jobId, - Date.now().toString(), - maxAge.toString(), - maxCount.toString(), - error?.message || '', - dedupPrefix, - ); - } - - async retryJob(jobId: string, queue: string, retryAt?: Date): Promise { - const keys = this.#getKeys(queue); - const now = Date.now(); - - await this.#connection.eval( - RETRY_JOB_SCRIPT, - 5, - keys.data, - keys.active, - keys.pending, - keys.delayed, - keys.overlay, - jobId, - retryAt ? retryAt.getTime().toString() : '0', - now.toString(), - ); - } - - async getJob(jobId: string, queue: string): Promise { - const keys = this.#getKeys(queue); - - const result = await this.#connection.eval( - GET_JOB_SCRIPT, - 7, - keys.data, - keys.pending, - keys.delayed, - keys.active, - keys.completed, - keys.failed, - keys.overlay, - jobId, - ); - - if (!result) { - return null; - } - - const record = JSON.parse(result as string) as Omit & { - data: string; - overlay?: string; - }; - - return { ...record, data: hydrateRedisJob(record.data, record.overlay) }; - } - - push(jobData: JobData): Promise { - return this.pushOn('default', jobData); - } - - pushLater(jobData: JobData, delay: number): Promise { - return this.pushLaterOn('default', jobData, delay); - } - - async pushLaterOn( - queue: string, - jobData: JobData, - delay: number, - ): Promise { - const keys = this.#getKeys(queue); - const executeAt = Date.now() + delay; - - if (jobData.dedup) { - const dedupKey = this.#getDedupKey(queue, jobData.dedup.id); - const [payloadData, payloadIsUndefined] = encodeRedisJobPayloadOverlay( - jobData.payload, - ); - const result = (await this.#connection.eval( - PUSH_DEDUP_JOB_SCRIPT, - 5, - keys.data, - keys.delayed, - dedupKey, - keys.pending, - keys.overlay, - jobData.id, - JSON.stringify(jobData), - executeAt.toString(), - (jobData.dedup.ttl ?? 0).toString(), - jobData.dedup.extend ? '1' : '0', - jobData.dedup.replace ? '1' : '0', - payloadData, - payloadIsUndefined, - )) as [string, string]; - return { outcome: result[0] as DedupOutcome, jobId: result[1] }; - } - + readonly #connection: Redis + readonly #ownsConnection: boolean + #workerId: string = '' + constructor(connection: Redis, ownsConnection: boolean = false) { + this.#connection = connection + this.#ownsConnection = ownsConnection + } + + #getKeys(queue: string) { + return { + data: `${redisKey}::${queue}::data`, + pending: `${redisKey}::${queue}::pending`, + delayed: `${redisKey}::${queue}::delayed`, + active: `${redisKey}::${queue}::active`, + overlay: `${redisKey}::${queue}::metadata`, + completed: `${redisKey}::${queue}::completed`, + completedIndex: `${redisKey}::${queue}::completed::index`, + failed: `${redisKey}::${queue}::failed`, + failedIndex: `${redisKey}::${queue}::failed::index`, + } + } + + #getDedupKey(queue: string, dedupId: string): string { + return `${this.#getDedupPrefix(queue)}${dedupId}` + } + + #getDedupPrefix(queue: string): string { + return `${redisKey}::${queue}::dedup::` + } + + setWorkerId(workerId: string): void { + this.#workerId = workerId + } + + async destroy(): Promise { + if (this.#ownsConnection) { + await this.#connection.quit() + } + } + + pop(): Promise { + return this.popFrom('default') + } + + async popFrom(queue: string): Promise { + const keys = this.#getKeys(queue) + const now = Date.now() + + const result = await this.#connection.eval( + ACQUIRE_JOB_SCRIPT, + 5, + keys.data, + keys.pending, + keys.active, + keys.delayed, + keys.overlay, + this.#workerId, + now.toString() + ) + + if (!result) { + return null + } + + const { data, overlay, acquiredAt } = JSON.parse(result as string) as { + data: string + overlay?: string + acquiredAt: number + } + + return { ...hydrateRedisJob(data, overlay), acquiredAt } + } + + async completeJob(jobId: string, queue: string, removeOnComplete?: JobRetention): Promise { + const keys = this.#getKeys(queue) + const dedupPrefix = this.#getDedupPrefix(queue) + const { keep, maxAge, maxCount } = resolveRetention(removeOnComplete) + + if (!keep) { await this.#connection.eval( - PUSH_DELAYED_JOB_SCRIPT, - 3, - keys.data, - keys.delayed, - keys.overlay, - jobData.id, - JSON.stringify(jobData), - executeAt.toString(), - ); - } - - async pushOn(queue: string, jobData: JobData): Promise { - const keys = this.#getKeys(queue); - const priority = jobData.priority ?? DEFAULT_PRIORITY; - const timestamp = Date.now(); - const score = calculateScore(priority, timestamp); - - if (jobData.dedup) { - const dedupKey = this.#getDedupKey(queue, jobData.dedup.id); - const [payloadData, payloadIsUndefined] = encodeRedisJobPayloadOverlay( - jobData.payload, - ); - const result = (await this.#connection.eval( - PUSH_DEDUP_JOB_SCRIPT, - 5, - keys.data, - keys.pending, - dedupKey, - keys.delayed, - keys.overlay, - jobData.id, - JSON.stringify(jobData), - score.toString(), - (jobData.dedup.ttl ?? 0).toString(), - jobData.dedup.extend ? '1' : '0', - jobData.dedup.replace ? '1' : '0', - payloadData, - payloadIsUndefined, - )) as [string, string]; - return { outcome: result[0] as DedupOutcome, jobId: result[1] }; - } - + REMOVE_JOB_SCRIPT, + 3, + keys.data, + keys.active, + keys.overlay, + jobId, + dedupPrefix + ) + return + } + + await this.#connection.eval( + FINALIZE_JOB_SCRIPT, + 5, + keys.data, + keys.active, + keys.completed, + keys.completedIndex, + keys.overlay, + jobId, + Date.now().toString(), + maxAge.toString(), + maxCount.toString(), + '', + dedupPrefix + ) + } + + async failJob( + jobId: string, + queue: string, + error?: Error, + removeOnFail?: JobRetention + ): Promise { + const keys = this.#getKeys(queue) + const dedupPrefix = this.#getDedupPrefix(queue) + const { keep, maxAge, maxCount } = resolveRetention(removeOnFail) + + if (!keep) { await this.#connection.eval( - PUSH_JOB_SCRIPT, - 3, - keys.data, - keys.pending, - keys.overlay, - jobData.id, - JSON.stringify(jobData), - score.toString(), - ); - } - - pushMany(jobs: JobData[]): Promise { - return this.pushManyOn('default', jobs); - } - - async pushManyOn(queue: string, jobs: JobData[]): Promise { - if (jobs.length === 0) return; - - if (jobs.some((j) => j.dedup)) { - throw new Error( - 'dedup is not supported in batch dispatch; use single dispatch', - ); - } - - const keys = this.#getKeys(queue); - const now = Date.now(); - const multi = this.#connection.multi(); - - for (const job of jobs) { - const priority = job.priority ?? DEFAULT_PRIORITY; - const score = calculateScore(priority, now); - multi.hdel(keys.overlay, job.id); - multi.hset(keys.data, job.id, JSON.stringify(job)); - multi.zadd(keys.pending, score, job.id); - } - - await multi.exec(); - } - - size(): Promise { - return this.sizeOf('default'); - } - - sizeOf(queue: string): Promise { - const keys = this.#getKeys(queue); - return this.#connection.zcard(keys.pending); - } - - async recoverStalledJobs( - queue: string, - stalledThreshold: number, - maxStalledCount: number, - ): Promise { - const keys = this.#getKeys(queue); - const now = Date.now(); - - const recovered = await this.#connection.eval( - RECOVER_STALLED_JOBS_SCRIPT, - 4, - keys.data, - keys.active, - keys.pending, - keys.overlay, - now.toString(), - stalledThreshold.toString(), - maxStalledCount.toString(), - this.#getDedupPrefix(queue), - ); - - return recovered as number; - } - - async upsertSchedule(config: ScheduleConfig): Promise { - const id = config.id ?? randomUUID(); - const now = Date.now(); - const scheduleKey = `${schedulesKey}::${id}`; - const [existingRunCount, existingCreatedAt, existingNextRunAt] = - await this.#connection.hmget( - scheduleKey, - 'run_count', - 'created_at', - 'next_run_at', - ); - - const scheduleData: Record = { - id, - name: config.name, - payload: JSON.stringify(config.payload), - timezone: config.timezone, - status: 'active', - run_count: existingRunCount ?? '0', - created_at: existingCreatedAt ?? now.toString(), - }; - - if (config.cronExpression !== undefined) - scheduleData.cron_expression = config.cronExpression; - if (config.everyMs !== undefined) - scheduleData.every_ms = config.everyMs.toString(); - if (config.from !== undefined) - scheduleData.from_date = config.from.getTime().toString(); - if (config.to !== undefined) - scheduleData.to_date = config.to.getTime().toString(); - if (config.limit !== undefined) - scheduleData.run_limit = config.limit.toString(); - - const multi = this.#connection - .multi() - .hdel( - scheduleKey, - 'cron_expression', - 'every_ms', - 'from_date', - 'to_date', - 'run_limit', - ) - .hset(scheduleKey, scheduleData) - .sadd(schedulesIndexKey, id); - - if (existingNextRunAt) { - multi.zadd( - schedulesDueKey, - Number.parseInt(existingNextRunAt, 10), - id, - ); - } - - await multi.exec(); - - return id; - } - - /** - * @deprecated Use `upsertSchedule` instead. - */ - createSchedule(config: ScheduleConfig): Promise { - return this.upsertSchedule(config); - } - - async getSchedule(id: string): Promise { - const scheduleKey = `${schedulesKey}::${id}`; - const data = await this.#connection.hgetall(scheduleKey); - + REMOVE_JOB_SCRIPT, + 3, + keys.data, + keys.active, + keys.overlay, + jobId, + dedupPrefix + ) + return + } + + await this.#connection.eval( + FINALIZE_JOB_SCRIPT, + 5, + keys.data, + keys.active, + keys.failed, + keys.failedIndex, + keys.overlay, + jobId, + Date.now().toString(), + maxAge.toString(), + maxCount.toString(), + error?.message || '', + dedupPrefix + ) + } + + async retryJob(jobId: string, queue: string, retryAt?: Date): Promise { + const keys = this.#getKeys(queue) + const now = Date.now() + + await this.#connection.eval( + RETRY_JOB_SCRIPT, + 5, + keys.data, + keys.active, + keys.pending, + keys.delayed, + keys.overlay, + jobId, + retryAt ? retryAt.getTime().toString() : '0', + now.toString() + ) + } + + async getJob(jobId: string, queue: string): Promise { + const keys = this.#getKeys(queue) + + const result = await this.#connection.eval( + GET_JOB_SCRIPT, + 7, + keys.data, + keys.pending, + keys.delayed, + keys.active, + keys.completed, + keys.failed, + keys.overlay, + jobId + ) + + if (!result) { + return null + } + + const record = JSON.parse(result as string) as Omit & { + data: string + overlay?: string + } + + return { ...record, data: hydrateRedisJob(record.data, record.overlay) } + } + + push(jobData: JobData): Promise { + return this.pushOn('default', jobData) + } + + pushLater(jobData: JobData, delay: number): Promise { + return this.pushLaterOn('default', jobData, delay) + } + + async pushLaterOn(queue: string, jobData: JobData, delay: number): Promise { + const keys = this.#getKeys(queue) + const executeAt = Date.now() + delay + + if (jobData.dedup) { + const dedupKey = this.#getDedupKey(queue, jobData.dedup.id) + const [payloadData, payloadIsUndefined] = encodeRedisJobPayloadOverlay(jobData.payload) + const result = (await this.#connection.eval( + PUSH_DEDUP_JOB_SCRIPT, + 5, + keys.data, + keys.delayed, + dedupKey, + keys.pending, + keys.overlay, + jobData.id, + JSON.stringify(jobData), + executeAt.toString(), + (jobData.dedup.ttl ?? 0).toString(), + jobData.dedup.extend ? '1' : '0', + jobData.dedup.replace ? '1' : '0', + payloadData, + payloadIsUndefined + )) as [string, string] + return { outcome: result[0] as DedupOutcome, jobId: result[1] } + } + + await this.#connection.eval( + PUSH_DELAYED_JOB_SCRIPT, + 3, + keys.data, + keys.delayed, + keys.overlay, + jobData.id, + JSON.stringify(jobData), + executeAt.toString() + ) + } + + async pushOn(queue: string, jobData: JobData): Promise { + const keys = this.#getKeys(queue) + const priority = jobData.priority ?? DEFAULT_PRIORITY + const timestamp = Date.now() + const score = calculateScore(priority, timestamp) + + if (jobData.dedup) { + const dedupKey = this.#getDedupKey(queue, jobData.dedup.id) + const [payloadData, payloadIsUndefined] = encodeRedisJobPayloadOverlay(jobData.payload) + const result = (await this.#connection.eval( + PUSH_DEDUP_JOB_SCRIPT, + 5, + keys.data, + keys.pending, + dedupKey, + keys.delayed, + keys.overlay, + jobData.id, + JSON.stringify(jobData), + score.toString(), + (jobData.dedup.ttl ?? 0).toString(), + jobData.dedup.extend ? '1' : '0', + jobData.dedup.replace ? '1' : '0', + payloadData, + payloadIsUndefined + )) as [string, string] + return { outcome: result[0] as DedupOutcome, jobId: result[1] } + } + + await this.#connection.eval( + PUSH_JOB_SCRIPT, + 3, + keys.data, + keys.pending, + keys.overlay, + jobData.id, + JSON.stringify(jobData), + score.toString() + ) + } + + pushMany(jobs: JobData[]): Promise { + return this.pushManyOn('default', jobs) + } + + async pushManyOn(queue: string, jobs: JobData[]): Promise { + if (jobs.length === 0) return + + if (jobs.some((j) => j.dedup)) { + throw new Error('dedup is not supported in batch dispatch; use single dispatch') + } + + const keys = this.#getKeys(queue) + const now = Date.now() + const multi = this.#connection.multi() + + for (const job of jobs) { + const priority = job.priority ?? DEFAULT_PRIORITY + const score = calculateScore(priority, now) + multi.hdel(keys.overlay, job.id) + multi.hset(keys.data, job.id, JSON.stringify(job)) + multi.zadd(keys.pending, score, job.id) + } + + await multi.exec() + } + + size(): Promise { + return this.sizeOf('default') + } + + sizeOf(queue: string): Promise { + const keys = this.#getKeys(queue) + return this.#connection.zcard(keys.pending) + } + + async recoverStalledJobs( + queue: string, + stalledThreshold: number, + maxStalledCount: number + ): Promise { + const keys = this.#getKeys(queue) + const now = Date.now() + + const recovered = await this.#connection.eval( + RECOVER_STALLED_JOBS_SCRIPT, + 4, + keys.data, + keys.active, + keys.pending, + keys.overlay, + now.toString(), + stalledThreshold.toString(), + maxStalledCount.toString(), + this.#getDedupPrefix(queue) + ) + + return recovered as number + } + + async upsertSchedule(config: ScheduleConfig): Promise { + const id = config.id ?? randomUUID() + const now = Date.now() + const scheduleKey = `${schedulesKey}::${id}` + const [existingRunCount, existingCreatedAt, existingNextRunAt] = await this.#connection.hmget( + scheduleKey, + 'run_count', + 'created_at', + 'next_run_at' + ) + + const scheduleData: Record = { + id, + name: config.name, + payload: JSON.stringify(config.payload), + timezone: config.timezone, + status: 'active', + run_count: existingRunCount ?? '0', + created_at: existingCreatedAt ?? now.toString(), + } + + if (config.cronExpression !== undefined) scheduleData.cron_expression = config.cronExpression + if (config.everyMs !== undefined) scheduleData.every_ms = config.everyMs.toString() + if (config.from !== undefined) scheduleData.from_date = config.from.getTime().toString() + if (config.to !== undefined) scheduleData.to_date = config.to.getTime().toString() + if (config.limit !== undefined) scheduleData.run_limit = config.limit.toString() + + const multi = this.#connection + .multi() + .hdel(scheduleKey, 'cron_expression', 'every_ms', 'from_date', 'to_date', 'run_limit') + .hset(scheduleKey, scheduleData) + .sadd(schedulesIndexKey, id) + + if (existingNextRunAt) { + multi.zadd(schedulesDueKey, Number.parseInt(existingNextRunAt, 10), id) + } + + await multi.exec() + + return id + } + + /** + * @deprecated Use `upsertSchedule` instead. + */ + createSchedule(config: ScheduleConfig): Promise { + return this.upsertSchedule(config) + } + + async getSchedule(id: string): Promise { + const scheduleKey = `${schedulesKey}::${id}` + const data = await this.#connection.hgetall(scheduleKey) + + if (!data || Object.keys(data).length === 0) { + return null + } + + return this.#hashToScheduleData(data) + } + + async listSchedules(options?: ScheduleListOptions): Promise { + const ids = await this.#connection.smembers(schedulesIndexKey) + if (ids.length === 0) { + return [] + } + + const pipeline = this.#connection.pipeline() + + for (const id of ids) { + pipeline.hgetall(`${schedulesKey}::${id}`) + } + + const results = await pipeline.exec() + if (!results) { + return [] + } + + const schedules: ScheduleData[] = [] + + for (const [, data] of results) { if (!data || Object.keys(data).length === 0) { - return null; - } - - return this.#hashToScheduleData(data); - } - - async listSchedules(options?: ScheduleListOptions): Promise { - const ids = await this.#connection.smembers(schedulesIndexKey); - if (ids.length === 0) { - return []; - } - - const pipeline = this.#connection.pipeline(); - - for (const id of ids) { - pipeline.hgetall(`${schedulesKey}::${id}`); + continue } - const results = await pipeline.exec(); - if (!results) { - return []; - } - - const schedules: ScheduleData[] = []; - - for (const [, data] of results) { - if (!data || Object.keys(data).length === 0) { - continue; - } + const schedule = this.#hashToScheduleData(data as Record) - const schedule = this.#hashToScheduleData( - data as Record, - ); - - // Filter by status if provided - if (options?.status && schedule.status !== options.status) { - continue; - } - - schedules.push(schedule); - } - - return schedules; - } - - async updateSchedule( - id: string, - updates: Partial< - Pick - >, - ): Promise { - const scheduleKey = `${schedulesKey}::${id}`; - const data: Record = {}; - - if (updates.status !== undefined) data.status = updates.status; - if (updates.nextRunAt !== undefined) { - data.next_run_at = updates.nextRunAt - ? updates.nextRunAt.getTime().toString() - : ''; + // Filter by status if provided + if (options?.status && schedule.status !== options.status) { + continue } - if (updates.lastRunAt !== undefined) { - data.last_run_at = updates.lastRunAt - ? updates.lastRunAt.getTime().toString() - : ''; - } - if (updates.runCount !== undefined) - data.run_count = updates.runCount.toString(); - - if (Object.keys(data).length === 0) return; - - const multi = this.#connection.multi().hset(scheduleKey, data); - if (updates.nextRunAt) { - multi.zadd(schedulesDueKey, updates.nextRunAt.getTime(), id); - } else if (updates.nextRunAt === null || updates.status === 'paused') { - multi.zrem(schedulesDueKey, id); + schedules.push(schedule) + } + + return schedules + } + + async updateSchedule( + id: string, + updates: Partial> + ): Promise { + const scheduleKey = `${schedulesKey}::${id}` + const data: Record = {} + + if (updates.status !== undefined) data.status = updates.status + if (updates.nextRunAt !== undefined) { + data.next_run_at = updates.nextRunAt ? updates.nextRunAt.getTime().toString() : '' + } + if (updates.lastRunAt !== undefined) { + data.last_run_at = updates.lastRunAt ? updates.lastRunAt.getTime().toString() : '' + } + if (updates.runCount !== undefined) data.run_count = updates.runCount.toString() + + if (Object.keys(data).length === 0) return + + const multi = this.#connection.multi().hset(scheduleKey, data) + + if (updates.nextRunAt) { + multi.zadd(schedulesDueKey, updates.nextRunAt.getTime(), id) + } else if (updates.nextRunAt === null || updates.status === 'paused') { + multi.zrem(schedulesDueKey, id) + } + + if (updates.status === 'active' && updates.nextRunAt === undefined) { + const existing = await this.#connection.hget(scheduleKey, 'next_run_at') + if (existing) { + multi.zadd(schedulesDueKey, Number.parseInt(existing, 10), id) } - - if (updates.status === 'active' && updates.nextRunAt === undefined) { - const existing = await this.#connection.hget( - scheduleKey, - 'next_run_at', - ); - if (existing) { - multi.zadd(schedulesDueKey, Number.parseInt(existing, 10), id); - } + } + + await multi.exec() + } + + async deleteSchedule(id: string): Promise { + const scheduleKey = `${schedulesKey}::${id}` + await this.#connection + .multi() + .del(scheduleKey) + .srem(schedulesIndexKey, id) + .zrem(schedulesDueKey, id) + .exec() + } + + async migrate(): Promise { + await this.backfillDueIndex() + } + + async claimDueSchedule(): Promise { + const now = Date.now() + const result = await this.#connection.eval( + CLAIM_SCHEDULE_SCRIPT, + 2, + schedulesDueKey, + `${schedulesKey}::`, + now.toString() + ) + + if (!result) { + return null + } + + const data = JSON.parse(result as string) as Record + + // If cron expression, we need to recalculate next_run_at properly. + // The Lua script only handles simple interval; cron needs JS cron-parser. + // This is safe because the schedule is already claimed (run_count incremented). + if (data.cron_expression) { + const { CronExpressionParser } = await import('cron-parser') + const cron = CronExpressionParser.parse(data.cron_expression, { + currentDate: new Date(now), + tz: data.timezone || 'UTC', + }) + const nextRun = cron.next().toDate().getTime() + + const runCount = Number.parseInt(data.run_count || '0', 10) + 1 + const runLimit = data.run_limit ? Number.parseInt(data.run_limit, 10) : null + const toDate = data.to_date ? Number.parseInt(data.to_date, 10) : null + + let newNextRunAt: number | string = nextRun + + if (runLimit !== null && runCount >= runLimit) { + newNextRunAt = '' + } else if (toDate && nextRun > toDate) { + newNextRunAt = '' } - await multi.exec(); - } - - async deleteSchedule(id: string): Promise { - const scheduleKey = `${schedulesKey}::${id}`; - await this.#connection - .multi() - .del(scheduleKey) - .srem(schedulesIndexKey, id) - .zrem(schedulesDueKey, id) - .exec(); - } - - async #ensureDueIndex(): Promise { - if (this.#dueIndexReady) return; - await this.backfillDueIndex(); - this.#dueIndexReady = true; - } - - async claimDueSchedule(): Promise { - await this.#ensureDueIndex(); - - const now = Date.now(); - const result = await this.#connection.eval( - CLAIM_SCHEDULE_SCRIPT, - 2, - schedulesDueKey, - `${schedulesKey}::`, - now.toString(), - ); - - if (!result) { - return null; - } + const scheduleKey = `${schedulesKey}::${data.id}` + const multi = this.#connection + .multi() + .hset(scheduleKey, 'next_run_at', newNextRunAt.toString()) - const data = JSON.parse(result as string) as Record; - - // If cron expression, we need to recalculate next_run_at properly. - // The Lua script only handles simple interval; cron needs JS cron-parser. - // This is safe because the schedule is already claimed (run_count incremented). - if (data.cron_expression) { - const { CronExpressionParser } = await import('cron-parser'); - const cron = CronExpressionParser.parse(data.cron_expression, { - currentDate: new Date(now), - tz: data.timezone || 'UTC', - }); - const nextRun = cron.next().toDate().getTime(); - - const runCount = Number.parseInt(data.run_count || '0', 10) + 1; - const runLimit = data.run_limit - ? Number.parseInt(data.run_limit, 10) - : null; - const toDate = data.to_date ? Number.parseInt(data.to_date, 10) : null; - - let newNextRunAt: number | string = nextRun; - - if (runLimit !== null && runCount >= runLimit) { - newNextRunAt = ''; - } else if (toDate && nextRun > toDate) { - newNextRunAt = ''; - } - - const scheduleKey = `${schedulesKey}::${data.id}`; - const multi = this.#connection - .multi() - .hset(scheduleKey, 'next_run_at', newNextRunAt.toString()); - - if (typeof newNextRunAt === 'number') { - multi.zadd(schedulesDueKey, newNextRunAt, data.id); - } else { - multi.zrem(schedulesDueKey, data.id); - } - - await multi.exec(); + if (typeof newNextRunAt === 'number') { + multi.zadd(schedulesDueKey, newNextRunAt, data.id) + } else { + multi.zrem(schedulesDueKey, data.id) } - return this.#hashToScheduleData(data); - } - - async backfillDueIndex(): Promise { - const ids = await this.#connection.smembers(schedulesIndexKey); - if (ids.length === 0) return 0; - - const pipeline = this.#connection.pipeline(); - for (const id of ids) { - pipeline.hmget(`${schedulesKey}::${id}`, 'next_run_at', 'status'); - } - const results = await pipeline.exec(); - if (!results) return 0; - - const addPipeline = this.#connection.pipeline(); - let count = 0; - - for (let i = 0; i < ids.length; i++) { - const [err, values] = results[i]; - if (err || !values) continue; - const [nextRunAt, status] = values as [string | null, string | null]; - if (nextRunAt && status === 'active') { - addPipeline.zadd( - schedulesDueKey, - Number.parseInt(nextRunAt, 10), - ids[i], - ); - count++; - } + await multi.exec() + } + + return this.#hashToScheduleData(data) + } + + async backfillDueIndex(): Promise { + const ids = await this.#connection.smembers(schedulesIndexKey) + if (ids.length === 0) return 0 + + const pipeline = this.#connection.pipeline() + for (const id of ids) { + pipeline.hmget(`${schedulesKey}::${id}`, 'next_run_at', 'status') + } + const results = await pipeline.exec() + if (!results) return 0 + + const addPipeline = this.#connection.pipeline() + let count = 0 + + for (let i = 0; i < ids.length; i++) { + const [err, values] = results[i] + if (err || !values) continue + const [nextRunAt, status] = values as [string | null, string | null] + if (nextRunAt && status === 'active') { + addPipeline.zadd(schedulesDueKey, Number.parseInt(nextRunAt, 10), ids[i]) + count++ } - - if (count > 0) await addPipeline.exec(); - return count; - } - - #hashToScheduleData(data: Record): ScheduleData { - return { - id: data.id, - name: data.name, - payload: JSON.parse(data.payload || '{}'), - cronExpression: data.cron_expression || null, - everyMs: data.every_ms ? Number.parseInt(data.every_ms, 10) : null, - timezone: data.timezone || 'UTC', - from: data.from_date - ? new Date(Number.parseInt(data.from_date, 10)) - : null, - to: data.to_date ? new Date(Number.parseInt(data.to_date, 10)) : null, - limit: data.run_limit ? Number.parseInt(data.run_limit, 10) : null, - runCount: Number.parseInt(data.run_count || '0', 10), - nextRunAt: data.next_run_at - ? new Date(Number.parseInt(data.next_run_at, 10)) - : null, - lastRunAt: data.last_run_at - ? new Date(Number.parseInt(data.last_run_at, 10)) - : null, - status: (data.status as 'active' | 'paused') || 'active', - createdAt: data.created_at - ? new Date(Number.parseInt(data.created_at, 10)) - : new Date(), - }; - } + } + + if (count > 0) await addPipeline.exec() + return count + } + + #hashToScheduleData(data: Record): ScheduleData { + return { + id: data.id, + name: data.name, + payload: JSON.parse(data.payload || '{}'), + cronExpression: data.cron_expression || null, + everyMs: data.every_ms ? Number.parseInt(data.every_ms, 10) : null, + timezone: data.timezone || 'UTC', + from: data.from_date ? new Date(Number.parseInt(data.from_date, 10)) : null, + to: data.to_date ? new Date(Number.parseInt(data.to_date, 10)) : null, + limit: data.run_limit ? Number.parseInt(data.run_limit, 10) : null, + runCount: Number.parseInt(data.run_count || '0', 10), + nextRunAt: data.next_run_at ? new Date(Number.parseInt(data.next_run_at, 10)) : null, + lastRunAt: data.last_run_at ? new Date(Number.parseInt(data.last_run_at, 10)) : null, + status: (data.status as 'active' | 'paused') || 'active', + createdAt: data.created_at ? new Date(Number.parseInt(data.created_at, 10)) : new Date(), + } + } } diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index 2f8cedf..82e00cb 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -1,4 +1,4 @@ -import { REDIS_DEDUP_LUA, REDIS_JOB_STORAGE_LUA } from './redis_job_storage.js'; +import { REDIS_DEDUP_LUA, REDIS_JOB_STORAGE_LUA } from './redis_job_storage.js' /** * Lua script for pushing a job to the queue. @@ -18,7 +18,7 @@ ${REDIS_JOB_STORAGE_LUA} redis.call('ZADD', pending_key, score, job_id) return 1 -`; +` /** * Lua script for pushing a dedup job. @@ -80,7 +80,7 @@ ${REDIS_DEDUP_LUA} redis.call('PEXPIRE', dedup_key, ttl) end return {'added', job_id} -`; +` /** * Lua script for pushing a delayed job. @@ -100,7 +100,7 @@ ${REDIS_JOB_STORAGE_LUA} redis.call('ZADD', delayed_key, execute_at, job_id) return 1 -`; +` /** * Lua script for atomic job acquisition. @@ -158,7 +158,7 @@ ${REDIS_JOB_STORAGE_LUA} return encode_job_result(job_data, overlay_key, job_id, { acquiredAt = now }) -`; +` /** * Lua script for removing a job completely (no history). @@ -193,7 +193,7 @@ ${REDIS_JOB_STORAGE_LUA} delete_job_data(data_key, overlay_key, job_id) return 1 -`; +` /** * Lua script for finalizing a job in history. @@ -277,7 +277,7 @@ ${REDIS_JOB_STORAGE_LUA} end return 1 -`; +` /** * Lua script for retrying a job. @@ -330,7 +330,7 @@ ${REDIS_JOB_STORAGE_LUA} end return 1 -`; +` /** * Lua script for recovering stalled jobs. @@ -399,7 +399,7 @@ ${REDIS_JOB_STORAGE_LUA} end return recovered -`; +` /** * Lua script for getting a job record with its status. @@ -458,7 +458,7 @@ ${REDIS_JOB_STORAGE_LUA} finishedAt = finished_at, error = error_msg }) -`; +` /** * Lua script for atomically claiming a due schedule using a sorted set index. @@ -505,6 +505,14 @@ export const CLAIM_SCHEDULE_SCRIPT = ` if schedule.status ~= 'active' then redis.call('ZREM', due_key, id) else + -- Hash is the source of truth for next_run_at. + -- If the ZSET score is stale, repair it and skip this candidate. + local hash_nra = schedule.next_run_at + if not hash_nra or hash_nra == '' then + redis.call('ZREM', due_key, id) + elseif tonumber(hash_nra) > now then + redis.call('ZADD', due_key, tonumber(hash_nra), id) + else local run_count = tonumber(schedule.run_count or '0') local run_limit = schedule.run_limit and tonumber(schedule.run_limit) or nil local to_date = schedule.to_date and tonumber(schedule.to_date) or nil @@ -550,7 +558,8 @@ export const CLAIM_SCHEDULE_SCRIPT = ` -- Return the schedule data (before update) as JSON return cjson.encode(schedule) end + end end end end -`; +` diff --git a/src/drivers/sync_adapter.ts b/src/drivers/sync_adapter.ts index d97fa55..e94aa7b 100644 --- a/src/drivers/sync_adapter.ts +++ b/src/drivers/sync_adapter.ts @@ -118,6 +118,10 @@ export class SyncAdapter implements Adapter { return Promise.resolve() } + migrate(): Promise { + return Promise.resolve() + } + upsertSchedule(_config: ScheduleConfig): Promise { // No-op: schedules don't make sense for sync adapter // Return a fake ID so code doesn't break in dev diff --git a/tests/_mocks/memory_adapter.ts b/tests/_mocks/memory_adapter.ts index fa64e31..40f40e5 100644 --- a/tests/_mocks/memory_adapter.ts +++ b/tests/_mocks/memory_adapter.ts @@ -293,6 +293,10 @@ export class MemoryAdapter implements Adapter { return Promise.resolve() } + migrate(): Promise { + return Promise.resolve() + } + async upsertSchedule(config: ScheduleConfig): Promise { const id = config.id ?? randomUUID() const existing = this.#schedules.get(id) diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index e061869..5e3209a 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -99,9 +99,6 @@ test.group('Adapter | Redis', (group) => { await adapter.updateSchedule(id, { nextRunAt: futureRunAt }) } - // Warm the due-index backfill so it doesn't count against the spy - await adapter.claimDueSchedule() - const { result: claimed, writes } = await withRedisWriteSpy({ connection, run: () => adapter.claimDueSchedule(), @@ -607,6 +604,92 @@ test.group('Adapter | Redis', (group) => { assert.isNull(await connection.hget(metadataKey, 'metadata-stalled-uuid-1')) assert.isNull(await adapter.getJob('metadata-stalled-uuid-1', queue)) }) + + test('backfillDueIndex populates ZSET for pre-existing schedules', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + + // Simulate pre-upgrade schedule data: write hash + index directly, skip ZSET + const id = 'pre-existing-schedule' + const pastRunAt = (Date.now() - 5_000).toString() + await connection + .multi() + .hset(`schedules::${id}`, { + id, + name: 'LegacyJob', + payload: '{}', + status: 'active', + every_ms: '60000', + timezone: 'UTC', + next_run_at: pastRunAt, + last_run_at: '', + run_count: '0', + created_at: Date.now().toString(), + }) + .sadd('schedules::index', id) + .exec() + + // Without backfill, ZSET has no entry so claim returns null + const beforeBackfill = await adapter.claimDueSchedule() + assert.isNull(beforeBackfill) + + await adapter.backfillDueIndex() + + const afterBackfill = await adapter.claimDueSchedule() + assert.isNotNull(afterBackfill) + assert.equal(afterBackfill!.id, id) + }) + + test('backfillDueIndex is idempotent', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + + await adapter.upsertSchedule({ + id: 'idempotent-schedule', + name: 'TestJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }) + await adapter.updateSchedule('idempotent-schedule', { + nextRunAt: new Date(Date.now() + 30_000), + }) + + // Clear the ZSET so backfill has work to do + await connection.del('schedules::due') + + const first = await adapter.backfillDueIndex() + const second = await adapter.backfillDueIndex() + + assert.isAbove(first, 0) + assert.equal(second, first) + + const score = await connection.zscore('schedules::due', 'idempotent-schedule') + assert.isNotNull(score) + }) + + test('stale ZSET score is self-healed during claim', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + const id = 'stale-score-schedule' + const futureRunAt = Date.now() + 60_000 + + await adapter.upsertSchedule({ + id, + name: 'StaleJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }) + await adapter.updateSchedule(id, { nextRunAt: new Date(futureRunAt) }) + + // Corrupt the ZSET score to a past value while hash still says future + await connection.zadd('schedules::due', Date.now() - 10_000, id) + + const claimed = await adapter.claimDueSchedule() + assert.isNull(claimed, 'should not claim when hash says schedule is not due yet') + + // ZSET score should have been repaired to match the hash + const repairedScore = await connection.zscore('schedules::due', id) + assert.equal(Number(repairedScore), futureRunAt) + }) }) test.group('Adapter | Knex (SQLite)', (group) => { From bc0226dc518f718b427330b4360a903198343085 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 09:26:14 +0000 Subject: [PATCH 05/14] 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. --- .../adapter-aware-workers-and-schedules.md | 17 +++ README.md | 20 +++ src/drivers/kysely_adapter.ts | 2 + src/drivers/redis_adapter.ts | 51 +++++-- tests/_utils/register_driver_test_suite.ts | 7 + tests/adapter.spec.ts | 139 ++++++++++++++++-- tests/queue_manager.spec.ts | 2 + 7 files changed, 212 insertions(+), 26 deletions(-) diff --git a/.changelog/adapter-aware-workers-and-schedules.md b/.changelog/adapter-aware-workers-and-schedules.md index 87790e3..24f7a11 100644 --- a/.changelog/adapter-aware-workers-and-schedules.md +++ b/.changelog/adapter-aware-workers-and-schedules.md @@ -66,3 +66,20 @@ its configured Adapter. When a schedule does not call `.with()`, its Adapter is resolved from the job's `adapter` option, then from the Adapter configured for the job's queue, and finally from the queue manager default. An explicit `.with()` always takes precedence. + +### Run Adapter Migrations Before Starting Workers + +The `Adapter` contract now includes an idempotent `migrate()` lifecycle method. Built-in adapters +without data migrations implement it as a no-op; custom adapters must implement it as well. + +Redis now claims schedules through the derived `schedules::due` sorted-set index. Deployments +upgrading from an earlier version must rebuild that index before workers start: + +```typescript +await QueueManager.init(config) +await QueueManager.use('redis').migrate() +``` + +Existing Redis schedules will not fire from the new index until this migration runs. The migration +scans all schedules, is safe to repeat, and should remain an explicit deployment step rather than +part of schedule polling. diff --git a/README.md b/README.md index 1fbcaaa..9c0bc37 100644 --- a/README.md +++ b/README.md @@ -334,6 +334,26 @@ const connection = new Redis({ host: 'localhost' }) const adapter = redis(connection) ``` +#### Migrating Redis schedules after an upgrade + +The Redis adapter uses a `schedules::due` sorted-set index to find due schedules. When upgrading +from a version that predates this index, run the adapter migration once during deployment, before +starting any workers: + +```typescript +import { QueueManager, Worker } from '@boringnode/queue' + +await QueueManager.init(config) +await QueueManager.use('redis').migrate() + +const worker = new Worker(config) +await worker.start(['default']) +``` + +The migration is idempotent and rebuilds the derived index from the canonical schedule hashes. +Existing Redis schedules will not fire through the new index until it has run. Do not run the +`O(number of schedules)` migration from the worker polling loop. + ### Knex (PostgreSQL, MySQL, SQLite) ```typescript diff --git a/src/drivers/kysely_adapter.ts b/src/drivers/kysely_adapter.ts index e09504c..7ddbe59 100644 --- a/src/drivers/kysely_adapter.ts +++ b/src/drivers/kysely_adapter.ts @@ -122,6 +122,8 @@ export class KyselyAdapter implements Adapter { // The Kysely instance is always owned by the application. } + async migrate(): Promise {} + async pop(): Promise { return this.popFrom('default') } diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 2edc06c..21e2f47 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -464,6 +464,8 @@ export class RedisAdapter implements Adapter { if (existingNextRunAt) { multi.zadd(schedulesDueKey, Number.parseInt(existingNextRunAt, 10), id) + } else { + multi.zrem(schedulesDueKey, id) } await multi.exec() @@ -544,18 +546,33 @@ export class RedisAdapter implements Adapter { if (Object.keys(data).length === 0) return - const multi = this.#connection.multi().hset(scheduleKey, data) - - if (updates.nextRunAt) { - multi.zadd(schedulesDueKey, updates.nextRunAt.getTime(), id) - } else if (updates.nextRunAt === null || updates.status === 'paused') { - multi.zrem(schedulesDueKey, id) + let dueStatus = updates.status + let dueAt = updates.nextRunAt === undefined ? undefined : (updates.nextRunAt?.getTime() ?? null) + + if ( + (updates.status !== undefined || updates.nextRunAt !== undefined) && + (dueStatus === undefined || dueAt === undefined) + ) { + const [existingStatus, existingNextRunAt] = await this.#connection.hmget( + scheduleKey, + 'status', + 'next_run_at' + ) + if (dueStatus === undefined) { + dueStatus = existingStatus === 'paused' ? 'paused' : 'active' + } + if (dueAt === undefined) { + dueAt = existingNextRunAt ? Number.parseInt(existingNextRunAt, 10) : null + } } - if (updates.status === 'active' && updates.nextRunAt === undefined) { - const existing = await this.#connection.hget(scheduleKey, 'next_run_at') - if (existing) { - multi.zadd(schedulesDueKey, Number.parseInt(existing, 10), id) + const multi = this.#connection.multi().hset(scheduleKey, data) + + if (updates.status !== undefined || updates.nextRunAt !== undefined) { + if (dueStatus === 'active' && dueAt !== null && dueAt !== undefined) { + multi.zadd(schedulesDueKey, dueAt, id) + } else { + multi.zrem(schedulesDueKey, id) } } @@ -634,7 +651,10 @@ export class RedisAdapter implements Adapter { async backfillDueIndex(): Promise { const ids = await this.#connection.smembers(schedulesIndexKey) - if (ids.length === 0) return 0 + if (ids.length === 0) { + await this.#connection.del(schedulesDueKey) + return 0 + } const pipeline = this.#connection.pipeline() for (const id of ids) { @@ -643,20 +663,21 @@ export class RedisAdapter implements Adapter { const results = await pipeline.exec() if (!results) return 0 - const addPipeline = this.#connection.pipeline() + const rebuild = this.#connection.multi().del(schedulesDueKey) let count = 0 for (let i = 0; i < ids.length; i++) { const [err, values] = results[i] if (err || !values) continue const [nextRunAt, status] = values as [string | null, string | null] - if (nextRunAt && status === 'active') { - addPipeline.zadd(schedulesDueKey, Number.parseInt(nextRunAt, 10), ids[i]) + const score = nextRunAt ? Number.parseInt(nextRunAt, 10) : Number.NaN + if (Number.isFinite(score) && status === 'active') { + rebuild.zadd(schedulesDueKey, score, ids[i]) count++ } } - if (count > 0) await addPipeline.exec() + await rebuild.exec() return count } diff --git a/tests/_utils/register_driver_test_suite.ts b/tests/_utils/register_driver_test_suite.ts index a8ba74c..5080067 100644 --- a/tests/_utils/register_driver_test_suite.ts +++ b/tests/_utils/register_driver_test_suite.ts @@ -21,6 +21,13 @@ interface DriverTestSuiteOptions { export function registerDriverTestSuite(options: DriverTestSuiteOptions) { const { test } = options + test('migrate should be safe to call repeatedly', async () => { + const adapter = await options.createAdapter() + + await adapter.migrate() + await adapter.migrate() + }) + test('popFrom should return null when queue is empty', async ({ assert }) => { const adapter = await options.createAdapter() adapter.setWorkerId('worker-1') diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 0d5286f..4edc857 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -629,7 +629,7 @@ test.group('Adapter | Redis', (group) => { assert.isNull(await adapter.getJob('metadata-stalled-uuid-1', queue)) }) - test('backfillDueIndex populates ZSET for pre-existing schedules', async ({ assert }) => { + test('migrate makes pre-existing schedules claimable from the due index', async ({ assert }) => { const adapter = new RedisAdapter(connection) // Simulate pre-upgrade schedule data: write hash + index directly, skip ZSET @@ -656,7 +656,10 @@ test.group('Adapter | Redis', (group) => { const beforeBackfill = await adapter.claimDueSchedule() assert.isNull(beforeBackfill) - await adapter.backfillDueIndex() + await adapter.migrate() + + const score = await connection.zscore('schedules::due', id) + assert.equal(Number(score), Number(pastRunAt)) const afterBackfill = await adapter.claimDueSchedule() assert.isNotNull(afterBackfill) @@ -665,6 +668,7 @@ test.group('Adapter | Redis', (group) => { test('backfillDueIndex is idempotent', async ({ assert }) => { const adapter = new RedisAdapter(connection) + const nextRunAt = Date.now() + 30_000 await adapter.upsertSchedule({ id: 'idempotent-schedule', @@ -674,20 +678,23 @@ test.group('Adapter | Redis', (group) => { timezone: 'UTC', }) await adapter.updateSchedule('idempotent-schedule', { - nextRunAt: new Date(Date.now() + 30_000), + nextRunAt: new Date(nextRunAt), }) - // Clear the ZSET so backfill has work to do - await connection.del('schedules::due') + await connection + .multi() + .del('schedules::due') + .zadd('schedules::due', Date.now() - 10_000, 'orphaned-schedule') + .exec() - const first = await adapter.backfillDueIndex() - const second = await adapter.backfillDueIndex() + await adapter.backfillDueIndex() + const firstMembers = await connection.zrange('schedules::due', 0, -1, 'WITHSCORES') - assert.isAbove(first, 0) - assert.equal(second, first) + await adapter.backfillDueIndex() + const secondMembers = await connection.zrange('schedules::due', 0, -1, 'WITHSCORES') - const score = await connection.zscore('schedules::due', 'idempotent-schedule') - assert.isNotNull(score) + assert.deepEqual(firstMembers, ['idempotent-schedule', nextRunAt.toString()]) + assert.deepEqual(secondMembers, firstMembers) }) test('stale ZSET score is self-healed during claim', async ({ assert }) => { @@ -714,6 +721,116 @@ test.group('Adapter | Redis', (group) => { const repairedScore = await connection.zscore('schedules::due', id) assert.equal(Number(repairedScore), futureRunAt) }) + + test('schedule lifecycle keeps the due index aligned with canonical state', async ({ + assert, + }) => { + const adapter = new RedisAdapter(connection) + const id = 'lifecycle-schedule' + const firstRunAt = Date.now() + 30_000 + const pausedRunAt = firstRunAt + 30_000 + + await adapter.upsertSchedule({ + id, + name: 'LifecycleJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }) + await adapter.updateSchedule(id, { nextRunAt: new Date(firstRunAt) }) + assert.equal(Number(await connection.zscore('schedules::due', id)), firstRunAt) + + await adapter.updateSchedule(id, { + status: 'paused', + nextRunAt: new Date(pausedRunAt), + }) + assert.isNull(await connection.zscore('schedules::due', id)) + + await adapter.updateSchedule(id, { nextRunAt: new Date(pausedRunAt + 30_000) }) + assert.isNull(await connection.zscore('schedules::due', id)) + + await adapter.updateSchedule(id, { status: 'active' }) + assert.equal(Number(await connection.zscore('schedules::due', id)), pausedRunAt + 30_000) + + await adapter.updateSchedule(id, { nextRunAt: null }) + assert.isNull(await connection.zscore('schedules::due', id)) + + await adapter.updateSchedule(id, { nextRunAt: new Date(firstRunAt) }) + await adapter.deleteSchedule(id) + assert.isNull(await connection.zscore('schedules::due', id)) + }) + + test('interval claims update the due index from the canonical hash', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + const id = 'interval-index-schedule' + + await adapter.upsertSchedule({ + id, + name: 'IntervalJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }) + await adapter.updateSchedule(id, { nextRunAt: new Date(Date.now() - 1_000) }) + + assert.equal((await adapter.claimDueSchedule())?.id, id) + + const schedule = await adapter.getSchedule(id) + const score = await connection.zscore('schedules::due', id) + assert.isNotNull(schedule!.nextRunAt) + assert.equal(Number(score), schedule!.nextRunAt!.getTime()) + }) + + test('cron claims update the due index from the canonical hash', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + const id = 'cron-index-schedule' + + await adapter.upsertSchedule({ + id, + name: 'CronJob', + payload: {}, + cronExpression: '* * * * *', + timezone: 'UTC', + }) + await adapter.updateSchedule(id, { nextRunAt: new Date(Date.now() - 1_000) }) + + assert.equal((await adapter.claimDueSchedule())?.id, id) + + const schedule = await adapter.getSchedule(id) + const score = await connection.zscore('schedules::due', id) + assert.isNotNull(schedule!.nextRunAt) + assert.equal(Number(score), schedule!.nextRunAt!.getTime()) + }) + + test('exhausted schedules are removed from the due index', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + + for (const config of [ + { id: 'limited-interval', everyMs: 60_000, limit: 1 }, + { id: 'limited-cron', cronExpression: '* * * * *', limit: 1 }, + { id: 'ended-interval', everyMs: 60_000, to: new Date(Date.now() + 1_000) }, + ]) { + await adapter.upsertSchedule({ + ...config, + name: 'ExhaustedJob', + payload: {}, + timezone: 'UTC', + }) + await adapter.updateSchedule(config.id, { nextRunAt: new Date(Date.now() - 1_000) }) + + assert.equal((await adapter.claimDueSchedule())?.id, config.id) + assert.isNull((await adapter.getSchedule(config.id))!.nextRunAt) + assert.isNull(await connection.zscore('schedules::due', config.id)) + } + }) + + test('claim removes a due index member whose canonical hash is missing', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + await connection.zadd('schedules::due', Date.now() - 1_000, 'missing-schedule') + + assert.isNull(await adapter.claimDueSchedule()) + assert.isNull(await connection.zscore('schedules::due', 'missing-schedule')) + }) }) test.group('Adapter | Knex (SQLite)', (group) => { diff --git a/tests/queue_manager.spec.ts b/tests/queue_manager.spec.ts index 46dde6d..df6b1f0 100644 --- a/tests/queue_manager.spec.ts +++ b/tests/queue_manager.spec.ts @@ -263,6 +263,7 @@ test.group('QueueManager', () => { destroy: async () => { destroyedCount++ }, + migrate: async () => {}, upsertSchedule: async () => 'schedule-id', createSchedule: async () => 'schedule-id', getSchedule: async () => null, @@ -330,6 +331,7 @@ test.group('QueueManager', () => { size: async () => 0, sizeOf: async () => 0, destroy: async () => {}, + migrate: async () => {}, upsertSchedule: async () => 'schedule-id', createSchedule: async () => 'schedule-id', getSchedule: async () => null, From b8c4d7cdb3c4de488d2cd00c98aab3c341ff2d4c Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 11:46:38 +0000 Subject: [PATCH 06/14] chore: remove released changelog entries --- .../adapter-aware-workers-and-schedules.md | 85 ------------------- .changelog/consistent-job-runtime.md | 30 ------- .changelog/hot-reloading-jobs.md | 44 ---------- 3 files changed, 159 deletions(-) delete mode 100644 .changelog/adapter-aware-workers-and-schedules.md delete mode 100644 .changelog/consistent-job-runtime.md delete mode 100644 .changelog/hot-reloading-jobs.md diff --git a/.changelog/adapter-aware-workers-and-schedules.md b/.changelog/adapter-aware-workers-and-schedules.md deleted file mode 100644 index 24f7a11..0000000 --- a/.changelog/adapter-aware-workers-and-schedules.md +++ /dev/null @@ -1,85 +0,0 @@ -# Adapter-Aware Workers and Schedules - -## New Features - -### Select an Adapter per Worker - -Workers can now listen on a specific registered Adapter with `worker.adapter`. When omitted, the -worker continues to use the queue manager's default Adapter. - -```typescript -const config = { - default: 'redis', - adapters: { - redis: redis(redisConfig), - database: knex(databaseConfig), - }, - worker: { - adapter: 'database', - concurrency: 5, - }, -} - -const worker = new Worker(config) -await worker.start(['default', 'emails']) -``` - -This makes it possible to run separate workers for queues stored by different Adapters. - -### Store and Access Schedules on a Specific Adapter - -Schedules can now select their owning Adapter with `.with()`: - -```typescript -await CleanupJob.schedule({ days: 30 }).id('daily-cleanup').with('redis').cron('0 0 * * *') -``` - -`Schedule.find()` and `Schedule.list()` accept an Adapter selector when accessing schedules outside -the default Adapter: - -```typescript -const schedule = await Schedule.find('daily-cleanup', { adapter: 'redis' }) -const schedules = await Schedule.list({ status: 'active' }, { adapter: 'redis' }) -``` - -A returned `Schedule` retains the selected Adapter for subsequent `pause()`, `resume()`, `delete()`, -and `trigger()` calls. Jobs dispatched by a schedule stay on the Adapter that owns that schedule. - -### Identify Jobs Dispatched by a Schedule - -Scheduled jobs now include their originating schedule ID in `JobData.scheduleId`. Jobs can access it -while executing through `this.context.scheduleId`: - -```typescript -async execute() { - console.log(this.context.scheduleId) -} -``` - -The value is `undefined` for jobs that were not dispatched by a schedule. - -## Upgrade Notes - -Start a Worker for every Adapter that owns schedules. A Worker only claims schedules and jobs from -its configured Adapter. - -When a schedule does not call `.with()`, its Adapter is resolved from the job's `adapter` option, -then from the Adapter configured for the job's queue, and finally from the queue manager default. -An explicit `.with()` always takes precedence. - -### Run Adapter Migrations Before Starting Workers - -The `Adapter` contract now includes an idempotent `migrate()` lifecycle method. Built-in adapters -without data migrations implement it as a no-op; custom adapters must implement it as well. - -Redis now claims schedules through the derived `schedules::due` sorted-set index. Deployments -upgrading from an earlier version must rebuild that index before workers start: - -```typescript -await QueueManager.init(config) -await QueueManager.use('redis').migrate() -``` - -Existing Redis schedules will not fire from the new index until this migration runs. The migration -scans all schedules, is safe to repeat, and should remain an explicit deployment step rather than -part of schedule polling. diff --git a/.changelog/consistent-job-runtime.md b/.changelog/consistent-job-runtime.md deleted file mode 100644 index efdcf4f..0000000 --- a/.changelog/consistent-job-runtime.md +++ /dev/null @@ -1,30 +0,0 @@ -# Consistent Job Dispatch and Execution - -## Improvements - -All job dispatch paths now apply the same routing and job options. This includes `dispatch()`, -`dispatchMany()`, manual schedule triggers, and schedules claimed by workers. - -The routing order is now consistent across these paths: - -1. Fluent overrides such as `.toQueue()` and `.with()` -2. Static `Job.options` -3. The Adapter configured for the selected queue -4. The queue manager's default Adapter - -Queue, Adapter, priority, custom job name, creation timestamp, and schedule provenance are therefore -preserved consistently regardless of how a job is dispatched. Static job options are resolved when -the fluent builder runs, so changes made between builder creation and execution are applied. - -The Sync adapter and Worker execution paths now also share the same job lifecycle behavior, -including context construction, dependency injection through `jobFactory`, execution wrappers, -timeouts, retries, failed hooks, and tracing. - -## Upgrade Notes - -A job routed to a queue with `queues..adapter` now uses that Adapter when neither `.with()` nor -`Job.options.adapter` selects another one. Previously, some dispatch paths could incorrectly fall -back to the queue manager's default Adapter. Verify that a Worker is running for every Adapter used -by queue configuration. - -Explicit fluent options continue to take precedence over static job options. diff --git a/.changelog/hot-reloading-jobs.md b/.changelog/hot-reloading-jobs.md deleted file mode 100644 index d23c314..0000000 --- a/.changelog/hot-reloading-jobs.md +++ /dev/null @@ -1,44 +0,0 @@ -# Hot Reloading Jobs - -## New Feature - -Workers can now execute the latest saved version of a job without restarting during development. - -Enable `hotReload` when initializing the queue manager. Jobs discovered from `locations` will then -be resolved from their module again before every execution. - -```typescript -await QueueManager.init({ - default: 'redis', - adapters: { - redis: redis({ host: 'localhost', port: 6379 }), - }, - locations: ['./app/jobs/**/*.ts'], - hotReload: process.env.NODE_ENV === 'development', -}) -``` - -Hot reload integrates with [Hot Hook](https://github.com/Julien-R44/hot-hook). The queue provides -the dynamic import boundary, while the application remains responsible for installing and -initializing Hot Hook. AdonisJS applications can use `node ace serve --hmr`; standalone worker -processes must initialize Hot Hook themselves. - -`Locator.registerFromGlob()` also accepts the option directly: - -```typescript -await Locator.registerFromGlob(['./app/jobs/**/*.ts'], { hotReload: true }) -``` - -## Upgrade Notes - -Hot reload is disabled by default and should only be enabled in development. - -Only jobs discovered from `locations` or registered with `Locator.registerFromGlob()` can be -reloaded. Jobs registered manually with `Locator.register()` do not have a module path to reload. - -Changes to the set of registered jobs still require a restart. This includes adding, deleting, -moving, or renaming a job, as well as changing its configured `name`. A job that is already running -keeps its current implementation; the next execution receives the updated version. - -Avoid import-time side effects in hot-reloaded job modules, since their module code can execute -again after an invalidation. From b6ef1e8fb5dbb099b173f3532406e3b08a2449ad Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 11:47:11 +0000 Subject: [PATCH 07/14] docs: add Redis schedule index changelog --- .changelog/redis-schedule-due-index.md | 29 ++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) create mode 100644 .changelog/redis-schedule-due-index.md diff --git a/.changelog/redis-schedule-due-index.md b/.changelog/redis-schedule-due-index.md new file mode 100644 index 0000000..f37432b --- /dev/null +++ b/.changelog/redis-schedule-due-index.md @@ -0,0 +1,29 @@ +# Indexed Redis Schedule Claims + +## Performance Improvement + +The Redis adapter now maintains a `schedules::due` sorted-set index scored by each schedule's +`next_run_at`. Claiming the next due schedule uses this index instead of scanning every stored +schedule, so polling no longer grows linearly with the total schedule count. + +Schedule hashes remain the source of truth. Creating, updating, pausing, resuming, deleting, and +claiming schedules maintain the derived index, while claiming repairs stale entries when the hash +and index disagree. + +## Upgrade Notes + +This change requires an explicit migration for existing Redis schedules. The `Adapter` contract now +includes an idempotent `migrate()` method; built-in adapters without migrations implement it as a +no-op, and custom adapters must implement it as well. + +Run the migration once during deployment, before starting workers or any process that creates or +updates schedules: + +```typescript +await QueueManager.init(config) +await QueueManager.use('redis').migrate() +``` + +Existing Redis schedules will not fire through the new index until the migration has completed. +The migration scans all schedules and should remain an explicit deployment step rather than run in +the worker polling loop. From ec32035e9e5800cb8bfdd6d356ab1233dff3b244 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 11:59:24 +0000 Subject: [PATCH 08/14] 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. --- src/drivers/redis_adapter.ts | 109 ++++++--------------- src/drivers/redis_scripts.ts | 103 ++++++++++++++++++++ tests/adapter.spec.ts | 177 +++++++++++++++++++++++++++++++++++ 3 files changed, 308 insertions(+), 81 deletions(-) diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 21e2f47..17f3069 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -16,6 +16,7 @@ import { resolveRetention } from '../utils.js' import { encodeRedisJobPayloadOverlay, hydrateRedisJob } from './redis_job_storage.js' import { ACQUIRE_JOB_SCRIPT, + BACKFILL_SCHEDULE_DUE_INDEX_SCRIPT, CLAIM_SCHEDULE_SCRIPT, FINALIZE_JOB_SCRIPT, GET_JOB_SCRIPT, @@ -26,6 +27,8 @@ import { REMOVE_JOB_SCRIPT, RENEW_JOBS_SCRIPT, RETRY_JOB_SCRIPT, + UPDATE_SCHEDULE_SCRIPT, + UPSERT_SCHEDULE_SCRIPT, } from './redis_scripts.js' const redisKey = 'jobs' @@ -433,12 +436,6 @@ export class RedisAdapter implements Adapter { const id = config.id ?? randomUUID() const now = Date.now() const scheduleKey = `${schedulesKey}::${id}` - const [existingRunCount, existingCreatedAt, existingNextRunAt] = await this.#connection.hmget( - scheduleKey, - 'run_count', - 'created_at', - 'next_run_at' - ) const scheduleData: Record = { id, @@ -446,8 +443,6 @@ export class RedisAdapter implements Adapter { payload: JSON.stringify(config.payload), timezone: config.timezone, status: 'active', - run_count: existingRunCount ?? '0', - created_at: existingCreatedAt ?? now.toString(), } if (config.cronExpression !== undefined) scheduleData.cron_expression = config.cronExpression @@ -456,19 +451,16 @@ export class RedisAdapter implements Adapter { if (config.to !== undefined) scheduleData.to_date = config.to.getTime().toString() if (config.limit !== undefined) scheduleData.run_limit = config.limit.toString() - const multi = this.#connection - .multi() - .hdel(scheduleKey, 'cron_expression', 'every_ms', 'from_date', 'to_date', 'run_limit') - .hset(scheduleKey, scheduleData) - .sadd(schedulesIndexKey, id) - - if (existingNextRunAt) { - multi.zadd(schedulesDueKey, Number.parseInt(existingNextRunAt, 10), id) - } else { - multi.zrem(schedulesDueKey, id) - } - - await multi.exec() + await this.#connection.eval( + UPSERT_SCHEDULE_SCRIPT, + 3, + scheduleKey, + schedulesIndexKey, + schedulesDueKey, + id, + now.toString(), + JSON.stringify(scheduleData) + ) return id } @@ -546,37 +538,14 @@ export class RedisAdapter implements Adapter { if (Object.keys(data).length === 0) return - let dueStatus = updates.status - let dueAt = updates.nextRunAt === undefined ? undefined : (updates.nextRunAt?.getTime() ?? null) - - if ( - (updates.status !== undefined || updates.nextRunAt !== undefined) && - (dueStatus === undefined || dueAt === undefined) - ) { - const [existingStatus, existingNextRunAt] = await this.#connection.hmget( - scheduleKey, - 'status', - 'next_run_at' - ) - if (dueStatus === undefined) { - dueStatus = existingStatus === 'paused' ? 'paused' : 'active' - } - if (dueAt === undefined) { - dueAt = existingNextRunAt ? Number.parseInt(existingNextRunAt, 10) : null - } - } - - const multi = this.#connection.multi().hset(scheduleKey, data) - - if (updates.status !== undefined || updates.nextRunAt !== undefined) { - if (dueStatus === 'active' && dueAt !== null && dueAt !== undefined) { - multi.zadd(schedulesDueKey, dueAt, id) - } else { - multi.zrem(schedulesDueKey, id) - } - } - - await multi.exec() + await this.#connection.eval( + UPDATE_SCHEDULE_SCRIPT, + 2, + scheduleKey, + schedulesDueKey, + id, + JSON.stringify(data) + ) } async deleteSchedule(id: string): Promise { @@ -650,35 +619,13 @@ export class RedisAdapter implements Adapter { } async backfillDueIndex(): Promise { - const ids = await this.#connection.smembers(schedulesIndexKey) - if (ids.length === 0) { - await this.#connection.del(schedulesDueKey) - return 0 - } - - const pipeline = this.#connection.pipeline() - for (const id of ids) { - pipeline.hmget(`${schedulesKey}::${id}`, 'next_run_at', 'status') - } - const results = await pipeline.exec() - if (!results) return 0 - - const rebuild = this.#connection.multi().del(schedulesDueKey) - let count = 0 - - for (let i = 0; i < ids.length; i++) { - const [err, values] = results[i] - if (err || !values) continue - const [nextRunAt, status] = values as [string | null, string | null] - const score = nextRunAt ? Number.parseInt(nextRunAt, 10) : Number.NaN - if (Number.isFinite(score) && status === 'active') { - rebuild.zadd(schedulesDueKey, score, ids[i]) - count++ - } - } - - await rebuild.exec() - return count + return (await this.#connection.eval( + BACKFILL_SCHEDULE_DUE_INDEX_SCRIPT, + 3, + schedulesIndexKey, + schedulesDueKey, + `${schedulesKey}::` + )) as number } #hashToScheduleData(data: Record): ScheduleData { diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index 8427b8d..0af0776 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -491,6 +491,109 @@ ${REDIS_JOB_STORAGE_LUA} }) ` +const SCHEDULE_DUE_INDEX_LUA = ` + local function sync_schedule_due_index(schedule_key, due_key, id) + local status = redis.call('HGET', schedule_key, 'status') + local next_run_at = redis.call('HGET', schedule_key, 'next_run_at') + local score = next_run_at and tonumber(next_run_at) or nil + + if status == 'active' and score then + redis.call('ZADD', due_key, score, id) + else + redis.call('ZREM', due_key, id) + end + end +` + +/** + * Atomically upserts schedule configuration while preserving runtime fields + * and synchronizing the derived due index from the resulting hash. + */ +export const UPSERT_SCHEDULE_SCRIPT = ` + local schedule_key = KEYS[1] + local schedules_index_key = KEYS[2] + local due_key = KEYS[3] + local id = ARGV[1] + local now = ARGV[2] + local schedule = cjson.decode(ARGV[3]) + +${SCHEDULE_DUE_INDEX_LUA} + + local run_count = redis.call('HGET', schedule_key, 'run_count') or '0' + local created_at = redis.call('HGET', schedule_key, 'created_at') or now + + redis.call( + 'HDEL', + schedule_key, + 'cron_expression', + 'every_ms', + 'from_date', + 'to_date', + 'run_limit' + ) + + for field, value in pairs(schedule) do + redis.call('HSET', schedule_key, field, value) + end + + redis.call('HSET', schedule_key, 'run_count', run_count, 'created_at', created_at) + redis.call('SADD', schedules_index_key, id) + sync_schedule_due_index(schedule_key, due_key, id) + + return id +` + +/** + * Atomically updates schedule runtime fields and synchronizes the derived due + * index from the resulting canonical hash. + */ +export const UPDATE_SCHEDULE_SCRIPT = ` + local schedule_key = KEYS[1] + local due_key = KEYS[2] + local id = ARGV[1] + local updates = cjson.decode(ARGV[2]) + +${SCHEDULE_DUE_INDEX_LUA} + + for field, value in pairs(updates) do + redis.call('HSET', schedule_key, field, value) + end + + sync_schedule_due_index(schedule_key, due_key, id) + + return 1 +` + +/** + * Atomically rebuilds the derived due index from canonical schedule hashes. + * This is an explicit O(N) migration and blocks concurrent Redis commands + * until the complete index reflects one consistent point in time. + */ +export const BACKFILL_SCHEDULE_DUE_INDEX_SCRIPT = ` + local schedules_index_key = KEYS[1] + local due_key = KEYS[2] + local schedule_key_prefix = KEYS[3] + local ids = redis.call('SMEMBERS', schedules_index_key) + local count = 0 + + redis.call('DEL', due_key) + + for i = 1, #ids do + local id = ids[i] + local schedule_key = schedule_key_prefix .. id + local status = redis.call('HGET', schedule_key, 'status') + local next_run_at = redis.call('HGET', schedule_key, 'next_run_at') + local score = next_run_at and tonumber(next_run_at) or nil + + if status == 'active' and score then + redis.call('ZADD', due_key, score, id) + count = count + 1 + end + end + + return count +` + /** * Lua script for atomically claiming a due schedule using a sorted set index. * diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 4edc857..c79981e 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -697,6 +697,68 @@ test.group('Adapter | Redis', (group) => { assert.deepEqual(secondMembers, firstMembers) }) + test('backfillDueIndex rebuilds the index in one atomic Redis command', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + const id = 'atomic-backfill-schedule' + + await connection + .multi() + .hset(`schedules::${id}`, { + id, + status: 'active', + next_run_at: (Date.now() + 30_000).toString(), + }) + .sadd('schedules::index', id) + .exec() + + const { writes } = await withRedisWriteSpy({ + connection, + run: () => adapter.backfillDueIndex(), + }) + + assert.equal(writes, 1) + }) + + test('concurrent migration and schedule writes leave the due index canonical', async ({ + assert, + cleanup, + }) => { + const secondConnection = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + }) + cleanup(async () => { + await secondConnection.quit() + }) + + const adapter = new RedisAdapter(connection) + const secondAdapter = new RedisAdapter(secondConnection) + + for (let i = 0; i < 20; i++) { + const id = `migration-write-schedule-${i}` + const nextRunAt = Date.now() + 30_000 + i + + await Promise.all([ + adapter.backfillDueIndex(), + secondAdapter.upsertSchedule({ + id, + name: 'MigrationWriteJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }), + ]) + await Promise.all([ + adapter.backfillDueIndex(), + secondAdapter.updateSchedule(id, { nextRunAt: new Date(nextRunAt) }), + ]) + + assert.equal(Number(await connection.zscore('schedules::due', id)), nextRunAt) + } + }) + test('stale ZSET score is self-healed during claim', async ({ assert }) => { const adapter = new RedisAdapter(connection) const id = 'stale-score-schedule' @@ -760,6 +822,121 @@ test.group('Adapter | Redis', (group) => { assert.isNull(await connection.zscore('schedules::due', id)) }) + test('schedule mutations update canonical state and its index in one command', async ({ + assert, + }) => { + const adapter = new RedisAdapter(connection) + const id = 'atomic-schedule-mutation' + + const { writes: upsertWrites } = await withRedisWriteSpy({ + connection, + run: () => + adapter.upsertSchedule({ + id, + name: 'AtomicJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }), + }) + assert.equal(upsertWrites, 1) + + const { writes: updateWrites } = await withRedisWriteSpy({ + connection, + run: () => adapter.updateSchedule(id, { nextRunAt: new Date(Date.now() + 30_000) }), + }) + assert.equal(updateWrites, 1) + }) + + test('concurrent resume and next-run updates leave the due index canonical', async ({ + assert, + cleanup, + }) => { + const secondConnection = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + }) + cleanup(async () => { + await secondConnection.quit() + }) + + const adapter = new RedisAdapter(connection) + const secondAdapter = new RedisAdapter(secondConnection) + const id = 'concurrent-resume-schedule' + + await adapter.upsertSchedule({ + id, + name: 'ConcurrentJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }) + + for (let i = 0; i < 20; i++) { + const nextRunAt = Date.now() + 30_000 + i + await adapter.updateSchedule(id, { status: 'paused' }) + + await Promise.all([ + adapter.updateSchedule(id, { nextRunAt: new Date(nextRunAt) }), + secondAdapter.updateSchedule(id, { status: 'active' }), + ]) + + const schedule = await adapter.getSchedule(id) + assert.equal(schedule!.status, 'active') + assert.equal(Number(await connection.zscore('schedules::due', id)), nextRunAt) + } + }) + + test('concurrent upsert and next-run updates leave the due index canonical', async ({ + assert, + cleanup, + }) => { + const secondConnection = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + }) + cleanup(async () => { + await secondConnection.quit() + }) + + const adapter = new RedisAdapter(connection) + const secondAdapter = new RedisAdapter(secondConnection) + const id = 'concurrent-upsert-schedule' + + await adapter.upsertSchedule({ + id, + name: 'ConcurrentJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }) + + for (let i = 0; i < 20; i++) { + const nextRunAt = Date.now() + 60_000 + i + await adapter.updateSchedule(id, { status: 'paused' }) + + await Promise.all([ + adapter.updateSchedule(id, { nextRunAt: new Date(nextRunAt) }), + secondAdapter.upsertSchedule({ + id, + name: 'ConcurrentJob', + payload: { iteration: i }, + everyMs: 60_000, + timezone: 'UTC', + }), + ]) + + const schedule = await adapter.getSchedule(id) + assert.equal(schedule!.status, 'active') + assert.equal(schedule!.nextRunAt!.getTime(), nextRunAt) + assert.equal(Number(await connection.zscore('schedules::due', id)), nextRunAt) + } + }) + test('interval claims update the due index from the canonical hash', async ({ assert }) => { const adapter = new RedisAdapter(connection) const id = 'interval-index-schedule' From cf901373d575bcaf708656cf514219b934f0081e Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 12:30:20 +0000 Subject: [PATCH 09/14] 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. --- src/drivers/redis_adapter.ts | 24 ++++----- src/drivers/redis_scripts.ts | 48 +++++++++++++++-- tests/adapter.spec.ts | 102 +++++++++++++++++++++++++++++++++++ 3 files changed, 159 insertions(+), 15 deletions(-) diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 17f3069..9005459 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -19,6 +19,7 @@ import { BACKFILL_SCHEDULE_DUE_INDEX_SCRIPT, CLAIM_SCHEDULE_SCRIPT, FINALIZE_JOB_SCRIPT, + FINALIZE_CRON_SCHEDULE_SCRIPT, GET_JOB_SCRIPT, PUSH_DEDUP_JOB_SCRIPT, PUSH_DELAYED_JOB_SCRIPT, @@ -601,18 +602,17 @@ export class RedisAdapter implements Adapter { newNextRunAt = '' } - const scheduleKey = `${schedulesKey}::${data.id}` - const multi = this.#connection - .multi() - .hset(scheduleKey, 'next_run_at', newNextRunAt.toString()) - - if (typeof newNextRunAt === 'number') { - multi.zadd(schedulesDueKey, newNextRunAt, data.id) - } else { - multi.zrem(schedulesDueKey, data.id) - } - - await multi.exec() + await this.#connection.eval( + FINALIZE_CRON_SCHEDULE_SCRIPT, + 2, + `${schedulesKey}::${data.id}`, + schedulesDueKey, + data.id, + runCount.toString(), + now.toString(), + data.cron_expression, + newNextRunAt.toString() + ) } return this.#hashToScheduleData(data) diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index 0af0776..15e7a61 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -564,6 +564,47 @@ ${SCHEDULE_DUE_INDEX_LUA} return 1 ` +/** + * Finalizes the JS-calculated next run for a cron claim only when the same + * claimed occurrence still owns the canonical hash. + */ +export const FINALIZE_CRON_SCHEDULE_SCRIPT = ` + local schedule_key = KEYS[1] + local due_key = KEYS[2] + local id = ARGV[1] + local expected_run_count = ARGV[2] + local expected_last_run_at = ARGV[3] + local expected_cron_expression = ARGV[4] + local next_run_at = ARGV[5] + +${SCHEDULE_DUE_INDEX_LUA} + + if redis.call('EXISTS', schedule_key) == 0 then + redis.call('ZREM', due_key, id) + return 0 + end + + local status = redis.call('HGET', schedule_key, 'status') + local run_count = redis.call('HGET', schedule_key, 'run_count') + local last_run_at = redis.call('HGET', schedule_key, 'last_run_at') + local current_next_run_at = redis.call('HGET', schedule_key, 'next_run_at') + local cron_expression = redis.call('HGET', schedule_key, 'cron_expression') + + if status ~= 'active' + or run_count ~= expected_run_count + or last_run_at ~= expected_last_run_at + or current_next_run_at ~= '' + or cron_expression ~= expected_cron_expression then + sync_schedule_due_index(schedule_key, due_key, id) + return 0 + end + + redis.call('HSET', schedule_key, 'next_run_at', next_run_at) + sync_schedule_due_index(schedule_key, due_key, id) + + return 1 +` + /** * Atomically rebuilds the derived due index from canonical schedule hashes. * This is an explicit O(N) migration and blocks concurrent Redis commands @@ -642,10 +683,11 @@ export const CLAIM_SCHEDULE_SCRIPT = ` -- Hash is the source of truth for next_run_at. -- If the ZSET score is stale, repair it and skip this candidate. local hash_nra = schedule.next_run_at - if not hash_nra or hash_nra == '' then + local hash_score = hash_nra and tonumber(hash_nra) or nil + if not hash_score then redis.call('ZREM', due_key, id) - elseif tonumber(hash_nra) > now then - redis.call('ZADD', due_key, tonumber(hash_nra), id) + elseif hash_score > now then + redis.call('ZADD', due_key, hash_score, id) else local run_count = tonumber(schedule.run_count or '0') local run_limit = schedule.run_limit and tonumber(schedule.run_limit) or nil diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index c79981e..4ab1421 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -979,6 +979,108 @@ test.group('Adapter | Redis', (group) => { assert.equal(Number(score), schedule!.nextRunAt!.getTime()) }) + test('cron finalization does not undo a concurrent pause or delete', async ({ + assert, + cleanup, + }) => { + const secondConnection = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + }) + cleanup(async () => { + await secondConnection.quit() + }) + + const adapter = new RedisAdapter(connection) + const secondAdapter = new RedisAdapter(secondConnection) + + for (const mutation of ['pause', 'delete'] as const) { + const id = `cron-finalize-${mutation}` + await adapter.upsertSchedule({ + id, + name: 'CronFinalizeJob', + payload: {}, + cronExpression: '* * * * *', + timezone: 'UTC', + }) + await adapter.updateSchedule(id, { nextRunAt: new Date(Date.now() - 1_000) }) + + const originalEval = connection.eval.bind(connection) + let releaseClaim!: () => void + let claimReturned!: () => void + const claimReleased = new Promise((resolve) => { + releaseClaim = resolve + }) + const claimHasReturned = new Promise((resolve) => { + claimReturned = resolve + }) + let gateNextEval = true + + connection.eval = (async (...args: Parameters) => { + const result = await originalEval(...args) + if (gateNextEval) { + gateNextEval = false + claimReturned() + await claimReleased + } + return result + }) as typeof connection.eval + + const claim = adapter.claimDueSchedule() + await claimHasReturned + + if (mutation === 'pause') { + await secondAdapter.updateSchedule(id, { status: 'paused' }) + } else { + await secondAdapter.deleteSchedule(id) + } + + releaseClaim() + await claim + connection.eval = originalEval + + if (mutation === 'pause') { + assert.equal((await adapter.getSchedule(id))!.status, 'paused') + } else { + assert.isNull(await adapter.getSchedule(id)) + } + assert.isNull(await connection.zscore('schedules::due', id)) + } + }) + + test('claim removes a malformed due score and continues to a valid schedule', async ({ + assert, + }) => { + const adapter = new RedisAdapter(connection) + const malformedId = 'malformed-next-run-at' + const validId = 'valid-after-malformed' + + await connection + .multi() + .hset(`schedules::${malformedId}`, { + id: malformedId, + status: 'active', + next_run_at: 'not-a-number', + }) + .sadd('schedules::index', malformedId) + .zadd('schedules::due', Date.now() - 2_000, malformedId) + .exec() + + await adapter.upsertSchedule({ + id: validId, + name: 'ValidJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }) + await adapter.updateSchedule(validId, { nextRunAt: new Date(Date.now() - 1_000) }) + + assert.equal((await adapter.claimDueSchedule())!.id, validId) + assert.isNull(await connection.zscore('schedules::due', malformedId)) + }) + test('exhausted schedules are removed from the due index', async ({ assert }) => { const adapter = new RedisAdapter(connection) From 992ce18cbf22d39937bea94fa13a8dd7bb8978bb Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 12:37:30 +0000 Subject: [PATCH 10/14] fix(redis): reject superseded cron finalization --- src/drivers/redis_adapter.ts | 1 + src/drivers/redis_scripts.ts | 19 ++++++++-- tests/adapter.spec.ts | 69 ++++++++++++++++++++++++++++++++++++ 3 files changed, 86 insertions(+), 3 deletions(-) diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 9005459..417b2ad 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -611,6 +611,7 @@ export class RedisAdapter implements Adapter { runCount.toString(), now.toString(), data.cron_expression, + data.config_revision || '', newNextRunAt.toString() ) } diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index 15e7a61..eb1b0c8 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -521,6 +521,7 @@ ${SCHEDULE_DUE_INDEX_LUA} local run_count = redis.call('HGET', schedule_key, 'run_count') or '0' local created_at = redis.call('HGET', schedule_key, 'created_at') or now + local config_revision = tonumber(redis.call('HGET', schedule_key, 'config_revision') or '0') + 1 redis.call( 'HDEL', @@ -536,7 +537,16 @@ ${SCHEDULE_DUE_INDEX_LUA} redis.call('HSET', schedule_key, field, value) end - redis.call('HSET', schedule_key, 'run_count', run_count, 'created_at', created_at) + redis.call( + 'HSET', + schedule_key, + 'run_count', + run_count, + 'created_at', + created_at, + 'config_revision', + tostring(config_revision) + ) redis.call('SADD', schedules_index_key, id) sync_schedule_due_index(schedule_key, due_key, id) @@ -575,7 +585,8 @@ export const FINALIZE_CRON_SCHEDULE_SCRIPT = ` local expected_run_count = ARGV[2] local expected_last_run_at = ARGV[3] local expected_cron_expression = ARGV[4] - local next_run_at = ARGV[5] + local expected_config_revision = ARGV[5] + local next_run_at = ARGV[6] ${SCHEDULE_DUE_INDEX_LUA} @@ -589,12 +600,14 @@ ${SCHEDULE_DUE_INDEX_LUA} local last_run_at = redis.call('HGET', schedule_key, 'last_run_at') local current_next_run_at = redis.call('HGET', schedule_key, 'next_run_at') local cron_expression = redis.call('HGET', schedule_key, 'cron_expression') + local config_revision = redis.call('HGET', schedule_key, 'config_revision') or '' if status ~= 'active' or run_count ~= expected_run_count or last_run_at ~= expected_last_run_at or current_next_run_at ~= '' - or cron_expression ~= expected_cron_expression then + or cron_expression ~= expected_cron_expression + or config_revision ~= expected_config_revision then sync_schedule_due_index(schedule_key, due_key, id) return 0 end diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 4ab1421..98f54cd 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -1050,6 +1050,75 @@ test.group('Adapter | Redis', (group) => { } }) + test('cron finalization does not apply a calculation from superseded configuration', async ({ + assert, + cleanup, + }) => { + const secondConnection = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + }) + cleanup(async () => { + await secondConnection.quit() + }) + + const adapter = new RedisAdapter(connection) + const secondAdapter = new RedisAdapter(secondConnection) + const id = 'cron-finalize-superseded-config' + + await adapter.upsertSchedule({ + id, + name: 'OriginalCronJob', + payload: {}, + cronExpression: '0 9 * * *', + timezone: 'UTC', + }) + await adapter.updateSchedule(id, { nextRunAt: new Date(Date.now() - 1_000) }) + + const originalEval = connection.eval.bind(connection) + let releaseClaim!: () => void + let claimReturned!: () => void + const claimReleased = new Promise((resolve) => { + releaseClaim = resolve + }) + const claimHasReturned = new Promise((resolve) => { + claimReturned = resolve + }) + let gateNextEval = true + + connection.eval = (async (...args: Parameters) => { + const result = await originalEval(...args) + if (gateNextEval) { + gateNextEval = false + claimReturned() + await claimReleased + } + return result + }) as typeof connection.eval + + const claim = adapter.claimDueSchedule() + await claimHasReturned + await secondAdapter.upsertSchedule({ + id, + name: 'UpdatedCronJob', + payload: {}, + cronExpression: '0 9 * * *', + timezone: 'America/New_York', + }) + + releaseClaim() + await claim + connection.eval = originalEval + + const schedule = await adapter.getSchedule(id) + assert.equal(schedule!.name, 'UpdatedCronJob') + assert.equal(schedule!.timezone, 'America/New_York') + assert.isNull(schedule!.nextRunAt) + assert.isNull(await connection.zscore('schedules::due', id)) + }) + test('claim removes a malformed due score and continues to a valid schedule', async ({ assert, }) => { From 738d64c6146454a2ffce4da47199ba0f3ddd9655 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 12:49:36 +0000 Subject: [PATCH 11/14] style(redis): clarify claim script nesting --- src/drivers/redis_scripts.ts | 78 ++++++++++++++++++------------------ 1 file changed, 39 insertions(+), 39 deletions(-) diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index eb1b0c8..2cd534f 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -702,51 +702,51 @@ export const CLAIM_SCHEDULE_SCRIPT = ` elseif hash_score > now then redis.call('ZADD', due_key, hash_score, id) else - local run_count = tonumber(schedule.run_count or '0') - local run_limit = schedule.run_limit and tonumber(schedule.run_limit) or nil - local to_date = schedule.to_date and tonumber(schedule.to_date) or nil + local run_count = tonumber(schedule.run_count or '0') + local run_limit = schedule.run_limit and tonumber(schedule.run_limit) or nil + local to_date = schedule.to_date and tonumber(schedule.to_date) or nil - -- Check limits - if (run_limit and run_count >= run_limit) or (to_date and now > to_date) then - redis.call('ZREM', due_key, id) - else - -- This schedule is claimable - atomically update it - local new_run_count = run_count + 1 - - -- Calculate new next_run_at (simple interval-based for now) - -- Complex cron calculation happens in the caller - local new_next_run_at = '' - local every_ms = schedule.every_ms and tonumber(schedule.every_ms) or nil - if every_ms then - new_next_run_at = tostring(now + every_ms) - end + -- Check limits + if (run_limit and run_count >= run_limit) or (to_date and now > to_date) then + redis.call('ZREM', due_key, id) + else + -- This schedule is claimable - atomically update it + local new_run_count = run_count + 1 + + -- Calculate new next_run_at (simple interval-based for now) + -- Complex cron calculation happens in the caller + local new_next_run_at = '' + local every_ms = schedule.every_ms and tonumber(schedule.every_ms) or nil + if every_ms then + new_next_run_at = tostring(now + every_ms) + end - -- Check if we've hit the limit after this run - if run_limit and new_run_count >= run_limit then - new_next_run_at = '' - end + -- Check if we've hit the limit after this run + if run_limit and new_run_count >= run_limit then + new_next_run_at = '' + end - -- Check if past end date - if to_date and new_next_run_at ~= '' and tonumber(new_next_run_at) > to_date then - new_next_run_at = '' - end + -- Check if past end date + if to_date and new_next_run_at ~= '' and tonumber(new_next_run_at) > to_date then + new_next_run_at = '' + end - -- Update the schedule atomically - redis.call('HSET', schedule_key, - 'next_run_at', new_next_run_at, - 'last_run_at', tostring(now), - 'run_count', tostring(new_run_count)) + -- Update the schedule atomically + redis.call('HSET', schedule_key, + 'next_run_at', new_next_run_at, + 'last_run_at', tostring(now), + 'run_count', tostring(new_run_count)) + + -- Update or remove from ZSET + if new_next_run_at ~= '' then + redis.call('ZADD', due_key, tonumber(new_next_run_at), id) + else + redis.call('ZREM', due_key, id) + end - -- Update or remove from ZSET - if new_next_run_at ~= '' then - redis.call('ZADD', due_key, tonumber(new_next_run_at), id) - else - redis.call('ZREM', due_key, id) + -- Return the schedule data (before update) as JSON + return cjson.encode(schedule) end - - -- Return the schedule data (before update) as JSON - return cjson.encode(schedule) - end end end end From 4acedbdca69f60934ff6cf3cd9e18dabb71d8810 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 13:01:21 +0000 Subject: [PATCH 12/14] docs: note Redis schedule race safeguards --- .changelog/redis-schedule-due-index.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.changelog/redis-schedule-due-index.md b/.changelog/redis-schedule-due-index.md index f37432b..d661a83 100644 --- a/.changelog/redis-schedule-due-index.md +++ b/.changelog/redis-schedule-due-index.md @@ -10,6 +10,11 @@ Schedule hashes remain the source of truth. Creating, updating, pausing, resumin claiming schedules maintain the derived index, while claiming repairs stale entries when the hash and index disagree. +Schedule hash and index writes are atomic, and index consistency is preserved across concurrent +lifecycle changes. Cron finalization now rejects stale calculations when a schedule is paused, +deleted, or reconfigured while its next run is being calculated. Claiming also discards malformed +due scores instead of allowing one corrupt entry to block later schedules. + ## Upgrade Notes This change requires an explicit migration for existing Redis schedules. The `Adapter` contract now From e043cdba955bdff6e377a91fffee521f1c11e401 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 13:20:33 +0000 Subject: [PATCH 13/14] fix(redis): preserve schedules across concurrent updates --- src/drivers/redis_adapter.ts | 2 - src/drivers/redis_scripts.ts | 27 +++++++---- tests/adapter.spec.ts | 92 ++++++++++++++++++++++++++++++++++++ 3 files changed, 110 insertions(+), 11 deletions(-) diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 417b2ad..0c20c85 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -608,8 +608,6 @@ export class RedisAdapter implements Adapter { `${schedulesKey}::${data.id}`, schedulesDueKey, data.id, - runCount.toString(), - now.toString(), data.cron_expression, data.config_revision || '', newNextRunAt.toString() diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index 2cd534f..5af310d 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -565,6 +565,11 @@ export const UPDATE_SCHEDULE_SCRIPT = ` ${SCHEDULE_DUE_INDEX_LUA} + if redis.call('EXISTS', schedule_key) == 0 then + redis.call('ZREM', due_key, id) + return 0 + end + for field, value in pairs(updates) do redis.call('HSET', schedule_key, field, value) end @@ -582,11 +587,9 @@ export const FINALIZE_CRON_SCHEDULE_SCRIPT = ` local schedule_key = KEYS[1] local due_key = KEYS[2] local id = ARGV[1] - local expected_run_count = ARGV[2] - local expected_last_run_at = ARGV[3] - local expected_cron_expression = ARGV[4] - local expected_config_revision = ARGV[5] - local next_run_at = ARGV[6] + local expected_cron_expression = ARGV[2] + local expected_config_revision = ARGV[3] + local next_run_at = ARGV[4] ${SCHEDULE_DUE_INDEX_LUA} @@ -596,15 +599,11 @@ ${SCHEDULE_DUE_INDEX_LUA} end local status = redis.call('HGET', schedule_key, 'status') - local run_count = redis.call('HGET', schedule_key, 'run_count') - local last_run_at = redis.call('HGET', schedule_key, 'last_run_at') local current_next_run_at = redis.call('HGET', schedule_key, 'next_run_at') local cron_expression = redis.call('HGET', schedule_key, 'cron_expression') local config_revision = redis.call('HGET', schedule_key, 'config_revision') or '' if status ~= 'active' - or run_count ~= expected_run_count - or last_run_at ~= expected_last_run_at or current_next_run_at ~= '' or cron_expression ~= expected_cron_expression or config_revision ~= expected_config_revision then @@ -612,6 +611,16 @@ ${SCHEDULE_DUE_INDEX_LUA} return 0 end + local run_count = tonumber(redis.call('HGET', schedule_key, 'run_count') or '0') + local run_limit = tonumber(redis.call('HGET', schedule_key, 'run_limit') or '') + local to_date = tonumber(redis.call('HGET', schedule_key, 'to_date') or '') + local next_score = tonumber(next_run_at) + + if (run_limit and run_count >= run_limit) + or (to_date and next_score and next_score > to_date) then + next_run_at = '' + end + redis.call('HSET', schedule_key, 'next_run_at', next_run_at) sync_schedule_due_index(schedule_key, due_key, id) diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 98f54cd..8914783 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -1119,6 +1119,98 @@ test.group('Adapter | Redis', (group) => { assert.isNull(await connection.zscore('schedules::due', id)) }) + test('cron finalization survives a concurrent manual trigger metadata update', async ({ + assert, + cleanup, + }) => { + const secondConnection = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + }) + cleanup(async () => { + await secondConnection.quit() + }) + + const adapter = new RedisAdapter(connection) + const secondAdapter = new RedisAdapter(secondConnection) + const id = 'cron-finalize-concurrent-trigger' + + await adapter.upsertSchedule({ + id, + name: 'CronTriggerJob', + payload: {}, + cronExpression: '* * * * *', + timezone: 'UTC', + }) + await adapter.updateSchedule(id, { nextRunAt: new Date(Date.now() - 1_000) }) + + const originalEval = connection.eval.bind(connection) + let releaseClaim!: () => void + let claimReturned!: () => void + const claimReleased = new Promise((resolve) => { + releaseClaim = resolve + }) + const claimHasReturned = new Promise((resolve) => { + claimReturned = resolve + }) + let gateNextEval = true + + connection.eval = (async (...args: Parameters) => { + const result = await originalEval(...args) + if (gateNextEval) { + gateNextEval = false + claimReturned() + await claimReleased + } + return result + }) as typeof connection.eval + + const claim = adapter.claimDueSchedule() + await claimHasReturned + await secondAdapter.updateSchedule(id, { + runCount: 2, + lastRunAt: new Date(), + }) + + releaseClaim() + await claim + connection.eval = originalEval + + const schedule = await adapter.getSchedule(id) + assert.equal(schedule!.runCount, 2) + assert.isNotNull(schedule!.nextRunAt) + assert.equal( + Number(await connection.zscore('schedules::due', id)), + schedule!.nextRunAt!.getTime() + ) + }) + + test('updating a deleted schedule does not recreate or index it', async ({ assert }) => { + const adapter = new RedisAdapter(connection) + const id = 'update-after-delete' + + await adapter.upsertSchedule({ + id, + name: 'DeletedScheduleJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + }) + await adapter.deleteSchedule(id) + + await adapter.updateSchedule(id, { + status: 'active', + nextRunAt: new Date(Date.now() - 1_000), + runCount: 0, + }) + + assert.isNull(await adapter.getSchedule(id)) + assert.isNull(await connection.zscore('schedules::due', id)) + assert.isNull(await adapter.claimDueSchedule()) + }) + test('claim removes a malformed due score and continues to a valid schedule', async ({ assert, }) => { From c1bcb9433946ff1f1556cd1e7410ce42584c1533 Mon Sep 17 00:00:00 2001 From: Romain Lanz Date: Wed, 26 Aug 2026 13:36:15 +0000 Subject: [PATCH 14/14] fix(redis): guard cron claim ownership --- src/drivers/redis_adapter.ts | 5 +- src/drivers/redis_scripts.ts | 21 +++++-- tests/adapter.spec.ts | 110 ++++++++++++++++++++++++++++++++++- 3 files changed, 130 insertions(+), 6 deletions(-) diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index 0c20c85..9740434 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -565,12 +565,14 @@ export class RedisAdapter implements Adapter { async claimDueSchedule(): Promise { const now = Date.now() + const claimToken = randomUUID() const result = await this.#connection.eval( CLAIM_SCHEDULE_SCRIPT, 2, schedulesDueKey, `${schedulesKey}::`, - now.toString() + now.toString(), + claimToken ) if (!result) { @@ -610,6 +612,7 @@ export class RedisAdapter implements Adapter { data.id, data.cron_expression, data.config_revision || '', + claimToken, newNextRunAt.toString() ) } diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index 5af310d..351fbf1 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -530,7 +530,8 @@ ${SCHEDULE_DUE_INDEX_LUA} 'every_ms', 'from_date', 'to_date', - 'run_limit' + 'run_limit', + 'claim_token' ) for field, value in pairs(schedule) do @@ -589,7 +590,8 @@ export const FINALIZE_CRON_SCHEDULE_SCRIPT = ` local id = ARGV[1] local expected_cron_expression = ARGV[2] local expected_config_revision = ARGV[3] - local next_run_at = ARGV[4] + local expected_claim_token = ARGV[4] + local next_run_at = ARGV[5] ${SCHEDULE_DUE_INDEX_LUA} @@ -602,11 +604,16 @@ ${SCHEDULE_DUE_INDEX_LUA} local current_next_run_at = redis.call('HGET', schedule_key, 'next_run_at') local cron_expression = redis.call('HGET', schedule_key, 'cron_expression') local config_revision = redis.call('HGET', schedule_key, 'config_revision') or '' + local claim_token = redis.call('HGET', schedule_key, 'claim_token') or '' if status ~= 'active' or current_next_run_at ~= '' or cron_expression ~= expected_cron_expression - or config_revision ~= expected_config_revision then + or config_revision ~= expected_config_revision + or claim_token ~= expected_claim_token then + if claim_token == expected_claim_token then + redis.call('HDEL', schedule_key, 'claim_token') + end sync_schedule_due_index(schedule_key, due_key, id) return 0 end @@ -622,6 +629,7 @@ ${SCHEDULE_DUE_INDEX_LUA} end redis.call('HSET', schedule_key, 'next_run_at', next_run_at) + redis.call('HDEL', schedule_key, 'claim_token') sync_schedule_due_index(schedule_key, due_key, id) return 1 @@ -674,6 +682,7 @@ export const CLAIM_SCHEDULE_SCRIPT = ` local due_key = KEYS[1] local prefix = KEYS[2] local now = tonumber(ARGV[1]) + local claim_token = ARGV[2] while true do local candidates = redis.call('ZRANGEBYSCORE', due_key, '-inf', tostring(now), 'LIMIT', 0, 1) @@ -721,6 +730,7 @@ export const CLAIM_SCHEDULE_SCRIPT = ` else -- This schedule is claimable - atomically update it local new_run_count = run_count + 1 + local new_claim_token = '' -- Calculate new next_run_at (simple interval-based for now) -- Complex cron calculation happens in the caller @@ -728,6 +738,8 @@ export const CLAIM_SCHEDULE_SCRIPT = ` local every_ms = schedule.every_ms and tonumber(schedule.every_ms) or nil if every_ms then new_next_run_at = tostring(now + every_ms) + elseif schedule.cron_expression then + new_claim_token = claim_token end -- Check if we've hit the limit after this run @@ -744,7 +756,8 @@ export const CLAIM_SCHEDULE_SCRIPT = ` redis.call('HSET', schedule_key, 'next_run_at', new_next_run_at, 'last_run_at', tostring(now), - 'run_count', tostring(new_run_count)) + 'run_count', tostring(new_run_count), + 'claim_token', new_claim_token) -- Update or remove from ZSET if new_next_run_at ~= '' then diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 8914783..6d07e8c 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -1119,7 +1119,7 @@ test.group('Adapter | Redis', (group) => { assert.isNull(await connection.zscore('schedules::due', id)) }) - test('cron finalization survives a concurrent manual trigger metadata update', async ({ + test('cron finalization survives a concurrent runtime metadata update', async ({ assert, cleanup, }) => { @@ -1187,6 +1187,114 @@ test.group('Adapter | Redis', (group) => { ) }) + test('stale cron finalization cannot modify a recreated schedule claim', async ({ + assert, + cleanup, + }) => { + const secondConnection = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + }) + cleanup(async () => { + await secondConnection.quit() + }) + + const adapter = new RedisAdapter(connection) + const secondAdapter = new RedisAdapter(secondConnection) + const id = 'cron-finalize-recreated-claim' + const cronExpression = '0 9 * * *' + + await adapter.upsertSchedule({ + id, + name: 'OriginalCronJob', + payload: {}, + cronExpression, + timezone: 'UTC', + }) + await adapter.updateSchedule(id, { nextRunAt: new Date(Date.now() - 1_000) }) + + const originalEval = connection.eval.bind(connection) + let releaseFirstClaim!: () => void + let firstClaimReturned!: () => void + const firstClaimReleased = new Promise((resolve) => { + releaseFirstClaim = resolve + }) + const firstClaimHasReturned = new Promise((resolve) => { + firstClaimReturned = resolve + }) + let gateFirstEval = true + + connection.eval = (async (...args: Parameters) => { + const result = await originalEval(...args) + if (gateFirstEval) { + gateFirstEval = false + firstClaimReturned() + await firstClaimReleased + } + return result + }) as typeof connection.eval + + const firstClaim = adapter.claimDueSchedule() + await firstClaimHasReturned + + await secondAdapter.deleteSchedule(id) + await secondAdapter.upsertSchedule({ + id, + name: 'RecreatedCronJob', + payload: {}, + cronExpression, + timezone: 'America/New_York', + }) + await secondAdapter.updateSchedule(id, { nextRunAt: new Date(Date.now() - 1_000) }) + + const originalSecondEval = secondConnection.eval.bind(secondConnection) + let releaseSecondClaim!: () => void + let secondClaimReturned!: () => void + const secondClaimReleased = new Promise((resolve) => { + releaseSecondClaim = resolve + }) + const secondClaimHasReturned = new Promise((resolve) => { + secondClaimReturned = resolve + }) + let gateSecondEval = true + + secondConnection.eval = (async (...args: Parameters) => { + const result = await originalSecondEval(...args) + if (gateSecondEval) { + gateSecondEval = false + secondClaimReturned() + await secondClaimReleased + } + return result + }) as typeof secondConnection.eval + + const secondClaim = secondAdapter.claimDueSchedule() + await secondClaimHasReturned + + releaseFirstClaim() + await firstClaim + + const awaitingSecondFinalization = await adapter.getSchedule(id) + assert.equal(awaitingSecondFinalization!.name, 'RecreatedCronJob') + assert.equal(awaitingSecondFinalization!.timezone, 'America/New_York') + assert.isNull(awaitingSecondFinalization!.nextRunAt) + assert.isNull(await connection.zscore('schedules::due', id)) + + releaseSecondClaim() + await secondClaim + connection.eval = originalEval + secondConnection.eval = originalSecondEval + + const finalized = await adapter.getSchedule(id) + assert.isNotNull(finalized!.nextRunAt) + assert.equal( + Number(await connection.zscore('schedules::due', id)), + finalized!.nextRunAt!.getTime() + ) + }) + test('updating a deleted schedule does not recreate or index it', async ({ assert }) => { const adapter = new RedisAdapter(connection) const id = 'update-after-delete'