diff --git a/backend/tests/test_download_ref_count.py b/backend/tests/test_download_ref_count.py new file mode 100644 index 00000000..3c1bbe26 --- /dev/null +++ b/backend/tests/test_download_ref_count.py @@ -0,0 +1,116 @@ +"""E3 S2 characterization net — MediaAsset.ref_count enqueue/release accounting (B-sub-5). + +`ref_count` tracks how many DownloadJobs still hold a shared cache asset; when it hits zero a READY +asset's file + row are freed. Enqueue +1's it and release_asset -1's it. Today those are Python +read-modify-writes (`asset.ref_count = (asset.ref_count or 0) ± 1`), which lose an update under two +concurrent enqueues/releases and can delete a file out from under a live holder. The fix makes the +increment/decrement atomic at the DB; this net locks the OBSERVABLE single-threaded semantics so the +conversion is provably behaviour-equivalent (concurrency itself can't be exercised in one session). + +DB lane (`db` fixture) — skipped without Postgres. +""" +from sqlalchemy import select, text + +from app.downloads import service +from app.models import DownloadJob, MediaAsset, User + + +def _user(db, email="ref@example.com"): + u = User(email=email) + db.add(u) + db.flush() + return u + + +def _ref_count(db, asset_id) -> int | None: + # Read straight from the row — the atomic-UPDATE fix leaves the ORM copy stale on purpose. + return db.execute( + text("SELECT ref_count FROM media_assets WHERE id = :id"), {"id": asset_id} + ).scalar() + + +def _holding_asset(db, *, status="ready", ref_count=1, rel_path=None) -> MediaAsset: + asset = MediaAsset( + source_kind="url", source_ref=f"r{ref_count}-{status}", format_sig="sig", + status=status, ref_count=ref_count, rel_path=rel_path, + ) + db.add(asset) + db.flush() + return asset + + +def _job(db, user_id, asset_id, status="done") -> DownloadJob: + j = DownloadJob( + user_id=user_id, asset_id=asset_id, source_kind="url", source_ref="r", + status=status, queue_pos=1, profile_snapshot={}, + ) + db.add(j) + db.flush() + return j + + +def test_two_enqueues_share_one_asset_and_stack_ref_count(db): + u = _user(db) + j1 = service.enqueue(db, u.id, "url", "http://example.com/v", {}) + j2 = service.enqueue(db, u.id, "url", "http://example.com/v", {}) + # One shared cache asset, two jobs on it, ref_count == 2. + assert j1.asset_id == j2.asset_id + assets = db.execute(select(MediaAsset)).scalars().all() + assert len(assets) == 1 + assert _ref_count(db, j1.asset_id) == 2 + + +def test_release_keeps_asset_until_last_holder_then_deletes_row(db): + u = _user(db) + asset = _holding_asset(db, status="ready", ref_count=2) + a = _job(db, u.id, asset.id, status="done") + b = _job(db, u.id, asset.id, status="done") + db.commit() + + service.release_asset(db, a) + db.commit() + assert _ref_count(db, asset.id) == 1 + assert db.get(MediaAsset, asset.id) is not None # still held by b + + service.release_asset(db, b) + db.commit() + # Last holder gone → the ready asset's row is freed (rel_path None, so no file to unlink). + assert db.execute(select(MediaAsset)).scalars().all() == [] + + +def test_release_is_a_noop_for_a_non_holding_job(db): + u = _user(db) + asset = _holding_asset(db, status="ready", ref_count=1) + canceled = _job(db, u.id, asset.id, status="canceled") # canceled ∉ HOLDING + db.commit() + + service.release_asset(db, canceled) + db.commit() + assert _ref_count(db, asset.id) == 1 # untouched — a canceled job already released while holding + + +def test_release_never_drives_ref_count_negative(db): + u = _user(db) + # error-status asset (kept at zero, unlike ready) so we isolate the clamp from the delete branch. + asset = _holding_asset(db, status="error", ref_count=0) # already at zero (defensive) + j = _job(db, u.id, asset.id, status="error") # error ∈ HOLDING, so release runs + db.commit() + + service.release_asset(db, j) + db.commit() + # The count clamps at 0 (never negative) even when released with nothing left to release. + assert _ref_count(db, asset.id) == 0 + assert db.get(MediaAsset, asset.id) is not None + + +def test_errored_asset_is_not_deleted_at_zero(db): + u = _user(db) + asset = _holding_asset(db, status="error", ref_count=1) + j = _job(db, u.id, asset.id, status="error") + db.commit() + + service.release_asset(db, j) + db.commit() + # An errored row is cheap (no file) and may be re-enqueued → never deleted here, only ready is. + assert _ref_count(db, asset.id) == 0 + assert db.get(MediaAsset, asset.id) is not None