siftlode/backend/tests/test_download_ref_count.py

"""Characterization net — MediaAsset.ref_count enqueue/release accounting.

`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