feat(scheduler): order due sources by last_checked_at — most overdue first
select_due_sources returned rows in undefined order (Postgres-determined, typically PK). At tick rates that outpace download-queue throughput, a freshly-rerun source could keep getting re-queued ahead of one that's still waiting for its first attempt this cycle. Operator-flagged 2026-05-30: > if there are 8 hours before a source is due again and 40 full time > downloads can happen in that period that means that there's a chance > the first one to fire gets back into the download queue before item 41 > has a chance to get downloaded. Added `ORDER BY last_checked_at ASC NULLS FIRST, id` to the due-source SELECT. Never-checked sources go first, then longest-since-checked, then ties broken by id. Combined with Celery's FIFO `download` queue, the oldest-overdue source in each tick now reaches a worker before any fresher one. Test pins the ordering: a NULL-last_checked source, a 4-hour-overdue source, and a 2-min-overdue source come back in that exact order from select_due_sources. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -116,6 +116,13 @@ async def select_due_sources(session: AsyncSession) -> list[Source]:
|
|||||||
whose platform is currently in a rate-limit cooldown are excluded — the
|
whose platform is currently in a rate-limit cooldown are excluded — the
|
||||||
cooldown is the preventive half of the burst-prevention pair (per-source
|
cooldown is the preventive half of the burst-prevention pair (per-source
|
||||||
consecutive_failures backoff handles the offending source itself).
|
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(
|
rows = (await session.execute(
|
||||||
select(Source)
|
select(Source)
|
||||||
@@ -123,6 +130,7 @@ async def select_due_sources(session: AsyncSession) -> list[Source]:
|
|||||||
.join(Artist, Source.artist_id == Artist.id)
|
.join(Artist, Source.artist_id == Artist.id)
|
||||||
.where(Source.enabled.is_(True))
|
.where(Source.enabled.is_(True))
|
||||||
.where(Artist.auto_check.is_(True))
|
.where(Artist.auto_check.is_(True))
|
||||||
|
.order_by(Source.last_checked_at.asc().nulls_first(), Source.id)
|
||||||
)).scalars().all()
|
)).scalars().all()
|
||||||
|
|
||||||
cooldowns = await _platforms_in_cooldown(session)
|
cooldowns = await _platforms_in_cooldown(session)
|
||||||
|
|||||||
@@ -233,6 +233,45 @@ async def test_select_ignores_expired_cooldown(db):
|
|||||||
assert any(s.url == "https://cd-exp" for s in due)
|
assert any(s.url == "https://cd-exp" for s in due)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_select_orders_most_overdue_first_then_id(db):
|
||||||
|
"""Within the due set, sources are ordered by last_checked_at ASC NULLS
|
||||||
|
FIRST. Combined with Celery FIFO on the download queue, the most-overdue
|
||||||
|
source in each tick reaches a worker first — preventing the 'a freshly-
|
||||||
|
rerun source keeps cutting in line ahead of one that's still waiting'
|
||||||
|
starvation pattern when queue throughput is below the tick population."""
|
||||||
|
artist = await _seed_artist(db, interval=60, name="order-test")
|
||||||
|
now = datetime.now(UTC)
|
||||||
|
# Newest check — least overdue (but still past its 60s interval).
|
||||||
|
s_new = Source(
|
||||||
|
artist_id=artist.id, platform="patreon", url="https://order-new",
|
||||||
|
enabled=True, consecutive_failures=0,
|
||||||
|
last_checked_at=now - timedelta(minutes=2),
|
||||||
|
)
|
||||||
|
# Oldest check — most overdue among checked sources.
|
||||||
|
s_old = Source(
|
||||||
|
artist_id=artist.id, platform="patreon", url="https://order-old",
|
||||||
|
enabled=True, consecutive_failures=0,
|
||||||
|
last_checked_at=now - timedelta(hours=4),
|
||||||
|
)
|
||||||
|
# Never checked — wins the ordering (NULLS FIRST).
|
||||||
|
s_never = Source(
|
||||||
|
artist_id=artist.id, platform="patreon", url="https://order-never",
|
||||||
|
enabled=True, consecutive_failures=0,
|
||||||
|
)
|
||||||
|
db.add_all([s_new, s_old, s_never])
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
due = await select_due_sources(db)
|
||||||
|
# Filter to just our test seeds (concurrent tests may add others).
|
||||||
|
urls = [s.url for s in due if s.url.startswith("https://order-")]
|
||||||
|
assert urls == [
|
||||||
|
"https://order-never",
|
||||||
|
"https://order-old",
|
||||||
|
"https://order-new",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_set_platform_cooldown_upserts(db):
|
async def test_set_platform_cooldown_upserts(db):
|
||||||
"""Calling set_platform_cooldown twice on the same platform updates
|
"""Calling set_platform_cooldown twice on the same platform updates
|
||||||
|
|||||||
Reference in New Issue
Block a user