"""Chained scheduled jobs — ordering as a declaration, not a minute gap. At least eight ordering dependencies in this codebase were expressed as the gap between two cron minute fields, with the intent recorded only in prose: *"每天 07:00 UTC(在 ai_spending.period_rollup_refresh 06:30 之后)"*, *"紧随扫描任务"*, *"排在预编译之后:先有证据,再谈覆盖率"*. **A minute gap is a promise the platform cannot keep.** The predecessor overruns it and the successor reads stale data, silently. Production makes overrun the normal case rather than the exception: in the pre-dawn window a job waits for one of four heavy concurrency slots before it even starts — measured, ``cmdb.grants`` for aws-sg queued **1 384 seconds** in front of a nine-second run — and ``cmdb.roles`` for aws-primary averages **1 055 seconds** against a nine-minute gap to the job behind it. A chained row says what was meant: *run me when that one finishes*. It carries no cron and no interval; the worker attaches no APScheduler trigger for it. Two mechanisms, deliberately separated -------------------------------------- **The fast path** is the executor: when a job reaches a terminal status it spawns its children immediately. This is a latency optimisation. **The correctness guarantee** is :func:`plan_advances`, run by the ``scheduler.chain_advance`` cron. It compares timestamps rather than listening for events, so it converges no matter *how* the fast path was missed — a deploy that killed the worker mid-chain, a Redis blip, a process OOM, a row an operator chained up five minutes ago. Anything event-driven has a hole exactly the size of "the event didn't happen"; a reconciler does not. Ordering, not a success precondition ------------------------------------ A chain advances on **every terminal status except ``interrupted``** — ``ok``, ``skipped`` and ``failed`` all release the successor. This surprises people, so: the eight pairs above all run their successor *unconditionally today*, because it has its own cron. Blocking on failure would be a behaviour change that **reduces** coverage — a failed ``leaver_assurance`` would newly stop the roster reconcile from running at all, and a failed ``period_rollup_refresh`` would stop the drift check from ever reporting. Neither is an improvement; both would be a new outage shape introduced by a refactor that was supposed to be safe. What a chain buys is that the successor starts *after* the predecessor is genuinely done, instead of after a clock guess. ``interrupted`` is the exception because the process is on its way out: spawning work during a shutdown drain is pointless, and the reconciler picks the chain back up on the next boot. """ from __future__ import annotations import logging from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException from app.scheduled_jobs.models import ( TRIGGER_CHAINED, ScheduledJob, ScheduledJobRun, ) logger = logging.getLogger(__name__) #: Statuses that release the next job in a chain. See the module #: docstring — this is an ordering edge, not a success precondition. CHAIN_ADVANCING_STATUSES = ("ok", "skipped", "failed") #: Hard ceiling on how deep a chain may be. A cycle cannot be created #: through the API (:func:`assert_no_cycle` rejects it) or through a #: reviewed migration, so this bound exists for the one remaining way in #: — a direct DB edit — and for the ordinary case of someone building a #: chain longer than any real pipeline. Walking a cycle without a bound #: is an infinite loop inside a cron. MAX_CHAIN_DEPTH = 24 @dataclass(slots=True, frozen=True) class ChainAdvance: """One chained job the reconciler believes is overdue.""" job_id: UUID job_name: str parent_id: UUID parent_name: str #: When the child last ran, or ``None`` if it never has. last_run_at: datetime | None reason: str async def load_parent_map(session: AsyncSession) -> dict[str, str | None]: """``{job_id: parent_id or None}`` over every row, as strings. One query. Used for cycle detection and depth bounding, both of which need the whole graph rather than one edge. """ rows = ( await session.execute( select(ScheduledJob.id, ScheduledJob.chain_after_job_id), ) ).all() return {str(row[0]): (str(row[1]) if row[1] else None) for row in rows} def chain_depth( job_id: str, parent_map: dict[str, str | None], ) -> int | None: """Hops from ``job_id`` up to a clock-triggered root. ``0`` for a root. ``None`` when the walk exceeds :data:`MAX_CHAIN_DEPTH` or revisits a node — i.e. the graph is cyclic or absurdly deep, and the caller must refuse to act rather than loop. """ seen: set[str] = set() depth = 0 cursor: str | None = job_id while cursor is not None: if cursor in seen or depth > MAX_CHAIN_DEPTH: return None seen.add(cursor) parent = parent_map.get(cursor) if parent is None: return depth depth += 1 cursor = parent return depth def assert_no_cycle( *, job_id: UUID, new_parent_id: UUID | None, parent_map: dict[str, str | None], ) -> None: """Reject a chain edit that would create a cycle or self-reference. Validated at the write boundary because a cycle is not a state the runtime can recover from gracefully: every member would keep re-triggering the next forever, each run burning a concurrency slot and whatever vendor quota its handler touches. Cheaper to make it unrepresentable. """ if new_parent_id is None: return if str(new_parent_id) == str(job_id): raise AppException( code="SCHEDULED_JOB_CHAIN_SELF", message="一个任务不能接在自己后面。", status_code=400, ) # Walk up from the proposed parent: if we reach this job, the edge # closes a loop. probe = dict(parent_map) probe[str(job_id)] = str(new_parent_id) if chain_depth(str(job_id), probe) is None: raise AppException( code="SCHEDULED_JOB_CHAIN_CYCLE", message=( "这条依赖会形成环(或链过长)。任务链必须能一路回溯到一个" "按时钟触发的起点。" ), status_code=400, ) async def children_of( session: AsyncSession, parent_id: UUID, *, enabled_only: bool = True, ) -> list[ScheduledJob]: """Enabled chained rows whose parent is ``parent_id``.""" query = select(ScheduledJob).where( ScheduledJob.chain_after_job_id == parent_id, ScheduledJob.trigger_type == TRIGGER_CHAINED, ) if enabled_only: query = query.where(ScheduledJob.enabled.is_(True)) return list( (await session.execute(query.order_by(ScheduledJob.name))) .scalars() .all() ) async def spawn_children( session: AsyncSession, *, parent_id: UUID, parent_status: str, parent_run_id: UUID | None, ) -> list[str]: """Fire every chained successor of a job that just finished. Returns the names spawned, for the parent's run summary. Detached via ``background.spawn`` on purpose. ``execute_job`` must never be awaited in-band from inside another execution: the concurrency gate is acquired around every run, so a parent that waited for its child while holding a slot could deadlock the gate against itself. The existing in-process fan-out (``run_cost_facts_monthclose``) detaches for the same reason, and the gate's no-self-deadlock argument depends on it staying true. """ if parent_status not in CHAIN_ADVANCING_STATUSES: return [] children = await children_of(session, parent_id) if not children: return [] parent_map = await load_parent_map(session) from app.core import background from app.scheduled_jobs.service import execute_job spawned: list[str] = [] for child in children: if chain_depth(str(child.id), parent_map) is None: logger.error( "refusing to advance chain into %s: its parent chain is " "cyclic or deeper than %d hops", child.name, MAX_CHAIN_DEPTH, ) continue trigger = ( f"chain:{parent_run_id}" if parent_run_id else "chain:unknown" ) background.spawn( execute_job(child.id, triggered_by=trigger), name=f"chain-{child.name}", ) spawned.append(child.name) return spawned async def plan_advances( session: AsyncSession, *, now: datetime | None = None, ) -> list[ChainAdvance]: """Chained jobs whose parent has moved on without them. The reconciler's whole judgement, expressed as a comparison of two timestamps rather than a memory of an event: * the parent has run at least once, and * the child has never run, or last ran **before** the parent's most recent run, and * neither end is running right now — not the parent (its own completion will spawn the child in a moment, so firing here would double it) and not the child itself (its clock is written when it finishes, so a run in flight still reads as behind its parent; the re-fire can only lose the single-flight lock race and leave a junk ``skipped`` row). The condition self-clears: once the child runs, its ``last_run_at`` moves past the parent's and it stops being overdue. That is what makes this safe to run every few minutes. **Except for a successor that cannot run at all.** Its clock never moves, so the predicate never clears and the reconciler fires it on every tick forever — burning a place in the caller's per-tick cap that a genuine advance needs. There are two ways to be unable to run and this query excludes both: ``enabled = 0``, and a row bound to a disabled or deleted integration, which ``execute_job`` turns away at its pre-gate before any work happens. The second exclusion was missing when the pre-gate shipped, which is the whole reason it is spelled out here. It is deliberately *not* fixed by having the pre-gate move the clock: the compliance posture gate reads ``last_run_at`` to decide whether a control's signal is fresh, and a job advertising a fresh clock while doing no work is the exact failure #2217 closed. Excluding is cheap; lying is not. Nothing is lost by excluding: the moment the integration comes back, the successor is still behind its parent and the next tick releases it. One hop per tick, by design. Advancing a whole chain inside one tick would mean firing a grandchild before its parent's fresh run had finished — the exact inversion this module exists to prevent. The executor's direct spawn is what makes multi-hop catch-up rare; when it does happen (a deploy landing mid-chain), the chain walks itself forward one tick at a time. """ now = now or datetime.now(UTC) parent = ScheduledJob.__table__.alias("parent") rows = ( await session.execute( select( ScheduledJob.id, ScheduledJob.name, ScheduledJob.last_run_at, parent.c.id, parent.c.name, parent.c.last_run_at, ScheduledJob.config, ) .join(parent, parent.c.id == ScheduledJob.chain_after_job_id) .where( ScheduledJob.enabled.is_(True), ScheduledJob.trigger_type == TRIGGER_CHAINED, parent.c.last_run_at.isnot(None), ) # Longest-waiting first. The caller caps how many advances it # fires per tick, and an unordered result lets the database # hand back the same slice every time — turning a cap that was # meant to bound a burst into indefinite starvation for # whichever member falls outside it. MySQL and TiDB both sort # NULL first on ASC, which is the order we want anyway: a # successor that has never run is the most overdue there is. .order_by(ScheduledJob.last_run_at.asc()) ) ).all() if not rows: return [] # Who is mid-run right now? Both ends matter, for different reasons. # # A **parent** mid-run is about to spawn its children itself, so # firing here would double them. # # A **child** mid-run is subtler and was missed at first: a job's # ``last_run_at`` is written at the END of its run, so for the whole # duration of that run it still reads as behind its parent. Without # this the reconciler re-fires it on every tick while it is already # running; each attempt loses the single-flight lock race and leaves # a ``skipped`` row behind. Bounded, since the lock frees when the # run ends — but one triad member measures 17 minutes, so it is a # steady drip of junk into a history capped at 100 rows, and it makes # "single-flight lock busy" fire routinely when it should mean a real # collision worth investigating. involved_ids = {row[3] for row in rows} | {row[0] for row in rows} running_now = { r[0] for r in ( await session.execute( select(ScheduledJobRun.job_id).where( ScheduledJobRun.job_id.in_(involved_ids), ScheduledJobRun.status == "running", ), ) ).all() } parent_map = await load_parent_map(session) # One query for the whole integration table, not one per candidate: # a fleet re-chained at once would otherwise turn a reconciler tick # into N round trips. from app.scheduled_jobs.hygiene import ( config_integration_id_of, load_integration_states, ) integration_states = await load_integration_states(session) advances: list[ChainAdvance] = [] for ( job_id, job_name, last_run_at, p_id, p_name, p_last_run, config, ) in rows: if p_id in running_now or job_id in running_now: continue pinned = config_integration_id_of(config) if pinned is not None and not integration_states.get(pinned, False): # Bound to an integration that is disabled or gone. The # pre-gate would turn this run away without doing any work, # and — because a suppressed skip does not move the clock — # it would still read as overdue on the next tick, and the # next, forever. Skip it silently; it is not a fault, and # ``GET /scheduled-jobs/hygiene`` is where a reader sees # these rows named. continue p_last = _as_utc(p_last_run) mine = _as_utc(last_run_at) if p_last is None: continue if mine is not None and mine >= p_last: continue if chain_depth(str(job_id), parent_map) is None: logger.error( "chain_advance skipping %s: cyclic or over-deep parent chain", job_name, ) continue advances.append( ChainAdvance( job_id=job_id, job_name=job_name, parent_id=p_id, parent_name=p_name, last_run_at=mine, reason=( "never ran" if mine is None else "last ran before the parent's most recent run" ), ), ) return advances def _as_utc(value: datetime | None) -> datetime | None: """MySQL/TiDB strip tzinfo on round-trip even for ``timezone=True`` columns, so a freshly-loaded timestamp arrives naive and comparing it against an aware one raises ``TypeError``.""" if value is None: return None return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) def _blocked_reason( job: ScheduledJob, integration_states: dict[str, bool], ) -> str | None: """Why ``job`` cannot advance right now, or ``None`` if it can. Mirrors the *cannot-run* exclusions in :func:`plan_advances` — same order, and the same ``config_integration_id_of`` predicate — because a panel that disagreed with the reconciler about which members are stuck would be worse than one that said nothing. Deliberately silent about the transient exclusion the reconciler also applies (a run already in flight at either end). That one clears itself within one run and is not a state anybody needs to act on; reporting it would make a healthy chain flicker "blocked" every night, which is how a real signal gets learned as noise. """ from app.scheduled_jobs.hygiene import config_integration_id_of if not job.enabled: return "disabled" pinned = config_integration_id_of(job.config or {}) if pinned is None: return None state = integration_states.get(pinned) if state is None: return "integration_missing" if not state: return "integration_disabled" return None async def chain_graph(session: AsyncSession) -> dict[str, Any]: """Every chain in the ledger, rooted at its clock-triggered job. Backs the admin console's chain panel. Each root carries its ordered members with the measured duration of each hop, because the question an operator actually has is "how long does this chain take, and where does it spend the time" — the answer to which is what decides whether the chain fits in its window. """ from app.scheduled_jobs.hygiene import ( handler_weight, load_integration_states, ) jobs = list((await session.execute(select(ScheduledJob))).scalars().all()) integration_states = await load_integration_states(session) by_id = {str(j.id): j for j in jobs} parent_map = {str(j.id): (str(j.chain_after_job_id) if j.chain_after_job_id else None) for j in jobs} children_by_parent: dict[str, list[str]] = {} for jid, pid in parent_map.items(): if pid is None: continue if by_id[jid].trigger_type != TRIGGER_CHAINED: continue children_by_parent.setdefault(pid, []).append(jid) for kids in children_by_parent.values(): kids.sort(key=lambda jid: by_id[jid].name) # Average duration per job over the recent ledger — one grouped query # for every job that participates in a chain. participating = set(children_by_parent) | { jid for kids in children_by_parent.values() for jid in kids } durations: dict[str, int] = {} rescued: dict[str, int] = {} if participating: from datetime import timedelta from sqlalchemy import func participating_ids = [UUID(j) for j in participating] rows = ( await session.execute( select( ScheduledJobRun.job_id, func.avg(ScheduledJobRun.duration_ms), ) .where( ScheduledJobRun.job_id.in_(participating_ids), ScheduledJobRun.status.in_(("ok", "skipped")), ScheduledJobRun.duration_ms.isnot(None), ) .group_by(ScheduledJobRun.job_id), ) ).all() durations = {str(r[0]): int(r[1] or 0) for r in rows} # How often the reconciler had to rescue each member in the last # week. The fast path (the predecessor spawning its successor on # completion) is supposed to be how a chain moves; the reconciler # is the safety net. A member that needs rescuing every day is # telling you the fast path is broken FOR IT — a deploy landing # in its window every night, a predecessor whose write-back keeps # failing — and without this number nobody would ever look, # because the chain still advances and everything reads green. since = datetime.now(UTC) - timedelta(days=7) rescue_rows = ( await session.execute( select(ScheduledJobRun.job_id, func.count()) .where( ScheduledJobRun.job_id.in_(participating_ids), ScheduledJobRun.triggered_by == "chain_advance", ScheduledJobRun.started_at >= since, ) .group_by(ScheduledJobRun.job_id), ) ).all() rescued = {str(r[0]): int(r[1] or 0) for r in rescue_rows} def _node(jid: str, depth: int) -> dict[str, Any]: job = by_id[jid] return { "id": jid, "name": job.name, "display_name": job.display_name, "handler_key": job.handler_key, "enabled": bool(job.enabled), "weight": handler_weight(job.handler_key), #: Why this member is not advancing, or ``None``. #: #: A chain whose members all read "enabled" but which has #: silently stopped moving is the worst thing this panel can #: show, and both causes are invisible in the job list. #: ``plan_advances`` excludes a member bound to a disabled or #: deleted integration — correctly, since ``execute_job`` #: would turn it away anyway — and without saying so here the #: panel would render a healthy-looking chain that does #: nothing. Naming it turns "why did this stop" from a DB #: session into a glance. "blocked_reason": _blocked_reason(job, integration_states), "trigger_type": job.trigger_type, "cron_expr": job.cron_expr, "interval_seconds": job.interval_seconds, "timezone": job.timezone, "depth": depth, "avg_duration_ms": durations.get(jid, 0), #: Times the reconciler had to release this member in the #: last 7 days because the predecessor's own spawn did not. "rescued_7d": rescued.get(jid, 0), "last_status": job.last_status, "last_run_at": ( last_run.isoformat() if (last_run := _as_utc(job.last_run_at)) is not None else None ), } chains: list[dict[str, Any]] = [] for root_id, _kids in sorted( children_by_parent.items(), key=lambda kv: by_id[kv[0]].name, ): root = by_id[root_id] # Only real roots start a chain listing; a chained middle link is # rendered inside its own root's chain, not as a chain of its own. if root.trigger_type == TRIGGER_CHAINED: continue members: list[dict[str, Any]] = [_node(root_id, 0)] frontier = [(kid, 1) for kid in children_by_parent.get(root_id, [])] guard = 0 while frontier and guard <= MAX_CHAIN_DEPTH * 8: guard += 1 jid, depth = frontier.pop(0) members.append(_node(jid, depth)) frontier.extend( (kid, depth + 1) for kid in children_by_parent.get(jid, []) ) chains.append( { "root_id": root_id, "root_name": root.name, "root_display_name": root.display_name, "members": members, "total_avg_duration_ms": sum( m["avg_duration_ms"] for m in members ), "member_count": len(members), "rescued_7d": sum(m["rescued_7d"] for m in members), }, ) orphans = [ { "id": str(j.id), "name": j.name, "display_name": j.display_name, "enabled": bool(j.enabled), } for j in jobs if j.trigger_type == TRIGGER_CHAINED and j.chain_after_job_id is None ] return { "checked_at": datetime.now(UTC).isoformat(), "chains": sorted(chains, key=lambda c: -c["total_avg_duration_ms"]), # A chained row with no parent can never fire. It happens when a # parent is deleted (the FK sets this NULL rather than cascading # the delete), so it is a real state worth naming rather than an # impossible one. "orphans": orphans, } __all__ = [ "CHAIN_ADVANCING_STATUSES", "MAX_CHAIN_DEPTH", "ChainAdvance", "assert_no_cycle", "chain_depth", "chain_graph", "children_of", "load_parent_map", "plan_advances", "spawn_children", ]