diff --git a/cron/occurrences.py b/cron/occurrences.py index 846419ba54..59e5d02cca 100644 --- a/cron/occurrences.py +++ b/cron/occurrences.py @@ -31,7 +31,7 @@ def completed_occurrence(job, instant): try: with _transaction() as conn: rows = conn.execute( - "SELECT finished_at, claimed_at FROM executions " + "SELECT id, finished_at, claimed_at FROM executions " "WHERE job_id=? AND scheduled_instant=? " "AND status='completed'", (str(job['id']), instant) ).fetchall() @@ -40,6 +40,13 @@ def completed_occurrence(job, instant): # Legacy or malformed timestamps remain proof; only positively identified poison # rows — completions recorded before their claimed occurrence — are ignored. if completed_at is None or datetime.fromisoformat(completed_at) >= earliest_real: + # Both dedup gates (due scan and fire claim) consume the slot on True without a + # run or a ledger row, so this line is the only trace the skip leaves (#111414). + logger.warning( + "Job '%s' (%s): scheduled occurrence %s was already completed by execution " + "%s (finished %s); skipping the due slot without a new run", + job.get("name", job.get("id")), job.get("id"), instant, row["id"], + row["finished_at"] or row["claimed_at"]) return True return False except Exception: diff --git a/tests/cron/test_scheduled_occurrence.py b/tests/cron/test_scheduled_occurrence.py index 4ff40505c4..0de70a3a53 100644 --- a/tests/cron/test_scheduled_occurrence.py +++ b/tests/cron/test_scheduled_occurrence.py @@ -178,3 +178,34 @@ def test_completion_before_occurrence_does_not_prove_slot_completed(tmp_path, mo executions.finish_execution(legitimate['id'], success=True) assert completed_occurrence({'id': 'job'}, slot) + + +def test_completed_occurrence_skip_names_job_slot_and_row(tmp_path, monkeypatch, caplog): + """A due slot the dedup gate consumes (already completed, e.g. after a jobs.json rollback) + leaves no run and no ledger row, so the skip itself must be logged with the job, the + instant and the completed execution (#111414: a consumed slot left zero trace).""" + import logging + from datetime import timedelta + + from cron import executions, jobs + from hermes_time import now + + monkeypatch.setattr(executions, 'EXECUTIONS_FILE', tmp_path / 'executions.db') + slot = (now() - timedelta(minutes=2)).isoformat() + with jobs.use_cron_store(tmp_path / 'cron'): + stored = jobs.create_job(prompt='test', schedule='every 4h') + rows = jobs.load_jobs() + rows[0]['next_run_at'] = slot + jobs.save_jobs(rows) + completed = executions.create_execution(stored['id'], source='builtin', scheduled_instant=slot) + executions.finish_execution(completed['id'], success=True) + + with caplog.at_level(logging.WARNING, logger='cron.occurrences'): + assert jobs.get_due_jobs() == [] + + assert jobs.load_jobs()[0]['next_run_at'] != slot + skip = [r.getMessage() for r in caplog.records if completed['id'] in r.getMessage()] + assert len(skip) == 1, caplog.text + assert stored['id'] in skip[0] + from cron.occurrences import scheduled_instant + assert scheduled_instant(slot) in skip[0]