fc3i: tests for recover_stalled_task_runs + prune_task_runs

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/test_maintenance.py

@@ -93,3 +93,174 @@ def test_cleanup_old_deletes_finished_old(db_sync):
 
     remaining = {t.source_path for t in db_sync.query(ImportTask).all()}
     assert remaining == {"/import/b.jpg", "/import/c.jpg"}
+
+
+# --- FC-3i: task_run sweep + retention -----------------------------
+
+
+def _make_task_run(db_sync, *, status, started_at, finished_at=None,
+                   error_type=None):
+    from backend.app.models import TaskRun
+    row = TaskRun(
+        celery_task_id="x",
+        queue="ml",
+        task_name="backend.app.tasks.fake.t",
+        target_id=1,
+        started_at=started_at,
+        finished_at=finished_at,
+        duration_ms=1000 if finished_at else None,
+        status=status,
+        error_type=error_type,
+        error_message="x" if status in ("error", "timeout") else None,
+    )
+    db_sync.add(row)
+    db_sync.flush()
+    return row.id
+
+
+def test_recover_stalled_task_runs_flips_old_running_to_error(db_sync):
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import recover_stalled_task_runs
+
+    now = datetime.now(UTC)
+    stale_id = _make_task_run(
+        db_sync, status="running", started_at=now - timedelta(minutes=10),
+    )
+    db_sync.commit()
+
+    recovered = recover_stalled_task_runs.apply().get()
+    assert recovered == 1
+
+    db_sync.expire_all()
+    status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == stale_id)
+    ).scalar_one()
+    error_type = db_sync.execute(
+        select(TaskRun.error_type).where(TaskRun.id == stale_id)
+    ).scalar_one()
+    assert status == "error"
+    assert error_type == "RecoverySweep"
+
+
+def test_recover_stalled_task_runs_skips_fresh_running(db_sync):
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import recover_stalled_task_runs
+
+    now = datetime.now(UTC)
+    fresh_id = _make_task_run(
+        db_sync, status="running", started_at=now - timedelta(seconds=30),
+    )
+    db_sync.commit()
+
+    recovered = recover_stalled_task_runs.apply().get()
+    assert recovered == 0
+
+    db_sync.expire_all()
+    status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == fresh_id)
+    ).scalar_one()
+    assert status == "running"
+
+
+def test_prune_task_runs_deletes_ok_older_than_24h(db_sync):
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import prune_task_runs
+
+    now = datetime.now(UTC)
+    old_id = _make_task_run(
+        db_sync, status="ok",
+        started_at=now - timedelta(hours=30),
+        finished_at=now - timedelta(hours=29),
+    )
+    recent_id = _make_task_run(
+        db_sync, status="ok",
+        started_at=now - timedelta(hours=2),
+        finished_at=now - timedelta(hours=1),
+    )
+    db_sync.commit()
+
+    result = prune_task_runs.apply().get()
+    assert result["ok_deleted"] == 1
+
+    db_sync.expire_all()
+    surviving_ids = set(db_sync.execute(
+        select(TaskRun.id).where(TaskRun.id.in_([old_id, recent_id]))
+    ).scalars().all())
+    assert surviving_ids == {recent_id}
+
+
+def test_prune_task_runs_deletes_failures_older_than_7d(db_sync):
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import prune_task_runs
+
+    now = datetime.now(UTC)
+    old_id = _make_task_run(
+        db_sync, status="error", error_type="OldError",
+        started_at=now - timedelta(days=10),
+        finished_at=now - timedelta(days=9),
+    )
+    db_sync.commit()
+
+    result = prune_task_runs.apply().get()
+    assert result["failures_deleted"] >= 1
+
+    db_sync.expire_all()
+    surviving = db_sync.execute(
+        select(TaskRun.id).where(TaskRun.id == old_id)
+    ).scalar_one_or_none()
+    assert surviving is None
+
+
+def test_prune_task_runs_keeps_recent_failures(db_sync):
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import prune_task_runs
+
+    now = datetime.now(UTC)
+    recent_id = _make_task_run(
+        db_sync, status="error", error_type="RecentError",
+        started_at=now - timedelta(days=6),
+        finished_at=now - timedelta(days=5),
+    )
+    db_sync.commit()
+
+    prune_task_runs.apply().get()
+
+    db_sync.expire_all()
+    surviving = db_sync.execute(
+        select(TaskRun.id).where(TaskRun.id == recent_id)
+    ).scalar_one_or_none()
+    assert surviving == recent_id
+
+
+def test_prune_task_runs_never_deletes_running(db_sync):
+    """Even a 30-day-old running row stays — recovery sweep is the
+    mechanism that flips them; prune doesn't touch in-flight."""
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import prune_task_runs
+
+    now = datetime.now(UTC)
+    ancient_id = _make_task_run(
+        db_sync, status="running",
+        started_at=now - timedelta(days=30),
+    )
+    db_sync.commit()
+
+    prune_task_runs.apply().get()
+
+    db_sync.expire_all()
+    surviving = db_sync.execute(
+        select(TaskRun.id).where(TaskRun.id == ancient_id)
+    ).scalar_one_or_none()
+    assert surviving == ancient_id