diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 347590b..6e917c3 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -28,6 +28,7 @@ from .credential_service import CredentialService from .gallery_dl import GalleryDLService, SourceConfig from .importer import Importer from .patreon_resolver import resolve_campaign_id +from .scheduler_service import set_platform_cooldown log = logging.getLogger(__name__) @@ -276,18 +277,27 @@ class DownloadService: } await self._update_source_health( source_id=ctx["source_id"], status=status, error_message=ev.error, + error_type=dl_result.error_type.value if dl_result.error_type else None, ) await self.async_session.commit() return event_id async def _update_source_health( self, *, source_id: int, status: str, error_message: str | None, + error_type: str | None = 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 + + When error_type == 'rate_limited', also stamps a platform-wide + cooldown via scheduler_service.set_platform_cooldown so the next + scan tick skips every source on this platform until the cooldown + expires. Preventive half of the burst-prevention pair — + consecutive_failures still backs the offending source off across + ticks. """ source = (await self.async_session.execute( select(Source).where(Source.id == source_id) @@ -299,6 +309,8 @@ class DownloadService: elif status == "error": source.consecutive_failures = (source.consecutive_failures or 0) + 1 source.last_error = error_message + if error_type == "rate_limited": + await set_platform_cooldown(self.async_session, source.platform) elif status == "skipped": source.last_error = None source.last_checked_at = now diff --git a/backend/app/services/scheduler_service.py b/backend/app/services/scheduler_service.py index e70c87f..3c42ca1 100644 --- a/backend/app/services/scheduler_service.py +++ b/backend/app/services/scheduler_service.py @@ -9,6 +9,7 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -23,6 +24,18 @@ MAX_BACKOFF_EXPONENT = 6 # stamp is older than a few minutes. SCHEDULER_LAST_TICK_KEY = "scheduler_last_tick_at" +# AppSetting key prefix for per-platform rate-limit cooldowns. When a +# download surfaces ErrorType.RATE_LIMITED, every other source on the same +# platform is deferred for PLATFORM_RATE_LIMIT_COOLDOWN_SECONDS so the next +# scan tick doesn't fire a burst of due same-platform sources back into the +# same limit. Per-source consecutive_failures backoff still applies on top +# of this — but this is PREVENTIVE (kills the same-tick burst from N due +# sources hammering the platform at once), while consecutive_failures is +# REACTIVE (slows the offender down over many cycles). Operator-confirmed +# 2026-05-30. +PLATFORM_COOLDOWN_KEY_PREFIX = "platform_cooldown:" +PLATFORM_RATE_LIMIT_COOLDOWN_SECONDS = 900 # 15 min + def compute_effective_interval( source: Source, artist: Artist, settings: ImportSettings, @@ -45,10 +58,64 @@ def compute_effective_interval( return max(MIN_INTERVAL_SECONDS, min(MAX_INTERVAL_SECONDS, raw)) +async def set_platform_cooldown( + session: AsyncSession, platform: str, + seconds: int = PLATFORM_RATE_LIMIT_COOLDOWN_SECONDS, +) -> None: + """Stamp a cooldown expiry on the given platform so select_due_sources + skips every source on that platform until it expires. + + Called when a download surfaces ErrorType.RATE_LIMITED so the other + sources on the same platform don't all retry into the same rate limit. + Caller is responsible for committing the session. + + Uses INSERT...ON CONFLICT DO UPDATE so two concurrent workers hitting + the same platform's rate limit don't race: a SELECT-then-INSERT pattern + would let the loser's whole transaction (including the source-health + update + event finalize) roll back on a unique-violation, stranding + that event. Atomic upsert avoids that. + """ + now = datetime.now(UTC) + expires_at = (now + timedelta(seconds=seconds)).isoformat() + key = f"{PLATFORM_COOLDOWN_KEY_PREFIX}{platform}" + stmt = pg_insert(AppSetting.__table__).values( + key=key, value=expires_at, updated_at=now, + ).on_conflict_do_update( + index_elements=["key"], + set_={"value": expires_at, "updated_at": now}, + ) + await session.execute(stmt) + + +async def _platforms_in_cooldown(session: AsyncSession) -> dict[str, datetime]: + """Return {platform: expires_at} for platforms whose cooldown is still + in the future. Expired rows are ignored (a future maintenance sweep can + delete them; they don't affect routing decisions on their own).""" + rows = (await session.execute( + select(AppSetting.key, AppSetting.value) + .where(AppSetting.key.startswith(PLATFORM_COOLDOWN_KEY_PREFIX)) + )).all() + if not rows: + return {} + now = datetime.now(UTC) + active: dict[str, datetime] = {} + for key, value in rows: + try: + expires_at = datetime.fromisoformat(value) + except (ValueError, TypeError): + continue + if expires_at > now: + active[key[len(PLATFORM_COOLDOWN_KEY_PREFIX):]] = expires_at + return active + + async def select_due_sources(session: AsyncSession) -> list[Source]: """Sources where (enabled, artist.auto_check) and now >= last_checked_at + effective_interval. - Never-checked sources (last_checked_at IS NULL) are always due. + Never-checked sources (last_checked_at IS NULL) are always due. Sources + whose platform is currently in a rate-limit cooldown are excluded — the + cooldown is the preventive half of the burst-prevention pair (per-source + consecutive_failures backoff handles the offending source itself). """ rows = (await session.execute( select(Source) @@ -58,11 +125,14 @@ async def select_due_sources(session: AsyncSession) -> list[Source]: .where(Artist.auto_check.is_(True)) )).scalars().all() + cooldowns = await _platforms_in_cooldown(session) settings = await ImportSettings.load(session) now = datetime.now(UTC) due: list[Source] = [] for s in rows: + if s.platform in cooldowns: + continue interval = compute_effective_interval(s, s.artist, settings) if s.last_checked_at is None: due.append(s) @@ -134,9 +204,12 @@ async def scheduler_status(session: AsyncSession) -> dict: elif next_due_at is None or nca < next_due_at: next_due_at = nca + cooldowns = await _platforms_in_cooldown(session) + return { "last_tick_at": last_tick_at, "next_due_at": next_due_at.isoformat() if next_due_at else None, "due_now": due_now, "auto_sources": len(rows), + "platform_cooldowns": {p: dt.isoformat() for p, dt in cooldowns.items()}, } diff --git a/tests/test_scheduler_service.py b/tests/test_scheduler_service.py index 4c5e298..03aab3a 100644 --- a/tests/test_scheduler_service.py +++ b/tests/test_scheduler_service.py @@ -147,3 +147,113 @@ async def test_select_includes_past_due(db): await db.commit() due = await select_due_sources(db) assert any(s.url == "https://sel-past" for s in due) + + +# --- platform-rate-limit cooldown ----------------------------------------- + + +@pytest.mark.asyncio +async def test_select_skips_sources_on_platform_in_cooldown(db): + """After set_platform_cooldown('patreon'), select_due_sources excludes + every patreon source — even ones that would otherwise be past-due.""" + from backend.app.services.scheduler_service import set_platform_cooldown + + artist = await _seed_artist(db, interval=60, name="cooldown-p") + db.add(Source( + artist_id=artist.id, platform="patreon", url="https://cd-p", + enabled=True, consecutive_failures=0, + last_checked_at=datetime.now(UTC) - timedelta(seconds=3600), + )) + await db.commit() + + # Sanity-pre: due before the cooldown is set. + due = await select_due_sources(db) + assert any(s.url == "https://cd-p" for s in due) + + await set_platform_cooldown(db, "patreon") + await db.commit() + + due = await select_due_sources(db) + assert all(s.url != "https://cd-p" for s in due) + + +@pytest.mark.asyncio +async def test_cooldown_is_per_platform_other_platforms_unaffected(db): + """A cooldown on patreon doesn't block subscribestar sources from + being due — cooldowns are scoped to the platform that hit the limit.""" + from backend.app.services.scheduler_service import set_platform_cooldown + + artist = await _seed_artist(db, interval=60, name="cooldown-mixed") + db.add_all([ + Source( + artist_id=artist.id, platform="patreon", url="https://mix-p", + enabled=True, consecutive_failures=0, + last_checked_at=datetime.now(UTC) - timedelta(seconds=3600), + ), + Source( + artist_id=artist.id, platform="subscribestar", url="https://mix-ss", + enabled=True, consecutive_failures=0, + last_checked_at=datetime.now(UTC) - timedelta(seconds=3600), + ), + ]) + await db.commit() + + await set_platform_cooldown(db, "patreon") + await db.commit() + + due = await select_due_sources(db) + urls = {s.url for s in due} + assert "https://mix-ss" in urls + assert "https://mix-p" not in urls + + +@pytest.mark.asyncio +async def test_select_ignores_expired_cooldown(db): + """An expired cooldown row doesn't filter — routing treats expired + same as not-set. The stale row stays around until a future sweep + prunes it.""" + from backend.app.models import AppSetting + from backend.app.services.scheduler_service import ( + PLATFORM_COOLDOWN_KEY_PREFIX, + ) + + artist = await _seed_artist(db, interval=60, name="cooldown-expired") + db.add(Source( + artist_id=artist.id, platform="patreon", url="https://cd-exp", + enabled=True, consecutive_failures=0, + last_checked_at=datetime.now(UTC) - timedelta(seconds=3600), + )) + db.add(AppSetting( + key=f"{PLATFORM_COOLDOWN_KEY_PREFIX}patreon", + value=(datetime.now(UTC) - timedelta(seconds=10)).isoformat(), + )) + await db.commit() + + due = await select_due_sources(db) + assert any(s.url == "https://cd-exp" for s in due) + + +@pytest.mark.asyncio +async def test_set_platform_cooldown_upserts(db): + """Calling set_platform_cooldown twice on the same platform updates + the row in place rather than inserting a duplicate (AppSetting's PK + is `key`).""" + from sqlalchemy import select as sa_select + + from backend.app.models import AppSetting + from backend.app.services.scheduler_service import ( + PLATFORM_COOLDOWN_KEY_PREFIX, + set_platform_cooldown, + ) + + await set_platform_cooldown(db, "patreon", seconds=60) + await db.commit() + await set_platform_cooldown(db, "patreon", seconds=600) + await db.commit() + + rows = (await db.execute( + sa_select(AppSetting).where( + AppSetting.key == f"{PLATFORM_COOLDOWN_KEY_PREFIX}patreon" + ) + )).scalars().all() + assert len(rows) == 1