From 87fb5347223fb7fc4d73f01538fea99383130da5 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 21 May 2026 16:00:40 -0400 Subject: [PATCH] =?UTF-8?q?fc3d:=20DownloadService=20finalize=20hook=20?= =?UTF-8?q?=E2=80=94=20Source.consecutive=5Ffailures/last=5Ferror/last=5Fc?= =?UTF-8?q?hecked=5Fat?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/services/download_service.py | 58 ++++++++++++-- tests/test_download_service.py | 98 ++++++++++++++++++++++++ 2 files changed, 150 insertions(+), 6 deletions(-) diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index b62dcd0..347590b 100644 --- a/backend/app/services/download_service.py +++ b/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( - source_id=source_id, status="skipped", - error="source disabled", - finished_at=datetime.now(UTC), - ) + ev = DownloadEvent(source_id=source_id, status="pending") 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() diff --git a/tests/test_download_service.py b/tests/test_download_service.py index 578f346..15ad337 100644 --- a/tests/test_download_service.py +++ b/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