fix(downloads): atomic ref_count increment/decrement (E3 S2, B-sub-5)
MediaAsset.ref_count (how many jobs hold a shared cache asset) was a Python read-modify-write at four sites (enqueue, enqueue_edit, release_asset, the redownload route). Two concurrent enqueues of the same (source,format) both read the same value and write the same +1 → the count ends up one short → the first release drives it to zero and deletes the file out from under a live holder. Replace all four with atomic DB ops: incr_ref_count (`ref_count = ref_count + 1`) and decr_ref_count (`greatest(ref_count - 1, 0)` RETURNING the new value, so the release decides whether to free the asset from the post-decrement count while holding the row's write lock — Postgres serialises concurrent releases on that lock, closing the double-free). Pure DB-atomicity change, no semantic shift. Green under the E3 S2 ref_count net (5) + full lane (356 pass), ruff clean.
This commit is contained in:
@@ -112,6 +112,34 @@ def next_queue_pos(db: Session) -> int:
|
||||
return (db.execute(select(func.coalesce(func.max(DownloadJob.queue_pos), 0))).scalar() or 0) + 1
|
||||
|
||||
|
||||
def incr_ref_count(db: Session, asset_id: int) -> None:
|
||||
"""Atomically +1 an asset's hold count AT THE DB. A Python read-modify-write
|
||||
(`asset.ref_count = asset.ref_count + 1`) loses an update when two enqueues of the same
|
||||
(source,format) run in parallel — both read the same value and write the same +1 — leaving the
|
||||
count one short, so the first release can free a file another job still holds (B-sub-5)."""
|
||||
db.execute(
|
||||
update(MediaAsset)
|
||||
.where(MediaAsset.id == asset_id)
|
||||
.values(ref_count=MediaAsset.ref_count + 1)
|
||||
.execution_options(synchronize_session=False)
|
||||
)
|
||||
|
||||
|
||||
def decr_ref_count(db: Session, asset_id: int) -> int:
|
||||
"""Atomically -1 an asset's hold count (clamped at 0) and RETURN the new value. The caller decides
|
||||
whether to free the asset from the returned count while still holding the row's write lock, so two
|
||||
concurrent releases can't both observe the same pre-decrement value and double-free — Postgres
|
||||
serialises the two UPDATEs on the row lock (B-sub-5). synchronize_session=False: the ORM copy is
|
||||
intentionally left stale; callers read the returned value, not `asset.ref_count`."""
|
||||
return db.execute(
|
||||
update(MediaAsset)
|
||||
.where(MediaAsset.id == asset_id)
|
||||
.values(ref_count=func.greatest(MediaAsset.ref_count - 1, 0))
|
||||
.returning(MediaAsset.ref_count)
|
||||
.execution_options(synchronize_session=False)
|
||||
).scalar_one()
|
||||
|
||||
|
||||
def enqueue(
|
||||
db: Session,
|
||||
user_id: int,
|
||||
@@ -143,7 +171,7 @@ def enqueue(
|
||||
queue_pos=next_queue_pos(db),
|
||||
)
|
||||
db.add(job)
|
||||
asset.ref_count = (asset.ref_count or 0) + 1
|
||||
incr_ref_count(db, asset.id)
|
||||
|
||||
# Cache hit: the file already exists — no worker round-trip needed.
|
||||
if asset.status == "ready":
|
||||
@@ -212,7 +240,7 @@ def enqueue_edit(
|
||||
queue_pos=next_queue_pos(db),
|
||||
)
|
||||
db.add(job)
|
||||
asset.ref_count = (asset.ref_count or 0) + 1
|
||||
incr_ref_count(db, asset.id)
|
||||
|
||||
if asset.status == "ready": # identical clip already produced for this user → instant
|
||||
job.status = "done"
|
||||
@@ -273,8 +301,8 @@ def release_asset(db: Session, job: DownloadJob) -> None:
|
||||
asset = db.get(MediaAsset, job.asset_id)
|
||||
if asset is None:
|
||||
return
|
||||
asset.ref_count = max(0, (asset.ref_count or 0) - 1)
|
||||
if asset.ref_count == 0 and asset.status == "ready":
|
||||
new_count = decr_ref_count(db, asset.id)
|
||||
if new_count == 0 and asset.status == "ready":
|
||||
if asset.rel_path:
|
||||
storage.delete_asset_files(settings.download_root, asset.rel_path)
|
||||
db.delete(asset)
|
||||
|
||||
@@ -523,7 +523,7 @@ def redownload_download(
|
||||
# last holder (a co-held asset is left intact for its other holders).
|
||||
service.release_asset(db, job)
|
||||
|
||||
asset.ref_count = (asset.ref_count or 0) + 1
|
||||
service.incr_ref_count(db, asset.id)
|
||||
job.asset_id = asset.id
|
||||
job.status = "queued"
|
||||
job.progress = 0
|
||||
|
||||
Reference in New Issue
Block a user