From 5711b4d3224c02a2d7ec3eb698b50c3cfa8faf69 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 19 Apr 2026 07:55:44 +0000 Subject: [PATCH 1/2] feat(background_tasks): add Celery + Redis task queue with admin UI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a new background_tasks module: Celery app with Redis broker/result backend, a TaskExecution table populated via Celery signals, an admin API + Inertia UI that lists executions with filters, and a retry action gated to failed/stuck rows. Other modules declare tasks by shipping a tasks.py — autodiscovery picks them up without framework changes. Infra: dedicated docker/worker.Dockerfile for worker + beat services, redis service added to docker-compose, .dockerignore, Makefile targets for local and containerized workers. https://claude.ai/code/session_013VMtYvyhYDHEDHGKWAQqPf --- .dockerignore | 34 +++ .env.example | 10 + Makefile | 15 +- docker-compose.yml | 67 +++++ docker/worker.Dockerfile | 51 ++++ ...4d2_add_background_tasks_task_execution.py | 115 ++++++++ host/pyproject.toml | 2 + .../background_tasks/__init__.py | 1 + .../background_tasks/_signal_support.py | 93 +++++++ .../background_tasks/celery_app.py | 98 +++++++ .../background_tasks/constants.py | 84 ++++++ .../background_tasks/contracts/__init__.py | 1 + .../background_tasks/contracts/events.py | 26 ++ .../background_tasks/contracts/schemas.py | 66 +++++ .../background_tasks/contracts/service.py | 32 +++ .../background_tasks/background_tasks/deps.py | 19 ++ .../background_tasks/endpoints/__init__.py | 0 .../background_tasks/endpoints/api_admin.py | 62 +++++ .../background_tasks/endpoints/views.py | 71 +++++ .../background_tasks/locales/en.json | 57 ++++ .../background_tasks/models.py | 86 ++++++ .../background_tasks/module.py | 96 +++++++ .../background_tasks/pages/Detail.tsx | 215 +++++++++++++++ .../background_tasks/pages/Index.tsx | 202 ++++++++++++++ .../pages/components/ExecutionRow.tsx | 115 ++++++++ .../background_tasks/pages/constants.ts | 49 ++++ .../background_tasks/py.typed | 0 .../background_tasks/service.py | 156 +++++++++++ .../background_tasks/services.py | 25 ++ .../background_tasks/settings.py | 66 +++++ .../background_tasks/signals.py | 232 ++++++++++++++++ .../background_tasks/sync_db.py | 68 +++++ .../background_tasks/tasks.py | 102 +++++++ modules/background_tasks/package.json | 16 ++ modules/background_tasks/pyproject.toml | 25 ++ .../background_tasks/tests/test_admin_api.py | 107 ++++++++ .../background_tasks/tests/test_service.py | 198 ++++++++++++++ .../background_tasks/tests/test_signals.py | 220 +++++++++++++++ modules/background_tasks/tsconfig.json | 11 + package-lock.json | 17 ++ packages/i18n/src/generated-resources.ts | 42 +++ packages/i18n/src/keys.generated.ts | 252 +++++++++++------- pyproject.toml | 3 +- scripts/run_worker.py | 20 ++ 44 files changed, 3127 insertions(+), 100 deletions(-) create mode 100644 .dockerignore create mode 100644 docker/worker.Dockerfile create mode 100644 host/migrations/versions/6df7645cc4d2_add_background_tasks_task_execution.py create mode 100644 modules/background_tasks/background_tasks/__init__.py create mode 100644 modules/background_tasks/background_tasks/_signal_support.py create mode 100644 modules/background_tasks/background_tasks/celery_app.py create mode 100644 modules/background_tasks/background_tasks/constants.py create mode 100644 modules/background_tasks/background_tasks/contracts/__init__.py create mode 100644 modules/background_tasks/background_tasks/contracts/events.py create mode 100644 modules/background_tasks/background_tasks/contracts/schemas.py create mode 100644 modules/background_tasks/background_tasks/contracts/service.py create mode 100644 modules/background_tasks/background_tasks/deps.py create mode 100644 modules/background_tasks/background_tasks/endpoints/__init__.py create mode 100644 modules/background_tasks/background_tasks/endpoints/api_admin.py create mode 100644 modules/background_tasks/background_tasks/endpoints/views.py create mode 100644 modules/background_tasks/background_tasks/locales/en.json create mode 100644 modules/background_tasks/background_tasks/models.py create mode 100644 modules/background_tasks/background_tasks/module.py create mode 100644 modules/background_tasks/background_tasks/pages/Detail.tsx create mode 100644 modules/background_tasks/background_tasks/pages/Index.tsx create mode 100644 modules/background_tasks/background_tasks/pages/components/ExecutionRow.tsx create mode 100644 modules/background_tasks/background_tasks/pages/constants.ts create mode 100644 modules/background_tasks/background_tasks/py.typed create mode 100644 modules/background_tasks/background_tasks/service.py create mode 100644 modules/background_tasks/background_tasks/services.py create mode 100644 modules/background_tasks/background_tasks/settings.py create mode 100644 modules/background_tasks/background_tasks/signals.py create mode 100644 modules/background_tasks/background_tasks/sync_db.py create mode 100644 modules/background_tasks/background_tasks/tasks.py create mode 100644 modules/background_tasks/package.json create mode 100644 modules/background_tasks/pyproject.toml create mode 100644 modules/background_tasks/tests/test_admin_api.py create mode 100644 modules/background_tasks/tests/test_service.py create mode 100644 modules/background_tasks/tests/test_signals.py create mode 100644 modules/background_tasks/tsconfig.json create mode 100644 scripts/run_worker.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..be06fc56 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,34 @@ +# Python / uv +__pycache__/ +*.py[cod] +*$py.class +.venv/ +.uv/ +.pytest_cache/ +.ruff_cache/ +.mypy_cache/ +.ty_cache/ +*.egg-info/ +dist/ +build/ + +# Node / Vite +node_modules/ +host/client_app/dist/ +packages/*/dist/ +.npm/ + +# Editors / OS +.vscode/ +.idea/ +.DS_Store + +# Secrets / local data +.env +*.db +*.sqlite +.memray/ + +# Tests / docs not needed at runtime +tests/e2e/ +docs/ diff --git a/.env.example b/.env.example index 5d18ea43..68490c6b 100644 --- a/.env.example +++ b/.env.example @@ -46,3 +46,13 @@ SM_USERS_COOKIE_SECURE=false # Multi-tenancy (default off). # SM_MULTI_TENANT=false # SM_TENANT_HEADER= + +# BackgroundTasks module — Celery + Redis task queue. +# In docker-compose these point at the `redis` service host (redis://redis:6379/*). +SM_BG_TASKS_BROKER_URL=redis://localhost:6379/0 +SM_BG_TASKS_RESULT_BACKEND=redis://localhost:6379/1 +# How long a task can stay in `running` without a heartbeat before it's +# flipped to `stuck` by the beat sweep. +SM_BG_TASKS_STUCK_AFTER_SECONDS=300 +# Purge terminal rows older than this (days) — keeps history bounded. +SM_BG_TASKS_RETENTION_DAYS=14 diff --git a/Makefile b/Makefile index e1b9db5a..867c45af 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: install install-py install-js dev dev-api dev-ui build test test-py test-js test-e2e bench memray-run memray-flamegraph loadtest loadtest-memray lint doctor migrate migration downgrade migration-history docker-up docker-down kill new-module gen-pages sync-module-deps ci-python-lint ci-python-typecheck ci-js-lint ci-js-typecheck ci-check-file-size +.PHONY: install install-py install-js dev dev-api dev-ui build test test-py test-js test-e2e bench memray-run memray-flamegraph loadtest loadtest-memray lint doctor migrate migration downgrade migration-history docker-up docker-down kill new-module gen-pages sync-module-deps ci-python-lint ci-python-typecheck ci-js-lint ci-js-typecheck ci-check-file-size worker beat worker-docker # Install install: @@ -138,7 +138,18 @@ kill: # Docker docker-up: - docker compose up -d + docker compose up -d postgres redis docker-down: docker compose down + +# Celery — local dev (fast reload, no container rebuild) +worker: ## Run a Celery worker locally against $(SM_BG_TASKS_BROKER_URL) + uv run celery -A scripts.run_worker:celery worker -l info + +beat: ## Run the Celery beat scheduler locally + uv run celery -A scripts.run_worker:celery beat -l info + +# Celery — containerized (matches prod image) +worker-docker: ## Build + run the worker + beat services in docker + docker compose up --build worker beat diff --git a/docker-compose.yml b/docker-compose.yml index 2c62e492..f1d90f8a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -9,6 +9,73 @@ services: - "5432:5432" volumes: - pgdata:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U sm -d simple_module"] + interval: 5s + timeout: 5s + retries: 10 + + redis: + image: redis:7-alpine + ports: + - "6379:6379" + volumes: + - redisdata:/data + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 10 + + worker: + build: + context: . + dockerfile: docker/worker.Dockerfile + env_file: .env + environment: + SM_BG_TASKS_BROKER_URL: redis://redis:6379/0 + SM_BG_TASKS_RESULT_BACKEND: redis://redis:6379/1 + SM_DATABASE_URL: ${SM_DATABASE_URL:-postgresql+asyncpg://sm:sm@postgres:5432/simple_module} + depends_on: + redis: + condition: service_healthy + postgres: + condition: service_healthy + command: + - "uv" + - "run" + - "celery" + - "-A" + - "scripts.run_worker:celery" + - "worker" + - "-l" + - "info" + - "--concurrency=4" + + beat: + build: + context: . + dockerfile: docker/worker.Dockerfile + env_file: .env + environment: + SM_BG_TASKS_BROKER_URL: redis://redis:6379/0 + SM_BG_TASKS_RESULT_BACKEND: redis://redis:6379/1 + SM_DATABASE_URL: ${SM_DATABASE_URL:-postgresql+asyncpg://sm:sm@postgres:5432/simple_module} + depends_on: + redis: + condition: service_healthy + worker: + condition: service_started + command: + - "uv" + - "run" + - "celery" + - "-A" + - "scripts.run_worker:celery" + - "beat" + - "-l" + - "info" volumes: pgdata: + redisdata: diff --git a/docker/worker.Dockerfile b/docker/worker.Dockerfile new file mode 100644 index 00000000..ab8bc0f8 --- /dev/null +++ b/docker/worker.Dockerfile @@ -0,0 +1,51 @@ +# Celery worker image for the BackgroundTasks module. +# +# Serves both the ``worker`` and ``beat`` services in docker-compose — they +# differ only by command. A separate image (rather than reusing a web one) +# keeps the runtime surface narrow: no Vite, no static assets, no hot-reload +# tooling. Everything needed is ``uv sync`` + the project source. + +FROM python:3.12-slim AS base + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + UV_LINK_MODE=copy \ + UV_COMPILE_BYTECODE=1 \ + UV_SYSTEM_PYTHON=1 + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + curl \ + ca-certificates \ + build-essential \ + && rm -rf /var/lib/apt/lists/* \ + && pip install --no-cache-dir uv + +WORKDIR /app + +# Dependency-only layer so code changes don't bust the wheel cache. +COPY pyproject.toml uv.lock ./ +COPY framework/ framework/ +COPY modules/ modules/ +COPY host/ host/ +COPY scripts/ scripts/ + +# Install the full workspace (all framework + module packages) without dev +# extras. Production workers never need pytest / ruff / ty. +RUN uv sync --all-packages --frozen --no-dev + +# Drop privileges. uv writes to /app/.venv so it needs write access until +# sync completes; chown after install to keep the build step rootful. +RUN useradd --system --uid 10001 --home /app --shell /usr/sbin/nologin worker \ + && chown -R worker:worker /app +USER worker + +# Container-native healthcheck: ask the worker to ping itself via Celery's +# control protocol. Fails fast if the worker has crashed but the PID lingers. +ENV CELERY_APP=scripts.run_worker:celery +HEALTHCHECK --interval=30s --timeout=10s --start-period=30s --retries=3 \ + CMD uv run celery -A $CELERY_APP inspect ping -d celery@$HOSTNAME || exit 1 + +# Default command is the worker; docker-compose overrides to `beat` for the +# scheduler service. +CMD ["uv", "run", "celery", "-A", "scripts.run_worker:celery", "worker", "-l", "info"] diff --git a/host/migrations/versions/6df7645cc4d2_add_background_tasks_task_execution.py b/host/migrations/versions/6df7645cc4d2_add_background_tasks_task_execution.py new file mode 100644 index 00000000..920c9d68 --- /dev/null +++ b/host/migrations/versions/6df7645cc4d2_add_background_tasks_task_execution.py @@ -0,0 +1,115 @@ +"""add background_tasks_task_execution + +Revision ID: 6df7645cc4d2 +Revises: b7e1af4c9d02 +Create Date: 2026-04-19 07:44:06.824688 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "6df7645cc4d2" +down_revision: str | None = "b7e1af4c9d02" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "background_tasks_task_execution", + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("(CURRENT_TIMESTAMP)"), + nullable=False, + ), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column("updated_by", sa.String(length=255), nullable=True), + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("celery_task_id", sa.String(length=64), nullable=True), + sa.Column("task_name", sa.String(length=255), nullable=False), + sa.Column("status", sa.String(length=20), nullable=False), + sa.Column("queue", sa.String(length=64), nullable=False), + sa.Column("args", sa.JSON(), nullable=True), + sa.Column("kwargs", sa.JSON(), nullable=True), + sa.Column("result", sa.JSON(), nullable=True), + sa.Column("traceback", sa.String(), nullable=True), + sa.Column("exception_type", sa.String(length=255), nullable=True), + sa.Column("worker", sa.String(length=255), nullable=True), + sa.Column("retries", sa.Integer(), nullable=False), + sa.Column("retried_from_id", sa.Uuid(), nullable=True), + sa.Column("queued_at", sa.DateTime(), nullable=True), + sa.Column("started_at", sa.DateTime(), nullable=True), + sa.Column("finished_at", sa.DateTime(), nullable=True), + sa.Column("heartbeat_at", sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint( + ["retried_from_id"], + ["background_tasks_task_execution.id"], + name=op.f( + "fk_background_tasks_task_execution_retried_from_id_background_tasks_task_execution" + ), + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_background_tasks_task_execution")), + ) + op.create_index( + op.f("ix_background_tasks_task_execution_celery_task_id"), + "background_tasks_task_execution", + ["celery_task_id"], + unique=False, + ) + op.create_index( + op.f("ix_background_tasks_task_execution_retried_from_id"), + "background_tasks_task_execution", + ["retried_from_id"], + unique=False, + ) + op.create_index( + op.f("ix_background_tasks_task_execution_status"), + "background_tasks_task_execution", + ["status"], + unique=False, + ) + op.create_index( + "ix_background_tasks_task_execution_status_queued", + "background_tasks_task_execution", + ["status", "queued_at"], + unique=False, + ) + op.create_index( + op.f("ix_background_tasks_task_execution_task_name"), + "background_tasks_task_execution", + ["task_name"], + unique=False, + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index( + op.f("ix_background_tasks_task_execution_task_name"), + table_name="background_tasks_task_execution", + ) + op.drop_index( + "ix_background_tasks_task_execution_status_queued", + table_name="background_tasks_task_execution", + ) + op.drop_index( + op.f("ix_background_tasks_task_execution_status"), + table_name="background_tasks_task_execution", + ) + op.drop_index( + op.f("ix_background_tasks_task_execution_retried_from_id"), + table_name="background_tasks_task_execution", + ) + op.drop_index( + op.f("ix_background_tasks_task_execution_celery_task_id"), + table_name="background_tasks_task_execution", + ) + op.drop_table("background_tasks_task_execution") + # ### end Alembic commands ### diff --git a/host/pyproject.toml b/host/pyproject.toml index ed15268a..994d3fcb 100644 --- a/host/pyproject.toml +++ b/host/pyproject.toml @@ -8,6 +8,7 @@ dependencies = [ "auth", "dashboard", "products", + "background-tasks", "python-multipart>=0.0.6", ] @@ -16,3 +17,4 @@ simple-module-hosting = { workspace = true } auth = { workspace = true } products = { workspace = true } dashboard = { workspace = true } +background-tasks = { workspace = true } diff --git a/modules/background_tasks/background_tasks/__init__.py b/modules/background_tasks/background_tasks/__init__.py new file mode 100644 index 00000000..d065990b --- /dev/null +++ b/modules/background_tasks/background_tasks/__init__.py @@ -0,0 +1 @@ +"""BackgroundTasks module — Celery + Redis task queue with admin UI.""" diff --git a/modules/background_tasks/background_tasks/_signal_support.py b/modules/background_tasks/background_tasks/_signal_support.py new file mode 100644 index 00000000..7cf05663 --- /dev/null +++ b/modules/background_tasks/background_tasks/_signal_support.py @@ -0,0 +1,93 @@ +"""Signal-handler helpers shared across :mod:`background_tasks.signals`. + +Kept in a separate module so :mod:`.signals` stays under the 300-line cap. +These helpers are intentionally side-effect-free except for the DB write in +``upsert_by_celery_id`` — signal handlers compose them. +""" + +from __future__ import annotations + +import traceback as tb_mod +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from background_tasks.models import TaskExecution + + +def now_utc() -> datetime: + return datetime.now(UTC) + + +def coerce_args_kwargs(args: Any, kwargs: Any) -> tuple[list[Any], dict[str, Any]]: + """Normalise Celery args/kwargs payloads to JSON-ready shapes. + + Celery sometimes hands us tuples/frozensets that our JSON column can't + store; also tolerates ``None`` so a partial payload won't wedge the row. + """ + safe_args: list[Any] = list(args) if args else [] + safe_kwargs: dict[str, Any] = dict(kwargs) if kwargs else {} + return safe_args, safe_kwargs + + +def upsert_by_celery_id( + session: Session, + *, + celery_task_id: str | None, + task_name: str, + defaults: dict[str, Any], +) -> TaskExecution: + """Fetch the row for ``celery_task_id``, or create it with ``defaults``. + + Handlers never race on the same row within a worker process — Celery + serialises per-task signals — so this is upsert-by-read, no advisory + locks needed. + """ + row: TaskExecution | None = None + if celery_task_id: + stmt = select(TaskExecution).where(TaskExecution.celery_task_id == celery_task_id) + row = session.execute(stmt).scalar_one_or_none() + if row is None: + # Callers sometimes include ``task_name`` in defaults too — let their + # value win, but ensure a name is always set. + row_kwargs = {"task_name": task_name, **defaults} + row = TaskExecution(celery_task_id=celery_task_id, **row_kwargs) + session.add(row) + else: + for key, value in defaults.items(): + setattr(row, key, value) + return row + + +def task_name_of(sender: Any, task: Any) -> str: + """Best-effort resolution of a task's registered name from signal args.""" + return ( + (isinstance(sender, str) and sender) + or getattr(task, "name", None) + or getattr(sender, "name", None) + or "unknown" + ) + + +def jsonable_result(result: Any) -> dict[str, Any] | None: + """Wrap arbitrary task results in a JSON-storable shape.""" + if result is None: + return None + if isinstance(result, dict): + return result + try: + return {"value": result} + except Exception: + return {"value": repr(result)} + + +def render_traceback(einfo: Any, exception: BaseException | None) -> str | None: + """Prefer Celery's pre-formatted traceback; fall back to the live exception.""" + tb = getattr(einfo, "traceback", None) + if tb: + return str(tb) + if exception is not None: + return "".join(tb_mod.format_exception(type(exception), exception, exception.__traceback__)) + return None diff --git a/modules/background_tasks/background_tasks/celery_app.py b/modules/background_tasks/background_tasks/celery_app.py new file mode 100644 index 00000000..f1989125 --- /dev/null +++ b/modules/background_tasks/background_tasks/celery_app.py @@ -0,0 +1,98 @@ +"""Build and configure the Celery application. + +Invoked from two places: + +- :meth:`BackgroundTasksModule.on_startup` in the web process, so the API + can enqueue tasks via ``app.state.background_tasks.celery.send_task(...)``. +- ``scripts/run_worker.py`` at worker boot, so the worker process uses an + identical config (broker URL, signal handlers, autodiscovered tasks). + +Task discovery uses :py:meth:`celery.Celery.autodiscover_tasks` with the +top-level package name of every installed module (enumerated via the +``simple_module`` entry-point group). Any module that ships a ``tasks.py`` +is picked up automatically — no framework hook, no per-module registration. +""" + +from __future__ import annotations + +import logging +from importlib.metadata import entry_points + +from celery import Celery +from celery.schedules import schedule + +from background_tasks.constants import ( + INTERNAL_TASK_PURGE_OLD, + INTERNAL_TASK_SWEEP_STUCK, + MODULE_NAME, +) +from background_tasks.settings import BackgroundTasksSettings + +logger = logging.getLogger(__name__) + +_ENTRY_POINT_GROUP = "simple_module" + + +def _discover_task_packages() -> list[str]: + """Return the top-level package name of every installed simple_module. + + Celery's ``autodiscover_tasks`` imports ``.tasks`` for each name + returned here, so any module that ships a ``tasks.py`` registers its + tasks without a per-module hook. + """ + packages: set[str] = set() + for ep in entry_points(group=_ENTRY_POINT_GROUP): + # ep.value looks like "background_tasks.module:BackgroundTasksModule". + # We want just the top-level package name. + module_path = ep.value.split(":", 1)[0] + top_level = module_path.split(".", 1)[0] + packages.add(top_level) + return sorted(packages) + + +def build_celery(settings: BackgroundTasksSettings) -> Celery: + """Construct a Celery app wired to the project's Redis broker.""" + celery = Celery(MODULE_NAME) + + celery.conf.update( + broker_url=settings.broker_url, + result_backend=settings.result_backend, + task_default_queue=settings.task_default_queue, + # ``task_track_started`` gives us the ``STARTED`` state so + # ``task_prerun`` can flip our row to ``running``. + task_track_started=True, + # ``task_acks_late`` + ``worker_prefetch_multiplier=1`` gives us + # at-least-once semantics, which matches how the admin UI asks users + # to think about retries — duplicates beat lost work. + task_acks_late=True, + worker_prefetch_multiplier=1, + task_serializer="json", + result_serializer="json", + accept_content=["json"], + timezone="UTC", + enable_utc=True, + broker_connection_retry_on_startup=True, + beat_schedule={ + "background-tasks-sweep-stuck": { + "task": INTERNAL_TASK_SWEEP_STUCK, + "schedule": schedule(settings.stuck_sweep_interval_seconds), + }, + "background-tasks-purge-old": { + "task": INTERNAL_TASK_PURGE_OLD, + "schedule": schedule(settings.purge_interval_seconds), + }, + }, + beat_scheduler="celery.beat:PersistentScheduler", + ) + + # Autodiscover across every installed simple_module — a module just ships + # a `tasks.py` and the worker picks it up. + packages = _discover_task_packages() + logger.info("Celery autodiscover_tasks across: %s", packages) + celery.autodiscover_tasks(packages, related_name="tasks", force=True) + + # Side-effect import: connects signal handlers to this Celery instance's + # ``celery.signals.*`` globals. Safe to import repeatedly. + from background_tasks import signals # noqa: F401 + + return celery diff --git a/modules/background_tasks/background_tasks/constants.py b/modules/background_tasks/background_tasks/constants.py new file mode 100644 index 00000000..e65e85b6 --- /dev/null +++ b/modules/background_tasks/background_tasks/constants.py @@ -0,0 +1,84 @@ +"""Single source of truth for background_tasks module string literals. + +Anything that would otherwise become a magic string — table names, env-var +prefixes, permission codes, route segments, page identifiers, task-status +values — lives here so models, signals, services, endpoints, tests, and +the frontend all agree on one spelling. +""" + +from __future__ import annotations + +from enum import StrEnum + +# ── Module identity ───────────────────────────────────────────── +MODULE_NAME = "background_tasks" +MODULE_DISPLAY_NAME = "BackgroundTasks" +TABLE_PREFIX = "background_tasks_" +TABLE_TASK_EXECUTION = f"{TABLE_PREFIX}task_execution" + +# ── Env / settings ────────────────────────────────────────────── +ENV_PREFIX = "SM_BG_TASKS_" + +# ── Permissions ───────────────────────────────────────────────── +PERM_GROUP = "Background Tasks" +PERM_VIEW = "background_tasks.view" +PERM_MANAGE = "background_tasks.manage" + +# ── Routes ────────────────────────────────────────────────────── +API_PREFIX = "/api/background_tasks" +VIEW_PREFIX = "/admin/background-tasks" +ADMIN_ROUTER_PREFIX = "/admin" + +# ── Menu ──────────────────────────────────────────────────────── +MENU_LABEL = "Background Tasks" +MENU_ICON = "activity" +MENU_ORDER = 80 + +# ── Page identifiers ──────────────────────────────────────────── +# Kept as literals at the call site (see endpoints/views.py) so +# ``make doctor``'s SM003/SM004 static analysis can match pages to renders. + + +# ── Task statuses ─────────────────────────────────────────────── +class TaskStatus(StrEnum): + """Lifecycle state of a single task execution row.""" + + PENDING = "pending" + RUNNING = "running" + SUCCESS = "success" + FAILED = "failed" + STUCK = "stuck" + REVOKED = "revoked" + RETRYING = "retrying" + + +RETRYABLE_STATUSES: frozenset[TaskStatus] = frozenset({TaskStatus.FAILED, TaskStatus.STUCK}) +TERMINAL_STATUSES: frozenset[TaskStatus] = frozenset( + {TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.STUCK, TaskStatus.REVOKED} +) +RUNNING_STATUSES: frozenset[TaskStatus] = frozenset( + {TaskStatus.PENDING, TaskStatus.RUNNING, TaskStatus.RETRYING} +) + +# ── Defaults ──────────────────────────────────────────────────── +DEFAULT_QUEUE = "default" +DEFAULT_BROKER_URL = "redis://localhost:6379/0" +DEFAULT_RESULT_BACKEND = "redis://localhost:6379/1" +DEFAULT_STUCK_AFTER_SECONDS = 300 +DEFAULT_STUCK_SWEEP_INTERVAL_SECONDS = 60 +DEFAULT_PURGE_INTERVAL_SECONDS = 60 * 60 * 24 +DEFAULT_RETENTION_DAYS = 14 +DEFAULT_MAX_RETRIES = 3 + +# ── Internal task names ───────────────────────────────────────── +INTERNAL_TASK_SWEEP_STUCK = "background_tasks.sweep_stuck_tasks" +INTERNAL_TASK_PURGE_OLD = "background_tasks.purge_old_executions" + +# ── Celery signal knobs ───────────────────────────────────────── +# Celery's built-in state strings don't line up 1:1 with TaskStatus, so we +# map at the signal layer. These are the keys we read off Celery state. +CELERY_STATE_STARTED = "STARTED" +CELERY_STATE_SUCCESS = "SUCCESS" +CELERY_STATE_FAILURE = "FAILURE" +CELERY_STATE_RETRY = "RETRY" +CELERY_STATE_REVOKED = "REVOKED" diff --git a/modules/background_tasks/background_tasks/contracts/__init__.py b/modules/background_tasks/background_tasks/contracts/__init__.py new file mode 100644 index 00000000..7f44f7d6 --- /dev/null +++ b/modules/background_tasks/background_tasks/contracts/__init__.py @@ -0,0 +1 @@ +"""Public contracts for the BackgroundTasks module.""" diff --git a/modules/background_tasks/background_tasks/contracts/events.py b/modules/background_tasks/background_tasks/contracts/events.py new file mode 100644 index 00000000..71b2b3d5 --- /dev/null +++ b/modules/background_tasks/background_tasks/contracts/events.py @@ -0,0 +1,26 @@ +"""Public events emitted by the BackgroundTasks module.""" + +from __future__ import annotations + +import uuid +from dataclasses import dataclass + +from simple_module_core.events import Event + + +@dataclass +class TaskFailed(Event): + """A task transitioned to ``failed`` — subscribers may alert/log.""" + + task_execution_id: uuid.UUID + task_name: str + exception_type: str | None + + +@dataclass +class TaskRetried(Event): + """A failed/stuck task was manually retried via the admin UI.""" + + original_id: uuid.UUID + new_id: uuid.UUID + task_name: str diff --git a/modules/background_tasks/background_tasks/contracts/schemas.py b/modules/background_tasks/background_tasks/contracts/schemas.py new file mode 100644 index 00000000..d6cecfce --- /dev/null +++ b/modules/background_tasks/background_tasks/contracts/schemas.py @@ -0,0 +1,66 @@ +"""DTOs returned by the BackgroundTasks admin API / views.""" + +from __future__ import annotations + +import uuid +from datetime import datetime +from typing import Any + +from pydantic import ConfigDict +from sqlmodel import SQLModel + +from background_tasks.constants import TaskStatus + + +class TaskExecutionListItem(SQLModel): + """Compact row for the admin listing table.""" + + model_config = ConfigDict(from_attributes=True) + + id: uuid.UUID + celery_task_id: str | None = None + task_name: str + status: TaskStatus + queue: str + retries: int + worker: str | None = None + queued_at: datetime | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + exception_type: str | None = None + retried_from_id: uuid.UUID | None = None + + +class TaskExecutionDetail(SQLModel): + """Full row for the detail page (adds args/kwargs/result/traceback).""" + + model_config = ConfigDict(from_attributes=True) + + id: uuid.UUID + celery_task_id: str | None = None + task_name: str + status: TaskStatus + queue: str + args: list[Any] = [] + kwargs: dict[str, Any] = {} + result: dict[str, Any] | None = None + traceback: str | None = None + exception_type: str | None = None + worker: str | None = None + retries: int = 0 + retried_from_id: uuid.UUID | None = None + queued_at: datetime | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + heartbeat_at: datetime | None = None + + +class TaskExecutionListResponse(SQLModel): + """Paginated response shape used by both the API and Inertia views.""" + + items: list[TaskExecutionListItem] + total: int + page: int + per_page: int + status: TaskStatus | None = None + task_name: str | None = None diff --git a/modules/background_tasks/background_tasks/contracts/service.py b/modules/background_tasks/background_tasks/contracts/service.py new file mode 100644 index 00000000..6966af70 --- /dev/null +++ b/modules/background_tasks/background_tasks/contracts/service.py @@ -0,0 +1,32 @@ +"""Public service interface for the BackgroundTasks module. + +Other modules depend on this Protocol, not on the concrete implementation. +""" + +from __future__ import annotations + +import uuid +from typing import Protocol + +from background_tasks.constants import TaskStatus +from background_tasks.contracts.schemas import ( + TaskExecutionDetail, + TaskExecutionListResponse, +) + + +class IBackgroundTaskService(Protocol): + """Surface exposed to callers outside the module.""" + + async def list( + self, + *, + status: TaskStatus | None = None, + task_name: str | None = None, + page: int = 1, + per_page: int = 20, + ) -> TaskExecutionListResponse: ... + + async def get(self, execution_id: uuid.UUID) -> TaskExecutionDetail | None: ... + + async def retry(self, execution_id: uuid.UUID) -> TaskExecutionDetail: ... diff --git a/modules/background_tasks/background_tasks/deps.py b/modules/background_tasks/background_tasks/deps.py new file mode 100644 index 00000000..a39cc1bc --- /dev/null +++ b/modules/background_tasks/background_tasks/deps.py @@ -0,0 +1,19 @@ +"""FastAPI dependencies for the BackgroundTasks module.""" + +from __future__ import annotations + +from fastapi import Depends, Request +from simple_module_core.events import EventBus +from simple_module_db.deps import get_db +from sqlalchemy.ext.asyncio import AsyncSession + +from background_tasks.service import BackgroundTaskService + + +async def get_background_task_service( + request: Request, + db: AsyncSession = Depends(get_db), +) -> BackgroundTaskService: + services = request.app.state.background_tasks + bus: EventBus = request.app.state.sm.event_bus + return BackgroundTaskService(db=db, celery=services.celery, event_bus=bus) diff --git a/modules/background_tasks/background_tasks/endpoints/__init__.py b/modules/background_tasks/background_tasks/endpoints/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/modules/background_tasks/background_tasks/endpoints/api_admin.py b/modules/background_tasks/background_tasks/endpoints/api_admin.py new file mode 100644 index 00000000..39f14469 --- /dev/null +++ b/modules/background_tasks/background_tasks/endpoints/api_admin.py @@ -0,0 +1,62 @@ +"""Admin REST endpoints for BackgroundTasks.""" + +from __future__ import annotations + +import uuid + +from fastapi import APIRouter, Depends, HTTPException, Query +from simple_module_hosting.permissions import RequiresPermission + +from background_tasks.constants import ( + ADMIN_ROUTER_PREFIX, + MODULE_NAME, + PERM_MANAGE, + PERM_VIEW, + TaskStatus, +) +from background_tasks.contracts.schemas import ( + TaskExecutionDetail, + TaskExecutionListResponse, +) +from background_tasks.deps import get_background_task_service +from background_tasks.service import BackgroundTaskService + +router = APIRouter( + prefix=ADMIN_ROUTER_PREFIX, + dependencies=[Depends(RequiresPermission(PERM_VIEW))], + tags=[f"{MODULE_NAME}-admin"], +) + + +@router.get("/executions", response_model=TaskExecutionListResponse) +async def list_executions( + status: TaskStatus | None = Query(default=None), + task_name: str | None = Query(default=None), + page: int = Query(default=1, ge=1), + per_page: int = Query(default=20, ge=1, le=200), + service: BackgroundTaskService = Depends(get_background_task_service), +) -> TaskExecutionListResponse: + return await service.list(status=status, task_name=task_name, page=page, per_page=per_page) + + +@router.get("/executions/{execution_id}", response_model=TaskExecutionDetail) +async def get_execution( + execution_id: uuid.UUID, + service: BackgroundTaskService = Depends(get_background_task_service), +) -> TaskExecutionDetail: + detail = await service.get(execution_id) + if detail is None: + raise HTTPException(status_code=404, detail="Task execution not found") + return detail + + +@router.post( + "/executions/{execution_id}/retry", + response_model=TaskExecutionDetail, + dependencies=[Depends(RequiresPermission(PERM_MANAGE))], +) +async def retry_execution( + execution_id: uuid.UUID, + service: BackgroundTaskService = Depends(get_background_task_service), +) -> TaskExecutionDetail: + return await service.retry(execution_id) diff --git a/modules/background_tasks/background_tasks/endpoints/views.py b/modules/background_tasks/background_tasks/endpoints/views.py new file mode 100644 index 00000000..6bfd7070 --- /dev/null +++ b/modules/background_tasks/background_tasks/endpoints/views.py @@ -0,0 +1,71 @@ +"""Inertia view endpoints for the BackgroundTasks admin UI.""" + +from __future__ import annotations + +import uuid + +from fastapi import APIRouter, Depends, HTTPException, Query +from inertia import InertiaResponse +from simple_module_hosting.inertia_deps import InertiaDep +from simple_module_hosting.permissions import RequiresPermission + +from background_tasks.constants import ( + PERM_VIEW, + TaskStatus, +) +from background_tasks.deps import get_background_task_service +from background_tasks.service import BackgroundTaskService + +router = APIRouter(dependencies=[Depends(RequiresPermission(PERM_VIEW))]) + +PER_PAGE = 20 + + +@router.get("/", response_model=None) +async def index( + inertia: InertiaDep, + status: TaskStatus | None = Query(default=None), + task_name: str = Query(default="", alias="q"), + page: int = Query(default=1, ge=1), + service: BackgroundTaskService = Depends(get_background_task_service), +) -> InertiaResponse: + response = await service.list( + status=status, + task_name=task_name or None, + page=page, + per_page=PER_PAGE, + ) + return await inertia.render( + "BackgroundTasks/Index", + { + "executions": [i.model_dump(mode="json") for i in response.items], + "pagination": { + "page": response.page, + "per_page": response.per_page, + "total": response.total, + }, + "filters": { + "status": status.value if status else "", + "task_name": task_name, + }, + }, + ) + + +@router.get("/{execution_id}", response_model=None) +async def detail( + execution_id: str, + inertia: InertiaDep, + service: BackgroundTaskService = Depends(get_background_task_service), +) -> InertiaResponse: + try: + eid = uuid.UUID(execution_id) + except ValueError as exc: + raise HTTPException(status_code=404) from exc + row = await service.get(eid) + if row is None: + raise HTTPException(status_code=404) + return await inertia.render( + "BackgroundTasks/Detail", + {"execution": row.model_dump(mode="json")}, + ) diff --git a/modules/background_tasks/background_tasks/locales/en.json b/modules/background_tasks/background_tasks/locales/en.json new file mode 100644 index 00000000..d945b15d --- /dev/null +++ b/modules/background_tasks/background_tasks/locales/en.json @@ -0,0 +1,57 @@ +{ + "index": { + "title": "Background Tasks", + "description": "Monitor task executions and retry failed or stuck jobs.", + "search_placeholder": "Search by task name…", + "count_one": "{count} task", + "count_other": "{count} tasks", + "empty_title": "No task executions yet", + "empty_description": "Tasks appear here as soon as modules enqueue work.", + "no_match": "No tasks match \"{query}\"" + }, + "filters": { + "status_label": "Status", + "status_all": "All statuses" + }, + "status": { + "pending": "Pending", + "running": "Running", + "success": "Success", + "failed": "Failed", + "stuck": "Stuck", + "revoked": "Revoked", + "retrying": "Retrying" + }, + "table": { + "task": "Task", + "status": "Status", + "queue": "Queue", + "queued_at": "Queued", + "duration": "Duration", + "worker": "Worker", + "actions": "Actions", + "view": "View", + "retry": "Retry" + }, + "detail": { + "title": "Task execution", + "back_button": "Back to tasks", + "args": "Arguments", + "kwargs": "Keyword arguments", + "result": "Result", + "traceback": "Traceback", + "meta": "Details", + "retried_from": "Retried from", + "no_traceback": "No traceback recorded." + }, + "retry_dialog": { + "title": "Retry this task?", + "description": "A new task execution will be enqueued with the same arguments. The original row is kept for history.", + "cancel": "Cancel", + "confirm": "Retry task" + }, + "toasts": { + "retried": "Task \"{name}\" re-enqueued", + "retry_failed": "Failed to retry task" + } +} diff --git a/modules/background_tasks/background_tasks/models.py b/modules/background_tasks/background_tasks/models.py new file mode 100644 index 00000000..259a68fb --- /dev/null +++ b/modules/background_tasks/background_tasks/models.py @@ -0,0 +1,86 @@ +"""SQLModel tables for the BackgroundTasks module.""" + +from __future__ import annotations + +import uuid +from datetime import datetime +from typing import Any + +from simple_module_db.base import create_module_base +from simple_module_db.mixins import AuditMixin +from sqlalchemy import JSON, Column, Index, String +from sqlmodel import Field + +from background_tasks.constants import ( + DEFAULT_QUEUE, + MODULE_NAME, + TABLE_TASK_EXECUTION, + TaskStatus, +) + +# Provider is auto-detected from SM_DATABASE_URL (falls back to SQLite). +# On PostgreSQL this gives the module its own `background_tasks` schema; on +# SQLite all modules share one schema, so __tablename__ is prefixed for +# isolation. +Base = create_module_base(MODULE_NAME) + + +class TaskExecution(Base, AuditMixin, table=True): # ty: ignore[unsupported-base] + """Persistent record of a single Celery task execution. + + We keep our own row (separate from Celery's volatile result backend) + because the admin UI needs stable history, retried-from chains, and a + "stuck since" timestamp that Celery's state machine doesn't model. + """ + + __tablename__ = TABLE_TASK_EXECUTION + + id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) + + # Celery's own UUID for the running task. Nullable because we create the + # row *before* enqueue in the retry flow, then stamp it on send. + celery_task_id: str | None = Field(default=None, index=True, max_length=64) + + task_name: str = Field(index=True, max_length=255) + # Stored as a plain string (the enum's ``value``) rather than a native + # SQL ENUM so autogenerate produces a VARCHAR column and Postgres + # doesn't own a ``taskstatus`` type that we'd need a separate migration + # to evolve. + status: TaskStatus = Field( + default=TaskStatus.PENDING, + sa_column=Column(String(20), index=True, nullable=False, default=TaskStatus.PENDING.value), + ) + queue: str = Field(default=DEFAULT_QUEUE, max_length=64) + + args: list[Any] = Field(default_factory=list, sa_column=Column(JSON)) + kwargs: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + result: dict[str, Any] | None = Field(default=None, sa_column=Column(JSON)) + + traceback: str | None = None + exception_type: str | None = Field(default=None, max_length=255) + + worker: str | None = Field(default=None, max_length=255) + retries: int = 0 + + # Self-reference so the UI can show "retried from " chains. + retried_from_id: uuid.UUID | None = Field( + default=None, + foreign_key=f"{TABLE_TASK_EXECUTION}.id", + index=True, + ) + + queued_at: datetime | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + # Heartbeat is refreshed on every signal that touches a running task. + # `stuck_after_seconds` older than ``now`` ⇒ `sweep_stuck_tasks` flips + # the row to `stuck`. + heartbeat_at: datetime | None = None + + __table_args__ = ( + Index( + "ix_background_tasks_task_execution_status_queued", + "status", + "queued_at", + ), + ) diff --git a/modules/background_tasks/background_tasks/module.py b/modules/background_tasks/background_tasks/module.py new file mode 100644 index 00000000..1621a94b --- /dev/null +++ b/modules/background_tasks/background_tasks/module.py @@ -0,0 +1,96 @@ +"""BackgroundTasks module definition.""" + +from __future__ import annotations + +import importlib.resources +import logging +from pathlib import Path +from typing import TYPE_CHECKING + +from fastapi import APIRouter +from simple_module_core.menu import MenuItem, MenuRegistry, MenuSection +from simple_module_core.module import ModuleBase, ModuleMeta +from simple_module_core.permissions import PermissionRegistry + +from background_tasks.constants import ( + API_PREFIX, + MENU_ICON, + MENU_LABEL, + MENU_ORDER, + MODULE_DISPLAY_NAME, + MODULE_NAME, + PERM_GROUP, + PERM_MANAGE, + PERM_VIEW, + VIEW_PREFIX, +) + +if TYPE_CHECKING: + from fastapi import FastAPI + + +logger = logging.getLogger(__name__) + + +class BackgroundTasksModule(ModuleBase): + """Celery + Redis task queue with an admin UI for retrying failed/stuck tasks.""" + + meta = ModuleMeta( + name=MODULE_DISPLAY_NAME, + route_prefix=API_PREFIX, + view_prefix=VIEW_PREFIX, + depends_on=["Users"], + ) + + def register_settings(self, app: FastAPI) -> None: + from background_tasks.services import BackgroundTasksServices + from background_tasks.settings import BackgroundTasksSettings + + services = BackgroundTasksServices(settings=BackgroundTasksSettings()) + app.state.background_tasks = services + + def register_permissions(self, registry: PermissionRegistry) -> None: + registry.add_group(PERM_GROUP, [PERM_VIEW, PERM_MANAGE]) + + def register_menu_items(self, registry: MenuRegistry) -> None: + registry.add( + MenuItem( + label=MENU_LABEL, + url=VIEW_PREFIX, + icon=MENU_ICON, + order=MENU_ORDER, + section=MenuSection.SIDEBAR, + roles=["admin"], + ) + ) + + def register_routes(self, api_router: APIRouter, view_router: APIRouter) -> None: + from background_tasks.endpoints.api_admin import router as api_admin + from background_tasks.endpoints.views import router as views + + api_router.include_router(api_admin) + view_router.include_router(views) + + def locale_dirs(self) -> dict[str, Path]: + base = Path(str(importlib.resources.files(__package__) / "locales")) + return {MODULE_NAME: base} + + async def on_startup(self, app: FastAPI) -> None: + """Build the Celery app, install signal handlers, hand-off to worker.""" + from background_tasks.celery_app import build_celery + + services = app.state.background_tasks + # build_celery imports `signals` for side effects and runs + # `autodiscover_tasks` across every installed module. + services.celery = build_celery(services.settings) + logger.info( + "BackgroundTasks: Celery app ready (broker=%s, queue=%s)", + services.settings.broker_url, + services.settings.task_default_queue, + ) + + async def on_shutdown(self, app: FastAPI) -> None: + services = getattr(app.state, "background_tasks", None) + celery = getattr(services, "celery", None) if services else None + if celery is not None: + celery.close() diff --git a/modules/background_tasks/background_tasks/pages/Detail.tsx b/modules/background_tasks/background_tasks/pages/Detail.tsx new file mode 100644 index 00000000..fb2dd7d1 --- /dev/null +++ b/modules/background_tasks/background_tasks/pages/Detail.tsx @@ -0,0 +1,215 @@ +import { Link, router, usePage } from '@inertiajs/react'; +import { PageShell } from '@simple-module/ui/components/PageShell'; +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, + AlertDialogTrigger, +} from '@simple-module/ui/components/ui/alert-dialog'; +import { Badge } from '@simple-module/ui/components/ui/badge'; +import { Button } from '@simple-module/ui/components/ui/button'; +import { Card } from '@simple-module/ui/components/ui/card'; +import { usePermissions } from '@simple-module/ui/hooks/use-permissions'; +import { AuthenticatedLayout } from '@simple-module/ui/layouts/AuthenticatedLayout'; +import { fetchWithCsrf } from '@simple-module/ui/lib/csrf'; +import { ArrowLeft, RefreshCcw } from 'lucide-react'; +import { toast } from 'sonner'; +import { + API_BASE, + RETRYABLE_STATUSES, + STATUS_BADGE_VARIANT, + type TaskStatus, + VIEW_BASE, +} from './constants'; + +interface Execution { + id: string; + celery_task_id: string | null; + task_name: string; + status: TaskStatus; + queue: string; + args: unknown[]; + kwargs: Record; + result: Record | null; + traceback: string | null; + exception_type: string | null; + worker: string | null; + retries: number; + retried_from_id: string | null; + queued_at: string | null; + started_at: string | null; + finished_at: string | null; + heartbeat_at: string | null; +} + +interface Props { + execution: Execution; +} + +function statusLabel(status: TaskStatus): string { + return status[0].toUpperCase() + status.slice(1); +} + +function pretty(value: unknown): string { + if (value === null || value === undefined) return '—'; + try { + return JSON.stringify(value, null, 2); + } catch { + return String(value); + } +} + +function formatTs(ts: string | null): string { + return ts ? new Date(ts).toLocaleString() : '—'; +} + +function Detail() { + const { execution } = usePage<{ props: Props }>().props as unknown as Props; + const { can } = usePermissions(); + const canRetry = can('background_tasks.manage'); + const retryable = RETRYABLE_STATUSES.has(execution.status) && canRetry; + + async function handleRetry() { + try { + const res = await fetchWithCsrf(`${API_BASE}/executions/${execution.id}/retry`, { + method: 'POST', + }); + if (!res.ok) { + const body = await res.json().catch(() => ({})); + throw new Error(body.detail || `HTTP ${res.status}`); + } + const created = (await res.json()) as Execution; + toast.success(`Task "${execution.task_name}" re-enqueued`); + router.visit(`${VIEW_BASE}/${created.id}`); + } catch (err) { + toast.error(err instanceof Error ? err.message : 'Failed to retry task'); + } + } + + return ( + + + {retryable && ( + + + + + + + Retry this task? + + A new task execution will be enqueued with the same arguments. + + + + Cancel + Retry task + + + + )} + + } + > +
+ +

Details

+
+
+
Status
+
+ + {statusLabel(execution.status)} + +
+
+ + + + + + + + + + {execution.retried_from_id && ( +
+
Retried from
+
+ + {execution.retried_from_id.slice(0, 8)}… + +
+
+ )} +
+ + +
+ +

Arguments

+ 
+              {pretty(execution.args)}
+
+ + +

Keyword arguments

+ 
+              {pretty(execution.kwargs)}
+
+ + {execution.result !== null && ( + +

Result

+ 
+                {pretty(execution.result)}
+
+ + )} + +

Traceback

+ {execution.traceback ? ( + 
+                {execution.traceback}
+
+ ) : ( +

No traceback recorded.

+ )} + +
+
+ + ); +} + +function Row({ label, value, mono }: { label: string; value: string; mono?: boolean }) { + return ( +
+
{label}
+
{value}
+
+ ); +} + +Detail.layout = (page: React.ReactNode) => {page}; +export default Detail; diff --git a/modules/background_tasks/background_tasks/pages/Index.tsx b/modules/background_tasks/background_tasks/pages/Index.tsx new file mode 100644 index 00000000..2fa4a1d3 --- /dev/null +++ b/modules/background_tasks/background_tasks/pages/Index.tsx @@ -0,0 +1,202 @@ +import { router, usePage } from '@inertiajs/react'; +import { PageShell } from '@simple-module/ui/components/PageShell'; +import { Button } from '@simple-module/ui/components/ui/button'; +import { Card } from '@simple-module/ui/components/ui/card'; +import { Input } from '@simple-module/ui/components/ui/input'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '@simple-module/ui/components/ui/select'; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from '@simple-module/ui/components/ui/table'; +import { usePermissions } from '@simple-module/ui/hooks/use-permissions'; +import { AuthenticatedLayout } from '@simple-module/ui/layouts/AuthenticatedLayout'; +import { fetchWithCsrf } from '@simple-module/ui/lib/csrf'; +import { Activity, Search } from 'lucide-react'; +import { useCallback, useEffect, useMemo, useState } from 'react'; +import { toast } from 'sonner'; +import { type Execution, ExecutionRow, statusLabel } from './components/ExecutionRow'; +import { API_BASE, STATUS_ORDER, VIEW_BASE } from './constants'; + +interface Pagination { + page: number; + per_page: number; + total: number; +} + +interface Filters { + status: string; + task_name: string; +} + +interface Props { + executions: Execution[]; + pagination: Pagination; + filters: Filters; +} + +const STATUS_ALL = '__all__'; + +function Index() { + const { + executions, + pagination, + filters: initialFilters, + } = usePage<{ props: Props }>().props as unknown as Props; + + const { can } = usePermissions(); + const canRetry = can('background_tasks.manage'); + + const [search, setSearch] = useState(initialFilters.task_name ?? ''); + const [status, setStatus] = useState(initialFilters.status || STATUS_ALL); + + const totalPages = useMemo( + () => Math.ceil(pagination.total / pagination.per_page), + [pagination.total, pagination.per_page], + ); + + const navigate = useCallback( + (opts: { page?: number; q?: string; status?: string }) => { + const params: Record = {}; + const q = opts.q ?? search; + const s = opts.status ?? status; + const page = opts.page ?? 1; + if (q) params.q = q; + if (s && s !== STATUS_ALL) params.status = s; + if (page > 1) params.page = String(page); + router.get(VIEW_BASE, params, { preserveState: true, preserveScroll: true }); + }, + [search, status], + ); + + useEffect(() => { + if (search === (initialFilters.task_name ?? '')) return; + const timeout = setTimeout(() => navigate({ q: search }), 300); + return () => clearTimeout(timeout); + }, [search, initialFilters.task_name, navigate]); + + async function handleRetry(execution: Execution) { + try { + const res = await fetchWithCsrf(`${API_BASE}/executions/${execution.id}/retry`, { + method: 'POST', + }); + if (!res.ok) { + const body = await res.json().catch(() => ({})); + throw new Error(body.detail || `HTTP ${res.status}`); + } + toast.success(`Task "${execution.task_name}" re-enqueued`); + router.reload({ only: ['executions', 'pagination'] }); + } catch (err) { + toast.error(err instanceof Error ? err.message : 'Failed to retry task'); + } + } + + return ( + +
+
+ + setSearch(e.target.value)} + className="pl-9" + /> +
+ +
+ + + + + + Task + Status + Queue + Queued + Duration + Worker + Actions + + + + {executions.map((e) => ( + + ))} + {executions.length === 0 && ( + + +
+ +

+ {search + ? `No tasks match "${search}"` + : 'No task executions yet. Tasks appear here as soon as modules enqueue work.'} +

+
+ + + )} + +
+ + + {totalPages > 1 && ( +
+ + + Page {pagination.page} of {totalPages} + + +
+ )} + + ); +} + +Index.layout = (page: React.ReactNode) => {page}; +export default Index; diff --git a/modules/background_tasks/background_tasks/pages/components/ExecutionRow.tsx b/modules/background_tasks/background_tasks/pages/components/ExecutionRow.tsx new file mode 100644 index 00000000..1bbbbd08 --- /dev/null +++ b/modules/background_tasks/background_tasks/pages/components/ExecutionRow.tsx @@ -0,0 +1,115 @@ +import { Link } from '@inertiajs/react'; +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, + AlertDialogTrigger, +} from '@simple-module/ui/components/ui/alert-dialog'; +import { Badge } from '@simple-module/ui/components/ui/badge'; +import { Button } from '@simple-module/ui/components/ui/button'; +import { TableCell, TableRow } from '@simple-module/ui/components/ui/table'; +import { RefreshCcw } from 'lucide-react'; +import { RETRYABLE_STATUSES, STATUS_BADGE_VARIANT, type TaskStatus, VIEW_BASE } from '../constants'; + +export interface Execution { + id: string; + celery_task_id: string | null; + task_name: string; + status: TaskStatus; + queue: string; + retries: number; + worker: string | null; + queued_at: string | null; + started_at: string | null; + finished_at: string | null; + exception_type: string | null; + retried_from_id: string | null; +} + +function statusLabel(status: TaskStatus): string { + return status[0].toUpperCase() + status.slice(1); +} + +function durationMs(started: string | null, finished: string | null): string { + if (!started) return '—'; + const end = finished ? new Date(finished) : new Date(); + const ms = end.getTime() - new Date(started).getTime(); + if (ms < 0) return '—'; + if (ms < 1000) return `${ms}ms`; + if (ms < 60_000) return `${(ms / 1000).toFixed(1)}s`; + return `${Math.floor(ms / 60_000)}m ${Math.floor((ms % 60_000) / 1000)}s`; +} + +interface Props { + execution: Execution; + canRetry: boolean; + onRetry: (execution: Execution) => void; +} + +export function ExecutionRow({ execution, canRetry, onRetry }: Props) { + const retryable = RETRYABLE_STATUSES.has(execution.status) && canRetry; + return ( + + +
+ + {execution.task_name} + + {execution.exception_type && ( + {execution.exception_type} + )} +
+ + + + {statusLabel(execution.status)} + + + + {execution.queue} + + + {execution.queued_at ? new Date(execution.queued_at).toLocaleString() : '—'} + + + {durationMs(execution.started_at, execution.finished_at)} + + + {execution.worker || '—'} + + + {retryable ? ( + + + + + + + Retry this task? + + A new task execution will be enqueued with the same arguments. The original row is + kept for history. + + + + Cancel + onRetry(execution)}>Retry task + + + + ) : ( + — + )} + + + ); +} + +export { statusLabel }; diff --git a/modules/background_tasks/background_tasks/pages/constants.ts b/modules/background_tasks/background_tasks/pages/constants.ts new file mode 100644 index 00000000..a5af3b66 --- /dev/null +++ b/modules/background_tasks/background_tasks/pages/constants.ts @@ -0,0 +1,49 @@ +// Frontend mirror of background_tasks/constants.py. +// Kept small and hand-maintained — the backend file is the source of truth. + +export const API_BASE = '/api/background_tasks/admin'; +export const VIEW_BASE = '/admin/background-tasks'; + +export const TASK_STATUS = { + PENDING: 'pending', + RUNNING: 'running', + SUCCESS: 'success', + FAILED: 'failed', + STUCK: 'stuck', + REVOKED: 'revoked', + RETRYING: 'retrying', +} as const; + +export type TaskStatus = (typeof TASK_STATUS)[keyof typeof TASK_STATUS]; + +export const RETRYABLE_STATUSES: ReadonlySet = new Set([ + TASK_STATUS.FAILED, + TASK_STATUS.STUCK, +]); + +export const TERMINAL_STATUSES: ReadonlySet = new Set([ + TASK_STATUS.SUCCESS, + TASK_STATUS.FAILED, + TASK_STATUS.STUCK, + TASK_STATUS.REVOKED, +]); + +export const STATUS_BADGE_VARIANT: Record = { + [TASK_STATUS.PENDING]: 'outline', + [TASK_STATUS.RUNNING]: 'secondary', + [TASK_STATUS.SUCCESS]: 'secondary', + [TASK_STATUS.FAILED]: 'destructive', + [TASK_STATUS.STUCK]: 'destructive', + [TASK_STATUS.REVOKED]: 'outline', + [TASK_STATUS.RETRYING]: 'outline', +}; + +export const STATUS_ORDER: TaskStatus[] = [ + TASK_STATUS.PENDING, + TASK_STATUS.RUNNING, + TASK_STATUS.RETRYING, + TASK_STATUS.SUCCESS, + TASK_STATUS.FAILED, + TASK_STATUS.STUCK, + TASK_STATUS.REVOKED, +]; diff --git a/modules/background_tasks/background_tasks/py.typed b/modules/background_tasks/background_tasks/py.typed new file mode 100644 index 00000000..e69de29b diff --git a/modules/background_tasks/background_tasks/service.py b/modules/background_tasks/background_tasks/service.py new file mode 100644 index 00000000..7c3c4170 --- /dev/null +++ b/modules/background_tasks/background_tasks/service.py @@ -0,0 +1,156 @@ +"""BackgroundTaskService — admin listing, detail, and retry.""" + +from __future__ import annotations + +import uuid +from typing import TYPE_CHECKING + +from fastapi import HTTPException, status +from simple_module_core.events import EventBus +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from background_tasks.constants import ( + RETRYABLE_STATUSES, + TaskStatus, +) +from background_tasks.contracts.events import TaskRetried +from background_tasks.contracts.schemas import ( + TaskExecutionDetail, + TaskExecutionListItem, + TaskExecutionListResponse, +) +from background_tasks.models import TaskExecution + +if TYPE_CHECKING: + from celery import Celery + + +class BackgroundTaskService: + """List, fetch, and retry task executions. + + The service treats the DB as the system of record and the Celery app + as pure transport — ``retry`` loads the original row, enqueues a new + Celery task with the same args/kwargs, and inserts a fresh + ``TaskExecution`` linked to the original via ``retried_from_id``. + """ + + def __init__( + self, + db: AsyncSession, + celery: Celery | None, + event_bus: EventBus, + ) -> None: + self.db = db + self.celery = celery + self.event_bus = event_bus + + async def list( + self, + *, + status: TaskStatus | None = None, + task_name: str | None = None, + page: int = 1, + per_page: int = 20, + ) -> TaskExecutionListResponse: + """Return a paginated listing, newest-first, with optional filters.""" + page = max(page, 1) + per_page = max(1, min(per_page, 200)) + + filters = [] + if status is not None: + filters.append(TaskExecution.status == status) + if task_name: + filters.append(TaskExecution.task_name.ilike(f"%{task_name}%")) + + base = select(TaskExecution) + for f in filters: + base = base.where(f) + + count_query = select(func.count()).select_from(TaskExecution) + for f in filters: + count_query = count_query.where(f) + total = (await self.db.execute(count_query)).scalar() or 0 + + query = ( + base.order_by(TaskExecution.queued_at.desc().nulls_last()) + .offset((page - 1) * per_page) + .limit(per_page) + ) + rows = (await self.db.execute(query)).scalars().all() + + return TaskExecutionListResponse( + items=[TaskExecutionListItem.model_validate(r) for r in rows], + total=total, + page=page, + per_page=per_page, + status=status, + task_name=task_name, + ) + + async def get(self, execution_id: uuid.UUID) -> TaskExecutionDetail | None: + row = await self.db.get(TaskExecution, execution_id) + if row is None: + return None + return TaskExecutionDetail.model_validate(row) + + async def retry(self, execution_id: uuid.UUID) -> TaskExecutionDetail: + """Re-enqueue a failed or stuck task. + + The original row is immutable; we insert a new ``TaskExecution`` row + carrying ``retried_from_id`` so the detail page can show the chain. + """ + row = await self.db.get(TaskExecution, execution_id) + if row is None: + raise HTTPException(status_code=404, detail="Task execution not found") + + # ``row.status`` is a plain string when loaded from SQLite (the + # column is VARCHAR), a TaskStatus when set via the model. StrEnum + # compares equal to its string value either way. + if row.status not in RETRYABLE_STATUSES: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=( + f"Task execution status is {str(row.status)!r}; " + "only failed or stuck tasks can be retried." + ), + ) + + if self.celery is None: + # Only happens if on_startup hasn't run — e.g. in a test that + # instantiates the service without wiring Celery. Fail loudly + # rather than silently swallowing the retry. + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Celery is not initialized; cannot enqueue retry.", + ) + + async_result = self.celery.send_task( + row.task_name, + args=list(row.args or []), + kwargs=dict(row.kwargs or {}), + queue=row.queue, + ) + + new_row = TaskExecution( + celery_task_id=async_result.id, + task_name=row.task_name, + status=TaskStatus.PENDING, + queue=row.queue, + args=list(row.args or []), + kwargs=dict(row.kwargs or {}), + retried_from_id=row.id, + ) + self.db.add(new_row) + await self.db.flush() + await self.db.refresh(new_row) + + await self.event_bus.publish( + TaskRetried( + original_id=row.id, + new_id=new_row.id, + task_name=row.task_name, + ) + ) + + return TaskExecutionDetail.model_validate(new_row) diff --git a/modules/background_tasks/background_tasks/services.py b/modules/background_tasks/background_tasks/services.py new file mode 100644 index 00000000..316a92fa --- /dev/null +++ b/modules/background_tasks/background_tasks/services.py @@ -0,0 +1,25 @@ +"""Module-scoped state container for background_tasks. + +Stored as ``app.state.background_tasks`` by +:meth:`BackgroundTasksModule.register_settings`. Not frozen — ``on_startup`` +populates :attr:`celery` once the Celery app is built. Convention: set once +during boot, treat as read-only after. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from celery import Celery + + from background_tasks.settings import BackgroundTasksSettings + + +@dataclass +class BackgroundTasksServices: + """BackgroundTasks singletons. Single slot at ``app.state.background_tasks``.""" + + settings: BackgroundTasksSettings + celery: Celery | None = None diff --git a/modules/background_tasks/background_tasks/settings.py b/modules/background_tasks/background_tasks/settings.py new file mode 100644 index 00000000..07521744 --- /dev/null +++ b/modules/background_tasks/background_tasks/settings.py @@ -0,0 +1,66 @@ +"""BackgroundTasks module settings loaded from SM_BG_TASKS_* env vars.""" + +from __future__ import annotations + +import os + +from pydantic import model_validator +from pydantic_settings import BaseSettings, SettingsConfigDict +from simple_module_core.environments import NON_PROD_ENVIRONMENTS + +from background_tasks.constants import ( + DEFAULT_BROKER_URL, + DEFAULT_MAX_RETRIES, + DEFAULT_PURGE_INTERVAL_SECONDS, + DEFAULT_QUEUE, + DEFAULT_RESULT_BACKEND, + DEFAULT_RETENTION_DAYS, + DEFAULT_STUCK_AFTER_SECONDS, + DEFAULT_STUCK_SWEEP_INTERVAL_SECONDS, + ENV_PREFIX, +) + + +class BackgroundTasksSettings(BaseSettings): + """Configuration for the Celery + Redis task runner.""" + + model_config = SettingsConfigDict(env_prefix=ENV_PREFIX, env_file=".env", extra="ignore") + + broker_url: str = DEFAULT_BROKER_URL + result_backend: str = DEFAULT_RESULT_BACKEND + task_default_queue: str = DEFAULT_QUEUE + + # A task that has been ``running`` longer than this without a heartbeat is + # flipped to ``stuck`` by the beat sweep. 5 min is long enough to cover + # normal slow jobs while still surfacing wedged workers within one UI + # refresh. + stuck_after_seconds: int = DEFAULT_STUCK_AFTER_SECONDS + stuck_sweep_interval_seconds: int = DEFAULT_STUCK_SWEEP_INTERVAL_SECONDS + purge_interval_seconds: int = DEFAULT_PURGE_INTERVAL_SECONDS + + retention_days: int = DEFAULT_RETENTION_DAYS + max_retries: int = DEFAULT_MAX_RETRIES + + @model_validator(mode="after") + def _forbid_localhost_broker_in_production(self) -> BackgroundTasksSettings: + """Fail boot if production is still pointed at the dev default broker. + + A localhost broker in prod means the web container is talking to its + own 6379 instead of the shared Redis service — tasks would silently + queue to a broker no worker reads. + """ + env = os.environ.get("SM_ENVIRONMENT", "development") + if env in NON_PROD_ENVIRONMENTS: + return self + bad = [] + if "localhost" in self.broker_url or "127.0.0.1" in self.broker_url: + bad.append(f"{ENV_PREFIX}BROKER_URL") + if "localhost" in self.result_backend or "127.0.0.1" in self.result_backend: + bad.append(f"{ENV_PREFIX}RESULT_BACKEND") + if bad: + names = ", ".join(bad) + raise ValueError( + f"{names} must not point at localhost when SM_ENVIRONMENT={env!r}. " + "Set these to the Redis service host (e.g. redis://redis:6379/0)." + ) + return self diff --git a/modules/background_tasks/background_tasks/signals.py b/modules/background_tasks/background_tasks/signals.py new file mode 100644 index 00000000..1c67f798 --- /dev/null +++ b/modules/background_tasks/background_tasks/signals.py @@ -0,0 +1,232 @@ +"""Celery signal handlers that keep the ``TaskExecution`` table in sync. + +Each signal writes one row: publish creates ``pending``, prerun flips to +``running`` with a heartbeat, postrun/success/failure/retry/revoked update +the terminal columns. We update an existing row (matched by +``celery_task_id``) so a single task's lifecycle stays on one row instead +of spawning a new row per signal. + +Signals are sync — see :mod:`.sync_db` for why we maintain a separate sync +engine, and :mod:`._signal_support` for the shared helpers. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from celery import signals + +from background_tasks._signal_support import ( + coerce_args_kwargs, + jsonable_result, + now_utc, + render_traceback, + task_name_of, + upsert_by_celery_id, +) +from background_tasks.constants import DEFAULT_QUEUE, TaskStatus +from background_tasks.sync_db import sync_session + +logger = logging.getLogger(__name__) + + +# ── Enqueue ───────────────────────────────────────────────────── + + +@signals.before_task_publish.connect +def on_task_publish( + sender: str | None = None, + headers: dict[str, Any] | None = None, + body: Any = None, + routing_key: str | None = None, + **_kwargs: Any, +) -> None: + """Record a row the moment a task is pushed onto the broker.""" + task_id = (headers or {}).get("id") + task_name = sender or (headers or {}).get("task") or "unknown" + + # ``body`` on the publish signal is ``(args, kwargs, options)`` for the + # standard Celery protocol; guard against edge shapes. + args_in, kwargs_in = [], {} + if isinstance(body, list | tuple) and len(body) >= 2: + args_in, kwargs_in = body[0], body[1] + args, kwargs = coerce_args_kwargs(args_in, kwargs_in) + + try: + with sync_session() as session: + upsert_by_celery_id( + session, + celery_task_id=task_id, + task_name=task_name, + defaults={ + "task_name": task_name, + "status": TaskStatus.PENDING, + "queue": routing_key or DEFAULT_QUEUE, + "args": args, + "kwargs": kwargs, + "queued_at": now_utc(), + }, + ) + except Exception: + logger.exception("on_task_publish failed for task_id=%s", task_id) + + +# ── Execution lifecycle ───────────────────────────────────────── + + +@signals.task_prerun.connect +def on_task_prerun( + sender: Any = None, + task_id: str | None = None, + task: Any = None, + args: Any = None, + kwargs: Any = None, + **_k: Any, +) -> None: + """Flip the row to ``running`` and start the heartbeat.""" + task_name = task_name_of(sender, task) + args_n, kwargs_n = coerce_args_kwargs(args, kwargs) + now = now_utc() + try: + with sync_session() as session: + upsert_by_celery_id( + session, + celery_task_id=task_id, + task_name=task_name, + defaults={ + "task_name": task_name, + "status": TaskStatus.RUNNING, + "args": args_n, + "kwargs": kwargs_n, + "started_at": now, + "heartbeat_at": now, + }, + ) + except Exception: + logger.exception("on_task_prerun failed for task_id=%s", task_id) + + +@signals.task_postrun.connect +def on_task_postrun( + sender: Any = None, + task_id: str | None = None, + task: Any = None, + **_k: Any, +) -> None: + """Refresh the heartbeat on normal completion. + + Terminal status is written by ``task_success`` / ``task_failure`` / + ``task_retry`` which fire *before* postrun; postrun only refreshes the + heartbeat so the sweep doesn't immediately flip a just-finished row. + """ + task_name = task_name_of(sender, task) + try: + with sync_session() as session: + upsert_by_celery_id( + session, + celery_task_id=task_id, + task_name=task_name, + defaults={"heartbeat_at": now_utc()}, + ) + except Exception: + logger.exception("on_task_postrun failed for task_id=%s", task_id) + + +@signals.task_success.connect +def on_task_success(sender: Any = None, result: Any = None, **_k: Any) -> None: + task_id = getattr(sender.request, "id", None) if sender is not None else None + task_name = getattr(sender, "name", None) or "unknown" + try: + with sync_session() as session: + upsert_by_celery_id( + session, + celery_task_id=task_id, + task_name=task_name, + defaults={ + "status": TaskStatus.SUCCESS, + "result": jsonable_result(result), + "finished_at": now_utc(), + "traceback": None, + "exception_type": None, + }, + ) + except Exception: + logger.exception("on_task_success failed for task_id=%s", task_id) + + +@signals.task_failure.connect +def on_task_failure( + sender: Any = None, + task_id: str | None = None, + exception: BaseException | None = None, + einfo: Any = None, + **_k: Any, +) -> None: + task_name = getattr(sender, "name", None) or "unknown" + tb = render_traceback(einfo, exception) + exc_type = type(exception).__name__ if exception is not None else None + try: + with sync_session() as session: + upsert_by_celery_id( + session, + celery_task_id=task_id, + task_name=task_name, + defaults={ + "status": TaskStatus.FAILED, + "traceback": tb, + "exception_type": exc_type, + "finished_at": now_utc(), + }, + ) + except Exception: + logger.exception("on_task_failure failed for task_id=%s", task_id) + + +@signals.task_retry.connect +def on_task_retry( + sender: Any = None, + request: Any = None, + reason: Any = None, + **_k: Any, +) -> None: + task_id = getattr(request, "id", None) if request is not None else None + task_name = getattr(sender, "name", None) or "unknown" + retries = int(getattr(request, "retries", 0) or 0) + try: + with sync_session() as session: + row = upsert_by_celery_id( + session, + celery_task_id=task_id, + task_name=task_name, + defaults={ + "status": TaskStatus.RETRYING, + "retries": retries, + "heartbeat_at": now_utc(), + }, + ) + # Preserve the latest reason for the UI without nulling a + # previously-captured traceback. + if reason is not None: + row.traceback = f"retry: {reason!r}" + except Exception: + logger.exception("on_task_retry failed for task_id=%s", task_id) + + +@signals.task_revoked.connect +def on_task_revoked(sender: Any = None, request: Any = None, **_k: Any) -> None: + task_id = getattr(request, "id", None) if request is not None else None + task_name = getattr(sender, "name", None) or "unknown" + try: + with sync_session() as session: + upsert_by_celery_id( + session, + celery_task_id=task_id, + task_name=task_name, + defaults={ + "status": TaskStatus.REVOKED, + "finished_at": now_utc(), + }, + ) + except Exception: + logger.exception("on_task_revoked failed for task_id=%s", task_id) diff --git a/modules/background_tasks/background_tasks/sync_db.py b/modules/background_tasks/background_tasks/sync_db.py new file mode 100644 index 00000000..3c7c9d34 --- /dev/null +++ b/modules/background_tasks/background_tasks/sync_db.py @@ -0,0 +1,68 @@ +"""Sync SQLAlchemy session factory for Celery signal handlers. + +Celery signals are called synchronously from inside the worker / web-process +hot path. Building an event-loop just to await a query is both ugly and +deadlock-prone (web-process signals can fire from inside an already-running +loop). We instead maintain a second, sync engine pointed at the same DB URL +and borrow a short-lived session from it whenever a signal fires. + +The engine is lazily built on first use and process-global. It's cheap — +SQLAlchemy's connection pool is per-process so the signal path reuses +connections after the first signal. +""" + +from __future__ import annotations + +import logging +import os +from collections.abc import Iterator +from contextlib import contextmanager + +from sqlalchemy import create_engine +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Session, sessionmaker + +logger = logging.getLogger(__name__) + +_engine: Engine | None = None +_session_factory: sessionmaker[Session] | None = None + + +def _sync_url(async_url: str) -> str: + """Convert an async SQLAlchemy URL to its sync driver equivalent. + + Mirrors ``host/migrations/env.py`` so signals use the same URL shape + Alembic does. + """ + return async_url.replace("+aiosqlite", "").replace("+asyncpg", "+psycopg2") + + +def _build_engine() -> Engine: + url = os.environ.get("SM_DATABASE_URL", "sqlite:///./app.db") + sync_url = _sync_url(url) + # Small pool — signals fire sequentially per worker process. + return create_engine(sync_url, pool_pre_ping=True, pool_size=2, max_overflow=3) + + +def get_sync_session_factory() -> sessionmaker[Session]: + """Return the process-global sync session factory, building it once.""" + global _engine, _session_factory + if _session_factory is None: + _engine = _build_engine() + _session_factory = sessionmaker(bind=_engine, expire_on_commit=False) + return _session_factory + + +@contextmanager +def sync_session() -> Iterator[Session]: + """Open a short-lived sync session; commit on success, rollback on error.""" + factory = get_sync_session_factory() + session = factory() + try: + yield session + session.commit() + except Exception: + session.rollback() + raise + finally: + session.close() diff --git a/modules/background_tasks/background_tasks/tasks.py b/modules/background_tasks/background_tasks/tasks.py new file mode 100644 index 00000000..99d8dd94 --- /dev/null +++ b/modules/background_tasks/background_tasks/tasks.py @@ -0,0 +1,102 @@ +"""Internal Celery tasks registered by the background_tasks module itself. + +- ``sweep_stuck_tasks`` flips ``running`` rows whose heartbeat has gone stale + to ``stuck`` so the UI surfaces them and they become eligible for retry. +- ``purge_old_executions`` deletes terminal rows older than the configured + retention so the history table stays bounded. + +Both are driven by Celery beat; see :mod:`.celery_app`. +""" + +from __future__ import annotations + +import logging +import os +from datetime import UTC, datetime, timedelta + +from celery import shared_task +from sqlalchemy import delete, select, update + +from background_tasks.constants import ( + DEFAULT_RETENTION_DAYS, + DEFAULT_STUCK_AFTER_SECONDS, + INTERNAL_TASK_PURGE_OLD, + INTERNAL_TASK_SWEEP_STUCK, + TERMINAL_STATUSES, + TaskStatus, +) +from background_tasks.models import TaskExecution +from background_tasks.settings import BackgroundTasksSettings +from background_tasks.sync_db import sync_session + +logger = logging.getLogger(__name__) + + +def _load_settings() -> BackgroundTasksSettings: + """Build settings inside the worker process. + + The web-process ``register_settings`` attached them to ``app.state``, + but the worker never sees that ``app.state``. Re-parsing from env here + is the same deterministic source of truth. + """ + return BackgroundTasksSettings() + + +@shared_task(name=INTERNAL_TASK_SWEEP_STUCK) +def sweep_stuck_tasks() -> int: + """Flip ``running`` rows with stale heartbeats to ``stuck``. + + Returns the number of rows flipped — handy for metrics/log scraping. + """ + # Avoid import-time coupling to settings (beat schedules this task + # before the worker has hydrated settings in some edge cases). + stuck_after = int( + os.environ.get("SM_BG_TASKS_STUCK_AFTER_SECONDS", DEFAULT_STUCK_AFTER_SECONDS) + ) + cutoff = datetime.now(UTC) - timedelta(seconds=stuck_after) + + with sync_session() as session: + stmt = ( + update(TaskExecution) + .where( + TaskExecution.status == TaskStatus.RUNNING, + TaskExecution.heartbeat_at < cutoff, + ) + .values(status=TaskStatus.STUCK, finished_at=datetime.now(UTC)) + ) + result = session.execute(stmt) + count = result.rowcount or 0 + if count: + logger.warning("sweep_stuck_tasks flipped %d row(s) to stuck", count) + return count + + +@shared_task(name=INTERNAL_TASK_PURGE_OLD) +def purge_old_executions() -> int: + """Delete terminal rows older than the retention window.""" + retention_days = int(os.environ.get("SM_BG_TASKS_RETENTION_DAYS", DEFAULT_RETENTION_DAYS)) + cutoff = datetime.now(UTC) - timedelta(days=retention_days) + + with sync_session() as session: + # Read first so the log line is accurate; the delete is one + # statement either way. + to_delete = ( + session.execute( + select(TaskExecution.id) + .where(TaskExecution.status.in_(list(TERMINAL_STATUSES))) + .where(TaskExecution.finished_at < cutoff) + ) + .scalars() + .all() + ) + + if not to_delete: + return 0 + + session.execute(delete(TaskExecution).where(TaskExecution.id.in_(to_delete))) + logger.info( + "purge_old_executions removed %d row(s) older than %d days", + len(to_delete), + retention_days, + ) + return len(to_delete) diff --git a/modules/background_tasks/package.json b/modules/background_tasks/package.json new file mode 100644 index 00000000..dbe6e873 --- /dev/null +++ b/modules/background_tasks/package.json @@ -0,0 +1,16 @@ +{ + "name": "@simple-module/background-tasks", + "version": "0.1.0", + "private": true, + "description": "Frontend assets for the BackgroundTasks module", + "peerDependencies": { + "react": "^19.0.0", + "react-dom": "^19.0.0", + "@inertiajs/react": "^2.0.0", + "@simple-module/ui": "*" + }, + "devDependencies": { + "@simple-module/tsconfig": "*" + }, + "dependencies": {} +} diff --git a/modules/background_tasks/pyproject.toml b/modules/background_tasks/pyproject.toml new file mode 100644 index 00000000..203f8ba8 --- /dev/null +++ b/modules/background_tasks/pyproject.toml @@ -0,0 +1,25 @@ +[project] +name = "background-tasks" +version = "0.1.0" +description = "Celery + Redis task queue with admin UI for monitoring and retrying failed/stuck tasks" +authors = [] +requires-python = ">=3.12" +dependencies = [ + "simple-module-core", + "simple-module-db", + "simple-module-hosting", + "celery[redis]>=5.4", + "redis>=5", +] + +[project.entry-points.simple_module] +background_tasks = "background_tasks.module:BackgroundTasksModule" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.uv.sources] +simple-module-core = { workspace = true } +simple-module-db = { workspace = true } +simple-module-hosting = { workspace = true } diff --git a/modules/background_tasks/tests/test_admin_api.py b/modules/background_tasks/tests/test_admin_api.py new file mode 100644 index 00000000..011739ea --- /dev/null +++ b/modules/background_tasks/tests/test_admin_api.py @@ -0,0 +1,107 @@ +"""End-to-end tests for /api/background_tasks/admin/* using the full app fixture.""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime +from unittest.mock import MagicMock + +import httpx +import pytest +from background_tasks.constants import TaskStatus +from background_tasks.models import TaskExecution + +ADMIN_BASE = "/api/background_tasks/admin" + + +@pytest.fixture(autouse=True) +async def _stub_celery(app) -> None: + """Replace the real Celery instance with a MagicMock for the admin suite. + + ``BackgroundTasksModule.on_startup`` builds a live Celery app which tries + to talk to a real Redis broker on the first ``send_task`` call. The admin + tests only care that ``retry`` flows through the API — mocking keeps the + test hermetic. + """ + celery = MagicMock(name="Celery") + celery.send_task.return_value.id = "mocked-celery-id" + app.state.background_tasks.celery = celery + + +async def _seed_failed(app, **overrides) -> TaskExecution: + row = TaskExecution( + celery_task_id=str(uuid.uuid4()), + task_name=overrides.get("task_name", "demo.failed"), + status=overrides.get("status", TaskStatus.FAILED), + queue="default", + args=overrides.get("args", []), + kwargs=overrides.get("kwargs", {}), + queued_at=datetime.now(UTC), + ) + async with app.state.sm.db.session_factory() as session: + session.add(row) + await session.commit() + await session.refresh(row) + return row + + +class TestListExecutions: + async def test_unauthenticated_request_is_rejected(self, client: httpx.AsyncClient): + resp = await client.get(f"{ADMIN_BASE}/executions", follow_redirects=False) + # The middleware redirects unauthenticated API callers to the login page. + assert resp.status_code in {302, 401, 403} + + async def test_admin_sees_paginated_list(self, app, authenticated_client: httpx.AsyncClient): + await _seed_failed(app, task_name="demo.a") + await _seed_failed(app, task_name="demo.b", status=TaskStatus.SUCCESS) + + resp = await authenticated_client.get(f"{ADMIN_BASE}/executions") + assert resp.status_code == 200 + body = resp.json() + assert body["total"] == 2 + assert {i["task_name"] for i in body["items"]} == {"demo.a", "demo.b"} + + async def test_status_filter_narrows_results( + self, app, authenticated_client: httpx.AsyncClient + ): + await _seed_failed(app, task_name="demo.ok", status=TaskStatus.SUCCESS) + await _seed_failed(app, task_name="demo.bad", status=TaskStatus.FAILED) + + resp = await authenticated_client.get( + f"{ADMIN_BASE}/executions", params={"status": "failed"} + ) + assert resp.status_code == 200 + names = [i["task_name"] for i in resp.json()["items"]] + assert names == ["demo.bad"] + + +class TestGetExecution: + async def test_returns_detail(self, app, authenticated_client: httpx.AsyncClient): + row = await _seed_failed(app) + + resp = await authenticated_client.get(f"{ADMIN_BASE}/executions/{row.id}") + assert resp.status_code == 200 + assert resp.json()["task_name"] == "demo.failed" + + async def test_missing_returns_404(self, authenticated_client: httpx.AsyncClient): + resp = await authenticated_client.get(f"{ADMIN_BASE}/executions/{uuid.uuid4()}") + assert resp.status_code == 404 + + +class TestRetryExecution: + async def test_retry_failed_creates_new_row(self, app, authenticated_client: httpx.AsyncClient): + original = await _seed_failed(app, args=[7], kwargs={"mode": "fast"}) + + resp = await authenticated_client.post(f"{ADMIN_BASE}/executions/{original.id}/retry") + assert resp.status_code == 200 + body = resp.json() + assert body["retried_from_id"] == str(original.id) + assert body["status"] == TaskStatus.PENDING.value + app.state.background_tasks.celery.send_task.assert_called_once() + + async def test_retry_success_row_is_rejected( + self, app, authenticated_client: httpx.AsyncClient + ): + row = await _seed_failed(app, status=TaskStatus.SUCCESS) + resp = await authenticated_client.post(f"{ADMIN_BASE}/executions/{row.id}/retry") + assert resp.status_code == 409 diff --git a/modules/background_tasks/tests/test_service.py b/modules/background_tasks/tests/test_service.py new file mode 100644 index 00000000..da2898a4 --- /dev/null +++ b/modules/background_tasks/tests/test_service.py @@ -0,0 +1,198 @@ +"""Unit tests for BackgroundTaskService — list, get, retry logic.""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime, timedelta +from typing import Any +from unittest.mock import MagicMock + +import pytest +from background_tasks.constants import TaskStatus +from background_tasks.contracts.events import TaskRetried +from background_tasks.models import TaskExecution +from background_tasks.service import BackgroundTaskService +from fastapi import HTTPException +from simple_module_core.events import EventBus +from sqlalchemy.ext.asyncio import AsyncSession + + +def _make_row( + *, + task_name: str = "demo.task", + status: TaskStatus = TaskStatus.FAILED, + args: list[Any] | None = None, + kwargs: dict[str, Any] | None = None, + queue: str = "default", + queued_at: datetime | None = None, +) -> TaskExecution: + return TaskExecution( + celery_task_id=str(uuid.uuid4()), + task_name=task_name, + status=status, + queue=queue, + args=args or [], + kwargs=kwargs or {}, + queued_at=queued_at or datetime.now(UTC), + ) + + +@pytest.fixture +def event_bus() -> EventBus: + return EventBus() + + +@pytest.fixture +def mock_celery() -> MagicMock: + celery = MagicMock(name="Celery") + celery.send_task.return_value.id = "new-celery-id-123" + return celery + + +@pytest.fixture +def service( + db_session: AsyncSession, event_bus: EventBus, mock_celery: MagicMock +) -> BackgroundTaskService: + return BackgroundTaskService(db=db_session, celery=mock_celery, event_bus=event_bus) + + +class TestList: + async def test_returns_paginated_rows_newest_first( + self, db_session: AsyncSession, service: BackgroundTaskService + ): + now = datetime.now(UTC) + older = _make_row(task_name="demo.a", queued_at=now - timedelta(minutes=5)) + newer = _make_row(task_name="demo.b", queued_at=now) + db_session.add_all([older, newer]) + await db_session.flush() + + resp = await service.list() + assert resp.total == 2 + assert [i.task_name for i in resp.items] == ["demo.b", "demo.a"] + + async def test_filters_by_status( + self, db_session: AsyncSession, service: BackgroundTaskService + ): + db_session.add_all( + [ + _make_row(task_name="demo.ok", status=TaskStatus.SUCCESS), + _make_row(task_name="demo.bad", status=TaskStatus.FAILED), + ] + ) + await db_session.flush() + + resp = await service.list(status=TaskStatus.FAILED) + assert [i.task_name for i in resp.items] == ["demo.bad"] + + async def test_filters_by_task_name_substring( + self, db_session: AsyncSession, service: BackgroundTaskService + ): + db_session.add_all( + [ + _make_row(task_name="orders.send_receipt"), + _make_row(task_name="mailer.send_invite"), + ] + ) + await db_session.flush() + + resp = await service.list(task_name="receipt") + assert [i.task_name for i in resp.items] == ["orders.send_receipt"] + + +class TestGet: + async def test_returns_none_for_missing_id(self, service: BackgroundTaskService): + assert await service.get(uuid.uuid4()) is None + + async def test_returns_detail_for_existing_row( + self, db_session: AsyncSession, service: BackgroundTaskService + ): + row = _make_row(args=[1, "two"], kwargs={"three": 3}) + db_session.add(row) + await db_session.flush() + await db_session.refresh(row) + + detail = await service.get(row.id) + assert detail is not None + assert detail.args == [1, "two"] + assert detail.kwargs == {"three": 3} + + +class TestRetry: + async def test_retry_failed_task_enqueues_and_creates_row( + self, + db_session: AsyncSession, + service: BackgroundTaskService, + mock_celery: MagicMock, + event_bus: EventBus, + ): + received: list[TaskRetried] = [] + + async def _on_retried(event: TaskRetried) -> None: + received.append(event) + + event_bus.subscribe(TaskRetried, _on_retried) + + original = _make_row( + task_name="orders.ship", + status=TaskStatus.FAILED, + args=[42], + kwargs={"priority": "high"}, + queue="orders", + ) + db_session.add(original) + await db_session.flush() + await db_session.refresh(original) + + detail = await service.retry(original.id) + + mock_celery.send_task.assert_called_once_with( + "orders.ship", + args=[42], + kwargs={"priority": "high"}, + queue="orders", + ) + assert detail.retried_from_id == original.id + assert detail.status == TaskStatus.PENDING + assert detail.celery_task_id == "new-celery-id-123" + assert len(received) == 1 + assert received[0].original_id == original.id + assert received[0].new_id == detail.id + + async def test_retry_stuck_task_is_allowed( + self, db_session: AsyncSession, service: BackgroundTaskService + ): + row = _make_row(status=TaskStatus.STUCK) + db_session.add(row) + await db_session.flush() + await db_session.refresh(row) + + detail = await service.retry(row.id) + assert detail.retried_from_id == row.id + + async def test_retry_rejects_non_retryable_status( + self, db_session: AsyncSession, service: BackgroundTaskService + ): + row = _make_row(status=TaskStatus.SUCCESS) + db_session.add(row) + await db_session.flush() + await db_session.refresh(row) + + with pytest.raises(HTTPException) as exc: + await service.retry(row.id) + assert exc.value.status_code == 409 + + async def test_retry_missing_row_404s(self, service: BackgroundTaskService): + with pytest.raises(HTTPException) as exc: + await service.retry(uuid.uuid4()) + assert exc.value.status_code == 404 + + async def test_retry_without_celery_503s(self, db_session: AsyncSession, event_bus: EventBus): + service = BackgroundTaskService(db=db_session, celery=None, event_bus=event_bus) + row = _make_row(status=TaskStatus.FAILED) + db_session.add(row) + await db_session.flush() + await db_session.refresh(row) + + with pytest.raises(HTTPException) as exc: + await service.retry(row.id) + assert exc.value.status_code == 503 diff --git a/modules/background_tasks/tests/test_signals.py b/modules/background_tasks/tests/test_signals.py new file mode 100644 index 00000000..735a1e13 --- /dev/null +++ b/modules/background_tasks/tests/test_signals.py @@ -0,0 +1,220 @@ +"""Tests for Celery signal handlers: upsert path and terminal transitions. + +Signals are sync, so we point :mod:`background_tasks.sync_db` at a temporary +SQLite file via ``SM_DATABASE_URL`` for the duration of each test. The +module-level engine cache is also reset so each test gets a fresh engine. +""" + +from __future__ import annotations + +import uuid +from collections.abc import Iterator +from pathlib import Path + +import pytest +from background_tasks import sync_db +from background_tasks._signal_support import upsert_by_celery_id +from background_tasks.constants import TaskStatus +from background_tasks.models import TaskExecution +from background_tasks.signals import ( + on_task_failure, + on_task_prerun, + on_task_publish, + on_task_retry, + on_task_revoked, + on_task_success, +) +from sqlalchemy import select + + +@pytest.fixture +def sync_sqlite(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]: + """Point sync_db at a fresh on-disk SQLite file, create the schema, reset cache.""" + db_file = tmp_path / "signals.db" + monkeypatch.setenv("SM_DATABASE_URL", f"sqlite:///{db_file}") + # Reset the process-global cache so we rebuild an engine bound to ``db_file``. + sync_db._engine = None + sync_db._session_factory = None + + factory = sync_db.get_sync_session_factory() + TaskExecution.metadata.create_all(factory.kw["bind"]) + + yield db_file + + # Dispose so the file can be unlinked cleanly and so later tests don't + # keep a stale engine reference. + if sync_db._engine is not None: + sync_db._engine.dispose() + sync_db._engine = None + sync_db._session_factory = None + + +def _first_row(factory) -> TaskExecution | None: + with factory() as session: + return session.execute(select(TaskExecution)).scalar_one_or_none() + + +class TestPublishSignal: + def test_records_pending_row_on_publish(self, sync_sqlite: Path): + task_id = str(uuid.uuid4()) + on_task_publish( + sender="demo.echo", + headers={"id": task_id, "task": "demo.echo"}, + body=([1, 2], {"x": "y"}, {}), + routing_key="default", + ) + factory = sync_db.get_sync_session_factory() + row = _first_row(factory) + assert row is not None + assert row.celery_task_id == task_id + assert row.task_name == "demo.echo" + assert row.status == TaskStatus.PENDING + assert row.args == [1, 2] + assert row.kwargs == {"x": "y"} + assert row.queue == "default" + assert row.queued_at is not None + + +class TestLifecycleSignals: + def test_prerun_flips_to_running_and_sets_heartbeat(self, sync_sqlite: Path): + task_id = str(uuid.uuid4()) + on_task_publish( + sender="demo.echo", + headers={"id": task_id, "task": "demo.echo"}, + body=([], {}, {}), + ) + + class _Task: + name = "demo.echo" + + on_task_prerun(sender=_Task, task_id=task_id, task=_Task(), args=[], kwargs={}) + factory = sync_db.get_sync_session_factory() + row = _first_row(factory) + assert row is not None + assert row.status == TaskStatus.RUNNING + assert row.started_at is not None + assert row.heartbeat_at is not None + + def test_failure_stores_traceback_and_exception_type(self, sync_sqlite: Path): + task_id = str(uuid.uuid4()) + on_task_publish( + sender="demo.boom", + headers={"id": task_id, "task": "demo.boom"}, + body=([], {}, {}), + ) + + class _Task: + name = "demo.boom" + + class _Einfo: + traceback = "Traceback (most recent call last):\n raise RuntimeError('boom')" + + err = RuntimeError("boom") + on_task_failure(sender=_Task, task_id=task_id, exception=err, einfo=_Einfo()) + + factory = sync_db.get_sync_session_factory() + row = _first_row(factory) + assert row is not None + assert row.status == TaskStatus.FAILED + assert row.exception_type == "RuntimeError" + assert row.traceback is not None + assert "boom" in row.traceback + + def test_success_captures_result(self, sync_sqlite: Path): + task_id = str(uuid.uuid4()) + on_task_publish( + sender="demo.ok", + headers={"id": task_id, "task": "demo.ok"}, + body=([], {}, {}), + ) + + class _Request: + id = task_id + + class _Task: + name = "demo.ok" + request = _Request() + + on_task_success(sender=_Task(), result={"answer": 42}) + + factory = sync_db.get_sync_session_factory() + row = _first_row(factory) + assert row is not None + assert row.status == TaskStatus.SUCCESS + assert row.result == {"answer": 42} + + def test_retry_bumps_counter_and_sets_retrying(self, sync_sqlite: Path): + task_id = str(uuid.uuid4()) + on_task_publish( + sender="demo.retry", + headers={"id": task_id, "task": "demo.retry"}, + body=([], {}, {}), + ) + + class _Request: + id = task_id + retries = 2 + + class _Task: + name = "demo.retry" + + on_task_retry(sender=_Task, request=_Request(), reason="transient") + + factory = sync_db.get_sync_session_factory() + row = _first_row(factory) + assert row is not None + assert row.status == TaskStatus.RETRYING + assert row.retries == 2 + assert row.traceback == "retry: 'transient'" + + def test_revoked_marks_terminal(self, sync_sqlite: Path): + task_id = str(uuid.uuid4()) + on_task_publish( + sender="demo.revoke", + headers={"id": task_id, "task": "demo.revoke"}, + body=([], {}, {}), + ) + + class _Request: + id = task_id + + class _Task: + name = "demo.revoke" + + on_task_revoked(sender=_Task, request=_Request()) + + factory = sync_db.get_sync_session_factory() + row = _first_row(factory) + assert row is not None + assert row.status == TaskStatus.REVOKED + assert row.finished_at is not None + + +class TestUpsertHelper: + def test_updates_existing_row_in_place(self, sync_sqlite: Path): + factory = sync_db.get_sync_session_factory() + cid = str(uuid.uuid4()) + with factory() as session: + upsert_by_celery_id( + session, + celery_task_id=cid, + task_name="demo.upsert", + defaults={ + "status": TaskStatus.PENDING, + "args": [], + "kwargs": {}, + }, + ) + session.commit() + with factory() as session: + upsert_by_celery_id( + session, + celery_task_id=cid, + task_name="demo.upsert", + defaults={"status": TaskStatus.SUCCESS}, + ) + session.commit() + with factory() as session: + rows = session.execute(select(TaskExecution)).scalars().all() + assert len(rows) == 1 + assert rows[0].status == TaskStatus.SUCCESS diff --git a/modules/background_tasks/tsconfig.json b/modules/background_tasks/tsconfig.json new file mode 100644 index 00000000..a83ec2d7 --- /dev/null +++ b/modules/background_tasks/tsconfig.json @@ -0,0 +1,11 @@ +{ + "extends": "@simple-module/tsconfig/base.json", + "compilerOptions": { + "baseUrl": ".", + "paths": { + "@/*": ["./background_tasks/*"], + "@simple-module/ui/*": ["../../packages/ui/src/*"] + } + }, + "include": ["background_tasks/**/*.ts", "background_tasks/**/*.tsx"] +} diff --git a/package-lock.json b/package-lock.json index 795abc66..5ea89ebf 100644 --- a/package-lock.json +++ b/package-lock.json @@ -78,6 +78,19 @@ "react-dom": "^19.0.0" } }, + "modules/background_tasks": { + "name": "@simple-module/background-tasks", + "version": "0.1.0", + "devDependencies": { + "@simple-module/tsconfig": "*" + }, + "peerDependencies": { + "@inertiajs/react": "^2.0.0", + "@simple-module/ui": "*", + "react": "^19.0.0", + "react-dom": "^19.0.0" + } + }, "modules/dashboard": { "name": "@simple-module/dashboard", "version": "0.1.0", @@ -4370,6 +4383,10 @@ "resolved": "modules/auth", "link": true }, + "node_modules/@simple-module/background-tasks": { + "resolved": "modules/background_tasks", + "link": true + }, "node_modules/@simple-module/dashboard": { "resolved": "modules/dashboard", "link": true diff --git a/packages/i18n/src/generated-resources.ts b/packages/i18n/src/generated-resources.ts index 18d4a801..002323c9 100644 --- a/packages/i18n/src/generated-resources.ts +++ b/packages/i18n/src/generated-resources.ts @@ -1,10 +1,52 @@ // AUTO-GENERATED by simple_module_hosting.i18n_manifest — do not edit by hand. // Regenerate by booting the host in development mode. + export default { translation: { 'auth.errors.missing_permission': '', 'auth.errors.not_authenticated': '', + 'background_tasks.detail.args': '', + 'background_tasks.detail.back_button': '', + 'background_tasks.detail.kwargs': '', + 'background_tasks.detail.meta': '', + 'background_tasks.detail.no_traceback': '', + 'background_tasks.detail.result': '', + 'background_tasks.detail.retried_from': '', + 'background_tasks.detail.title': '', + 'background_tasks.detail.traceback': '', + 'background_tasks.filters.status_all': '', + 'background_tasks.filters.status_label': '', + 'background_tasks.index.count_one': '', + 'background_tasks.index.count_other': '', + 'background_tasks.index.description': '', + 'background_tasks.index.empty_description': '', + 'background_tasks.index.empty_title': '', + 'background_tasks.index.no_match': '', + 'background_tasks.index.search_placeholder': '', + 'background_tasks.index.title': '', + 'background_tasks.retry_dialog.cancel': '', + 'background_tasks.retry_dialog.confirm': '', + 'background_tasks.retry_dialog.description': '', + 'background_tasks.retry_dialog.title': '', + 'background_tasks.status.failed': '', + 'background_tasks.status.pending': '', + 'background_tasks.status.retrying': '', + 'background_tasks.status.revoked': '', + 'background_tasks.status.running': '', + 'background_tasks.status.stuck': '', + 'background_tasks.status.success': '', + 'background_tasks.table.actions': '', + 'background_tasks.table.duration': '', + 'background_tasks.table.queue': '', + 'background_tasks.table.queued_at': '', + 'background_tasks.table.retry': '', + 'background_tasks.table.status': '', + 'background_tasks.table.task': '', + 'background_tasks.table.view': '', + 'background_tasks.table.worker': '', + 'background_tasks.toasts.retried': '', + 'background_tasks.toasts.retry_failed': '', 'dashboard.home.description': '', 'dashboard.home.description_body': '', 'dashboard.home.stats.active_users': '', diff --git a/packages/i18n/src/keys.generated.ts b/packages/i18n/src/keys.generated.ts index d0ee311e..1a57dbcd 100644 --- a/packages/i18n/src/keys.generated.ts +++ b/packages/i18n/src/keys.generated.ts @@ -4,145 +4,203 @@ export const keys = { auth: { errors: { - missing_permission: 'auth.errors.missing_permission', - not_authenticated: 'auth.errors.not_authenticated', + missing_permission: "auth.errors.missing_permission", + not_authenticated: "auth.errors.not_authenticated", + }, + }, + background_tasks: { + detail: { + args: "background_tasks.detail.args", + back_button: "background_tasks.detail.back_button", + kwargs: "background_tasks.detail.kwargs", + meta: "background_tasks.detail.meta", + no_traceback: "background_tasks.detail.no_traceback", + result: "background_tasks.detail.result", + retried_from: "background_tasks.detail.retried_from", + title: "background_tasks.detail.title", + traceback: "background_tasks.detail.traceback", + }, + filters: { + status_all: "background_tasks.filters.status_all", + status_label: "background_tasks.filters.status_label", + }, + index: { + count: "background_tasks.index.count", + count_one: "background_tasks.index.count_one", + count_other: "background_tasks.index.count_other", + description: "background_tasks.index.description", + empty_description: "background_tasks.index.empty_description", + empty_title: "background_tasks.index.empty_title", + no_match: "background_tasks.index.no_match", + search_placeholder: "background_tasks.index.search_placeholder", + title: "background_tasks.index.title", + }, + retry_dialog: { + cancel: "background_tasks.retry_dialog.cancel", + confirm: "background_tasks.retry_dialog.confirm", + description: "background_tasks.retry_dialog.description", + title: "background_tasks.retry_dialog.title", + }, + status: { + failed: "background_tasks.status.failed", + pending: "background_tasks.status.pending", + retrying: "background_tasks.status.retrying", + revoked: "background_tasks.status.revoked", + running: "background_tasks.status.running", + stuck: "background_tasks.status.stuck", + success: "background_tasks.status.success", + }, + table: { + actions: "background_tasks.table.actions", + duration: "background_tasks.table.duration", + queue: "background_tasks.table.queue", + queued_at: "background_tasks.table.queued_at", + retry: "background_tasks.table.retry", + status: "background_tasks.table.status", + task: "background_tasks.table.task", + view: "background_tasks.table.view", + worker: "background_tasks.table.worker", + }, + toasts: { + retried: "background_tasks.toasts.retried", + retry_failed: "background_tasks.toasts.retry_failed", }, }, dashboard: { home: { - description: 'dashboard.home.description', - description_body: 'dashboard.home.description_body', + description: "dashboard.home.description", + description_body: "dashboard.home.description_body", stats: { - active_users: 'dashboard.home.stats.active_users', - modules: 'dashboard.home.stats.modules', - products: 'dashboard.home.stats.products', - total_users: 'dashboard.home.stats.total_users', + active_users: "dashboard.home.stats.active_users", + modules: "dashboard.home.stats.modules", + products: "dashboard.home.stats.products", + total_users: "dashboard.home.stats.total_users", }, system_info: { - health_checks: 'dashboard.home.system_info.health_checks', - modules: 'dashboard.home.system_info.modules', - python_version: 'dashboard.home.system_info.python_version', + health_checks: "dashboard.home.system_info.health_checks", + modules: "dashboard.home.system_info.modules", + python_version: "dashboard.home.system_info.python_version", }, - system_info_title: 'dashboard.home.system_info_title', - title: 'dashboard.home.title', - welcome_card_title: 'dashboard.home.welcome_card_title', - welcome_message: 'dashboard.home.welcome_message', + system_info_title: "dashboard.home.system_info_title", + title: "dashboard.home.title", + welcome_card_title: "dashboard.home.welcome_card_title", + welcome_message: "dashboard.home.welcome_message", }, }, host: { error: { - forbidden_description: 'host.error.forbidden_description', - forbidden_title: 'host.error.forbidden_title', - generic_description: 'host.error.generic_description', - generic_title: 'host.error.generic_title', - go_back: 'host.error.go_back', - go_home: 'host.error.go_home', - not_found_description: 'host.error.not_found_description', - not_found_title: 'host.error.not_found_title', - server_error_description: 'host.error.server_error_description', - server_error_title: 'host.error.server_error_title', + forbidden_description: "host.error.forbidden_description", + forbidden_title: "host.error.forbidden_title", + generic_description: "host.error.generic_description", + generic_title: "host.error.generic_title", + go_back: "host.error.go_back", + go_home: "host.error.go_home", + not_found_description: "host.error.not_found_description", + not_found_title: "host.error.not_found_title", + server_error_description: "host.error.server_error_description", + server_error_title: "host.error.server_error_title", }, landing: { - badge: 'host.landing.badge', - cta_dashboard: 'host.landing.cta_dashboard', - cta_docs: 'host.landing.cta_docs', - cta_get_started: 'host.landing.cta_get_started', + badge: "host.landing.badge", + cta_dashboard: "host.landing.cta_dashboard", + cta_docs: "host.landing.cta_docs", + cta_get_started: "host.landing.cta_get_started", features: { - auth_description: 'host.landing.features.auth_description', - auth_title: 'host.landing.features.auth_title', - devtools_description: 'host.landing.features.devtools_description', - devtools_title: 'host.landing.features.devtools_title', - diagnostics_description: 'host.landing.features.diagnostics_description', - diagnostics_title: 'host.landing.features.diagnostics_title', - inertia_description: 'host.landing.features.inertia_description', - inertia_title: 'host.landing.features.inertia_title', - module_system_description: 'host.landing.features.module_system_description', - module_system_title: 'host.landing.features.module_system_title', - schema_description: 'host.landing.features.schema_description', - schema_title: 'host.landing.features.schema_title', + auth_description: "host.landing.features.auth_description", + auth_title: "host.landing.features.auth_title", + devtools_description: "host.landing.features.devtools_description", + devtools_title: "host.landing.features.devtools_title", + diagnostics_description: "host.landing.features.diagnostics_description", + diagnostics_title: "host.landing.features.diagnostics_title", + inertia_description: "host.landing.features.inertia_description", + inertia_title: "host.landing.features.inertia_title", + module_system_description: "host.landing.features.module_system_description", + module_system_title: "host.landing.features.module_system_title", + schema_description: "host.landing.features.schema_description", + schema_title: "host.landing.features.schema_title", }, - hero_subtitle: 'host.landing.hero_subtitle', - hero_title_line1: 'host.landing.hero_title_line1', - hero_title_line2: 'host.landing.hero_title_line2', + hero_subtitle: "host.landing.hero_subtitle", + hero_title_line1: "host.landing.hero_title_line1", + hero_title_line2: "host.landing.hero_title_line2", }, }, products: { browse: { - count: 'products.browse.count', - count_one: 'products.browse.count_one', - count_other: 'products.browse.count_other', - create_button: 'products.browse.create_button', - description: 'products.browse.description', - empty_description: 'products.browse.empty_description', - empty_title: 'products.browse.empty_title', - new_button: 'products.browse.new_button', - no_match: 'products.browse.no_match', - search_placeholder: 'products.browse.search_placeholder', - title: 'products.browse.title', + count: "products.browse.count", + count_one: "products.browse.count_one", + count_other: "products.browse.count_other", + create_button: "products.browse.create_button", + description: "products.browse.description", + empty_description: "products.browse.empty_description", + empty_title: "products.browse.empty_title", + new_button: "products.browse.new_button", + no_match: "products.browse.no_match", + search_placeholder: "products.browse.search_placeholder", + title: "products.browse.title", }, create: { - description: 'products.create.description', - submit_button: 'products.create.submit_button', - submitting_button: 'products.create.submitting_button', - title: 'products.create.title', + description: "products.create.description", + submit_button: "products.create.submit_button", + submitting_button: "products.create.submitting_button", + title: "products.create.title", }, delete_dialog: { - cancel: 'products.delete_dialog.cancel', - confirm: 'products.delete_dialog.confirm', - description: 'products.delete_dialog.description', - title: 'products.delete_dialog.title', + cancel: "products.delete_dialog.cancel", + confirm: "products.delete_dialog.confirm", + description: "products.delete_dialog.description", + title: "products.delete_dialog.title", }, edit: { - back_button: 'products.edit.back_button', - description: 'products.edit.description', - submit_button: 'products.edit.submit_button', - submitting_button: 'products.edit.submitting_button', - title: 'products.edit.title', + back_button: "products.edit.back_button", + description: "products.edit.description", + submit_button: "products.edit.submit_button", + submitting_button: "products.edit.submitting_button", + title: "products.edit.title", }, errors: { - not_found: 'products.errors.not_found', + not_found: "products.errors.not_found", }, form: { - active_label: 'products.form.active_label', - cancel_button: 'products.form.cancel_button', - description_label: 'products.form.description_label', - description_placeholder: 'products.form.description_placeholder', - name_label: 'products.form.name_label', - name_placeholder: 'products.form.name_placeholder', - price_label: 'products.form.price_label', - price_placeholder: 'products.form.price_placeholder', + active_label: "products.form.active_label", + cancel_button: "products.form.cancel_button", + description_label: "products.form.description_label", + description_placeholder: "products.form.description_placeholder", + name_label: "products.form.name_label", + name_placeholder: "products.form.name_placeholder", + price_label: "products.form.price_label", + price_placeholder: "products.form.price_placeholder", }, table: { - actions: 'products.table.actions', - active: 'products.table.active', - description: 'products.table.description', - inactive: 'products.table.inactive', - name: 'products.table.name', - price: 'products.table.price', - status: 'products.table.status', + actions: "products.table.actions", + active: "products.table.active", + description: "products.table.description", + inactive: "products.table.inactive", + name: "products.table.name", + price: "products.table.price", + status: "products.table.status", }, toasts: { - created: 'products.toasts.created', - delete_failed: 'products.toasts.delete_failed', - deleted: 'products.toasts.deleted', - updated: 'products.toasts.updated', + created: "products.toasts.created", + delete_failed: "products.toasts.delete_failed", + deleted: "products.toasts.deleted", + updated: "products.toasts.updated", }, validation: { - name_required: 'products.validation.name_required', - name_too_long: 'products.validation.name_too_long', - price_positive: 'products.validation.price_positive', - price_required: 'products.validation.price_required', + name_required: "products.validation.name_required", + name_too_long: "products.validation.name_too_long", + price_positive: "products.validation.price_positive", + price_required: "products.validation.price_required", }, }, ui: { errors: { - generic_description: 'ui.errors.generic_description', - generic_title: 'ui.errors.generic_title', - go_home_button: 'ui.errors.go_home_button', - reload_button: 'ui.errors.reload_button', + generic_description: "ui.errors.generic_description", + generic_title: "ui.errors.generic_title", + go_home_button: "ui.errors.go_home_button", + reload_button: "ui.errors.reload_button", }, switcher: { - label: 'ui.switcher.label', + label: "ui.switcher.label", }, }, } as const; diff --git a/pyproject.toml b/pyproject.toml index 478569dd..2ec5166e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,6 +64,7 @@ extra-paths = [ "modules/dashboard", "modules/products", "modules/users", + "modules/background_tasks", "host", ] @@ -95,7 +96,7 @@ invalid-argument-type = "ignore" [tool.pytest.ini_options] asyncio_mode = "auto" -testpaths = ["framework/core/tests", "framework/db/tests", "framework/hosting/tests", "framework/testing/tests", "host/tests", "modules/auth/tests", "modules/dashboard/tests", "modules/products/tests", "modules/users/tests", "tests/integration", "tests/e2e", "tests/benchmarks"] +testpaths = ["framework/core/tests", "framework/db/tests", "framework/hosting/tests", "framework/testing/tests", "host/tests", "modules/auth/tests", "modules/dashboard/tests", "modules/products/tests", "modules/users/tests", "modules/background_tasks/tests", "tests/integration", "tests/e2e", "tests/benchmarks"] markers = [ "e2e: end-to-end tests requiring a live browser", "perf: performance benchmarks (opt-in; run via `make bench`)", diff --git a/scripts/run_worker.py b/scripts/run_worker.py new file mode 100644 index 00000000..9eb20f87 --- /dev/null +++ b/scripts/run_worker.py @@ -0,0 +1,20 @@ +"""Entry point for the Celery worker and beat services. + +Usage:: + + uv run celery -A scripts.run_worker:celery worker -l info + uv run celery -A scripts.run_worker:celery beat -l info + +Both the web process and the worker go through the same +:func:`background_tasks.celery_app.build_celery` factory, so the broker +config, autodiscovered tasks, and signal handlers stay in lockstep. +""" + +from __future__ import annotations + +from background_tasks.celery_app import build_celery +from background_tasks.settings import BackgroundTasksSettings + +# Module-level name ``celery`` is what ``celery -A scripts.run_worker:celery`` +# looks for. Keep it stable. +celery = build_celery(BackgroundTasksSettings()) From b2fbd967f2bf70fd5fc195b22c97a4c577248a34 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 19 Apr 2026 11:57:55 +0000 Subject: [PATCH 2/2] refactor(background_tasks): simplify after code review Code-quality and efficiency pass on the Celery module: - reuse framework's ENTRY_POINT_GROUP in celery_app - drop dead constants (CELERY_STATE_*, RUNNING_STATUSES) - drop dead 503 branch in service.retry (celery is always set by on_startup) - merge list+count into one window-function query - purge_old_executions now issues one DELETE instead of SELECT-then-IN - dispose the sync engine on shutdown so lifespan restarts don't leak pools - collapse six signal-handler try/except bodies via a shared _apply helper - extract shared retry flow (fetch + toast) and RetryConfirmDialog so Index.tsx and Detail.tsx stop duplicating the same ~20 lines each https://claude.ai/code/session_013VMtYvyhYDHEDHGKWAQqPf --- .../background_tasks/_signal_support.py | 26 ++- .../background_tasks/celery_app.py | 7 +- .../background_tasks/constants.py | 12 - .../background_tasks/module.py | 10 +- .../background_tasks/pages/Detail.tsx | 139 +++++------ .../background_tasks/pages/Index.tsx | 85 +++---- .../pages/components/ExecutionRow.tsx | 58 +---- .../pages/components/RetryConfirmDialog.tsx | 38 +++ .../background_tasks/pages/retry.ts | 43 ++++ .../background_tasks/service.py | 48 ++-- .../background_tasks/signals.py | 216 ++++++++---------- .../background_tasks/sync_db.py | 14 ++ .../background_tasks/tasks.py | 35 ++- .../background_tasks/tests/test_service.py | 11 - .../background_tasks/tests/test_signals.py | 3 +- 15 files changed, 344 insertions(+), 401 deletions(-) create mode 100644 modules/background_tasks/background_tasks/pages/components/RetryConfirmDialog.tsx create mode 100644 modules/background_tasks/background_tasks/pages/retry.ts diff --git a/modules/background_tasks/background_tasks/_signal_support.py b/modules/background_tasks/background_tasks/_signal_support.py index 7cf05663..f5c813fa 100644 --- a/modules/background_tasks/background_tasks/_signal_support.py +++ b/modules/background_tasks/background_tasks/_signal_support.py @@ -36,11 +36,11 @@ def upsert_by_celery_id( session: Session, *, celery_task_id: str | None, - task_name: str, defaults: dict[str, Any], ) -> TaskExecution: - """Fetch the row for ``celery_task_id``, or create it with ``defaults``. + """Fetch the row for ``celery_task_id``, or create it from ``defaults``. + ``defaults`` must include ``task_name`` (it's NOT NULL on the table). Handlers never race on the same row within a worker process — Celery serialises per-task signals — so this is upsert-by-read, no advisory locks needed. @@ -50,10 +50,7 @@ def upsert_by_celery_id( stmt = select(TaskExecution).where(TaskExecution.celery_task_id == celery_task_id) row = session.execute(stmt).scalar_one_or_none() if row is None: - # Callers sometimes include ``task_name`` in defaults too — let their - # value win, but ensure a name is always set. - row_kwargs = {"task_name": task_name, **defaults} - row = TaskExecution(celery_task_id=celery_task_id, **row_kwargs) + row = TaskExecution(celery_task_id=celery_task_id, **defaults) session.add(row) else: for key, value in defaults.items(): @@ -61,7 +58,7 @@ def upsert_by_celery_id( return row -def task_name_of(sender: Any, task: Any) -> str: +def task_name_of(sender: Any, task: Any = None) -> str: """Best-effort resolution of a task's registered name from signal args.""" return ( (isinstance(sender, str) and sender) @@ -71,6 +68,21 @@ def task_name_of(sender: Any, task: Any) -> str: ) +def task_id_from(sender: Any = None, request: Any = None) -> str | None: + """Pull the Celery task UUID out of whichever signal-arg carries it. + + Celery's signals pass the task id via three different shapes depending + on the signal: ``task_id=`` kwarg, ``request.id``, or ``sender.request.id``. + Centralising the probe keeps the per-handler code boring. + """ + if request is not None: + tid = getattr(request, "id", None) + if tid: + return tid + sender_request = getattr(sender, "request", None) + return getattr(sender_request, "id", None) if sender_request is not None else None + + def jsonable_result(result: Any) -> dict[str, Any] | None: """Wrap arbitrary task results in a JSON-storable shape.""" if result is None: diff --git a/modules/background_tasks/background_tasks/celery_app.py b/modules/background_tasks/background_tasks/celery_app.py index f1989125..e7aa87d0 100644 --- a/modules/background_tasks/background_tasks/celery_app.py +++ b/modules/background_tasks/background_tasks/celery_app.py @@ -20,6 +20,7 @@ from celery import Celery from celery.schedules import schedule +from simple_module_core.discovery import ENTRY_POINT_GROUP from background_tasks.constants import ( INTERNAL_TASK_PURGE_OLD, @@ -30,8 +31,6 @@ logger = logging.getLogger(__name__) -_ENTRY_POINT_GROUP = "simple_module" - def _discover_task_packages() -> list[str]: """Return the top-level package name of every installed simple_module. @@ -41,9 +40,7 @@ def _discover_task_packages() -> list[str]: tasks without a per-module hook. """ packages: set[str] = set() - for ep in entry_points(group=_ENTRY_POINT_GROUP): - # ep.value looks like "background_tasks.module:BackgroundTasksModule". - # We want just the top-level package name. + for ep in entry_points(group=ENTRY_POINT_GROUP): module_path = ep.value.split(":", 1)[0] top_level = module_path.split(".", 1)[0] packages.add(top_level) diff --git a/modules/background_tasks/background_tasks/constants.py b/modules/background_tasks/background_tasks/constants.py index e65e85b6..99c54dbb 100644 --- a/modules/background_tasks/background_tasks/constants.py +++ b/modules/background_tasks/background_tasks/constants.py @@ -56,9 +56,6 @@ class TaskStatus(StrEnum): TERMINAL_STATUSES: frozenset[TaskStatus] = frozenset( {TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.STUCK, TaskStatus.REVOKED} ) -RUNNING_STATUSES: frozenset[TaskStatus] = frozenset( - {TaskStatus.PENDING, TaskStatus.RUNNING, TaskStatus.RETRYING} -) # ── Defaults ──────────────────────────────────────────────────── DEFAULT_QUEUE = "default" @@ -73,12 +70,3 @@ class TaskStatus(StrEnum): # ── Internal task names ───────────────────────────────────────── INTERNAL_TASK_SWEEP_STUCK = "background_tasks.sweep_stuck_tasks" INTERNAL_TASK_PURGE_OLD = "background_tasks.purge_old_executions" - -# ── Celery signal knobs ───────────────────────────────────────── -# Celery's built-in state strings don't line up 1:1 with TaskStatus, so we -# map at the signal layer. These are the keys we read off Celery state. -CELERY_STATE_STARTED = "STARTED" -CELERY_STATE_SUCCESS = "SUCCESS" -CELERY_STATE_FAILURE = "FAILURE" -CELERY_STATE_RETRY = "RETRY" -CELERY_STATE_REVOKED = "REVOKED" diff --git a/modules/background_tasks/background_tasks/module.py b/modules/background_tasks/background_tasks/module.py index 1621a94b..c98d5cb0 100644 --- a/modules/background_tasks/background_tasks/module.py +++ b/modules/background_tasks/background_tasks/module.py @@ -90,7 +90,9 @@ async def on_startup(self, app: FastAPI) -> None: ) async def on_shutdown(self, app: FastAPI) -> None: - services = getattr(app.state, "background_tasks", None) - celery = getattr(services, "celery", None) if services else None - if celery is not None: - celery.close() + from background_tasks.sync_db import dispose_sync_engine + + services = app.state.background_tasks + if services.celery is not None: + services.celery.close() + dispose_sync_engine() diff --git a/modules/background_tasks/background_tasks/pages/Detail.tsx b/modules/background_tasks/background_tasks/pages/Detail.tsx index fb2dd7d1..7d754d07 100644 --- a/modules/background_tasks/background_tasks/pages/Detail.tsx +++ b/modules/background_tasks/background_tasks/pages/Detail.tsx @@ -1,31 +1,15 @@ import { Link, router, usePage } from '@inertiajs/react'; import { PageShell } from '@simple-module/ui/components/PageShell'; -import { - AlertDialog, - AlertDialogAction, - AlertDialogCancel, - AlertDialogContent, - AlertDialogDescription, - AlertDialogFooter, - AlertDialogHeader, - AlertDialogTitle, - AlertDialogTrigger, -} from '@simple-module/ui/components/ui/alert-dialog'; import { Badge } from '@simple-module/ui/components/ui/badge'; import { Button } from '@simple-module/ui/components/ui/button'; import { Card } from '@simple-module/ui/components/ui/card'; import { usePermissions } from '@simple-module/ui/hooks/use-permissions'; import { AuthenticatedLayout } from '@simple-module/ui/layouts/AuthenticatedLayout'; -import { fetchWithCsrf } from '@simple-module/ui/lib/csrf'; import { ArrowLeft, RefreshCcw } from 'lucide-react'; -import { toast } from 'sonner'; -import { - API_BASE, - RETRYABLE_STATUSES, - STATUS_BADGE_VARIANT, - type TaskStatus, - VIEW_BASE, -} from './constants'; +import type { ReactNode } from 'react'; +import { RetryConfirmDialog } from './components/RetryConfirmDialog'; +import { RETRYABLE_STATUSES, STATUS_BADGE_VARIANT, type TaskStatus, VIEW_BASE } from './constants'; +import { retryExecution } from './retry'; interface Execution { id: string; @@ -47,10 +31,6 @@ interface Execution { heartbeat_at: string | null; } -interface Props { - execution: Execution; -} - function statusLabel(status: TaskStatus): string { return status[0].toUpperCase() + status.slice(1); } @@ -68,27 +48,42 @@ function formatTs(ts: string | null): string { return ts ? new Date(ts).toLocaleString() : '—'; } +function JsonCard({ title, children }: { title: string; children: ReactNode }) { + return ( + +

{title}

+ {children} + + ); +} + +function JsonBlock({ value }: { value: unknown }) { + return ( + 
+      {pretty(value)}
+
+ ); +} + +function Row({ label, value, mono }: { label: string; value: string; mono?: boolean }) { + return ( +
+
{label}
+
{value}
+
+ ); +} + function Detail() { - const { execution } = usePage<{ props: Props }>().props as unknown as Props; + const { execution } = usePage<{ props: { execution: Execution } }>().props as unknown as { + execution: Execution; + }; const { can } = usePermissions(); - const canRetry = can('background_tasks.manage'); - const retryable = RETRYABLE_STATUSES.has(execution.status) && canRetry; + const retryable = RETRYABLE_STATUSES.has(execution.status) && can('background_tasks.manage'); async function handleRetry() { - try { - const res = await fetchWithCsrf(`${API_BASE}/executions/${execution.id}/retry`, { - method: 'POST', - }); - if (!res.ok) { - const body = await res.json().catch(() => ({})); - throw new Error(body.detail || `HTTP ${res.status}`); - } - const created = (await res.json()) as Execution; - toast.success(`Task "${execution.task_name}" re-enqueued`); - router.visit(`${VIEW_BASE}/${created.id}`); - } catch (err) { - toast.error(err instanceof Error ? err.message : 'Failed to retry task'); - } + const created = await retryExecution(execution); + if (created) router.visit(`${VIEW_BASE}/${created.id}`); } return ( @@ -104,26 +99,15 @@ function Detail() { {retryable && ( - - + Retry task - - - - Retry this task? - - A new task execution will be enqueued with the same arguments. - - - - Cancel - Retry task - - - + } + onConfirm={handleRetry} + /> )} } @@ -166,28 +150,18 @@ function Detail() {
- -

Arguments

- 
-              {pretty(execution.args)}
-
- - -

Keyword arguments

- 
-              {pretty(execution.kwargs)}
-
- + + + + + + {execution.result !== null && ( - -

Result

- 
-                {pretty(execution.result)}
-
- + + + )} - -

Traceback

+ {execution.traceback ? ( 
                 {execution.traceback}
@@ -195,21 +169,12 @@ function Detail() {
             ) : (
               
No traceback recorded.

             )}
-          
+
); } -function Row({ label, value, mono }: { label: string; value: string; mono?: boolean }) { - return ( -
-
{label}
-
{value}
-
- ); -} - Detail.layout = (page: React.ReactNode) => {page}; export default Detail; diff --git a/modules/background_tasks/background_tasks/pages/Index.tsx b/modules/background_tasks/background_tasks/pages/Index.tsx index 2fa4a1d3..b3e8733d 100644 --- a/modules/background_tasks/background_tasks/pages/Index.tsx +++ b/modules/background_tasks/background_tasks/pages/Index.tsx @@ -20,12 +20,11 @@ import { } from '@simple-module/ui/components/ui/table'; import { usePermissions } from '@simple-module/ui/hooks/use-permissions'; import { AuthenticatedLayout } from '@simple-module/ui/layouts/AuthenticatedLayout'; -import { fetchWithCsrf } from '@simple-module/ui/lib/csrf'; import { Activity, Search } from 'lucide-react'; -import { useCallback, useEffect, useMemo, useState } from 'react'; -import { toast } from 'sonner'; -import { type Execution, ExecutionRow, statusLabel } from './components/ExecutionRow'; -import { API_BASE, STATUS_ORDER, VIEW_BASE } from './constants'; +import { useEffect, useState } from 'react'; +import { ExecutionRow, statusLabel } from './components/ExecutionRow'; +import { STATUS_ORDER, VIEW_BASE } from './constants'; +import { type Execution, retryExecution } from './retry'; interface Pagination { page: number; @@ -33,19 +32,22 @@ interface Pagination { total: number; } -interface Filters { - status: string; - task_name: string; -} - interface Props { executions: Execution[]; pagination: Pagination; - filters: Filters; + filters: { status: string; task_name: string }; } const STATUS_ALL = '__all__'; +function pushFilters(filters: { status: string; task_name: string }, page: number): void { + const params: Record = {}; + if (filters.task_name) params.q = filters.task_name; + if (filters.status && filters.status !== STATUS_ALL) params.status = filters.status; + if (page > 1) params.page = String(page); + router.get(VIEW_BASE, params, { preserveState: true, preserveScroll: true }); +} + function Index() { const { executions, @@ -57,47 +59,23 @@ function Index() { const canRetry = can('background_tasks.manage'); const [search, setSearch] = useState(initialFilters.task_name ?? ''); - const [status, setStatus] = useState(initialFilters.status || STATUS_ALL); - - const totalPages = useMemo( - () => Math.ceil(pagination.total / pagination.per_page), - [pagination.total, pagination.per_page], - ); - - const navigate = useCallback( - (opts: { page?: number; q?: string; status?: string }) => { - const params: Record = {}; - const q = opts.q ?? search; - const s = opts.status ?? status; - const page = opts.page ?? 1; - if (q) params.q = q; - if (s && s !== STATUS_ALL) params.status = s; - if (page > 1) params.page = String(page); - router.get(VIEW_BASE, params, { preserveState: true, preserveScroll: true }); - }, - [search, status], - ); + const totalPages = Math.ceil(pagination.total / pagination.per_page); + const statusValue = initialFilters.status || STATUS_ALL; + // Debounce search: any change from the server-provided value kicks off a + // page-1 navigation 300ms after the user stops typing. useEffect(() => { if (search === (initialFilters.task_name ?? '')) return; - const timeout = setTimeout(() => navigate({ q: search }), 300); + const timeout = setTimeout( + () => pushFilters({ status: statusValue, task_name: search }, 1), + 300, + ); return () => clearTimeout(timeout); - }, [search, initialFilters.task_name, navigate]); + }, [search, initialFilters.task_name, statusValue]); async function handleRetry(execution: Execution) { - try { - const res = await fetchWithCsrf(`${API_BASE}/executions/${execution.id}/retry`, { - method: 'POST', - }); - if (!res.ok) { - const body = await res.json().catch(() => ({})); - throw new Error(body.detail || `HTTP ${res.status}`); - } - toast.success(`Task "${execution.task_name}" re-enqueued`); - router.reload({ only: ['executions', 'pagination'] }); - } catch (err) { - toast.error(err instanceof Error ? err.message : 'Failed to retry task'); - } + const created = await retryExecution(execution); + if (created) router.reload({ only: ['executions', 'pagination'] }); } return ( @@ -116,11 +94,8 @@ function Index() { />