diff --git a/cron/jobs.py b/cron/jobs.py index 3865e02b3ef..cc9d1351d75 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -568,7 +568,10 @@ def _recoverable_oneshot_run_at( if not run_at: return None - run_at_dt = _ensure_aware(datetime.fromisoformat(run_at)) + try: + run_at_dt = _ensure_aware(datetime.fromisoformat(run_at)) + except Exception: + return None if run_at_dt >= now - timedelta(seconds=ONESHOT_GRACE_SECONDS): return run_at return None @@ -630,9 +633,11 @@ def compute_next_run(schedule: Dict[str, Any], last_run_at: Optional[str] = None if minutes is None: return None if last_run_at: - # Next run is last_run + interval - last = _ensure_aware(datetime.fromisoformat(last_run_at)) - next_run = last + timedelta(minutes=minutes) + try: + last = _ensure_aware(datetime.fromisoformat(last_run_at)) + next_run = last + timedelta(minutes=minutes) + except Exception: + next_run = now + timedelta(minutes=minutes) else: # First run is now + interval next_run = now + timedelta(minutes=minutes) @@ -657,7 +662,10 @@ def compute_next_run(schedule: Dict[str, Any], last_run_at: Optional[str] = None # rather than to an arbitrary restart time. base_time = now if last_run_at: - base_time = _ensure_aware(datetime.fromisoformat(last_run_at)) + try: + base_time = _ensure_aware(datetime.fromisoformat(last_run_at)) + except Exception: + base_time = now cron = croniter(expr, base_time) next_run = cron.get_next(datetime) return next_run.isoformat() @@ -1702,6 +1710,63 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]: if not isinstance(rj.get("schedule"), dict): rj["schedule"] = {} needs_save = True + + # Normalize malformed "next_run_at" records (direct jobs.json edit, + # corruption, migration, or buggy writer). If present but not a valid + # ISO string, datetime.fromisoformat(next_run) later raises and aborts + # the entire scan *before* save_jobs(). Healthy siblings then lose any + # fast-forwarded next_run_at (same class of bug as bad "id" or "schedule"). + # Strip the bad value so the existing "no next_run_at" recovery path + # recomputes a sane value and persists it for this job. + for j in jobs: + nr = j.get("next_run_at") + if nr is not None: + if not isinstance(nr, str): + j.pop("next_run_at", None) + needs_save = True + else: + try: + datetime.fromisoformat(nr) + except Exception: + j.pop("next_run_at", None) + needs_save = True + for rj in raw_jobs: + nr = rj.get("next_run_at") + if nr is not None: + if not isinstance(nr, str): + rj.pop("next_run_at", None) + needs_save = True + else: + try: + datetime.fromisoformat(nr) + except Exception: + rj.pop("next_run_at", None) + needs_save = True + + # Same treatment for last_run_at (used as base in recovery / compute_next_run). + for j in jobs: + lr = j.get("last_run_at") + if lr is not None and not isinstance(lr, str): + j.pop("last_run_at", None) + needs_save = True + elif isinstance(lr, str): + try: + datetime.fromisoformat(lr) + except Exception: + j.pop("last_run_at", None) + needs_save = True + for rj in raw_jobs: + lr = rj.get("last_run_at") + if lr is not None and not isinstance(lr, str): + rj.pop("last_run_at", None) + needs_save = True + elif isinstance(lr, str): + try: + datetime.fromisoformat(lr) + except Exception: + rj.pop("last_run_at", None) + needs_save = True + # Resolve the one-shot running-claim stale-recovery TTL once per scan # (derived from HERMES_CRON_TIMEOUT). See _oneshot_run_claim_ttl_seconds. _run_claim_ttl = _oneshot_run_claim_ttl_seconds() diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index ada767e7e0d..0c9eef4c0a9 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -1446,6 +1446,63 @@ class TestMarkJobRunConcurrency: ) +class TestBadNextRunAtRecovery: + """Regression: malformed next_run_at must not crash the due scan or starve siblings. + + Mirrors the id-less and non-dict-schedule patterns: a single bad persisted + record in jobs.json must not abort _get_due_jobs_locked before save. + """ + + def test_bad_next_run_at_does_not_crash_or_block_sibling_jobs(self, tmp_cron_dir): + """One job with unparseable next_run_at + one healthy due sibling. + + get_due_jobs must succeed and return the healthy job; the bad record + must be repaired (next_run_at cleared so recovery can set a sane value). + """ + from datetime import timezone, timedelta as td + now = datetime.now(timezone.utc) + past = (now - td(seconds=30)).isoformat() + future = (now + td(days=1)).isoformat() + + # Bad record: next_run_at is not a valid ISO string (e.g. from hand-edit or corruption) + # Healthy sibling is past due with good schedule. + bad_job = { + "id": "bad-next", + "schedule": {"kind": "interval", "minutes": 60}, + "next_run_at": "not-a-valid-iso-timestamp!!!", + "enabled": True, + "created_at": past, + } + good_job = { + "id": "good-sibling", + "schedule": {"kind": "interval", "minutes": 5}, + "next_run_at": past, + "enabled": True, + "created_at": past, + } + save_jobs([bad_job, good_job]) + + # Must not raise + due = get_due_jobs() + + # The healthy job must still be returned + ids = [j["id"] for j in due] + assert "good-sibling" in ids, f"healthy sibling missing from due jobs: {ids}" + assert "bad-next" not in ids # bad one may be repaired and/or not yet due after repair + + # Bad job should have been auto-repaired (next_run_at stripped or fixed) + repaired = get_job("bad-next") + assert repaired is not None + nr = repaired.get("next_run_at") + if nr is not None: + # If still present it must now be parseable + datetime.fromisoformat(nr) + + # Calling again must remain stable (no crash on re-scan) + due2 = get_due_jobs() + assert any(j["id"] == "good-sibling" for j in due2) + + class TestSaveJobOutput: def test_creates_output_file(self, tmp_cron_dir): output_file = save_job_output("test123", "# Results\nEverything ok.")