fix(scheduler): cron parse guard + last-tick watermark [修复组②] - #21
Conversation
C2: an invalid cron_expression made croniter raise inside _is_due, which was swallowed by a bare `except Exception: return False` with zero logging — the schedule went permanently silent with no diagnostic trace. Now logs a WARNING (schedule id, name, offending expression) once per (schedule_id, cron_expression) pair per process; re-warns if the expression is later edited. Behavior stays "not due" (skip); the schedule is not disabled in DB. C4: the 61s due-check window was decoupled from actual tick cadence (sleep(60) + loop body time), so a fire could be dispatched by two consecutive ticks near the boundary, and a slow loop body (>1s) could silently miss a fire. Replaced with a process-local last_tick watermark: each tick evaluates the half-open interval (last_tick, now], which is disjoint and gapless across consecutive ticks, so double-dispatch is structurally impossible and a slow tick just widens its own window (catch-up) instead of losing a fire. Multiple fire times in one window coalesce into a single dispatch (logged at debug). First tick after process start only establishes the watermark and dispatches nothing, so a restart never replays everything missed while the process was down. Deviation from the original ask: a per-schedule dispatch is now wrapped in its own try/except. Without it, one schedule's dispatch raising would propagate out of the tick's for-loop and skip the `last_tick = now` update entirely, so the next tick's re-widened window could re-dispatch schedules that already fired successfully earlier in the same tick — reopening the exact double-dispatch bug C4 is meant to close. Verified empirically: test_scheduler_loop_dispatch_failure_does_not_stall_watermark fails (sched-ok dispatched twice) without the try/except and passes with it.
|
✅ Health: 8.6 📋 At a glance 🚨 Change risk: 9.2/10 (high)
🔎 More signals (1)🔥 Hotspot touched (1)
👀 Suggested reviewers @xujinghua 📊 Full report · ⭐ Star Repowise · 📥 Install bot · Last updated 2026-07-18 14:33 UTC |
|
Warning Review limit reached
Next review available in: 2 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the scheduler loop to use a gapless, window-based approach with a process-local watermark to prevent missed or double-dispatched schedules, and improves error handling for unparseable cron expressions. The review feedback is highly constructive, pointing out a potential scheduler stall if cron.get_next raises an exception, suggesting the addition of timezone support, and recommending initializing last_tick before the loop to simplify logic and eliminate an unnecessary 60-second startup delay.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| def _fires_in_window( | ||
| cron_expression: str, | ||
| schedule_id: str, | ||
| window_start: datetime, | ||
| window_end: datetime, | ||
| *, | ||
| name: str | None = None, | ||
| ) -> int: | ||
| """Count cron fire times in the half-open interval (window_start, window_end]. | ||
|
|
||
| AUDIT C2: a cron_expression croniter can't parse used to be swallowed by | ||
| a bare `except Exception: return False` — the schedule went permanently | ||
| silent with zero log trace. Now it warns once per (schedule_id, | ||
| cron_expression) pair and is treated as "not due" (0 fires) so one bad | ||
| schedule can't crash the loop or take down the others. | ||
| """ | ||
| try: | ||
| base = now - timedelta(seconds=61) | ||
| cron = croniter(cron_expression, base) | ||
| cron = croniter(cron_expression, window_start) | ||
| except Exception as exc: | ||
| warn_key = (schedule_id, cron_expression) | ||
| if warn_key not in _warned_bad_cron: | ||
| _warned_bad_cron.add(warn_key) | ||
| logger.warning( | ||
| "schedule %s (%s) has an unparseable cron_expression %r; " | ||
| "skipping until fixed: %s", | ||
| schedule_id, name or "?", cron_expression, exc, | ||
| ) | ||
| return 0 | ||
|
|
||
| count = 0 | ||
| while True: | ||
| next_fire = cron.get_next(datetime) | ||
| return next_fire <= now | ||
| except Exception: | ||
| return False | ||
| if next_fire > window_end: | ||
| break | ||
| count += 1 | ||
| return count |
There was a problem hiding this comment.
If croniter successfully parses the expression but cron.get_next(datetime) raises an exception (e.g., due to logically impossible date combinations like leap years, DST transitions, or bugs in croniter's calculation), the exception will propagate out of _fires_in_window.
Since the call to _fires_in_window in _scheduler_loop is not wrapped in a try-except block, this exception will propagate to the outer loop, preventing other schedules from being evaluated in the current tick, and preventing last_tick from being updated. In subsequent ticks, the window will widen and the same failing schedule will be evaluated again, leading to a permanent stall of the entire scheduler.
Wrapping the entire calculation (including the while loop) in the try-except block ensures complete robustness. Additionally, we can add support for the schedule's timezone by converting the UTC window boundaries to the target timezone using zoneinfo.ZoneInfo.
def _fires_in_window(
cron_expression: str,
schedule_id: str,
window_start: datetime,
window_end: datetime,
*,
name: str | None = None,
timezone_str: str = "UTC",
) -> int:
"""Count cron fire times in the half-open interval (window_start, window_end].
AUDIT C2: a cron_expression croniter can't parse used to be swallowed by
a bare `except Exception: return False` — the schedule went permanently
silent with zero log trace. Now it warns once per (schedule_id,
cron_expression) pair and is treated as "not due" (0 fires) so one bad
schedule can't crash the loop or take down the others.
"""
try:
from zoneinfo import ZoneInfo
tz = ZoneInfo(timezone_str)
except Exception:
tz = timezone.utc
try:
local_start = window_start.astimezone(tz)
local_end = window_end.astimezone(tz)
cron = croniter(cron_expression, local_start)
count = 0
while True:
next_fire = cron.get_next(datetime)
if next_fire > local_end:
break
count += 1
return count
except Exception as exc:
warn_key = (schedule_id, cron_expression)
if warn_key not in _warned_bad_cron:
_warned_bad_cron.add(warn_key)
logger.warning(
"schedule %s (%s) has an unparseable or failing cron_expression %r; "
"skipping until fixed: %s",
schedule_id, name or "?", cron_expression, exc,
)
return 0| "source_id": sched.source_id, | ||
| "cron_expression": sched.cron_expression, | ||
| "parameters": sched.parameters, | ||
| "name": sched.name, | ||
| } |
There was a problem hiding this comment.
The CronSchedule model has a timezone field (defaulting to "UTC"), but the scheduler currently ignores it and evaluates all cron expressions in UTC. This means a schedule configured for a specific local timezone (e.g., America/New_York) will fire at the wrong time (e.g., 9 AM UTC instead of 9 AM EST/EDT).
We can easily fix this by retrieving the timezone from the database and converting the UTC window boundaries (window_start and window_end) to the schedule's target timezone before passing them to croniter.
| "source_id": sched.source_id, | |
| "cron_expression": sched.cron_expression, | |
| "parameters": sched.parameters, | |
| "name": sched.name, | |
| } | |
| "source_id": sched.source_id, | |
| "cron_expression": sched.cron_expression, | |
| "parameters": sched.parameters, | |
| "name": sched.name, | |
| "timezone": sched.timezone, | |
| } |
| fire_count = _fires_in_window( | ||
| sched["cron_expression"], | ||
| sched["schedule_id"], | ||
| last_tick, | ||
| now, | ||
| name=sched["name"], | ||
| ) |
There was a problem hiding this comment.
Pass the retrieved timezone from the schedule dictionary to _fires_in_window so that cron expressions are evaluated in the correct local timezone.
fire_count = _fires_in_window(
sched["cron_expression"],
sched["schedule_id"],
last_tick,
now,
name=sched["name"],
timezone_str=sched.get("timezone", "UTC"),
)| last_tick: datetime | None = None | ||
| while True: | ||
| try: | ||
| await asyncio.sleep(60) | ||
| now = datetime.now(timezone.utc) | ||
| now = _now() | ||
|
|
||
| if last_tick is None: | ||
| # First tick after process start: establish the watermark | ||
| # without dispatching, so a restart never replays everything | ||
| # that fired while the process was down. | ||
| last_tick = now | ||
| continue |
There was a problem hiding this comment.
Currently, last_tick is initialized to None, and on the first tick after process start, it is set to now and the loop continues. This introduces an extra 60-second delay on startup before any schedules are evaluated (first evaluation happens 120 seconds after startup), and adds conditional logic inside the loop.
If we instead initialize last_tick = _now() immediately before entering the while True loop:
- The first evaluation will happen on the first tick (60 seconds after startup).
- The "no catch-up storm" guarantee is still perfectly preserved because the first window will be exactly
(startup_time, startup_time + 60s]. - The loop logic is simplified by removing the
if last_tick is None:check.
Note: If you apply this, remember to update or remove the test test_scheduler_loop_first_tick_establishes_watermark_no_dispatch since the first tick will now evaluate the first 60-second window.
| last_tick: datetime | None = None | |
| while True: | |
| try: | |
| await asyncio.sleep(60) | |
| now = datetime.now(timezone.utc) | |
| now = _now() | |
| if last_tick is None: | |
| # First tick after process start: establish the watermark | |
| # without dispatching, so a restart never replays everything | |
| # that fired while the process was down. | |
| last_tick = now | |
| continue | |
| last_tick = _now() | |
| while True: | |
| try: | |
| await asyncio.sleep(60) | |
| now = _now() |
1 participant
修复组② — 调度器 (账本 C2/C4)
Sonnet 实施, Fable 审计通过 (窗口数学 + 故障语义 + deviation 反证测试逐项核过)。
except: return False换成 warning (schedule id + name + 表达式 + 异常), 每(schedule_id, cron_expression)进程内只警一次, 表达式被编辑后再坏会重新警。行为保持"跳过", 不动 DB(last_tick, now]— 双派发在数学上不可能, 慢 tick 自动加宽窗口补漏而非丢失。窗口内多个 fire 合并单次派发 (debug 记数)。进程启动首 tick 只立水位不派发 (重启无补发风暴)test_scheduler_loop_dispatch_failure_does_not_stall_watermark)Test
tests/unit/test_scheduler.py21/21 (冻结时钟 +_now()seam, 零真实 sleep): 坏 cron 警一次/改后重警、窗口内恰好一发、相邻窗口零双发、3 个漏发合并一发、首 tick 不发、派发失败不卡水位