fc3d: scheduler_service — pure logic for due-source selection + backoff
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,80 @@
|
|||||||
|
"""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.ext.asyncio import AsyncSession
|
||||||
|
from sqlalchemy.orm import selectinload
|
||||||
|
|
||||||
|
from ..models import Artist, ImportSettings, Source
|
||||||
|
|
||||||
|
MIN_INTERVAL_SECONDS = 60
|
||||||
|
MAX_INTERVAL_SECONDS = 86400
|
||||||
|
MAX_BACKOFF_EXPONENT = 6
|
||||||
|
|
||||||
|
|
||||||
|
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 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.
|
||||||
|
"""
|
||||||
|
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 session.execute(
|
||||||
|
select(ImportSettings).where(ImportSettings.id == 1)
|
||||||
|
)).scalar_one()
|
||||||
|
|
||||||
|
now = datetime.now(UTC)
|
||||||
|
due: list[Source] = []
|
||||||
|
for s in rows:
|
||||||
|
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)
|
||||||
@@ -0,0 +1,149 @@
|
|||||||
|
"""FC-3d: scheduler_service unit tests.
|
||||||
|
|
||||||
|
Covers compute_effective_interval (override/artist/default precedence,
|
||||||
|
backoff exponent at failure counts 0/1/3/6/10, floor and ceiling clamp)
|
||||||
|
and select_due_sources (auto_check off, never-checked, enabled=False,
|
||||||
|
due/not-due boundary).
|
||||||
|
"""
|
||||||
|
from datetime import UTC, datetime, timedelta
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from backend.app.models import Artist, ImportSettings, Source
|
||||||
|
from backend.app.services.scheduler_service import (
|
||||||
|
compute_effective_interval,
|
||||||
|
select_due_sources,
|
||||||
|
)
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.integration
|
||||||
|
|
||||||
|
|
||||||
|
# --- compute_effective_interval (pure, no DB) -----------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def _src(override=None, failures=0):
|
||||||
|
return Source(
|
||||||
|
artist_id=1, platform="patreon", url="https://x", enabled=True,
|
||||||
|
check_interval_override=override, consecutive_failures=failures,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _art(interval=None, auto=True):
|
||||||
|
return Artist(name="A", slug="a", auto_check=auto, check_interval_seconds=interval)
|
||||||
|
|
||||||
|
|
||||||
|
def _settings(default=28800):
|
||||||
|
return ImportSettings(id=1, download_schedule_default_seconds=default)
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_uses_source_override_when_set():
|
||||||
|
assert compute_effective_interval(_src(override=600), _art(interval=3600), _settings()) == 600
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_falls_back_to_artist_interval():
|
||||||
|
assert compute_effective_interval(_src(), _art(interval=3600), _settings()) == 3600
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_falls_back_to_global_default():
|
||||||
|
assert compute_effective_interval(_src(), _art(), _settings(default=900)) == 900
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_backoff_zero_failures_is_identity():
|
||||||
|
assert compute_effective_interval(_src(override=600), _art(), _settings()) == 600
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_backoff_one_failure_doubles():
|
||||||
|
assert compute_effective_interval(_src(override=600, failures=1), _art(), _settings()) == 1200
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_backoff_three_failures_x8():
|
||||||
|
assert compute_effective_interval(_src(override=600, failures=3), _art(), _settings()) == 4800
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_backoff_caps_at_exponent_six():
|
||||||
|
# 600 * 2^6 = 38400; 2^7 would be 76800 but exponent caps at 6.
|
||||||
|
assert compute_effective_interval(_src(override=600, failures=6), _art(), _settings()) == 38400
|
||||||
|
assert compute_effective_interval(_src(override=600, failures=10), _art(), _settings()) == 38400
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_floor_at_60_seconds():
|
||||||
|
# base 30 with 0 failures would be 30s; floor lifts to 60.
|
||||||
|
assert compute_effective_interval(_src(override=30), _art(), _settings()) == 60
|
||||||
|
|
||||||
|
|
||||||
|
def test_interval_ceiling_at_86400_seconds():
|
||||||
|
# base 28800 * 2^6 = 1843200; ceiling caps at 86400.
|
||||||
|
assert compute_effective_interval(_src(failures=6), _art(), _settings(default=28800)) == 86400
|
||||||
|
|
||||||
|
|
||||||
|
# --- select_due_sources (real session) ------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
async def _seed_artist(db, *, auto=True, interval=None, name="A"):
|
||||||
|
a = Artist(name=name, slug=name.lower(), auto_check=auto, check_interval_seconds=interval)
|
||||||
|
db.add(a)
|
||||||
|
await db.flush()
|
||||||
|
return a
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_select_skips_disabled_sources(db):
|
||||||
|
artist = await _seed_artist(db, name="sel-disabled")
|
||||||
|
db.add(Source(
|
||||||
|
artist_id=artist.id, platform="patreon", url="https://sel-disabled",
|
||||||
|
enabled=False, consecutive_failures=0,
|
||||||
|
))
|
||||||
|
await db.commit()
|
||||||
|
due = await select_due_sources(db)
|
||||||
|
assert all(s.url != "https://sel-disabled" for s in due)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_select_skips_artist_auto_check_false(db):
|
||||||
|
artist = await _seed_artist(db, auto=False, name="sel-noauto")
|
||||||
|
db.add(Source(
|
||||||
|
artist_id=artist.id, platform="patreon", url="https://sel-noauto",
|
||||||
|
enabled=True, consecutive_failures=0,
|
||||||
|
))
|
||||||
|
await db.commit()
|
||||||
|
due = await select_due_sources(db)
|
||||||
|
assert all(s.url != "https://sel-noauto" for s in due)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_select_includes_never_checked(db):
|
||||||
|
artist = await _seed_artist(db, name="sel-never")
|
||||||
|
src = Source(
|
||||||
|
artist_id=artist.id, platform="patreon", url="https://sel-never",
|
||||||
|
enabled=True, consecutive_failures=0,
|
||||||
|
)
|
||||||
|
db.add(src)
|
||||||
|
await db.commit()
|
||||||
|
due = await select_due_sources(db)
|
||||||
|
assert any(s.url == "https://sel-never" for s in due)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_select_excludes_not_yet_due(db):
|
||||||
|
artist = await _seed_artist(db, interval=3600, name="sel-recent")
|
||||||
|
db.add(Source(
|
||||||
|
artist_id=artist.id, platform="patreon", url="https://sel-recent",
|
||||||
|
enabled=True, consecutive_failures=0,
|
||||||
|
last_checked_at=datetime.now(UTC) - timedelta(seconds=60),
|
||||||
|
))
|
||||||
|
await db.commit()
|
||||||
|
due = await select_due_sources(db)
|
||||||
|
assert all(s.url != "https://sel-recent" for s in due)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_select_includes_past_due(db):
|
||||||
|
artist = await _seed_artist(db, interval=60, name="sel-past")
|
||||||
|
db.add(Source(
|
||||||
|
artist_id=artist.id, platform="patreon", url="https://sel-past",
|
||||||
|
enabled=True, consecutive_failures=0,
|
||||||
|
last_checked_at=datetime.now(UTC) - timedelta(seconds=3600),
|
||||||
|
))
|
||||||
|
await db.commit()
|
||||||
|
due = await select_due_sources(db)
|
||||||
|
assert any(s.url == "https://sel-past" for s in due)
|
||||||
Reference in New Issue
Block a user