"""Runtime-tunable sweep-guard tolerance for CMDB cloud sync. The CMDB sweep guard refuses to soft-delete rows when the cloud returned fewer rows than the previous successful sync's baseline by more than ``tolerance_percent``. Three call sites read this value: * :func:`app.domains.cmdb.sync._safe_sweep` — resource sweep (``cmdb.resource_sync`` cron + admin force-sweep endpoint). * :func:`app.domains.cmdb.lifecycle_service.sync_scopes_for_integration` — scope close-sweep. * :func:`app.domains.cmdb.lifecycle_service.sync_roles_for_integration` — role soft-delete sweep. Resolution order: 1. ``platform_settings['cmdb_sync.sweep_drop_tolerance_percent']`` — admin-editable JSON value (int 0-100). Edit via the Platform Settings admin page; effective on the next sweep call (the resolver reads the row fresh each invocation — no caching, no restart). This is the recommended ops surface. 2. ``settings.CMDB_SYNC_SWEEP_DROP_TOLERANCE_PERCENT`` — ``.env`` bootstrap fallback. Used when the platform_settings row is missing (fresh DB before migration 182 lands, or row deleted by accident) or malformed (non-int / out-of-band value). Both bounds ``[0, 100]``. The semantics is "% of drop we will tolerate before refusing to sweep": * 0 — STRICTEST: any drop trips (``upserted < baseline`` refuses). * 20 — default: absorbs drops up to 20%; >20% trips. * 100 — DISABLED (drop-ratio leg only — zero-rows guard companion remains unconditional): the floor collapses to 0 so no drop ratio ever trips. Useful only as a stop-gap during a known cloud-wide regression; never leave it at 100 long-term. Why a separate module: ``_safe_sweep`` and the lifecycle-service sweep paths live in different files but must read the SAME tolerance every tick (otherwise a partial change of one knob would create a confusing asymmetry where scope sweeps and resource sweeps disagree). Putting the resolver here keeps the canonical key + plausibility band in one place. """ from __future__ import annotations from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.integrations.models import PlatformSetting PLATFORM_SETTING_SWEEP_DROP_TOLERANCE = "cmdb_sync.sweep_drop_tolerance_percent" # Plausibility band — same as the pydantic-settings validator on the # .env fallback. A value outside this range is treated as malformed # and triggers fallback. _MIN_PERCENT = 0 _MAX_PERCENT = 100 def _coerce_int_percent(value: Any) -> int | None: """Best-effort coerce a JSON-stored value to an int in ``[0, 100]``. ``platform_settings.value`` is JSON; TiDB's driver could surface a number as int / float / Decimal, and a manually-edited row could carry a string (``"20"``). Anything we can't make sense of, or any value outside the plausibility band, returns ``None`` so the caller falls back to the settings default. ``bool`` is rejected explicitly (it's an ``int`` subclass — ``True`` would silently coerce to 1). """ if value is None or isinstance(value, bool): return None if isinstance(value, int): as_int = value elif isinstance(value, float): # Accept ``20.0`` (JSON serialisation can produce floats for # whole numbers depending on the driver); reject ``20.5`` since # the contract is a percentage as an integer. if not value.is_integer(): return None as_int = int(value) elif isinstance(value, str): try: as_int = int(value) except (TypeError, ValueError): return None else: return None if not (_MIN_PERCENT <= as_int <= _MAX_PERCENT): return None return as_int async def get_sweep_drop_tolerance_percent(db: AsyncSession) -> int: """Return the active sweep drop tolerance as an int percent. DB-first, settings fallback. Always returns a value in [0, 100] — the settings field carries the same validator, so a fallback hit cannot escape the band either. Callers pass the resolved percentage to :func:`drop_exceeds_tolerance`. The comparison uses exact integer cross multiplication, avoiding the downward-rounding hole that previously let small catalogs lose far more than the configured percentage. Resolving the percentage once and reusing it for both the decision and message also closes the TOCTOU window where an admin edit could make them disagree. """ row = await db.get(PlatformSetting, PLATFORM_SETTING_SWEEP_DROP_TOLERANCE) if row is not None: coerced = _coerce_int_percent(row.value) if coerced is not None: return coerced return settings.CMDB_SYNC_SWEEP_DROP_TOLERANCE_PERCENT def drop_exceeds_tolerance( *, baseline: int, observed: int, tolerance_percent: int, ) -> bool: """Return whether the observed drop is strictly over tolerance. Integer cross multiplication preserves the exact ratio. In particular, a 2 → 1 snapshot is a 50% drop and must not become acceptable merely because ``int(2 * 0.8)`` rounds the required count down to one. """ if baseline <= 0 or observed >= baseline: return False return observed * 100 < baseline * (100 - tolerance_percent) def minimum_observed_within_tolerance( *, baseline: int, tolerance_percent: int, ) -> int: """Smallest integer observation whose drop is within tolerance.""" numerator = baseline * (100 - tolerance_percent) return (numerator + 99) // 100 # ── Cloud role-catalog freshness ───────────────────────────────────── async def cloud_catalog_health( session, integration_id: str, ) -> dict[str, object]: """Is this integration's role catalog still being refreshed? The role catalog is what the employee's 「申请权限」 picker and the admin's promote/demote surface both read. It is refreshed only by ``cmdb.roles_sync``, and a stale catalog fails silently in the worst possible direction: a role that exists on the cloud is simply absent from the dropdown, so the request cannot be made and nobody learns why. Prod 2026-08-12: ``cmdb.roles.tencent-primary`` failed once on 2026-05-19 and was left **disabled**. The catalog sat frozen at that day's snapshot for nearly three months. Re-running the sync by hand upserted 1031 roles and retired 15 — i.e. nothing was broken, the job had just stopped and no surface said so. A disabled cron is indistinguishable from a deliberate configuration unless something reads it back, which is what this does. Returns ``last_synced_at`` / ``stale_days`` from the catalog rows themselves plus the cron's own state, so a caller can tell "the job is off" from "the job runs but errors" from "healthy". """ from datetime import UTC, datetime from sqlalchemy import func, select from app.domains.cmdb.models import CmdbCloudRole from app.scheduled_jobs.models import ScheduledJob last_synced = (await session.execute( select(func.max(CmdbCloudRole.last_synced_at)).where( CmdbCloudRole.integration_id == str(integration_id), CmdbCloudRole.deleted_at.is_(None), ) )).scalar() jobs = (await session.execute( select(ScheduledJob).where( ScheduledJob.handler_key == "cmdb.roles_sync", ) )).scalars().all() job = next( ( j for j in jobs if str((j.config or {}).get("integration_id") or "") == str(integration_id) ), None, ) stale_days: int | None = None if last_synced is not None: now = datetime.now(UTC).replace(tzinfo=None) base = ( last_synced.replace(tzinfo=None) if last_synced.tzinfo else last_synced ) stale_days = max(0, (now - base).days) return { "last_synced_at": ( last_synced.isoformat() if last_synced is not None else None ), "stale_days": stale_days, "cron_name": job.name if job is not None else None, # ``False`` here is the case that went unnoticed for three # months; ``None`` means no roles cron exists for this # integration at all, which is equally worth showing. "cron_enabled": bool(job.enabled) if job is not None else None, "cron_status": job.last_status if job is not None else None, "cron_last_run_at": ( job.last_run_at.isoformat() if job is not None and job.last_run_at is not None else None ), "cron_last_error": ( (job.last_error or "")[:500] if job is not None else None ), }