peter
d73cdcee97
The models now live in their units' packages, and a core model names a unit's by string (User.subscriptions -> Subscription). The download child imports neither catalog nor app.tables before db.get, so mapper configuration failed and every download, clip, edit and upload job died with child_exit. Each entry point now imports app.tables, and a test imports each one in a clean interpreter.
129 lines
5.8 KiB
Python
129 lines
5.8 KiB
Python
"""Download/edit child process — runs ONE job's heavy work (yt-dlp download or ffmpeg edit) in its
|
|
own process, launched and supervised by the worker (`app.worker._run_supervised`).
|
|
|
|
Why a separate process: a worker THREAD can't be killed, a C-level spin (a catastrophic regex, a
|
|
tight yt-dlp retry loop) can't even be cooperatively cancelled, and one such thread starves the
|
|
others via the GIL. Running the work in a child means the supervisor can SIGTERM/SIGKILL it (and its
|
|
process group, so ffmpeg grandchildren die too) on a stall / pause / wall-clock breach.
|
|
|
|
Contract: the asset is already claimed ('downloading') by the parent. This process does the work and
|
|
writes the terminal job/asset state itself (done / error, or a cooperative pause via the in-flight
|
|
progress hook), then exits 0. An unexpected crash exits non-zero and the parent fails the job.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
|
|
import app.tables # noqa: F401 — a fresh interpreter: every table, before the first query
|
|
from app import logs
|
|
from app import logsink
|
|
from app.config import settings
|
|
from app.db import SessionLocal
|
|
from app.downloads import edit as editmod
|
|
from app.downloads import formats
|
|
from app.downloads.status import JobKind
|
|
from app.downloads.models import DownloadJob
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
|
|
def _deprioritize() -> None:
|
|
"""Lower THIS child's CPU + IO priority so its yt-dlp/ffmpeg work — every descendant inherits the
|
|
handicap — can never starve the co-located API, Postgres, or Plex playback. Run once at startup,
|
|
before any heavy work. Every step is best-effort: a failed renice must never fail a download."""
|
|
nice_inc = settings.download_child_nice
|
|
if nice_inc > 0:
|
|
try:
|
|
os.nice(nice_inc) # relative increment; the child starts at the worker's nice (0)
|
|
except OSError as exc:
|
|
log.info("could not renice download child: %s", str(exc)[:120])
|
|
cls = settings.download_child_ionice_class
|
|
if cls > 0 and shutil.which("ionice"): # util-linux; absent on some minimal images
|
|
try:
|
|
subprocess.run(
|
|
["ionice", "-c", str(cls), "-p", str(os.getpid())],
|
|
capture_output=True,
|
|
timeout=5,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.SubprocessError) as exc:
|
|
log.info("could not ionice download child: %s", str(exc)[:120])
|
|
# Cap OpenMP-parallel ffmpeg filters to the same budget as the encoder threads, so a filter
|
|
# graph can't fan out across every core behind ffmpeg's own `-threads` cap.
|
|
os.environ.setdefault("OMP_NUM_THREADS", str(editmod.encode_thread_cap()))
|
|
|
|
|
|
def main() -> None:
|
|
logs.configure_logging("runner")
|
|
_deprioritize()
|
|
if len(sys.argv) != 2:
|
|
# T201: a usage line is for the person who typed the command. Logging is already
|
|
# configured here, and in a deployment it writes JSON records — the one reader of this
|
|
# line wants the bare sentence, not a record with a request id they do not have.
|
|
print( # noqa: T201
|
|
"usage: python -m app.downloads.runner <job_id>", file=sys.stderr
|
|
)
|
|
raise SystemExit(2)
|
|
job_id = int(sys.argv[1])
|
|
# This whole process exists to run ONE job — tag every line it prints with it, so the child's
|
|
# output joins up with the worker's and the API's view of the same download.
|
|
logs.set_job(job_id)
|
|
logsink.start() # flushed at exit (atexit) — this process has no shutdown hook of its own
|
|
|
|
with SessionLocal() as db:
|
|
job = db.get(DownloadJob, job_id)
|
|
if job is None:
|
|
return
|
|
asset_id = job.asset_id
|
|
job_kind = job.job_kind
|
|
spec = dict(job.profile_snapshot or {})
|
|
source_kind, source_ref = job.source_kind, job.source_ref
|
|
user_id = job.user_id
|
|
if asset_id is None:
|
|
return
|
|
|
|
_run(job_id, asset_id, job_kind, source_kind, source_ref, spec, user_id)
|
|
|
|
|
|
def _run(job_id, asset_id, job_kind, source_kind, source_ref, spec, user_id) -> None:
|
|
"""Run the job, and be the last stop before Python's own exception hook.
|
|
|
|
What gets past the pipeline is a failure it could not even record (the database refused the
|
|
write that fails the job). Python's hook would print it to stderr raw — and its context is the
|
|
download's own failure, whose text is the URL or ffmpeg's stderr naming the file. Here it is
|
|
one line with the whole chain scrubbed; the non-zero exit then has the worker fail the job
|
|
(`cause=child_exit`)."""
|
|
# Imported here (not at module top) so the child pays the yt-dlp import cost only when it has
|
|
# real work, and to keep the import graph obvious: runner is a thin entry over the pipeline.
|
|
from app.downloads.pipeline import _convert_upload, _download, _process_clip, _process_edit
|
|
|
|
with formats.hiding(source_ref):
|
|
try:
|
|
if job_kind == JobKind.EDIT:
|
|
_process_edit(job_id, asset_id)
|
|
elif job_kind == JobKind.UPLOAD:
|
|
_convert_upload(job_id, asset_id)
|
|
elif job_kind == JobKind.CLIP:
|
|
_process_clip(job_id, asset_id)
|
|
else:
|
|
_download(job_id, asset_id, source_kind, source_ref, spec, user_id)
|
|
except Exception as exc: # noqa: BLE001 — whatever it is, it is reported here or raw
|
|
# TRY400: the stack goes with it scrubbed (`scrubbed_stack`); `exc_info` is raw.
|
|
log.error( # noqa: TRY400
|
|
"the download child for job %s failed",
|
|
job_id,
|
|
extra={
|
|
**logs.event(
|
|
"download.child.failed", job_id=job_id, error_type=type(exc).__name__
|
|
),
|
|
**logs.scrubbed_stack(exc, formats.log_safe),
|
|
},
|
|
)
|
|
raise SystemExit(1) from None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|