peter
815083bcc4
- The queue keeps a done job by when it became done (ready_at), not by its last write: renaming, keeping or moving an old file, or the GC's expiry warning, no longer brings it back as just finished. - GC expiry and eviction re-read each file under the asset→job locks the user routes take, and commit one file at a time: a delete, a re-download or a keep made while the GC runs is honoured, never a second release that frees a file another user still holds. - A file records the origin it was placed under (download_jobs.source_key, migration 0105): an edit whose parent expired still goes back to its folder, not to Uploads. - A folder a file is being put in is held (FOR SHARE) until it lands; a delete of it waits and takes that file home too, instead of failing on the foreign key. - The hash route reads at most two files at once (429 past that) and holds no connection meanwhile. - 0104's backfill dates old files by their request date only: no notice prod wrote names its job. - The old DOWNLOAD_ROOT still finds the files; a files root that is not a mounted volume stops the worker and is logged by the API; the volume-copy recipe looks the project-prefixed names up. - Stale comments about the /downloads mount and the Downloads page.
142 lines
6.1 KiB
Python
142 lines
6.1 KiB
Python
"""Migration 0102's data step, DB-lane: every existing job lands in its source folder, and a done
|
|
file's expiry moves from the shared asset to the user's job.
|
|
|
|
The schema half is checked by test_migrations.py; a row placed in the wrong folder is invisible there.
|
|
"""
|
|
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
|
|
|
|
def test_existing_jobs_are_placed_and_keep_their_expiry(db, make_user):
|
|
alice, bob = make_user().id, make_user().id
|
|
cfg = Config("alembic.ini")
|
|
command.downgrade(cfg, "0101_plex_text_plain")
|
|
try:
|
|
with engine.begin() as conn:
|
|
|
|
def asset(kind, ref, expires=None):
|
|
return conn.execute(
|
|
text(
|
|
"INSERT INTO media_assets (source_kind, source_ref, format_sig, status,"
|
|
" ref_count, expires_at) VALUES (:k, :r, :sig, 'ready', 1, :e)"
|
|
" RETURNING id"
|
|
),
|
|
{"k": kind, "r": ref, "sig": kind + ref, "e": expires},
|
|
).scalar_one()
|
|
|
|
def job(user, kind, ref, asset_id, status="done", spec=None, parent=None):
|
|
return conn.execute(
|
|
text(
|
|
"INSERT INTO download_jobs (user_id, asset_id, source_kind, source_ref,"
|
|
" profile_snapshot, status, edit_spec, source_job_id)"
|
|
" VALUES (:u, :a, :k, :r, '{}', :s, CAST(:spec AS json), :p) RETURNING id"
|
|
),
|
|
{"u": user, "a": asset_id, "k": kind, "r": ref, "s": status,
|
|
"spec": spec, "p": parent},
|
|
).scalar_one() # fmt: skip
|
|
|
|
yt = asset("youtube", "abcdefghijk", "2030-01-02T00:00:00+00:00")
|
|
tt = asset("url", "https://www.tiktok.com/@a/video/1", "2030-01-03T00:00:00+00:00")
|
|
up = asset("upload", "u1")
|
|
cl = asset("clip", "c1", "2030-01-04T00:00:00+00:00")
|
|
j_yt = job(alice, "youtube", "abcdefghijk", yt)
|
|
j_yt_bob = job(bob, "youtube", "abcdefghijk", yt, status="queued")
|
|
j_tt = job(alice, "url", "https://www.tiktok.com/@a/video/1", tt)
|
|
j_up = job(alice, "upload", "u1", up)
|
|
j_cl = job(alice, "clip", "c1", cl, spec='{"source": "plex", "ref": "9"}')
|
|
j_ed = job(alice, "edit", "e1", cl, parent=j_tt)
|
|
finally:
|
|
command.upgrade(cfg, "head")
|
|
|
|
with engine.connect() as conn:
|
|
rows = conn.execute(
|
|
text(
|
|
"SELECT j.id, f.user_id, f.source_key, f.retention_days, j.expires_at::date"
|
|
" FROM download_jobs j JOIN file_folders f ON f.id = j.folder_id"
|
|
)
|
|
).all()
|
|
folders = conn.execute(text("SELECT count(*) FROM file_folders")).scalar()
|
|
got = {r[0]: (r[1], r[2], r[3], str(r[4]) if r[4] else None) for r in rows}
|
|
assert got == {
|
|
j_yt: (alice, "youtube", 30, "2030-01-02"),
|
|
j_yt_bob: (bob, "youtube", 30, None), # not done yet: no expiry until it is
|
|
j_tt: (alice, "site:tiktok.com", 30, "2030-01-03"),
|
|
j_up: (alice, "upload", None, None), # an upload never expires
|
|
j_cl: (alice, "plex", 30, "2030-01-04"),
|
|
j_ed: (alice, "site:tiktok.com", 30, "2030-01-04"),
|
|
}
|
|
# alice: youtube, tiktok, upload, plex; bob: youtube.
|
|
assert folders == 5
|
|
|
|
|
|
def test_the_admins_retention_is_kept_and_the_downgrade_gives_the_asset_back_its_expiry(
|
|
db, make_user
|
|
):
|
|
alice, bob = make_user().id, make_user().id
|
|
cfg = Config("alembic.ini")
|
|
command.downgrade(cfg, "0101_plex_text_plain")
|
|
try:
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO system_config (key, value) VALUES ('download_retention_days', '7')"
|
|
" ON CONFLICT (key) DO UPDATE SET value = '7'"
|
|
)
|
|
)
|
|
yt = conn.execute(
|
|
text(
|
|
"INSERT INTO media_assets (source_kind, source_ref, format_sig, status,"
|
|
" ref_count, expires_at) VALUES ('youtube', 'abcdefghijk', 'sig', 'ready', 2,"
|
|
" '2030-01-02T00:00:00+00:00') RETURNING id"
|
|
)
|
|
).scalar_one()
|
|
for user in (alice, bob):
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO download_jobs (user_id, asset_id, source_kind, source_ref,"
|
|
" profile_snapshot, status) VALUES (:u, :a, 'youtube', 'abcdefghijk',"
|
|
" '{}', 'done')"
|
|
),
|
|
{"u": user, "a": yt},
|
|
)
|
|
command.upgrade(cfg, "head")
|
|
with engine.connect() as conn:
|
|
retention = conn.execute(
|
|
text(
|
|
"SELECT DISTINCT retention_days FROM file_folders WHERE source_key = 'youtube'"
|
|
)
|
|
).all()
|
|
|
|
def asset_expiry_after_downgrade(alice_at, bob_at):
|
|
with engine.begin() as conn:
|
|
for user, at in ((alice, alice_at), (bob, bob_at)):
|
|
conn.execute(
|
|
text(
|
|
"UPDATE download_jobs SET expires_at = CAST(:at AS timestamptz)"
|
|
" WHERE asset_id = :a AND user_id = :u"
|
|
),
|
|
{"at": at, "a": yt, "u": user},
|
|
)
|
|
command.downgrade(cfg, "0101_plex_text_plain")
|
|
with engine.connect() as conn:
|
|
got = conn.execute(
|
|
text("SELECT expires_at::date FROM media_assets WHERE id = :a"), {"a": yt}
|
|
).scalar()
|
|
command.upgrade(cfg, "head")
|
|
return str(got) if got else None
|
|
|
|
# The shared file goes when its last holder's copy would; one holder keeping it keeps it.
|
|
latest = asset_expiry_after_downgrade(
|
|
"2030-02-01T00:00:00+00:00", "2030-03-01T00:00:00+00:00"
|
|
)
|
|
kept_by_one = asset_expiry_after_downgrade("2030-02-01T00:00:00+00:00", None)
|
|
finally:
|
|
command.upgrade(cfg, "head")
|
|
assert retention == [(7,)]
|
|
assert latest == "2030-03-01"
|
|
assert kept_by_one is None
|