svcforge/services/reconciler/main.py

"""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()