diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index 39af6a58..e3916408 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -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) diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index 2b1c08c9..cc435535 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -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