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