fix(downloads): worker claim fairness + bounded job listings (R12 S5)
#2 — _claim_job excluded users at their concurrency limit only in Python, AFTER a fixed LIMIT-50 fetch window. One user flooding the queue with >50 queued jobs while at their cap filled the whole window, chosen came back None, and every other user's work past position 50 was never seen. Exclude capped users IN THE SQL (per-user max_concurrent from the quota row, else the sysconfig default; unlimited users never capped), keeping the Python re-check as a race guard that can only reject, never over-claim. #3 — list_downloads and shared_with_me loaded every job for a user with no bound. Cap both at 500 newest-first (a shared _MAX_LIST, matching the admin listing's existing cap); the Download Center renders newest-first and 500 is far beyond any real manual-download history, so no frontend change is needed. Tests: DB-backed — a capped user's 60-job flood no longer starves a user behind the window; oldest-by-queue_pos claim, all-capped->None, unlimited never excluded; and the list cap bounds the result set.
This commit is contained in:
@@ -40,6 +40,11 @@ admin_router = APIRouter(prefix="/api/admin/downloads", tags=["admin-downloads"]
|
||||
|
||||
_VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$")
|
||||
|
||||
# Cap the per-user job listings so a heavy library can't return an unbounded result set. The
|
||||
# Download Center renders newest-first; 500 is far beyond any real manual-download history (and
|
||||
# matches the admin listing's own cap).
|
||||
_MAX_LIST = 500
|
||||
|
||||
|
||||
# --- source + serialization ----------------------------------------------------------------
|
||||
|
||||
@@ -334,6 +339,7 @@ def list_downloads(
|
||||
select(DownloadJob)
|
||||
.where(DownloadJob.user_id == user.id)
|
||||
.order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc())
|
||||
.limit(_MAX_LIST)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
@@ -380,6 +386,7 @@ def shared_with_me(
|
||||
.join(DownloadShare, DownloadShare.job_id == DownloadJob.id)
|
||||
.where(DownloadShare.to_user_id == user.id)
|
||||
.order_by(DownloadShare.created_at.desc())
|
||||
.limit(_MAX_LIST)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
@@ -881,7 +888,7 @@ def admin_list_downloads(
|
||||
) -> list[dict]:
|
||||
jobs = (
|
||||
db.execute(
|
||||
select(DownloadJob).order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()).limit(500)
|
||||
select(DownloadJob).order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()).limit(_MAX_LIST)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
|
||||
+20
-3
@@ -177,19 +177,36 @@ def _recover_orphans() -> None:
|
||||
def _claim_job() -> int | None:
|
||||
"""Claim the oldest queued job whose user is under their per-user concurrency limit.
|
||||
|
||||
Fetches a window of locked candidates (FOR UPDATE SKIP LOCKED) and picks the first eligible
|
||||
one, so one user flooding the queue can't starve others beyond their max_concurrent."""
|
||||
Users already AT their concurrency limit are excluded IN THE SQL, so one user flooding the
|
||||
queue can't monopolize the fetch window and starve everyone behind them. (The old fixed
|
||||
LIMIT-50 window did exactly that: 50 queued jobs from a single capped user filled the window,
|
||||
`chosen` came back None, and every other user's work — sitting past position 50 — was never
|
||||
seen.) The per-user limit is the quota row's `max_concurrent` (else the sysconfig default);
|
||||
`unlimited` users are never capped. A final Python `at_concurrency_limit` re-check still guards
|
||||
the tiny race between the exclusion subquery and the claim (it can only ever reject, never
|
||||
over-claim), so correctness never depends on the two observations agreeing."""
|
||||
with SessionLocal() as db:
|
||||
default_mc = sysconfig.get_int(db, "download_default_max_concurrent")
|
||||
rows = db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT id, user_id FROM download_jobs
|
||||
WHERE status='queued'
|
||||
AND user_id NOT IN (
|
||||
SELECT r.user_id
|
||||
FROM download_jobs r
|
||||
LEFT JOIN download_quotas q ON q.user_id = r.user_id
|
||||
WHERE r.status='running' AND r.user_id IS NOT NULL
|
||||
GROUP BY r.user_id, q.max_concurrent, q.unlimited
|
||||
HAVING COALESCE(q.unlimited, false) = false
|
||||
AND count(*) >= COALESCE(q.max_concurrent, :dmc)
|
||||
)
|
||||
ORDER BY queue_pos, id
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 50
|
||||
"""
|
||||
)
|
||||
),
|
||||
{"dmc": default_mc},
|
||||
).fetchall()
|
||||
chosen = next(
|
||||
(jid for jid, uid in rows if not quota.at_concurrency_limit(db, uid)), None
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
"""DB-backed tests for R12 · S5 backend residue — the two download limits:
|
||||
|
||||
- **#2 claim fairness** (`worker._claim_job`): users already AT their per-user concurrency limit
|
||||
are excluded IN THE SQL, so one user flooding the queue past the fetch window can't starve
|
||||
everyone behind them. The old fixed `LIMIT 50` window did exactly that.
|
||||
- **#3 list cap** (`routes.downloads.list_downloads` / `shared_with_me`): the per-user listings
|
||||
are bounded so a heavy library can't return an unbounded result set.
|
||||
|
||||
Uses the `db` fixture (real Postgres) — skipped when no database is reachable. `_claim_job` runs in
|
||||
its OWN session, so every test COMMITS its setup before calling it.
|
||||
"""
|
||||
from app.models import DownloadJob, DownloadQuota, User
|
||||
from app.worker import _claim_job
|
||||
|
||||
|
||||
def _job(db, user_id, queue_pos, status="queued"):
|
||||
j = DownloadJob(
|
||||
user_id=user_id,
|
||||
source_kind="youtube",
|
||||
source_ref="x" * 11,
|
||||
status=status,
|
||||
queue_pos=queue_pos,
|
||||
profile_snapshot={},
|
||||
)
|
||||
db.add(j)
|
||||
return j
|
||||
|
||||
|
||||
def _quota(db, user_id, max_concurrent, unlimited=False):
|
||||
db.add(
|
||||
DownloadQuota(
|
||||
user_id=user_id,
|
||||
max_bytes=10**12,
|
||||
max_concurrent=max_concurrent,
|
||||
max_jobs=10**6,
|
||||
unlimited=unlimited,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
# --- #2 · claim fairness --------------------------------------------------------------------
|
||||
|
||||
def test_capped_user_flood_does_not_starve_users_behind_the_window(db):
|
||||
a = User(email="flooder@example.com")
|
||||
b = User(email="waiter@example.com")
|
||||
db.add_all([a, b])
|
||||
db.flush()
|
||||
_quota(db, a.id, max_concurrent=1) # A: 1 concurrent, already running one -> all A's queued are ineligible
|
||||
_quota(db, b.id, max_concurrent=3)
|
||||
_job(db, a.id, queue_pos=0, status="running")
|
||||
for i in range(1, 61): # A floods the FRONT with far more than the 50-row fetch window
|
||||
_job(db, a.id, queue_pos=i)
|
||||
bjob = _job(db, b.id, queue_pos=100) # B sits behind the flood
|
||||
db.commit()
|
||||
|
||||
assert _claim_job() == bjob.id # B is served despite A's 60-job flood filling the old window
|
||||
db.expire_all()
|
||||
assert db.get(DownloadJob, bjob.id).status == "running"
|
||||
|
||||
|
||||
def test_eligible_user_claims_oldest_by_queue_pos(db):
|
||||
u = User(email="solo@example.com")
|
||||
db.add(u)
|
||||
db.flush()
|
||||
_quota(db, u.id, max_concurrent=2)
|
||||
first = _job(db, u.id, queue_pos=1)
|
||||
_job(db, u.id, queue_pos=2)
|
||||
db.commit()
|
||||
assert _claim_job() == first.id
|
||||
|
||||
|
||||
def test_all_candidates_capped_returns_none(db):
|
||||
u = User(email="capped@example.com")
|
||||
db.add(u)
|
||||
db.flush()
|
||||
_quota(db, u.id, max_concurrent=1)
|
||||
_job(db, u.id, queue_pos=0, status="running") # at the cap
|
||||
_job(db, u.id, queue_pos=1) # queued but ineligible
|
||||
db.commit()
|
||||
assert _claim_job() is None
|
||||
|
||||
|
||||
def test_unlimited_user_is_never_excluded(db):
|
||||
u = User(email="unlimited@example.com")
|
||||
db.add(u)
|
||||
db.flush()
|
||||
_quota(db, u.id, max_concurrent=1, unlimited=True)
|
||||
for i in range(3): # 3 running, well past max_concurrent — but unlimited bypasses the cap
|
||||
_job(db, u.id, queue_pos=i, status="running")
|
||||
q = _job(db, u.id, queue_pos=10)
|
||||
db.commit()
|
||||
assert _claim_job() == q.id
|
||||
|
||||
|
||||
# --- #3 · list cap --------------------------------------------------------------------------
|
||||
|
||||
def test_list_downloads_is_capped(db, monkeypatch):
|
||||
from app.routes import downloads as dl
|
||||
|
||||
u = User(email="heavy@example.com")
|
||||
db.add(u)
|
||||
db.flush()
|
||||
for i in range(3):
|
||||
_job(db, u.id, queue_pos=i, status="done")
|
||||
db.commit()
|
||||
|
||||
monkeypatch.setattr(dl, "_MAX_LIST", 2)
|
||||
rows = dl.list_downloads(user=u, db=db)
|
||||
assert len(rows) == 2 # bounded by the cap, not the 3 rows on disk
|
||||
Reference in New Issue
Block a user