peter
e824fe0ae9
Locks the observable single-threaded ref_count accounting before the atomic-SQL conversion: two enqueues of one (source,format) share a single asset and stack ref_count to 2; release keeps the asset until the last holder then frees the ready row; a non-holding (canceled) job's release is a no-op; the decrement clamps at 0; an errored asset is kept at zero (only ready is freed). Green on the current RMW code (5 pass) — the fix must keep every assertion.
117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
"""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
|