fix(worker): obey the stored concurrency and log level, live

The admin page's download_worker_concurrency and log_level were stored but never reached the worker: it read the env once at boot (worker.py) and never looked again. The worker now re-reads both every 10 s (_sync_knobs): extra loop threads start at once, surplus ones retire after their current job, and a changed level is applied to the process. An unreadable database keeps what already runs.
2026-10-04 17:53:19 +02:00
parent 2a699e26ba
commit 33dec56e45
2 changed files with 168 additions and 10 deletions
@@ -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)


@@ -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 == []