fix(files): review round 2, the server — locks the GC and folders read under, the mount check, the old mount name

- GC: the expiry warning marks each file conditionally on the expiry it saw, one commit each (a file
  deleted meanwhile no longer fails the run; one re-timed keeps its fresh expiry unwarned); eviction
  counts the disk again when a candidate was freed meanwhile, so it never evicts for space already free.
- Folders: a retention change locks the folder and its files, and a finishing file reads the retention
  under a share lock, so neither misses the other; a folder delete locks the files it takes home (a
  file moved out meanwhile stays put, one deleted meanwhile no longer fails the delete).
- Upload: the folder is checked before the probe and held only once the job names it.
- Dedup enqueue locks the cached asset, so a release meanwhile yields a fresh one instead of a 500.
- Mount check reads the kernel's mount table instead of comparing devices (the vfs storage driver gave
  a mounted volume the device of /), and only inside a container; the API also logs a fresh, empty
  volume in the library's place.
- Compose files fall back to DOWNLOAD_HOST_PATH; the upgrade notes cover the from-source pull.
- Folder routes call owned_or_404 directly.
This commit is contained in:
2026-10-13 02:32:49 +02:00
parent 8b2260c9a8
commit 54a6bbf07f
15 changed files with 543 additions and 93 deletions
+33 -3
View File
@@ -136,7 +136,14 @@ def done_expiry(db: Session, folder_id: int | None, kept: bool) -> datetime | No
folder that keeps files)."""
if kept or folder_id is None:
return None
days = db.execute(select(FileFolder.retention_days).where(FileFolder.id == folder_id)).scalar()
# Read under a share lock, held to the caller's commit: a retention change (`set_retention` locks
# the folder) waits for this file and restamps it too, or this read waits for the change and
# takes the new value — never the old value stamped on a file the change did not see.
days = db.execute(
select(FileFolder.retention_days)
.where(FileFolder.id == folder_id)
.with_for_update(read=True)
).scalar()
return None if days is None else datetime.now(timezone.utc) + timedelta(days=days)
@@ -208,18 +215,30 @@ def set_retention(db: Session, folder: FileFolder, days: int | None) -> None:
"""Change how long `folder` keeps a file (None = until deleted). The done files already in it
move by the difference — each keeps the age it has — but none ends up closer than the GC's
warning window, so a shortened retention still warns before it takes anything."""
# Wait out a file being put in it or stamped from it right now (`own_folder`, `done_expiry`), so
# that file is among the ones restamped below. Re-read: another change may have landed first.
db.execute(
select(FileFolder)
.where(FileFolder.id == folder.id)
.with_for_update()
.execution_options(populate_existing=True)
)
old = folder.retention_days
if old == days:
return
folder.retention_days = days
now = datetime.now(timezone.utc)
floor = now + timedelta(days=sysconfig.get_int(db, "download_gc_grace_days"))
# Locked as they are read: one moved out or deleted meanwhile is no longer this folder's to time.
jobs = db.execute(
select(DownloadJob).where(
select(DownloadJob)
.where(
DownloadJob.folder_id == folder.id,
DownloadJob.status == JobStatus.DONE,
DownloadJob.kept.is_(False),
)
.with_for_update()
.execution_options(populate_existing=True)
).scalars()
for job in jobs:
if days is None:
@@ -238,7 +257,18 @@ def delete_own(db: Session, folder: FileFolder) -> int:
raise FolderError("file_folder_not_own")
# Wait out a file being put in it right now (`own_folder`), then take that one home too.
db.execute(select(FileFolder.id).where(FileFolder.id == folder.id).with_for_update())
jobs = db.execute(select(DownloadJob).where(DownloadJob.folder_id == folder.id)).scalars().all()
# Locked as they are read: a file moved out meanwhile stays where it went, and one deleted
# meanwhile is not written back (its gone row would fail the delete).
jobs = (
db.execute(
select(DownloadJob)
.where(DownloadJob.folder_id == folder.id)
.with_for_update()
.execution_options(populate_existing=True)
)
.scalars()
.all()
)
for job in jobs:
job.folder_id = source_folder(db, job.user_id, source_key_for(db, job)).id
restamp(db, job)
+86 -49
View File
@@ -22,7 +22,7 @@ import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
from sqlalchemy import func, or_, select
from sqlalchemy import func, or_, select, update
from sqlalchemy.orm import Session
from app import logs, sysconfig
@@ -82,6 +82,59 @@ def _relock_expired(
return job if still else None
def _mark_warned(db: Session, job_id: int, expires_at: datetime) -> DownloadJob | None:
"""Mark one file warned if it still goes when the pass saw it go — not deleted, kept, moved or
re-timed since (each of those sets a fresh expiry, and its own warning to come) — and return it.
The UPDATE's row lock holds it so until the commit."""
hit = db.execute(
update(DownloadJob)
.where(
DownloadJob.id == job_id,
DownloadJob.status == JobStatus.DONE,
DownloadJob.kept.is_(False),
DownloadJob.expiry_warned.is_(False),
DownloadJob.expires_at == expires_at,
)
.values(expiry_warned=True)
.returning(DownloadJob.id)
.execution_options(synchronize_session=False)
).scalar_one_or_none()
return db.get(DownloadJob, job_id) if hit is not None else None
def _on_disk(db: Session) -> int:
"""The bytes the cache holds now (every asset whose file is on disk)."""
return (
db.execute(
select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(quota.ON_DISK)
).scalar()
or 0
)
def _relock_evictable(db: Session, asset_id: int) -> tuple[MediaAsset | None, list[DownloadJob]]:
"""Re-read an eviction candidate and its holders under their locks (the asset, then its jobs —
the order the user routes take): between the pass's SELECT and here a holder may have kept it,
or every holder deleted it."""
asset = db.execute(
select(MediaAsset)
.where(MediaAsset.id == asset_id)
.with_for_update()
.execution_options(populate_existing=True)
).scalar_one_or_none()
holders = (
db.execute(
select(DownloadJob)
.where(DownloadJob.asset_id == asset_id)
.with_for_update()
.execution_options(populate_existing=True)
)
.scalars()
.all()
)
return asset, list(holders)
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)."""
@@ -176,29 +229,28 @@ def run_download_gc(db: Session) -> dict:
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.
# 1. Pre-expiry warnings — one per file, the first run inside the grace window. Each mark is
# conditional on the expiry the SELECT saw and committed on its own: a file deleted since is
# passed over (a write to its gone row would fail the whole run), and one kept, moved or re-timed
# since keeps its fresh expiry unwarned.
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,
)
soon = db.execute(
select(DownloadJob.id, DownloadJob.expires_at).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:
).all()
db.rollback() # the pass-1 snapshot holds no locks; each mark takes its own below
for job_id, expires_at in soon:
job = _mark_warned(db, job_id, expires_at)
if job is not None:
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
_notify(db, job, asset, "download_expiring")
warned += 1
db.commit()
# 2. Expiry — a file past its folder's retention leaves the library (warned in pass 1).
@@ -224,19 +276,14 @@ def run_download_gc(db: Session) -> dict:
# 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
)
total = _on_disk(db)
if total > cap:
kept_assets = select(DownloadJob.asset_id).where(
DownloadJob.kept.is_(True), DownloadJob.asset_id.is_not(None)
)
candidates = (
candidate_ids = (
db.execute(
select(MediaAsset)
select(MediaAsset.id)
.where(
MediaAsset.status == "ready",
MediaAsset.source_kind != "upload",
@@ -247,30 +294,20 @@ def run_download_gc(db: Session) -> dict:
.scalars()
.all()
)
candidate_ids = [a.id for a in candidates]
db.rollback()
for asset_id in candidate_ids:
if total <= cap:
break
# Locked and re-read like an expiry: a holder may have kept it, or every holder
# deleted it (the file is then already gone and counted out) since the SELECT.
asset = db.execute(
select(MediaAsset)
.where(MediaAsset.id == asset_id)
.with_for_update()
.execution_options(populate_existing=True)
).scalar_one_or_none()
holders = (
db.execute(
select(DownloadJob)
.where(DownloadJob.asset_id == asset_id)
.with_for_update()
.execution_options(populate_existing=True)
)
.scalars()
.all()
)
if asset is None or asset.status != "ready" or any(j.kept for j in holders):
# deleted it since the SELECT.
asset, holders = _relock_evictable(db, asset_id)
if asset is None or asset.status != "ready":
# Its file went (or is being fetched afresh) since the sum: the disk is counted
# again, or the pass would evict another user's file for space already free.
db.commit()
total = _on_disk(db)
continue
if any(j.kept for j in holders):
db.commit()
continue
total -= asset.size_bytes or 0
+23 -1
View File
@@ -65,6 +65,23 @@ def reference_url(job, asset) -> str | None:
return None
def library_left_behind(db: Session, files_root: str) -> bool:
"""True when the library names files and none of the newest of them is under `files_root` — a
fresh, empty volume mounted in the library's place (a compose file changed under an install, its
old volume left behind). The mount check cannot see that: the new volume is a real mount."""
rels = (
db.execute(
select(MediaAsset.rel_path)
.where(MediaAsset.status == "ready", MediaAsset.rel_path.is_not(None))
.order_by(MediaAsset.id.desc())
.limit(20)
)
.scalars()
.all()
)
return bool(rels) and not any(storage.abs_path(files_root, r).exists() for r in rels if r)
def get_or_create_asset(
db: Session, source_kind: str, source_ref: str, format_sig: str
) -> MediaAsset:
@@ -77,7 +94,12 @@ def get_or_create_asset(
MediaAsset.source_ref == source_ref,
MediaAsset.format_sig == format_sig,
)
asset = db.execute(stmt).scalar_one_or_none()
# Locked to the caller's commit: a release or the GC freeing this asset meanwhile (they lock it
# too) either waits for the new hold, or has deleted the row by the time this read returns — and
# the caller then gets a fresh one, never a job pointing at a row that is gone.
asset = db.execute(
stmt.with_for_update().execution_options(populate_existing=True)
).scalar_one_or_none()
if asset is not None:
# A previously-FAILED shared asset must be re-attempted for the new job about to hold it —
# reset it to pending (+clear the error) so the worker actually re-downloads instead of
+43 -11
View File
@@ -68,20 +68,52 @@ _FOLD_MAP = str.maketrans(
)
def files_root_problem(root: str, system_root: str = "/") -> str | None:
"""Why `root` cannot be the files library's volume, or None when it can. A root that sits on the
container's own filesystem is the image's empty directory, not a mounted volume: the library
would answer 404 for every file, and what a download writes would go when the container does —
the state an install is in when its compose file still mounts the library somewhere else."""
def _mount_points(mountinfo: str) -> list[str]:
"""The mount points a /proc/<pid>/mountinfo table lists (its fifth field, octal escapes undone)."""
points = []
for line in mountinfo.splitlines():
fields = line.split(" ")
if len(fields) > 4:
points.append(re.sub(r"\\([0-7]{3})", lambda m: chr(int(m.group(1), 8)), fields[4]))
return points
def _in_container() -> bool:
return Path("/.dockerenv").exists() or Path("/run/.containerenv").exists()
def _read_mountinfo() -> str | None:
try:
return Path("/proc/self/mountinfo").read_text()
# silent-ok: no mount table to read (not Linux) — the caller then cannot tell, and refuses nothing.
except OSError:
return None
def files_root_problem(
root: str, mountinfo: str | None = None, in_container: bool | None = None
) -> str | None:
"""Why `root` cannot be the files library's volume, or None when it can. In a container, a root
that is no mount of its own (nor inside one) is the image's empty directory: the library would
answer 404 for every file, and what a download writes would go when the container does — the
state an install is in when its compose file still mounts the library somewhere else.
Asked of the kernel's mount table, not by comparing devices: a storage driver that keeps the
container's root on the host's own disk (vfs, the fallback under Docker-in-LXC on older ZFS) gives
a mounted volume the same device as `/`. Outside a container a plain directory is a fine root."""
path = Path(root)
if not path.is_dir():
return f"{root} does not exist"
try:
if path.stat().st_dev == Path(system_root).stat().st_dev:
return f"{root} is not a mounted volume (it is the container's own empty directory)"
# silent-ok: the failure IS the answer — returned as the reason, which the caller logs.
except OSError as e:
return f"{root} cannot be read ({e})"
if in_container if in_container is not None else _in_container():
table = mountinfo if mountinfo is not None else _read_mountinfo()
if table is not None:
real = os.path.realpath(root)
mounted = any(
m != "/" and (real == m or real.startswith(m.rstrip("/") + "/"))
for m in _mount_points(table)
)
if not mounted:
return f"{root} is not a mounted volume (it is the container's own empty directory)"
if not os.access(path, os.W_OK):
return f"{root} is not writable"
return None
+4 -2
View File
@@ -70,7 +70,9 @@ def ingest_upload(
rejected upload leaves nothing behind and the caller just cleans up its temp file."""
quota.check_upload(db, user_id, size) # the file's size is known → charge it up front
if folder_id is not None:
folders.upload_target(db, user_id, folder_id)
# Refused before any work; held only once the job is about to name it (below), not through
# the probe and the poster, which a rename or delete of the folder would wait out.
folders.upload_target(db, user_id, folder_id, hold=False)
data = probe.probe_json(src_path)
vcodec, acodec, width, height, duration = probe.facts(data)
@@ -138,7 +140,7 @@ def ingest_upload(
db.add(job)
folders.place(db, job)
if folder_id is not None:
job.folder_id = folder_id
job.folder_id = folders.upload_target(db, user_id, folder_id).id
if not convert:
folders.stamp_done(db, job)
service.incr_ref_count(db, asset.id)
+13
View File
@@ -28,6 +28,7 @@ log = logs.get_logger(__name__)
from app import bodylimit, collation, logsink, state, sysconfig # noqa: E402 — logging is configured first
from app.config import settings # noqa: E402 — logging is configured first
from app.db import SessionLocal, engine # noqa: E402 — logging is configured first
from app.downloads import service as download_service # noqa: E402 — logging is configured first
from app.downloads import storage # noqa: E402 — logging is configured first
from app.plex import stream as plex_stream # noqa: E402 — logging is configured first
from app.plex import subtitles as plex_subtitles # noqa: E402 — logging is configured first
@@ -84,6 +85,18 @@ async def lifespan(app: FastAPI):
# With the traceback: this catch spans a connection, a query and a coercion at boot,
# and `%s` of the exception does not say which of them raised.
log.exception("could not read the stored log level; staying at %s", current_level())
if not problem:
try:
left = download_service.library_left_behind(db, settings.files_root)
except Exception: # a database not reachable yet: the check is a hint, never a gate
log.exception("could not check the files library against its database")
else:
if left:
log.critical(
"The files library is empty but the database names files: the volume "
"mounted at %s is not the one that holds them. See docs/self-hosting.md.",
settings.files_root,
)
if not state.is_configured(db):
token = state.rotate_setup_token(db)
url = f"{settings.app_base}/setup?token={token}"
+2 -6
View File
@@ -339,10 +339,6 @@ def _serialize_folder(folder: FileFolder, count: int, size: int) -> dict:
}
def _own_folder_or_404(db: Session, user: User, folder_id: int) -> FileFolder:
return owned_or_404(db, FileFolder, folder_id, user, what="folder")
@router.get("/folders")
def list_folders(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]:
"""The caller's folders with how many files each holds and their size. A source folder its
@@ -407,7 +403,7 @@ def update_folder(
) -> dict:
"""Rename one of the caller's own folders (a source folder's name is its origin's), and/or set
how long any of their folders keeps a file (`folders.set_retention`)."""
folder = _own_folder_or_404(db, user, folder_id)
folder = owned_or_404(db, FileFolder, folder_id, user, what="folder")
sent = body.model_fields_set
if "name" in sent:
if folder.kind != folders.FolderKind.OWN:
@@ -457,7 +453,7 @@ def delete_folder(
folder_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
"""Delete one of the caller's own folders; its files go back to their source folders."""
folder = _own_folder_or_404(db, user, folder_id)
folder = owned_or_404(db, FileFolder, folder_id, user, what="folder")
try:
moved = folders.delete_own(db, folder)
except folders.FolderError as e: