diff --git a/cron/jobs.py b/cron/jobs.py index bd9be129cd..84bb8fefbf 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -855,6 +855,26 @@ def _ensure_aware(dt: datetime) -> datetime: return dt.astimezone(target_tz) +def _elapsed_seconds(later: datetime, earlier: datetime) -> float: + """Return elapsed seconds between aware instants, independent of wall time.""" + return (later.astimezone(timezone.utc) - earlier.astimezone(timezone.utc)).total_seconds() + + +def _instant_after(left: datetime, right: datetime) -> bool: + """Whether *left* is a later absolute instant than *right*.""" + return left.astimezone(timezone.utc) > right.astimezone(timezone.utc) + + +def _instant_at_or_before(left: datetime, right: datetime) -> bool: + """Whether *left* is at or before *right* as an absolute instant.""" + return left.astimezone(timezone.utc) <= right.astimezone(timezone.utc) + + +def _instant_before(left: datetime, right: datetime) -> bool: + """Whether *left* is an earlier absolute instant than *right*.""" + return left.astimezone(timezone.utc) < right.astimezone(timezone.utc) + + def _parse_aware(value: Any) -> Optional[datetime]: """``_ensure_aware(datetime.fromisoformat(value))``, or None when *value* is not a parseable ISO string.""" @@ -888,7 +908,7 @@ def _recoverable_oneshot_run_at( return None run_at = schedule.get("run_at") run_at_dt = _parse_aware(run_at) if run_at else None - if run_at_dt is not None and run_at_dt >= now - timedelta(seconds=ONESHOT_GRACE_SECONDS): + if run_at_dt is not None and _elapsed_seconds(now, run_at_dt) <= ONESHOT_GRACE_SECONDS: return run_at return None @@ -958,7 +978,7 @@ def _job_is_stale_error_recurring( last_run_dt = _parse_aware(last_run) if last_run else None if last_run_dt is None: return False - age_seconds = (now - last_run_dt).total_seconds() + age_seconds = _elapsed_seconds(now, last_run_dt) if age_seconds < 0: return False grace = _compute_grace_seconds(schedule) @@ -2142,7 +2162,7 @@ def _claim_is_live(claim: Any, now: datetime, ttl_seconds: float) -> bool: if not isinstance(claim, dict) or not claim.get("at"): return False claimed_at = _parse_aware(claim["at"]) - if claimed_at is None or not (0 <= (now - claimed_at).total_seconds() < ttl_seconds): + if claimed_at is None or not (0 <= _elapsed_seconds(now, claimed_at) < ttl_seconds): return False return not _claim_owner_is_dead(claim) @@ -2900,7 +2920,7 @@ def _repair_timezone_shifted_cron(d: _DueJob) -> bool: offset change meeting the same conditions SKIPS the pending occurrence; accepted as rare.""" now = d.scan.now if not ( - d.next_run_dt <= now + _instant_at_or_before(d.next_run_dt, now) and _timezone_offset_mismatch(d.raw_next_run_dt, now) and _stored_wall_clock_is_future(d.raw_next_run_dt, now) ): @@ -2928,7 +2948,7 @@ def _rearm_stale_error_recurring(d: _DueJob) -> datetime: now = d.scan.now if not ( d.kind in ("cron", "interval") - and d.next_run_dt > now + and _instant_after(d.next_run_dt, now) and _job_is_stale_error_recurring(d.job, d.schedule, now) ): return d.next_run_dt @@ -2938,7 +2958,10 @@ def _rearm_stale_error_recurring(d: _DueJob) -> datetime: else: recovered_next = d.recompute_next() recovered_next_dt = _parse_aware(recovered_next) if recovered_next else None - if not (recovered_next and recovered_next_dt is not None and recovered_next_dt < d.next_run_dt): + if not ( + recovered_next and recovered_next_dt is not None + and _instant_before(recovered_next_dt, d.next_run_dt) + ): return d.next_run_dt jid = d.job.get("id") logger.warning( @@ -2991,13 +3014,13 @@ def _fast_forward_missed_recurring(d: _DueJob, grace: int) -> bool: protects the crash window before mark_job_run and covers the external fire_due path, which never calls advance_next_run. mark_job_run re-anchors on completion, so the value is provisional. """ - if (d.scan.now - d.next_run_dt).total_seconds() <= grace: + if _elapsed_seconds(d.scan.now, d.next_run_dt) <= grace: return False new_next = d.recompute_next() if not new_next: return False d.scan.persist(d.job["id"], next_run_at=new_next) - if (_ensure_aware(datetime.fromisoformat(new_next)) > d.scan.now + if (_instant_after(_ensure_aware(datetime.fromisoformat(new_next)), d.scan.now) and not _cron_config_number("catch_up_missed", True, lambda value: value is not False)): logger.info( "Job '%s' missed its scheduled time (%s, grace=%ds). " @@ -3019,7 +3042,7 @@ def _retire_expired_oneshot(d: _DueJob) -> bool: and recovery never revives them; only the due scan used to dispatch them hours late). With no claim stamped, retire it with a diagnostic (never silently delete). A claim may mean a run is still in flight elsewhere — skip but keep the record so its mark_job_run can land.""" - if (d.scan.now - d.next_run_dt).total_seconds() <= ONESHOT_GRACE_SECONDS: + if _elapsed_seconds(d.scan.now, d.next_run_dt) <= ONESHOT_GRACE_SECONDS: return False if not (d.job.get("run_claim") or d.job.get("fire_claim")): _write_missed_oneshot_diagnostic(d.job, d.next_run) @@ -3128,7 +3151,7 @@ def _evaluate_due_job(job: Dict[str, Any], scan: _DueScan, run_claim_ttl: float) if kind == "cron" and not manual_run and _repair_timezone_shifted_cron(d): return False d.next_run_dt = _rearm_stale_error_recurring(d) - if d.next_run_dt > now: + if _instant_after(d.next_run_dt, now): return False # Only the dispatch snapshot carries this field; never infer it from a later stamp. @@ -3154,7 +3177,7 @@ def _evaluate_due_job(job: Dict[str, Any], scan: _DueScan, run_claim_ttl: float) # late catch-up. Recurring only — expired one-shots were retired above; manual triggers aren't # late. if not manual_run and recurring: - lateness = max(0.0, (now - d.next_run_dt).total_seconds()) + lateness = max(0.0, _elapsed_seconds(now, d.next_run_dt)) # See #99879. dispatch_stamp = { "scheduled_at": next_run, diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index 1431191cd0..648252366f 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -295,7 +295,8 @@ def fire_overdue_jobs( return 0 from cron.jobs import ( - ONESHOT_GRACE_SECONDS, _ensure_aware, _hermes_now, is_job_runnable, load_jobs, + ONESHOT_GRACE_SECONDS, _elapsed_seconds, _ensure_aware, _hermes_now, + is_job_runnable, load_jobs, ) if now is None: @@ -312,7 +313,7 @@ def fire_overdue_jobs( due_dt = _ensure_aware(datetime.fromisoformat(next_run_at)) except (ValueError, TypeError): continue - overdue_seconds = (now - due_dt).total_seconds() + overdue_seconds = _elapsed_seconds(now, due_dt) if overdue_seconds < grace_minutes * 60: continue job_id = str(job.get("id") or "") diff --git a/tests/cron/test_due_scan_dst_fold.py b/tests/cron/test_due_scan_dst_fold.py new file mode 100644 index 0000000000..5fbe2f2d5f --- /dev/null +++ b/tests/cron/test_due_scan_dst_fold.py @@ -0,0 +1,110 @@ +"""Due-scan comparisons must use absolute instants across a repeated DST hour.""" + +from datetime import datetime, timezone +from zoneinfo import ZoneInfo + +import hermes_time +import pytest + +from cron import jobs + + +NEW_YORK = ZoneInfo("America/New_York") + + +@pytest.fixture +def dst_cron_store(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("HERMES_TIMEZONE", "America/New_York") + hermes_time.reset_cache() + monkeypatch.setattr(jobs, "CRON_DIR", tmp_path / "cron") + monkeypatch.setattr(jobs, "JOBS_FILE", tmp_path / "cron" / "jobs.json") + monkeypatch.setattr(jobs, "OUTPUT_DIR", tmp_path / "cron" / "output") + yield tmp_path + hermes_time.reset_cache() + + +def _instant(hour, minute): + return datetime(2026, 11, 1, hour, minute, tzinfo=timezone.utc).astimezone(NEW_YORK) + + +def _job(next_run_at, *, kind="interval", expr=None): + schedule = {"kind": kind, "minutes": 60} + if expr is not None: + schedule["expr"] = expr + return { + "id": "dst-job", + "name": "dst-job", + "prompt": "fixture", + "schedule": schedule, + "next_run_at": next_run_at.isoformat(), + "last_run_at": None, + "enabled": True, + "state": "scheduled", + "repeat": {"times": None, "completed": 0}, + "deliver": "local", + } + + +def _due_at(monkeypatch, now, scheduled): + monkeypatch.setattr(jobs, "_hermes_now", lambda: now) + jobs.save_jobs([_job(scheduled)]) + return [row["id"] for row in jobs.get_due_jobs()] + + +def test_fold_one_interval_is_not_due_during_fold_zero(dst_cron_store, monkeypatch): + now = _instant(5, 1) # 01:01 EDT, fold=0 + scheduled = _instant(6, 0) # 01:00 EST, fold=1 + assert now.fold == 0 + assert scheduled.fold == 1 + assert scheduled.timestamp() - now.timestamp() == 59 * 60 + + assert _due_at(monkeypatch, now, scheduled) == [] + + +def test_interval_runs_when_due_in_either_fold(dst_cron_store, monkeypatch): + first_fold_now = _instant(5, 1) + first_fold_slot = _instant(5, 0) + second_fold_now = _instant(6, 1) + second_fold_slot = _instant(6, 0) + + assert first_fold_now.fold == first_fold_slot.fold == 0 + assert second_fold_now.fold == second_fold_slot.fold == 1 + assert _due_at(monkeypatch, first_fold_now, first_fold_slot) == ["dst-job"] + assert _due_at(monkeypatch, second_fold_now, second_fold_slot) == ["dst-job"] + + +def test_cron_future_occurrence_is_not_due_in_prior_fold(dst_cron_store, monkeypatch): + now = _instant(5, 31) # 01:31 EDT, fold=0 + scheduled = _instant(6, 30) # 01:30 EST, fold=1 + assert scheduled.timestamp() > now.timestamp() + + monkeypatch.setattr(jobs, "_hermes_now", lambda: now) + jobs.save_jobs([_job(scheduled, kind="cron", expr="30 1 * * *")]) + assert jobs.get_due_jobs() == [] + + +def test_hourly_interval_does_not_recur_each_minute_during_fallback(dst_cron_store, monkeypatch): + """A one-hour cadence remains hourly across both occurrences of local 01:00.""" + from datetime import timedelta + + start = datetime(2026, 11, 1, 4, 0, tzinfo=timezone.utc) # 00:00 EDT + initial_now = (start - timedelta(hours=1)).astimezone(NEW_YORK) + monkeypatch.setattr(jobs, "_hermes_now", lambda: initial_now) + job = jobs.create_job( + prompt="fixture", schedule="every 60m", model="fixture", deliver="local") + fired_at = [] + + for elapsed_minutes in range(181): + now = (start + timedelta(minutes=elapsed_minutes)).astimezone(NEW_YORK) + monkeypatch.setattr(jobs, "_hermes_now", lambda now=now: now) + due = jobs.get_due_jobs() + if due: + fired_at.append(now.timestamp()) + assert [row["id"] for row in due] == [job["id"]] + assert jobs.claim_job_for_fire(job["id"]) is True + assert jobs.mark_job_run(job["id"], success=True) is True + + assert fired_at == [ + (start + timedelta(hours=offset)).timestamp() for offset in range(4) + ] diff --git a/tests/cron/test_misfire_catchup.py b/tests/cron/test_misfire_catchup.py index a786b6075c..dd2659bc0e 100644 --- a/tests/cron/test_misfire_catchup.py +++ b/tests/cron/test_misfire_catchup.py @@ -8,7 +8,8 @@ gateway housekeeping, claims and fires those jobs after a grace window. """ import threading -from datetime import timedelta +from datetime import datetime, timedelta, timezone +from zoneinfo import ZoneInfo import pytest @@ -66,6 +67,27 @@ def _park_in_past(job_id, minutes): class TestFireOverdueJobs: + def test_future_fold_one_slot_is_not_overdue_in_fold_zero(self, tmp_cron_dir, monkeypatch): + """A repeated-hour slot stays future in the external-provider backstop too.""" + import hermes_time + + monkeypatch.setenv("HERMES_TIMEZONE", "America/New_York") + hermes_time.reset_cache() + try: + zone = ZoneInfo("America/New_York") + now = datetime(2026, 11, 1, 5, 40, tzinfo=timezone.utc).astimezone(zone) + next_run = datetime(2026, 11, 1, 6, 0, tzinfo=timezone.utc).astimezone(zone) + monkeypatch.setattr("cron.jobs._hermes_now", lambda: now) + create_job(prompt="p", schedule="every 1h") + rows = load_jobs() + rows[0]["next_run_at"] = next_run.isoformat() + save_jobs(rows) + + assert now.fold == 0 and next_run.fold == 1 + assert fire_overdue_jobs(RecordingProvider(), now=now) == 0 + finally: + hermes_time.reset_cache() + def test_noop_for_builtin_provider(self, tmp_cron_dir): """The in-process ticker self-heals past-due jobs — the sweep must never double-dispatch under it."""