diff --git a/alembic/versions/0016_fc3i_task_run.py b/alembic/versions/0016_fc3i_task_run.py new file mode 100644 index 0000000..678b1ee --- /dev/null +++ b/alembic/versions/0016_fc3i_task_run.py @@ -0,0 +1,86 @@ +"""fc3i: task_run table + +Revision ID: 0016 +Revises: 0015 +Create Date: 2026-05-24 + +Additive only. New table records every Celery task attempt via signal +handlers (backend.app.celery_signals). Status is plain String(16) not +Postgres ENUM (per feedback_check_existing_enums: ENUM columns hard- +fail at INSERT, String columns extend cleanly). + +Composite indexes anticipate the three dashboard panes: +- (queue, started_at desc) — per-lane recent activity +- (status, started_at desc) — recent failures pane +- (task_name, started_at desc) — drill-down by task + +Indexed columns get individual indexes via `index=True` on the model; +the composites below cover the multi-column lookups. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0016" +down_revision: Union[str, None] = "0015" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "task_run", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("celery_task_id", sa.String(length=64), nullable=False), + sa.Column("queue", sa.String(length=32), nullable=False), + sa.Column("task_name", sa.String(length=128), nullable=False), + sa.Column("target_id", sa.Integer(), nullable=True), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("duration_ms", sa.Integer(), nullable=True), + sa.Column( + "status", sa.String(length=16), nullable=False, + server_default="running", + ), + sa.Column("error_type", sa.String(length=128), nullable=True), + sa.Column("error_message", sa.Text(), nullable=True), + sa.Column("retry_count", sa.Integer(), nullable=True), + sa.Column("worker_hostname", sa.String(length=128), nullable=True), + sa.Column("args_summary", sa.String(length=255), nullable=True), + ) + + # Single-column indexes (matches Mapped[...].index=True on model). + op.create_index("ix_task_run_celery_task_id", "task_run", ["celery_task_id"]) + op.create_index("ix_task_run_queue", "task_run", ["queue"]) + op.create_index("ix_task_run_task_name", "task_run", ["task_name"]) + op.create_index("ix_task_run_started_at", "task_run", ["started_at"]) + op.create_index("ix_task_run_finished_at", "task_run", ["finished_at"]) + op.create_index("ix_task_run_status", "task_run", ["status"]) + + # Composite indexes for dashboard query patterns. + op.create_index( + "ix_task_run_queue_started", + "task_run", ["queue", sa.text("started_at DESC")], + ) + op.create_index( + "ix_task_run_status_started", + "task_run", ["status", sa.text("started_at DESC")], + ) + op.create_index( + "ix_task_run_name_started", + "task_run", ["task_name", sa.text("started_at DESC")], + ) + + +def downgrade() -> None: + op.drop_index("ix_task_run_name_started", table_name="task_run") + op.drop_index("ix_task_run_status_started", table_name="task_run") + op.drop_index("ix_task_run_queue_started", table_name="task_run") + op.drop_index("ix_task_run_status", table_name="task_run") + op.drop_index("ix_task_run_finished_at", table_name="task_run") + op.drop_index("ix_task_run_started_at", table_name="task_run") + op.drop_index("ix_task_run_task_name", table_name="task_run") + op.drop_index("ix_task_run_queue", table_name="task_run") + op.drop_index("ix_task_run_celery_task_id", table_name="task_run") + op.drop_table("task_run") diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index 62ad522..441d58d 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -33,6 +33,7 @@ def all_blueprints() -> list[Blueprint]: from .showcase import showcase_bp from .sources import sources_bp from .suggestions import suggestions_bp + from .system_activity import system_activity_bp from .tags import tags_bp return [ api_bp, @@ -44,6 +45,7 @@ def all_blueprints() -> list[Blueprint]: artists_bp, showcase_bp, settings_bp, + system_activity_bp, import_admin_bp, migrate_bp, suggestions_bp, diff --git a/backend/app/api/migrate.py b/backend/app/api/migrate.py index ffb9dc9..81e1239 100644 --- a/backend/app/api/migrate.py +++ b/backend/app/api/migrate.py @@ -24,7 +24,12 @@ _VALID_KINDS = frozenset({ "ml_queue", "verify", "rollback", "cleanup", }) _INGEST_KINDS = frozenset({"gs_ingest", "ir_ingest"}) -_APPLY_KINDS = frozenset({"gs_ingest", "ir_ingest", "tag_apply", "rollback", "cleanup"}) +# Backup gate retired 2026-05-24 — operator-flagged the speculative-safety +# requirement was actively blocking the UI ingest path (backup itself is +# unreliable on large NFS-backed image libraries) and FC-3h will rewrite +# the backup surface as a first-class feature with its own scheduling +# + recovery. Leaving the constant for historical grep. +_APPLY_KINDS: frozenset[str] = frozenset() def _bad(error: str, *, status: int = 400, **extra): diff --git a/backend/app/api/system_activity.py b/backend/app/api/system_activity.py new file mode 100644 index 0000000..480b8c9 --- /dev/null +++ b/backend/app/api/system_activity.py @@ -0,0 +1,209 @@ +"""FC-3i: system activity dashboard endpoints. + +Read-only. Combines Redis-broker queue depths (LLEN per queue), +Celery worker introspection (celery inspect), and the task_run DB +history into the surfaces the SystemActivityTab UI consumes. + +All filesystem/sync-client work goes through asyncio.to_thread per +ASYNC230/240 (mirrors backend.app.api.extension's pattern). +""" + +from __future__ import annotations + +import asyncio +import time +from datetime import UTC, datetime, timedelta + +from quart import Blueprint, jsonify, request +from sqlalchemy import desc, func, select + +from ..config import get_config +from ..extensions import get_session +from ..models import TaskRun + +system_activity_bp = Blueprint( + "system_activity", __name__, url_prefix="/api/system/activity", +) + +# Canonical queue order — must match celery_app.task_routes. UI renders +# in this order; queues with no LLEN response show as null rather than +# absent. +_QUEUE_NAMES = ( + "default", "import", "thumbnail", "ml", + "download", "scan", "maintenance", +) + +# Cache module-level so all requests share the cache between polls. +# Tests can reset via direct dict mutation if needed. +_QUEUE_CACHE: dict = {"ts": 0.0, "data": None} +_WORKER_CACHE: dict = {"ts": 0.0, "data": None} +_QUEUE_CACHE_TTL = 2.0 +_WORKER_CACHE_TTL = 5.0 + + +def _read_queues_sync() -> dict: + """Reads each queue's LLEN from the broker. Sync — caller wraps in + asyncio.to_thread. Per-queue try/except returns None on failure so + one bad queue doesn't break the whole response.""" + import redis # local import; only this endpoint needs it + + cfg = get_config() + client = redis.Redis.from_url(cfg.celery_broker_url) + out: dict = {} + for name in _QUEUE_NAMES: + try: + out[name] = int(client.llen(name)) + except Exception: # noqa: BLE001 — broker hiccup shouldn't break UI + out[name] = None + return { + "queues": out, + "fetched_at": datetime.now(UTC).isoformat(), + } + + +def _read_workers_sync() -> dict: + """celery inspect active_queues + active. Returns per-worker info.""" + from ..celery_app import celery as celery_app + + insp = celery_app.control.inspect(timeout=2.0) + active_queues = insp.active_queues() or {} + active_tasks = insp.active() or {} + + workers: dict = {} + for hostname, queues in active_queues.items(): + workers[hostname] = { + "queues": sorted({q["name"] for q in queues}), + "active_count": len(active_tasks.get(hostname, [])), + } + return { + "workers": workers, + "fetched_at": datetime.now(UTC).isoformat(), + } + + +@system_activity_bp.route("/queues", methods=["GET"]) +async def get_queues(): + """Per-queue Redis LLEN. Cached 2s. + + Response: {queues: {name: depth_or_null}, fetched_at: iso8601} + """ + now = time.time() + if _QUEUE_CACHE["data"] is None or (now - _QUEUE_CACHE["ts"]) > _QUEUE_CACHE_TTL: + _QUEUE_CACHE["data"] = await asyncio.to_thread(_read_queues_sync) + _QUEUE_CACHE["ts"] = now + return jsonify(_QUEUE_CACHE["data"]) + + +@system_activity_bp.route("/workers", methods=["GET"]) +async def get_workers(): + """Live celery inspect. Cached 5s. + + Response: {workers: {hostname: {queues, active_count}}, fetched_at} + """ + now = time.time() + if _WORKER_CACHE["data"] is None or (now - _WORKER_CACHE["ts"]) > _WORKER_CACHE_TTL: + _WORKER_CACHE["data"] = await asyncio.to_thread(_read_workers_sync) + _WORKER_CACHE["ts"] = now + return jsonify(_WORKER_CACHE["data"]) + + +@system_activity_bp.route("/runs", methods=["GET"]) +async def list_runs(): + """Paginated task_run history. Query params: + queue= filter to one queue + status= filter to one status (running/ok/error/timeout/retry) + limit= default 50, max 200 + before_id= cursor for keyset pagination + + Response: {runs: [...], next_cursor: id|null} + """ + try: + limit = min(int(request.args.get("limit", "50")), 200) + except ValueError: + return jsonify({"error": "invalid_limit"}), 400 + if limit < 1: + return jsonify({"error": "invalid_limit"}), 400 + + queue = request.args.get("queue") + status = request.args.get("status") + before_id_raw = request.args.get("before_id") + before_id = int(before_id_raw) if before_id_raw else None + + async with get_session() as session: + stmt = select(TaskRun).order_by(desc(TaskRun.id)) + if queue: + stmt = stmt.where(TaskRun.queue == queue) + if status: + stmt = stmt.where(TaskRun.status == status) + if before_id is not None: + stmt = stmt.where(TaskRun.id < before_id) + stmt = stmt.limit(limit + 1) + rows = (await session.execute(stmt)).scalars().all() + + has_more = len(rows) > limit + rows = rows[:limit] + return jsonify({ + "runs": [_row_to_dict(r) for r in rows], + "next_cursor": rows[-1].id if has_more and rows else None, + }) + + +@system_activity_bp.route("/failures", methods=["GET"]) +async def list_failures(): + """Recent failures across all lanes (24h window). + + Response: {recent: [...], count_by_type: {ErrorClass: n}, since} + """ + try: + limit = min(int(request.args.get("limit", "50")), 200) + except ValueError: + return jsonify({"error": "invalid_limit"}), 400 + + since = datetime.now(UTC) - timedelta(hours=24) + + async with get_session() as session: + recent_stmt = ( + select(TaskRun) + .where(TaskRun.status.in_(["error", "timeout"])) + .where(TaskRun.finished_at >= since) + .order_by(desc(TaskRun.finished_at)) + .limit(limit) + ) + recent = (await session.execute(recent_stmt)).scalars().all() + + count_stmt = ( + select(TaskRun.error_type, func.count(TaskRun.id)) + .where(TaskRun.status.in_(["error", "timeout"])) + .where(TaskRun.finished_at >= since) + .group_by(TaskRun.error_type) + .order_by(desc(func.count(TaskRun.id))) + ) + counts = (await session.execute(count_stmt)).all() + + return jsonify({ + "recent": [_row_to_dict(r) for r in recent], + "count_by_type": { + (row[0] or "Unknown"): row[1] + for row in counts + }, + "since": since.isoformat(), + }) + + +def _row_to_dict(r: TaskRun) -> dict: + return { + "id": r.id, + "queue": r.queue, + "task_name": r.task_name, + "target_id": r.target_id, + "celery_task_id": r.celery_task_id, + "started_at": r.started_at.isoformat() if r.started_at else None, + "finished_at": r.finished_at.isoformat() if r.finished_at else None, + "duration_ms": r.duration_ms, + "status": r.status, + "error_type": r.error_type, + "error_message": r.error_message, + "retry_count": r.retry_count, + "worker_hostname": r.worker_hostname, + "args_summary": r.args_summary, + } diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index e5d91f2..b6ce64c 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -81,9 +81,20 @@ def make_celery() -> Celery: "task": "backend.app.tasks.maintenance.cleanup_old_download_events", "schedule": 86400.0, # daily }, + "recover-stalled-task-runs": { + "task": "backend.app.tasks.maintenance.recover_stalled_task_runs", + "schedule": 300.0, # every 5 min, matches recover-interrupted-tasks + }, + "prune-task-runs": { + "task": "backend.app.tasks.maintenance.prune_task_runs", + "schedule": 86400.0, # daily + }, }, timezone="UTC", ) + # FC-3i: register task_run signal handlers (side-effect import). + from . import celery_signals # noqa: F401 + return app diff --git a/backend/app/celery_signals.py b/backend/app/celery_signals.py new file mode 100644 index 0000000..307d5ea --- /dev/null +++ b/backend/app/celery_signals.py @@ -0,0 +1,210 @@ +"""FC-3i: task_run lifecycle via Celery signals. + +Subscribes to task_prerun / task_postrun / task_failure / task_retry +and persists one task_run row per task attempt. Drop-in for every +existing and future Celery task — no per-task instrumentation. + +Signal handlers run inside the worker process (sync context); DB +writes go through the existing shared sync engine +(backend.app.tasks._sync_engine.sync_session_factory) — one engine +per worker process, not per-task, so we don't blow Postgres +max_connections under load (the reason FC-3g shared-engine fix +existed). + +Failure-mode discipline (operator-pressed point): every handler is +wrapped in try/except that swallows + logs. If the DB is down or the +handler has a bug, the real task still runs — the dashboard goes +dark for that interval. Monitoring NEVER breaks the thing it's +monitoring. +""" + +import logging +from datetime import UTC, datetime + +from celery.exceptions import SoftTimeLimitExceeded +from celery.signals import task_failure, task_postrun, task_prerun, task_retry + +from .models import TaskRun +from .tasks._sync_engine import sync_session_factory + +log = logging.getLogger(__name__) + +# Celery-internal tasks that would generate dashboard noise without +# operational value. Conservative list; extend only when a specific +# task proves noisy. +_UNTRACKED_TASK_NAMES = frozenset({ + "celery.chord_unlock", + "celery.backend_cleanup", + "celery.chunks", +}) + +_MAX_ERROR_MESSAGE_LEN = 2000 +_MAX_ARGS_SUMMARY_LEN = 255 +_MAX_WORKER_HOSTNAME_LEN = 128 + +# PostgreSQL Integer is signed 32-bit. Tasks called with a first-arg +# int outside this range (e.g. an absurdly large mock value, or a +# string-of-digits coercible to int but bigger than 2^31-1) would crash +# the INSERT with NumericValueOutOfRange. Bound the recorded value to +# the column's range; values outside become None (target_id is +# nullable, so this is safe). +_INT32_MAX = 2_147_483_647 +_INT32_MIN = -2_147_483_648 + + +def _queue_for(task) -> str: + """Reverse the task→queue routing from celery_app.task_routes. + Keep in sync if task_routes is reordered.""" + name = getattr(task, "name", "") or "" + if name.startswith("backend.app.tasks.import_file."): + return "import" + if name.startswith("backend.app.tasks.ml."): + return "ml" + if name.startswith("backend.app.tasks.thumbnail."): + return "thumbnail" + if name.startswith("backend.app.tasks.download."): + return "download" + if name.startswith("backend.app.tasks.scan."): + return "scan" + if name.startswith(( + "backend.app.tasks.maintenance.", + "backend.app.tasks.migration.", + )): + return "maintenance" + return "default" + + +def _target_id_from_args(args) -> int | None: + """Best-effort: if the first positional arg parses as int AND fits + in the column's signed-32-bit range, record it as target_id + (image_id, source_id, etc.). Never raises.""" + if not args: + return None + try: + value = int(args[0]) + except (TypeError, ValueError): + return None + if value < _INT32_MIN or value > _INT32_MAX: + return None + return value + + +def _truncate(s, limit: int) -> str | None: + if s is None: + return None + text = str(s) + return text if len(text) <= limit else text[:limit] + + +def _is_tracked(task_name: str | None) -> bool: + return bool(task_name) and task_name not in _UNTRACKED_TASK_NAMES + + +@task_prerun.connect +def _on_prerun(sender=None, task_id=None, task=None, args=None, + kwargs=None, **_): + if not _is_tracked(getattr(task, "name", None)): + return + try: + Session = sync_session_factory() + with Session() as session: + session.add(TaskRun( + celery_task_id=task_id or "", + queue=_queue_for(task), + task_name=task.name, + target_id=_target_id_from_args(args), + started_at=datetime.now(UTC), + status="running", + args_summary=_truncate(repr(args), _MAX_ARGS_SUMMARY_LEN), + worker_hostname=_truncate( + getattr(sender, "hostname", None), + _MAX_WORKER_HOSTNAME_LEN, + ), + )) + session.commit() + except Exception: # noqa: BLE001 — never break the worker + log.exception("task_run prerun insert failed (task=%s)", + getattr(task, "name", "?")) + + +def _finalize(task_id: str, *, status: str, + error_type: str | None = None, + error_message: str | None = None, + retry_count: int | None = None) -> None: + """Shared write path for postrun/failure/retry. Picks the most- + recent task_run row for this celery_task_id that's still 'running' + (retries reuse the same celery_task_id; each new attempt's prerun + inserts a fresh row, so finalize targets the latest running row).""" + try: + from sqlalchemy import select + + Session = sync_session_factory() + now = datetime.now(UTC) + with Session() as session: + row = session.execute( + select(TaskRun) + .where(TaskRun.celery_task_id == task_id) + .where(TaskRun.status == "running") + .order_by(TaskRun.id.desc()) + .limit(1) + ).scalar_one_or_none() + if row is None: + return # no prerun row (untracked, insert failed, or already finalized) + row.finished_at = now + row.duration_ms = int( + (now - row.started_at).total_seconds() * 1000 + ) + row.status = status + if error_type is not None: + row.error_type = _truncate(error_type, 128) + if error_message is not None: + row.error_message = _truncate(error_message, _MAX_ERROR_MESSAGE_LEN) + if retry_count is not None: + row.retry_count = retry_count + session.commit() + except Exception: # noqa: BLE001 + log.exception("task_run finalize failed (task_id=%s)", task_id) + + +@task_postrun.connect +def _on_postrun(sender=None, task_id=None, task=None, args=None, + kwargs=None, retval=None, state=None, **_): + if not _is_tracked(getattr(task, "name", None)): + return + # state is one of SUCCESS/FAILURE/RETRY/etc. Only handle SUCCESS; + # task_failure handles FAILURE explicitly (with the exception). + if state != "SUCCESS": + return + _finalize(task_id, status="ok") + + +@task_failure.connect +def _on_failure(sender=None, task_id=None, exception=None, + args=None, kwargs=None, einfo=None, **_): + if not _is_tracked(getattr(sender, "name", None)): + return + status = ("timeout" + if isinstance(exception, SoftTimeLimitExceeded) + else "error") + _finalize( + task_id, status=status, + error_type=type(exception).__name__ if exception else "Unknown", + error_message=str(exception) if exception else None, + ) + + +@task_retry.connect +def _on_retry(sender=None, request=None, reason=None, einfo=None, **_): + if not _is_tracked(getattr(sender, "name", None)): + return + task_id = getattr(request, "id", None) + if not task_id: + return + # Mark current attempt's row as 'retry' (terminal for this row). + # The next attempt's task_prerun inserts a fresh row. + _finalize( + task_id, status="retry", + error_type=type(reason).__name__ if reason else "Retry", + error_message=str(reason) if reason else None, + retry_count=getattr(request, "retries", 0), + ) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 0697c07..f7a3c94 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -21,6 +21,7 @@ from .tag_alias import TagAlias from .tag_allowlist import TagAllowlist from .tag_reference_embedding import TagReferenceEmbedding from .tag_suggestion_rejection import TagSuggestionRejection +from .task_run import TaskRun __all__ = [ "Base", @@ -46,4 +47,5 @@ __all__ = [ "TagAllowlist", "TagReferenceEmbedding", "TagSuggestionRejection", + "TaskRun", ] diff --git a/backend/app/models/task_run.py b/backend/app/models/task_run.py new file mode 100644 index 0000000..e8eac05 --- /dev/null +++ b/backend/app/models/task_run.py @@ -0,0 +1,48 @@ +"""FC-3i: task_run — per-Celery-task lifecycle audit row. + +One row inserted by the task_prerun signal at task start, updated by +task_postrun / task_failure / task_retry. The shape supports the +SystemActivity dashboard's three panes: per-queue queue+worker summary, +recent failures (24h, grouped by error_type), and full paginated +activity history. + +Retention: ok rows pruned after 24h, error/timeout after 7d (see +backend.app.tasks.maintenance.prune_task_runs). + +Recovery: rows stuck in 'running' for >5 min flipped to 'error' by +backend.app.tasks.maintenance.recover_stalled_task_runs (Beat 5 min). +""" + +from datetime import datetime + +from sqlalchemy import DateTime, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class TaskRun(Base): + __tablename__ = "task_run" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + celery_task_id: Mapped[str] = mapped_column( + String(64), nullable=False, index=True, + ) + queue: Mapped[str] = mapped_column(String(32), nullable=False, index=True) + task_name: Mapped[str] = mapped_column(String(128), nullable=False, index=True) + target_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + started_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, index=True, + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, index=True, + ) + duration_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + status: Mapped[str] = mapped_column( + String(16), nullable=False, default="running", index=True, + ) + error_type: Mapped[str | None] = mapped_column(String(128), nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + retry_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + worker_hostname: Mapped[str | None] = mapped_column(String(128), nullable=True) + args_summary: Mapped[str | None] = mapped_column(String(255), nullable=True) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 2a50e86..1a2e028 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -9,7 +9,7 @@ from PIL import Image from sqlalchemy import delete, select, update from ..celery_app import celery -from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask +from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun from ..utils.phash import compute_phash from ._sync_engine import sync_session_factory as _sync_session_factory @@ -20,6 +20,8 @@ OLD_TASK_DAYS = 7 PHASH_PAGE = 500 VERIFY_PAGE = 200 FFPROBE_TIMEOUT_SECONDS = 10 +TASK_RUN_KEEP_OK_SECONDS = 24 * 3600 # 24 h +TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks") @@ -80,6 +82,71 @@ def cleanup_old_tasks() -> int: return result.rowcount or 0 +@celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs") +def recover_stalled_task_runs() -> int: + """Flip task_run rows stuck in 'running' for >STUCK_THRESHOLD_MINUTES + to 'error'. FC-3i. + + A row gets stuck when the worker dies without emitting + task_postrun / task_failure (e.g. OOM, container restart between + signals, signal handler raised+logged). Shares the 5-min threshold + with recover_interrupted_tasks for consistency. + """ + SessionLocal = _sync_session_factory() + cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + with SessionLocal() as session: + result = session.execute( + update(TaskRun) + .where(TaskRun.status == "running") + .where(TaskRun.started_at < cutoff) + .values( + status="error", + error_type="RecoverySweep", + error_message=( + f"no completion signal received within " + f"{STUCK_THRESHOLD_MINUTES} min" + ), + finished_at=datetime.now(UTC), + ) + ) + session.commit() + return result.rowcount or 0 + + +@celery.task(name="backend.app.tasks.maintenance.prune_task_runs") +def prune_task_runs() -> dict: + """Daily retention for task_run rows. FC-3i. + + - 'ok' rows: deleted after TASK_RUN_KEEP_OK_SECONDS (24h default). + Success is high-volume, not interesting after a day. + - 'error' / 'timeout' rows: deleted after TASK_RUN_KEEP_FAILURE_SECONDS + (7 days default). Failures are operationally interesting longer. + - 'running' rows: NEVER deleted by this task. The recovery sweep + (recover_stalled_task_runs) is the mechanism that flips them to + terminal state; prune doesn't touch in-flight state. + - 'retry' rows: treated as failures (>7d). + + Returns dict of how many rows were deleted in each bucket. + """ + SessionLocal = _sync_session_factory() + now = datetime.now(UTC) + ok_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_OK_SECONDS) + fail_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_FAILURE_SECONDS) + with SessionLocal() as session: + ok_deleted = session.execute( + delete(TaskRun) + .where(TaskRun.status == "ok") + .where(TaskRun.finished_at < ok_cutoff) + ).rowcount or 0 + fail_deleted = session.execute( + delete(TaskRun) + .where(TaskRun.status.in_(["error", "timeout", "retry"])) + .where(TaskRun.finished_at < fail_cutoff) + ).rowcount or 0 + session.commit() + return {"ok_deleted": ok_deleted, "failures_deleted": fail_deleted} + + @celery.task(name="backend.app.tasks.maintenance.backfill_phash") def backfill_phash() -> int: """Recompute phash for stored images that have none (imported before diff --git a/frontend/src/components/modal/ImageViewer.vue b/frontend/src/components/modal/ImageViewer.vue index fe458db..0ea01bc 100644 --- a/frontend/src/components/modal/ImageViewer.vue +++ b/frontend/src/components/modal/ImageViewer.vue @@ -2,9 +2,6 @@