diff --git a/backend/alembic/versions/0038_asset_gc_notified.py b/backend/alembic/versions/0038_asset_gc_notified.py new file mode 100644 index 0000000..19b317f --- /dev/null +++ b/backend/alembic/versions/0038_asset_gc_notified.py @@ -0,0 +1,29 @@ +"""media_assets.gc_notified flag for one-shot pre-expiry warnings + +Revision ID: 0038_asset_gc_notified +Revises: 0037_dl_builtin_profiles +Create Date: 2026-07-03 + +The download GC warns each holder once, a grace window before an asset's TTL expiry; this flag +records that so subsequent GC runs don't re-notify until the asset is deleted. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0038_asset_gc_notified" +down_revision: Union[str, None] = "0037_dl_builtin_profiles" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "media_assets", + sa.Column("gc_notified", sa.Boolean(), server_default="false", nullable=False), + ) + + +def downgrade() -> None: + op.drop_column("media_assets", "gc_notified") diff --git a/backend/app/config.py b/backend/app/config.py index cb44539..e521bcd 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -142,6 +142,9 @@ class Settings(BaseSettings): download_worker_concurrency: int = 2 # How often (empty pending queue) the worker polls for a new job, in seconds. download_poll_seconds: int = 5 + # Total cache size cap across ALL users (bytes). When >0 and exceeded, the GC evicts the + # least-recently-accessed ready assets until under it. 0 = unlimited (disk is the limit). + download_total_max_bytes: int = 0 # Default per-user quota (admin-tunable; a per-user download_quotas row overrides). download_default_max_bytes: int = 5_368_709_120 # 5 GiB download_default_max_concurrent: int = 2 diff --git a/backend/app/downloads/gc.py b/backend/app/downloads/gc.py new file mode 100644 index 0000000..570d130 --- /dev/null +++ b/backend/app/downloads/gc.py @@ -0,0 +1,119 @@ +"""Download retention garbage collector (a scheduler job). + +Three passes each run: + 1. Pre-expiry warning — for ready assets expiring within the grace window, notify each holder + once (gc_notified flag) so a download isn't silently deleted. + 2. Expiry deletion — delete assets past their TTL (files + row). The FK SET NULL nulls the + holders' jobs' asset_id, so those jobs surface as "expired" (re-downloadable) in the UI. + 3. LRU eviction — if a total cache cap is set and exceeded, delete least-recently-accessed + ready assets until back under it. + +Runs in the API process (which mounts DOWNLOAD_ROOT). Pure disk/DB work — no YouTube quota. +""" +import logging +from datetime import datetime, timedelta, timezone + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app import sysconfig +from app.config import settings +from app.downloads import storage +from app.models import DownloadJob, MediaAsset +from app.notifications import create_notification + +log = logging.getLogger("siftlode.downloads.gc") + + +def _holders(db: Session, asset_id: int) -> list[int]: + return list( + db.execute( + select(DownloadJob.user_id) + .where(DownloadJob.asset_id == asset_id) + .distinct() + ).scalars() + ) + + +def _notify_holders(db: Session, asset: MediaAsset, ntype: str) -> None: + title = asset.title or "download" + for uid in _holders(db, asset.id): + create_notification( + db, + user_id=uid, + type=ntype, + title=title, # English fallback; the client renders a translated message from data + data={"kind": ntype, "title": asset.title, "video_id": asset.source_ref}, + commit=False, + ) + + +def _delete_asset(db: Session, asset: MediaAsset, ntype: str) -> None: + if asset.rel_path: + storage.delete_asset_files(settings.download_root, asset.rel_path) + _notify_holders(db, asset, ntype) + db.delete(asset) # jobs.asset_id -> NULL via FK ON DELETE SET NULL + + +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 = 0 + + # 1. Pre-expiry warnings. + warn_before = now + timedelta(days=grace_days) + soon = db.execute( + select(MediaAsset).where( + MediaAsset.status == "ready", + MediaAsset.gc_notified.is_(False), + MediaAsset.expires_at.is_not(None), + MediaAsset.expires_at > now, + MediaAsset.expires_at <= warn_before, + ) + ).scalars().all() + for asset in soon: + _notify_holders(db, asset, "download_expiring") + asset.gc_notified = True + warned += 1 + if soon: + db.commit() + + # 2. Expiry deletion (ready assets past TTL, and any errored leftovers with a TTL). + dead = db.execute( + select(MediaAsset).where( + MediaAsset.status.in_(("ready", "error")), + MediaAsset.expires_at.is_not(None), + MediaAsset.expires_at <= now, + ) + ).scalars().all() + for asset in dead: + _delete_asset(db, asset, "download_expired") + expired += 1 + if dead: + db.commit() + + # 3. LRU eviction over the total cache cap. + 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( + MediaAsset.status == "ready" + ) + ).scalar() or 0 + if total > cap: + candidates = db.execute( + select(MediaAsset) + .where(MediaAsset.status == "ready") + .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 + _delete_asset(db, asset, "download_evicted") + evicted += 1 + db.commit() + + result = {"warned": warned, "expired": expired, "evicted": evicted} + log.info("download_gc %s", result) + return result diff --git a/backend/app/downloads/quota.py b/backend/app/downloads/quota.py new file mode 100644 index 0000000..fb224cf --- /dev/null +++ b/backend/app/downloads/quota.py @@ -0,0 +1,129 @@ +"""Per-user download limits + footprint accounting. + +A user's *footprint* is the total size of the ready assets their library references (cache hits +count — it's "how much disk you're responsible for", not "how much you personally downloaded"). +Limits come from a per-user DownloadQuota row if the admin set one, else the sysconfig defaults. +`unlimited` (e.g. the admin's own account) bypasses the byte cap. + +Enforcement split: max_jobs + max_bytes are checked at enqueue (they bound holdings); per-user +max_concurrent is enforced by the worker at claim time (it bounds simultaneous downloads). +""" +from dataclasses import dataclass + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app import sysconfig +from app.models import DownloadJob, DownloadQuota, MediaAsset + +# Jobs a user is considered to be "holding" (occupy a job slot / footprint). Terminal failures +# (error, canceled) don't count. +_HOLDING = ("queued", "running", "paused", "done") +_RUNNING = ("queued", "running") + + +@dataclass +class QuotaLimits: + max_bytes: int + max_concurrent: int + max_jobs: int + unlimited: bool + + +class QuotaExceeded(Exception): + """Raised by check_enqueue; `reason` is one of max_jobs|max_bytes (i18n key on the client).""" + + def __init__(self, reason: str, limit: int, current: int): + self.reason = reason + self.limit = limit + self.current = current + super().__init__(f"download quota exceeded: {reason} ({current}/{limit})") + + +def resolve(db: Session, user_id: int) -> QuotaLimits: + row = db.get(DownloadQuota, user_id) + if row is not None: + return QuotaLimits(row.max_bytes, row.max_concurrent, row.max_jobs, row.unlimited) + return QuotaLimits( + max_bytes=sysconfig.get_int(db, "download_default_max_bytes"), + max_concurrent=sysconfig.get_int(db, "download_default_max_concurrent"), + max_jobs=sysconfig.get_int(db, "download_default_max_jobs"), + unlimited=False, + ) + + +def footprint(db: Session, user_id: int) -> int: + """Total bytes of the distinct ready assets this user's active jobs reference.""" + held_assets = ( + select(DownloadJob.asset_id) + .where( + DownloadJob.user_id == user_id, + DownloadJob.status.in_(_HOLDING), + DownloadJob.asset_id.is_not(None), + ) + .distinct() + ) + return ( + db.execute( + select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where( + MediaAsset.id.in_(held_assets), MediaAsset.status == "ready" + ) + ).scalar() + or 0 + ) + + +def _count(db: Session, user_id: int, states) -> int: + return ( + db.execute( + select(func.count()) + .select_from(DownloadJob) + .where(DownloadJob.user_id == user_id, DownloadJob.status.in_(states)) + ).scalar() + or 0 + ) + + +def active_jobs(db: Session, user_id: int) -> int: + return _count(db, user_id, _HOLDING) + + +def running_jobs(db: Session, user_id: int) -> int: + return _count(db, user_id, _RUNNING) + + +def at_concurrency_limit(db: Session, user_id: int) -> bool: + """Worker-side gate: is this user already downloading their max? Unlimited users never are.""" + lim = resolve(db, user_id) + if lim.unlimited: + return False + running = _count(db, user_id, ("running",)) + return running >= lim.max_concurrent + + +def check_enqueue(db: Session, user_id: int) -> None: + """Raise QuotaExceeded if the user can't take another job (holdings caps). Concurrency is + enforced later by the worker, so a big queue is allowed; it just drains max_concurrent-wide.""" + lim = resolve(db, user_id) + if lim.unlimited: + return + n = active_jobs(db, user_id) + if n >= lim.max_jobs: + raise QuotaExceeded("max_jobs", lim.max_jobs, n) + used = footprint(db, user_id) + if used >= lim.max_bytes: + raise QuotaExceeded("max_bytes", lim.max_bytes, used) + + +def usage(db: Session, user_id: int) -> dict: + """Compact usage snapshot for the UI (footprint + counts vs limits).""" + lim = resolve(db, user_id) + return { + "footprint_bytes": footprint(db, user_id), + "active_jobs": active_jobs(db, user_id), + "running_jobs": running_jobs(db, user_id), + "max_bytes": lim.max_bytes, + "max_concurrent": lim.max_concurrent, + "max_jobs": lim.max_jobs, + "unlimited": lim.unlimited, + } diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index e832d6b..a446cc2 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -12,7 +12,7 @@ from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session -from app.downloads import formats +from app.downloads import formats, quota from app.models import DownloadJob, MediaAsset @@ -67,7 +67,10 @@ def enqueue( profile_id: int | None = None, display_name: str | None = None, ) -> DownloadJob: - """Queue a download for `user_id`. Returns the DownloadJob (already `done` on a cache hit).""" + """Queue a download for `user_id`. Returns the DownloadJob (already `done` on a cache hit). + + Raises quota.QuotaExceeded if the user is at their job-count or footprint cap.""" + quota.check_enqueue(db, user_id) spec = formats.normalize(spec) sig = formats.format_sig(spec) asset = get_or_create_asset(db, source_kind, source_ref, sig) diff --git a/backend/app/models.py b/backend/app/models.py index be145e0..e91d0d8 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -731,6 +731,10 @@ class MediaAsset(Base, TimestampMixin): nfo_written: Mapped[bool] = mapped_column( Boolean, default=False, server_default="false" ) + # Set once the GC has warned holders of the upcoming expiry, so it doesn't warn every run. + gc_notified: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) # Denormalized naming/metadata for the Plex tree + ad-hoc display. title: Mapped[str | None] = mapped_column(Text) uploader: Mapped[str | None] = mapped_column(String(255)) diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index cdec9f3..8c0e506 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -12,6 +12,7 @@ from sqlalchemy import select from app import progress, quota from app.config import settings from app.db import SessionLocal +from app.downloads.gc import run_download_gc from app.models import SchedulerSetting from app.notifications import create_notification from app.state import is_sync_paused @@ -59,6 +60,7 @@ JOB_INTERVALS: dict[str, int] = { "maintenance": settings.maintenance_interval_minutes, "demo_reset": settings.demo_reset_minutes, "explore_cleanup": settings.explore_cleanup_minutes, + "download_gc": settings.download_gc_minutes, } # Sane bounds for an admin-set interval (minutes). @@ -276,6 +278,12 @@ def _explore_cleanup_job() -> None: _job("explore_cleanup", purge_ephemeral) +def _download_gc_job() -> None: + # Retention GC for the download center: pre-expiry warnings, TTL deletion, LRU eviction. + # Pure disk/DB work (no YouTube quota). + _job("download_gc", run_download_gc) + + # job_id -> wrapper. The single source of truth for which jobs exist and how to run one, # shared by start_scheduler (recurring registration) and trigger_job (manual "run now"). JOB_FUNCS: dict[str, Callable[[], None]] = { @@ -289,6 +297,7 @@ JOB_FUNCS: dict[str, Callable[[], None]] = { "maintenance": _maintenance_job, "demo_reset": _demo_reset_job, "explore_cleanup": _explore_cleanup_job, + "download_gc": _download_gc_job, } diff --git a/backend/app/sysconfig.py b/backend/app/sysconfig.py index 1febd37..ed6c79b 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -67,6 +67,7 @@ SPECS: tuple[ConfigSpec, ...] = ( # --- Access / registration --- ConfigSpec("allow_registration", "bool", "access", "allow_registration"), # --- Download center (yt-dlp) — per-user defaults + retention/GC + layout --- + ConfigSpec("download_total_max_bytes", "int", "downloads", "download_total_max_bytes", min=0), ConfigSpec("download_default_max_bytes", "int", "downloads", "download_default_max_bytes", min=0), ConfigSpec("download_default_max_concurrent", "int", "downloads", "download_default_max_concurrent", min=1, max=20), ConfigSpec("download_default_max_jobs", "int", "downloads", "download_default_max_jobs", min=1, max=100_000), diff --git a/backend/app/worker.py b/backend/app/worker.py index 5276e0b..b3a81ee 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -28,7 +28,7 @@ from sqlalchemy import text from app.config import settings from app.db import SessionLocal -from app.downloads import formats, service, storage +from app.downloads import formats, quota, service, storage from app.models import DownloadJob, MediaAsset log = logging.getLogger("siftlode.worker") @@ -83,24 +83,37 @@ def _recover_orphans() -> None: def _claim_job() -> int | None: + """Claim the oldest queued job whose user is under their per-user concurrency limit. + + Fetches a window of locked candidates (FOR UPDATE SKIP LOCKED) and picks the first eligible + one, so one user flooding the queue can't starve others beyond their max_concurrent.""" with SessionLocal() as db: - row = db.execute( + rows = db.execute( text( """ - UPDATE download_jobs SET status='running', phase='starting', updated_at=now() - WHERE id = ( - SELECT id FROM download_jobs - WHERE status='queued' - ORDER BY queue_pos, id - FOR UPDATE SKIP LOCKED - LIMIT 1 - ) - RETURNING id + SELECT id, user_id FROM download_jobs + WHERE status='queued' + ORDER BY queue_pos, id + FOR UPDATE SKIP LOCKED + LIMIT 50 """ ) - ).fetchone() + ).fetchall() + chosen = next( + (jid for jid, uid in rows if not quota.at_concurrency_limit(db, uid)), None + ) + if chosen is None: + db.commit() # release the row locks; nothing eligible right now + return None + db.execute( + text( + "UPDATE download_jobs SET status='running', phase='starting', updated_at=now() " + "WHERE id = :id" + ), + {"id": chosen}, + ) db.commit() - return row[0] if row else None + return chosen def _claim_asset(asset_id: int) -> bool: