feat(downloads): bulk retry respects cooldown; single-source RETRY overrides
Today's platform-cooldown commit (61ce1ce) only filtered the scan tick
— manual /api/sources/<id>/check still bypassed it. Operator-flagged
2026-05-30: clicked "Retry failed" on a Patreon failure pile and saw
every one go 'queued' without realising the cooldown wasn't in the
loop. Bulk retry with N sources on a cooled-down platform bowls right
back into the rate limit the cooldown is trying to prevent.
**Backend (`/api/sources/<id>/check`):**
- Reads optional `?force=true` query flag.
- Without force: queries `active_platform_cooldowns` (renamed from the
private `_platforms_in_cooldown` since it's now a cross-module API).
If the source's platform is in cooldown, returns **202** with
`{status: 'deferred', platform, cooldown_until}` — no event created,
no dispatch.
- With force: cooldown skipped entirely.
- In-flight guard always applies (no point creating duplicate pendings).
**Frontend (`sourcesStore.checkNow(id, {force=false})`):** new optional
`force` flag → adds `?force=true` to the URL.
**Frontend (`DownloadsTab`):**
- `onRetrySource` (single-source RETRY click): passes `force: true` →
explicit operator override, useful for rapid auth-fix testing.
- `onRetryAll` (RETRY ALL + MaintenanceMenu "Retry failed"): no force →
cooldown respected. Tallies `deferred` alongside `queued` /
`already_running`; toast reads e.g. *"5 queued, 12 deferred
(cooldown), 3 already running"*. That count is the operator's
diagnostic answer for "is rate-limit the cause of most failures?"
(12-of-20 deferred → yes; 0 deferred → no).
**Auto-resume:** no new sweep needed. Deferred sources still have stale
`last_checked_at`, so the next scan tick after the cooldown AppSetting
expires picks them up via `select_due_sources` (which already filters
on `active_platform_cooldowns`).
Tests: two new — deferred-on-cooldown returns 202 with the right body
and no dispatch; force=true overrides the cooldown and creates the
event normally.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -5,7 +5,7 @@ from sqlalchemy import select
|
|||||||
|
|
||||||
from ..extensions import get_session
|
from ..extensions import get_session
|
||||||
from ..models import DownloadEvent, Source
|
from ..models import DownloadEvent, Source
|
||||||
from ..services.scheduler_service import scheduler_status
|
from ..services.scheduler_service import active_platform_cooldowns, scheduler_status
|
||||||
from ..services.source_service import (
|
from ..services.source_service import (
|
||||||
KNOWN_PLATFORMS,
|
KNOWN_PLATFORMS,
|
||||||
ArtistNotFoundError,
|
ArtistNotFoundError,
|
||||||
@@ -125,7 +125,16 @@ async def check_source(source_id: int):
|
|||||||
"""FC-3c: enqueue a download for this source.
|
"""FC-3c: enqueue a download for this source.
|
||||||
|
|
||||||
Returns 202 with the new DownloadEvent id. If a pending/running
|
Returns 202 with the new DownloadEvent id. If a pending/running
|
||||||
event already exists for this source, returns 409 with that id."""
|
event already exists for this source, returns 409 with that id. If
|
||||||
|
the source's platform is currently in a rate-limit cooldown, returns
|
||||||
|
**202 with `{status: "deferred", cooldown_until, platform}`** and
|
||||||
|
does NOT create an event or dispatch — the bulk retry path uses this
|
||||||
|
to avoid bowling N sources right back into the rate limit the
|
||||||
|
cooldown is preventing. Single-click "retry this one source" passes
|
||||||
|
`?force=true` to override the cooldown (operator-explicit, useful
|
||||||
|
for rapid auth-fix testing). The in-flight guard always applies.
|
||||||
|
"""
|
||||||
|
force = (request.args.get("force") or "").lower() in ("1", "true", "yes")
|
||||||
async with get_session() as session:
|
async with get_session() as session:
|
||||||
source = (await session.execute(
|
source = (await session.execute(
|
||||||
select(Source).where(Source.id == source_id)
|
select(Source).where(Source.id == source_id)
|
||||||
@@ -135,6 +144,19 @@ async def check_source(source_id: int):
|
|||||||
if not source.enabled:
|
if not source.enabled:
|
||||||
return _bad("source_disabled", detail="enable the source first")
|
return _bad("source_disabled", detail="enable the source first")
|
||||||
|
|
||||||
|
# Cooldown gate (unless explicitly overridden). Checked before the
|
||||||
|
# in-flight guard because a deferred retry doesn't need to create
|
||||||
|
# or check for an event at all.
|
||||||
|
if not force:
|
||||||
|
cooldowns = await active_platform_cooldowns(session)
|
||||||
|
expires_at = cooldowns.get(source.platform)
|
||||||
|
if expires_at is not None:
|
||||||
|
return jsonify({
|
||||||
|
"status": "deferred",
|
||||||
|
"platform": source.platform,
|
||||||
|
"cooldown_until": expires_at.isoformat(),
|
||||||
|
}), 202
|
||||||
|
|
||||||
in_flight = (await session.execute(
|
in_flight = (await session.execute(
|
||||||
select(DownloadEvent.id).where(
|
select(DownloadEvent.id).where(
|
||||||
DownloadEvent.source_id == source_id,
|
DownloadEvent.source_id == source_id,
|
||||||
|
|||||||
@@ -87,10 +87,15 @@ async def set_platform_cooldown(
|
|||||||
await session.execute(stmt)
|
await session.execute(stmt)
|
||||||
|
|
||||||
|
|
||||||
async def _platforms_in_cooldown(session: AsyncSession) -> dict[str, datetime]:
|
async def active_platform_cooldowns(session: AsyncSession) -> dict[str, datetime]:
|
||||||
"""Return {platform: expires_at} for platforms whose cooldown is still
|
"""Return {platform: expires_at} for platforms whose cooldown is still
|
||||||
in the future. Expired rows are ignored (a future maintenance sweep can
|
in the future. Expired rows are ignored (a future maintenance sweep can
|
||||||
delete them; they don't affect routing decisions on their own)."""
|
delete them; they don't affect routing decisions on their own).
|
||||||
|
|
||||||
|
Exposed beyond scheduler_service so the manual check endpoint
|
||||||
|
(`/api/sources/<id>/check`) can defer bulk retries that would bowl
|
||||||
|
into the same rate limit the cooldown is preventing.
|
||||||
|
"""
|
||||||
rows = (await session.execute(
|
rows = (await session.execute(
|
||||||
select(AppSetting.key, AppSetting.value)
|
select(AppSetting.key, AppSetting.value)
|
||||||
.where(AppSetting.key.startswith(PLATFORM_COOLDOWN_KEY_PREFIX))
|
.where(AppSetting.key.startswith(PLATFORM_COOLDOWN_KEY_PREFIX))
|
||||||
@@ -133,7 +138,7 @@ async def select_due_sources(session: AsyncSession) -> list[Source]:
|
|||||||
.order_by(Source.last_checked_at.asc().nulls_first(), Source.id)
|
.order_by(Source.last_checked_at.asc().nulls_first(), Source.id)
|
||||||
)).scalars().all()
|
)).scalars().all()
|
||||||
|
|
||||||
cooldowns = await _platforms_in_cooldown(session)
|
cooldowns = await active_platform_cooldowns(session)
|
||||||
settings = await ImportSettings.load(session)
|
settings = await ImportSettings.load(session)
|
||||||
|
|
||||||
now = datetime.now(UTC)
|
now = datetime.now(UTC)
|
||||||
@@ -212,7 +217,7 @@ async def scheduler_status(session: AsyncSession) -> dict:
|
|||||||
elif next_due_at is None or nca < next_due_at:
|
elif next_due_at is None or nca < next_due_at:
|
||||||
next_due_at = nca
|
next_due_at = nca
|
||||||
|
|
||||||
cooldowns = await _platforms_in_cooldown(session)
|
cooldowns = await active_platform_cooldowns(session)
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"last_tick_at": last_tick_at,
|
"last_tick_at": last_tick_at,
|
||||||
|
|||||||
@@ -171,10 +171,12 @@ async function refresh() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Retry a single failing source (re-runs its whole feed) then refresh the
|
// Retry a single failing source (re-runs its whole feed) then refresh the
|
||||||
// rollup + stats so the operator sees it move.
|
// rollup + stats so the operator sees it move. Passes force=true so the
|
||||||
|
// platform cooldown is bypassed — single-source click is an explicit
|
||||||
|
// operator override, useful for rapid auth-fix or fixture testing.
|
||||||
async function onRetrySource(source) {
|
async function onRetrySource(source) {
|
||||||
try {
|
try {
|
||||||
await sourcesStore.checkNow(source.id)
|
await sourcesStore.checkNow(source.id, { force: true })
|
||||||
toast({ text: `Retry queued for ${source.artist_name || source.platform}`, type: 'success' })
|
toast({ text: `Retry queued for ${source.artist_name || source.platform}`, type: 'success' })
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
if (e?.body?.download_event_id) {
|
if (e?.body?.download_event_id) {
|
||||||
@@ -187,15 +189,24 @@ async function onRetrySource(source) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Bulk retry — leaves cooldown enforcement ON so N failing sources on
|
||||||
|
// the same platform don't all retry into the rate limit the cooldown is
|
||||||
|
// preventing. Sources deferred by cooldown will be picked up by the
|
||||||
|
// next scan tick after the AppSetting expires. Toast tallies the three
|
||||||
|
// outcomes so the operator can quickly read whether cooldown is the
|
||||||
|
// dominant failure mode ("12 deferred (cooldown)" → yes, rate limit is
|
||||||
|
// the issue).
|
||||||
async function onRetryAll(sources) {
|
async function onRetryAll(sources) {
|
||||||
retryingAll.value = true
|
retryingAll.value = true
|
||||||
let ok = 0
|
let ok = 0
|
||||||
let conflict = 0
|
let conflict = 0
|
||||||
|
let deferred = 0
|
||||||
try {
|
try {
|
||||||
for (const s of sources) {
|
for (const s of sources) {
|
||||||
try {
|
try {
|
||||||
await sourcesStore.checkNow(s.id)
|
const body = await sourcesStore.checkNow(s.id)
|
||||||
ok += 1
|
if (body?.status === 'deferred') deferred += 1
|
||||||
|
else ok += 1
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
if (e?.body?.download_event_id) conflict += 1
|
if (e?.body?.download_event_id) conflict += 1
|
||||||
}
|
}
|
||||||
@@ -206,6 +217,7 @@ async function onRetryAll(sources) {
|
|||||||
}
|
}
|
||||||
const parts = []
|
const parts = []
|
||||||
if (ok) parts.push(`${ok} queued`)
|
if (ok) parts.push(`${ok} queued`)
|
||||||
|
if (deferred) parts.push(`${deferred} deferred (cooldown)`)
|
||||||
if (conflict) parts.push(`${conflict} already running`)
|
if (conflict) parts.push(`${conflict} already running`)
|
||||||
toast({ text: parts.join(', ') || 'Nothing to retry', type: 'info' })
|
toast({ text: parts.join(', ') || 'Nothing to retry', type: 'info' })
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -67,10 +67,17 @@ export const useSourcesStore = defineStore('sources', () => {
|
|||||||
// FC-3c: trigger a download for one source. Returns {download_event_id,status}.
|
// FC-3c: trigger a download for one source. Returns {download_event_id,status}.
|
||||||
const checkingIds = ref(new Set())
|
const checkingIds = ref(new Set())
|
||||||
|
|
||||||
async function checkNow(id) {
|
// force=true bypasses the platform-rate-limit cooldown gate. Single-
|
||||||
|
// source RETRY clicks pass it (operator-explicit override, useful for
|
||||||
|
// rapid auth-fix testing); bulk RETRY ALL / MaintenanceMenu retries
|
||||||
|
// leave it off so the cooldown does its preventive job.
|
||||||
|
async function checkNow(id, { force = false } = {}) {
|
||||||
checkingIds.value = new Set(checkingIds.value).add(id)
|
checkingIds.value = new Set(checkingIds.value).add(id)
|
||||||
try {
|
try {
|
||||||
return await api.post(`/api/sources/${id}/check`)
|
const url = force
|
||||||
|
? `/api/sources/${id}/check?force=true`
|
||||||
|
: `/api/sources/${id}/check`
|
||||||
|
return await api.post(url)
|
||||||
} finally {
|
} finally {
|
||||||
const next = new Set(checkingIds.value)
|
const next = new Set(checkingIds.value)
|
||||||
next.delete(id)
|
next.delete(id)
|
||||||
|
|||||||
@@ -63,3 +63,55 @@ async def test_check_409_when_already_running(client, db, seed, monkeypatch):
|
|||||||
assert resp.status_code == 409
|
assert resp.status_code == 409
|
||||||
body = await resp.get_json()
|
body = await resp.get_json()
|
||||||
assert body["download_event_id"] == existing.id
|
assert body["download_event_id"] == existing.id
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_check_defers_when_platform_in_cooldown(client, db, seed, monkeypatch):
|
||||||
|
"""Bulk retries land here without ?force — when the source's platform
|
||||||
|
has an active cooldown, return 202 with status='deferred' instead of
|
||||||
|
creating an event. Lets the bulk path tally deferred-vs-queued in
|
||||||
|
the toast so the operator can see rate-limit-induced failures at a
|
||||||
|
glance."""
|
||||||
|
from backend.app.services.scheduler_service import set_platform_cooldown
|
||||||
|
|
||||||
|
delays: list = []
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"backend.app.tasks.download.download_source.delay",
|
||||||
|
lambda *a, **k: delays.append(a),
|
||||||
|
)
|
||||||
|
|
||||||
|
await set_platform_cooldown(db, seed.platform, seconds=900)
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
resp = await client.post(f"/api/sources/{seed.id}/check")
|
||||||
|
assert resp.status_code == 202
|
||||||
|
body = await resp.get_json()
|
||||||
|
assert body["status"] == "deferred"
|
||||||
|
assert body["platform"] == seed.platform
|
||||||
|
assert body["cooldown_until"]
|
||||||
|
assert delays == [] # no dispatch when deferred
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_check_force_overrides_cooldown(client, db, seed, monkeypatch):
|
||||||
|
"""Single-source RETRY click sends ?force=true — explicit operator
|
||||||
|
override, used for rapid auth-fix testing without waiting on the
|
||||||
|
cooldown."""
|
||||||
|
from backend.app.services.scheduler_service import set_platform_cooldown
|
||||||
|
|
||||||
|
delays: list = []
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"backend.app.tasks.download.download_source.delay",
|
||||||
|
lambda *a, **k: delays.append(a),
|
||||||
|
)
|
||||||
|
|
||||||
|
await set_platform_cooldown(db, seed.platform, seconds=900)
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
resp = await client.post(f"/api/sources/{seed.id}/check?force=true")
|
||||||
|
assert resp.status_code == 202
|
||||||
|
body = await resp.get_json()
|
||||||
|
assert body["status"] == "pending"
|
||||||
|
assert isinstance(body["download_event_id"], int)
|
||||||
|
assert len(delays) == 1
|
||||||
|
assert delays[0] == (seed.id,)
|
||||||
|
|||||||
Reference in New Issue
Block a user