"""Worker scheduler-liveness watchdog — contract pins. History: * Round-20 (2026-05-12) — watchdog originally watched the FASTEST enabled interval-based job's staleness. That assumption held until PR #316 (2026-05-15) found ``identity_sync.dispatch`` silently wedged for 9 days while the watchdog read green: the import-bug only nuked ONE job's APScheduler dispatch, the faster ``provisioning.dispatch`` kept firing, and "fastest-only" measurement missed the per-job blind spot entirely. * PR #316 — switched to **worst-stale** measurement: the helper now picks the job with the highest ``stale_seconds / interval_seconds`` ratio across all enabled interval rows, and returns ``(stale_seconds, interval_seconds, job_name)`` so the wedge log can identify the responsible job by name. This file pins the contracts that keep the watchdog from either under-firing (incident path) or over-firing (false-positive restart loop): T1 — Returns ``None`` when there are no enabled interval-based jobs to measure against. Cron-only worker configurations have no fast liveness signal; the watchdog must abstain. Tripping here would mean a permanent restart loop on any worker that doesn't happen to have an interval job. T2 — Staleness is measured from the last DISPATCH, not the last completion: ``max(last_run_at, latest run.started_at)``, falling back to ``created_at`` for a job never dispatched at all. A job never dispatched still ages into a wedge verdict (the original T2 intent); a job dispatched on time but never completing does NOT (the 2026-08-17 crash-loop — see the two T2 tests). T3 — Returns the actual elapsed seconds for a job that has run at least once. Critically, the result is identical whether the column round-trips as tz-aware or tz-naive — TiDB's ``DateTime(timezone=True)`` strips tzinfo, and a naive-vs-aware subtract bug would either raise TypeError (watchdog never trips, defeating its purpose) or silently return wrong values. T4 — **Worst-stale** selection: when multiple interval jobs are enabled, the helper measures against the one with the highest stale-to-cadence ratio, NOT the fastest cadence. This is the contract that catches a per-job wedge (one handler missing from the worker's import list, etc.) — pre-PR-#316 this exact case escaped the watchdog for 9 days in prod. The loop's grace-window + threshold-comparison logic is straight-line code with no branches that need pinning — covered by code review and the inline comments in ``app/worker.py``. """ from __future__ import annotations import datetime import uuid from contextlib import asynccontextmanager import pytest from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun from app.worker import ( SCHEDULER_WATCHDOG_STALE_THRESHOLD_SECONDS, SCHEDULER_WATCHDOG_STARTUP_GRACE_SECONDS, _measure_scheduler_staleness, ) def _wire_factory(monkeypatch, db_session) -> None: """Point ``async_session_factory`` at the test session so the helper observes the seeded rows. The helper opens its own session via that factory rather than accepting a session arg.""" @asynccontextmanager async def _factory(): yield db_session monkeypatch.setattr("app.core.database.async_session_factory", _factory) @pytest.mark.asyncio async def test_staleness_returns_none_on_cron_only_config( db_session, monkeypatch, ): """A worker config with only cron jobs (no interval-based) has no fast-cadence liveness signal. The helper must return None so the watchdog abstains. If this regresses to "0" or "inf" the watchdog enters a permanent restart loop on cron-only deployments.""" _wire_factory(monkeypatch, db_session) db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.cron_only", display_name="Cron only", handler_key="test.handler", trigger_type="cron", cron_expr="0 0 * * *", enabled=True, )) await db_session.flush() result = await _measure_scheduler_staleness() assert result is None, ( f"helper must return None on cron-only config, got {result!r}" ) @pytest.mark.asyncio async def test_staleness_measures_never_dispatched_job_from_created_at( db_session, monkeypatch, ): """An interval job that has never been DISPATCHED at all (no run rows, ``last_run_at IS NULL``) is measured from ``created_at``. This is the T2 intent — "a job that should have fired by now and hasn't is indistinguishable from a wedge" — with the clock started at row creation rather than reported as an instant ``+inf``. An old never-dispatched row still trips (here: created 1h ago on a 30s cadence ⇒ ratio 120, far past any threshold), so the wedge the original T2 protected against is still caught. """ _wire_factory(monkeypatch, db_session) now = datetime.datetime.now(datetime.UTC) db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.fast", display_name="Fast", handler_key="test.handler", trigger_type="interval", interval_seconds=30, enabled=True, last_run_at=None, created_at=now - datetime.timedelta(hours=1), )) await db_session.flush() result = await _measure_scheduler_staleness() assert result is not None stale, interval, name = result assert 3590 <= stale <= 3610, ( "never-dispatched job must age from created_at, got " f"staleness={stale!r}" ) assert interval == 30 assert name == "test.fast", ( "name must come back so the wedge log can identify the job; " f"got {name!r}" ) @pytest.mark.asyncio async def test_staleness_counts_dispatch_not_completion( db_session, monkeypatch, ): """**The 2026-08-17 contract.** A job that is dispatched on time but never COMPLETES must not read as stale. ``scheduled_jobs.last_run_at`` is written only in the scheduler's terminal write-back, so a handler that outlives the watchdog's grace window leaves it NULL forever. Measuring completions made the watchdog kill a worker whose scheduler was provably alive — and the restart guaranteed the handler never finished, so the next boot read NULL again. That deadlock crash-looped prod every ~5m20s for ~16h (60s ``intelligence.cloud_scope_baseline_bootstrap`` over 271 cloud scopes), truncating every long-running cloud-cost ingest mid-flight and surfacing as a bogus "vendor pagination truncated" cost alert. ``scheduled_job_runs.started_at`` is written at dispatch — for skipped and abandoned runs too — so it is green exactly when APScheduler is alive. """ _wire_factory(monkeypatch, db_session) now = datetime.datetime.now(datetime.UTC) job_id = uuid.uuid4() db_session.add(ScheduledJob( id=job_id, name="test.never_completes", display_name="Never completes", handler_key="test.handler", trigger_type="interval", interval_seconds=60, enabled=True, # Never reached a terminal write-back, so both stay untouched. last_run_at=None, created_at=now - datetime.timedelta(hours=16), )) # Dispatched every minute regardless — the scheduler is healthy. for age in (180, 120, 60, 5): db_session.add(ScheduledJobRun( id=uuid.uuid4(), job_id=job_id, started_at=now - datetime.timedelta(seconds=age), status="running" if age == 5 else "skipped", triggered_by="scheduler", )) await db_session.flush() result = await _measure_scheduler_staleness() assert result is not None stale, _interval, name = result assert name == "test.never_completes" assert stale < 60, ( "a job dispatched 5s ago must not read as stale; got " f"staleness={stale!r}. If this regresses, a slow handler " "crash-loops the worker and nothing long-running ever finishes." ) @pytest.mark.asyncio async def test_staleness_picks_worst_stale_job_by_ratio( db_session, monkeypatch, ): """**The PR #316 contract.** When multiple interval jobs are enabled, the helper measures against the WORST-STALE job (highest stale_seconds / interval_seconds ratio), not the fastest by interval. This catches per-job wedges where a single handler is silently dropped while the rest of the scheduler runs healthily — exactly the 9-day ``identity_sync.dispatch`` blind spot that motivated this change. Setup mirrors the prod incident in miniature: * Healthy 30s job that ran 5s ago (ratio ~0.17) * 60s job that hasn't run in 1 hour (ratio = 60) The 60s job wins on ratio even though it's not the fastest. """ _wire_factory(monkeypatch, db_session) now = datetime.datetime.now(datetime.UTC) # Healthy fastest job — pre-PR-#316 this would mask the wedge. db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.healthy_fast", display_name="Healthy fast", handler_key="test.handler", trigger_type="interval", interval_seconds=30, enabled=True, last_run_at=now - datetime.timedelta(seconds=5), )) # Wedged slower job — what watchdog must now catch. db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.wedged_medium", display_name="Wedged medium", handler_key="test.handler", trigger_type="interval", interval_seconds=60, enabled=True, last_run_at=now - datetime.timedelta(hours=1), )) await db_session.flush() result = await _measure_scheduler_staleness() assert result is not None stale, interval, name = result assert name == "test.wedged_medium", ( f"watchdog must pick the worst-stale-ratio job, got {name!r}; " "if this picks 'test.healthy_fast' the regression is the " "exact 2026-05 identity_sync.dispatch blind-spot" ) assert interval == 60 assert 3590 <= stale <= 3610 @pytest.mark.asyncio async def test_staleness_returns_interval_for_caller_to_size_grace( db_session, monkeypatch, ): """The (stale, interval, name) tuple lets the watchdog scale its grace window with the worst-stale job's cadence. Pre-PR-#316 the helper returned only (stale, interval) — PR #316 added the name so the wedge log can identify the responsible job without a separate DB round-trip during the panic. """ _wire_factory(monkeypatch, db_session) now = datetime.datetime.now(datetime.UTC) db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.medium", display_name="Medium", handler_key="test.handler", trigger_type="interval", interval_seconds=1800, # 30 min cadence enabled=True, last_run_at=now - datetime.timedelta(minutes=5), )) await db_session.flush() result = await _measure_scheduler_staleness() assert result is not None stale, interval, name = result assert interval == 1800, ( "interval must come back so the watchdog can compute its " "effective grace as max(static_grace, 2 * interval)" ) assert 295 <= stale <= 305 # ~300s assert name == "test.medium" @pytest.mark.asyncio async def test_staleness_ignores_disabled_jobs( db_session, monkeypatch, ): """Disabled jobs are not part of the wedge signal. A disabled-but- stale job must not cause the watchdog to fire — the operator legitimately turned it off and expects no liveness contract. """ _wire_factory(monkeypatch, db_session) now = datetime.datetime.now(datetime.UTC) db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.disabled_stale", display_name="Disabled stale", handler_key="test.handler", trigger_type="interval", interval_seconds=30, enabled=False, last_run_at=now - datetime.timedelta(days=10), )) db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.enabled_healthy", display_name="Enabled healthy", handler_key="test.handler", trigger_type="interval", interval_seconds=60, enabled=True, last_run_at=now - datetime.timedelta(seconds=5), )) await db_session.flush() result = await _measure_scheduler_staleness() assert result is not None _, _, name = result assert name == "test.enabled_healthy", ( "disabled rows must be filtered before ratio comparison; " f"got {name!r}" ) @pytest.mark.asyncio async def test_staleness_per_job_grace_does_not_mask_faster_wedge( db_session, monkeypatch, ): """**PR #316 round-2 self-audit fix.** A slow never-ran job (e.g. 13h cadence) must not consume the entire watchdog grace window and mask a faster job's wedge. Without per-job grace filtering inside the helper, the slow job's ratio (inf) would always win the worst-stale selection, and the caller would see ``inf > threshold`` but compare against the SLOW job's 26h grace — which the worker hasn't yet exceeded — leaving the wedge undetected. Setup mirrors the scenario: * Slow 13h job, never ran (legitimate — needs 26h grace). * 60s job, last ran 1 hour ago (a wedge). At 2h uptime, slow is still in grace; 60s is past. Helper should return the 60s job's measurement, NOT the slow job's inf. """ _wire_factory(monkeypatch, db_session) now = datetime.datetime.now(datetime.UTC) db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.slow_never_ran", display_name="Slow", handler_key="test.handler", trigger_type="interval", interval_seconds=46800, # 13h enabled=True, last_run_at=None, )) db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.fast_wedged", display_name="Fast wedged", handler_key="test.handler", trigger_type="interval", interval_seconds=60, enabled=True, last_run_at=now - datetime.timedelta(hours=1), )) await db_session.flush() # 2h uptime: slow's per-job grace (26h) NOT yet elapsed, fast's # (120s static-floor) long past. result = await _measure_scheduler_staleness(uptime_seconds=7200) assert result is not None _, interval, name = result assert name == "test.fast_wedged", ( f"per-job grace filter must exclude slow-and-young; got {name!r}. " "If this reports the slow job, the 9-day blind spot regression " "for any concurrent fast-job wedge is back." ) assert interval == 60 @pytest.mark.asyncio async def test_staleness_returns_none_when_every_job_in_grace( db_session, monkeypatch, ): """Boot-time edge: when uptime hasn't yet reached even the fastest job's per-job grace, the helper returns None and the watchdog loop's ``measurement is None`` branch keeps waiting. """ _wire_factory(monkeypatch, db_session) db_session.add(ScheduledJob( id=uuid.uuid4(), name="test.slow", display_name="Slow", handler_key="test.handler", trigger_type="interval", interval_seconds=3600, enabled=True, last_run_at=None, )) await db_session.flush() # 30s uptime — slow job needs max(static_grace_300s, 2 * 3600) = # 7200s, so still in grace. result = await _measure_scheduler_staleness(uptime_seconds=30) assert result is None, ( "every-job-in-grace must surface as None so the watchdog " "loop skips rather than fires on the only candidate" ) def test_watchdog_thresholds_are_reasonable(): """Sanity-check the threshold constants don't drift into either: * too aggressive (grace < 60s) — a slow boot would crash-loop * too tolerant (threshold > 3600s) — a real wedge wastes hours before recovery """ assert SCHEDULER_WATCHDOG_STARTUP_GRACE_SECONDS >= 120, ( "grace too short — risks tight crash loops on slow boots" ) assert SCHEDULER_WATCHDOG_STARTUP_GRACE_SECONDS <= 900, ( "grace too long — masks wedges discovered shortly after deploy" ) assert SCHEDULER_WATCHDOG_STALE_THRESHOLD_SECONDS >= 600, ( "threshold too aggressive — transient DB hiccups would trip" ) assert SCHEDULER_WATCHDOG_STALE_THRESHOLD_SECONDS <= 3600, ( "threshold too tolerant — real wedges should recover within " "an hour, not days like the 2026-05 incident" )