fix(redis): revoke cron claims when the next run is updated

A cron claim clears next_run_at and stores a claim token. The worker then
computes the next run in JS and finalizes it only if next_run_at is still
empty, the cron expression and config revision are unchanged, and the
token still matches.

updateSchedule({ nextRunAt: null }) writes the same empty next_run_at but
kept the token and the revision, so the finalizer could not tell the
difference and wrote the next run back. updateSchedule() now removes the
claim token whenever it sets next_run_at.

Author: RomainLanz

.changelog/redis-schedule-due-index.md

@@ -11,10 +11,10 @@ claiming schedules maintain the derived index, while claiming repairs stale entr
 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 deleted or
-reconfigured while its next run is being calculated. Paused schedules retain their next run but stay
-out of the due index until resumed. Claiming also discards malformed due scores instead of allowing
-one corrupt entry to block later schedules.
+lifecycle changes. Cron finalization now rejects stale calculations when a schedule is deleted,
+reconfigured, or has its next run cleared during the calculation. Paused schedules retain their
+next run but stay out of the due index until resumed. Claiming also discards malformed due scores
+instead of allowing one corrupt entry to block later schedules.
 
 ## Bug Fix
 

src/drivers/redis_scripts.ts

@@ -575,6 +575,12 @@ ${SCHEDULE_DUE_INDEX_LUA}
     redis.call('HSET', schedule_key, field, value)
   end
 
+  -- An explicit next run supersedes an in-flight cron claim, so the claim's
+  -- finalization must not overwrite it.
+  if updates.next_run_at ~= nil then
+    redis.call('HDEL', schedule_key, 'claim_token')
+  end
+
   sync_schedule_due_index(schedule_key, due_key, id)
 
   return 1

tests/adapter.spec.ts

@@ -1333,6 +1333,64 @@ test.group('Adapter | Redis', (group) => {
     )
   })
 
+  test('cron finalization does not undo a cleared next run', 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-cleared-next-run'
+
+    await adapter.upsertSchedule({
+      id,
+      name: 'CronClearedJob',
+      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<void>((resolve) => {
+      releaseClaim = resolve
+    })
+    const claimHasReturned = new Promise<void>((resolve) => {
+      claimReturned = resolve
+    })
+    let gateNextEval = true
+
+    connection.eval = (async (...args: Parameters<typeof connection.eval>) => {
+      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, { nextRunAt: null })
+
+    releaseClaim()
+    await claim
+    connection.eval = originalEval
+
+    const schedule = await adapter.getSchedule(id)
+    assert.isNull(schedule!.nextRunAt)
+    assert.isNull(await connection.zscore('schedules::due', id))
+  })
+
   test('stale cron finalization cannot modify a recreated schedule claim', async ({
     assert,
     cleanup,