"""FC-3d: pure logic for deciding which sources are due for a check. The Celery tick task wraps `select_due_sources` and fires `download_source.delay()` per result; this module knows nothing about Celery, gallery-dl, or any side effect. """ 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 from ..models import AppSetting, Artist, ImportSettings, Source MIN_INTERVAL_SECONDS = 60 MAX_INTERVAL_SECONDS = 86400 MAX_BACKOFF_EXPONENT = 6 # AppSetting key stamped every time the Beat tick fires (see scan.py). The # tick runs every 60s; the UI flags the scheduler as stalled if the last # 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, ) -> int: """Return the seconds between scheduled checks for one source. Precedence: source.check_interval_override > artist.check_interval_seconds > settings.download_schedule_default_seconds. Multiplied by 2 ** min(consecutive_failures, MAX_BACKOFF_EXPONENT) and clamped to [MIN_INTERVAL_SECONDS, MAX_INTERVAL_SECONDS]. """ base = ( source.check_interval_override or artist.check_interval_seconds or settings.download_schedule_default_seconds ) exponent = min(max(0, source.consecutive_failures or 0), MAX_BACKOFF_EXPONENT) factor = 2 ** exponent raw = base * factor 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 active_platform_cooldowns(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). Exposed beyond scheduler_service so the manual check endpoint (`/api/sources//check`) can defer bulk retries that would bowl into the same rate limit the cooldown is preventing. """ 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. 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). Ordering: last_checked_at ASC NULLS FIRST, then id. Never-checked sources go first, then the longest-since-checked, so the most overdue sources hit Celery's FIFO download queue first. Anti-starvation: if queue throughput ever falls below the tick rate, a freshly-rerun source can't keep cutting in line ahead of one that hasn't been checked at all. Operator-confirmed 2026-05-30. """ rows = (await session.execute( select(Source) .options(selectinload(Source.artist)) .join(Artist, Source.artist_id == Artist.id) .where(Source.enabled.is_(True)) .where(Artist.auto_check.is_(True)) .order_by(Source.last_checked_at.asc().nulls_first(), Source.id) )).scalars().all() cooldowns = await active_platform_cooldowns(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) continue elapsed = (now - s.last_checked_at).total_seconds() if elapsed >= interval: due.append(s) return due def compute_next_check_at( source: Source, artist: Artist, settings: ImportSettings, ) -> datetime | None: """Return the projected datetime of the next check, or None if never checked.""" if source.last_checked_at is None: return None interval = compute_effective_interval(source, artist, settings) return source.last_checked_at + timedelta(seconds=interval) async def record_tick(session: AsyncSession) -> None: """Stamp the current time on the SCHEDULER_LAST_TICK_KEY AppSetting. Called once per Beat tick so the UI can prove the scheduler is alive. Commits its own write so the stamp survives even if the rest of the tick errors out. """ now_iso = datetime.now(UTC).isoformat() row = (await session.execute( select(AppSetting).where(AppSetting.key == SCHEDULER_LAST_TICK_KEY) )).scalar_one_or_none() if row is None: session.add(AppSetting(key=SCHEDULER_LAST_TICK_KEY, value=now_iso)) else: row.value = now_iso await session.commit() async def scheduler_status(session: AsyncSession) -> dict: """Summarise scheduler health for the dashboard. Returns last_tick_at (when Beat last fired), next_due_at (earliest upcoming scheduled check across enabled auto-check sources), due_now (how many are due right now), and auto_sources (total under schedule). """ last_tick_at = (await session.execute( select(AppSetting.value).where(AppSetting.key == SCHEDULER_LAST_TICK_KEY) )).scalar_one_or_none() rows = (await session.execute( select(Source) .options(selectinload(Source.artist)) .join(Artist, Source.artist_id == Artist.id) .where(Source.enabled.is_(True)) .where(Artist.auto_check.is_(True)) )).scalars().all() settings = await ImportSettings.load(session) now = datetime.now(UTC) due_now = 0 next_due_at: datetime | None = None for s in rows: if s.last_checked_at is None: due_now += 1 continue nca = compute_next_check_at(s, s.artist, settings) if nca is None or nca <= now: due_now += 1 elif next_due_at is None or nca < next_due_at: next_due_at = nca cooldowns = await active_platform_cooldowns(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()}, }