"""The control loop. Every other service here is edge-triggered: a tenant POSTs, a row appears, a worker claims it. That is correct only as long as nothing is missed, and things are missed — a worker SIGKILLed holding a lease, an operator running `helm uninstall` by hand, a pod dying between the CAS and the enqueue. Nothing sends an event for any of it, because the thing that would have sent it is the thing that died. So: level-triggered. Every 60 seconds, compare the world to the database and enqueue what is missing. The four checks below do not know what went wrong, or whether anything did; they are the same code on the happy path and after an outage. That is why each is written as a query for work rather than a reaction to an event. Three rules hold the design together: * **Singleton.** `replicas: 1`, `strategy: Recreate`. Two reconcilers double-enqueue drift and race on TTL. No leader election on purpose — the right lease for that lives in Postgres next to the data, and until there is a second replica to elect between, an election is a subsystem that can only fail. The `SvcforgeReconcilerStale` alert notices when the one pod is gone. * **Each check is independent.** A helm binary that cannot reach the API server must not stop TTLs from expiring. * **Enqueue, never act.** The reconciler diagnoses and workers treat: it writes task rows and instance states and never calls `helm install`. Reading is the exception, since seeing reality is the job. """ from __future__ import annotations import asyncio from collections.abc import Awaitable, Callable from dataclasses import dataclass import typer from svcforge_core.adapters.clock import Clock, SystemClock from svcforge_core.adapters.helm import HelmProvisioner, Provisioner from svcforge_core.adapters.notify import LogNotifier, Notifier from svcforge_core.domain.catalog import load_catalog from svcforge_core.domain.models import CatalogEntry from svcforge_core.domain.windows import BadWindow, parse_window, schedule_upgrade_at from svcforge_core.obs import ( INSTANCES, QUEUE_DEPTH, RECONCILER_LAST_TICK, get_logger, setup, start_metrics_server, tracer, ) from svcforge_core.repo.db import DictPool, make_pool from svcforge_core.repo.instances import InstanceRepo from svcforge_core.repo.reconcile import ReconcileRepo from svcforge_core.repo.tasks import TaskRepo from svcforge_core.runtime import install_stop_signals, sleep_or_stop from svcforge_core.settings import Settings, load_settings log = get_logger("svcforge.reconciler") @dataclass(frozen=True) class ReconcilerDeps: """Everything a check is allowed to touch. Built once in main(), passed down. Same shape as `WorkerDeps` and for the same reason: checks take `deps` instead of reaching for globals, so the integration tests run every check against a real Postgres and a `FakeProvisioner` with no cluster in sight. """ pool: DictPool instances: InstanceRepo tasks: TaskRepo reconcile: ReconcileRepo provisioner: Provisioner notifier: Notifier clock: Clock catalog: dict[str, CatalogEntry] settings: Settings # Whose instances go first in the day-2 work list. Eating your own dog food is enforced # by an `order by` rather than a policy document — see InstanceRepo.list_upgradable. own_team: str # A config value, not a scheduler. Leave it at 1 until 1 is too slow. max_in_flight: int # --- The four checks --------------------------------------------------------------------- async def check_drift(deps: ReconcilerDeps) -> None: """The live helm releases versus what the database believes. The only check that looks outside Postgres, and the only one that catches someone running `helm uninstall` by hand or a drained node whose release never came back — where the DB still says `ready` and still hands the tenant an endpoint resolving to nothing. Two directions, two different answers: * **Release gone, DB says `ready`** -> re-enqueue provision. Safe because provisioning is `helm upgrade --install` against a deterministic release name. * **Release exists, DB knows nothing** -> log at error and stop. **Never delete in v1.** "The DB knows nothing" is one query against one database, and the release might belong to another team, another tool, or a half-finished migration. A human decides. """ with tracer().start_as_current_span("helm.list"): releases = await deps.provisioner.list_releases() live = {(r.name, r.namespace) for r in releases} for inst in await deps.reconcile.ready_instances(): if (inst.release_name, inst.namespace) in live: continue reason = f"drift: helm release {inst.release_name} missing from namespace {inst.namespace}" task_id = await deps.reconcile.enqueue_reprovision(inst.id, reason) if task_id is None: continue # already being dealt with, or the row moved under us log.warning( "drift.release_missing", instance_id=str(inst.id), team=inst.team, release=inst.release_name, namespace=inst.namespace, task_id=task_id, ) try: await deps.notifier.send( "drift.release_missing", f"re-provisioning {inst.id}: release {inst.release_name} vanished", {"instance_id": str(inst.id), "team": inst.team}, ) except Exception: # The task is committed; the notification is a courtesy. A webhook timing out # must not abandon the rest of the sweep — the instances after this one have the # same problem and nobody else is coming to find them. log.exception("notify.failed", instance_id=str(inst.id)) known = await deps.reconcile.known_releases() for name, namespace in sorted(live - known): # error, not warning: a resource nobody owns and nobody is billing for. It repeats # every 60s until a human deletes or adopts it, which is the intended pressure. log.error("drift.orphan_release", release=name, namespace=namespace, action="none (v1 never deletes)") async def check_lease_expiry(deps: ReconcilerDeps) -> None: """Tasks whose worker died -> back to `queued`. A lease, not a lock: no lock survives a power cut. A worker SIGKILLed mid-provision leaves `state='running'` with `locked_by` set and nobody running it, and cleanup code in the worker cannot help because the worker is what died. `locked_at` plus a timeout is the only thing that recovers the row. The 5-minute default must exceed the longest a healthy task can hold a lease, or a still-running provision is handed to a second worker. Handlers are idempotent so that is survivable, but it costs a duplicated helm run — hence `lease_seconds` > helm's `--timeout`. """ freed = await deps.tasks.reset_expired_leases(deps.settings.lease_seconds) if freed: log.warning("lease.expired", tasks_freed=freed, lease_seconds=deps.settings.lease_seconds) async def check_ttl(deps: ReconcilerDeps) -> None: """Expired instances -> `deleting`, plus a deprovision task. What stops a demo cluster becoming a permanent cloud bill, and the sweep the API's DELETE route depends on: DELETE CASes and enqueues in two statements, and a crash between them lands here on the next tick. Idempotent by construction — the work list excludes anything with a queued or running deprovision, and the CAS and insert share one transaction. Without that, a deprovision taking longer than 60 seconds collects a new task every tick. """ for inst in await deps.reconcile.due_for_deprovision(): task_id = await deps.reconcile.enqueue_deprovision(inst.id) if task_id is None: continue log.info( "ttl.expired", instance_id=str(inst.id), team=inst.team, expires_at=inst.expires_at.isoformat() if inst.expires_at else None, previous_state=inst.state.value, task_id=task_id, ) async def check_version_drift(deps: ReconcilerDeps) -> None: """The day-2 rollout: the work-list query, one service type at a time. Everything that makes it safe is somewhere else, which is the point: * `list_upgradable` limits to `max_in_flight` and returns nothing while `rollout_state='halted'`, so a bad chart stops after one tenant. * `schedule_upgrade_at` turns the maintenance window into a `run_after` and the queue does the waiting, in `where run_after <= now()`. No scheduler here, and there must not be one: a task parked in Postgres until 03:00 Sunday survives a restart, a timer does not. * `security: true` in the catalog bypasses the window. A bad window spec is one instance's problem: log it and move on. Failing the check would let one tenant's typo freeze everyone's security rollout. """ now = deps.clock.now() for service_type, entry in deps.catalog.items(): candidates = await deps.instances.list_upgradable( service_type=service_type, catalog_version=entry.chart_version, own_team=deps.own_team, max_in_flight=deps.max_in_flight, ) for candidate in candidates: inst = candidate.instance try: window = parse_window(candidate.maintenance_window) except BadWindow: log.exception( "upgrade.bad_window", instance_id=str(inst.id), team=inst.team, maintenance_window=candidate.maintenance_window, ) continue run_after = schedule_upgrade_at(window, security=entry.security, now=now) task_id = await deps.reconcile.enqueue_upgrade(inst.id, run_after) if task_id is None: continue # already queued or running; this is the max_in_flight guard log.info( "upgrade.scheduled", instance_id=str(inst.id), team=inst.team, service_type=service_type, from_version=inst.chart_version, to_version=entry.chart_version, run_after=run_after.isoformat(), security=entry.security, task_id=task_id, ) CHECKS: dict[str, Callable[[ReconcilerDeps], Awaitable[None]]] = { "drift": check_drift, "lease_expiry": check_lease_expiry, "ttl": check_ttl, "version_drift": check_version_drift, } # --- The tick ---------------------------------------------------------------------------- async def tick(deps: ReconcilerDeps) -> None: """One pass: all four checks, then the gauges, then the heartbeat. Checks first, gauges second, so `svcforge_queue_depth` reports what this tick left behind rather than what preceded its own work. The heartbeat is set unconditionally. It answers "is the loop running", not "is everything fine" — the checks have their own alerts. Gating it on success would make `SvcforgeReconcilerStale` fire for a helm blip and mean two things at once, and an alert that means two things gets muted. The whole tick runs in one span, a considered exception to "manual spans wrap helm calls only". That rule keeps the API from hand-rolling spans `opentelemetry-instrument` already makes; nothing auto-instruments the reconciler, so without this it emits no traces at all and — since `inject_traceparent` serialises the *active* context — every task it enqueues would carry a null `traceparent` and be unjoinable to the tick that created it. """ with tracer().start_as_current_span("reconciler.tick"): await _run_checks(deps) RECONCILER_LAST_TICK.set(deps.clock.now().timestamp()) log.info("tick.done") async def _run_checks(deps: ReconcilerDeps) -> None: """The four checks and the gauges. Split out so `tick` reads as span + heartbeat.""" for name, check in CHECKS.items(): try: await check(deps) except Exception: # the tick is the error boundary # The swallow is the design. The four checks share nothing but a database # handle, and a level-triggered loop is only worth having if it keeps running: # an unreachable cluster must not stop TTLs expiring, and one tenant's broken # window spec must not stop drift detection. This means "this check achieved # nothing for 60 seconds", never "the reconciler stops". log.exception("check.failed", check=name) try: QUEUE_DEPTH.set(await deps.reconcile.queue_depth()) counts = await deps.reconcile.instance_counts() for state, n in counts.items(): INSTANCES.labels(state=state).set(n) except Exception: # gauges are diagnostics; a failed read is not a failed tick log.exception("gauges.failed") async def run_reconciler(deps: ReconcilerDeps, stop: asyncio.Event) -> None: """Tick, sleep, repeat, until told to stop. Tick first, then sleep: a just-restarted pod should reconcile now, not in sixty seconds. Fixed interval rather than fixed period, so a tick that overruns delays the next one instead of stacking a second on top — which for a singleton is exactly the concurrent reconciler `replicas: 1` exists to prevent. """ while not stop.is_set(): await tick(deps) await sleep_or_stop(stop, deps.settings.reconcile_interval_s) def build_deps( pool: DictPool, settings: Settings, own_team: str, max_in_flight: int, ) -> ReconcilerDeps: """Wire the real collaborators. The only place that names concrete classes.""" return ReconcilerDeps( pool=pool, instances=InstanceRepo(pool), tasks=TaskRepo(pool), reconcile=ReconcileRepo(pool), provisioner=HelmProvisioner(helm_bin=settings.helm_bin, timeout_s=int(settings.helm_timeout_s)), notifier=LogNotifier(), clock=SystemClock(), catalog=load_catalog(settings.catalog_path), settings=settings, own_team=own_team, max_in_flight=max_in_flight, ) async def _amain(once: bool, own_team: str, max_in_flight: int) -> None: settings = load_settings() setup("svcforge-reconciler", settings) pool = make_pool(settings.runtime_dsn, settings.pool_min_size, settings.pool_max_size) await pool.open(wait=True) deps = build_deps(pool, settings, own_team, max_in_flight) try: if once: # One pass and exit: the acceptance path, and how to drive a reconcile by hand. # No metrics server — nothing would ever scrape it. await tick(deps) return # settings.metrics_port, like the worker. SVCFORGE_METRICS_PORT overrides it through # pydantic rather than a second CLI option, so the port has one definition. start_metrics_server(settings.metrics_port) stop = asyncio.Event() install_stop_signals(stop) await run_reconciler(deps, stop) finally: await pool.close() app = typer.Typer(add_completion=False, help="svcforge reconciler: the control loop.") @app.command() def main( once: bool = typer.Option(False, "--once", help="Run one tick and exit."), own_team: str = typer.Option( "platform", envvar="SVCFORGE_OWN_TEAM", help="Team whose instances upgrade first." ), max_in_flight: int = typer.Option( 1, envvar="SVCFORGE_MAX_IN_FLIGHT", min=1, help="Concurrent upgrades across the fleet." ), ) -> None: """Run the reconciler.""" # One asyncio.run, at the top, never nested. Everything below it is already async. asyncio.run(_amain(once, own_team, max_in_flight)) if __name__ == "__main__": app()