fix(cron): don't silently skip a due run after a timezone-offset migration
Upgrading from a UTC-scheduling build to one that honours the profile timezone (Europe/Brussels) left daily cron jobs sitting in jobs.json with pre-migration instants — e.g. next_run_at "2026-09-02T04:00:00+00:00" for expr "0 4 * * *". _ensure_aware normalizes that to 06:00+02, which the expression excludes, so the stale-expression guard (#93049) read it as a direct jobs.json edit, logged exactly that, and re-anchored to tomorrow without firing. The due occurrence disappeared with no error anywhere. The guard only asked "is the stored instant an occurrence of the current expr?", never "why not?" — and the two possible answers demand opposite actions. Add _classify_stale_cron_next_run, which distinguishes them by whether normalization itself moved the wall clock: * expr_edit — wall clock unchanged (or the stored wall clock is not an occurrence either): the instant is genuinely excluded by the current expression. Re-anchor without firing, exactly as before. * timezone_migration — the stored value's own wall clock IS a legal occurrence and it only left the lattice because _ensure_aware converted it to a different offset. Fall through and fire the overdue run once. Because every value written by this build carries the configured offset, a real expr edit leaves the wall clock untouched and can never be reclassified as a migration, so the #93049 protection is intact. At-most-once is unchanged: the fire flows through the normal due path and the usual advance_next_run / mark_job_run re-anchor rewrites next_run_at in the current offset, so the legacy instant is never read again. Future local wall-clock occurrences are untouched — not-yet-due rows never reach the guard, and the #28934 offset-repair branch still runs first for a still-future stored wall clock. The migration case is classified explicitly rather than retried broadly: it logs cron.timezone_migration.catch_up with the stored and normalized instants plus both offsets, and increments a probe-visible counter (get_timezone_migration_catchup_stats, timezone_migration_catchups.jsonl) kept separate from catch_up_occurrences so an operator can tell "the upgrade backlog is draining" from "runs are missing their grace window". Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
+128
-3
@@ -1410,6 +1410,101 @@ def _cron_next_run_matches_expr(
|
||||
return True
|
||||
|
||||
|
||||
# Classification results for a due cron instant that is NOT an occurrence of
|
||||
# the job's current expression (see _classify_stale_cron_next_run).
|
||||
STALE_CRON_MATCH = "match"
|
||||
STALE_CRON_TIMEZONE_MIGRATION = "timezone_migration"
|
||||
STALE_CRON_EXPR_EDIT = "expr_edit"
|
||||
|
||||
|
||||
def _classify_stale_cron_next_run(
|
||||
schedule: Dict[str, Any],
|
||||
raw_next_run_dt: datetime,
|
||||
next_run_dt: datetime,
|
||||
) -> str:
|
||||
"""Explain WHY a stored ``next_run_at`` misses the current cron lattice.
|
||||
|
||||
``_cron_next_run_matches_expr`` answers "does the stored instant occur in
|
||||
the current expression?" but not "why not?", and the two answers call for
|
||||
opposite actions:
|
||||
|
||||
* ``expr_edit`` — a direct ``jobs.json`` edit changed ``schedule.expr``
|
||||
while leaving ``next_run_at`` computed under the old one (#93049). The
|
||||
stored instant is a time the current expression *excludes*, so it must
|
||||
be re-anchored WITHOUT firing.
|
||||
* ``timezone_migration`` — the expression never changed; only the stored
|
||||
value's *offset representation* did. Upgrading from a UTC-scheduling
|
||||
build to one that honours the profile timezone leaves legacy rows like
|
||||
``2026-09-02T04:00:00+00:00`` for ``0 4 * * *``; normalizing to
|
||||
Europe/Brussels turns that into ``06:00+02``, which the expression
|
||||
excludes. Treating it as a stale edit re-anchored to tomorrow and
|
||||
silently skipped a due occurrence that had never fired.
|
||||
|
||||
The discriminator is whether *normalization itself* moved the wall clock.
|
||||
Cron expressions describe local wall-clock intent, so a stored instant
|
||||
whose OWN wall clock is a legal occurrence, and which only left the
|
||||
lattice because ``_ensure_aware`` converted it into a different offset, is
|
||||
a representation migration — not a schedule edit. When the offsets agree
|
||||
(the common case, including every value this build wrote) the wall clock
|
||||
is unchanged, so a genuine ``expr`` edit can never be misread as a
|
||||
migration.
|
||||
"""
|
||||
if _cron_next_run_matches_expr(schedule, next_run_dt):
|
||||
return STALE_CRON_MATCH
|
||||
wall_clock_shifted = (
|
||||
raw_next_run_dt.replace(tzinfo=None) != next_run_dt.replace(tzinfo=None)
|
||||
)
|
||||
if wall_clock_shifted and _cron_next_run_matches_expr(schedule, raw_next_run_dt):
|
||||
return STALE_CRON_TIMEZONE_MIGRATION
|
||||
return STALE_CRON_EXPR_EDIT
|
||||
|
||||
|
||||
# Durable, probe-visible counter for offset-representation migrations caught
|
||||
# on the fire path. Distinct from `catch_up_occurrences` (which counts runs
|
||||
# skipped past their grace window) because this one means "an upgrade rewrote
|
||||
# how next_run_at is represented" — an operator seeing it climb after a deploy
|
||||
# is seeing the migration drain, and seeing it climb steadily afterwards is
|
||||
# seeing a timezone that keeps changing under the store.
|
||||
_timezone_migration_catchups: int = 0
|
||||
_TIMEZONE_MIGRATION_CATCHUP_HISTORY = 20
|
||||
_timezone_migration_catchups_recent: list = []
|
||||
|
||||
|
||||
def _record_timezone_migration_catchup(
|
||||
job: Dict[str, Any],
|
||||
raw_next_run_dt: datetime,
|
||||
next_run_dt: datetime,
|
||||
) -> None:
|
||||
"""Persist a countable signal for one offset-migration catch-up fire."""
|
||||
global _timezone_migration_catchups
|
||||
entry = {
|
||||
"job_id": job.get("id"),
|
||||
"name": job.get("name") or job.get("id"),
|
||||
"expr": (job.get("schedule") or {}).get("expr"),
|
||||
"stored_next_run_at": raw_next_run_dt.isoformat(),
|
||||
"normalized_next_run_at": next_run_dt.isoformat(),
|
||||
"fired_at": _hermes_now().isoformat(),
|
||||
}
|
||||
_timezone_migration_catchups += 1
|
||||
_timezone_migration_catchups_recent.append(entry)
|
||||
del _timezone_migration_catchups_recent[:-_TIMEZONE_MIGRATION_CATCHUP_HISTORY]
|
||||
try:
|
||||
path = _current_cron_store().cron_dir / "timezone_migration_catchups.jsonl"
|
||||
_ensure_cron_dir(path.parent)
|
||||
with open(path, "a", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(entry) + "\n")
|
||||
except Exception as exc: # never let telemetry break a tick
|
||||
logger.debug("Could not append timezone-migration-catchup record: %s", exc)
|
||||
|
||||
|
||||
def get_timezone_migration_catchup_stats() -> Dict[str, Any]:
|
||||
"""Probe-visible snapshot of offset-migration catch-up fires."""
|
||||
return {
|
||||
"timezone_migration_catchups": _timezone_migration_catchups,
|
||||
"recent": list(_timezone_migration_catchups_recent),
|
||||
}
|
||||
|
||||
|
||||
def compute_next_run(schedule: Dict[str, Any], last_run_at: Optional[str] = None) -> Optional[str]:
|
||||
"""
|
||||
Compute the next run time for a schedule.
|
||||
@@ -4046,9 +4141,18 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
|
||||
# so re-anchor before either can fire. Recomputation uses the
|
||||
# current expression, so this converges — it cannot defer
|
||||
# forever.
|
||||
if not manual_run and kind == "cron" and not _cron_next_run_matches_expr(
|
||||
schedule, next_run_dt
|
||||
):
|
||||
#
|
||||
# Not every mismatch is an edit, though: an offset-representation
|
||||
# migration (UTC-scheduling build -> profile-timezone build)
|
||||
# moves a legacy instant off the lattice without the expression
|
||||
# ever changing, and re-anchoring THAT silently swallowed a due
|
||||
# occurrence. Classify first, and only the edit case skips.
|
||||
stale_class = (
|
||||
_classify_stale_cron_next_run(schedule, raw_next_run_dt, next_run_dt)
|
||||
if not manual_run and kind == "cron"
|
||||
else STALE_CRON_MATCH
|
||||
)
|
||||
if stale_class == STALE_CRON_EXPR_EDIT:
|
||||
new_next = compute_next_run(schedule, now.isoformat())
|
||||
logger.info(
|
||||
"Job '%s' next_run_at %s does not match its current "
|
||||
@@ -4066,6 +4170,27 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
|
||||
needs_save = True
|
||||
break
|
||||
continue
|
||||
if stale_class == STALE_CRON_TIMEZONE_MIGRATION:
|
||||
# Fall through to the normal due path: the occurrence is
|
||||
# real and overdue, so it fires ONCE here and the usual
|
||||
# advance/mark_job_run re-anchor writes the value back in
|
||||
# the current offset. At-most-once is preserved because
|
||||
# nothing re-reads the legacy instant after that.
|
||||
logger.warning(
|
||||
"cron.timezone_migration.catch_up job='%s' id=%s expr=%r "
|
||||
"stored=%s normalized=%s — stored next_run_at carries a "
|
||||
"pre-migration UTC offset (%s, now %s) and is a legal "
|
||||
"occurrence at its own wall clock; firing the due run "
|
||||
"instead of re-anchoring past it.",
|
||||
job.get("name", job.get("id", "?")),
|
||||
job.get("id"),
|
||||
schedule.get("expr"),
|
||||
next_run,
|
||||
next_run_dt.isoformat(),
|
||||
raw_next_run_dt.utcoffset(),
|
||||
now.utcoffset(),
|
||||
)
|
||||
_record_timezone_migration_catchup(job, raw_next_run_dt, next_run_dt)
|
||||
|
||||
# For recurring jobs, check if the scheduled time is stale
|
||||
# (gateway was down and missed the window). Fast-forward to
|
||||
|
||||
@@ -0,0 +1,213 @@
|
||||
"""Timezone-migration silent misfire on the cron fire path.
|
||||
|
||||
Production incident: after upgrading from a build that scheduled in UTC to
|
||||
one that honours the profile timezone (Europe/Brussels), daily cron jobs
|
||||
stopped running. Their ``jobs.json`` rows still held pre-migration instants
|
||||
like ``2026-09-02T04:00:00+00:00`` for expr ``0 4 * * *``. ``_ensure_aware``
|
||||
normalizes that to ``06:00+02``, which ``0 4 * * *`` excludes, so the
|
||||
stale-expression guard (#93049) classified it as a direct ``jobs.json`` edit,
|
||||
logged exactly that, and re-anchored to tomorrow WITHOUT firing — the due
|
||||
occurrence vanished with no failure anywhere.
|
||||
|
||||
The fix classifies the mismatch instead of assuming an edit: an instant whose
|
||||
own wall clock is a legal occurrence, and which only left the lattice because
|
||||
normalization changed its offset, is a representation migration and fires.
|
||||
|
||||
These exercise the real store against a temp ``HERMES_HOME`` (no mocks) per
|
||||
the E2E-over-mocks discipline for file-touching code.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_home(tmp_path, monkeypatch):
|
||||
"""Isolated HERMES_HOME so jobs.json doesn't touch the real store."""
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
yield tmp_path
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_migration_counters(monkeypatch):
|
||||
"""Module-level telemetry counters must not leak between tests."""
|
||||
from cron import jobs as J
|
||||
|
||||
monkeypatch.setattr(J, "_timezone_migration_catchups", 0)
|
||||
monkeypatch.setattr(J, "_timezone_migration_catchups_recent", [])
|
||||
yield
|
||||
|
||||
|
||||
# Europe/Brussels is +02:00 on this date; the legacy row was written by a
|
||||
# build that scheduled everything at the UTC offset.
|
||||
_BRUSSELS_NOW = datetime.fromisoformat("2026-09-02T06:05:00+02:00")
|
||||
_LEGACY_UTC_NEXT_RUN = "2026-09-02T04:00:00+00:00"
|
||||
_DAILY_0400 = "0 4 * * *"
|
||||
|
||||
|
||||
def _write_cron_job(expr: str, next_run_at: str, name: str = "t") -> str:
|
||||
"""Persist a cron job with a pinned next_run_at (the legacy-row shape)."""
|
||||
from cron.jobs import create_job, load_jobs, save_jobs
|
||||
|
||||
job = create_job(prompt="x", schedule="every 5m", name=name)
|
||||
jobs = load_jobs()
|
||||
for j in jobs:
|
||||
if j["id"] == job["id"]:
|
||||
j["schedule"] = {"kind": "cron", "expr": expr}
|
||||
j["next_run_at"] = next_run_at
|
||||
save_jobs(jobs)
|
||||
return job["id"]
|
||||
|
||||
|
||||
def test_legacy_utc_offset_next_run_still_fires(temp_home, monkeypatch):
|
||||
"""The incident case: a pre-migration +00:00 instant for a Brussels
|
||||
``0 4 * * *`` job must fire its due occurrence, not be re-anchored away."""
|
||||
from cron.jobs import get_due_jobs, get_timezone_migration_catchup_stats
|
||||
|
||||
monkeypatch.setattr("cron.jobs._hermes_now", lambda: _BRUSSELS_NOW)
|
||||
jid = _write_cron_job(_DAILY_0400, _LEGACY_UTC_NEXT_RUN)
|
||||
|
||||
due = get_due_jobs()
|
||||
|
||||
assert jid in [j["id"] for j in due]
|
||||
stats = get_timezone_migration_catchup_stats()
|
||||
assert stats["timezone_migration_catchups"] == 1
|
||||
record = stats["recent"][0]
|
||||
assert record["job_id"] == jid
|
||||
assert record["expr"] == _DAILY_0400
|
||||
assert record["stored_next_run_at"] == _LEGACY_UTC_NEXT_RUN
|
||||
assert record["normalized_next_run_at"] == "2026-09-02T06:00:00+02:00"
|
||||
|
||||
|
||||
def test_legacy_offset_catchup_fires_at_most_once(temp_home, monkeypatch):
|
||||
"""The catch-up run is a single fire: once the scheduler advances the
|
||||
job, the legacy instant is gone and a second scan finds nothing due."""
|
||||
from cron.jobs import advance_next_run, get_due_jobs, get_job
|
||||
|
||||
monkeypatch.setattr("cron.jobs._hermes_now", lambda: _BRUSSELS_NOW)
|
||||
jid = _write_cron_job(_DAILY_0400, _LEGACY_UTC_NEXT_RUN)
|
||||
|
||||
assert jid in [j["id"] for j in get_due_jobs()]
|
||||
assert advance_next_run(jid) is True
|
||||
|
||||
# Re-anchored to tomorrow's occurrence, expressed in the configured zone.
|
||||
assert get_job(jid)["next_run_at"] == "2026-09-03T04:00:00+02:00"
|
||||
assert [j["id"] for j in get_due_jobs() if j["id"] == jid] == []
|
||||
|
||||
|
||||
def test_genuine_expr_edit_still_reanchors_without_firing(temp_home, monkeypatch):
|
||||
"""#93049 protection intact: a stale instant in the CURRENT offset (no
|
||||
representation change) is still treated as an edit and does not fire."""
|
||||
from cron.jobs import get_due_jobs, get_job, get_timezone_migration_catchup_stats
|
||||
|
||||
monkeypatch.setattr("cron.jobs._hermes_now", lambda: _BRUSSELS_NOW)
|
||||
# Stored at the configured offset, but the expr was edited to 09:00.
|
||||
jid = _write_cron_job("0 9 * * *", "2026-09-02T04:00:00+02:00")
|
||||
|
||||
due = get_due_jobs()
|
||||
|
||||
assert [j["id"] for j in due if j["id"] == jid] == []
|
||||
assert get_job(jid)["next_run_at"] == "2026-09-02T09:00:00+02:00"
|
||||
assert (
|
||||
get_timezone_migration_catchup_stats()["timezone_migration_catchups"] == 0
|
||||
)
|
||||
|
||||
|
||||
def test_expr_edit_on_a_legacy_offset_row_still_does_not_fire(temp_home, monkeypatch):
|
||||
"""A legacy +00:00 row whose expr was ALSO edited must not fire: the
|
||||
stored wall clock is not an occurrence of the new expression either, so
|
||||
the migration escape hatch does not open."""
|
||||
from cron.jobs import get_due_jobs, get_job, get_timezone_migration_catchup_stats
|
||||
|
||||
monkeypatch.setattr("cron.jobs._hermes_now", lambda: _BRUSSELS_NOW)
|
||||
jid = _write_cron_job("0 9 * * *", _LEGACY_UTC_NEXT_RUN)
|
||||
|
||||
due = get_due_jobs()
|
||||
|
||||
assert [j["id"] for j in due if j["id"] == jid] == []
|
||||
assert get_job(jid)["next_run_at"] == "2026-09-02T09:00:00+02:00"
|
||||
assert (
|
||||
get_timezone_migration_catchup_stats()["timezone_migration_catchups"] == 0
|
||||
)
|
||||
|
||||
|
||||
def test_future_local_wall_clock_is_left_scheduled(temp_home, monkeypatch):
|
||||
"""A legacy row whose normalized instant has not arrived yet is simply
|
||||
not due — no catch-up, no re-anchor, no telemetry."""
|
||||
from cron.jobs import get_due_jobs, get_job, get_timezone_migration_catchup_stats
|
||||
|
||||
before_due = datetime.fromisoformat("2026-09-02T05:00:00+02:00")
|
||||
monkeypatch.setattr("cron.jobs._hermes_now", lambda: before_due)
|
||||
jid = _write_cron_job(_DAILY_0400, _LEGACY_UTC_NEXT_RUN)
|
||||
|
||||
due = get_due_jobs()
|
||||
|
||||
assert [j["id"] for j in due if j["id"] == jid] == []
|
||||
assert get_job(jid)["next_run_at"] == _LEGACY_UTC_NEXT_RUN
|
||||
assert (
|
||||
get_timezone_migration_catchup_stats()["timezone_migration_catchups"] == 0
|
||||
)
|
||||
|
||||
|
||||
def test_future_stored_wall_clock_still_takes_the_offset_repair_path(
|
||||
temp_home, monkeypatch
|
||||
):
|
||||
"""#28934 regression: a westward TZ move (+10 -> +02) that makes a still-
|
||||
future wall clock look due recomputes rather than firing early, and is
|
||||
NOT reclassified as a migration catch-up."""
|
||||
from cron.jobs import get_due_jobs, get_job, get_timezone_migration_catchup_stats
|
||||
|
||||
scan_time = datetime.fromisoformat("2026-09-02T14:00:00+02:00")
|
||||
monkeypatch.setattr("cron.jobs._hermes_now", lambda: scan_time)
|
||||
jid = _write_cron_job("0 21 * * *", "2026-09-02T21:00:00+10:00")
|
||||
|
||||
due = get_due_jobs()
|
||||
|
||||
assert [j["id"] for j in due if j["id"] == jid] == []
|
||||
assert get_job(jid)["next_run_at"] == "2026-09-02T21:00:00+02:00"
|
||||
assert (
|
||||
get_timezone_migration_catchup_stats()["timezone_migration_catchups"] == 0
|
||||
)
|
||||
|
||||
|
||||
def test_classifier_separates_migration_from_edit(temp_home):
|
||||
"""Unit-level: the three classifications the fire path branches on."""
|
||||
from cron.jobs import (
|
||||
STALE_CRON_EXPR_EDIT,
|
||||
STALE_CRON_MATCH,
|
||||
STALE_CRON_TIMEZONE_MIGRATION,
|
||||
_classify_stale_cron_next_run,
|
||||
)
|
||||
|
||||
daily = {"kind": "cron", "expr": _DAILY_0400}
|
||||
raw_legacy = datetime.fromisoformat(_LEGACY_UTC_NEXT_RUN)
|
||||
normalized = datetime.fromisoformat("2026-09-02T06:00:00+02:00")
|
||||
on_lattice = datetime.fromisoformat("2026-09-02T04:00:00+02:00")
|
||||
|
||||
# Stored instant already occurs under the current expression.
|
||||
assert (
|
||||
_classify_stale_cron_next_run(daily, on_lattice, on_lattice)
|
||||
== STALE_CRON_MATCH
|
||||
)
|
||||
# Only the offset representation changed.
|
||||
assert (
|
||||
_classify_stale_cron_next_run(daily, raw_legacy, normalized)
|
||||
== STALE_CRON_TIMEZONE_MIGRATION
|
||||
)
|
||||
# Wall clock never moved, so a mismatch can only be a schedule edit.
|
||||
assert (
|
||||
_classify_stale_cron_next_run(
|
||||
{"kind": "cron", "expr": "0 9 * * *"}, on_lattice, on_lattice
|
||||
)
|
||||
== STALE_CRON_EXPR_EDIT
|
||||
)
|
||||
# Wall clock moved, but the stored wall clock is not an occurrence either.
|
||||
assert (
|
||||
_classify_stale_cron_next_run(
|
||||
{"kind": "cron", "expr": "0 9 * * *"}, raw_legacy, normalized
|
||||
)
|
||||
== STALE_CRON_EXPR_EDIT
|
||||
)
|
||||
Reference in New Issue
Block a user