diff --git a/alembic/versions/0090_service_seen.py b/alembic/versions/0090_service_seen.py new file mode 100644 index 0000000..f00c43d --- /dev/null +++ b/alembic/versions/0090_service_seen.py @@ -0,0 +1,64 @@ +"""service_seen — the learned roster that makes a stopped part observable. + +Milestone 365. Nothing in FabledCurator knew what was SUPPOSED to be running: +`celery inspect` reports the workers that answer, so a dead worker was a +shorter list rather than a red light, and the only surface that could tell an +operator otherwise was Portainer. This table is the memory that turns an +absence into something the app can see. + +Keyed on the queue set for a celery role and on agent_id for the GPU agent — +NOT on the celery worker name, which here is `celery@` and is +minted fresh on every deploy. See the model docstring for why that choice is +the whole design. + +## First migration on the collapsed baseline + +0089 is the single generated baseline that replaced revisions 0001..0089 +(milestone 328). This is the first revision written on top of it, so it is +also the first evidence that the chain steps forward from the collapse rather +than merely reproducing the schema — which nothing had demonstrated yet. + +An existing install is at 0089 because it ran the real 0089; a fresh one is at +0089 because it ran the baseline. Both arrive here identically, which was the +property the collapse was designed around. + +Revision ID: 0090 +Revises: 0089 +Create Date: 2026-09-02 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0090" +down_revision: Union[str, None] = "0089" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "service_seen", + sa.Column("key", sa.String(length=128), nullable=False), + sa.Column("kind", sa.String(length=16), nullable=False), + sa.Column("display_name", sa.String(length=64), nullable=False), + sa.Column( + "first_seen_at", sa.DateTime(timezone=True), + server_default=sa.text("now()"), nullable=False, + ), + sa.Column( + "last_seen_at", sa.DateTime(timezone=True), + server_default=sa.text("now()"), nullable=False, + ), + sa.Column("details", sa.JSON(), nullable=False), + sa.PrimaryKeyConstraint("key", name=op.f("pk_service_seen")), + ) + # No secondary indexes, deliberately: one row per moving part means every + # read is a handful of rows and an index would be write cost buying + # nothing (#3301 removed seven of exactly that shape). + + +def downgrade() -> None: + op.drop_table("service_seen") diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index f500acd..075e15d 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -38,6 +38,7 @@ def all_blueprints() -> list[Blueprint]: from .suggestions import suggestions_bp from .system_activity import system_activity_bp from .system_backup import system_backup_bp + from .system_health import system_health_bp from .tags import tags_bp from .thumbnails import thumbnails_bp return [ @@ -51,6 +52,7 @@ def all_blueprints() -> list[Blueprint]: showcase_bp, settings_bp, system_activity_bp, + system_health_bp, system_backup_bp, admin_bp, cleanup_bp, diff --git a/backend/app/api/gpu.py b/backend/app/api/gpu.py index 8ee7bb9..f0723fb 100644 --- a/backend/app/api/gpu.py +++ b/backend/app/api/gpu.py @@ -21,6 +21,7 @@ from ..services.gallery_service import image_url from ..services.ml.gpu_jobs import GpuJobService, error_dedupe_statements from ..services.ml.gpu_triage import classify_reason, recover_defective_image from ..services.ml.regions import RegionService +from ..services.service_roster import touch_service gpu_bp = Blueprint("gpu", __name__, url_prefix="/api/gpu") @@ -256,6 +257,18 @@ async def lease(): if not await _agent_authed(session): return jsonify({"error": "unauthorized"}), 401 jobs = await GpuJobService(session).lease(agent_id, batch_size=batch) + # The agent cannot be polled — it is HTTP-only and pulls from here, so + # web never dials it. A lease IS the check-in, and until milestone 365 + # it was thrown away: an agent sitting idle with nothing to lease left + # no trace at all and was indistinguishable from one switched off a + # week ago. Recorded on the call that was already happening. + await touch_service( + session, + key=f"agent:{agent_id}", + kind="agent", + display_name="GPU agent" if agent_id == "agent" else f"GPU agent ({agent_id})", + details={"agent_id": agent_id, "last_call": "lease", "leased": len(jobs)}, + ) ml = await MLSettings.load(session) # image rows for url/mime in one shot ids = [j.image_record_id for j in jobs] @@ -329,6 +342,13 @@ async def heartbeat(): if not await _agent_authed(session): return jsonify({"error": "unauthorized"}), 401 n = await GpuJobService(session).heartbeat(agent_id, job_ids) + await touch_service( + session, + key=f"agent:{agent_id}", + kind="agent", + display_name="GPU agent" if agent_id == "agent" else f"GPU agent ({agent_id})", + details={"agent_id": agent_id, "last_call": "heartbeat", "extended": n}, + ) await session.commit() return jsonify({"extended": n}) diff --git a/backend/app/api/system_health.py b/backend/app/api/system_health.py new file mode 100644 index 0000000..e767ede --- /dev/null +++ b/backend/app/api/system_health.py @@ -0,0 +1,192 @@ +"""Is every part of FabledCurator running? One verdict, one endpoint. + +Milestone 365. The nav indicator and the System page both read this and +nothing else — composing a verdict is this module's job, not the UI's. + +## Two kinds of part, answered two different ways + +**Learned** — celery roles and the GPU agent, from `service_seen`. The +question is "how long since it checked in", and these are the parts that can +be ABSENT, which is the whole point: `celery inspect` alone reports presence, +so a dead worker is a shorter list rather than a red light. + +**Probed live** — Postgres and Redis. Always expected, never learned, and a +last-seen for them would be actively misleading: that Redis answered thirty +seconds ago says nothing about now. + +## This endpoint must never fail because something it checks has failed + +The inversion is easy to write by accident and it destroys the feature exactly +when it is needed — a 500 when Redis is down, instead of `redis: down`. Every +probe is wrapped, every wait has a deadline (rule 156), and the roster refresh +swallows its own errors. The worst case is a part reported `unknown`, which is +a true statement. +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from datetime import UTC, datetime + +from quart import Blueprint, jsonify +from sqlalchemy import select, text + +from ..config import get_config +from ..extensions import get_session +from ..models import ServiceSeen +from ..services.service_roster import refresh_if_stale + +log = logging.getLogger(__name__) + +system_health_bp = Blueprint("system_health", __name__, url_prefix="/api/system") + +# How long a learned part may go quiet before it is doubted, then disbelieved. +# +# These are deliberately generous, and the reason is a deploy rather than a +# worker: `docker compose up -d` rolls start-first, so a role is briefly served +# by two containers and then by neither while the old one drains. Thresholds +# tight enough to catch a crash in seconds would paint the page red every time +# the stack is updated, and an alarm that cries wolf on every deploy is one +# nobody reads. Tune down only after watching a real deploy pass through. +STALE_AFTER_SECONDS = 90 +DOWN_AFTER_SECONDS = 300 + +# Probes cross a process boundary, so they carry deadlines. A hung Postgres +# must make this endpoint say "postgres: down", not hang alongside it. +PROBE_TIMEOUT_SECONDS = 2.0 + +_OK, _STALE, _DOWN, _UNKNOWN = "ok", "stale", "down", "unknown" + +# Worst-first, so an overall verdict is just the max. +_SEVERITY = {_OK: 0, _UNKNOWN: 1, _STALE: 2, _DOWN: 3} + + +def _age_state(age_seconds: float) -> str: + if age_seconds >= DOWN_AFTER_SECONDS: + return _DOWN + if age_seconds >= STALE_AFTER_SECONDS: + return _STALE + return _OK + + +def _describe_learned(name: str, state: str, age: float, details: dict) -> str: + """Say what the state MEANS. A red chip tells an operator less than a + sentence does at the moment they are deciding whether to go and look.""" + if state == _OK: + replicas = details.get("replicas") + if replicas and replicas > 1: + return f"{name} is running ({replicas} replicas)" + return f"{name} is running" + mins = int(age // 60) + ago = f"{mins} min" if mins else f"{int(age)}s" + if state == _STALE: + return f"{name} has not checked in for {ago}" + return f"{name} has not checked in for {ago} — treat it as stopped" + + +async def _probe_postgres(session) -> dict: + started = time.monotonic() + try: + await asyncio.wait_for( + session.execute(text("SELECT 1")), timeout=PROBE_TIMEOUT_SECONDS + ) + except Exception as exc: # noqa: BLE001 — a probe reports, it never raises + return { + "key": "postgres", "kind": "datastore", "name": "PostgreSQL", + "state": _DOWN, "detail": f"not answering: {type(exc).__name__}", + } + return { + "key": "postgres", "kind": "datastore", "name": "PostgreSQL", "state": _OK, + "detail": "answering", "latency_ms": round((time.monotonic() - started) * 1000, 1), + } + + +def _ping_redis_sync() -> None: + import redis # local import; mirrors system_activity's pattern + + client = redis.Redis.from_url( + get_config().celery_broker_url, + socket_connect_timeout=PROBE_TIMEOUT_SECONDS, + socket_timeout=PROBE_TIMEOUT_SECONDS, + ) + client.ping() + + +async def _probe_redis() -> dict: + started = time.monotonic() + try: + await asyncio.wait_for( + asyncio.to_thread(_ping_redis_sync), timeout=PROBE_TIMEOUT_SECONDS * 2 + ) + except Exception as exc: # noqa: BLE001 + return { + "key": "redis", "kind": "datastore", "name": "Redis", + "state": _DOWN, + "detail": f"not answering: {type(exc).__name__} — queues and workers " + f"cannot be reached either", + } + return { + "key": "redis", "kind": "datastore", "name": "Redis", "state": _OK, + "detail": "answering", "latency_ms": round((time.monotonic() - started) * 1000, 1), + } + + +@system_health_bp.route("/health", methods=["GET"]) +async def system_health(): + """Every part, its state, and one overall verdict. + + Response: {overall, parts: [{key, kind, name, state, detail, last_seen_at, + …}], checked_at} + """ + parts: list[dict] = [] + now = datetime.now(UTC) + + async with get_session() as session: + # Postgres first, and if it is unreachable nothing else can be read — + # say so rather than failing, because "the database is down" is the + # single most useful thing this endpoint can ever report. + pg = await _probe_postgres(session) + parts.append(pg) + + if pg["state"] == _OK: + # Rate-limited inside; see service_roster on why the web process + # is the right observer. + try: + await refresh_if_stale(session) + await session.commit() + except Exception: # noqa: BLE001 + log.warning("system health: roster refresh failed", exc_info=True) + + rows = ( + await session.execute(select(ServiceSeen).order_by(ServiceSeen.display_name)) + ).scalars().all() + for row in rows: + age = (now - row.last_seen_at).total_seconds() + state = _age_state(age) + parts.append({ + "key": row.key, + "kind": row.kind, + "name": row.display_name, + "state": state, + "detail": _describe_learned(row.display_name, state, age, row.details or {}), + "last_seen_at": row.last_seen_at.isoformat(), + "first_seen_at": row.first_seen_at.isoformat(), + **{k: v for k, v in (row.details or {}).items() if k != "agent_id"}, + }) + + parts.append(await _probe_redis()) + + overall = max((p["state"] for p in parts), key=lambda s: _SEVERITY[s], default=_UNKNOWN) + return jsonify({ + "overall": overall, + "parts": sorted(parts, key=lambda p: (-_SEVERITY[p["state"]], p["name"])), + "checked_at": now.isoformat(), + # So the UI can explain a `stale` without hard-coding the same numbers + # in a second place. + "thresholds": { + "stale_after_seconds": STALE_AFTER_SECONDS, + "down_after_seconds": DOWN_AFTER_SECONDS, + }, + }) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index a11d267..fe4bf54 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -32,6 +32,7 @@ from .presentation_review import PresentationReview from .series_chapter import SeriesChapter from .series_page import SeriesPage from .series_suggestion import SeriesSuggestion +from .service_seen import ServiceSeen from .source import Source from .subscribestar_failed_media import SubscribeStarFailedMedia from .subscribestar_seen_media import SubscribeStarSeenMedia @@ -63,6 +64,7 @@ __all__ = [ "SeriesChapter", "SeriesPage", "SeriesSuggestion", + "ServiceSeen", "ImageRecord", "ImageProvenance", "ImageRegion", diff --git a/backend/app/models/service_seen.py b/backend/app/models/service_seen.py new file mode 100644 index 0000000..69af2b8 --- /dev/null +++ b/backend/app/models/service_seen.py @@ -0,0 +1,88 @@ +"""service_seen — the learned roster of FabledCurator's own moving parts. + +Nothing else in this application knows what is SUPPOSED to be running. +`celery inspect` reports the workers that answer, so a stopped worker is a +shorter list rather than a red light, and Postgres and Redis have no +representation at all. That is why the only place an operator could see a +dead service was Portainer, which knows the intended set (milestone 365). + +This table is the memory that makes an absence observable: every part that +has ever checked in, and when it last did. A row that stops advancing is a +part that stopped. + +## Why the key is not the hostname + +`_read_workers_sync()` returns celery's worker names, which here are +`celery@`. Those are minted fresh on every deploy. Keyed on +them, this table would record a death and a birth every time the stack is +updated — and a status page that goes red on every deploy is a status page +nobody reads, which is worse than not having one. + +So a celery role is keyed on its **queue set**, which is assigned per role in +docker-compose.yml (`CELERY_QUEUES`) and survives container replacement: + + default,import,thumbnail,download -> worker + maintenance,scan -> scheduler (celery worker --beat) + ml -> ml-worker + +Two replicas of one role share a queue set and are therefore ONE row — which +is right, because the question being answered is "is that role being served", +not "how many containers exist". The replica count and their hostnames go in +`details`, where they can change without the identity changing. + +The GPU agent is keyed on its `agent_id`, the identity its lease protocol +already uses (`api/gpu.py`). + +## What is NOT in here + +Postgres and Redis. They are always expected and never learned, and a +last-seen for them would be actively misleading — that one answered thirty +seconds ago says nothing about now. They are probed live at request time. + +## kind + +Plain `String`, not a Postgres ENUM and not CHECK-gated, matching +`gpu_job.status` and `backup_run.status`. The value set here is expected to +grow as parts are added, and a constraint swap per new kind (rule 36) would +be cost with no invariant behind it. + + celery — a worker role, keyed on its queue set + agent — a GPU agent, keyed on its agent_id +""" + +from datetime import datetime + +from sqlalchemy import JSON, DateTime, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class ServiceSeen(Base): + __tablename__ = "service_seen" + + # No indexes beyond the primary key, deliberately. This table holds one row + # per moving part — a handful, forever — so every query against it is a + # full read of a few rows and an index would be write cost buying nothing + # (the lesson of #3301, which removed seven redundant ones). + key: Mapped[str] = mapped_column(String(128), primary_key=True) + kind: Mapped[str] = mapped_column(String(16), nullable=False) + + # What to call it in the UI. Derived from the queue set where it is + # recognised, and falling back to the raw queue list where it is not — a + # deployment that slices its queues differently should still show something + # true rather than a name this code invented for it. + display_name: Mapped[str] = mapped_column(String(64), nullable=False) + + first_seen_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), + ) + last_seen_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), + ) + + # The parts that change without changing identity: replica hostnames, + # active task counts, the queues actually being served. Kept as a blob + # because it is displayed and never queried — giving it columns would + # invite filtering on it, which is what the activity endpoints are for. + details: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) diff --git a/backend/app/services/service_roster.py b/backend/app/services/service_roster.py new file mode 100644 index 0000000..5b16f3b --- /dev/null +++ b/backend/app/services/service_roster.py @@ -0,0 +1,175 @@ +"""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 + +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 + +# Queue set -> the name an operator recognises. Sorted-tuple keys, because the +# order celery reports them in is not guaranteed. +# +# A deployment that slices CELERY_QUEUES differently falls through to the raw +# queue list rather than being given a name this table 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] = { + ("default", "download", "import", "thumbnail"): "Worker", + ("maintenance", "scan"): "Scheduler", + ("ml",): "ML worker", +} + + +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) + active_queues = insp.active_queues() or {} + active_tasks = insp.active() or {} + + grouped: dict[tuple[str, ...], dict] = {} + for hostname, queues in active_queues.items(): + key = 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 * 2, + ) + 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) diff --git a/frontend/src/components/TopNav.vue b/frontend/src/components/TopNav.vue index 467278b..8b99c2b 100644 --- a/frontend/src/components/TopNav.vue +++ b/frontend/src/components/TopNav.vue @@ -5,9 +5,12 @@ FabledCurator - + {{ health.icon }} - + @@ -64,13 +67,15 @@