svcforge/services/reconciler/main.py

"""The control loop.

Every other service in svcforge is edge-triggered: a tenant POSTs, a row appears, a worker
claims it. Edge-triggered systems are correct exactly as long as nothing is ever missed —
and things are missed. A worker is SIGKILLed holding a lease. An operator runs
`helm uninstall` by hand. A pod dies between the CAS and the enqueue. Nobody sends an event
for any of that, 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 or care what went wrong, or whether anything
did; they are the same code on the happy path and after an outage. That property is the
entire reason this service exists, and it is why each check is written as a *query for
work*, never as a reaction to an event.

Three rules hold the design together:

* **Singleton.** `replicas: 1`, `strategy: Recreate` in the chart. Two reconcilers
  double-enqueue drift and race on TTL. There is no leader election here on purpose: the
  correct lease for that lives in Postgres next to the data, not in a Redis lock, and
  until there is a second replica to elect between, an election is a subsystem that can
  only fail. One pod, and the `SvcforgeReconcilerStale` alert is what notices it is gone.
* **Each check is independent.** One failing check must not skip the other three. A helm
  binary that cannot reach the API server must not stop TTLs from expiring.
* **Enqueue, never act.** The reconciler diagnoses; workers treat. It writes task rows and
  instance states, and never calls `helm install`. The one exception is reading — the drift
  check lists the live releases, because seeing reality is the job.
"""

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass

import typer

from services._runtime import install_stop_signals, sleep_or_stop
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.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` for the same reason: the checks take `deps` instead of
    reaching for globals, so the integration tests below run every check against a real
    Postgres and a `FakeProvisioner` without a cluster anywhere 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.

    This is the only check that looks outside Postgres, and the only one that can catch the
    failure nothing else can: someone ran `helm uninstall` by hand, or a node was drained
    and the release never came back. The DB still says `ready` and still hands the tenant an
    endpoint that resolves to nothing.

    Two directions, two very different answers:

    * **Release gone, DB says `ready`** -> re-enqueue provision. Safe, because provisioning
      is `helm upgrade --install` against a deterministic release name: converging on
      desired state, not a blind re-install.
    * **Release exists, DB knows nothing** -> log at error with release and namespace, and
      stop. **Never delete in v1.** The reconciler's view of "the DB knows nothing" is one
      query against one database; the release might belong to another team, another tool,
      or a migration half-finished. Deleting on that evidence is how an automated system
      takes down production faster than any human could. A human reads the log and 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 already committed; the notification is a courtesy. A webhook
            # timing out must not abandon the rest of the sweep — the instances after this
            # one in the loop 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: this is a resource nobody is billing for and nobody owns.
        # It will sit here every 60s until a human deletes it or adopts it. That 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. No lock survives a power cut: a worker SIGKILLed mid-provision
    leaves `state='running'` with `locked_by` set and nobody running it, and no amount of
    cleanup code in the worker helps, because the worker is the part that died. `locked_at`
    plus a timeout is the only thing that recovers the row, which is why `locked_at` exists.

    The 5-minute default must exceed the longest a healthy task can hold a lease, or the
    reconciler hands a still-running provision to a second worker. Handlers are idempotent,
    so that is survivable, though it still costs a duplicated helm run — which is why
    `lease_seconds` sits above 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.

    The line item that stops a demo cluster from becoming a permanent cloud bill. Also the
    sweep the API's DELETE route depends on: it CASes to `deleting` and enqueues in two
    statements, and a crash in between lands here on the next tick.

    Idempotent by construction — the work list excludes anything that already has a queued
    or running deprovision, and the CAS and the insert share one transaction. Without that
    guard, a deprovision that takes longer than 60 seconds gets a second task on the next
    tick, and a third on the tick after.
    """
    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 this 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 tenant's maintenance window into a `run_after`; the
      queue does the waiting, in `where run_after <= now()`. There is no scheduler here and
      there must not be one — a task parked in Postgres until 03:00 Sunday survives a
      reconciler restart, and an in-memory timer does not.
    * `security: true` in the catalog bypasses the window. A CVE with a public exploit does
      not wait until Sunday.

    A bad window spec is this instance's problem, not the fleet's: log it and move to the
    next one. Failing the whole 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: `svcforge_queue_depth` is read straight after the checks
    that add to the queue, so the value scraped is the value the tick left behind rather
    than one from before its own work.

    The heartbeat is set unconditionally, and that is deliberate. 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 inside one span, which is a considered exception to "manual spans go
    around helm calls only". That rule exists so the API does not hand-roll spans that
    `opentelemetry-instrument` already creates for it. Nothing auto-instruments the
    reconciler: without a span here it emits no traces at all, and — because
    `inject_traceparent` serialises the *active* context — every task it enqueues would be
    written with a null `traceparent` and be unjoinable to the tick that decided to create
    it. One span per tick is what makes "why was this instance re-provisioned at 03:00?" a
    question the traces can answer.
    """
    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. These four checks share nothing but a database
            # handle, and the value of a level-triggered loop is that it keeps running: an
            # unreachable cluster must not stop TTLs from expiring, and one tenant's broken
            # window spec must not stop drift detection. This means "this check achieved
            # nothing for 60 seconds", which the log says out loud. It never means "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 pod that has just been restarted should reconcile now, not in
    sixty seconds. Fixed interval rather than a fixed period — a tick that overruns simply
    delays the next one, instead of stacking a second tick on top of the first, which for a
    singleton would be 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 you drive a reconcile by hand
            # from a shell. No metrics server — nothing would ever scrape it.
            await tick(deps)
            return

        # settings.metrics_port, like the worker. SVCFORGE_METRICS_PORT still 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()