diff --git a/backend/app/worker.py b/backend/app/worker.py index 3aa9fe94..815c2859 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -176,7 +176,14 @@ def _recover_orphans() -> None: def _claim_job() -> int | None: - """Claim the oldest queued job whose user is under their per-user concurrency limit. + """Claim the oldest queued job whose user is under their per-user concurrency limit AND whose + shared asset isn't already being downloaded by another worker. + + The asset exclusion (`NOT EXISTS … media_assets.status='downloading'`) stops the same-asset ping- + pong (B-sub-3): without it, a second worker claims a queued job that shares an in-flight asset, + finds it busy, requeues it, and immediately re-claims it — burning a worker slot behind one + in-flight asset and starving everyone else. Excluded here, that job simply waits until its asset + settles, so `_process_job` no longer needs to sleep on the busy/lost-race paths. 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 @@ -193,6 +200,10 @@ def _claim_job() -> int | None: """ SELECT id, user_id FROM download_jobs WHERE status='queued' + AND NOT EXISTS ( + SELECT 1 FROM media_assets a + WHERE a.id = download_jobs.asset_id AND a.status = 'downloading' + ) AND user_id NOT IN ( SELECT r.user_id FROM download_jobs r @@ -463,15 +474,16 @@ def _process_job(job_id: int) -> None: _set_job(job_id, status="error", error=asset_error or "Download failed", phase=None) return if asset_status == "downloading": - # Another worker owns this asset; wait and let the job be reclaimed later. + # Another worker won this asset between our claim and here (a narrow race — the claim SQL + # excludes downloading-asset jobs). Requeue and let it be reclaimed once the asset settles; + # no sleep — the claim exclusion keeps it out of the window until then, so no busy-loop. _requeue_if_running(job_id, "waiting") - time.sleep(2) return # asset_status == "pending": try to win the download / edit. if not _claim_asset(asset_id): + # Lost the pending->downloading race; requeue (same reasoning — the exclusion prevents a spin). _requeue_if_running(job_id, "waiting") - time.sleep(2) return # The asset is now ours ('downloading'): run the heavy work in a supervised child process. diff --git a/backend/tests/test_download_limits.py b/backend/tests/test_download_limits.py index 1b1a1663..1c5edf1d 100644 --- a/backend/tests/test_download_limits.py +++ b/backend/tests/test_download_limits.py @@ -9,13 +9,14 @@ 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.models import DownloadJob, DownloadQuota, MediaAsset, User from app.worker import _claim_job -def _job(db, user_id, queue_pos, status="queued"): +def _job(db, user_id, queue_pos, status="queued", asset_id=None): j = DownloadJob( user_id=user_id, + asset_id=asset_id, source_kind="youtube", source_ref="x" * 11, status=status, @@ -26,6 +27,15 @@ def _job(db, user_id, queue_pos, status="queued"): return j +def _asset(db, status, ref): + a = MediaAsset( + source_kind="url", source_ref=ref, format_sig="s", status=status, ref_count=1 + ) + db.add(a) + db.flush() + return a + + def _quota(db, user_id, max_concurrent, unlimited=False): db.add( DownloadQuota( @@ -104,6 +114,27 @@ def test_stored_zero_concurrency_is_floored_not_starving(db): assert db.get(DownloadJob, claimed).user_id == a.id # A's own first job runs (floored to 1) +def test_job_whose_asset_is_already_downloading_is_skipped(db): + """B-sub-3: a queued job whose shared asset is ALREADY `downloading` (another worker owns it) must + not be claimed — otherwise a second worker claims it, finds the asset busy, requeues, and ping- + pongs on it, starving everyone behind. The claim SQL excludes it, so the job behind it (on a + different, free asset) is served instead.""" + u = User(email="asset-excl@example.com") + db.add(u) + db.flush() + _quota(db, u.id, max_concurrent=5) # high → the ASSET rule, not the per-user rule, does the excluding + busy = _asset(db, status="downloading", ref="busy") + free = _asset(db, status="pending", ref="free") + _job(db, u.id, queue_pos=1, status="running", asset_id=busy.id) # the holder actually downloading it + competitor = _job(db, u.id, queue_pos=2, asset_id=busy.id) # same asset — must be skipped + behind = _job(db, u.id, queue_pos=3, asset_id=free.id) # different asset — should be served + db.commit() + + assert _claim_job() == behind.id # the competitor on the busy asset was skipped + db.expire_all() + assert db.get(DownloadJob, competitor.id).status == "queued" # still queued, never claimed + + def test_unlimited_user_is_never_excluded(db): u = User(email="unlimited@example.com") db.add(u)