fc3d: DownloadService finalize hook — Source.consecutive_failures/last_error/last_checked_at

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

backend/app/services/download_service.py

@@ -134,14 +134,14 @@ class DownloadService:
             raise LookupError(f"source {source_id} not found")
 
         if not source.enabled:
-            ev = DownloadEvent(
+            ev = DownloadEvent(source_id=source_id, status="pending")
-                source_id=source_id, status="skipped",
-                error="source disabled",
-                finished_at=datetime.now(UTC),
-            )
             self.async_session.add(ev)
             await self.async_session.commit()
             await self.async_session.refresh(ev)
+            await self._finalize_event_and_source(
+                event_id=ev.id, source_id=source_id, status="skipped",
+                error_message="source disabled",
+            )
             return {"status": "skipped", "event_id": ev.id}
 
         existing = (await self.async_session.execute(
@@ -258,7 +258,8 @@ class DownloadService:
         run_stats["quarantined_count"] = dl_result.files_quarantined
         stderr_summary = self.gdl._extract_errors_warnings(dl_result.stderr)
 
-        ev.status = "ok" if (dl_result.success and import_summary["errors"] == 0) else "error"
+        status = "ok" if (dl_result.success and import_summary["errors"] == 0) else "error"
+        ev.status = status
         ev.finished_at = datetime.now(UTC)
         ev.files_count = import_summary["attached"]
         ev.bytes_downloaded = bytes_downloaded
@@ -273,5 +274,50 @@ class DownloadService:
             "quarantined_paths": dl_result.quarantined_paths or None,
             "import_summary": import_summary,
         }
+        await self._update_source_health(
+            source_id=ctx["source_id"], status=status, error_message=ev.error,
+        )
         await self.async_session.commit()
         return event_id
+
+    async def _update_source_health(
+        self, *, source_id: int, status: str, error_message: str | None,
+    ) -> None:
+        """FC-3d: update Source.{consecutive_failures, last_error, last_checked_at}.
+
+        ok      -> failures = 0, error = None, checked_at = now
+        error   -> failures += 1, error = error_message, checked_at = now
+        skipped -> failures unchanged, error = None, checked_at = now
+        """
+        source = (await self.async_session.execute(
+            select(Source).where(Source.id == source_id)
+        )).scalar_one()
+        now = datetime.now(UTC)
+        if status == "ok":
+            source.consecutive_failures = 0
+            source.last_error = None
+        elif status == "error":
+            source.consecutive_failures = (source.consecutive_failures or 0) + 1
+            source.last_error = error_message
+        elif status == "skipped":
+            source.last_error = None
+        source.last_checked_at = now
+
+    async def _finalize_event_and_source(
+        self, *, event_id: int, source_id: int, status: str,
+        error_message: str | None,
+    ) -> None:
+        """Finalize the event AND the source-health columns in one
+        transactional step. Used by the skipped early-out path and by
+        unit tests that exercise the source-health writes directly.
+        """
+        ev = (await self.async_session.execute(
+            select(DownloadEvent).where(DownloadEvent.id == event_id)
+        )).scalar_one()
+        ev.status = status
+        ev.finished_at = datetime.now(UTC)
+        ev.error = error_message
+        await self._update_source_health(
+            source_id=source_id, status=status, error_message=error_message,
+        )
+        await self.async_session.commit()

tests/test_download_service.py

@@ -272,3 +272,101 @@ async def test_patreon_retry_caches_campaign_id(
         select(Source.config_overrides).where(Source.id == source.id)
     ).scalar_one()
     assert overrides.get("patreon_campaign_id") == "99"
+
+
+# --- FC-3d: finalize hook updates Source health columns -------------------
+
+
+async def _seed_source_with_health(db, *, failures=0, last_error=None, suffix="fz"):
+    artist = Artist(name=f"alice-{suffix}", slug=f"alice-{suffix}")
+    db.add(artist)
+    await db.flush()
+    source = Source(
+        artist_id=artist.id, platform="patreon",
+        url=f"https://patreon.com/alice-{suffix}", enabled=True,
+        consecutive_failures=failures, last_error=last_error,
+    )
+    db.add(source)
+    await db.flush()
+    event = DownloadEvent(source_id=source.id, status="running")
+    db.add(event)
+    await db.commit()
+    return source.id, event.id
+
+
+@pytest.mark.asyncio
+async def test_finalize_ok_resets_failures_and_clears_error(db):
+    from backend.app.services.download_service import DownloadService
+
+    source_id, event_id = await _seed_source_with_health(
+        db, failures=3, last_error="prev", suffix="ok",
+    )
+    svc = DownloadService(
+        async_session=db, sync_session=None,
+        gdl=None, importer=None, cred_service=None,
+    )
+    await svc._finalize_event_and_source(
+        event_id=event_id, source_id=source_id, status="ok",
+        error_message=None,
+    )
+
+    row = (await db.execute(
+        select(
+            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
+        ).where(Source.id == source_id)
+    )).one()
+    assert row.consecutive_failures == 0
+    assert row.last_error is None
+    assert row.last_checked_at is not None
+
+
+@pytest.mark.asyncio
+async def test_finalize_error_increments_failures_and_sets_error(db):
+    from backend.app.services.download_service import DownloadService
+
+    source_id, event_id = await _seed_source_with_health(
+        db, failures=2, last_error=None, suffix="err",
+    )
+    svc = DownloadService(
+        async_session=db, sync_session=None,
+        gdl=None, importer=None, cred_service=None,
+    )
+    await svc._finalize_event_and_source(
+        event_id=event_id, source_id=source_id, status="error",
+        error_message="boom",
+    )
+
+    row = (await db.execute(
+        select(
+            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
+        ).where(Source.id == source_id)
+    )).one()
+    assert row.consecutive_failures == 3
+    assert row.last_error == "boom"
+    assert row.last_checked_at is not None
+
+
+@pytest.mark.asyncio
+async def test_finalize_skipped_preserves_failures_clears_error(db):
+    from backend.app.services.download_service import DownloadService
+
+    source_id, event_id = await _seed_source_with_health(
+        db, failures=4, last_error="stale", suffix="skip",
+    )
+    svc = DownloadService(
+        async_session=db, sync_session=None,
+        gdl=None, importer=None, cred_service=None,
+    )
+    await svc._finalize_event_and_source(
+        event_id=event_id, source_id=source_id, status="skipped",
+        error_message=None,
+    )
+
+    row = (await db.execute(
+        select(
+            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
+        ).where(Source.id == source_id)
+    )).one()
+    assert row.consecutive_failures == 4
+    assert row.last_error is None
+    assert row.last_checked_at is not None