"""Seed the chain reconciler and convert the first six ordering pairs. Revision ID: 616 Revises: 615 Create Date: 2026-09-06 Migration 615 added the column; this one uses it, on the pairs where the conversion is provably behaviour-preserving. **The bar for inclusion.** A pair qualifies only when the successor's own schedule already sits in the **same window on the same day** as its predecessor, so chaining changes *when inside that window* it runs and nothing else. Two documented pairs deliberately fail that bar and are left on their crons: * ``auditor.evidence_precompile`` (03:20 on the 2nd) → ``auditor.compliance_posture_summary`` (09:00 on the **3rd**). Chaining would move a monthly report a day earlier. That is a product-visible change, not an infrastructure one, and it is not this migration's to make. * ``cmdb.cost_facts_monthclose`` (5th & 12th) → ``cmdb.cost_recon.aliyun_bq`` (14th) — the two-day gap after the *second* pass is the point of the schedule, not an accident of it. Also excluded: ``cmdb.efficiency_owner_digest``. It is **weekly** while the ``efficiency_compute`` rows it consumes are daily; chaining a weekly job behind a daily one silently makes it daily, which for a digest means seven times the notifications. **Each edge is applied only if the successor still carries the cron the code seeded** — same discipline as migration 418 and 613. An operator who has already retuned one of these rows made a decision this migration knows nothing about. **What a chained successor gives up:** its own clock. If its parent is disabled, it stops running too. For every pair here that is the honest semantics (there is no fresh data to consume), it is now *visible* in the console's chain panel instead of implied by two cron expressions, and the reconciler reports it. Downgrade restores each original cron exactly. """ from __future__ import annotations import json import uuid import sqlalchemy as sa from alembic import op revision = "616" down_revision = "615" branch_labels = None depends_on = None CHAIN_ADVANCE_JOB = { "name": "scheduler.chain_advance", "display_name": "任务链推进对账", "description": ( "每 5 分钟找出「上游已经跑过、自己却还没跟上」的 chained 任务并触发它。" "执行器在上游结束时会直接拉起下游 —— 那是快路径,只管延迟;这条任务" "才是保证:它比较的是两个时间戳而不是记住一个事件,所以无论快路径是" "怎么漏掉的(发布打断、worker OOM、Redis 抖动、刚配好的依赖)都会收敛。" "每跳只推进一层,单次上限 12 个。" ), "handler_key": "scheduler.chain_advance", "cron_expr": "*/5 * * * *", "timezone": "UTC", } #: ``(successor_name, predecessor_name, expected_current_cron)``. #: Order matters only for readability; each edge is independent. CHAIN_EDGES: tuple[tuple[str, str, str], ...] = ( # 06:30 → 07:00 UTC. The drift check reads the rollup this releases; # the rollup averages 107 s and has peaked at 296 s against a 30-min # gap, so today it works by a wide margin — until the gate makes it # wait. ( "ai_spending.cost_drift_check", "ai_spending.period_rollup_refresh", "0 7 * * *", ), # Hourly :00 → :10. "紧随扫描任务" in the handler's own description. ( "feed.notify_celebrants_and_leads", "feed.scan_today_celebrations", "10 * * * *", ), # 06:30 → 07:10 Asia/Shanghai. "排在 hr.leave_account_sync 之后运行", # and that sync averages 178 s. ( "hr.leave_insight_snapshot", "hr.leave_account_sync", "10 7 * * *", ), # 03:47 → 04:10 UTC. The successor's description already spells the # dependency out including the predecessor's clock time, which is # exactly the kind of comment this column exists to replace. ( "offboarding.account_assurance_rosters", "identity_sync.leaver_assurance", "10 4 * * *", ), # 04:00 → 04:45 → 05:00 UTC, a three-link chain. usage_sync's # description says the other two must "先跑完,所以调度顺序排在它后面". ( "bailian.api_key_sync", "bailian.workspace_sync", "45 4 * * *", ), ( "bailian.usage_sync", "bailian.api_key_sync", "0 5 * * *", ), ) def _table_exists(bind, table: str) -> bool: return bool( bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t LIMIT 1", ), {"t": table}, ).scalar(), ) def _column_exists(bind, table: str, column: str) -> bool: return bool( bind.execute( sa.text( "SELECT 1 FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " "AND COLUMN_NAME = :c LIMIT 1", ), {"t": table, "c": column}, ).scalar(), ) def _job_id(bind, name: str) -> str | None: return bind.execute( sa.text("SELECT id FROM scheduled_jobs WHERE name = :n LIMIT 1"), {"n": name}, ).scalar() def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "scheduled_jobs"): return if not _column_exists(bind, "scheduled_jobs", "chain_after_job_id"): # 615 did not run (fresh DB mid-chain, or a partial apply). Nothing # here is meaningful without the column; 615 will fail loudly on # its own rather than being papered over here. return if not _job_id(bind, CHAIN_ADVANCE_JOB["name"]): bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " " trigger_type, cron_expr, timezone, enabled, config, " " run_count, failure_count, created_at, updated_at) " "VALUES (:id, :n, :dn, :d, :hk, 'cron', :ce, :tz, 1, :cfg, " " 0, 0, NOW(), NOW())", ), { "id": str(uuid.uuid4()), "n": CHAIN_ADVANCE_JOB["name"], "dn": CHAIN_ADVANCE_JOB["display_name"], "d": CHAIN_ADVANCE_JOB["description"], "hk": CHAIN_ADVANCE_JOB["handler_key"], "ce": CHAIN_ADVANCE_JOB["cron_expr"], "tz": CHAIN_ADVANCE_JOB["timezone"], "cfg": json.dumps({}), }, ) for successor, predecessor, expected_cron in CHAIN_EDGES: parent_id = _job_id(bind, predecessor) if parent_id is None: # The predecessor is not seeded in this deployment, so the # ordering it was supposed to guarantee does not exist here # either. Leaving the successor on its cron is correct. continue bind.execute( sa.text( "UPDATE scheduled_jobs " "SET chain_after_job_id = :parent, trigger_type = 'chained', " " cron_expr = NULL, interval_seconds = NULL, " " updated_at = NOW() " "WHERE name = :n AND cron_expr = :expected " " AND chain_after_job_id IS NULL", ), {"parent": parent_id, "n": successor, "expected": expected_cron}, ) def downgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "scheduled_jobs"): return if _column_exists(bind, "scheduled_jobs", "chain_after_job_id"): for successor, predecessor, expected_cron in CHAIN_EDGES: parent_id = _job_id(bind, predecessor) if parent_id is None: continue # Mirror-image guard: restore only the rows this migration # actually converted, so a chain an operator built by hand # after the upgrade survives the downgrade. bind.execute( sa.text( "UPDATE scheduled_jobs " "SET chain_after_job_id = NULL, trigger_type = 'cron', " " cron_expr = :expected, updated_at = NOW() " "WHERE name = :n AND trigger_type = 'chained' " " AND chain_after_job_id = :parent", ), { "n": successor, "expected": expected_cron, "parent": parent_id, }, ) bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": CHAIN_ADVANCE_JOB["name"]}, )