diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index aa10e5ce..b8e91780 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -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() diff --git a/backend/app/worker.py b/backend/app/worker.py index 96c56956..48cdfcf8 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -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 diff --git a/backend/tests/test_download_limits.py b/backend/tests/test_download_limits.py new file mode 100644 index 00000000..84cd198e --- /dev/null +++ b/backend/tests/test_download_limits.py @@ -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