"""Master dispatcher handler for the Identity Sync Center. Registered as ``identity_sync.dispatch`` with the scheduled-jobs framework. Runs every 60 seconds (interval, not cron — cron lives on the per-target rows). Each tick: 1. Loads all targets whose ``next_run_at`` is in the past. 2. For each, acquires a per-target distributed lock. 3. Runs the target end-to-end via :meth:`IdentitySyncService.run_target`. 4. The service updates the target's ``last_run_*`` and ``next_run_at`` columns as part of the run finalisation. A target whose run takes longer than 60 s holds its per-target lock, so the next tick will see that lock as busy and skip — no double runs. The outer ``identity_sync.dispatch`` ScheduledJob also has a single-flight lock from the framework (``sched_job:identity_sync.dispatch``) which prevents two minute-ticks from overlapping if a slow tick takes longer than 60s. """ from __future__ import annotations import logging from datetime import datetime, timedelta from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) def _expected_cron_period( schedule_config: dict[str, Any], *, now: datetime, ) -> timedelta | None: """Expected gap between scheduled fires AROUND ``now``: (next fire − previous fire). The original watchdog computed the period as the gap between the next TWO fires from now — which misfires on non-uniform crons: a weekday-only ``0 9 * * 1-5`` checked on Saturday returns Mon→Tue = 1 day, so the perfectly-on-time Friday-morning sweep was flagged stale over the weekend. The prev↔next straddle measures the gap the CURRENT moment actually sits in (Fri→Mon = 3 days), so weekend checks stay green and Monday's check (sitting in the Mon→Tue gap once Monday fired… or the Fri→Mon gap before 9am) is exact. APScheduler's ``CronTrigger`` exposes no ``get_prev``; the previous fire is recovered by bisecting ``_compute_next_run`` (fire times are monotone in ``after``) over a bounded 62-day window. ~25 trigger evaluations per target on a 6-hourly job — negligible. A cron so sparse that no fire lands inside the window falls back to the next-two-fires gap (the old shape) rather than scanning years back. """ from app.domains.identity_sync.service import _compute_next_run n1 = _compute_next_run(schedule_config, after=now) if n1 is None: return None def _fallback() -> timedelta | None: n2 = _compute_next_run(schedule_config, after=n1) return (n2 - n1) if n2 is not None else None lo = now - timedelta(days=62) first = _compute_next_run(schedule_config, after=lo) if first is None or first > now: return _fallback() hi = now # Invariant: next_run(after=lo) <= now < next_run(after=hi). while (hi - lo) > timedelta(seconds=1): mid = lo + (hi - lo) / 2 nf = _compute_next_run(schedule_config, after=mid) if nf is not None and nf <= now: lo = mid else: hi = mid prev = _compute_next_run(schedule_config, after=lo) if prev is None or prev > now: return _fallback() return n1 - prev @register_scheduled_job( key="identity_sync.dispatch", display_name="数据同步中心 · 主调度", description=( "每 60 秒扫描启用中的同步目标,对到点(next_run_at <= now)" "的目标触发一次完整同步。每个目标 cron 表达式存在 " "identity_sync_targets.schedule_config.cron_expr 里;手动 / " "事件触发的同步不通过本任务,直接走 IdentitySyncService。" ), ) async def run_identity_sync_dispatch( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.identity_sync.service import IdentitySyncService svc = IdentitySyncService(db) summary = await svc.run_due_targets() # raise-on-failure discipline (feedback_handler_status_must_raise): # the dispatcher isolates per-target failures so one bad target # doesn't poison the others, but ANY failed target must surface # through the job run + failure notifier — a huilianyi sweep # failing nightly on a revoked 权限密钥 is the leaver-lockout # backstop silently dying, not an "ok" tick. ``failed`` = raised # out of run_target; ``ran_failed`` = run finalised without raising # but left something broken behind — the target is down # (status='failed': breaker / brake / config / no vendor call got # through) OR the target is up and specific people did not sync # (status='partial'). Both statuses live in # ``service._RUN_STATUSES_WITH_FAILURES``; see ``run_due_targets`` # for why ``partial`` is included rather than left to the # per-target alert_notifier. failed = (summary.get("failed") or []) + (summary.get("ran_failed") or []) if failed: preview = " | ".join(str(x)[:200] for x in failed[:3]) raise RuntimeError( f"identity_sync.dispatch: {len(failed)} due target(s) " f"failed this tick; first errors: {preview}" ) return summary @register_scheduled_job( key="identity_sync.run_items_cleanup", display_name="数据同步台账清理", description=( "每天删除超过保留期(config.retention_days,默认 90 天)的 " "identity_sync_run_items 逐条同步台账。运行行与状态行不受影响 —— " "台账是域内唯一无界增长的表(每次 cron 全量同步都会追加非 skip 行)。" ), ) async def run_identity_sync_run_items_cleanup( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.identity_sync.service import IdentitySyncService try: retention_days = int(config.get("retention_days") or 90) except (TypeError, ValueError): retention_days = 90 deleted = await IdentitySyncService(db).cleanup_run_items( retention_days=retention_days, ) return {"deleted": deleted, "retention_days": retention_days} @register_scheduled_job( key="identity_sync.cleanup_stale_runs", display_name="数据同步中心 · 幽灵运行行清理", description=( "每 15 分钟将卡在 status='running' 且超过阈值(config.stale_minutes,默认 30)" "的 identity_sync_runs 收尾为 'cancelled'。worker 硬退出 / OOM 会留下幽灵运行" "行(分布式锁 TTL 过期、运行行不会),web 启动看门狗只在开机扫且有 30 分钟下限," "本任务补运行期盲区。cancelled 是基础设施事件:不计入失败率、不触发失败告警;" "目标 next_run_at 不动(硬退出未推进游标),下一调度 tick 自然补跑。" ), ) async def run_identity_sync_cleanup_stale_runs( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.identity_sync.service import IdentitySyncService try: stale_minutes = int(config.get("stale_minutes") or 30) except (TypeError, ValueError): stale_minutes = 30 return await IdentitySyncService(db).cleanup_stale_runs( stale_minutes=stale_minutes, ) @register_scheduled_job( key="identity_sync.account_backstop", display_name="数据同步中心 · 离职账号兜底收敛", description=( "每日扫描已离职者(employment_status = left,已过本地末个工作日)在厂商侧仍" "活跃的账号,经账号管家 coordinator 幂等冻结(飞书 is_frozen=true)。" "入离职事件路为主,这是**兜底安全网** —— 骑既有 feishu_contact " "target 的启用状态(on_offboard≠keep),无独立开关。单次批量上限 " "config.batch_limit(默认 200)防误扫风暴;冻结失败会 raise 触达" "失败通知器。正常每日候选通常为 0(事件路已处理),兜底仅补漏。" ), ) async def run_account_backstop( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Freeze leavers whose Feishu account the event path missed. Idempotent (coordinator/runner returns ``deactivated`` on an already-frozen / absent account). Marks the identity-sync state row terminal on success so a frozen leaver isn't rescanned. Per-person commit keeps a mid-batch failure from losing prior progress. """ from sqlalchemy import select from app.core.coordinators import account_lifecycle_coordinator from app.domains.identity_sync.models import ( IdentitySyncState, IdentitySyncTarget, ) from app.identity.person_models import IdentityPerson # Gate on ``employment_status == 'left'`` ONLY — never 'leaver_pending'. # 'left' is set exclusively by ``HrService.transition_overdue_leavers`` # AFTER (a) the leaver's *local-timezone* last working day has rolled past # ``termination_date`` and (b) their Leaver-Blocked process cleared. By # contrast 'leaver_pending' is assigned at HRIS-confirm time for any # offboard whose endDate is < 30 days out — INCLUDING future-dated # leavers still serving their notice period. Freezing those here would # lock a still-active employee out of Feishu days before their real last # day — the exact class of bug the event path guards against via # ``_leaver_revoke_not_before`` (2026-05-28 Thalie Zhuang incident). # Gating on 'left' inherits both the local-TZ and Leaver-Blocked gates # transitively; a leaver_pending whose local date has genuinely passed is # flipped to 'left' by the hourly transition cron and caught next tick. terminal_statuses = ("deactivated", "deleted", "excluded") try: batch_limit = int(config.get("batch_limit") or 200) except (TypeError, ValueError): batch_limit = 200 # Rides the existing feishu_contact target's enablement — no separate # backstop switch (per the 2026-07-01 ADR). If it's off / keep, no-op. target = ( await db.execute( select(IdentitySyncTarget).where( IdentitySyncTarget.kind == "feishu_contact", IdentitySyncTarget.enabled.is_(True), IdentitySyncTarget.on_offboard != "keep", ) ) ).scalars().first() if target is None: return { "status": "skipped", "reason": "feishu_contact target not enabled / on_offboard=keep", "scanned": 0, "deactivated": 0, } rows = ( await db.execute( select(IdentityPerson, IdentitySyncState) .join( IdentitySyncState, IdentitySyncState.person_id == IdentityPerson.id, ) .where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "left", IdentitySyncState.target_id == target.id, IdentitySyncState.external_id.is_not(None), IdentitySyncState.last_status.notin_(terminal_statuses), ) .limit(batch_limit + 1) ) ).all() total_due = len(rows) capped = total_due > batch_limit if capped: rows = rows[:batch_limit] logger.warning( "account_backstop: %d leavers with live Feishu accounts — " "capping at %d this run (rest next tick)", total_due, batch_limit, ) deactivated = 0 no_email = 0 errors: list[str] = [] for person, state in rows: if not person.work_email: # A leaver row with a Feishu sync-state but no work_email can't be # resolved to a Feishu user — surface it (don't silently skip a # would-be freeze), leave the state for a human to fix the email. no_email += 1 logger.warning( "account_backstop: leaver person_id=%s has a live Feishu " "state but no work_email — skipped, needs manual review", person.id, ) continue res = await account_lifecycle_coordinator.deactivate_account( db, person, trace_id=str(person.id), ) feishu = (res or {}).get("feishu") or {} status = feishu.get("status") if status == "deactivated": state.last_status = "deactivated" deactivated += 1 elif status == "skipped": pass # target vanished mid-run; leave state for next tick else: errors.append(f"{person.work_email}: {feishu.get('error') or status}") await db.commit() summary = { # true backlog count (not truncated) so monitoring sees a growing # backlog even when a tick caps; ``processed`` = this tick's slice. "scanned": total_due, "processed": len(rows), "deactivated": deactivated, "no_email_skipped": no_email, "capped": capped, "errors": errors, } # Surface BOTH hard freeze failures AND leavers the backstop cannot # resolve (live Feishu account but no work_email). The latter is a # security-backstop gap — a departed employee retaining an active # account — so it must alert via the cron-failure notifier, not sit # silently behind a green run. Per-person deactivations already # committed above, so raising here only flags the run for attention. if errors or no_email: parts: list[str] = [] if errors: parts.append( f"{len(errors)} Feishu freeze failure(s); first: {errors[0][:200]}" ) if no_email: parts.append( f"{no_email} leaver(s) with a live Feishu account but no " f"work_email — cannot resolve to freeze, needs manual review" ) raise RuntimeError("account_backstop: " + " | ".join(parts)) return summary @register_scheduled_job( key="identity_sync.vendor_snapshot_refresh", display_name="数据同步中心 · 厂商快照每日刷新", description=( "每日对所有启用且 on_offboard∈(deactivate,delete) 的同步目标执行厂商" "快照刷新(refresh_vendor_snapshot,对厂商纯只读的名册拉取),让离职回收" "零信任对账(leaver_assurance)的厂商侧信号有新鲜证据可用。此前快照只有" "手动刷新入口,一放就馊(2026-07-17:huilianyi 快照停在 6-16、feishu 停在 " "7-2、其余 6 个目标从未刷过)——过期快照在加新鲜度闸后等于厂商侧信号静默" "失效,本任务就是让闭环真正闭上的那半圈。排在 assurance(03:47 UTC)之前" "跑;不支持快照的 kind(如 netsuite)记 skipped 不算失败;单目标失败不" "阻断其余目标,收尾统一 raise 触达失败通知器。单目标的 1205/1213/断连" "走共享的 run_with_lock_retry(新 session + innodb_lock_wait_timeout=120 " "重试),因为本任务与 03:00 的坐席同步机群在 identity_persons 上抢同一批" "外键父行锁(2026-08-08 zoom 快照即因此 1205 失败)。" ), ) async def run_identity_sync_vendor_snapshot_refresh( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Daily read-only roster pull per revocation-enabled target so the leaver-assurance vendor-side signal always has fresh evidence. Without this, snapshots only move when an admin clicks refresh — and a snapshot that predates a departure can't testify about it (the assurance job's freshness guard rightly ignores it), so Signal B decays to silence. Per-target isolation: one vendor being down must not stop the others' refresh; failures aggregate into one raise at the end so the framework failure notifier still pages IT. Each target's refresh additionally rides out a transient lock-wait via the shared ``run_with_lock_retry`` primitive. The snapshot write is an N-row INSERT into ``identity_sync_vendor_records`` whose ``matched_person_id`` FK makes TiDB take a pessimistic lock on every matched ``identity_persons`` row — the exact rows the 03:00 seat-sync fleet (jira / feishu / zoom, 130–160 s transactions) is locking through ``software_seat_allocations.user_id``. On 2026-08-08 the zoom target (621 records, first in the loop) lost that race and 1205'd out at the default 50 s, dropping the whole day's zoom snapshot — a re-run on a fresh session with ``innodb_lock_wait_timeout=120`` is the canonical fix, and the refresh is idempotent (wipe-and-replace, read-only on the vendor side) so re-running it is always safe. """ from sqlalchemy import select from app.core.exceptions import AppException from app.domains.identity_sync.models import IdentitySyncTarget from app.domains.identity_sync.service import IdentitySyncService from app.scheduled_jobs.handlers import run_with_lock_retry targets = (await db.execute( select(IdentitySyncTarget).where( IdentitySyncTarget.deleted_at.is_(None), IdentitySyncTarget.enabled.is_(True), IdentitySyncTarget.on_offboard.in_(("deactivate", "delete")), ) )).scalars().all() # Materialize scalars up front: a mid-loop rollback expires loaded ORM # rows (rollback always expires, regardless of expire_on_commit), and a # lazy re-load on ``target.slug`` inside an except block would blow up # the whole run for one bad target. target_rows = [(t.id, t.slug, t.kind) for t in targets] refreshed: list[str] = [] skipped: list[str] = [] errors: list[str] = [] for target_id, slug, kind in target_rows: # ``_tid`` is bound as a default so the closure captures THIS # iteration's target rather than the loop variable's final value. async def _refresh(session: AsyncSession, _tid=target_id): # Bind a service to whichever session the retry helper hands # us: attempt 1 is the framework session (identical to a # direct call), attempts 2+ are fresh sessions carrying a # 120 s lock-wait timeout. return await IdentitySyncService(session).refresh_vendor_snapshot( target_id=_tid, actor_id=None, ) try: result = await run_with_lock_retry( db, _refresh, label=f"identity_sync.vendor_snapshot_refresh[{slug}]", ) refreshed.append( f"{slug}: {result.get('counts')} " f"(matched={result.get('matched_users')})" ) except AppException as e: await db.rollback() if e.code == "IDENTITY_SYNC_SNAPSHOT_UNSUPPORTED": skipped.append(f"{slug}: kind={kind}") continue errors.append(f"{slug}: {str(e)[:200]}") except Exception as e: # noqa: BLE001 — per-target isolation await db.rollback() errors.append(f"{slug}: {str(e)[:200]}") summary = { "refreshed": refreshed, "skipped_unsupported": skipped, "errors": errors, } if errors: raise RuntimeError( "identity_sync.vendor_snapshot_refresh: " f"{len(errors)} target(s) failed " f"({len(refreshed)} refreshed ok): " + " | ".join(errors[:5]) ) return summary @register_scheduled_job( key="identity_sync.leaver_assurance", display_name="数据同步中心 · 离职回收零信任对账", description=( "每日反向核验:对每个启用且 on_offboard∈(deactivate,delete) 的同步目标," "扫描已离职者(employment_status='left')在该目标上是否仍**未确认回收**——" "两路信号:①Helix 侧 —— 该离职者有 external_id 但 last_status 不在 " "(deactivated/deleted/excluded);②厂商侧 —— vendor 快照 " "IdentitySyncVendorRecord 匹配到该离职者且 is_active=True,**且快照拍摄日" "严格晚于离职日**(拍在离职前的快照本来就该是 active,不能作为「未回收」的" "证词——2026-07-17 首跑 48 条里 36 条即此类过期快照误报)。二者任一命中即" "视为「离职者账号可能仍活跃」。external_account_registry 里有已知分类" "(shared_service/external_guest/contractor,如 CBG 保留)的离职者豁免——" "与 offboarding 写路径的保留语义一致,登记在案的保留不是回收缺口。这是独立" "于写路径的闭环:写路径(sweep/事件/backstop)声称回收成功是一面之词,本对账" "反向核对真相,命中即 raise → 调度框架失败通知器触达 IT(**只推 IT,不碰" "员工/经理**)。纯 DB 查询、不调厂商;厂商侧信号依赖快照新鲜度(由 " "identity_sync.vendor_snapshot_refresh 每日先行刷新)。" ), ) async def run_identity_sync_leaver_assurance( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Reverse-verify that departed employees have no lingering active vendor account — the zero-trust closed loop over the write path. The write path (nightly sweep out-of-scope deactivate, offboard event push, Feishu account_backstop) ASSERTS it revoked a leaver; this job independently CHECKS it, per enabled deactivate/delete target, from two angles and pages IT on any gap. DB-only (no vendor calls); raises so the scheduled-jobs framework failure notifier reaches IT admins (never the leaver / their manager). Two suppressions keep the findings TRUE (a zero-trust alarm that cries wolf gets muted, which is worse than no alarm): * Deliberately-retained accounts — a leaver whose ``work_email`` carries a KNOWN classification in the external-account registry (shared_service / external_guest / contractor, e.g. CBG-retained Feishu members) is exempt from BOTH signals: the offboarding write path intentionally skips revoking these, so flagging them daily just re-litigates a documented retention decision. * Stale snapshots (vendor signal only) — a snapshot row fetched ON or BEFORE the person's termination date shows them active because they genuinely WERE active when it was taken; it cannot testify about post-departure revocation. Only snapshots strictly newer than the termination date count. (First run 2026-07-17: 36 of 48 findings were this artifact — huilianyi's snapshot predated a month of departures.) """ from sqlalchemy import func as sql_func from sqlalchemy import select # Cross-domain read of the retention registry mirrors the offboarding # write path (offboarding/service.py _is_*_account_retained) — the # assurance check must honor the same keep semantics it verifies. from app.domains.asset.models import ( EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS, ExternalAccountRegistry, ) from app.domains.identity_sync.models import ( IdentitySyncState, IdentitySyncTarget, ) from app.domains.identity_sync.service import IdentitySyncService from app.identity.person_models import IdentityPerson try: max_report = int(config.get("max_report") or 50) except (TypeError, ValueError): max_report = 50 terminal_statuses = ("deactivated", "deleted", "excluded") retained_emails = ( select(sql_func.lower(ExternalAccountRegistry.email)) .where( ExternalAccountRegistry.email.is_not(None), ExternalAccountRegistry.classification.in_( EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS ), ) .scalar_subquery() ) # work_email is NOT NULL in the schema and the subquery filters NULL # registry emails, so this NOT IN cannot be NULL-poisoned. not_retained = sql_func.lower(IdentityPerson.work_email).notin_( retained_emails ) targets = (await db.execute( select(IdentitySyncTarget).where( IdentitySyncTarget.deleted_at.is_(None), IdentitySyncTarget.enabled.is_(True), IdentitySyncTarget.on_offboard.in_(("deactivate", "delete")), ) )).scalars().all() findings: list[str] = [] affected_slugs: set[str] = set() checked_targets = 0 svc = IdentitySyncService(db) for target in targets: checked_targets += 1 # Signal A (Helix-side): a leaver Helix linked to this target but # never drove to a terminal (deactivated/deleted/excluded) state — # i.e. we hold an external_id for them but haven't confirmed revoke. helix_rows = (await db.execute( select( IdentityPerson.work_email, IdentitySyncState.last_status, IdentitySyncState.last_error, ) .join( IdentitySyncState, IdentitySyncState.person_id == IdentityPerson.id, ) .where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "left", not_retained, IdentitySyncState.target_id == target.id, IdentitySyncState.external_id.is_not(None), IdentitySyncState.last_status.notin_(terminal_statuses), ) .limit(max_report + 1) )).all() for email, last_status, last_error in helix_rows: affected_slugs.add(target.slug) # Carry the state row's ``last_error`` into the finding. This # string IS the alert body IT reads at 03:47, and without the # reason it reads identically for two very different worlds: # the vendor rejected this ONE person, versus this whole # target is unauthorized and failing every write. On # 2026-08-14 (aws-ai, revoked SCIM token) it was the latter, # and the page said only "leaver still linked" — the reason # (``PATCH /Users/… (deactivate) → 401:``) was sitting on the # row the query already reads. Quoting it (trimmed) makes the # page self-explaining rather than sending the on-call to # correlate a different job's alert an hour earlier. why = (last_error or "").strip().replace("\n", " ") suffix = f", last_error={why[:120]}" if why else "" findings.append( f"[{target.slug}] {email or '?'}: leaver still linked " f"(last_status={last_status}, not revoked{suffix})" ) # Signal B (vendor-side): the stored vendor snapshot shows an ACTIVE # record matched to a leaver — catches a recreated / federation- # reprovisioned account the write path can't see via its stale id, # a 2061 (不推送) leaver the sweep never touches, and an orphan whose # account predates Helix managing this target. Sourced from the # SHARED population helper so this alarm and the write-side # leaver_revocation_reconcile can never disagree about the set — # the reconcile provably drives these findings to zero (freshness # guard + registry exemption live inside the helper). vendor_pairs = await svc.active_leaver_vendor_records( target_id=target.id, limit=max_report + 1, ) for person, vrec in vendor_pairs: affected_slugs.add(target.slug) findings.append( f"[{target.slug}] {person.work_email or '?'}: vendor snapshot " f"shows ACTIVE (key={vrec.external_key}) for a leaver" ) summary = { "checked_targets": checked_targets, "affected_targets": sorted(affected_slugs), "findings": findings[:max_report], "finding_count": len(findings), } if findings: preview = " | ".join(findings[:5]) raise RuntimeError( f"identity_sync.leaver_assurance: {len(findings)} leaver " f"account(s) not confirmed revoked on " f"{len(affected_slugs)} of {checked_targets} checked " f"target(s): {preview}" ) return summary @register_scheduled_job( key="identity_sync.leaver_revocation_reconcile", display_name="数据同步中心 · 离职回收主动对账收敛", description=( "每日主动收敛:对每个启用且 on_offboard∈(deactivate,delete) 的同步目标," "把厂商快照里仍活跃、且匹配到已离职者(employment_status='left',非 registry " "保留分类,快照拍摄日严格晚于离职日)的账号,按厂商自己的 external_key 直接" "在厂商侧停用/删除——**不经 _load_candidates/sweep**,因此能收掉写路径结构性" "收不到的两类:①SF customString24=2061(不推送)离职者(sweep 把他们排除在" "候选之外,永不回收);②孤儿账号(账号早于 Helix 接管该目标就存在,离职时没有" "任何事件下发到此目标,如 2026-07-09 接入 alicloud-sso 前离职的 94 人)。回收" "成功后写终态 state 行并把该厂商记录 is_active 置否,让当晚 leaver_assurance" "(03:47 UTC)看到已收敛而非重复告警。这是只读 leaver_assurance 的**写侧孪生**" "(共用同一 Signal-B 人群,证告警必被本任务清零)。排在快照刷新(03:00)之后、" "assurance(03:47)之前跑;单目标失败不阻断其余,收尾统一 raise 触达失败通知器" "(**只推 IT,不碰员工/经理**)。" ), ) async def run_identity_sync_leaver_revocation_reconcile( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Actively revoke leaver accounts the vendor snapshot still shows active — the WRITE sibling of the read-only ``leaver_assurance``. For each enabled deactivate/delete target, revoke (by the vendor's own ``external_key``) every account matched to a not-retained ``left`` person whose snapshot post-dates their departure — the SAME Signal-B set assurance alarms on, so this provably drives assurance to zero. It bypasses ``_load_candidates``, so it reaches the two gaps the sweep structurally cannot: 2061 (不推送) leavers and pre-onboarding orphans. Per-target isolation: one vendor being down must not stop the others. The service method commits per row, so a target-level exception (e.g. a runner build failure) leaves already-revoked rows durable; the handler rolls back the tainted unit of work and re-fetches the next target fresh (a rollback expires ORM rows). Failures aggregate into one raise so the framework failure notifier pages IT. Runs 03:25 UTC — after ``vendor_snapshot_refresh`` (03:00), before ``leaver_assurance`` (03:47). """ from sqlalchemy import select from app.domains.identity_sync.models import IdentitySyncTarget from app.domains.identity_sync.service import IdentitySyncService try: per_target_limit = int(config.get("limit") or 1000) except (TypeError, ValueError): per_target_limit = 1000 targets = (await db.execute( select(IdentitySyncTarget).where( IdentitySyncTarget.deleted_at.is_(None), IdentitySyncTarget.enabled.is_(True), IdentitySyncTarget.on_offboard.in_(("deactivate", "delete")), ) )).scalars().all() target_ids = [t.id for t in targets] svc = IdentitySyncService(db) revoked_total = 0 per_target: list[dict[str, Any]] = [] errors: list[str] = [] for target_id in target_ids: # Re-fetch each target fresh: a prior target's rollback expires # every loaded ORM row, so a held reference could lazy-reload # inside an except block and blow up the run for one bad target. target = await db.get(IdentitySyncTarget, target_id) if target is None: continue slug = target.slug try: result = await svc.revoke_leavers_on_target( target, limit=per_target_limit, ) revoked_total += result["revoked"] per_target.append({ "target": slug, "candidates": result["candidates"], "revoked": result["revoked"], "failed": result["failed"], "skipped": result["skipped"], }) if result["errors"]: errors.append( f"{slug}: {result['failed']} failed; " f"first: {result['errors'][0][:200]}" ) except Exception as e: # noqa: BLE001 — per-target isolation await db.rollback() logger.exception( "leaver_revocation_reconcile: target %s failed", slug, ) errors.append(f"{slug}: {str(e)[:200]}") summary = { "revoked": revoked_total, "per_target": [p for p in per_target if p["candidates"]], "errors": errors, } if errors: raise RuntimeError( "identity_sync.leaver_revocation_reconcile: " f"{len(errors)} target(s) had revocation failures " f"({revoked_total} revoked ok): " + " | ".join(errors[:5]) ) return summary @register_scheduled_job( key="identity_sync.staleness_watchdog", display_name="数据同步中心 · 沉默漏跑看门狗", description=( "每 6 小时检查所有启用的 cron 同步目标:若某目标距上一次**成功的定时" "全量 sweep**(trigger_source='scheduler' 且 ok/partial)已超过「一个 cron " "周期 + 宽限期(config.grace_hours,默认 6h)」,则 raise → 经调度框架失败" "通知器触达 IT。补的是既有失败通知器覆盖不到的盲区:那个通知器只在 run " "**跑了并失败**时报警,而定时 sweep 若被高频单人 run 顶掉 next_run_at 从而" "**根本没跑**(2026-07-14 张业祥飞书部门不同步事故),既无失败记录、也就" "无人被告警。target.last_run_at 会被每次单人 run 刷新、不能当「sweep 有没有" "跑」的信号——本任务专看 scheduler 触发的 run。\n" "告警区分两种形状(2026-09-06 起):**silent** = 压根没跑(本任务的本职," "别处无人报);**failing** = 跑了但失败(调度器 ran_failed 闸同时也会报," "目标的 runs 列表里有 error_message 可查)。此前两者共用「沉默漏跑」一种" "措辞,而 2026-08-20~23 的 netsuite 是天天跑、天天失败,措辞把排查引向了" "错误的方向。" ), ) async def run_identity_sync_staleness_watchdog( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Alert when an enabled cron sync target has gone longer than one cron period without a SUCCESSFUL scheduled full sweep. For each enabled ``cron`` target: compute the expected inter-fire period from its cron expr, then find the most recent ``scheduler``- triggered run that actually executed (``ok``/``partial``). If that run is older than ``period + grace_hours`` — or there is none and the target has existed longer than that threshold — the target is stale. Any stale target ⇒ ``raise`` so the scheduled-jobs framework's failure notifier pages IT admins (no dependency on the target's own ``alert_recipient_ids``, which may be unset). Staleness comes in two shapes and the alert names which one it is — ``result_summary`` carries them as ``silent`` / ``failing`` and their union as ``stale``: * **silent** — no ``scheduler`` run row at all since the last good sweep. This is the blind spot the job was built for (2026-07-14): the dispatcher never fired, so there is no failed run and no other alarm anywhere. ``target.last_run_at`` cannot detect it, because per-person runs refresh that column — which is why this job reads run rows. * **failing** — it fired on schedule and the sweep failed. The dispatcher's own ``ran_failed`` gate covers this too (once that gate was repaired on 2026-09-06 — it had been dead code, so for a while this watchdog really was the only alarm for the failing case). Until 2026-09-06 both shapes were reported with the *silent* wording, which is how prod's 2026-08-20..23 netsuite alert came to say a sweep had "silently stopped running" on a target that ran every night and failed every night. See the inline note on ``last_attempt`` below. """ from datetime import UTC from sqlalchemy import desc, select from app.domains.identity_sync.models import ( IdentitySyncRun, IdentitySyncTarget, ) try: grace_hours = float(config.get("grace_hours") or 6.0) except (TypeError, ValueError): grace_hours = 6.0 grace = timedelta(hours=grace_hours) now = datetime.now(UTC) targets = ( await db.execute( select(IdentitySyncTarget).where( IdentitySyncTarget.enabled.is_(True), IdentitySyncTarget.deleted_at.is_(None), ) ) ).scalars().all() checked = 0 # ``silent`` = the dispatcher never fired the sweep (this job's reason to # exist); ``failing`` = it fired and the sweep failed. ``stale`` stays the # union so existing readers of ``result_summary`` keep working. silent: list[str] = [] failing: list[str] = [] for t in targets: sc = t.schedule_config or {} if sc.get("trigger") != "cron" or not sc.get("cron_expr"): continue # Expected period between fires = (next fire − previous fire) # straddling now — NOT the gap between the next two slots, which # flagged a weekday-only cron's on-time Friday sweep as stale on # Saturday (upstream #1968 fixed the same bug by taking the MAX # gap over the next 8 fires; the straddle subsumes it — same # weekend behaviour, and midweek it detects a silently-missed # fire within one real period instead of one worst-case period). # See :func:`_expected_cron_period`. period = _expected_cron_period(sc, now=now) if period is None: continue threshold = period + grace checked += 1 # Last SCHEDULED run that actually executed. A cron lock-skip in # ``run_due_targets`` creates NO run row, so any scheduler row means # the sweep really ran; failed scheduler runs are covered by the # failure notifier, so here we look for the last ok/partial one. last = ( await db.execute( select(IdentitySyncRun) .where(IdentitySyncRun.target_id == t.id) .where(IdentitySyncRun.trigger_source == "scheduler") .where(IdentitySyncRun.status.in_(("ok", "partial"))) .where(IdentitySyncRun.finished_at.is_not(None)) .order_by(desc(IdentitySyncRun.started_at)) .limit(1) ) ).scalars().first() # ...and the last scheduled ATTEMPT of any outcome. The two together # separate the only two things staleness can mean, which the # ok/partial predicate alone conflates — and mis-describing which one # is happening is not cosmetic: on 2026-08-20..23 netsuite ran its # sweep every single night and failed it every single night, while # this job told IT the sweep had "silently stopped running". Two # weeks later that wording sent a reader to ``last_run_at`` (fresh, # because per-person runs move it), where the two facts read as a # contradiction and the watchdog looked like the broken thing. # * no attempt at all → SILENT: the dispatcher never fired it. # This is the blind spot the job exists for (2026-07-14, a # per-person run flood kept shoving ``next_run_at`` forward). # * attempts, none ok → FAILING: loud, diagnosable, and already # visible on the runs list with an error_message. # Both still raise — a target with no completed full sweep in over a # period is a real leaver-lockout gap either way, and after the # 2026-09-06 ``ran_failed`` fix the FAILING case double-covers with # the dispatcher rather than replacing it. last_attempt = ( await db.execute( select(IdentitySyncRun) .where(IdentitySyncRun.target_id == t.id) .where(IdentitySyncRun.trigger_source == "scheduler") .where(IdentitySyncRun.finished_at.is_not(None)) .order_by(desc(IdentitySyncRun.started_at)) .limit(1) ) ).scalars().first() def _utc(dt: datetime) -> datetime: # MySQL/TiDB DATETIME loses tzinfo on read — re-attach UTC. return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt def _age(dt: datetime) -> timedelta: return now - _utc(dt) def _attempted_since(dt: datetime | None) -> bool: """Did a scheduled run *execute* more recently than ``dt`` (or at all, when ``dt`` is None — never a successful sweep)? Both sides go through ``_utc``: comparing a naive against an aware datetime raises, and a TypeError here would take the whole watchdog down rather than mis-word one line. """ if last_attempt is None or last_attempt.started_at is None: return False if dt is None: return True return _utc(last_attempt.started_at) > _utc(dt) if last is not None and last.started_at is not None: if _age(last.started_at) > threshold: if _attempted_since(last.started_at): at = last_attempt.started_at # type: ignore[union-attr] failing.append( f"{t.slug}: scheduled sweep RAN BUT FAILED — last " f"attempt {at:%Y-%m-%d %H:%M}Z status=" f"{last_attempt.status!r}; last SUCCESSFUL sweep " # type: ignore[union-attr] f"{last.started_at:%Y-%m-%d %H:%M}Z " f"(> {threshold} ago; cron {sc.get('cron_expr')!r}). " f"See the target's runs list for the error" ) else: silent.append( f"{t.slug}: scheduled sweep DID NOT RUN since " f"{last.started_at:%Y-%m-%d %H:%M}Z " f"(> {threshold} ago; cron {sc.get('cron_expr')!r})" ) else: # Never ran a scheduled sweep. Key the "has it had a chance # to fire" clock on when cron mode was (most recently) # configured, NOT on created_at alone — flipping a # months-old manual target to cron would otherwise page # IMMEDIATELY, before its first fire ever arrives. # ``updated_at`` moves on the admin's mode flip (and on any # other edit — acceptable slack: worst case the alert waits # one extra period after an unrelated edit), so # max(created_at, updated_at) is the latest moment the # current schedule could have started counting from. anchors = [ d for d in (t.created_at, t.updated_at) if d is not None ] if anchors and min(_age(d) for d in anchors) > threshold: newest = min(anchors, key=_age) if _attempted_since(None): at = last_attempt.started_at # type: ignore[union-attr] failing.append( f"{t.slug}: no scheduled sweep has EVER succeeded — " f"last attempt {at:%Y-%m-%d %H:%M}Z status=" f"{last_attempt.status!r} " # type: ignore[union-attr] f"(cron config last touched {newest:%Y-%m-%d}; " f"cron {sc.get('cron_expr')!r}). " f"See the target's runs list for the error" ) else: silent.append( f"{t.slug}: no scheduled sweep ever " f"(cron config last touched {newest:%Y-%m-%d}; " f"cron {sc.get('cron_expr')!r})" ) stale = silent + failing summary = { "checked": checked, "stale": stale, "silent": silent, "failing": failing, "grace_hours": grace_hours, } if stale: # raise → scheduled-jobs framework failure notifier pages IT admins. # Stable message so the notifier's signature dedup suppresses repeats # of the same standing staleness until it resolves or the set changes. # # The headline names which of the two shapes is in play, because that # is the first thing the reader has to decide and the old wording # ("silently stopped running") answered it wrongly for the FAILING # case. Silent leads when both are present: it is the one nothing else # is reporting. if silent and failing: headline = ( f"{len(silent)} cron sync target(s) have silently stopped " f"running their scheduled full sweep, and {len(failing)} " f"have been running it and failing" ) elif silent: headline = ( f"{len(silent)} cron sync target(s) have silently stopped " f"running their scheduled full sweep" ) else: headline = ( f"{len(failing)} cron sync target(s) are running their " f"scheduled full sweep and failing it — no successful sweep " f"in over one cron period" ) preview = " | ".join(stale[:5]) raise RuntimeError( f"identity_sync.staleness_watchdog: {headline}: {preview}" ) return summary