diff --git a/backend/app/api/downloads.py b/backend/app/api/downloads.py index ee648ef..c3ae536 100644 --- a/backend/app/api/downloads.py +++ b/backend/app/api/downloads.py @@ -187,3 +187,20 @@ async def get_download(event_id: int): return jsonify({"error": "not_found"}), 404 event, source, artist = row return jsonify(_detail_record(event, source, artist)) + + +@downloads_bp.route("/recover-stalled", methods=["POST"]) +async def recover_stalled(): + """Trigger the recover_stalled_download_events sweep on demand. + + The same sweep runs every 5 min via Beat (see celery_app.beat_schedule); + this endpoint exists so the operator can force-clear stuck pending/ + running download_events from the Subscriptions → Downloads maintenance + menu without waiting for the next scheduled tick. + """ + # Local import: avoids registering maintenance tasks during blueprint + # import (Celery task discovery races with the API import otherwise). + from ..tasks.maintenance import recover_stalled_download_events + + recover_stalled_download_events.delay() + return jsonify({"queued": True}), 202 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/gallery_dl.py b/backend/app/services/gallery_dl.py index e149477..2883bd4 100644 --- a/backend/app/services/gallery_dl.py +++ b/backend/app/services/gallery_dl.py @@ -42,6 +42,15 @@ class ErrorType(StrEnum): UNKNOWN_ERROR = "unknown_error" +# Pinned to download_source's Celery soft_time_limit (900s, see +# tasks/download.py:32). Anything larger and Celery kills the task before +# subprocess.run can raise TimeoutExpired — leaving the DownloadEvent +# stranded for the recovery sweep instead of capturing a clean timeout +# error. Per-source bumps live in source.config_overrides for legitimately +# long syncs. Operator-confirmed 2026-05-30 (~40-min hang investigation). +_DEFAULT_GDL_TIMEOUT_SECONDS = 900 + + @dataclass class SourceConfig: content_types: list[str] = field(default_factory=lambda: ["all"]) @@ -51,7 +60,7 @@ class SourceConfig: filename_pattern: str | None = None skip_existing: bool = True save_metadata: bool = True - timeout: int = 3600 + timeout: int = _DEFAULT_GDL_TIMEOUT_SECONDS @classmethod def from_dict(cls, data: dict) -> SourceConfig: @@ -63,7 +72,7 @@ class SourceConfig: filename_pattern=data.get("filename_pattern"), skip_existing=data.get("skip_existing", True), save_metadata=data.get("save_metadata", True), - timeout=data.get("timeout", 3600), + timeout=data.get("timeout", _DEFAULT_GDL_TIMEOUT_SECONDS), ) 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/frontend/src/components/subscriptions/DownloadsTab.vue b/frontend/src/components/subscriptions/DownloadsTab.vue index a1f5a15..0307c63 100644 --- a/frontend/src/components/subscriptions/DownloadsTab.vue +++ b/frontend/src/components/subscriptions/DownloadsTab.vue @@ -22,7 +22,10 @@ mdi-refresh Refresh - +
@@ -207,6 +210,24 @@ async function onRetryAll(sources) { toast({ text: parts.join(', ') || 'Nothing to retry', type: 'info' }) } +// Manual trigger for recover_stalled_download_events. The sweep runs on +// Beat every 5 min; this lets the operator force-clear stuck pending/ +// running events on demand. Fire-and-forget: the API returns 202 once the +// task is dispatched, and we refresh after a few seconds so swept rows +// show up in the failing rollup. +async function onRecoverStalled() { + try { + await store.recoverStalled() + toast({ + text: 'Recovery sweep queued — refreshing in a few seconds', + type: 'success', + }) + setTimeout(refresh, 4000) + } catch (e) { + toast({ text: `Sweep failed: ${e?.detail || e?.message || e}`, type: 'error' }) + } +} + // Live auto-refresh: while any download is queued or running, poll the // stats + first page every 4s so the operator can watch events succeed/ // fail in real time without hitting Refresh. Polling stops automatically diff --git a/frontend/src/components/subscriptions/MaintenanceMenu.vue b/frontend/src/components/subscriptions/MaintenanceMenu.vue index b89d045..ffae2c9 100644 --- a/frontend/src/components/subscriptions/MaintenanceMenu.vue +++ b/frontend/src/components/subscriptions/MaintenanceMenu.vue @@ -9,14 +9,14 @@ diff --git a/frontend/src/stores/downloads.js b/frontend/src/stores/downloads.js index 940d124..ee274c6 100644 --- a/frontend/src/stores/downloads.js +++ b/frontend/src/stores/downloads.js @@ -88,10 +88,17 @@ export const useDownloadsStore = defineStore('downloads', () => { return activeEvents.value } + // POSTs to the download-recovery sweep endpoint (fire-and-forget — the + // Beat schedule also runs it every 5 min). The caller should refresh + // failing/stats a few seconds after this resolves to see swept rows. + async function recoverStalled() { + return api.post('/api/downloads/recover-stalled') + } + return { events, cursor, hasMore, filter, selected, loading, error, stats, activity, failing, activeEvents, loadFirst, loadMore, loadOne, applyFilter, closeDetail, loadStats, - loadActivity, loadFailing, loadActive, + loadActivity, loadFailing, loadActive, recoverStalled, } }) diff --git a/tests/test_api_downloads.py b/tests/test_api_downloads.py index b20f53b..e61e927 100644 --- a/tests/test_api_downloads.py +++ b/tests/test_api_downloads.py @@ -134,3 +134,23 @@ async def test_activity_returns_fixed_bucket_array(client, seed): async def test_activity_rejects_bogus_hours(client): resp = await client.get("/api/downloads/activity?hours=bogus") assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_recover_stalled_dispatches_task(client, monkeypatch): + """POST /api/downloads/recover-stalled returns 202 and dispatches the + recover_stalled_download_events task. The task body itself is exercised + in test_maintenance.py — this just covers the API → Celery hand-off.""" + from backend.app.tasks import maintenance + + called: list[tuple] = [] + monkeypatch.setattr( + maintenance.recover_stalled_download_events, "delay", + lambda *a, **kw: called.append((a, kw)), + ) + + resp = await client.post("/api/downloads/recover-stalled") + assert resp.status_code == 202 + body = await resp.get_json() + assert body == {"queued": True} + assert len(called) == 1 diff --git a/tests/test_api_sources.py b/tests/test_api_sources.py index c48ed65..28d1a11 100644 --- a/tests/test_api_sources.py +++ b/tests/test_api_sources.py @@ -39,7 +39,10 @@ async def test_schedule_status_shape(client): resp = await client.get("/api/sources/schedule-status") assert resp.status_code == 200 body = await resp.get_json() - assert set(body) == {"last_tick_at", "next_due_at", "due_now", "auto_sources"} + assert set(body) == { + "last_tick_at", "next_due_at", "due_now", "auto_sources", + "platform_cooldowns", + } assert isinstance(body["due_now"], int) assert isinstance(body["auto_sources"], int) diff --git a/tests/test_api_system_activity.py b/tests/test_api_system_activity.py index f5832ef..8191a8e 100644 --- a/tests/test_api_system_activity.py +++ b/tests/test_api_system_activity.py @@ -60,7 +60,10 @@ async def test_summary_returns_rollup_shape(client, monkeypatch): assert body["queued_total"] == 3 # only ml has a non-zero depth assert isinstance(body["running"], int) assert isinstance(body["failing"], int) - assert set(body["scheduler"]) == {"last_tick_at", "next_due_at", "due_now", "auto_sources"} + assert set(body["scheduler"]) == { + "last_tick_at", "next_due_at", "due_now", "auto_sources", + "platform_cooldowns", + } @pytest.mark.asyncio 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