Fix scheduler claim model to avoid silently dropping occurrences - #11
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: f004b4e1aa
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| or( | ||
| isNull(schedules.claimedAt), | ||
| lt(schedules.claimedAt, staleBeforeIso) | ||
| ) |
There was a problem hiding this comment.
Reclaiming rows when claimed_at is older than 10 minutes allows a second worker to process the same schedule while the first worker is still running, and this commit does not prevent the original worker from writing state after it has lost ownership. In scheduled.ts, monitor execution calls updateState(...) before lock ownership is checked, and updateState updates by id only, so a long-running first attempt can overwrite the newer worker’s state_json after the row is reclaimed. This can regress monitor state and cause duplicate/incorrect notifications whenever execution exceeds the stale-lock window.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 5944d73: updateState is now lock-guarded — the UPDATE WHERE matches the caller's claimed_at, returns { lockLost }. Monitor execution in scheduled.ts checks lockLost and logs a warning instead of silently overwriting state from a re-claimed row.
There was a problem hiding this comment.
Pull request overview
Updates the scheduler claiming mechanism to use an explicit claimed_at lock so next_run_at is only advanced after a run is confirmed successful (or retries are exhausted), preventing ambiguous D1 updates from silently skipping occurrences.
Changes:
- Replace “claim by advancing
next_run_at” with aclaimed_atlock + stale-lock reclamation. - Update worker completion paths to advance
next_run_aton success/exhaustion and to surfacelockLoston completion updates. - Add DB migration + schema updates for the new
claimed_atcolumn, plus extra scheduled-handler logging.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| apps/operator/src/services/schedule.ts | Implements claimed_at-based claiming, stale lock reclamation, and lock-guarded completion updates. |
| apps/operator/src/scheduled.ts | Logs claim errors with causes, logs “no due schedules”, and handles lockLost on success/failure recording. |
| apps/operator/src/db/schema.ts | Adds claimedAt column to the Drizzle schema. |
| apps/operator/migrations/0003_amused_marvex.sql | Adds claimed_at to the schedules table. |
| apps/operator/migrations/meta/_journal.json | Registers migration 0003 in the migration journal. |
| apps/operator/migrations/meta/0003_snapshot.json | Captures the updated schema snapshot including claimed_at. |
| .claude/commands/pr-open.md | Adds a helper command doc for opening GitHub PRs via gh. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // Handle failures — mark for retry | ||
| const completionTime = new Date(); | ||
| for (let i = 0; i < results.length; i++) { | ||
| const result = results[i]; | ||
| const schedule = claimed[i]; | ||
| if (result.status === "fulfilled") { | ||
| await scheduleService.markSuccess(schedule.id); | ||
| const { lockLost } = await scheduleService.markSuccess( | ||
| schedule, | ||
| completionTime | ||
| ); |
There was a problem hiding this comment.
markSuccess/markFailed are being passed a single completionTime captured after all tasks finish, and nextRunAt is computed from that time. For short-interval schedules (e.g., hourly), if execution runs past the next scheduled boundary, this will advance nextRunAt beyond the immediate next occurrence (silently skipping it). Consider computing the next occurrence from the schedule’s current nextRunAt (the occurrence being processed) or from the claim time instead of the batch completion time, so long-running executions don’t drop intermediate occurrences.
There was a problem hiding this comment.
Fixed in 5944d73: markSuccess / markFailed (exhausted) now compute the next occurrence from row.nextRunAt (the slot just processed), not from batch completion time. Long-running batches can no longer skip intermediate occurrences for short-interval schedules.
| async claimDueSchedules( | ||
| now: Date, | ||
| allowedChatId: string | ||
| ): Promise<ClaimedSchedule[]> { | ||
| const nowIso = now.toISOString(); |
There was a problem hiding this comment.
The new claimed_at locking / stale-lock reclaim behavior introduced in claimDueSchedules (and the claimedAt guards used by markSuccess/markFailed) is core correctness logic but isn’t covered by unit tests (schedule.test.ts currently covers computeNextRun + schema validation only). Adding unit tests for claiming when lock is free/stale, refusing to claim when the row changes concurrently, and returning lockLost when claimedAt doesn’t match would help prevent regressions.
There was a problem hiding this comment.
Agreed — these paths need real coverage. The existing suite has no D1/drizzle fixture (only function-level tests for computeNextRun + schema validation), so adding meaningful claim/lock tests means standing up Miniflare D1 emulation or an equivalent fake. Tracking as follow-up rather than landing brittle mocked tests in this PR.
| const lockGuard = | ||
| row.claimedAt === null | ||
| ? isNull(schedules.claimedAt) | ||
| : eq(schedules.claimedAt, row.claimedAt); | ||
| const result = await this.db | ||
| .update(schedules) | ||
| .set({ nextRunAt: nextRun.toISOString() }) | ||
| .where( | ||
| and(eq(schedules.id, row.id), eq(schedules.nextRunAt, row.nextRunAt)) | ||
| ); | ||
| .set({ claimedAt: nowIso }) | ||
| .where(and(eq(schedules.id, row.id), lockGuard)); |
There was a problem hiding this comment.
In claimDueSchedules, the UPDATE that sets claimedAt only guards on (id, claimed_at). If another worker completes the schedule between the SELECT and this UPDATE (advancing nextRunAt and clearing claimedAt back to NULL), this code can still successfully re-claim the row even though it is no longer due, leading to duplicate/early executions. Add a guard on nextRunAt as well (e.g., ensure schedules.nextRunAt still equals the selected row.nextRunAt, or otherwise re-check due-ness in the UPDATE WHERE).
There was a problem hiding this comment.
Fixed in 5944d73: the claim UPDATE now also guards on schedules.nextRunAt = row.nextRunAt, so a row completed concurrently between SELECT and UPDATE can't be re-claimed.
What
next_run_atfor an unrun occurrence, silently dropping it; switch to aclaimed_atlock that only advancesnext_run_atafter confirmed success/exhaustion.markSuccess/markFailednow returnlockLostso the worker can warn when it loses the lock before recording the outcome.0003_amused_marvex.sql(newclaimed_atcolumn) plus error-cause logging in the scheduled handler and a "no due schedules" tick log.How to test
pnpm typecheck
pnpm lint
pnpm test
pnpm format:check
Security review
No security-impacting changes.