diff --git a/.claude/settings.local.json b/.claude/settings.local.json index cc71e59..d27782b 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -34,7 +34,9 @@ "Bash(curl -s -b /tmp/fg_cookies.txt http://localhost:8000__TRACKED_VAR__)", "Bash(sqlite3 app.db \"PRAGMA foreign_keys=ON; SELECT scope_model_id FROM augmented_parts WHERE id=1862;\")", "Bash(sqlite3 app.db \"SELECT id, name FROM scope_models WHERE name='Viper';\")", - "Bash(sqlite3 app.db \"SELECT COUNT\\(*\\) FROM augmented_parts WHERE scope_model_id=\\(SELECT id FROM scope_models WHERE name='Viper'\\);\")" + "Bash(sqlite3 app.db \"SELECT COUNT\\(*\\) FROM augmented_parts WHERE scope_model_id=\\(SELECT id FROM scope_models WHERE name='Viper'\\);\")", + "Bash(sqlite3 app.db \"PRAGMA wal_checkpoint\\(FULL\\);\")", + "Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://127.0.0.1:8000/login)" ] } } diff --git a/alembic/versions/04425ce3d4d5_add_jobs_table.py b/alembic/versions/04425ce3d4d5_add_jobs_table.py new file mode 100644 index 0000000..ac221f5 --- /dev/null +++ b/alembic/versions/04425ce3d4d5_add_jobs_table.py @@ -0,0 +1,44 @@ +"""add jobs table + +Revision ID: 04425ce3d4d5 +Revises: a3f7c19b2d40 +Create Date: 2026-08-05 15:21:09.527442 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision: str = "04425ce3d4d5" +down_revision: Union[str, None] = "a3f7c19b2d40" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "jobs", + sa.Column("id", sa.String(length=32), nullable=False), + sa.Column("kind", sa.String(length=64), nullable=False), + sa.Column("status", sa.String(length=16), nullable=False), + sa.Column("message", sa.Text(), nullable=False), + sa.Column("result", sa.JSON(), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("(CURRENT_TIMESTAMP)"), + nullable=False, + ), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table("jobs") + # ### end Alembic commands ### diff --git a/app/models/__init__.py b/app/models/__init__.py index 4351951..223c6a9 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -14,6 +14,7 @@ from app.models.fit import Fit, FitLabel, FitNote from app.models.imported_part import ImportedPart from app.models.imported_part_config import ImportedPartConfig from app.models.iso_thread import IsoThread +from app.models.job import Job from app.models.manufacturer import Manufacturer from app.models.part_colour import PartColour from app.models.part_type import PartType @@ -40,6 +41,7 @@ __all__ = [ "ImportedPart", "ImportedPartConfig", "IsoThread", + "Job", "Manufacturer", "PartColour", "PartType", diff --git a/app/models/job.py b/app/models/job.py new file mode 100644 index 0000000..927bffd --- /dev/null +++ b/app/models/job.py @@ -0,0 +1,31 @@ +from datetime import datetime + +from sqlalchemy import JSON, DateTime, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.database import Base + + +class Job(Base): + """Tracks a long-running admin operation (sync, import, fit recompute). + + Persisted (rather than kept in a module-level dict) so that any worker + process/replica handling a status-poll request can see progress recorded + by whichever process is actually running the job. + """ + + __tablename__ = "jobs" + + id: Mapped[str] = mapped_column(String(32), primary_key=True) + kind: Mapped[str] = mapped_column(String(64), nullable=False) + status: Mapped[str] = mapped_column( + String(16), nullable=False, default="running" + ) # running | done | error + message: Mapped[str] = mapped_column(Text, nullable=False, default="Working…") + result: Mapped[dict | None] = mapped_column(JSON, nullable=True, default=None) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, default=None + ) diff --git a/app/services/jobs.py b/app/services/jobs.py index 271ceda..3dc9284 100644 --- a/app/services/jobs.py +++ b/app/services/jobs.py @@ -1,11 +1,12 @@ -"""Lightweight in-process tracking for long-running admin operations. +"""DB-backed tracking for long-running admin operations. -Jobs live in a module-level dict, so this only works correctly for a single -OS process (the app's current deployment: one uvicorn worker, no multi-process -or multi-replica setup — see ``entrypoint.sh`` / ``docker-compose.yml``). If -the app is ever scaled to multiple workers/replicas, this needs to move to -shared storage (a DB table, Redis) so a poll request landing on a different -process can still see the job. +Jobs are persisted to the ``jobs`` table rather than kept in a module-level +dict, so that a status-poll request landing on a *different* worker +process/replica than the one running the job can still see its progress — +required once the app runs with more than one uvicorn worker or replica +(a module-level dict is process-local and would report "not found" for a job +started elsewhere, even while that job is still running and writing to the +shared database). """ from __future__ import annotations @@ -13,62 +14,83 @@ from __future__ import annotations import asyncio import uuid from collections.abc import Awaitable, Callable -from dataclasses import dataclass, field -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta -_MAX_FINISHED_AGE_SECONDS = 3600 +from sqlalchemy import delete +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.database import AsyncSessionLocal +from app.models.job import Job + +_MAX_FINISHED_AGE = timedelta(hours=1) -@dataclass -class Job: - id: str - kind: str - status: str = "running" # running | done | error - message: str = "Working…" - result: object | None = None - created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) - finished_at: datetime | None = None +async def get(db: AsyncSession, job_id: str) -> Job | None: + return await db.get(Job, job_id) -_jobs: dict[str, Job] = {} +async def start(kind: str, run: Callable[[str], Awaitable[None]]) -> Job: + """Create a job row, launch ``run(job_id)`` in the background, and return + the job immediately. - -def get(job_id: str) -> Job | None: - return _jobs.get(job_id) - - -def _evict_stale() -> None: - now = datetime.now(UTC) - stale = [ - job_id - for job_id, job in _jobs.items() - if job.finished_at - and (now - job.finished_at).total_seconds() > _MAX_FINISHED_AGE_SECONDS - ] - for job_id in stale: - del _jobs[job_id] - - -def start(kind: str, run: Callable[[Job], Awaitable[None]]) -> Job: - """Create a job, launch ``run(job)`` in the background, and return it immediately. - - ``run`` is responsible for updating ``job.status``/``job.message``/``job.result`` - to reflect the outcome, including catching its own errors. Any exception that - escapes anyway is caught here as a last resort so the job doesn't hang forever - in "running" state. + ``run`` is responsible for calling ``mark_done``/``mark_error`` to record + the outcome, including on its own errors. Any exception that escapes + anyway is caught here as a last resort so the job doesn't hang forever in + "running" state. """ - _evict_stale() - job = Job(id=uuid.uuid4().hex, kind=kind) - _jobs[job.id] = job + async with AsyncSessionLocal() as db: + await _evict_stale(db) + job = Job(id=uuid.uuid4().hex, kind=kind, status="running", message="Working…") + db.add(job) + await db.commit() async def _runner() -> None: try: - await run(job) + await run(job.id) except Exception as exc: - job.status = "error" - job.message = f"{kind} failed: {exc}" - finally: - job.finished_at = datetime.now(UTC) + await mark_error(job.id, f"{kind} failed: {exc}") asyncio.create_task(_runner()) return job + + +async def mark_done(job_id: str, message: str, result: dict | None = None) -> None: + await _update(job_id, status="done", message=message, result=result) + + +async def mark_error(job_id: str, message: str) -> None: + await _update(job_id, status="error", message=message) + + +async def set_message(job_id: str, message: str) -> None: + await _update(job_id, message=message) + + +async def _update( + job_id: str, + *, + status: str | None = None, + message: str | None = None, + result: dict | None = None, +) -> None: + async with AsyncSessionLocal() as db: + job = await db.get(Job, job_id) + if job is None: + return + if status is not None: + job.status = status + if status in ("done", "error"): + job.finished_at = datetime.now(UTC) + if message is not None: + job.message = message + if result is not None: + job.result = result + await db.commit() + + +async def _evict_stale(db: AsyncSession) -> None: + cutoff = datetime.now(UTC) - _MAX_FINISHED_AGE + await db.execute( + delete(Job).where(Job.finished_at.is_not(None), Job.finished_at < cutoff) + ) + await db.commit() diff --git a/app/web/admin.py b/app/web/admin.py index bd37481..68624cb 100644 --- a/app/web/admin.py +++ b/app/web/admin.py @@ -1511,13 +1511,12 @@ async def upload_scope_import( data = await file.read() actor_id = actor.id - async def _run(job: jobs.Job) -> None: + async def _run(job_id: str) -> None: async with AsyncSessionLocal() as db: try: result = await scope_import_service.import_from_bytes(db, data) except Exception as exc: # noqa: BLE001 - job.status = "error" - job.message = f"Could not read spreadsheet: {exc}" + await jobs.mark_error(job_id, f"Could not read spreadsheet: {exc}") return actor_row = await user_crud.get(db, actor_id) await audit_crud.log( @@ -1532,10 +1531,16 @@ async def upload_scope_import( f"{result.errors} errors" ), ) - job.status = "done" - job.result = result + await jobs.mark_done( + job_id, + message=( + f"Scope import complete: {result.created} created, " + f"{result.updated} updated." + ), + result=result.model_dump(mode="json"), + ) - job = jobs.start("scope_import", _run) + job = await jobs.start("scope_import", _run) return templates.TemplateResponse( request, "admin/_job_progress.html", _scope_import_progress_ctx(job) ) @@ -1545,9 +1550,10 @@ async def upload_scope_import( async def scope_import_job_status( request: Request, job_id: str, + db: AsyncSession = Depends(get_db), _: User = Depends(web_require("augmented_parts:write")), ): - job = jobs.get(job_id) + job = await jobs.get(db, job_id) if job is None: return templates.TemplateResponse( request, @@ -1718,13 +1724,12 @@ async def sync_augmented_parts( ): actor_id = actor.id - async def _run(job: jobs.Job) -> None: + async def _run(job_id: str) -> None: async with AsyncSessionLocal() as db: try: result = await augmented_part_sync_service.sync(db) except Exception as exc: - job.status = "error" - job.message = f"Sync failed: {exc}" + await jobs.mark_error(job_id, f"Sync failed: {exc}") return actor_row = await user_crud.get(db, actor_id) await audit_crud.log( @@ -1738,18 +1743,21 @@ async def sync_augmented_parts( f"{result.skipped} skipped (no type match)" ), ) - job.status = "done" - job.message = ( - f"Sync complete: {result.created} created, {result.deleted} deleted" - + ( - f", {result.skipped} skipped (type not recognised)" - if result.skipped - else "" - ) - + "." + await jobs.mark_done( + job_id, + message=( + f"Sync complete: {result.created} created, " + f"{result.deleted} deleted" + + ( + f", {result.skipped} skipped (type not recognised)" + if result.skipped + else "" + ) + + "." + ), ) - job = jobs.start("augmented_part_sync", _run) + job = await jobs.start("augmented_part_sync", _run) return templates.TemplateResponse( request, "admin/_job_progress.html", @@ -1768,7 +1776,7 @@ async def augmented_parts_sync_job_status( db: AsyncSession = Depends(get_db), _: User = Depends(web_require("augmented_parts:write")), ): - job = jobs.get(job_id) + job = await jobs.get(db, job_id) if job is None: form_ctx = await _augmented_part_form_ctx(db) return templates.TemplateResponse( @@ -2483,7 +2491,7 @@ async def recompute_augmented_part_fits( ): actor_id = actor.id - async def _run(job: jobs.Job) -> None: + async def _run(job_id: str) -> None: async with AsyncSessionLocal() as db: result = await fit_matching_service.recompute_for_part( db, augmented_part_id @@ -2500,10 +2508,11 @@ async def recompute_augmented_part_fits( f"{result.created} created, {result.removed} removed" ), ) - job.status = "done" - job.message = f"{result.created} created, {result.removed} removed." + await jobs.mark_done( + job_id, f"{result.created} created, {result.removed} removed." + ) - job = jobs.start("fit_recompute_part", _run) + job = await jobs.start("fit_recompute_part", _run) return templates.TemplateResponse( request, "admin/_job_progress.html", @@ -2527,7 +2536,7 @@ async def augmented_part_fits_recompute_job_status( db: AsyncSession = Depends(get_db), actor: User = Depends(web_require("fits:write")), ): - job = jobs.get(job_id) + job = await jobs.get(db, job_id) if job is None: ctx = await _augmented_part_fits_ctx( db, @@ -2774,7 +2783,7 @@ async def recompute_all_fits( ): actor_id = actor.id - async def _run(job: jobs.Job) -> None: + async def _run(job_id: str) -> None: async with AsyncSessionLocal() as db: result = await fit_matching_service.recompute_all(db) actor_row = await user_crud.get(db, actor_id) @@ -2789,13 +2798,15 @@ async def recompute_all_fits( f"{result.removed} removed" ), ) - job.status = "done" - job.message = ( - f"Recompute complete: {result.created} created, " - f"{result.removed} removed." + await jobs.mark_done( + job_id, + message=( + f"Recompute complete: {result.created} created, " + f"{result.removed} removed." + ), ) - job = jobs.start("fit_recompute_all", _run) + job = await jobs.start("fit_recompute_all", _run) return templates.TemplateResponse( request, "admin/_job_progress_inline.html", @@ -2813,9 +2824,10 @@ async def recompute_all_fits( async def recompute_all_fits_job_status( request: Request, job_id: str, + db: AsyncSession = Depends(get_db), _: User = Depends(web_require("fits:write")), ): - job = jobs.get(job_id) + job = await jobs.get(db, job_id) if job is None: return HTMLResponse("Recompute job not found — it may have expired.") if job.status == "running": diff --git a/tests/test_admin_background_jobs.py b/tests/test_admin_background_jobs.py index ba60476..563b717 100644 --- a/tests/test_admin_background_jobs.py +++ b/tests/test_admin_background_jobs.py @@ -2,7 +2,10 @@ import asyncio import re import pytest +import pytest_asyncio +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from app.core.database import Base from app.crud import role as role_crud from app.models.attachment_point import AttachmentPoint from app.models.attachment_point_name import AttachmentPointName @@ -13,6 +16,25 @@ from tests.test_scope_import import _scope_row, _workbook _JOB_URL_RE = re.compile(r'hx-get="([^"]+/jobs/[^"]+)"') +@pytest_asyncio.fixture +async def db_sessionmaker(tmp_path): + """Override conftest's in-memory+StaticPool fixture for this module. + + These tests run a real background asyncio task (the job) concurrently + with polling requests, both hitting the DB at once. StaticPool pins every + session to one shared raw connection, which can't safely serve two + callers at the same time; a real file-backed DB with its own connection + pool (how SQLite/Postgres actually behave outside tests) can. + """ + db_path = tmp_path / "jobs_test.db" + engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + maker = async_sessionmaker(engine, expire_on_commit=False) + yield maker + await engine.dispose() + + async def _web_login(client, email, password="secret123"): resp = await client.post("/login", data={"email": email, "password": password}) assert resp.status_code == 303 @@ -39,6 +61,7 @@ async def test_scope_import_runs_in_background( client, seeded, make_user, db_sessionmaker, monkeypatch ): monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker) + monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker) async with db_sessionmaker() as db: perms = seeded["perms"] editor = await role_crud.create( @@ -69,6 +92,7 @@ async def test_augmented_parts_sync_runs_in_background( client, seeded, make_user, db_sessionmaker, monkeypatch ): monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker) + monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker) async with db_sessionmaker() as db: perms = seeded["perms"] editor = await role_crud.create( @@ -93,6 +117,7 @@ async def test_fit_recompute_for_part_runs_in_background( client, seeded, make_user, db_sessionmaker, monkeypatch ): monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker) + monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker) async with db_sessionmaker() as db: apn = AttachmentPointName(name="Objective") part_t = PartType(name="Scope") @@ -137,6 +162,7 @@ async def test_fit_recompute_all_runs_in_background( client, seeded, make_user, db_sessionmaker, monkeypatch ): monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker) + monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker) async with db_sessionmaker() as db: perms = seeded["perms"] editor = await role_crud.create(