peter
89d18b46a6
- A converted upload over the space left (not zero) fails with CONVERTED_TOO_BIG_MESSAGE, not "storage is full"; it classifies as quota_exceeded by the phrase it shares with the clip's. - The public watch page says "preparing" only for a job a worker will take: a suspended owner's queued job waits for as long as the suspension lasts, so the link says the video is gone, as for a paused rebuild. - The worker's claim docstring names the restart case: a running job put back in the queue waits.
728 lines
32 KiB
Python
728 lines
32 KiB
Python
"""Download worker process entry point.
|
|
|
|
Runs in a DEDICATED container (`command: python -m app.worker`, `WORKER_ENABLED=1`), separate
|
|
from the API so long yt-dlp downloads + ffmpeg postprocessing never block web requests.
|
|
|
|
Design:
|
|
* N loop threads (download_worker_concurrency) each claim a `queued` DownloadJob with
|
|
FOR UPDATE SKIP LOCKED and process it. N and the log level follow the settings page while the
|
|
worker runs (`_sync_knobs`), not only the env it booted with.
|
|
* The shared MediaAsset is the dedup unit. A job whose asset is already `ready` finishes
|
|
instantly (no re-download); a fresh asset is claimed pending→downloading by exactly one
|
|
worker (a DB compare-and-set), so two jobs for the same video+format never double-download.
|
|
* Live progress is written to the job row (short-lived sessions, throttled ~1/s) — the
|
|
frontend polls it (the worker is a separate process with no channel to the browser).
|
|
* Pause/cancel are cooperative: the progress hook re-reads the job status and aborts the
|
|
yt-dlp download if it's no longer `running`.
|
|
* Crash-safe: on startup, orphaned `running` jobs are requeued and `downloading` assets reset.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable
|
|
from datetime import datetime, timezone
|
|
from functools import partial
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import logs
|
|
from app import logsink
|
|
from app import sysconfig
|
|
from app.config import settings
|
|
from app.db import SessionLocal
|
|
from app.downloads import edit as editmod
|
|
from app.downloads import credentials, formats, quota, service
|
|
from app.downloads import errors as dlerrors
|
|
from app.downloads.jobdb import (
|
|
_edit_staging_dir,
|
|
_requeue_if_running,
|
|
_reset_asset_pending,
|
|
_staging_dir,
|
|
)
|
|
from app.downloads.status import STOPPED_BY_ROUTE, JobKind, JobStatus
|
|
from app.models import DISPLAY_TEXT_MAX, DownloadJob, MediaAsset
|
|
|
|
log = logs.get_logger(__name__)
|
|
_stop = threading.Event()
|
|
# How many loop threads may claim jobs. A thread whose index is at or above it stops after its
|
|
# current job, so lowering the knob never abandons a download halfway.
|
|
_target = 1
|
|
# How often the worker re-reads the knobs it obeys; the settings page says nothing about a restart.
|
|
_KNOB_SECONDS = 10.0
|
|
|
|
|
|
def _handle_signal(signum, frame):
|
|
_stop.set()
|
|
|
|
|
|
def _read_knobs() -> tuple[int, str] | None:
|
|
"""The stored concurrency and log level, or None when the database cannot answer — the caller
|
|
then keeps what already runs."""
|
|
try:
|
|
with SessionLocal() as db:
|
|
return (
|
|
max(1, sysconfig.get_int(db, "download_worker_concurrency")),
|
|
sysconfig.get_str(db, "log_level"),
|
|
)
|
|
except Exception: # noqa: BLE001 — any DB or config failure keeps the knobs already in force
|
|
log.warning(
|
|
"could not read the worker's settings; keeping the ones in force",
|
|
exc_info=True,
|
|
extra=logs.event("download.worker.knobs_unreadable"),
|
|
)
|
|
return None
|
|
|
|
|
|
def _sync_knobs(threads: dict[int, threading.Thread], level: str | None) -> str | None:
|
|
"""Apply the stored knobs: run loop threads up to the concurrency (surplus ones retire after
|
|
their job, see `_target`) and re-apply the log level when it changed. Returns the level in force."""
|
|
global _target
|
|
knobs = _read_knobs()
|
|
if knobs is not None:
|
|
target, stored_level = knobs
|
|
if target != _target:
|
|
log.info(
|
|
"download concurrency is now %d",
|
|
target,
|
|
extra=logs.event("download.worker.concurrency", before=_target, after=target),
|
|
)
|
|
_target = target
|
|
if stored_level != level:
|
|
logs.set_level(stored_level)
|
|
level = stored_level
|
|
for i in range(_target):
|
|
t = threads.get(i)
|
|
if t is None or not t.is_alive():
|
|
threads[i] = threading.Thread(target=_worker_loop, args=(i,), daemon=True)
|
|
threads[i].start()
|
|
return level
|
|
|
|
|
|
# --- claim + recovery -----------------------------------------------------------------------
|
|
|
|
|
|
def _wait_for_schema(timeout: float = 300.0) -> bool:
|
|
"""Block until the download tables exist before doing any DB work. Returns False if they never
|
|
appeared (or we were asked to stop while waiting).
|
|
|
|
The API applies migrations on its OWN startup, and the worker is a separate container that may
|
|
boot first (they start in parallel). Without this, the worker would hit a missing `download_jobs`
|
|
table and crash-loop until the API's migrations land. Poll patiently instead — and if the schema
|
|
is still absent when the timeout runs out, say so and let the caller exit, because "continuing"
|
|
only moved the same crash one call further down (into `_recover_orphans`), where it read as an
|
|
unrelated failure."""
|
|
deadline = time.monotonic() + timeout
|
|
while not _stop.is_set():
|
|
try:
|
|
with SessionLocal() as db:
|
|
db.execute(text("SELECT 1 FROM download_jobs LIMIT 1"))
|
|
return True
|
|
except Exception: # table not migrated yet / DB not up yet
|
|
if time.monotonic() > deadline:
|
|
# The traceback, not `str(exc)[:120]`: this is the last thing the worker says
|
|
# before it exits, and which driver call failed is the whole question.
|
|
log.exception(
|
|
"download schema still absent after %ss — the API's migrations have not "
|
|
"landed; exiting so the container restarts",
|
|
timeout,
|
|
)
|
|
return False
|
|
log.info("waiting for the database schema (API migrations)…")
|
|
_stop.wait(3.0)
|
|
return False
|
|
|
|
|
|
def _recover_orphans() -> None:
|
|
"""Reset state left behind by a previous crash so nothing is stuck."""
|
|
with SessionLocal() as db:
|
|
db.execute(text("UPDATE download_jobs SET status='queued' WHERE status='running'"))
|
|
db.execute(text("UPDATE media_assets SET status='pending' WHERE status='downloading'"))
|
|
db.commit()
|
|
|
|
|
|
def _claim_job() -> int | None:
|
|
"""Claim the oldest queued job whose user is under their per-user concurrency limit AND whose
|
|
shared asset isn't already being downloaded by another worker.
|
|
|
|
The asset exclusion (`NOT EXISTS … media_assets.status='downloading'`) stops the same-asset ping-
|
|
pong (B-sub-3): without it, a second worker claims a queued job that shares an in-flight asset,
|
|
finds it busy, requeues it, and immediately re-claims it — burning a worker slot behind one
|
|
in-flight asset and starving everyone else. Excluded here, that job simply waits until its asset
|
|
settles, so `_process_job` no longer needs to sleep on the busy/lost-race paths.
|
|
|
|
Users already AT their concurrency limit are excluded IN THE SQL, so one user flooding the
|
|
queue can't monopolize the fetch window and starve everyone behind them. (The old fixed
|
|
LIMIT-50 window did exactly that: 50 queued jobs from a single capped user filled the window,
|
|
`chosen` came back None, and every other user's work — sitting past position 50 — was never
|
|
seen.) The per-user limit is the quota row's `max_concurrent` (else the sysconfig default);
|
|
`unlimited` users are never capped. A final Python `at_concurrency_limit` re-check still guards
|
|
the tiny race between the exclusion subquery and the claim (it can only ever reject, never
|
|
over-claim), so correctness never depends on the two observations agreeing.
|
|
|
|
A suspended user's queued jobs are left where they are: suspension blocks their downloads (not
|
|
only their sign-in), and lifting it lets them run in their old place. A job already running
|
|
when the user is suspended finishes on that worker; one a worker restart puts back in the queue
|
|
(`_requeue_if_running`, `_recover_orphans`) waits with the rest."""
|
|
with SessionLocal() as db:
|
|
default_mc = sysconfig.get_int(db, "download_default_max_concurrent")
|
|
rows = db.execute(
|
|
text(
|
|
"""
|
|
SELECT id, user_id FROM download_jobs
|
|
WHERE status='queued'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM users u
|
|
WHERE u.id = download_jobs.user_id AND u.is_suspended
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM media_assets a
|
|
WHERE a.id = download_jobs.asset_id AND a.status = 'downloading'
|
|
)
|
|
AND user_id NOT IN (
|
|
SELECT r.user_id
|
|
FROM download_jobs r
|
|
LEFT JOIN download_quotas q ON q.user_id = r.user_id
|
|
WHERE r.status='running' AND r.user_id IS NOT NULL
|
|
GROUP BY r.user_id, q.max_concurrent, q.unlimited
|
|
HAVING COALESCE(q.unlimited, false) = false
|
|
AND count(*) >= GREATEST(COALESCE(q.max_concurrent, :dmc), 1)
|
|
)
|
|
ORDER BY queue_pos, id
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT 50
|
|
"""
|
|
),
|
|
{"dmc": default_mc},
|
|
).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 chosen
|
|
|
|
|
|
def _claim_asset(asset_id: int) -> bool:
|
|
"""Compare-and-set pending→downloading. True if THIS worker won the right to download."""
|
|
with SessionLocal() as db:
|
|
row = db.execute(
|
|
text(
|
|
"UPDATE media_assets SET status='downloading' "
|
|
"WHERE id = :id AND status='pending' RETURNING id"
|
|
),
|
|
{"id": asset_id},
|
|
).fetchone()
|
|
db.commit()
|
|
return row is not None
|
|
|
|
|
|
# --- supervised child execution + watchdog --------------------------------------------------
|
|
# The heavy work runs in a child process so a wedged yt-dlp/ffmpeg is HARD-killable. A worker thread
|
|
# can't be killed, a C-level spin (e.g. a catastrophic regex) can't even be cooperatively cancelled,
|
|
# and one such thread starves the rest via the GIL — the incident that motivated this. The parent
|
|
# thread supervises: it kills the child (and its process group, so ffmpeg grandchildren die too) on
|
|
# a route pause/cancel, on a no-progress stall, or past an absolute wall-clock cap.
|
|
#
|
|
# POSIX-only: the kill path uses process groups (start_new_session / os.killpg / os.getpgid), so the
|
|
# worker is Linux-container-only — never run `python -m app.worker` natively on Windows.
|
|
# Edits ALSO carry a second, tighter watchdog inside run_ffmpeg (a per-ffmpeg stall timeout with a
|
|
# cleaner "ffmpeg stalled" error); this process-level supervisor is the uniform backstop for both the
|
|
# download and edit paths, and the last line of defence for anything run_ffmpeg can't see.
|
|
|
|
_CHILD_OK = 0
|
|
|
|
|
|
def _spawn_download_child(job_id: int) -> subprocess.Popen:
|
|
"""Start the child that does the actual download/edit for `job_id`, in its OWN session/process
|
|
group so stopping it also stops any ffmpeg / helper processes yt-dlp spawned."""
|
|
return subprocess.Popen(
|
|
[sys.executable, "-m", "app.downloads.runner", str(job_id)],
|
|
start_new_session=True,
|
|
)
|
|
|
|
|
|
def _job_liveness(job_id: int) -> tuple[str | None, float | None]:
|
|
"""(status, seconds-since-last-row-update) for the job, computed with the DB clock so there is
|
|
no parent↔DB skew. A live download rewrites its row ~1/s; a stalled one never does. Returns
|
|
(None, None) if the row is gone."""
|
|
with SessionLocal() as db:
|
|
row = db.execute(
|
|
text(
|
|
"SELECT status, EXTRACT(EPOCH FROM (now() - updated_at)) "
|
|
"FROM download_jobs WHERE id = :id"
|
|
),
|
|
{"id": job_id},
|
|
).one_or_none()
|
|
if row is None:
|
|
return None, None
|
|
return row[0], (float(row[1]) if row[1] is not None else None)
|
|
|
|
|
|
def _signal_group(proc: subprocess.Popen, pgid: int | None, sig: int) -> None:
|
|
"""Signal the child's whole process group (falling back to the child alone), swallowing the race
|
|
where it has already exited."""
|
|
try:
|
|
if pgid is not None:
|
|
os.killpg(pgid, sig)
|
|
else:
|
|
proc.send_signal(sig)
|
|
except ProcessLookupError:
|
|
pass # silent-ok: the child already exited — the race this function is written around
|
|
except OSError as exc:
|
|
# Any OTHER refusal means the signal did not land and the child is still running, which
|
|
# is how a "cancelled" download keeps writing to a staging tree nobody is watching.
|
|
log.info(
|
|
"a download child could not be signalled",
|
|
extra=logs.event(
|
|
"download.child.signal_failed", signal=int(sig), reason=type(exc).__name__
|
|
),
|
|
)
|
|
|
|
|
|
def _stop_child(proc: subprocess.Popen, grace_s: float) -> None:
|
|
"""SIGTERM the child's process GROUP, then SIGKILL if it outlives the grace period. Shares the
|
|
terminate→wait→kill escalation with run_ffmpeg (`editmod.escalate_kill`); the only difference is
|
|
that we signal the whole group so ffmpeg grandchildren die too. `escalate_kill` no-ops if the
|
|
child already exited, and `getpgid` yields None for a reaped pid, so no separate poll guard."""
|
|
try:
|
|
pgid: int | None = os.getpgid(proc.pid)
|
|
# silent-ok: `getpgid` refuses for a reaped pid, which the docstring above already treats as
|
|
# the ordinary case — the escalation then signals the child alone and no-ops if it is gone.
|
|
except OSError:
|
|
pgid = None
|
|
editmod.escalate_kill(
|
|
proc,
|
|
terminate=lambda: _signal_group(proc, pgid, signal.SIGTERM),
|
|
kill=lambda: _signal_group(proc, pgid, signal.SIGKILL),
|
|
grace_s=grace_s,
|
|
label="download child",
|
|
)
|
|
|
|
|
|
def _stop_and_free(job_id: int, asset_id: int, proc: subprocess.Popen, grace_s: float) -> None:
|
|
"""Kill the child, free its claimed asset, and clear its staging — the common tail of every path
|
|
that stops a child (a route pause/cancel, a shutdown, an over-cap abandon)."""
|
|
_stop_child(proc, grace_s)
|
|
_reset_asset_pending(asset_id)
|
|
_cleanup_staging(asset_id)
|
|
|
|
|
|
def _fail_after_kill(
|
|
job_id: int,
|
|
asset_id: int,
|
|
proc: subprocess.Popen,
|
|
grace_s: float,
|
|
error: str,
|
|
error_code: str,
|
|
cause: str | None = None,
|
|
) -> bool:
|
|
"""`_stop_and_free` + mark the job errored — for the watchdog's stall/timeout kills and the
|
|
supervisor-error fallback. Returns whether the job failed here (`_fail_running`)."""
|
|
_stop_and_free(job_id, asset_id, proc, grace_s)
|
|
return _fail_running(job_id, error=error, error_code=error_code, cause=cause)
|
|
|
|
|
|
def _fail_running(
|
|
job_id: int,
|
|
*,
|
|
error: str,
|
|
error_code: str | None,
|
|
error_detail: str | None = None,
|
|
cause: str | None = None,
|
|
exit_code: int | None = None,
|
|
) -> bool:
|
|
"""Fail a claimed job — only while it is still `running` (`service.finalize_if_running`), so a
|
|
cancel/pause a route landed first is never turned back into `error`, which holds the asset the
|
|
route already released. `error_detail` is written every time: a failure never shows an earlier
|
|
failure's explanation.
|
|
|
|
`cause` is the worker's half of `download.job.failed`: the failures the child never got to
|
|
report because it was killed, never ran or died — a stall, the wall-clock cap, a non-zero exit
|
|
(with its `exit_code`). Without it those left no line at all. A caller that logs its own line
|
|
(a worker error, with its stack — `_report_worker_error`) passes none and uses the returned
|
|
verdict. Logged only when the job actually failed here: a route's cancel that landed first is
|
|
not a failure. Returns whether it failed here."""
|
|
with SessionLocal() as db:
|
|
failed = service.finalize_if_running(
|
|
db,
|
|
job_id,
|
|
status=JobStatus.ERROR,
|
|
error=error,
|
|
error_code=error_code,
|
|
error_detail=error_detail,
|
|
phase=None,
|
|
)
|
|
db.commit()
|
|
if cause is not None and failed:
|
|
log.warning(
|
|
"job %s failed in the worker: %s",
|
|
job_id,
|
|
cause,
|
|
extra=logs.event(
|
|
"download.job.failed",
|
|
job_id=job_id,
|
|
error_code=error_code,
|
|
cause=cause,
|
|
**({"exit_code": exit_code} if exit_code is not None else {}),
|
|
),
|
|
)
|
|
return failed
|
|
|
|
|
|
def _report_worker_error(job_id: int, exc: Exception, cause: str, fail: Callable[[], bool]) -> None:
|
|
"""A worker error on a job (the supervisor loop, or processing the job, raised): fail the job,
|
|
then ONE line with the stack. Its event says what the error did to the job —
|
|
`download.job.failed` when the job failed with it, `download.worker.error` when a route had
|
|
already stopped it (a cancel that landed first is not a failure, `_fail_running`).
|
|
|
|
The line goes out even when failing the job raises (the database is down): as
|
|
`download.worker.error`, since the job was not failed, and the raise goes on to the caller —
|
|
without `exc` as its context, which this line has already carried with its stack."""
|
|
failed = False
|
|
try:
|
|
failed = fail()
|
|
except BaseException as write_error:
|
|
write_error.__suppress_context__ = True
|
|
raise
|
|
finally:
|
|
log.error(
|
|
"job %s: worker error (%s)",
|
|
job_id,
|
|
cause,
|
|
exc_info=exc,
|
|
extra=logs.event(
|
|
"download.job.failed" if failed else "download.worker.error",
|
|
job_id=job_id,
|
|
cause=cause,
|
|
),
|
|
)
|
|
|
|
|
|
def _run_supervised(job_id: int, asset_id: int) -> None:
|
|
"""Run the heavy download/edit for an already-claimed asset in a child process and watchdog it.
|
|
|
|
A clean child exit means the child already wrote the terminal job/asset state (done/error, or a
|
|
cooperative pause via the in-child progress hook) — we leave it. We step in only when the child
|
|
can't finish or self-abort: a route paused/cancelled it (kill, free the asset, keep the route's
|
|
status), it made no progress for `stall_timeout` or ran past `wall_clock_max` (kill, fail the
|
|
job), the worker is shutting down (kill, requeue), or it crashed without setting state."""
|
|
poll_s = max(1, settings.download_supervisor_poll_seconds)
|
|
stall_s = settings.download_stall_timeout_seconds
|
|
wall_s = settings.download_wall_clock_max_seconds
|
|
grace_s = settings.download_kill_grace_seconds
|
|
|
|
try:
|
|
proc = _spawn_download_child(job_id)
|
|
except Exception: # spawn failed; don't leave the asset stuck 'downloading'
|
|
log.exception("failed to spawn download child for job %s", job_id)
|
|
_reset_asset_pending(asset_id)
|
|
_requeue_if_running(job_id, None)
|
|
return
|
|
|
|
started = time.monotonic()
|
|
reason: str | None = None
|
|
try:
|
|
while True:
|
|
try:
|
|
proc.wait(timeout=poll_s)
|
|
break # child finished on its own
|
|
# silent-ok: the timeout is the loop's clock, not a failure — it is what gives this
|
|
# thread a turn to check whether a route cancelled the job.
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
status, stale = _job_liveness(job_id)
|
|
if status is None or status in STOPPED_BY_ROUTE:
|
|
reason = "cancelled" # a route stopped it (or the row vanished)
|
|
break
|
|
if status == JobStatus.RUNNING:
|
|
if stale is not None and stale > stall_s:
|
|
reason = "stalled"
|
|
break
|
|
if _stop.is_set():
|
|
reason = "shutdown"
|
|
break
|
|
# Absolute wall-clock cap frees the worker slot even for a FINISHED child wedged in
|
|
# teardown. A still-running job over the cap is failed ("timeout"); a terminal one (already
|
|
# done/error) just gets its lingering child killed, its DB status left intact ("abandon").
|
|
if time.monotonic() - started > wall_s:
|
|
reason = "timeout" if status == JobStatus.RUNNING else "abandon"
|
|
break
|
|
# else: a terminal status under the cap (done/error, or a re-queued contended asset) means
|
|
# the child is exiting on its own — keep polling; NEVER kill a finishing child (that could
|
|
# drop its completion notice).
|
|
except Exception as exc: # noqa: BLE001 — a watchdog error must NEVER leave the child running unsupervised
|
|
# silent-ok: not silent — `_report_worker_error` is the report.
|
|
_report_worker_error(
|
|
job_id,
|
|
exc,
|
|
"supervisor_error",
|
|
lambda: _fail_after_kill(
|
|
job_id,
|
|
asset_id,
|
|
proc,
|
|
grace_s,
|
|
"The download was stopped after a worker error.",
|
|
"unknown",
|
|
),
|
|
)
|
|
return
|
|
|
|
if reason is None:
|
|
# Child exited. Reconcile the asset we claimed: free it if the child left it non-finalized —
|
|
# a crash/OOM, or an early return on a job deleted mid-flight — a no-op when it finalized
|
|
# normally. Then, if it died non-zero while the job is still 'running', fail the job so it is
|
|
# not left stuck. Guarded: the child is already dead, so a DB blip here must not propagate.
|
|
try:
|
|
_reset_asset_pending(asset_id)
|
|
if proc.returncode != _CHILD_OK:
|
|
status, _ = _job_liveness(job_id)
|
|
if status == JobStatus.RUNNING:
|
|
_cleanup_staging(asset_id)
|
|
_fail_running(
|
|
job_id,
|
|
error=f"The download process exited unexpectedly (code {proc.returncode}).",
|
|
error_code="unknown",
|
|
cause="child_exit",
|
|
exit_code=proc.returncode,
|
|
)
|
|
except Exception: # boot recovery is the backstop for a stuck 'running' job
|
|
log.exception("post-exit reconciliation failed for job %s", job_id)
|
|
return
|
|
|
|
if reason == "stalled":
|
|
_fail_after_kill(
|
|
job_id,
|
|
asset_id,
|
|
proc,
|
|
grace_s,
|
|
f"The download stalled (no progress for {int(stall_s)}s) and was stopped.",
|
|
"network",
|
|
cause="stalled",
|
|
)
|
|
return
|
|
if reason == "timeout":
|
|
_fail_after_kill(
|
|
job_id,
|
|
asset_id,
|
|
proc,
|
|
grace_s,
|
|
f"The download exceeded the {int(wall_s)}s time limit and was stopped.",
|
|
"network",
|
|
cause="timeout",
|
|
)
|
|
return
|
|
# cancelled / abandon / shutdown: stop + free the asset, but leave the job's own DB status.
|
|
_stop_and_free(job_id, asset_id, proc, grace_s)
|
|
if reason == "shutdown":
|
|
_requeue_if_running(job_id, None) # resume on the next boot
|
|
elif reason == "abandon":
|
|
log.warning(
|
|
"job %s: killed a child wedged past the wall-clock; DB status left intact",
|
|
job_id,
|
|
extra=logs.event("download.child.abandoned", job_id=job_id),
|
|
)
|
|
else: # cancelled
|
|
# Not `download.job.aborted`: that is the child's line, logged when it notices the pause
|
|
# itself — and it often has, and is only tearing down when this poll comes round. This one
|
|
# says the worker had to stop it.
|
|
log.info(
|
|
"job %s: child stopped (route pause/cancel)",
|
|
job_id,
|
|
extra=logs.event("download.child.stopped", job_id=job_id),
|
|
)
|
|
|
|
|
|
def _cleanup_staging(asset_id: int) -> None:
|
|
"""Best-effort removal of a killed child's leftover staging dirs (its own `finally` never ran).
|
|
The next attempt on this asset also wipes staging on start, so this is just tidiness/disk."""
|
|
for path in (_staging_dir(asset_id), _edit_staging_dir(asset_id)):
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
|
|
|
|
# --- processing -----------------------------------------------------------------------------
|
|
|
|
|
|
def _process_job(job_id: int) -> None:
|
|
"""Dispatch a claimed job. The trivial outcomes (cache hit, dead/contended asset) are handled
|
|
inline and cheaply; the actual heavy work (yt-dlp download / ffmpeg edit) is claimed here and
|
|
then run in a supervised CHILD PROCESS (`_run_supervised`) so a wedged yt-dlp/ffmpeg can be
|
|
hard-killed — a worker THREAD cannot be killed, nor (when it spins in a C call) even
|
|
cooperatively cancelled, and one such thread starves the others via the GIL."""
|
|
diagnosis = None # set exactly when the asset has already failed (asset_status == "error")
|
|
with SessionLocal() as db:
|
|
job = db.get(DownloadJob, job_id)
|
|
if job is None:
|
|
return
|
|
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
|
|
asset_id = asset.id if asset else None
|
|
asset_status = asset.status if asset else None
|
|
asset_error = asset.error if asset else None
|
|
if asset is not None and asset_status == "error":
|
|
# A job that meets an already-failed asset gets that failure's explanation — not a bare
|
|
# "Download failed". Only the text is on the asset, so it is diagnosed again, for this
|
|
# link, with whether a cookie goes with it asked of the same gate the download child
|
|
# used (`credentials.usable_cookie`) — as it stands now, so the advice fits this job's Retry.
|
|
url, cookie = _failed_link_and_cookie(db, job, asset)
|
|
diagnosis = dlerrors.diagnose(asset_error, url=url, cookie_attached=cookie is not None)
|
|
del cookie # only whether there is one matters here
|
|
|
|
if asset is None:
|
|
_fail_running(job_id, error="No media asset", error_code=None, cause="no_asset")
|
|
return
|
|
|
|
if asset_status == "ready":
|
|
_finish_from_cache(job_id, asset_id)
|
|
return
|
|
if diagnosis is not None:
|
|
_fail_running(
|
|
job_id,
|
|
error=asset_error or "Download failed",
|
|
error_code=diagnosis.code,
|
|
error_detail=diagnosis.detail,
|
|
cause="asset_failed",
|
|
)
|
|
return
|
|
if asset_status == "downloading":
|
|
# Another worker won this asset between our claim and here (a narrow race — the claim SQL
|
|
# excludes downloading-asset jobs). Requeue and let it be reclaimed once the asset settles;
|
|
# no sleep — the claim exclusion keeps it out of the window until then, so no busy-loop.
|
|
_requeue_if_running(job_id, "waiting")
|
|
return
|
|
|
|
# asset_status == "pending": try to win the download / edit.
|
|
if not _claim_asset(asset_id):
|
|
# Lost the pending->downloading race; requeue (same reasoning — the exclusion prevents a spin).
|
|
_requeue_if_running(job_id, "waiting")
|
|
return
|
|
|
|
# The asset is now ours ('downloading'): run the heavy work in a supervised child process.
|
|
_run_supervised(job_id, asset_id)
|
|
|
|
|
|
def _failed_link_and_cookie(
|
|
db: Session, job: DownloadJob, asset: MediaAsset
|
|
) -> tuple[str | None, str | None]:
|
|
"""The link a job's failed asset was fetched from, and the cookie that fetch would go with now.
|
|
A clip's link is its source's (a `clip` ref names the user's range, no page), asked of the
|
|
clip's own cookie gate, as its fetch was (`_clip_source`); a Plex clip fetches nothing."""
|
|
if job.job_kind == JobKind.CLIP:
|
|
spec = dict(job.edit_spec or {})
|
|
if spec.get("source") != "youtube":
|
|
return None, None
|
|
url = service.source_url("youtube", spec["ref"])
|
|
return url, credentials.usable_clip_cookie(db, job.user_id, url, asset, spec)
|
|
url = service.source_url(job.source_kind, job.source_ref)
|
|
spec = formats.normalize(job.profile_snapshot)
|
|
return url, credentials.usable_cookie(db, job.user_id, url, asset, spec)
|
|
|
|
|
|
def _finish_from_cache(job_id: int, asset_id: int) -> None:
|
|
with SessionLocal() as db:
|
|
job = db.get(DownloadJob, job_id)
|
|
asset = db.get(MediaAsset, asset_id)
|
|
if job and asset:
|
|
fields = {"status": JobStatus.DONE, "progress": 100, "phase": None}
|
|
# A clip queued under the placeholder takes the title its sibling's fetch found.
|
|
unnamed = not job.display_name or (
|
|
job.job_kind == JobKind.CLIP and job.display_name == service.clip_title(None, None)
|
|
)
|
|
if unnamed and asset.title:
|
|
# display_name is VARCHAR(255); a title can be longer (e.g. a recipe in the title).
|
|
fields["display_name"] = asset.title[:DISPLAY_TEXT_MAX]
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
# Atomic finalize: only marks the job done while it is still running, so a concurrent
|
|
# cancel/pause is never resurrected (see service.finalize_if_running).
|
|
service.finalize_if_running(db, job_id, **fields)
|
|
db.commit()
|
|
|
|
|
|
# --- loop -----------------------------------------------------------------------------------
|
|
|
|
|
|
def _worker_loop(idx: int) -> None:
|
|
while not _stop.is_set() and idx < _target:
|
|
try:
|
|
job_id = _claim_job()
|
|
except Exception:
|
|
log.exception("claim failed")
|
|
_stop.wait(settings.download_poll_seconds)
|
|
continue
|
|
if job_id is None:
|
|
_stop.wait(settings.download_poll_seconds)
|
|
continue
|
|
# Tag every line this job produces (including the crash below) with the job id: a worker runs
|
|
# N jobs at once on N threads, so without it the interleaved output is unreadable. The bind
|
|
# wraps the handler too — bound inside the `try`, it was already reset by the time the crash
|
|
# line was logged.
|
|
with logs.bind(job_id=job_id):
|
|
try:
|
|
_process_job(job_id)
|
|
except Exception as exc: # noqa: BLE001
|
|
try:
|
|
_report_worker_error(
|
|
job_id,
|
|
exc,
|
|
"crash",
|
|
partial(_fail_running, job_id, error="Worker error", error_code=None),
|
|
)
|
|
# silent-ok: not silent — `_report_worker_error` logged the crash before this
|
|
# re-raise. Failing the job hit the database, and letting that out would end this
|
|
# worker thread for good; boot recovery settles a job left `running`.
|
|
except Exception:
|
|
log.exception(
|
|
"could not fail job %s after its crash",
|
|
job_id,
|
|
extra=logs.event("download.worker.fail_write_failed", job_id=job_id),
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
logs.configure_logging("worker")
|
|
signal.signal(signal.SIGTERM, _handle_signal)
|
|
signal.signal(signal.SIGINT, _handle_signal)
|
|
|
|
if not settings.worker_enabled:
|
|
log.info("WORKER_ENABLED is off; worker exiting (nothing to do).")
|
|
return
|
|
|
|
# The API may still be applying migrations when we boot. If they never arrive, exit instead of
|
|
# pressing on into a guaranteed crash — with a FAILURE status, so the container comes back under
|
|
# any restart policy (`on-failure` too, not just the `unless-stopped` our composes ship) and the
|
|
# exit code doesn't claim success for a worker that never started working.
|
|
if not _wait_for_schema():
|
|
if _stop.is_set():
|
|
return # asked to shut down while waiting — that IS a clean exit
|
|
raise SystemExit(1)
|
|
# After the wait, not before: the DB is reachable now. The wait checks `download_jobs`, not
|
|
# `log_events`, so on the deploy that first adds the table the worker may start before the API
|
|
# has migrated — those few seconds of batches fail and say so on stderr; nothing else breaks.
|
|
logsink.start()
|
|
_recover_orphans()
|
|
global _target
|
|
_target = max(1, settings.download_worker_concurrency) # used only if the DB can't answer
|
|
threads: dict[int, threading.Thread] = {}
|
|
level = _sync_knobs(threads, None)
|
|
log.info("Download worker started (concurrency=%d, root=%s).", _target, settings.download_root)
|
|
while not _stop.wait(_KNOB_SECONDS):
|
|
level = _sync_knobs(threads, level)
|
|
log.info("Download worker stopping…")
|
|
for t in threads.values():
|
|
t.join(timeout=5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|