peter
bf99444ded
The Files library starts here. A user's every download, clip and upload now lands in a source folder made per origin (YouTube, Plex clips, uploads, one per other site), decided once in downloads/folders.py at every place a job is made. How long a file stays moves from the shared asset to the user's job: stamped from its folder when it is done, never for a file the user keeps. The GC warns each user once, then takes the expired file out of their library; the cache cap never evicts a kept file, and an evicted one leaves every holder's library. The media mount is /files now (FILES_ROOT, FILES_HOST_PATH, volume siftlode_files); docs/self-hosting.md says how to carry an old volume over. Migration 0102 places every existing job and carries each done file's expiry over from its asset. safe_abs_path takes a missing path and answers None, which drops one baselined type error.
298 lines
12 KiB
Python
298 lines
12 KiB
Python
"""Download retention garbage collector (a scheduler job).
|
|
|
|
Five passes each run:
|
|
1. Pre-expiry warning — a done file whose `expires_at` (its folder's retention, `folders`) falls
|
|
within the grace window, and that its user did not keep, warns its user once (`expiry_warned`).
|
|
2. Expiry — such a file past its `expires_at` leaves the library: the job is deleted and its hold
|
|
released, so the shared file goes with its last holder.
|
|
3. LRU eviction — if a total cache cap is set and exceeded, delete least-recently-accessed ready
|
|
assets (never an upload, never one a user kept) until back under it; each holder's file leaves
|
|
their library, with a notice.
|
|
4. Orphan reap — delete errored, fileless, unreferenced asset rows (they carry no TTL, so
|
|
passes 1-3 skip them; benign row-count hygiene).
|
|
5. Orphan FILE sweep — delete on-disk files/dirs no surviving asset owns (cruft left when a
|
|
re-download renames an asset, e.g. old accented channel dirs, or an upload's stranded file).
|
|
It skips the dot-prefixed system trees (all but `.uploads`) and anything newer than an hour,
|
|
so an in-flight download is never touched.
|
|
|
|
Runs in the API process (which mounts FILES_ROOT). Pure disk/DB work — no YouTube quota.
|
|
"""
|
|
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import logs, sysconfig
|
|
from app.config import settings
|
|
from app.downloads import quota, service, storage
|
|
from app.downloads.status import JobStatus
|
|
from app.models import DownloadJob, MediaAsset
|
|
from app.notifications import create_notification
|
|
|
|
log = logs.get_logger(__name__)
|
|
# The one dot-prefixed top-level tree the orphan sweep enters. The others are not part of the asset
|
|
# layout (user clips, Plex HLS/img caches, in-progress staging, editor storyboards) and are skipped.
|
|
_UPLOAD_TREE = ".uploads"
|
|
_ORPHAN_MIN_AGE_S = 3600 # only sweep files older than this, so an in-flight download is never hit
|
|
|
|
|
|
def _notify(db: Session, job: DownloadJob, asset: MediaAsset | None, ntype: str) -> None:
|
|
title = job.display_name or (asset.title if asset else None) or "download"
|
|
create_notification(
|
|
db,
|
|
user_id=job.user_id,
|
|
type=ntype,
|
|
title=title, # English fallback; the client renders a translated message from data
|
|
data={
|
|
"kind": ntype,
|
|
"title": title,
|
|
"video_id": asset.source_ref if asset else None,
|
|
"job_id": job.id,
|
|
},
|
|
commit=False,
|
|
)
|
|
|
|
|
|
def _drop_job(db: Session, job: DownloadJob, ntype: str) -> None:
|
|
"""Take a file out of its user's library: tell them, release the job's hold on the asset (the
|
|
file goes with the last holder), delete the row (its shares and links cascade)."""
|
|
_notify(db, job, db.get(MediaAsset, job.asset_id) if job.asset_id else None, ntype)
|
|
service.release_asset(db, job)
|
|
db.delete(job)
|
|
|
|
|
|
def _sweep_orphan_files(db: Session, files_root: str) -> int:
|
|
"""Delete media/sidecar files (and now-empty dirs) under FILES_ROOT that no `ready` asset
|
|
owns — cruft left when a re-download changes an asset's on-disk name (e.g. an old accented
|
|
channel dir, or a poster from a superseded naming scheme) — or, under `.uploads`, that no upload
|
|
row names at all. Skips the other dot-prefixed system trees and only removes files older than
|
|
`_ORPHAN_MIN_AGE_S`, so a just-completed or in-flight download is never deleted, and only files
|
|
NOT owned by a surviving asset."""
|
|
root = Path(files_root)
|
|
if not root.is_dir():
|
|
return 0
|
|
keep: set[Path] = set()
|
|
# An upload's file is on disk from ingest on, whatever its state (a queued conversion's original,
|
|
# a failed one kept for Retry), so every upload row keeps its file — not only a ready one.
|
|
for rel in db.execute(
|
|
select(MediaAsset.rel_path).where(
|
|
or_(MediaAsset.status == "ready", MediaAsset.source_kind == "upload"),
|
|
MediaAsset.rel_path.is_not(None),
|
|
)
|
|
).scalars():
|
|
media = (root / rel).resolve()
|
|
keep |= {media, media.with_suffix(".nfo"), media.with_suffix(".jpg")}
|
|
|
|
cutoff = time.time() - _ORPHAN_MIN_AGE_S
|
|
removed = 0
|
|
# Throttle the whole-tree walk: this runs in the API process, and an unpaced rglob+stat over a
|
|
# large download root monopolizes disk IO long enough to make web requests + playback stutter.
|
|
# Yield the IO channel after every `batch` files touched so foreground work interleaves.
|
|
pause_s = max(0, settings.download_gc_sweep_pause_ms) / 1000.0
|
|
batch = max(1, settings.download_gc_sweep_batch)
|
|
touched = 0
|
|
|
|
def _throttle() -> None:
|
|
nonlocal touched
|
|
touched += 1
|
|
if pause_s and touched % batch == 0:
|
|
time.sleep(pause_s)
|
|
|
|
unreclaimed = 0
|
|
for entry in root.iterdir():
|
|
# .edits / .plex-* / .staging / .storyboards + any dotfile — but `.uploads` is swept: a
|
|
# conversion stopped between placing the new file and the commit that points at it (or the
|
|
# unlink of the original after it) leaves a file no row names, and nothing else finds it.
|
|
if entry.name.startswith(".") and entry.name != _UPLOAD_TREE:
|
|
continue
|
|
files = [entry] if entry.is_file() else [p for p in entry.rglob("*") if p.is_file()]
|
|
for f in files:
|
|
_throttle()
|
|
try:
|
|
if f.resolve() in keep or f.stat().st_mtime > cutoff:
|
|
continue
|
|
f.unlink()
|
|
removed += 1
|
|
except OSError:
|
|
# silent-ok: counted, not logged per file — a broken mount would put one line
|
|
# under every orphan. The total is reported once at the end of the sweep.
|
|
unreclaimed += 1
|
|
if entry.is_dir(): # prune emptied dirs (deepest first), then the channel dir itself
|
|
for d in sorted(
|
|
(p for p in entry.rglob("*") if p.is_dir()),
|
|
key=lambda p: len(p.parts),
|
|
reverse=True,
|
|
):
|
|
try:
|
|
d.rmdir()
|
|
# silent-ok: `rmdir` refusing means the directory still holds files, which is
|
|
# the ordinary answer here — this prune only takes the ones that emptied.
|
|
except OSError:
|
|
pass
|
|
try:
|
|
entry.rmdir()
|
|
# silent-ok: the same prune one level up, at the channel directory.
|
|
except OSError:
|
|
pass
|
|
if unreclaimed:
|
|
log.warning(
|
|
"some orphaned download files could not be reclaimed",
|
|
extra=logs.event("download.gc.unreclaimed", files=unreclaimed, removed=removed),
|
|
)
|
|
return removed
|
|
|
|
|
|
def run_download_gc(db: Session) -> dict:
|
|
now = datetime.now(timezone.utc)
|
|
grace_days = sysconfig.get_int(db, "download_gc_grace_days")
|
|
warned = expired = evicted = reaped = 0
|
|
|
|
# 1. Pre-expiry warnings — one per file, the first run inside the grace window.
|
|
warn_before = now + timedelta(days=grace_days)
|
|
soon = (
|
|
db.execute(
|
|
select(DownloadJob).where(
|
|
DownloadJob.status == JobStatus.DONE,
|
|
DownloadJob.kept.is_(False),
|
|
DownloadJob.expiry_warned.is_(False),
|
|
DownloadJob.expires_at.is_not(None),
|
|
DownloadJob.expires_at > now,
|
|
DownloadJob.expires_at <= warn_before,
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for job in soon:
|
|
_notify(
|
|
db, job, db.get(MediaAsset, job.asset_id) if job.asset_id else None, "download_expiring"
|
|
)
|
|
job.expiry_warned = True
|
|
warned += 1
|
|
if soon:
|
|
db.commit()
|
|
|
|
# 2. Expiry — a file past its folder's retention leaves the library (warned in pass 1).
|
|
dead = (
|
|
db.execute(
|
|
select(DownloadJob).where(
|
|
DownloadJob.status == JobStatus.DONE,
|
|
DownloadJob.kept.is_(False),
|
|
DownloadJob.expires_at.is_not(None),
|
|
DownloadJob.expires_at <= now,
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for job in dead:
|
|
_drop_job(db, job, "download_expired")
|
|
expired += 1
|
|
if dead:
|
|
db.commit()
|
|
|
|
# 3. LRU eviction over the total cache cap: the disk's safety net. Never an upload (no source to
|
|
# fetch again) and never a file anyone kept; an evicted file leaves every holder's library.
|
|
cap = sysconfig.get_int(db, "download_total_max_bytes")
|
|
if cap > 0:
|
|
total = (
|
|
db.execute(
|
|
select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(quota.ON_DISK)
|
|
).scalar()
|
|
or 0
|
|
)
|
|
if total > cap:
|
|
kept_assets = select(DownloadJob.asset_id).where(
|
|
DownloadJob.kept.is_(True), DownloadJob.asset_id.is_not(None)
|
|
)
|
|
candidates = (
|
|
db.execute(
|
|
select(MediaAsset)
|
|
.where(
|
|
MediaAsset.status == "ready",
|
|
MediaAsset.source_kind != "upload",
|
|
MediaAsset.id.not_in(kept_assets),
|
|
)
|
|
.order_by(MediaAsset.last_access_at.asc().nulls_first())
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for asset in candidates:
|
|
if total <= cap:
|
|
break
|
|
total -= asset.size_bytes or 0
|
|
holders = (
|
|
db.execute(select(DownloadJob).where(DownloadJob.asset_id == asset.id))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for job in holders:
|
|
_notify(db, job, asset, "download_evicted")
|
|
db.delete(job)
|
|
if asset.rel_path:
|
|
storage.delete_asset_files(settings.files_root, asset.rel_path)
|
|
db.delete(asset)
|
|
evicted += 1
|
|
db.commit()
|
|
|
|
# 4. Reap orphaned errored rows (no file, no holder). get_or_create_asset resets a still-
|
|
# referenced errored row to pending on re-enqueue, so these only accumulate once every
|
|
# holder is gone; they carry no expires_at, so passes 1-3 never touch them.
|
|
orphans = (
|
|
db.execute(
|
|
select(MediaAsset).where(
|
|
MediaAsset.status == "error",
|
|
MediaAsset.ref_count == 0,
|
|
MediaAsset.rel_path.is_(None),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for asset in orphans:
|
|
db.delete(asset) # no file to unlink; FK SET NULL clears any stray job.asset_id
|
|
reaped += 1
|
|
if orphans:
|
|
db.commit()
|
|
|
|
# 4b. Reap unheld uploads. An upload carries no expires_at, is never evicted and never re-enqueued,
|
|
# so the passes above never free one that lost its last holder outside `release_asset` — a
|
|
# conversion that finished after its job was cancelled, or one cancelled mid-run (the asset
|
|
# was `downloading` then). Its file, ready or still the unconverted original, is at `rel_path`.
|
|
unheld = (
|
|
db.execute(
|
|
select(MediaAsset).where(
|
|
MediaAsset.source_kind == "upload",
|
|
MediaAsset.ref_count == 0,
|
|
MediaAsset.status != "downloading",
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for asset in unheld:
|
|
if asset.rel_path: # no holder is left to tell, so no notice — just the file and the row
|
|
storage.delete_asset_files(settings.files_root, asset.rel_path)
|
|
db.delete(asset)
|
|
reaped += 1
|
|
if unheld:
|
|
db.commit()
|
|
|
|
# 5. Orphan FILE sweep — remove on-disk cruft no surviving asset owns (old accented dirs, stale
|
|
# posters). Runs last so the keep-set reflects the assets that survived passes 2-4.
|
|
swept = _sweep_orphan_files(db, settings.files_root)
|
|
|
|
result = {
|
|
"warned": warned,
|
|
"expired": expired,
|
|
"evicted": evicted,
|
|
"reaped": reaped,
|
|
"swept": swept,
|
|
}
|
|
log.info("download_gc %s", result)
|
|
return result
|