Files
siftlode/backend/app/downloads/quota.py
T
peter 358d722ce9 fix(e12): close what round one broke, and revert two fixes that overreached
Round two of the review loop, on round one's own diff. Two of its findings were
regressions the fixes themselves introduced, which is exactly what the loop is for:

**A per-user refusal was writing onto a shared row.** MediaAssets are deduped across
users, and the generic failure handler puts the message on the asset — right for an
upstream failure, wrong for "your storage is full". Two users queuing the same video share
one asset, so the first user hitting their cap failed the second user's job with the first
one's private reason, telling someone with an empty library to free up space. The refusal
now raises `_OwnerRefused`, caught separately: the job fails, the asset goes back to
`pending` for whoever else holds it. Pinned by a test that drives `_download` directly —
the refusal happens before extraction, so nothing needs stubbing.

**The restore scripts restarted the app, undoing the restore.** `docker compose start`
re-runs the container command, and that runs `alembic upgrade head` — so on the rollback
path this documents as recommended (restore the dump, THEN pin the older tag) the newer
image immediately re-applied the migrations the restore had just undone, and on a FAILED
restore it put the app back on a half-restored schema. Both are the opposite of why the
stop was added. Nothing is restarted automatically now; the scripts say what to run and
when, and the docs' snippets say the same.

**Reverted rather than patched:** the OnboardingWizard toast (the wizard already renders
that failure inline, next to a Retry button — three recoveries for one failure), and
`useVerifyEntry`'s "each new link clears the other". The latter blanked the signed-out
card's only banner, where nothing renders the reset flag to replace it. The two states are
independent again; what changes is that `dismiss` peels ONE layer, which is what the
original finding was actually about.

Also: `Stats` uses the YouTube helper rather than the bare one, so the route's English 403
detail no longer beats the translated fallback under a Hungarian UI; `urlparse` guarded
alongside `.port` (an unclosed bracket raised just as late), the reason renamed `malformed`
since "bad port" misdescribes it; three docstrings narrowed to what the code actually does
(the worker's byte check is per-job, not an aggregate ledger, and not on the edit path;
the redirect residual leaks a reachability oracle, which the note had denied; the
invalidation helper does not cover suspend); the pasted Windows recipe wrapped so `throw`
can stop it; the bash one-liner's failure branch no longer exits 0; and the tests that were
missing for the round-one fixes — the account panel's error paths, the invalidation set,
and the shared-asset case above.

Gate green: vitest 685, pytest 950.
2026-09-05 04:02:57 +02:00

175 lines
7.0 KiB
Python

"""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).
DOWNLOAD jobs additionally consult `remaining_bytes` in the worker, because enqueue can only compare
the footprint as it stood when the job was created and earlier jobs may have filled the cap since.
That is a per-job ceiling, not an aggregate ledger: concurrent jobs of one user each read the same
remainder (in-flight bytes are not `ready`, so `footprint` cannot see them), so a wave of
max_concurrent downloads can still land over the cap — the deliberate trade, since a download's size
is unknown until it finishes. Edit jobs do not consult it at all (tracked as SEC-19).
"""
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:
# Floor max_concurrent at 1 (admin_set_quota already does on write; the sysconfig default's
# min is 1). A stored 0 is nonsensical for a concurrency limit AND would diverge the two
# "at limit" checks — the worker's claim-window exclusion vs at_concurrency_limit — since a
# 0-limit user is "capped" with zero running jobs, which the SQL exclusion can't see. Keep
# them consistent by never letting the effective limit drop below 1. (To BLOCK a user's
# downloads, suspend them or set max_jobs=0, not max_concurrent=0.)
return QuotaLimits(row.max_bytes, max(1, row.max_concurrent), row.max_jobs, row.unlimited)
return QuotaLimits(
max_bytes=sysconfig.get_int(db, "download_default_max_bytes"),
max_concurrent=max(1, 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 int(
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 remaining_bytes(db: Session, user_id: int) -> int | None:
"""Storage this user may still spend on ONE download: ``None`` for an unlimited user, else the
cap minus the current footprint, floored at 0.
Zero is a real answer, not "no limit" — it means the cap is already met and the caller must
REFUSE rather than proceed unbounded. `check_enqueue` cannot make that refusal on the caller's
behalf: it ran when the job was queued, while the user was still under the cap, and the jobs
that filled it have landed since. Keeping the arithmetic here (rather than inline in the worker)
is what keeps every max_bytes decision in this module."""
lim = resolve(db, user_id)
if lim.unlimited:
return None
return max(0, lim.max_bytes - footprint(db, user_id))
def check_upload(db: Session, user_id: int, add_bytes: int) -> None:
"""Like check_enqueue, but for an upload whose size is KNOWN at ingest: also reject when adding
this file would push the footprint over max_bytes. A download can't pre-check its size (unknown
until the worker finishes) so check_enqueue only compares the current footprint; an upload can,
so it's charged against the cap before it lands. Raises QuotaExceeded(max_jobs|max_bytes)."""
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)
projected = footprint(db, user_id) + add_bytes
if projected > lim.max_bytes:
raise QuotaExceeded("max_bytes", lim.max_bytes, projected)
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,
}