FabledCurator/backend/app/services/service_roster.py

"""The learned roster: which of FabledCurator's parts have checked in, and when.

Milestone 365. `celery inspect` answers "who is here"; this answers "who is
missing", which nothing in the application could do before — see
`models/service_seen.py` for why the identity is a queue set and not a
worker hostname.

## Who does the observing, and why it is the web process

Three candidates, and the choice matters more than the code:

* **A celery beat sweep.** Rejected. If the scheduler dies, the sweep stops,
  every row goes stale, and the page reports that everything is down when one
  thing is. An alarm that cannot distinguish "one part died" from "the
  observer died" is worse than no alarm.
* **A background task in web.** Rejected on a detail of how this deploys:
  hypercorn runs `--workers 4`, so a `before_serving` loop would be FOUR
  concurrent inspect loops hammering the broker, forever, per container.
* **Refresh on demand, rate-limited by the data itself.** Taken. Whichever web
  process happens to serve a health request refreshes the roster if it is
  older than REFRESH_TTL, and otherwise reads what is already there.

The third has the property the other two lack: **the observer is the thing
serving the page.** If web is down you get a browser error rather than a
confidently green page, which is the honest failure. It also self-limits
without coordination — the TTL lives in the row everybody can see.
"""

from __future__ import annotations

import asyncio
import logging

from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import ServiceSeen
from .worker_lanes import LANES, lane_for_node

log = logging.getLogger(__name__)

# How stale the roster may be before a health request refreshes it. Comfortably
# under the staleness thresholds that decide a service is missing, so the
# verdict is never limited by how often anyone looked.
REFRESH_TTL_SECONDS = 20.0

# celery inspect is a broker round trip and this sits on a request path, so it
# gets a deadline (rule 156). A broker that has stopped answering must make the
# roster stale — which is a true statement about the system — not hang the one
# page that exists to explain it.
INSPECT_TIMEOUT_SECONDS = 2.0

# How many broadcast round trips `_inspect_celery_sync` makes. Named, because
# the wrapper's budget is derived from it and the two must not drift.
#
# `active_queues()` and `active()` are separate broadcasts, and a broadcast
# with no `destination` cannot know how many replies to expect — so each one
# waits out its full timeout rather than returning on the last reply. The sync
# call therefore costs ~2 x INSPECT_TIMEOUT_SECONDS in the ordinary case, not
# once.
INSPECT_ROUND_TRIPS = 2

# Slack for the thread handoff. `asyncio.to_thread` hands work to the default
# executor, and on a loaded web process — the operator's showcase page pulling
# ninety thumbnails a second — the thread may not even be scheduled inside the
# budget, let alone finish.
#
# This exists because the wrapper used to allow `INSPECT_TIMEOUT_SECONDS * 2`,
# which LOOKS like a safety factor and is exactly the worst case with nothing
# left over. Observed on the operator's first consolidated deploy, 2026-09-23:
# a TimeoutError traceback per refresh while the two inspect calls were
# working perfectly. A budget equal to the work is a budget that fails under
# any load at all.
INSPECT_SLACK_SECONDS = 3.0

# Queue set -> the name an operator recognises. Sorted-tuple keys, because the
# order celery reports them in is not guaranteed.
#
# DERIVED from `worker_lanes.LANES` (milestone 422 step 1) rather than written
# out here. It was a hand-kept second copy of the same fact, and it had already
# drifted: `maintenance_long` is a live lane with four task routes pointing at
# it and a dedicated worker in the operator's stack, and this map did not know
# it — so the System tab labelled it `Worker (maintenance_long)`. One list of
# lanes now names them everywhere.
#
# A deployment that slices CELERY_QUEUES differently still falls through to the
# raw queue list rather than being given a name this code invented for it: a
# wrong-but-confident label on a status page is worse than an ugly true one.
ROLE_NAMES: dict[tuple[str, ...], str] = {
    lane.queue_key: lane.display_name for lane in LANES
}


def role_display_name(queues: tuple[str, ...]) -> str:
    known = ROLE_NAMES.get(queues)
    if known:
        return known
    return "Worker (" + ", ".join(queues) + ")"


def _inspect_celery_sync() -> dict[tuple[str, ...], dict]:
    """celery inspect, grouped by queue set rather than by worker.

    Returns {queue_set: {"hostnames": [...], "active": int}}. Two replicas of
    one role collapse into one entry on purpose — the question is whether the
    role is being served, not how many containers exist.
    """
    from ..celery_app import celery as celery_app

    insp = celery_app.control.inspect(timeout=INSPECT_TIMEOUT_SECONDS)
    # TWO broadcasts, each waiting out its own timeout — see
    # INSPECT_ROUND_TRIPS, which the caller's budget is derived from. Adding a
    # third call here without updating that constant puts the wrapper back
    # under the work it is waiting for.
    active_queues = insp.active_queues() or {}
    active_tasks = insp.active() or {}

    grouped: dict[tuple[str, ...], dict] = {}
    for hostname, queues in active_queues.items():
        # Keyed on the LANE's queue set when the node name identifies one, so
        # a lane keeps the same roster row whether or not it is consuming.
        #
        # Grouping on the ACTIVE queues alone meant a lane at cap 0 — which
        # cancels its consumers — reported an empty set, landed under the key
        # `celery:`, and rendered as a phantom row named `Worker ()` while its
        # real row went stale beside it. Both symptoms on the operator's
        # screen, 2026-09-23, from this one line.
        #
        # Deriving the key from `lane.queue_key` rather than inventing a new
        # one keeps every existing row: it is the same string the lane already
        # had while it was running.
        lane = lane_for_node(hostname)
        key = lane.queue_key if lane else tuple(sorted({q["name"] for q in queues}))
        entry = grouped.setdefault(key, {"hostnames": [], "active": 0})
        entry["hostnames"].append(hostname)
        entry["active"] += len(active_tasks.get(hostname, []))
    for entry in grouped.values():
        entry["hostnames"].sort()
    return grouped


async def touch_service(
    session: AsyncSession, *, key: str, kind: str, display_name: str, details: dict
) -> None:
    """Record that a part checked in just now.

    Upsert rather than read-modify-write: several web processes and several
    agents can be doing this at once, and the last writer is simply the most
    recent sighting. `first_seen_at` is deliberately NOT updated — it is the
    one field that answers "has this ever run", which the learned-roster design
    depends on.
    """
    stmt = pg_insert(ServiceSeen).values(
        key=key, kind=kind, display_name=display_name, details=details,
    )
    stmt = stmt.on_conflict_do_update(
        index_elements=[ServiceSeen.key],
        set_={
            "kind": stmt.excluded.kind,
            "display_name": stmt.excluded.display_name,
            "details": stmt.excluded.details,
            "last_seen_at": func.now(),
        },
    )
    await session.execute(stmt)


async def refresh_celery_roster(session: AsyncSession) -> None:
    """Inspect the broker and record what answered. Never raises.

    A failure here means the roster does not advance, and the rows going stale
    is then a TRUE report about a broker nobody can reach. Letting the
    exception out would instead break the health endpoint, which is the one
    thing that must keep answering when the stack is unwell.
    """
    try:
        grouped = await asyncio.wait_for(
            asyncio.to_thread(_inspect_celery_sync),
            timeout=(
                INSPECT_TIMEOUT_SECONDS * INSPECT_ROUND_TRIPS
                + INSPECT_SLACK_SECONDS
            ),
        )
    except Exception:
        log.warning("service roster: celery inspect failed; roster not refreshed", exc_info=True)
        return

    for queues, entry in grouped.items():
        await touch_service(
            session,
            key="celery:" + ",".join(queues),
            kind="celery",
            display_name=role_display_name(queues),
            details={
                "queues": list(queues),
                "hostnames": entry["hostnames"],
                "replicas": len(entry["hostnames"]),
                "active": entry["active"],
            },
        )


async def refresh_if_stale(session: AsyncSession) -> None:
    """Refresh the celery roster if nobody has for REFRESH_TTL_SECONDS.

    Rate-limited by the data rather than by a lock: the gate is the newest
    last_seen_at across the celery rows, which every web process can see. Two
    processes racing through the gate costs one redundant inspect and writes
    the same values twice, so the benign outcome needs no coordination to
    prevent.
    """
    newest = (
        await session.execute(
            select(func.max(ServiceSeen.last_seen_at)).where(ServiceSeen.kind == "celery")
        )
    ).scalar_one_or_none()

    if newest is not None:
        age = (await session.execute(select(func.now()))).scalar_one() - newest
        if age.total_seconds() < REFRESH_TTL_SECONDS:
            return

    await refresh_celery_roster(session)