feat(scheduler): platform-wide cooldown on RATE_LIMITED — burst prevention

The scan tick fired download_source.delay() for every due source without
grouping by platform; with multiple download workers, N due Patreon
sources could all hit Patreon's API in parallel and rate-limit each
other. Per-source consecutive_failures backoff REACTS to that (slows the
offender across cycles) but didn't PREVENT the first-tick burst.

When DownloadService._update_source_health sees a source error
classified as ErrorType.RATE_LIMITED, it now stamps an AppSetting row
`platform_cooldown:<platform>` with the cooldown expiry (now + 15 min,
PLATFORM_RATE_LIMIT_COOLDOWN_SECONDS). select_due_sources queries every
platform_cooldown:* key at the start of each tick and excludes every
source whose platform is in active cooldown. scheduler_status surfaces
active cooldowns as platform_cooldowns: {platform: expires_iso} so the
TopNav pipeline chip / activity summary can display them.

INSERT...ON CONFLICT DO UPDATE for the upsert so two workers racing
RATE_LIMITED responses on the same platform don't let one's
IntegrityError roll back the other's event-finalize transaction
(stranding the event for the recovery sweep). Atomic at the SQL level.

Tests cover: select_due_sources skips a platform in cooldown; other
platforms unaffected during single-platform cooldown; expired cooldown
rows don't filter; set_platform_cooldown is upsert-safe under repeated
calls.

Operator-flagged 2026-05-30 ("running multiple workers I don't know how
we'd keep the downloader from hitting a rate limit on a source").

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

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

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()},
     }

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