fix(downloads): exclude in-flight assets from the claim, drop the busy-wait (E3 S2, B-sub-3)

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

@@ -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.

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)