From 77f7a2341043773df4b7386cc75876125b2aeeb9 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 30 May 2026 12:08:43 -0400 Subject: [PATCH] =?UTF-8?q?feat(scheduler):=20order=20due=20sources=20by?= =?UTF-8?q?=20last=5Fchecked=5Fat=20=E2=80=94=20most=20overdue=20first?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/app/services/scheduler_service.py | 8 +++++ tests/test_scheduler_service.py | 39 +++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/backend/app/services/scheduler_service.py b/backend/app/services/scheduler_service.py index 3c42ca1..dd038a4 100644 --- a/backend/app/services/scheduler_service.py +++ b/backend/app/services/scheduler_service.py @@ -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 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) @@ -123,6 +130,7 @@ async def select_due_sources(session: AsyncSession) -> list[Source]: .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 _platforms_in_cooldown(session) diff --git a/tests/test_scheduler_service.py b/tests/test_scheduler_service.py index 03aab3a..e94c554 100644 --- a/tests/test_scheduler_service.py +++ b/tests/test_scheduler_service.py @@ -233,6 +233,45 @@ async def test_select_ignores_expired_cooldown(db): 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 async def test_set_platform_cooldown_upserts(db): """Calling set_platform_cooldown twice on the same platform updates