fix(observability): stop skipped runs from masking a stalled job

Two defects the review caught, both in the case this telemetry exists for.

The last-run clock was refreshed on every event, including runs that never
executed. A job repeatedly missed or skipped kept looking recently run, so the
"time since last run" alert the docs recommend would stay quiet through exactly
the outage it is meant to catch. Only a run that executed moves the clock now;
the skip is still counted, just not as a run.

Start times also leaked on a missed run. MAX_INSTANCES is emitted by the
scheduler before it submits, so there is nothing to release, and the previous
code treated MISSED the same way. MISSED comes from the executor after the
submit (`apscheduler/executors/base.py`), so its start time was recorded and
never freed, growing one entry per miss for the life of the process. The
comment claiming neither follows a submission was wrong for MISSED.

Also corrects the injected clock's annotation, which used `Final` in a parameter
position where it is not valid, and adds the missing `Final` on the lock
telemetry's logger lookup.

litellm/integrations/prometheus.py

@@ -3207,7 +3207,8 @@ class PrometheusLogger(CustomLogger):
     def record_scheduled_job_run(self, run: JobRun) -> None:
         """Publish one completed run of a scheduled background job."""
         self.litellm_scheduled_job_runs_total.labels(job_name=run.job_name, result=run.result.value).inc()
-        self.litellm_scheduled_job_last_run_timestamp.labels(job_name=run.job_name).set(time.time())
+        if run.did_execute:
+            self.litellm_scheduled_job_last_run_timestamp.labels(job_name=run.job_name).set(time.time())
         if run.duration_seconds is not None:
             self.litellm_scheduled_job_duration_seconds.labels(job_name=run.job_name).observe(run.duration_seconds)
         if run.items_processed:

litellm/proxy/common_utils/scheduled_job_metrics.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 
 import re
 import time
+from collections.abc import Callable
 from dataclasses import dataclass
 from enum import Enum
 from typing import TYPE_CHECKING, Final
@@ -65,6 +66,16 @@ class JobRun:
     duration_seconds: float | None
     items_processed: int | None
 
+    @property
+    def did_execute(self) -> bool:
+        """Whether the job body actually ran.
+
+        A skipped run must not refresh the last-run clock, or a job that never
+        executes keeps looking recently run and the "time since last run" alert
+        stays quiet through exactly the outage it exists to catch.
+        """
+        return self.result in (JobResult.SUCCESS, JobResult.ERROR)
+
 
 def _label_for(job_id: str) -> str:
     """The job id, unless APScheduler generated it."""
@@ -95,7 +106,7 @@ class ScheduledJobMetricsListener:
     on the job id alone would let those runs consume each other's start times.
     """
 
-    def __init__(self, *, monotonic: Final = time.monotonic) -> None:
+    def __init__(self, *, monotonic: Callable[[], float] = time.monotonic) -> None:
         self._monotonic: Final = monotonic
         self._started_at: dict[str, float] = {}  # mutable-ok: start times arrive one scheduler event at a time
 
@@ -124,14 +135,16 @@ class ScheduledJobMetricsListener:
                 self._started_at[self._key(event.job_id, scheduled)] = now
             return None
 
-        # Neither of these follows a submission of its own. MAX_INSTANCES in
-        # particular arrives while the previous run is still going, so popping
-        # before this point would steal that run's start time and drop the
-        # duration of exactly the overruns worth measuring.
-        if event.code == EVENT_JOB_MISSED:
-            return JobRun(job_name, JobResult.MISSED, None, None)
+        # MAX_INSTANCES is emitted by the scheduler before it submits, while the
+        # previous run is still going, so popping here would steal that run's
+        # start time and drop the duration of exactly the overruns worth
+        # measuring. MISSED comes from the executor after the submit, so its
+        # start time was recorded and has to be released or it leaks.
         if event.code == EVENT_JOB_MAX_INSTANCES:
             return JobRun(job_name, JobResult.MAX_INSTANCES, None, None)
+        if event.code == EVENT_JOB_MISSED:
+            self._started_at.pop(self._key(event.job_id, getattr(event, "scheduled_run_time", None)), None)
+            return JobRun(job_name, JobResult.MISSED, None, None)
 
         started_at: Final = self._started_at.pop(
             self._key(event.job_id, getattr(event, "scheduled_run_time", None)), None

litellm/proxy/db/db_transaction_queue/pod_lock_manager.py

@@ -26,7 +26,7 @@ def _record_lock_attempt(cronjob_id: str, result: LockAttemptResult) -> None:
     try:
         from litellm.integrations.prometheus import PrometheusLogger
 
-        logger = PrometheusLogger.get_instance()
+        logger: Final = PrometheusLogger.get_instance()
         if logger is not None:
             logger.record_cronjob_lock_attempt(cronjob_id, result)
     except Exception as e:  # noqa: BLE001  # telemetry must not stop a job from running

tests/test_litellm/integrations/test_prometheus_cronjob_lock_metrics.py

@@ -35,3 +35,43 @@ def test_the_lock_metric_documents_every_result_it_can_emit(monkeypatch):
     documented = {value.strip() for value in advertised.group(1).split(",")}
 
     assert documented == {result.value for result in LockAttemptResult}
+
+def _fresh_logger(monkeypatch):
+    from prometheus_client import REGISTRY
+
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(litellm, "success_callback", [])
+    for collector in list(REGISTRY._collector_to_names.keys()):
+        try:
+            REGISTRY.unregister(collector)
+        except Exception:
+            pass
+    return PrometheusLogger()
+
+
+def test_only_a_run_that_executed_refreshes_the_last_run_clock(monkeypatch):
+    """The recommended alert is time since last run, so a skipped run bumping the
+    clock would keep a job that never executes looking healthy."""
+    from prometheus_client import REGISTRY
+
+    from litellm.proxy.common_utils.scheduled_job_metrics import JobResult, JobRun
+
+    logger = _fresh_logger(monkeypatch)
+
+    def clock():
+        return REGISTRY.get_sample_value(
+            "litellm_scheduled_job_last_run_timestamp", {"job_name": "job"}
+        )
+
+    logger.record_scheduled_job_run(JobRun("job", JobResult.SUCCESS, 1.0, None))
+    after_success = clock()
+    assert after_success is not None and after_success > 0
+
+    logger.record_scheduled_job_run(JobRun("job", JobResult.MISSED, None, None))
+    logger.record_scheduled_job_run(JobRun("job", JobResult.MAX_INSTANCES, None, None))
+
+    assert clock() == after_success, "a skipped run must not move the last-run clock"
+    assert REGISTRY.get_sample_value(
+        "litellm_scheduled_job_runs_total", {"job_name": "job", "result": "missed"}
+    ) == 1.0, "the skip is still counted, just not as a run"
+

tests/test_litellm/proxy/common_utils/test_scheduled_job_metrics.py

@@ -17,6 +17,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler
 sys.path.insert(0, os.path.abspath("../../../.."))
 
 from litellm.proxy.common_utils.scheduled_job_metrics import (
+    JobRun,
     JobResult,
     ScheduledJobMetricsListener,
 )
@@ -337,3 +338,67 @@ def test_handle_swallows_a_failure_so_the_scheduler_keeps_running():
     with patch.object(ScheduledJobMetricsListener, "_publish", side_effect=RuntimeError("metrics down")):
         listener.handle(event)  # must not raise
 
+def test_a_skipped_run_does_not_refresh_the_last_run_clock():
+    """The recommended alert is time since last run. If a skipped run bumped the
+    clock, a job that never executes would look recently run and the alert would
+    stay quiet through exactly the outage this telemetry exists to catch."""
+    executed = JobRun("job", JobResult.SUCCESS, 1.0, None)
+    failed = JobRun("job", JobResult.ERROR, 1.0, None)
+    missed = JobRun("job", JobResult.MISSED, None, None)
+    overrun = JobRun("job", JobResult.MAX_INSTANCES, None, None)
+
+    assert executed.did_execute is True
+    assert failed.did_execute is True, "a job that ran and raised still ran"
+    assert missed.did_execute is False
+    assert overrun.did_execute is False
+
+
+def test_a_missed_run_releases_the_start_time_it_was_given():
+    """APScheduler emits MISSED from the executor, after the scheduler already
+    emitted SUBMITTED, so a start time was recorded for that run. Dropping the
+    event without releasing it leaks one entry per miss for the life of the
+    process."""
+    import datetime
+
+    from apscheduler.events import (
+        EVENT_JOB_MISSED,
+        EVENT_JOB_SUBMITTED,
+        JobExecutionEvent,
+        JobSubmissionEvent,
+    )
+
+    when = datetime.datetime(2026, 1, 1, 0, 0, 0)
+    listener = ScheduledJobMetricsListener()
+
+    listener._to_run(JobSubmissionEvent(EVENT_JOB_SUBMITTED, "slow_job", "default", [when]))
+    assert len(listener._started_at) == 1, "the submit records a start time"
+
+    run = listener._to_run(JobExecutionEvent(EVENT_JOB_MISSED, "slow_job", "default", when))
+
+    assert run is not None and run.result is JobResult.MISSED
+    assert listener._started_at == {}, "the missed run must not leave its start time behind"
+
+
+def test_an_overrun_keeps_the_running_jobs_start_time():
+    """MAX_INSTANCES is emitted before the submit, while the previous run is
+    still going, so it must not consume that run's start time."""
+    import datetime
+
+    from apscheduler.events import (
+        EVENT_JOB_EXECUTED,
+        EVENT_JOB_MAX_INSTANCES,
+        EVENT_JOB_SUBMITTED,
+        JobExecutionEvent,
+        JobSubmissionEvent,
+    )
+
+    when = datetime.datetime(2026, 1, 1, 0, 0, 0)
+    ticks = iter([100.0, 104.0])
+    listener = ScheduledJobMetricsListener(monotonic=lambda: next(ticks))
+
+    listener._to_run(JobSubmissionEvent(EVENT_JOB_SUBMITTED, "slow_job", "default", [when]))
+    listener._to_run(JobExecutionEvent(EVENT_JOB_MAX_INSTANCES, "slow_job", "default", when))
+    finished = listener._to_run(JobExecutionEvent(EVENT_JOB_EXECUTED, "slow_job", "default", when, retval=None))
+
+    assert finished is not None and finished.duration_seconds == pytest.approx(4.0)
+