diff --git a/backend/app/worker.py b/backend/app/worker.py index e2006bc9..4536cb04 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -5,7 +5,8 @@ from the API so long yt-dlp downloads + ffmpeg postprocessing never block web re Design: * N loop threads (download_worker_concurrency) each claim a `queued` DownloadJob with - FOR UPDATE SKIP LOCKED and process it. + 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. @@ -48,12 +49,60 @@ 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 ----------------------------------------------------------------------- @@ -576,7 +625,7 @@ def _finish_from_cache(job_id: int, asset_id: int) -> None: def _worker_loop(idx: int) -> None: - while not _stop.is_set(): + while not _stop.is_set() and idx < _target: try: job_id = _claim_job() except Exception: @@ -634,15 +683,15 @@ def main() -> None: # 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) + 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: + for t in threads.values(): t.join(timeout=5) diff --git a/backend/tests/test_worker_knobs.py b/backend/tests/test_worker_knobs.py new file mode 100644 index 00000000..b423510d --- /dev/null +++ b/backend/tests/test_worker_knobs.py @@ -0,0 +1,109 @@ +"""The worker obeys the two admin knobs that concern it — `download_worker_concurrency` and +`log_level` — as stored in the settings page, not only the env it booted with. It used to read +`settings.download_worker_concurrency` and `settings.log_level` once at start, so the admin page +showed values the worker never applied. No DB or real threads of work here: the stored values come +from a patched `sysconfig`, and the loop threads are stand-ins that only record their index.""" + +import threading + +import pytest + +from app import worker + + +class _NoSession: + def __enter__(self): + return None + + def __exit__(self, *exc): + return False + + +@pytest.fixture +def stored(monkeypatch): + """The settings page's values, served to the worker instead of a database.""" + values = {"download_worker_concurrency": 3, "log_level": "DEBUG"} + monkeypatch.setattr(worker, "SessionLocal", _NoSession) + monkeypatch.setattr(worker.sysconfig, "get_int", lambda _db, key: int(values[key])) + monkeypatch.setattr(worker.sysconfig, "get_str", lambda _db, key: str(values[key])) + return values + + +@pytest.fixture +def levels(monkeypatch): + applied: list[str] = [] + monkeypatch.setattr(worker.logs, "set_level", applied.append) + return applied + + +@pytest.fixture +def loops(monkeypatch): + """Stand-in loop threads: each records its index and parks until the worker stops.""" + started: list[int] = [] + lock = threading.Lock() + + def fake_loop(idx: int) -> None: + with lock: + started.append(idx) + worker._stop.wait() + + monkeypatch.setattr(worker, "_worker_loop", fake_loop) + yield started + worker._stop.set() + + +@pytest.fixture(autouse=True) +def _clean_stop(): + worker._stop.clear() + yield + worker._stop.clear() + + +def test_worker_boots_with_the_stored_concurrency_and_log_level(monkeypatch, stored, levels, loops): + monkeypatch.setattr(worker.settings, "download_worker_concurrency", 2) + monkeypatch.setattr(worker.settings, "worker_enabled", True) + monkeypatch.setattr(worker.logs, "configure_logging", lambda _source: None) + monkeypatch.setattr(worker.signal, "signal", lambda *_a: None) + monkeypatch.setattr(worker, "_wait_for_schema", lambda: True) + monkeypatch.setattr(worker.logsink, "start", lambda: None) + monkeypatch.setattr(worker, "_recover_orphans", lambda: None) + + threading.Timer(0.5, worker._stop.set).start() + worker.main() + + assert sorted(loops) == [0, 1, 2] # the stored 3, not the env's 2 + assert levels and levels[-1] == "DEBUG" + + +def test_a_changed_knob_applies_without_a_restart(stored, levels, loops): + threads: dict[int, threading.Thread] = {} + level = worker._sync_knobs(threads, None) + assert level == "DEBUG" + assert worker._target == 3 + + stored["download_worker_concurrency"] = 5 + stored["log_level"] = "WARNING" + level = worker._sync_knobs(threads, level) + + assert worker._target == 5 + assert sorted(threads) == [0, 1, 2, 3, 4] + assert level == "WARNING" and levels[-1] == "WARNING" + + # Lowering it retires the surplus threads after their current job: they see the lower target. + stored["download_worker_concurrency"] = 1 + worker._sync_knobs(threads, level) + assert worker._target == 1 + + +def test_an_unreadable_database_keeps_what_runs(monkeypatch, levels, loops): + def boom(): + raise RuntimeError("db down") + + monkeypatch.setattr(worker, "SessionLocal", boom) + monkeypatch.setattr(worker, "_target", 2) + threads: dict[int, threading.Thread] = {} + + assert worker._sync_knobs(threads, "INFO") == "INFO" + assert worker._target == 2 + assert sorted(threads) == [0, 1] # the loops still run on the boot value + assert levels == []