fix(tasks): keep failure reports and queue choice on the task spec

report_task_failure can run in the worker parent when a work horse is
killed, so it now enqueues notify_task_failure on the default queue
instead of loading the notification stack there.

enqueue_task defaults to the queue its spec names. The streaming worker
runs as PreloadingWorker, which imports the enabled scheduled tasks of
its queues, so the reaper's per-minute horse inherits its stack. Spec
tests now read config in a fresh interpreter instead of comparing a
value with itself.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

DEVELOPER_SETUP.md

@@ -144,7 +144,7 @@ uv run rq worker -c config --worker-class handler.rq_worker.RomMWorker --with-sc
 
 ```sh
 cd backend
-uv run rq worker -c config --worker-class handler.rq_worker.RomMWorker --with-scheduler streaming
+uv run rq worker -c config --worker-class handler.rq_worker.PreloadingWorker --with-scheduler streaming
 ```
 
 `--with-scheduler` releases delayed jobs, such as the rescans the filesystem watcher waits out, so each worker needs it. The recurring schedule is registered by the RQ cron process, which the workers then execute:

backend/handler/redis_handler.py

@@ -46,6 +46,7 @@ ALL_QUEUES: Final = (
     default_queue,
     low_prio_queue,
 )
+QUEUES_BY_NAME: Final = {queue.name: queue for queue in ALL_QUEUES}
 
 
 def __get_fake_server() -> Any:

backend/handler/rq_worker.py

@@ -53,3 +53,20 @@ class RomMWorker(Worker):
         with contextlib.suppress(Exception):
             job.execute_failure_callback(self.death_penalty_class, *exc_info)
         self.handle_exception(job, *exc_info)
+
+
+class PreloadingWorker(RomMWorker):
+    """RomMWorker that imports the enabled scheduled tasks of its queues up front.
+
+    For a dedicated queue whose tasks run often enough that every work horse
+    importing its task again costs more than holding it in the parent.
+    """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        from tasks.registry import SCHEDULED_TASKS, get_task
+
+        queues = set(self.queue_names())
+        for name, spec in SCHEDULED_TASKS.items():
+            if spec.enabled and spec.queue_name in queues:
+                get_task(name)

backend/tasks/registry.py

@@ -38,7 +38,7 @@ from config import (
     TASK_TIMEOUT,
 )
 from exceptions.task_exceptions import TaskNotFoundException
-from handler.redis_handler import STREAMING_QUEUE_NAME, low_prio_queue, scan_queue
+from handler.redis_handler import QUEUES_BY_NAME, STREAMING_QUEUE_NAME, scan_queue
 from handler.streaming.config import HOLD_CEILING_SECONDS, streaming_enabled
 from tasks.tasks import Task, TaskSpec, TaskType, run_task_by_name
 
@@ -277,7 +277,7 @@ def get_task(name: str) -> Task | None:
 def enqueue_task(
     name: str,
     *,
-    queue: Queue = low_prio_queue,
+    queue: Queue | None = None,
     task_kwargs: dict[str, Any] | None = None,
     run_by_user_id: int | None = None,
     **job_options: Any,
@@ -286,7 +286,7 @@ def enqueue_task(
 
     Args:
         name: The key the task is registered under.
-        queue: Which queue to enqueue on.
+        queue: Which queue to enqueue on, the one the spec names by default.
         task_kwargs: Forwarded to the task's ``run``, nested so that they cannot
             collide with the name of the task to run.
         run_by_user_id: Who ran it by hand, notified when it ends.
@@ -299,6 +299,7 @@ def enqueue_task(
     if spec is None:
         raise TaskNotFoundException(name)
 
+    queue = queue or QUEUES_BY_NAME[spec.queue_name]
     return queue.enqueue(
         run_task_by_name,
         kwargs={

backend/tasks/tasks.py

@@ -1,4 +1,3 @@
-import asyncio
 from abc import ABC, abstractmethod
 from collections.abc import Mapping
 from dataclasses import dataclass
@@ -12,7 +11,7 @@ from rq.timeouts import JobTimeoutException
 
 from config import TASK_RESULT_TTL, TASK_TIMEOUT
 from exceptions.task_exceptions import TaskNotFoundException
-from handler.redis_handler import QueuePrio
+from handler.redis_handler import QueuePrio, default_queue
 from logger.logger import log
 from utils.background_tasks import wait_for_background_tasks
 
@@ -84,12 +83,27 @@ def report_task_failure(
         reason = str(exc_value) or repr(exc_value)
 
     try:
-        asyncio.run(_notify_task_end(name, spec, run_by_user_id, error=reason))
+        # A job of its own, as this may run in the worker parent, which must not
+        # load the notification stack.
+        default_queue.enqueue(
+            notify_task_failure, name, run_by_user_id, reason, result_ttl=0
+        )
     except Exception:  # noqa: BLE001
         # Raising would stop the worker's sweep of the other orphaned jobs.
         log.error(f"Could not report failed task {job.id}", exc_info=True)
 
 
+async def notify_task_failure(
+    name: str, run_by_user_id: int | None, reason: str
+) -> None:
+    """Tell whoever ran a task, or the admins, why it failed."""
+    from tasks.registry import get_task_spec
+
+    spec = get_task_spec(name)
+    if spec is not None:
+        await _notify_task_end(name, spec, run_by_user_id, error=reason)
+
+
 async def _notify_task_end(
     name: str, spec: "TaskSpec", run_by_user_id: int | None, error: str | None = None
 ) -> None:

backend/tests/handler/test_rq_worker.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Iterator
+from dataclasses import replace
 from unittest.mock import MagicMock
 
 import pytest
@@ -7,7 +8,8 @@ import rq.scheduler
 from fakeredis import FakeRedis
 from rq.exceptions import AbandonedJobError
 
-from handler.rq_worker import RomMWorker, _DropPeriodicNoiseFilter
+from handler.rq_worker import PreloadingWorker, RomMWorker, _DropPeriodicNoiseFilter
+from tasks.registry import REAP_STREAMING_SESSIONS_SPEC
 from tasks.tasks import report_task_failure
 
 scheduler_log = logging.getLogger(rq.scheduler.__name__)
@@ -96,3 +98,32 @@ def test_a_failing_callback_still_reaches_the_handlers(mocker) -> None:
     worker.handle_work_horse_killed(job, 123, 9, None)
 
     handle_exception.assert_called_once()
+
+
+class TestPreloadingWorker:
+    @pytest.fixture
+    def get_task(self, mocker):
+        def spec(queue_name: str, enabled: bool):
+            return replace(
+                REAP_STREAMING_SESSIONS_SPEC, queue_name=queue_name, enabled=enabled
+            )
+
+        mocker.patch(
+            "tasks.registry.SCHEDULED_TASKS",
+            {
+                "on": spec("streaming", True),
+                "off": spec("streaming", False),
+                "elsewhere": spec("low", True),
+            },
+        )
+        return mocker.patch("tasks.registry.get_task")
+
+    def test_imports_the_enabled_tasks_of_its_queues(self, get_task) -> None:
+        PreloadingWorker(["streaming"], connection=FakeRedis(version=7))
+
+        get_task.assert_called_once_with("on")
+
+    def test_imports_nothing_for_a_queue_without_tasks(self, get_task) -> None:
+        PreloadingWorker(["scans"], connection=FakeRedis(version=7))
+
+        get_task.assert_not_called()

backend/tests/tasks/test_cleanup_audit_log.py

@@ -1,7 +1,6 @@
 from dataclasses import replace
 from datetime import datetime, timedelta, timezone
 
-from config import AUDIT_LOG_RETENTION_DAYS
 from handler.database import db_audit_event_handler
 from tasks.registry import CLEANUP_AUDIT_LOG_SPEC
 from tasks.scheduled import cleanup_audit_log
@@ -9,8 +8,7 @@ from tasks.scheduled.cleanup_audit_log import CleanupAuditLogTask
 
 
 class TestCleanupAuditLogTask:
-    def test_runs_daily_unless_events_are_kept_forever(self):
-        assert CLEANUP_AUDIT_LOG_SPEC.enabled is (AUDIT_LOG_RETENTION_DAYS > 0)
+    def test_runs_daily(self):
         assert CLEANUP_AUDIT_LOG_SPEC.cron_string == "30 4 * * *"
 
     async def test_deletes_batches_until_one_comes_back_short(self, mocker):

backend/tests/tasks/test_registry.py

@@ -1,3 +1,8 @@
+import os
+import subprocess
+import sys
+from pathlib import Path
+
 import pytest
 from rq.utils import import_attribute
 
@@ -77,6 +82,18 @@ class TestEnqueueTask:
         assert kwargs["job_id"] == "fixed"
         assert kwargs["unique"] is True
 
+    @pytest.mark.parametrize(
+        "name,queue_name",
+        [("cleanup_zip_cache", "low"), ("reap_streaming_sessions", "streaming")],
+    )
+    def test_defaults_to_the_queue_the_spec_names(self, mocker, name, queue_name):
+        queue = mocker.MagicMock()
+        mocker.patch.dict("tasks.registry.QUEUES_BY_NAME", {queue_name: queue})
+
+        enqueue_task(name)
+
+        queue.enqueue.assert_called_once()
+
     def test_an_unknown_name_is_refused_before_it_reaches_redis(self, queue):
         with pytest.raises(TaskNotFoundException):
             enqueue_task("no_such_task", queue=queue)
@@ -108,3 +125,59 @@ class TestEnqueueScheduledScan:
 
         job_timeout = scan_queue.enqueue.call_args.kwargs["job_timeout"]
         assert job_timeout == SCHEDULED_TASKS["scan_library"].timeout
+
+
+BACKEND_ROOT = Path(__file__).resolve().parents[2]
+
+
+@pytest.mark.parametrize(
+    "env,name,field,expected",
+    [
+        ({"AUDIT_LOG_RETENTION_DAYS": "0"}, "cleanup_audit_log", "enabled", False),
+        ({"AUDIT_LOG_RETENTION_DAYS": "30"}, "cleanup_audit_log", "enabled", True),
+        (
+            {"SCHEDULED_RESCAN_CRON": "5 4 * * 1"},
+            "scan_library",
+            "cron_string",
+            "5 4 * * 1",
+        ),
+        (
+            {
+                "LAUNCHBOX_API_ENABLED": "true",
+                "ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA": "false",
+            },
+            "update_launchbox_metadata",
+            "can_run_manually",
+            True,
+        ),
+        (
+            {
+                "LAUNCHBOX_API_ENABLED": "false",
+                "ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA": "false",
+            },
+            "update_launchbox_metadata",
+            "can_run_manually",
+            False,
+        ),
+    ],
+)
+def test_specs_follow_the_config_they_start_with(env, name, field, expected):
+    """Specs read config once at import, so each case needs a fresh interpreter."""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            "import sys\n"
+            "from tasks.registry import get_task_spec\n"
+            "print(repr(getattr(get_task_spec(sys.argv[1]), sys.argv[2])))\n",
+            name,
+            field,
+        ],
+        cwd=BACKEND_ROOT,
+        env={**os.environ, "PYTHONPATH": str(BACKEND_ROOT), **env},
+        capture_output=True,
+        text=True,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.strip().splitlines()[-1] == repr(expected)

backend/tests/tasks/test_task_func_paths.py

@@ -11,6 +11,7 @@ BACKEND_ROOT = Path(__file__).resolve().parents[2]
 # run_task_by_name, which resolves it from the registry.
 JOB_FUNC_PATHS = (
     "tasks.tasks.run_task_by_name",
+    "tasks.tasks.notify_task_failure",
     "endpoints.sockets.scan.scan_platforms",
     "tasks.registry.enqueue_scheduled_scan",
 )

backend/tests/tasks/test_tasks.py

@@ -16,6 +16,7 @@ from tasks.tasks import (
     RemoteFilePullTask,
     TaskSpec,
     TaskType,
+    notify_task_failure,
     report_task_failure,
     run_task_by_name,
 )
@@ -319,6 +320,24 @@ def _job(name="cleanup_missing_roms", run_by_user_id=None, **overrides):
 class TestReportTaskFailure:
     """The worker's exception handler reports every way a task can fail."""
 
+    @pytest.fixture(autouse=True)
+    def queue(self, mocker):
+        """Runs the enqueued notification inline, as a work horse would."""
+        queue = mocker.patch("tasks.tasks.default_queue")
+        queue.enqueue.side_effect = lambda func, *args, **_: asyncio.run(func(*args))
+        return queue
+
+    def test_notifies_from_a_job_of_its_own(self, mocker, queue):
+        _task(mocker)
+
+        report_task_failure(
+            _job(run_by_user_id=4), RuntimeError, RuntimeError("disk full"), None
+        )
+
+        queue.enqueue.assert_called_once_with(
+            notify_task_failure, "cleanup_missing_roms", 4, "disk full", result_ttl=0
+        )
+
     def test_tells_the_runner_why_it_failed(self, mocker, notify, notify_admins):
         _task(mocker)
 
@@ -390,12 +409,9 @@ class TestReportTaskFailure:
         get_task_spec.assert_not_called()
         notify.assert_not_awaited()
 
-    def test_never_raises_into_the_worker(self, mocker):
+    def test_never_raises_into_the_worker(self, mocker, queue):
         _task(mocker)
-        mocker.patch(
-            "handler.notification_handler.notify_admins",
-            AsyncMock(side_effect=RuntimeError("redis gone")),
-        )
+        queue.enqueue.side_effect = RuntimeError("redis gone")
 
         report_task_failure(_job(), RuntimeError, RuntimeError("boom"), None)
 

docker/init_scripts/init

@@ -263,19 +263,20 @@ start_bin_rq_cron() {
 # Commands to start RQ worker
 start_bin_rq_worker() {
 	info_log "Starting RQ worker"
-	start_rq_worker rq_worker high default low
+	start_rq_worker rq_worker RomMWorker high default low
 }
 
 # Scans get a worker of their own, see SCAN_QUEUE_NAME.
 start_bin_rq_scan_worker() {
 	info_log "Starting RQ scan worker"
-	start_rq_worker rq_scan_worker scans
+	start_rq_worker rq_scan_worker RomMWorker scans
 }
 
-# Streaming teardowns get a worker of their own, see STREAMING_QUEUE_NAME.
+# Streaming teardowns get a worker of their own, see STREAMING_QUEUE_NAME. It
+# preloads the reaper, which runs every minute when streaming is on.
 start_bin_rq_streaming_worker() {
 	info_log "Starting RQ streaming worker"
-	start_rq_worker rq_streaming_worker streaming
+	start_rq_worker rq_streaming_worker PreloadingWorker streaming
 }
 
 # Set PYTHONPATH so RQ can find the tasks module.
@@ -286,7 +287,8 @@ start_bin_rq_streaming_worker() {
 # wait out their delay.
 start_rq_worker() {
 	local name="$1"
-	shift
+	local worker_class="$2"
+	shift 2
 
 	local redis_url
 	redis_url="$(build_redis_url)"
@@ -295,7 +297,7 @@ start_rq_worker() {
 		RQ_REDIS_URL="${redis_url}" \
 		rq worker \
 		--path /backend \
-		--worker-class handler.rq_worker.RomMWorker \
+		--worker-class "handler.rq_worker.${worker_class}" \
 		--results-ttl "${TASK_RESULT_TTL:-86400}" \
 		--logging_level "${LOGLEVEL}" \
 		--with-scheduler \

docs/BACKEND_ARCHITECTURE.md

@@ -1441,7 +1441,7 @@ await notify_admins("custom", NotificationLevel.WARNING, title="Disk almost full
 
 A `NotificationKind` is translated by the client from `data`; a new one needs a describer in `frontend/src/v2/utils/notifications.ts` and locale keys. Until then, or for a one-off, pass any other kind with `title`/`body`/`link`/`icon`. Both helpers log and swallow failures, so a job never fails over reporting itself.
 
-A task reports its success from `run_task_by_name`. Its failure is reported by `report_task_failure`, an exception handler `RomMWorker` installs, so a timeout, a killed work horse or a dead worker notifies too, for cron runs as well as manual ones.
+A task reports its success from `run_task_by_name`. Its failure is reported by `report_task_failure`, an exception handler `RomMWorker` installs, so a timeout, a killed work horse or a dead worker notifies too, for cron runs as well as manual ones. The handler can run in the worker parent, so it enqueues `notify_task_failure` on the default queue rather than loading the notification stack there.
 
 ---
 

entrypoint.sh

@@ -96,13 +96,14 @@ PYTHONPATH="/app/backend:${PYTHONPATH-}" \
 # wait out their delay.
 start_rq_worker() {
 	local name="$1"
-	shift
+	local worker_class="$2"
+	shift 2
 
 	PYTHONPATH="/app/backend:${PYTHONPATH-}" \
 		RQ_REDIS_URL="${REDIS_URL}" \
 		rq worker \
 		--path /app/backend \
-		--worker-class handler.rq_worker.RomMWorker \
+		--worker-class "handler.rq_worker.${worker_class}" \
 		--pid "/tmp/${name}.pid" \
 		--logging_level "${LOGLEVEL:-INFO}" \
 		--with-scheduler \
@@ -110,15 +111,16 @@ start_rq_worker() {
 }
 
 echo "Starting RQ worker..."
-start_rq_worker rq_worker high default low
+start_rq_worker rq_worker RomMWorker high default low
 
 # Scans get a worker of their own, see SCAN_QUEUE_NAME.
 echo "Starting RQ scan worker..."
-start_rq_worker rq_scan_worker scans
+start_rq_worker rq_scan_worker RomMWorker scans
 
-# Streaming teardowns get a worker of their own, see STREAMING_QUEUE_NAME.
+# Streaming teardowns get a worker of their own, see STREAMING_QUEUE_NAME. It
+# preloads the reaper, which runs every minute when streaming is on.
 echo "Starting RQ streaming worker..."
-start_rq_worker rq_streaming_worker streaming
+start_rq_worker rq_streaming_worker PreloadingWorker streaming
 
 echo "Starting watcher..."
 watchfiles \