peter
54d19cc386
- yt-dlp's text is scrubbed wherever it is logged, not only in the adapter:
`formats.log_safe` takes out the URL, the `[extractor] <id>:` prefix and
file paths, and the job-failed, refused, edit-failed and backfill lines use
it. Done lines log the asset id, not the title-built path. The PII contract
now bans `rel`/`rel_path`/`title`.
- A traceback goes with a download/edit failure only when it is our own code
failing; for yt-dlp's DownloadError, our RuntimeErrors and OSErrors its foot
is the unscrubbed message.
- yt-dlp's `error` maps to WARNING: it fires on attempts the fallback client
set recovers from; the job's verdict is `download.job.failed`.
- The sink groups a bare `%s` pass-through by its text (numbers folded), and
a process's last batch closes the repeat window, so a short-lived download
child still reports what it dropped.
- Worker-side failures (stall, wall-clock, no asset, cached asset error) log
`download.job.failed` with a cause; they left no line at all before. The
crash, supervisor-error and non-zero-exit lines carry the code too.
- An edit failure is one coded line in `_fail_edit`, for every path.
- `logs.get_logger("__main__")` resolves to the `-m` module, so the worker
and the download child keep their names.
- TRACE lines carrying user content are gated on `logs.consented_trace()`,
not on the process level; a contract test pins it.
- The OAuth token refresh goes through `_upstream_failure`; transport errors
carry reason `transport`. Subscribe recognises a duplicate by reason code
only (the prose fallback was dead).
- Once-only guards for a stale sysconfig override and a refusing HLS session
directory; item_detail splits Plex failures from our own parse bugs.
- The playlist breaker line names the playlist and the action; comments that
were false (runner noqa, `_merge_vtt`, sink grouping) are corrected.
607 lines
27 KiB
Python
607 lines
27 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.
|
|
* 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 can't reach the API's in-process WebSocket registry.
|
|
* 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 datetime import datetime, timezone
|
|
|
|
from sqlalchemy import text
|
|
|
|
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, JobStatus
|
|
from app.models import DownloadJob, MediaAsset
|
|
|
|
log = logs.get_logger(__name__)
|
|
_stop = threading.Event()
|
|
|
|
|
|
def _handle_signal(signum, frame):
|
|
_stop.set()
|
|
|
|
|
|
# --- 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: # noqa: BLE001 — 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."""
|
|
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 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,
|
|
) -> None:
|
|
"""`_stop_and_free` + mark the job errored — for the watchdog's stall/timeout kills and the
|
|
supervisor-error fallback."""
|
|
_stop_and_free(job_id, asset_id, proc, grace_s)
|
|
_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,
|
|
) -> None:
|
|
"""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 or never ran — a stall, the wall-clock cap. Without it those left
|
|
no line at all. A caller that already logs its own coded line (a crash, with its stack) passes
|
|
none. Logged only when the job actually failed here: a route's cancel that landed first is not
|
|
a failure."""
|
|
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
|
|
),
|
|
)
|
|
|
|
|
|
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: # noqa: BLE001 — 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 == "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 == "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: # noqa: BLE001 — a watchdog error must NEVER leave the child running unsupervised
|
|
log.exception(
|
|
"supervisor loop failed for job %s; stopping the child",
|
|
job_id,
|
|
extra=logs.event("download.job.failed", job_id=job_id, cause="supervisor_error"),
|
|
)
|
|
_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 == "running":
|
|
_cleanup_staging(asset_id)
|
|
_fail_running(
|
|
job_id,
|
|
error=f"The download process exited unexpectedly (code {proc.returncode}).",
|
|
error_code="unknown",
|
|
)
|
|
log.warning(
|
|
"job %s: child exited %s without finishing",
|
|
job_id,
|
|
proc.returncode,
|
|
extra=logs.event(
|
|
"download.job.failed",
|
|
job_id=job_id,
|
|
cause="child_exit",
|
|
exit_code=proc.returncode,
|
|
),
|
|
)
|
|
except Exception: # noqa: BLE001 — 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
|
|
log.info(
|
|
"job %s: child stopped (route pause/cancel)",
|
|
job_id,
|
|
extra=logs.event("download.job.aborted", 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."""
|
|
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_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 = service.source_url(job.source_kind, job.source_ref)
|
|
cookie = credentials.usable_cookie(
|
|
db, job.user_id, url, asset, formats.normalize(job.profile_snapshot)
|
|
)
|
|
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 asset_status == "error":
|
|
_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 _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}
|
|
if not job.display_name 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[:255]
|
|
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():
|
|
try:
|
|
job_id = _claim_job()
|
|
except Exception: # noqa: BLE001
|
|
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: # noqa: BLE001
|
|
log.exception(
|
|
"processing job %s crashed",
|
|
job_id,
|
|
extra=logs.event("download.job.failed", job_id=job_id, cause="crash"),
|
|
)
|
|
_fail_running(job_id, error="Worker error", error_code=None)
|
|
|
|
|
|
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()
|
|
n = max(1, settings.download_worker_concurrency)
|
|
log.info("Download worker started (concurrency=%d, root=%s).", n, settings.download_root)
|
|
threads = [threading.Thread(target=_worker_loop, args=(i,), daemon=True) for i in range(n)]
|
|
for t in threads:
|
|
t.start()
|
|
while not _stop.is_set():
|
|
_stop.wait(1.0)
|
|
log.info("Download worker stopping…")
|
|
for t in threads:
|
|
t.join(timeout=5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|