From 7d89bc8b8eff9eda8674fd5c444b22c4ba8c89be Mon Sep 17 00:00:00 2001 From: npeter83 Date: Tue, 18 Aug 2026 21:31:18 +0200 Subject: [PATCH] fix(downloads): exclude in-flight assets from the claim, drop the busy-wait (E3 S2, B-sub-3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two jobs on the same shared asset (same video+format queued twice, or by two users) starved the worker pool: worker A downloads the asset; worker B claims the sibling job, finds the asset `downloading`, requeues it, sleep(2)s, and re-claims it on the next tick — burning a whole worker slot behind one in-flight asset for the entire download (at concurrency=2, the queue stalls). Fix: the claim SQL now excludes any job whose asset is already `downloading` (`NOT EXISTS … media_assets.status='downloading'`), so a competing sibling simply waits until the asset settles and the workers pick up unrelated jobs meanwhile. The `_process_job` busy/lost-race branches drop their `time.sleep(2)` — the exclusion keeps a requeued job out of the claim window, so there's no spin. TDD: the new test failed on the old claim (claimed the sibling), passes now. Full lane green, ruff clean. --- backend/app/worker.py | 20 ++++++++++++--- backend/tests/test_download_limits.py | 35 +++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 6 deletions(-) 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)