From 9e9f59c042137a8ea5513de4021994a9f8f54923 Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Sun, 23 Aug 2026 14:13:26 -0500 Subject: [PATCH 1/8] fix(scan): stop refusing scans a stalled scheduler left pending Every scan request has been refused with "A scan is already in progress" on instances whose rq-scheduler cannot make progress, and waiting does not help because nothing is running to wait for. The scheduler reads a job's function name before it takes the job out of its registry, so one job left behind by an older version, holding an argument that no longer deserializes, crashes it on every poll and stays there for the next one to trip on. Nothing scheduled runs again, and the watcher's delayed rescans pile up behind it. `_get_queued_scan_jobs` counted those as scans waiting to start, so the guard added in 99e214a8b refused every manual scan for as long as the scheduler stayed broken. Clear jobs the scheduler cannot read at startup, before it polls them again, and split scan discovery so the guard consults only what sits on a worker queue. Delayed scans still block nothing: the scheduler is the only thing that releases them, so one that is down must not be able to refuse scans. A recovered scheduler would release its whole backlog at once and run the same library scan over and over, so delayed scans more than an hour past due go too. Stopping a scan now drops delayed scans out of the scheduler's registry rather than only cancelling the job, which left the id in the registry for the scheduler to queue anyway once the delay was up. A worker killed mid-scan points at a job that can already be gone, which raised NoSuchJobError out of the socket handler with no reply to the client, and out of GET /tasks/status as a 500. Reading a job's status has the same problem once its hash expires, so it goes through a wrapper. Finally, name the scan in the way when refusing, and say when it is stopping rather than running, so the message is actionable. Fixes #4186 Co-Authored-By: Claude Opus 5 --- backend/endpoints/sockets/scan.py | 136 ++++++++++++--- backend/endpoints/tasks.py | 8 +- backend/handler/redis_handler.py | 19 ++- backend/startup.py | 11 ++ backend/tasks/tasks.py | 23 +++ backend/tests/endpoints/sockets/test_scan.py | 166 ++++++++++++++++++- backend/tests/tasks/test_tasks.py | 42 ++++- 7 files changed, 372 insertions(+), 33 deletions(-) diff --git a/backend/endpoints/sockets/scan.py b/backend/endpoints/sockets/scan.py index 41d5534af6..7d208a32b0 100644 --- a/backend/endpoints/sockets/scan.py +++ b/backend/endpoints/sockets/scan.py @@ -2,12 +2,14 @@ import asyncio from dataclasses import dataclass +from datetime import datetime, timedelta, timezone from itertools import batched, chain from typing import Any, Final import pydash import socketio # type: ignore from rq import Worker, get_current_job +from rq.exceptions import InvalidJobOperation, NoSuchJobError from rq.job import Job, JobStatus from sqlalchemy.exc import IntegrityError @@ -52,6 +54,7 @@ from handler.metadata.ss_handler import log_scan_summary as log_ss_scan_summary from handler.redis_handler import ( get_job_func_name, + get_job_status, high_prio_queue, low_prio_queue, redis_client, @@ -80,6 +83,10 @@ STOP_SCAN_FLAG: Final = "scan:stop" +# A delayed watcher scan this far past due was left behind by a scheduler that +# stopped releasing them, and the change it reacted to has long since settled. +STALE_SCHEDULED_SCAN_AGE: Final = timedelta(hours=1) + def _scan_platforms_func_name() -> str: """Fully qualified name RQ records for a directly enqueued scan. @@ -108,7 +115,13 @@ def _get_running_scan_job() -> Job | None: """ func_names = _scan_job_func_names() for worker in Worker.all(connection=redis_client): - job = worker.get_current_job() + # A worker killed mid-scan keeps pointing at its job until its own + # registration expires, and the job can be gone by then. + try: + job = worker.get_current_job() + except NoSuchJobError: + continue + if job is not None and get_job_func_name(job) in func_names: return job @@ -116,33 +129,109 @@ def _get_running_scan_job() -> Job | None: def _get_queued_scan_jobs() -> list[Job]: - """Scans waiting to run, not yet picked up by a worker. + """Scans sitting on a worker queue, waiting to be picked up. - Socket scans sit in the high priority queue, while watcher scans are delayed - through the scheduler before landing in the low priority queue. + Socket scans go to the high priority queue and watcher scans to the low + priority one, once the scheduler releases them. """ func_names = _scan_job_func_names() jobs: dict[str, Job] = {} for job in chain(high_prio_queue.get_jobs(), low_prio_queue.get_jobs()): - if isinstance(job, Job) and get_job_func_name(job) in func_names: + if ( + isinstance(job, Job) + and get_job_func_name(job) in func_names + and get_job_status(job) == JobStatus.QUEUED + ): jobs[job.id] = job - # The scheduler registry also holds the standing cron entry for the - # scheduled rescan, which is a schedule rather than a pending scan, so only - # delayed scan_platforms jobs count as queued here. + return list(jobs.values()) + + +def _get_scheduled_scan_jobs() -> list[Job]: + """Scans waiting out a delay in the scheduler, which only the watcher sets. + + The scheduler is the only thing that releases them, so one that is down + leaves them there for good and they cannot stand in for a scan in flight. + """ + # The registry also holds the standing cron entry for the scheduled rescan, + # which is a schedule rather than a pending scan, so only delayed + # scan_platforms jobs count here. scan_platforms_func_name = _scan_platforms_func_name() + jobs: dict[str, Job] = {} + for job in tasks_scheduler.get_jobs(): if ( isinstance(job, Job) and get_job_func_name(job) == scan_platforms_func_name - and job.get_status() in (JobStatus.SCHEDULED, JobStatus.QUEUED) + and get_job_status(job) in (JobStatus.SCHEDULED, JobStatus.QUEUED) ): jobs[job.id] = job return list(jobs.values()) +def _cancel_scheduled_scan_job(job: Job) -> None: + """Drop a delayed scan from the scheduler. + + Cancelling the job on its own leaves the id in the scheduler's registry, and + the scheduler queues it anyway once the delay is up, which runs the scan that + was just stopped. + """ + tasks_scheduler.cancel(job) + try: + job.cancel() + except InvalidJobOperation: + # Already cancelled; the registry entry was the part that mattered. + pass + + +def drop_stale_scheduled_scans() -> int: + """Drop delayed watcher scans that are long past due. + + Releasing a backlog of them at once, which is what a stalled scheduler does + the moment it recovers, would run the same library scan over and over. + + Returns: + int: How many scans were dropped. + """ + scan_platforms_func_name = _scan_platforms_func_name() + cutoff = datetime.now(timezone.utc) - STALE_SCHEDULED_SCAN_AGE + dropped = 0 + + for job, scheduled_at in tasks_scheduler.get_jobs(with_times=True): + if ( + not isinstance(job, Job) + or get_job_func_name(job) != scan_platforms_func_name + # The scheduler records due times as naive UTC. + or scheduled_at is None + or scheduled_at.replace(tzinfo=timezone.utc) > cutoff + ): + continue + + _cancel_scheduled_scan_job(job) + dropped += 1 + log.warning(f"Dropped scan scheduled for {scheduled_at} UTC, too long past due") + + return dropped + + +def _scan_job_label(job: Job) -> str: + """How to refer to a scan job when reporting it to a client.""" + return str(job.meta.get("task_name") or "A scan") + + +def _scan_in_flight_message(running: Job | None, queued: list[Job]) -> str: + """Say which scan is in the way, so the client knows what to wait on.""" + if running is None: + return f"{_scan_job_label(queued[0])} is already queued" + + if get_job_status(running) in (JobStatus.CANCELED, JobStatus.STOPPED): + return f"{_scan_job_label(running)} is still stopping, try again in a moment" + + return f"{_scan_job_label(running)} is already running" + + @dataclass class ScanStats: total_platforms: int = 0 @@ -1331,14 +1420,18 @@ async def scan_handler(sid: str, options: dict[str, Any]): # Without this, every request enqueues another full scan behind the running # one, and a client that lost the progress socket has no way to tell. - if not DEV_MODE and (_get_running_scan_job() or _get_queued_scan_jobs()): - log.info(f"{emoji.EMOJI_STOP_SIGN} Scan already in progress, ignoring request") - await socket_handler.socket_server.emit( - "scan:done_ko", - "A scan is already in progress", - to=sid, - ) - return + if not DEV_MODE: + running_job = _get_running_scan_job() + queued_jobs = _get_queued_scan_jobs() + if running_job is not None or queued_jobs: + message = _scan_in_flight_message(running_job, queued_jobs) + log.info(f"{emoji.EMOJI_STOP_SIGN} {message}, ignoring request") + await socket_handler.socket_server.emit( + "scan:done_ko", + message, + to=sid, + ) + return log.info(f"{emoji.EMOJI_MAGNIFYING_GLASS_TILTED_RIGHT} Scanning") @@ -1394,6 +1487,10 @@ async def stop_scan_handler(sid: str): for job in queued_jobs: job.cancel() + scheduled_jobs = _get_scheduled_scan_jobs() + for job in scheduled_jobs: + _cancel_scheduled_scan_job(job) + # A running scan cannot be interrupted from here, it polls the stop flag # between platforms and ROMs and unwinds itself. running_job = _get_running_scan_job() @@ -1401,11 +1498,12 @@ async def stop_scan_handler(sid: str): running_job.cancel() redis_client.set(STOP_SCAN_FLAG, 1) - if running_job is None and not queued_jobs: + if running_job is None and not queued_jobs and not scheduled_jobs: log.info(f"{emoji.EMOJI_STOP_BUTTON} No running scan to stop") return log.info( f"{emoji.EMOJI_STOP_BUTTON} Stopping scan " - f"({int(running_job is not None)} running, {len(queued_jobs)} queued)" + f"({int(running_job is not None)} running, {len(queued_jobs)} queued, " + f"{len(scheduled_jobs)} scheduled)" ) diff --git a/backend/endpoints/tasks.py b/backend/endpoints/tasks.py index d007dc3a9e..70cebea1da 100644 --- a/backend/endpoints/tasks.py +++ b/backend/endpoints/tasks.py @@ -278,7 +278,13 @@ async def get_tasks_status(request: Request) -> list[TaskStatusResponse]: # Get currently running jobs from workers workers = Worker.all(connection=redis_client) for worker in workers: - current_job = worker.get_current_job() + # A worker killed mid-job keeps pointing at it until its own + # registration expires, and the job can be gone by then. + try: + current_job = worker.get_current_job() + except NoSuchJobError: + continue + if current_job: all_tasks.append(_build_task_status_response(current_job)) diff --git a/backend/handler/redis_handler.py b/backend/handler/redis_handler.py index 9102439604..1205998f6d 100644 --- a/backend/handler/redis_handler.py +++ b/backend/handler/redis_handler.py @@ -5,8 +5,8 @@ from redis import Redis from redis.asyncio import Redis as AsyncRedis from rq import Queue -from rq.exceptions import DeserializationError -from rq.job import Job +from rq.exceptions import DeserializationError, InvalidJobOperation +from rq.job import Job, JobStatus from config import IS_PYTEST_RUN, REDIS_URL from logger.logger import log @@ -74,3 +74,18 @@ def get_job_func_name(job: Job, fallback: str = "") -> str: except DeserializationError: # Job data cannot be deserialized (e.g., function no longer exists) return fallback + + +def get_job_status(job: Job) -> JobStatus | None: + """Safely get the status of an RQ job, which is gone once its hash expires. + + Args: + job: The RQ Job object to get the status of + + Returns: + The job status, or None if the job no longer has one + """ + try: + return job.get_status() + except InvalidJobOperation: + return None diff --git a/backend/startup.py b/backend/startup.py index 4d58a41cf3..12fa9d9f42 100644 --- a/backend/startup.py +++ b/backend/startup.py @@ -16,6 +16,7 @@ SENTRY_DSN, TASK_TIMEOUT, ) +from endpoints.sockets.scan import drop_stale_scheduled_scans from handler.database import db_save_handler from handler.metadata.base_handler import ( MAME_XML_KEY, @@ -44,6 +45,7 @@ from tasks.scheduled.update_launchbox_metadata import update_launchbox_metadata_task from tasks.scheduled.update_switch_titledb import update_switch_titledb_task from tasks.sync_push_pull_task import sync_push_pull_task +from tasks.tasks import drop_unreadable_scheduled_jobs from utils import get_version from utils.cache import conditionally_set_cache from utils.context import initialize_context @@ -140,6 +142,15 @@ async def main() -> None: async with initialize_context(): log.info("Running startup tasks") + # A job the scheduler cannot read crashes it on every poll, so it has + # to go before the scheduler picks it up again, along with the scans that + # piled up behind it while nothing was being released. + try: + drop_unreadable_scheduled_jobs() + drop_stale_scheduled_scans() + except Exception: + log.exception("Failed to clean up the scheduler registry") + # Initialize scheduled tasks cleanup_netplay_task.init() cleanup_zip_cache_task.init() diff --git a/backend/tasks/tasks.py b/backend/tasks/tasks.py index 57d526264e..55211d88be 100644 --- a/backend/tasks/tasks.py +++ b/backend/tasks/tasks.py @@ -21,6 +21,29 @@ SCAN_LIBRARY_TASK_FUNC: Final = "tasks.scheduled.scan_library.scan_library_task.run" +def drop_unreadable_scheduled_jobs() -> int: + """Remove scheduled jobs whose payload can no longer be deserialized. + + The scheduler reads a job's function name before taking it out of the + registry, so one unreadable job left behind by an older version crashes it + on every poll and nothing scheduled ever runs again. + + Returns: + int: How many jobs were dropped. + """ + dropped = 0 + + for job in tasks_scheduler.get_jobs(): + if not isinstance(job, Job) or get_job_func_name(job): + continue + + tasks_scheduler.cancel(job) + dropped += 1 + log.warning(f"Dropped scheduled job {job.id}, its payload cannot be read") + + return dropped + + def update_job_meta(metadata: dict[str, Any]) -> None: """Update the current RQ job's meta data with update stats information""" try: diff --git a/backend/tests/endpoints/sockets/test_scan.py b/backend/tests/endpoints/sockets/test_scan.py index 4a39b0ec09..fe45780331 100644 --- a/backend/tests/endpoints/sockets/test_scan.py +++ b/backend/tests/endpoints/sockets/test_scan.py @@ -1,9 +1,11 @@ +from datetime import datetime, timedelta, timezone from itertools import count from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, Mock import pytest import socketio +from rq.exceptions import InvalidJobOperation, NoSuchJobError from rq.job import Job, JobStatus from endpoints.sockets import scan as scan_module @@ -1687,21 +1689,35 @@ def test_url_contains_fs_path_and_name(self, handler: FSRomsHandler): _job_ids = count() -def make_job(func_name: str, *, status=JobStatus.QUEUED): +def make_job(func_name: str, *, status=JobStatus.QUEUED, task_name: str | None = None): """An RQ job stub that scan job discovery will accept.""" job = MagicMock(spec=Job) job.id = f"job-{next(_job_ids)}" job.func_name = func_name job.get_status.return_value = status + job.meta = {"task_name": task_name} if task_name else {} return job def patch_scan_jobs( - mocker, *, running=None, high_queued=(), low_queued=(), scheduled=() -): - """Point every place scan discovery looks at a fixed set of jobs.""" + mocker, + *, + running=None, + high_queued=(), + low_queued=(), + scheduled=(), + worker_lost=False, +) -> MagicMock: + """Point every place scan discovery looks at a fixed set of jobs. + + Returns the patched scheduler cancel, the only thing that drops a delayed + scan out of the scheduler's registry. + """ worker = MagicMock() - worker.get_current_job.return_value = running + if worker_lost: + worker.get_current_job.side_effect = NoSuchJobError + else: + worker.get_current_job.return_value = running mocker.patch.object(scan_module.Worker, "all", return_value=[worker]) mocker.patch.object( scan_module.high_prio_queue, "get_jobs", return_value=list(high_queued) @@ -1712,6 +1728,7 @@ def patch_scan_jobs( mocker.patch.object( scan_module.tasks_scheduler, "get_jobs", return_value=list(scheduled) ) + return mocker.patch.object(scan_module.tasks_scheduler, "cancel") class TestScanConcurrency: @@ -1767,8 +1784,9 @@ async def test_refuses_when_a_watcher_scan_is_queued(self, mocker, emit): enqueue.assert_not_called() - async def test_refuses_when_a_watcher_scan_is_scheduled(self, mocker, emit): - # A watcher scan waits out its delay in the scheduler before it queues. + async def test_a_scheduled_watcher_scan_does_not_block(self, mocker, emit): + # A delayed watcher scan only moves when the scheduler releases it, so + # a scheduler that is down would refuse manual scans for good. patch_scan_jobs( mocker, scheduled=[make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED)], @@ -1777,7 +1795,79 @@ async def test_refuses_when_a_watcher_scan_is_scheduled(self, mocker, emit): await scan_handler("sid", {"type": "quick"}) - enqueue.assert_not_called() + enqueue.assert_called_once() + + async def test_a_cancelled_queued_scan_does_not_block(self, mocker, emit): + patch_scan_jobs( + mocker, + high_queued=[make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.CANCELED)], + ) + enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue") + + await scan_handler("sid", {"type": "quick"}) + + enqueue.assert_called_once() + + async def test_a_queued_scan_whose_status_is_gone_does_not_block( + self, mocker, emit + ): + # RQ raises rather than reporting a status once the job hash expires. + job = make_job(SCAN_PLATFORMS_FUNC) + job.get_status.side_effect = InvalidJobOperation + patch_scan_jobs(mocker, high_queued=[job]) + enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue") + + await scan_handler("sid", {"type": "quick"}) + + enqueue.assert_called_once() + + async def test_a_worker_holding_a_job_that_is_gone_does_not_block( + self, mocker, emit + ): + patch_scan_jobs(mocker, worker_lost=True) + enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue") + + await scan_handler("sid", {"type": "quick"}) + + enqueue.assert_called_once() + + async def test_refusal_names_the_scan_in_the_way(self, mocker, emit): + patch_scan_jobs( + mocker, + running=make_job( + SCAN_PLATFORMS_FUNC, status=JobStatus.STARTED, task_name="Quick Scan" + ), + ) + mocker.patch.object(scan_module.high_prio_queue, "enqueue") + + await scan_handler("sid", {"type": "quick"}) + + assert emit.await_args.args[1] == "Quick Scan is already running" + + async def test_refusal_says_a_stopping_scan_is_worth_retrying(self, mocker, emit): + # A stopped scan holds the worker until it unwinds, which reads as a + # scan in progress unless the client is told to come back. + patch_scan_jobs( + mocker, + running=make_job( + SCAN_PLATFORMS_FUNC, status=JobStatus.CANCELED, task_name="Quick Scan" + ), + ) + mocker.patch.object(scan_module.high_prio_queue, "enqueue") + + await scan_handler("sid", {"type": "quick"}) + + assert "still stopping" in emit.await_args.args[1] + + async def test_refusal_reports_a_queued_scan_as_queued(self, mocker, emit): + patch_scan_jobs( + mocker, high_queued=[make_job(SCAN_PLATFORMS_FUNC, task_name="Full Scan")] + ) + mocker.patch.object(scan_module.high_prio_queue, "enqueue") + + await scan_handler("sid", {"type": "quick"}) + + assert emit.await_args.args[1] == "Full Scan is already queued" async def test_refuses_when_the_scheduled_rescan_is_running(self, mocker, emit): # The scheduled rescan runs scan_platforms from inside its own task, so @@ -1930,12 +2020,28 @@ async def test_cancels_watcher_scans(self, mocker, emit, redis): # watcher's scan the moment the running one unwinds. low_queued = make_job(SCAN_PLATFORMS_FUNC) scheduled = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED) - patch_scan_jobs(mocker, low_queued=[low_queued], scheduled=[scheduled]) + scheduler_cancel = patch_scan_jobs( + mocker, low_queued=[low_queued], scheduled=[scheduled] + ) await stop_scan_handler("sid") low_queued.cancel.assert_called_once() scheduled.cancel.assert_called_once() + # Cancelling the job leaves its id in the scheduler, which queues it + # anyway once the delay is up. + scheduler_cancel.assert_called_once_with(scheduled) + + async def test_drops_a_scheduled_scan_that_was_already_cancelled( + self, mocker, emit, redis + ): + scheduled = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED) + scheduled.cancel.side_effect = InvalidJobOperation + scheduler_cancel = patch_scan_jobs(mocker, scheduled=[scheduled]) + + await stop_scan_handler("sid") + + scheduler_cancel.assert_called_once_with(scheduled) async def test_cancels_queued_scans_with_none_running(self, mocker, emit, redis): # Stopping a scan that has not been picked up yet must still drop it, @@ -1954,3 +2060,45 @@ async def test_no_scan_to_stop(self, mocker, emit, redis): await stop_scan_handler("sid") redis.set.assert_not_called() + + +class TestDropStaleScheduledScans: + """A recovered scheduler must not release a backlog of delayed scans.""" + + @staticmethod + def _scheduled(func_name: str, *, hours_late: float): + # The scheduler records due times as naive UTC. + due = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta( + hours=hours_late + ) + return (make_job(func_name, status=JobStatus.SCHEDULED), due) + + def _patch(self, mocker, jobs): + mocker.patch.object( + scan_module.tasks_scheduler, "get_jobs", return_value=list(jobs) + ) + return mocker.patch.object(scan_module.tasks_scheduler, "cancel") + + def test_drops_a_scan_long_past_due(self, mocker): + job, due = self._scheduled(SCAN_PLATFORMS_FUNC, hours_late=2) + scheduler_cancel = self._patch(mocker, [(job, due)]) + + assert scan_module.drop_stale_scheduled_scans() == 1 + scheduler_cancel.assert_called_once_with(job) + + def test_keeps_a_scan_still_waiting_out_its_delay(self, mocker): + scheduler_cancel = self._patch( + mocker, [self._scheduled(SCAN_PLATFORMS_FUNC, hours_late=-1)] + ) + + assert scan_module.drop_stale_scheduled_scans() == 0 + scheduler_cancel.assert_not_called() + + def test_leaves_the_standing_rescan_cron_entry_alone(self, mocker): + # The cron entry reschedules itself, so a missed run is not a backlog. + scheduler_cancel = self._patch( + mocker, [self._scheduled(scan_module.SCAN_LIBRARY_TASK_FUNC, hours_late=2)] + ) + + assert scan_module.drop_stale_scheduled_scans() == 0 + scheduler_cancel.assert_not_called() diff --git a/backend/tests/tasks/test_tasks.py b/backend/tests/tasks/test_tasks.py index 3ff7f69ab1..4dd3ea0ef8 100644 --- a/backend/tests/tasks/test_tasks.py +++ b/backend/tests/tasks/test_tasks.py @@ -1,11 +1,18 @@ -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch import httpx import pytest +from rq.exceptions import DeserializationError from rq.job import Job from exceptions.task_exceptions import SchedulerException -from tasks.tasks import PeriodicTask, RemoteFilePullTask, TaskType, tasks_scheduler +from tasks.tasks import ( + PeriodicTask, + RemoteFilePullTask, + TaskType, + drop_unreadable_scheduled_jobs, + tasks_scheduler, +) class ConcretePeriodicTask(PeriodicTask): @@ -322,3 +329,34 @@ async def test_run_disabled_but_forced(self, mock_ctx_httpx_client, disabled_tas result = await disabled_task.run(force=True) assert result == b"forced content" + + +class TestDropUnreadableScheduledJobs: + """One unreadable job crashes the scheduler on every poll until it is gone.""" + + @staticmethod + def _job(func_name: str | None): + job = MagicMock(spec=Job) + job.id = "job-1" + if func_name is None: + type(job).func_name = PropertyMock(side_effect=DeserializationError) + else: + job.func_name = func_name + return job + + @patch.object(tasks_scheduler, "cancel") + @patch.object(tasks_scheduler, "get_jobs") + def test_drops_a_job_whose_payload_cannot_be_read(self, get_jobs, cancel): + job = self._job(None) + get_jobs.return_value = [job] + + assert drop_unreadable_scheduled_jobs() == 1 + cancel.assert_called_once_with(job) + + @patch.object(tasks_scheduler, "cancel") + @patch.object(tasks_scheduler, "get_jobs") + def test_leaves_readable_jobs_scheduled(self, get_jobs, cancel): + get_jobs.return_value = [self._job("test.function")] + + assert drop_unreadable_scheduled_jobs() == 0 + cancel.assert_not_called() From 465879ecbfc6ab41604bfda2cf5329e2a65d6c68 Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Sun, 23 Aug 2026 14:34:17 -0500 Subject: [PATCH 2/8] chore(deps): upgrade rq to 2.11 The lockfile sat on 2.9.0 while the constraint allowed anything under 3.0, so the floor now names the version the code is tested against. 2.11 carries what the scheduler work ahead needs: RQScheduler acquires and refreshes its lock before enqueueing, each CronJob has a name and keeps the ids of the jobs it created, and calling create_cron() twice no longer duplicates jobs. 2.10 added webhook notifications and a stable scheduler identity, and 2.9.1 covers redis-py >= 8. 2.11.0 was published inside the rolling 7-day window, so it needs a per-package exclusion until 2026-08-24, when the window reaches it and the entry can go. The constraint stays under 3.0 deliberately. That release moves get_current_job() to contextvars, reworks dependency handling behind a ReadyJobRegistry, and changes the Worker.handle_job_success() signature, which RomMWorker subclasses around. Co-Authored-By: Claude Opus 5 --- pyproject.toml | 7 +++++-- uv.lock | 9 +++++---- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index c50fb843d7..6fbe370c16 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,7 +47,7 @@ dependencies = [ "python-magic ~= 0.4", "python-socketio ~= 5.16", "redis ~= 6.2", - "rq ~= 2.7", + "rq ~= 2.11", # TODO: Move back to upstream `rq-scheduler`, when support for username and SSL settings is added. # Related PR: https://github.com/rq/rq-scheduler/pull/325 "rq-scheduler @ git+https://github.com/adamantike/rq-scheduler.git@feat/script-options-username-ssl", @@ -146,7 +146,10 @@ package = false exclude-newer = "7 days" # vcrpy >= 8.2.0 is required for aiohttp 3.14 compatibility (the removal of # `AsyncStreamReaderMixin`); allow it past the rolling 7-day window. -exclude-newer-package = { vcrpy = "2026-06-17" } +# rq 2.11.0 carries the scheduler locking and cron job history the scheduler +# migration builds on. The rolling window reaches it on 2026-08-24, after which +# this entry can go. +exclude-newer-package = { vcrpy = "2026-06-17", rq = "2026-08-17" } [tool.ty.environment] root = ["./backend"] diff --git a/uv.lock b/uv.lock index 10bfbf1224..b9d5cecc6b 100644 --- a/uv.lock +++ b/uv.lock @@ -11,6 +11,7 @@ exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for exclude-newer-span = "P7D" [options.exclude-newer-package] +rq = "2026-08-18T05:00:00Z" vcrpy = "2026-06-18T05:00:00Z" [[package]] @@ -2319,7 +2320,7 @@ requires-dist = [ { name = "python-socketio", specifier = "~=5.16" }, { name = "pyyaml", specifier = "~=6.0" }, { name = "redis", specifier = "~=6.2" }, - { name = "rq", specifier = "~=2.7" }, + { name = "rq", specifier = "~=2.11" }, { name = "rq-scheduler", git = "https://github.com/adamantike/rq-scheduler.git?rev=feat%2Fscript-options-username-ssl" }, { name = "sentry-sdk", specifier = "~=2.32" }, { name = "sqlalchemy", extras = ["mariadb-connector", "mysql-connector", "postgresql-psycopg"], specifier = "~=2.0" }, @@ -2343,16 +2344,16 @@ provides-extras = ["dev", "test"] [[package]] name = "rq" -version = "2.9.0" +version = "2.11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "click" }, { name = "croniter" }, { name = "redis" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/15/5e/43a7a61f3ebfa79789c72bf442a47fd80bb1a743caeea47b2b833001f388/rq-2.9.0.tar.gz", hash = "sha256:db5dfc1e1fe80ef977fd557d4305107dcf99a80b33d381c9a06e8a2bb11730e5", size = 744959, upload-time = "2026-05-19T15:00:29.894Z" } +sdist = { url = "https://files.pythonhosted.org/packages/d7/8a/7fcf759b5b685b26ef029af24575359bf8f5b5b340d92b08bb193a4163b1/rq-2.11.0.tar.gz", hash = "sha256:1ed9db6b685707a7dfd535532a408f1a8f0a5ad720a307be6b844f4142999e78", size = 759304, upload-time = "2026-08-17T02:59:19.695Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/4d/a9/aa38ae2505e5dcb1e898f83a263e703b05829580f75ee86c5e480760ae1a/rq-2.9.0-py3-none-any.whl", hash = "sha256:665b9ad34e36ea15913e60d2a32e1775fef1e9aff907bcae297d6f70dccffed2", size = 120113, upload-time = "2026-05-19T15:00:27.511Z" }, + { url = "https://files.pythonhosted.org/packages/80/1e/f85cba91fe58c63812b4acce073fc448230381c4785e03832f455305ff82/rq-2.11.0-py3-none-any.whl", hash = "sha256:2f54a31375d7c1b5a1642acef4948809ce6484775b9741fe7edd9399ec9306a9", size = 126654, upload-time = "2026-08-17T02:59:17.112Z" }, ] [[package]] From e546ad74a6b68c530d4b13e6efb6f7227a4df183 Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Sun, 23 Aug 2026 14:57:17 -0500 Subject: [PATCH 3/8] refactor(tasks): schedule with RQ instead of rq-scheduler rq-scheduler has one release since 2023, 109 open issues, and RomM ran a fork of it for Redis username and SSL options that upstream never took. Everything it was here for now ships with RQ. Its two failure modes were the reason for the workarounds in 9e9f59c04. It reads a job's function name before taking the job out of its registry, so one unreadable job stalls it for good; RQ's scheduler removes every id it looked at whether or not the job could be read. And cancelling a delayed job needed both `Scheduler.cancel` and `Job.cancel`, because the first left the status alone and the second left the registry entry, which the scheduler then queued anyway; `Job.cancel` on its own now clears the scheduled registry. Delayed jobs, which is how the watcher defers a rescan, go onto the queue with `Queue.enqueue_in` and are released by the worker running with `--with-scheduler`. Periodic tasks are declared in `tasks/cron_config.py` and loaded by an `rq cron` process, replacing `rqscheduler`. A task is registered only when it is enabled and has a cron string, so the schedule-then-unschedule dance around the env toggles is gone, and with it `PeriodicTask.init/schedule/unschedule`. Jobs are enqueued by task name through `run_task_by_name`, resolved against a catalog in `tasks/registry.py`. Nothing pickles a task instance into Redis any more, which is what made a payload unreadable across upgrades in the first place. The catalog also gives the tasks endpoint one home for what a task is, rather than a list of its own. Scan discovery keys off the type a job carries rather than the scheduled rescan's function name, since every task now shares one entry point. That also drops SCAN_LIBRARY_TASK_FUNC, which only existed to name a function across an import cycle. The watcher's dedupe worked on `job.args[0]` while scans are enqueued with keyword arguments, so it never matched and every filesystem change scheduled another scan. It reads `platform_ids` from the payload now, and deliberately ignores a scan already running: that one may have walked past the folder that just changed. Startup clears what the old scheduler left in Redis, once. Co-Authored-By: Claude Opus 5 --- backend/endpoints/responses/__init__.py | 2 +- backend/endpoints/sockets/scan.py | 108 ++++---- backend/endpoints/tasks.py | 126 ++------- backend/exceptions/task_exceptions.py | 5 + backend/handler/redis_handler.py | 16 ++ backend/startup.py | 103 ++++---- backend/tasks/cron_config.py | 30 +++ backend/tasks/registry.py | 50 ++++ backend/tasks/scheduled/cleanup_netplay.py | 1 - .../scheduled/cleanup_orphaned_resources.py | 10 - backend/tasks/scheduled/cleanup_upload_tmp.py | 1 - backend/tasks/scheduled/cleanup_zip_cache.py | 1 - backend/tasks/scheduled/scan_library.py | 7 +- backend/tasks/tasks.py | 112 ++------ backend/tests/endpoints/sockets/test_scan.py | 128 +++++----- backend/tests/endpoints/test_tasks.py | 241 ++++++++---------- .../tasks/test_cleanup_orphaned_resources.py | 22 -- backend/tests/tasks/test_cleanup_zip_cache.py | 5 +- backend/tests/tasks/test_cron_config.py | 66 +++++ backend/tests/tasks/test_prevent_requeue.py | 48 ---- backend/tests/tasks/test_registry.py | 22 ++ backend/tests/tasks/test_scan_library.py | 4 +- backend/tests/tasks/test_tasks.py | 222 ++-------------- backend/watcher.py | 98 ++----- docker/init_scripts/init | 61 +++-- docs/BACKEND_ARCHITECTURE.md | 20 +- entrypoint.sh | 31 ++- pyproject.toml | 3 - uv.lock | 18 -- 29 files changed, 617 insertions(+), 944 deletions(-) create mode 100644 backend/tasks/cron_config.py create mode 100644 backend/tasks/registry.py create mode 100644 backend/tests/tasks/test_cron_config.py delete mode 100644 backend/tests/tasks/test_prevent_requeue.py create mode 100644 backend/tests/tasks/test_registry.py diff --git a/backend/endpoints/responses/__init__.py b/backend/endpoints/responses/__init__.py index e1dfe5042f..b8a7647e89 100644 --- a/backend/endpoints/responses/__init__.py +++ b/backend/endpoints/responses/__init__.py @@ -1,6 +1,6 @@ from typing import Literal, TypedDict, Union -from rq_scheduler.scheduler import JobStatus +from rq.job import JobStatus from tasks.tasks import TaskType diff --git a/backend/endpoints/sockets/scan.py b/backend/endpoints/sockets/scan.py index 7d208a32b0..81b4a6f90e 100644 --- a/backend/endpoints/sockets/scan.py +++ b/backend/endpoints/sockets/scan.py @@ -9,8 +9,9 @@ import pydash import socketio # type: ignore from rq import Worker, get_current_job -from rq.exceptions import InvalidJobOperation, NoSuchJobError +from rq.exceptions import NoSuchJobError from rq.job import Job, JobStatus +from rq.registry import ScheduledJobRegistry from sqlalchemy.exc import IntegrityError from config import DEV_MODE, REDIS_URL, SCAN_TIMEOUT, SCAN_WORKERS, TASK_RESULT_TTL @@ -74,7 +75,7 @@ from models.firmware import Firmware from models.platform import Platform from models.rom import Rom -from tasks.tasks import SCAN_LIBRARY_TASK_FUNC, tasks_scheduler, update_job_meta +from tasks.tasks import update_job_meta from utils import emoji from utils.audio_tags import remove_persisted_cover from utils.context import initialize_context @@ -83,8 +84,8 @@ STOP_SCAN_FLAG: Final = "scan:stop" -# A delayed watcher scan this far past due was left behind by a scheduler that -# stopped releasing them, and the change it reacted to has long since settled. +# A delayed watcher scan this far past due was left behind by an instance that +# was not running, and the change it reacted to has long since settled. STALE_SCHEDULED_SCAN_AGE: Final = timedelta(hours=1) @@ -97,14 +98,17 @@ def _scan_platforms_func_name() -> str: return f"{scan_platforms.__module__}.{scan_platforms.__name__}" -def _scan_job_func_names() -> frozenset[str]: - """Every job function name that ends up running a scan. +def _is_scan_job(job: Job) -> bool: + """Whether this job runs a scan. - Socket and watcher scans enqueue scan_platforms itself, while the scheduled - rescan enqueues its own task and calls scan_platforms in process. Both have - to be recognised or an in-flight scan goes unseen. + Socket and watcher scans enqueue scan_platforms itself. Task-driven scans go + through the task runner, which every task shares, so they are recognised by + the type their job carries instead. """ - return frozenset((_scan_platforms_func_name(), SCAN_LIBRARY_TASK_FUNC)) + if get_job_func_name(job) == _scan_platforms_func_name(): + return True + + return job.meta.get("task_type") == TaskType.SCAN def _get_running_scan_job() -> Job | None: @@ -113,7 +117,6 @@ def _get_running_scan_job() -> Job | None: A started job is no longer in the queue, so it can only be found by asking the workers what they are holding. """ - func_names = _scan_job_func_names() for worker in Worker.all(connection=redis_client): # A worker killed mid-scan keeps pointing at its job until its own # registration expires, and the job can be gone by then. @@ -122,7 +125,7 @@ def _get_running_scan_job() -> Job | None: except NoSuchJobError: continue - if job is not None and get_job_func_name(job) in func_names: + if job is not None and _is_scan_job(job): return job return None @@ -132,15 +135,14 @@ def _get_queued_scan_jobs() -> list[Job]: """Scans sitting on a worker queue, waiting to be picked up. Socket scans go to the high priority queue and watcher scans to the low - priority one, once the scheduler releases them. + priority one, once their delay is up. """ - func_names = _scan_job_func_names() jobs: dict[str, Job] = {} for job in chain(high_prio_queue.get_jobs(), low_prio_queue.get_jobs()): if ( isinstance(job, Job) - and get_job_func_name(job) in func_names + and _is_scan_job(job) and get_job_status(job) == JobStatus.QUEUED ): jobs[job.id] = job @@ -148,70 +150,57 @@ def _get_queued_scan_jobs() -> list[Job]: return list(jobs.values()) +def _scheduled_scan_registry() -> ScheduledJobRegistry: + """Where delayed scans wait, which is only ever the watcher's queue.""" + return ScheduledJobRegistry(queue=low_prio_queue) + + def _get_scheduled_scan_jobs() -> list[Job]: - """Scans waiting out a delay in the scheduler, which only the watcher sets. + """Scans waiting out a delay, which only the watcher sets. - The scheduler is the only thing that releases them, so one that is down - leaves them there for good and they cannot stand in for a scan in flight. + These never stand in for a scan in flight: a worker has to be running to + release them, so counting them would refuse scans on an idle instance. """ - # The registry also holds the standing cron entry for the scheduled rescan, - # which is a schedule rather than a pending scan, so only delayed - # scan_platforms jobs count here. - scan_platforms_func_name = _scan_platforms_func_name() - jobs: dict[str, Job] = {} + registry = _scheduled_scan_registry() + jobs = Job.fetch_many(registry.get_job_ids(), connection=redis_client) - for job in tasks_scheduler.get_jobs(): - if ( - isinstance(job, Job) - and get_job_func_name(job) == scan_platforms_func_name - and get_job_status(job) in (JobStatus.SCHEDULED, JobStatus.QUEUED) - ): - jobs[job.id] = job + return [job for job in jobs if job is not None and _is_scan_job(job)] - return list(jobs.values()) +def get_pending_scan_jobs() -> list[Job]: + """Scans that have not started yet: queued, or waiting out a delay. -def _cancel_scheduled_scan_job(job: Job) -> None: - """Drop a delayed scan from the scheduler. - - Cancelling the job on its own leaves the id in the scheduler's registry, and - the scheduler queues it anyway once the delay is up, which runs the scan that - was just stopped. + A scan already running is deliberately not one of these. It may have walked + past the folder that just changed, so a fresh scan is still warranted. """ - tasks_scheduler.cancel(job) - try: - job.cancel() - except InvalidJobOperation: - # Already cancelled; the registry entry was the part that mattered. - pass + return _get_queued_scan_jobs() + _get_scheduled_scan_jobs() def drop_stale_scheduled_scans() -> int: """Drop delayed watcher scans that are long past due. - Releasing a backlog of them at once, which is what a stalled scheduler does - the moment it recovers, would run the same library scan over and over. + Releasing a backlog of them at once, which is what an instance that was down + for a while does on start, would run the same library scan over and over. Returns: int: How many scans were dropped. """ - scan_platforms_func_name = _scan_platforms_func_name() + registry = _scheduled_scan_registry() cutoff = datetime.now(timezone.utc) - STALE_SCHEDULED_SCAN_AGE dropped = 0 - for job, scheduled_at in tasks_scheduler.get_jobs(with_times=True): - if ( - not isinstance(job, Job) - or get_job_func_name(job) != scan_platforms_func_name - # The scheduler records due times as naive UTC. - or scheduled_at is None - or scheduled_at.replace(tzinfo=timezone.utc) > cutoff - ): + for job in _get_scheduled_scan_jobs(): + try: + scheduled_at = registry.get_scheduled_time(job) + except NoSuchJobError: continue - _cancel_scheduled_scan_job(job) + if scheduled_at > cutoff: + continue + + job.cancel() dropped += 1 - log.warning(f"Dropped scan scheduled for {scheduled_at} UTC, too long past due") + log.warning(f"Dropped scan scheduled for {scheduled_at}, too long past due") return dropped @@ -1484,12 +1473,9 @@ async def stop_scan_handler(sid: str): # Queued scans have not started, so cancelling them is enough. They have to # go too: stopping only the running scan would hand the worker the next one. queued_jobs = _get_queued_scan_jobs() - for job in queued_jobs: - job.cancel() - scheduled_jobs = _get_scheduled_scan_jobs() - for job in scheduled_jobs: - _cancel_scheduled_scan_job(job) + for job in queued_jobs + scheduled_jobs: + job.cancel() # A running scan cannot be interrupted from here, it polls the stop flag # between platforms and ROMs and unwinds itself. diff --git a/backend/endpoints/tasks.py b/backend/endpoints/tasks.py index 70cebea1da..8816092e3e 100644 --- a/backend/endpoints/tasks.py +++ b/backend/endpoints/tasks.py @@ -1,5 +1,5 @@ from datetime import datetime, timezone -from typing import Any, TypedDict +from typing import Any, Final from fastapi import Body, HTTPException, Request from rq import Worker @@ -33,21 +33,8 @@ low_prio_queue, redis_client, ) -from tasks.manual.cleanup_missing_roms import cleanup_missing_roms_task -from tasks.manual.recompute_save_content_hashes import ( - recompute_save_content_hashes_task, -) -from tasks.manual.sync_folder_scan import sync_folder_scan_task -from tasks.scheduled.cleanup_orphaned_resources import cleanup_orphaned_resources_task -from tasks.scheduled.cleanup_zip_cache import cleanup_zip_cache_task -from tasks.scheduled.convert_images_to_webp import convert_images_to_webp_task -from tasks.scheduled.scan_library import scan_library_task -from tasks.scheduled.update_launchbox_metadata import update_launchbox_metadata_task -from tasks.scheduled.update_switch_titledb import update_switch_titledb_task -from tasks.tasks import ( - Task, - TaskType, -) +from tasks.registry import MANUAL_TASKS, SCHEDULED_TASKS +from tasks.tasks import Task, TaskType, run_task_by_name from utils.router import APIRouter router = APIRouter( @@ -56,84 +43,16 @@ ) -class ScheduledTask(TypedDict): - name: str - type: TaskType - task: Task - - -class ManualTask(ScheduledTask): - pass - - -scheduled_tasks: list[ScheduledTask] = [ - ScheduledTask( - { - "name": "scan_library", - "type": TaskType.SCAN, - "task": scan_library_task, - } - ), - ScheduledTask( - { - "name": "update_launchbox_metadata", - "type": TaskType.UPDATE, - "task": update_launchbox_metadata_task, - } - ), - ScheduledTask( - { - "name": "update_switch_titledb", - "type": TaskType.UPDATE, - "task": update_switch_titledb_task, - } - ), - ScheduledTask( - { - "name": "convert_images_to_webp", - "type": TaskType.CONVERSION, - "task": convert_images_to_webp_task, - } - ), - ScheduledTask( - { - "name": "cleanup_zip_cache", - "type": TaskType.CLEANUP, - "task": cleanup_zip_cache_task, - } - ), - ScheduledTask( - { - "name": "cleanup_orphaned_resources", - "type": TaskType.CLEANUP, - "task": cleanup_orphaned_resources_task, - } - ), -] - -manual_tasks: list[ManualTask] = [ - ManualTask( - { - "name": "cleanup_missing_roms", - "type": TaskType.CLEANUP, - "task": cleanup_missing_roms_task, - } - ), - ManualTask( - { - "name": "sync_folder_scan", - "type": TaskType.SYNC, - "task": sync_folder_scan_task, - } - ), - ManualTask( - { - "name": "recompute_save_content_hashes", - "type": TaskType.CLEANUP, - "task": recompute_save_content_hashes_task, - } - ), -] +# Scheduled tasks an admin can see and trigger. The rest of the catalog runs on +# its schedule without being surfaced. +VISIBLE_SCHEDULED_TASKS: Final = ( + "scan_library", + "update_launchbox_metadata", + "update_switch_titledb", + "convert_images_to_webp", + "cleanup_zip_cache", + "cleanup_orphaned_resources", +) def _build_task_info(name: str, task: Task) -> TaskInfo: @@ -242,11 +161,11 @@ async def list_tasks(request: Request) -> GroupedTasksDict: "watcher": [], } - for task in manual_tasks: - grouped_tasks["manual"].append(_build_task_info(task["name"], task["task"])) + for name, task in MANUAL_TASKS.items(): + grouped_tasks["manual"].append(_build_task_info(name, task)) - for task in scheduled_tasks: - grouped_tasks["scheduled"].append(_build_task_info(task["name"], task["task"])) + for name in VISIBLE_SCHEDULED_TASKS: + grouped_tasks["scheduled"].append(_build_task_info(name, SCHEDULED_TASKS[name])) # Add the adhoc watcher task grouped_tasks["watcher"].append( @@ -380,7 +299,10 @@ async def run_single_task( Returns: TaskExecutionResponse: Task execution response with details """ - all_tasks = {task["name"]: task["task"] for task in manual_tasks + scheduled_tasks} + all_tasks: dict[str, Task] = { + **MANUAL_TASKS, + **{name: SCHEDULED_TASKS[name] for name in VISIBLE_SCHEDULED_TASKS}, + } if task_name not in all_tasks: available_tasks = list(all_tasks.keys()) @@ -396,9 +318,11 @@ async def run_single_task( detail=f"Task '{task_name}' cannot be run", ) + # Enqueued by name, like the scheduled runs, so the payload carries no + # pickled task and the job is readable by whatever version picks it up. job = low_prio_queue.enqueue( - task_instance.run, - kwargs=task_kwargs or {}, + run_task_by_name, + kwargs={"name": task_name, **(task_kwargs or {})}, job_timeout=task_instance.timeout, result_ttl=TASK_RESULT_TTL, meta={ diff --git a/backend/exceptions/task_exceptions.py b/backend/exceptions/task_exceptions.py index c5f3da58b2..0ed51fd1dd 100644 --- a/backend/exceptions/task_exceptions.py +++ b/backend/exceptions/task_exceptions.py @@ -5,3 +5,8 @@ def __init__(self, message: str): def __repr__(self): return self.message + + +class TaskNotFoundException(SchedulerException): + def __init__(self, name: str): + super().__init__(f"No task is registered under the name '{name}'") diff --git a/backend/handler/redis_handler.py b/backend/handler/redis_handler.py index 1205998f6d..fa2761bed7 100644 --- a/backend/handler/redis_handler.py +++ b/backend/handler/redis_handler.py @@ -1,6 +1,7 @@ import os import sys from enum import Enum +from typing import Any from redis import Redis from redis.asyncio import Redis as AsyncRedis @@ -89,3 +90,18 @@ def get_job_status(job: Job) -> JobStatus | None: return job.get_status() except InvalidJobOperation: return None + + +def get_job_kwargs(job: Job) -> dict[str, Any] | None: + """Safely get the keyword arguments an RQ job was enqueued with. + + Args: + job: The RQ Job object to read + + Returns: + The keyword arguments, or None if the payload cannot be deserialized + """ + try: + return job.kwargs + except DeserializationError: + return None diff --git a/backend/startup.py b/backend/startup.py index 12fa9d9f42..4ad314ed51 100644 --- a/backend/startup.py +++ b/backend/startup.py @@ -8,11 +8,6 @@ from config import ( ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP, - ENABLE_SCHEDULED_RESCAN, - ENABLE_SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC, - ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA, - ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB, - ENABLE_SYNC_PUSH_PULL, SENTRY_DSN, TASK_TIMEOUT, ) @@ -27,25 +22,19 @@ PSP_SERIAL_INDEX_KEY, SCUMMVM_INDEX_KEY, ) -from handler.redis_handler import async_cache, low_prio_queue +from handler.redis_handler import ( + async_cache, + default_queue, + high_prio_queue, + low_prio_queue, + redis_client, +) from logger.logger import log from models.firmware import FIRMWARE_FIXTURES_DIR, KNOWN_BIOS_KEY from tasks.manual.recompute_save_content_hashes import ( recompute_save_content_hashes_task, ) -from tasks.scheduled.cleanup_netplay import cleanup_netplay_task -from tasks.scheduled.cleanup_orphaned_resources import cleanup_orphaned_resources_task -from tasks.scheduled.cleanup_upload_tmp import cleanup_upload_tmp_task -from tasks.scheduled.cleanup_zip_cache import cleanup_zip_cache_task from tasks.scheduled.convert_images_to_webp import convert_images_to_webp_task -from tasks.scheduled.scan_library import scan_library_task -from tasks.scheduled.sync_retroachievements_progress import ( - sync_retroachievements_progress_task, -) -from tasks.scheduled.update_launchbox_metadata import update_launchbox_metadata_task -from tasks.scheduled.update_switch_titledb import update_switch_titledb_task -from tasks.sync_push_pull_task import sync_push_pull_task -from tasks.tasks import drop_unreadable_scheduled_jobs from utils import get_version from utils.cache import conditionally_set_cache from utils.context import initialize_context @@ -135,6 +124,48 @@ def _enqueue_convert_images_to_webp() -> None: ) +# Keys the rq-scheduler process used before scheduling moved onto RQ itself. +# Everything it held is either obsolete or now owned by the cron config, so it +# only has to be cleared once, and this can go a release or two from now. +LEGACY_SCHEDULED_JOBS_KEY = "rq:scheduler:scheduled_jobs" +LEGACY_SCHEDULER_KEYS = ( + LEGACY_SCHEDULED_JOBS_KEY, + "rq:scheduler_lock", + "rq:scheduler", +) + + +def _drop_legacy_scheduler_state() -> None: + """Clear what the old scheduler left in Redis, jobs included.""" + try: + legacy_job_ids = { + job_id.decode() + for job_id in redis_client.zrange(LEGACY_SCHEDULED_JOBS_KEY, 0, -1) + } + if not legacy_job_ids: + return + + # A cron job the old scheduler had already queued lives in both places, + # and it still has to run, so only the orphans are deleted. + queued = set() + for queue in (high_prio_queue, default_queue, low_prio_queue): + queued.update(queue.get_job_ids()) + + orphans = legacy_job_ids - queued + if orphans: + redis_client.delete(*(f"rq:job:{job_id}" for job_id in orphans)) + + redis_client.delete(*LEGACY_SCHEDULER_KEYS) + for key in redis_client.scan_iter("rq:scheduler_instance:*"): + redis_client.delete(key) + + log.info( + f"Cleared {len(legacy_job_ids)} job(s) left behind by the old scheduler" + ) + except Exception: + log.exception("Failed to clear the old scheduler's leftovers") + + @tracer.start_as_current_span("main") async def main() -> None: """Run startup tasks.""" @@ -142,40 +173,18 @@ async def main() -> None: async with initialize_context(): log.info("Running startup tasks") - # A job the scheduler cannot read crashes it on every poll, so it has - # to go before the scheduler picks it up again, along with the scans that - # piled up behind it while nothing was being released. + # An instance that was down for a while comes back with every rescan + # its watcher queued still waiting, and releasing them all would run the + # same library scan over and over. try: - drop_unreadable_scheduled_jobs() drop_stale_scheduled_scans() except Exception: - log.exception("Failed to clean up the scheduler registry") - - # Initialize scheduled tasks - cleanup_netplay_task.init() - cleanup_zip_cache_task.init() - cleanup_upload_tmp_task.init() - cleanup_orphaned_resources_task.init() - - if ENABLE_SCHEDULED_RESCAN: - log.info("Starting scheduled rescan") - scan_library_task.init() - if ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB: - log.info("Starting scheduled update switch titledb") - update_switch_titledb_task.init() - if ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA: - log.info("Starting scheduled update launchbox metadata") - update_launchbox_metadata_task.init() + log.exception("Failed to check for stale scheduled scans") + + _drop_legacy_scheduler_state() + if ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP: - log.info("Starting scheduled convert images to webp") - convert_images_to_webp_task.init() _enqueue_convert_images_to_webp() - if ENABLE_SCHEDULED_RETROACHIEVEMENTS_PROGRESS_SYNC: - log.info("Starting scheduled RetroAchievements progress sync") - sync_retroachievements_progress_task.init() - if ENABLE_SYNC_PUSH_PULL: - log.info("Starting scheduled push-pull sync") - sync_push_pull_task.init() _enqueue_recompute_save_hashes_if_needed() diff --git a/backend/tasks/cron_config.py b/backend/tasks/cron_config.py new file mode 100644 index 0000000000..ea9f472369 --- /dev/null +++ b/backend/tasks/cron_config.py @@ -0,0 +1,30 @@ +"""The schedule `rq cron` runs, registered when that process starts. + +A task is registered only when it is enabled and has a cron string, so nothing +in Redis has to be unscheduled when a deployment turns one off: the next start +simply leaves it out. +""" + +from rq import cron + +from handler.redis_handler import QueuePrio +from logger.logger import log +from tasks.registry import SCHEDULED_TASKS +from tasks.tasks import run_task_by_name + +for name, task in SCHEDULED_TASKS.items(): + if not task.enabled or not task.cron_string: + continue + + cron.register( + run_task_by_name, + QueuePrio.LOW.value, + kwargs={"name": name}, + cron=task.cron_string, + job_timeout=task.timeout, + meta={ + "task_name": task.title, + "task_type": task.task_type.value, + }, + ) + log.info(f"Scheduled '{name}' at '{task.cron_string}'") diff --git a/backend/tasks/registry.py b/backend/tasks/registry.py new file mode 100644 index 0000000000..0330326c7b --- /dev/null +++ b/backend/tasks/registry.py @@ -0,0 +1,50 @@ +"""The catalog of tasks an admin can see, run, or have run on a schedule.""" + +from typing import Final + +from tasks.manual.cleanup_missing_roms import cleanup_missing_roms_task +from tasks.manual.recompute_save_content_hashes import ( + recompute_save_content_hashes_task, +) +from tasks.manual.sync_folder_scan import sync_folder_scan_task +from tasks.scheduled.cleanup_netplay import cleanup_netplay_task +from tasks.scheduled.cleanup_orphaned_resources import cleanup_orphaned_resources_task +from tasks.scheduled.cleanup_upload_tmp import cleanup_upload_tmp_task +from tasks.scheduled.cleanup_zip_cache import cleanup_zip_cache_task +from tasks.scheduled.convert_images_to_webp import convert_images_to_webp_task +from tasks.scheduled.scan_library import scan_library_task +from tasks.scheduled.sync_retroachievements_progress import ( + sync_retroachievements_progress_task, +) +from tasks.scheduled.update_launchbox_metadata import update_launchbox_metadata_task +from tasks.scheduled.update_switch_titledb import update_switch_titledb_task +from tasks.sync_push_pull_task import sync_push_pull_task +from tasks.tasks import PeriodicTask, Task + +# The keys are the names the API and the cron schedule address a task by, and +# they end up in the job payload, so they outlive any given release. Every task +# that runs on a schedule belongs here; which of them the API surfaces is the +# endpoint's business. +SCHEDULED_TASKS: Final[dict[str, PeriodicTask]] = { + "scan_library": scan_library_task, + "update_launchbox_metadata": update_launchbox_metadata_task, + "update_switch_titledb": update_switch_titledb_task, + "convert_images_to_webp": convert_images_to_webp_task, + "cleanup_zip_cache": cleanup_zip_cache_task, + "cleanup_orphaned_resources": cleanup_orphaned_resources_task, + "cleanup_netplay": cleanup_netplay_task, + "cleanup_upload_tmp": cleanup_upload_tmp_task, + "sync_retroachievements_progress": sync_retroachievements_progress_task, + "sync_push_pull": sync_push_pull_task, +} + +MANUAL_TASKS: Final[dict[str, Task]] = { + "cleanup_missing_roms": cleanup_missing_roms_task, + "sync_folder_scan": sync_folder_scan_task, + "recompute_save_content_hashes": recompute_save_content_hashes_task, +} + + +def get_task(name: str) -> Task | None: + """Look up a task by the name it is addressed by.""" + return SCHEDULED_TASKS.get(name) or MANUAL_TASKS.get(name) diff --git a/backend/tasks/scheduled/cleanup_netplay.py b/backend/tasks/scheduled/cleanup_netplay.py index 5eed908e0d..635ae07835 100644 --- a/backend/tasks/scheduled/cleanup_netplay.py +++ b/backend/tasks/scheduled/cleanup_netplay.py @@ -17,7 +17,6 @@ def __init__(self): async def run(self) -> None: if not self.enabled: - self.unschedule() return netplay_rooms = await netplay_handler.get_all() diff --git a/backend/tasks/scheduled/cleanup_orphaned_resources.py b/backend/tasks/scheduled/cleanup_orphaned_resources.py index 5f9f3e5c6c..3c25f99d38 100644 --- a/backend/tasks/scheduled/cleanup_orphaned_resources.py +++ b/backend/tasks/scheduled/cleanup_orphaned_resources.py @@ -4,7 +4,6 @@ from dataclasses import dataclass from anyio import Path as AnyioPath -from rq.job import Job from config import ( ENABLE_SCHEDULED_CLEANUP_ORPHANED_RESOURCES, @@ -86,15 +85,6 @@ def __init__(self): func="tasks.scheduled.cleanup_orphaned_resources.cleanup_orphaned_resources_task.run", ) - def init(self) -> Job | None: - # Without a cron string there is nothing to schedule, so drop any job - # left over from a previous configuration. - if not self.cron_string: - self.unschedule() - return None - - return super().init() - @initialize_context() async def run(self, force: bool = False) -> dict[str, int]: """Clean up orphaned resources. diff --git a/backend/tasks/scheduled/cleanup_upload_tmp.py b/backend/tasks/scheduled/cleanup_upload_tmp.py index 86a8137ec6..925329988e 100644 --- a/backend/tasks/scheduled/cleanup_upload_tmp.py +++ b/backend/tasks/scheduled/cleanup_upload_tmp.py @@ -20,7 +20,6 @@ def __init__(self): async def run(self) -> None: if not self.enabled: - self.unschedule() return if not ROM_UPLOAD_TMP_BASE.exists(): diff --git a/backend/tasks/scheduled/cleanup_zip_cache.py b/backend/tasks/scheduled/cleanup_zip_cache.py index f81436861e..f69a5d3b45 100644 --- a/backend/tasks/scheduled/cleanup_zip_cache.py +++ b/backend/tasks/scheduled/cleanup_zip_cache.py @@ -17,7 +17,6 @@ def __init__(self): async def run(self) -> None: if not self.enabled: - self.unschedule() return deleted = cleanup_stale_zips() diff --git a/backend/tasks/scheduled/scan_library.py b/backend/tasks/scheduled/scan_library.py index 0d0d8c97cc..c5b2325148 100644 --- a/backend/tasks/scheduled/scan_library.py +++ b/backend/tasks/scheduled/scan_library.py @@ -19,7 +19,7 @@ ) from handler.scan_handler import MetadataSource, ScanType from logger.logger import log -from tasks.tasks import SCAN_LIBRARY_TASK_FUNC, PeriodicTask, TaskType +from tasks.tasks import PeriodicTask, TaskType class ScanLibraryTask(PeriodicTask): @@ -31,15 +31,14 @@ def __init__(self): enabled=ENABLE_SCHEDULED_RESCAN, manual_run=False, cron_string=SCHEDULED_RESCAN_CRON, - func=SCAN_LIBRARY_TASK_FUNC, + func="tasks.scheduled.scan_library.scan_library_task.run", ) async def run(self) -> dict[str, str]: scan_stats = ScanStats() if not ENABLE_SCHEDULED_RESCAN: - log.info("Scheduled library scan not enabled, unscheduling...") - self.unschedule() + log.info("Scheduled library scan not enabled, skipping...") return scan_stats.to_dict() source_mapping: dict[str, bool] = { diff --git a/backend/tasks/tasks.py b/backend/tasks/tasks.py index 55211d88be..22218b33b6 100644 --- a/backend/tasks/tasks.py +++ b/backend/tasks/tasks.py @@ -1,47 +1,39 @@ from abc import ABC, abstractmethod from enum import Enum -from itertools import chain -from typing import Any, Final +from typing import Any import httpx from rq import get_current_job -from rq.job import Job -from rq_scheduler import Scheduler from config import TASK_TIMEOUT -from exceptions.task_exceptions import SchedulerException -from handler.redis_handler import get_job_func_name, low_prio_queue +from exceptions.task_exceptions import TaskNotFoundException from logger.logger import log from utils.context import ctx_httpx_client -tasks_scheduler = Scheduler(queue=low_prio_queue, connection=low_prio_queue.connection) -# Lives here rather than in the task module so scan job discovery can recognise -# the scheduled rescan without importing it, which would close an import cycle. -SCAN_LIBRARY_TASK_FUNC: Final = "tasks.scheduled.scan_library.scan_library_task.run" +async def run_task_by_name(name: str, **kwargs: Any) -> Any: + """Run the task registered under ``name``. + Every scheduled and manually triggered task is enqueued through here, so a + job payload holds a name rather than a pickled task, and nothing in Redis + depends on where the code that runs it lives. -def drop_unreadable_scheduled_jobs() -> int: - """Remove scheduled jobs whose payload can no longer be deserialized. - - The scheduler reads a job's function name before taking it out of the - registry, so one unreadable job left behind by an older version crashes it - on every poll and nothing scheduled ever runs again. + Args: + name: The key the task is registered under. + kwargs: Forwarded to the task's ``run``. Returns: - int: How many jobs were dropped. + Whatever the task returns. """ - dropped = 0 - - for job in tasks_scheduler.get_jobs(): - if not isinstance(job, Job) or get_job_func_name(job): - continue + # Imported here because the registry imports every task module, and those + # modules import this one. + from tasks.registry import get_task - tasks_scheduler.cancel(job) - dropped += 1 - log.warning(f"Dropped scheduled job {job.id}, its payload cannot be read") + task = get_task(name) + if task is None: + raise TaskNotFoundException(name) - return dropped + return await task.run(**kwargs) def update_job_meta(metadata: dict[str, Any]) -> None: @@ -107,75 +99,12 @@ async def run(self, *args: Any, **kwargs: Any) -> Any: ... class PeriodicTask(Task, ABC): - """Base class for periodic tasks that can be scheduled.""" + """Base class for tasks the cron scheduler runs on a schedule.""" def __init__(self, *args: Any, func: str, **kwargs: Any): super().__init__(*args, **kwargs) self.func = func - def _get_existing_job(self) -> Job | None: - existing_jobs = chain(tasks_scheduler.get_jobs(), low_prio_queue.get_jobs()) - for job in existing_jobs: - if isinstance(job, Job) and get_job_func_name(job) == self.func: - return job - - return None - - def init(self) -> Job | None: - """Initialize the task by scheduling or unscheduling it based on its state. - - Returns the scheduled job if it was successfully scheduled, or None if it was already - scheduled or unscheduled. - """ - job = self._get_existing_job() - - if self.enabled and not job: - return self.schedule() - elif job and not self.enabled: - self.unschedule() - return None - return None - - def schedule(self) -> Job | None: - """Schedule the task if it is enabled and not already scheduled. - - Returns the scheduled job if successful, or None otherwise. - """ - if not self.enabled: - raise SchedulerException(f"Scheduled {self.description} is not enabled.") - - if self._get_existing_job(): - log.info(f"{self.description.capitalize()} is already scheduled.") - return None - - if self.cron_string: - return tasks_scheduler.cron( - self.cron_string, - func=self.func, - repeat=None, - timeout=self.timeout, - meta={ - "task_name": self.title, - "task_type": self.task_type.value, - }, - ) - - return None - - def unschedule(self) -> bool: - """Unschedule the task if it is currently scheduled. - - Returns whether the unscheduling was successful. - """ - job = self._get_existing_job() - if not job: - log.info(f"{self.description.capitalize()} is not scheduled.") - return False - - tasks_scheduler.cancel(job) - log.info(f"{self.description.capitalize()} unscheduled.") - return True - class RemoteFilePullTask(PeriodicTask, ABC): """Base class for tasks that pull files from a remote URL.""" @@ -186,8 +115,7 @@ def __init__(self, *args: Any, url: str, **kwargs: Any): async def run(self, force: bool = False) -> Any: if not self.enabled and not force: - log.info(f"Scheduled {self.description} not enabled, unscheduling...") - self.unschedule() + log.info(f"Scheduled {self.description} not enabled, skipping...") return None log.info(f"Scheduled {self.description} started...") diff --git a/backend/tests/endpoints/sockets/test_scan.py b/backend/tests/endpoints/sockets/test_scan.py index fe45780331..b4eafa02fd 100644 --- a/backend/tests/endpoints/sockets/test_scan.py +++ b/backend/tests/endpoints/sockets/test_scan.py @@ -35,6 +35,7 @@ from models.firmware import Firmware from models.platform import Platform from models.rom import Rom +from tasks.tasks import TaskType def test_scan_stats(): @@ -1684,21 +1685,37 @@ def test_url_contains_fs_path_and_name(self, handler: FSRomsHandler): SCAN_PLATFORMS_FUNC = "endpoints.sockets.scan.scan_platforms" +TASK_RUNNER_FUNC = "tasks.tasks.run_task_by_name" CLEANUP_FUNC = "tasks.scheduled.cleanup_zip_cache.cleanup_zip_cache_task.run" _job_ids = count() -def make_job(func_name: str, *, status=JobStatus.QUEUED, task_name: str | None = None): +def make_job( + func_name: str, + *, + status=JobStatus.QUEUED, + task_name: str | None = None, + task_type: TaskType | None = None, +): """An RQ job stub that scan job discovery will accept.""" job = MagicMock(spec=Job) job.id = f"job-{next(_job_ids)}" job.func_name = func_name job.get_status.return_value = status - job.meta = {"task_name": task_name} if task_name else {} + job.meta = {} + if task_name: + job.meta["task_name"] = task_name + if task_type: + job.meta["task_type"] = task_type return job +def make_task_job(**kwargs): + """A scan that runs through the task runner, as the scheduled rescan does.""" + return make_job(TASK_RUNNER_FUNC, task_type=TaskType.SCAN, **kwargs) + + def patch_scan_jobs( mocker, *, @@ -1710,8 +1727,7 @@ def patch_scan_jobs( ) -> MagicMock: """Point every place scan discovery looks at a fixed set of jobs. - Returns the patched scheduler cancel, the only thing that drops a delayed - scan out of the scheduler's registry. + Returns the patched scheduled-scan registry. """ worker = MagicMock() if worker_lost: @@ -1725,10 +1741,13 @@ def patch_scan_jobs( mocker.patch.object( scan_module.low_prio_queue, "get_jobs", return_value=list(low_queued) ) - mocker.patch.object( - scan_module.tasks_scheduler, "get_jobs", return_value=list(scheduled) - ) - return mocker.patch.object(scan_module.tasks_scheduler, "cancel") + + scheduled_jobs = list(scheduled) + registry = MagicMock() + registry.get_job_ids.return_value = [job.id for job in scheduled_jobs] + mocker.patch.object(scan_module, "_scheduled_scan_registry", return_value=registry) + mocker.patch.object(scan_module.Job, "fetch_many", return_value=scheduled_jobs) + return registry class TestScanConcurrency: @@ -1870,30 +1889,15 @@ async def test_refusal_reports_a_queued_scan_as_queued(self, mocker, emit): assert emit.await_args.args[1] == "Full Scan is already queued" async def test_refuses_when_the_scheduled_rescan_is_running(self, mocker, emit): - # The scheduled rescan runs scan_platforms from inside its own task, so - # the worker reports the task's name rather than the scan's. - patch_scan_jobs(mocker, running=make_job(scan_module.SCAN_LIBRARY_TASK_FUNC)) + # Every task runs through the same runner, so the scheduled rescan is + # only recognisable by the type its job carries. + patch_scan_jobs(mocker, running=make_task_job()) enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue") await scan_handler("sid", {"type": "quick"}) enqueue.assert_not_called() - async def test_standing_rescan_cron_entry_does_not_block(self, mocker, emit): - # The cron entry sits in the scheduler for as long as the periodic task - # is enabled. It is a schedule, not a scan waiting to run. - patch_scan_jobs( - mocker, - scheduled=[ - make_job(scan_module.SCAN_LIBRARY_TASK_FUNC, status=JobStatus.SCHEDULED) - ], - ) - enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue") - - await scan_handler("sid", {"type": "quick"}) - - enqueue.assert_called_once() - async def test_ignores_unrelated_jobs(self, mocker, emit): # Only scans block scans; a cleanup or metadata task must not. patch_scan_jobs( @@ -1997,7 +2001,7 @@ async def test_sets_stop_flag_for_running_scheduled_rescan( ): # The flag is the only channel an in-flight scan polls, so missing the # scheduled rescan here makes stopping it a silent no-op. - running = make_job(scan_module.SCAN_LIBRARY_TASK_FUNC) + running = make_task_job() patch_scan_jobs(mocker, running=running) await stop_scan_handler("sid") @@ -2020,28 +2024,14 @@ async def test_cancels_watcher_scans(self, mocker, emit, redis): # watcher's scan the moment the running one unwinds. low_queued = make_job(SCAN_PLATFORMS_FUNC) scheduled = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED) - scheduler_cancel = patch_scan_jobs( - mocker, low_queued=[low_queued], scheduled=[scheduled] - ) + patch_scan_jobs(mocker, low_queued=[low_queued], scheduled=[scheduled]) await stop_scan_handler("sid") low_queued.cancel.assert_called_once() + # Cancelling a delayed job takes it out of the scheduled registry too, + # so nothing releases it once the delay is up. scheduled.cancel.assert_called_once() - # Cancelling the job leaves its id in the scheduler, which queues it - # anyway once the delay is up. - scheduler_cancel.assert_called_once_with(scheduled) - - async def test_drops_a_scheduled_scan_that_was_already_cancelled( - self, mocker, emit, redis - ): - scheduled = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED) - scheduled.cancel.side_effect = InvalidJobOperation - scheduler_cancel = patch_scan_jobs(mocker, scheduled=[scheduled]) - - await stop_scan_handler("sid") - - scheduler_cancel.assert_called_once_with(scheduled) async def test_cancels_queued_scans_with_none_running(self, mocker, emit, redis): # Stopping a scan that has not been picked up yet must still drop it, @@ -2063,42 +2053,40 @@ async def test_no_scan_to_stop(self, mocker, emit, redis): class TestDropStaleScheduledScans: - """A recovered scheduler must not release a backlog of delayed scans.""" + """A worker that starts after downtime must not release a backlog.""" @staticmethod - def _scheduled(func_name: str, *, hours_late: float): - # The scheduler records due times as naive UTC. - due = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta( - hours=hours_late - ) - return (make_job(func_name, status=JobStatus.SCHEDULED), due) - - def _patch(self, mocker, jobs): - mocker.patch.object( - scan_module.tasks_scheduler, "get_jobs", return_value=list(jobs) - ) - return mocker.patch.object(scan_module.tasks_scheduler, "cancel") + def _due(hours_late: float): + return datetime.now(timezone.utc) - timedelta(hours=hours_late) def test_drops_a_scan_long_past_due(self, mocker): - job, due = self._scheduled(SCAN_PLATFORMS_FUNC, hours_late=2) - scheduler_cancel = self._patch(mocker, [(job, due)]) + job = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED) + registry = patch_scan_jobs(mocker, scheduled=[job]) + registry.get_scheduled_time.return_value = self._due(2) assert scan_module.drop_stale_scheduled_scans() == 1 - scheduler_cancel.assert_called_once_with(job) + job.cancel.assert_called_once() def test_keeps_a_scan_still_waiting_out_its_delay(self, mocker): - scheduler_cancel = self._patch( - mocker, [self._scheduled(SCAN_PLATFORMS_FUNC, hours_late=-1)] - ) + job = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED) + registry = patch_scan_jobs(mocker, scheduled=[job]) + registry.get_scheduled_time.return_value = self._due(-1) assert scan_module.drop_stale_scheduled_scans() == 0 - scheduler_cancel.assert_not_called() + job.cancel.assert_not_called() - def test_leaves_the_standing_rescan_cron_entry_alone(self, mocker): - # The cron entry reschedules itself, so a missed run is not a backlog. - scheduler_cancel = self._patch( - mocker, [self._scheduled(scan_module.SCAN_LIBRARY_TASK_FUNC, hours_late=2)] - ) + def test_ignores_a_scan_that_left_the_registry(self, mocker): + job = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED) + registry = patch_scan_jobs(mocker, scheduled=[job]) + registry.get_scheduled_time.side_effect = NoSuchJobError + + assert scan_module.drop_stale_scheduled_scans() == 0 + job.cancel.assert_not_called() + + def test_leaves_jobs_that_are_not_scans_alone(self, mocker): + job = make_job(CLEANUP_FUNC, status=JobStatus.SCHEDULED) + registry = patch_scan_jobs(mocker, scheduled=[job]) + registry.get_scheduled_time.return_value = self._due(2) assert scan_module.drop_stale_scheduled_scans() == 0 - scheduler_cancel.assert_not_called() + job.cancel.assert_not_called() diff --git a/backend/tests/endpoints/test_tasks.py b/backend/tests/endpoints/test_tasks.py index b81e6f52fe..96a6ff0b7e 100644 --- a/backend/tests/endpoints/test_tasks.py +++ b/backend/tests/endpoints/test_tasks.py @@ -81,44 +81,37 @@ class TestListTasks: @patch("endpoints.tasks.ENABLE_RESCAN_ON_FILESYSTEM_CHANGE", True) @patch("endpoints.tasks.RESCAN_ON_FILESYSTEM_CHANGE_DELAY", 5) @patch( - "endpoints.tasks.manual_tasks", - [ - { - "name": "test_manual", - "type": TaskType.CLEANUP, - "task": Mock( - spec=Task, - task_type=TaskType.CLEANUP, - title="Manual Task", - description="Manual task", - enabled=True, - manual_run=True, - can_run_manually=True, - timeout=300, - cron_string=None, - ), - } - ], + "endpoints.tasks.MANUAL_TASKS", + { + "test_manual": Mock( + spec=Task, + task_type=TaskType.CLEANUP, + title="Manual Task", + description="Manual task", + enabled=True, + manual_run=True, + can_run_manually=True, + timeout=300, + cron_string=None, + ), + }, ) + @patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ("test_scheduled",)) @patch( - "endpoints.tasks.scheduled_tasks", - [ - { - "name": "test_scheduled", - "type": TaskType.UPDATE, - "task": Mock( - spec=Task, - task_type=TaskType.UPDATE, - title="Scheduled Task", - description="Scheduled task", - enabled=True, - manual_run=False, - can_run_manually=False, - timeout=300, - cron_string="0 0 * * *", - ), - } - ], + "endpoints.tasks.SCHEDULED_TASKS", + { + "test_scheduled": Mock( + spec=Task, + task_type=TaskType.UPDATE, + title="Scheduled Task", + description="Scheduled task", + enabled=True, + manual_run=False, + can_run_manually=False, + timeout=300, + cron_string="0 0 * * *", + ), + }, ) def test_list_tasks_success(self, client, access_token): """Test successful listing of all tasks""" @@ -166,8 +159,8 @@ def test_list_tasks_success(self, client, access_token): @patch("endpoints.tasks.ENABLE_RESCAN_ON_FILESYSTEM_CHANGE", False) @patch("endpoints.tasks.RESCAN_ON_FILESYSTEM_CHANGE_DELAY", 10) - @patch("endpoints.tasks.manual_tasks", []) - @patch("endpoints.tasks.scheduled_tasks", []) + @patch("endpoints.tasks.MANUAL_TASKS", {}) + @patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ()) def test_list_tasks_empty(self, client, access_token): """Test listing tasks when no tasks are available""" response = client.get( @@ -216,26 +209,22 @@ class TestRunSingleTask: @patch("endpoints.tasks.low_prio_queue.enqueue", return_value=create_mock_job()) @patch( - "endpoints.tasks.manual_tasks", - [ - { - "name": "test_task", - "type": TaskType.CLEANUP, - "task": Mock( - spec=Task, - task_type=TaskType.CLEANUP, - title="Test Task", - description="Test Description", - enabled=True, - manual_run=True, - can_run_manually=True, - timeout=300, - run=Mock(), - ), - } - ], + "endpoints.tasks.MANUAL_TASKS", + { + "test_task": Mock( + spec=Task, + task_type=TaskType.CLEANUP, + title="Test Task", + description="Test Description", + enabled=True, + manual_run=True, + can_run_manually=True, + timeout=300, + run=Mock(), + ), + }, ) - @patch("endpoints.tasks.scheduled_tasks", []) + @patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ()) def test_run_single_task_success(self, mock_queue, client, access_token): """Test successful running of a single task""" response = client.post( @@ -254,8 +243,8 @@ def test_run_single_task_success(self, mock_queue, client, access_token): mock_queue.assert_called_once() - @patch("endpoints.tasks.manual_tasks", []) - @patch("endpoints.tasks.scheduled_tasks", []) + @patch("endpoints.tasks.MANUAL_TASKS", {}) + @patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ()) def test_run_single_task_not_found(self, client, access_token): """Test running a non-existent task""" response = client.post( @@ -269,26 +258,22 @@ def test_run_single_task_not_found(self, client, access_token): @patch("endpoints.tasks.low_prio_queue") @patch( - "endpoints.tasks.manual_tasks", - [ - { - "name": "disabled_task", - "type": TaskType.CLEANUP, - "task": Mock( - spec=Task, - task_type=TaskType.CLEANUP, - title="Disabled Task", - description="Disabled Description", - enabled=False, - manual_run=True, - can_run_manually=False, - timeout=300, - run=Mock(), - ), - } - ], + "endpoints.tasks.MANUAL_TASKS", + { + "disabled_task": Mock( + spec=Task, + task_type=TaskType.CLEANUP, + title="Disabled Task", + description="Disabled Description", + enabled=False, + manual_run=True, + can_run_manually=False, + timeout=300, + run=Mock(), + ), + }, ) - @patch("endpoints.tasks.scheduled_tasks", []) + @patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ()) def test_run_single_task_disabled(self, mock_queue, client, access_token): """Test running a disabled task""" response = client.post( @@ -302,26 +287,22 @@ def test_run_single_task_disabled(self, mock_queue, client, access_token): @patch("endpoints.tasks.low_prio_queue") @patch( - "endpoints.tasks.manual_tasks", - [ - { - "name": "non_manual_task", - "type": TaskType.CLEANUP, - "task": Mock( - spec=Task, - task_type=TaskType.CLEANUP, - title="Non-Manual Task", - description="Non-Manual Description", - enabled=True, - manual_run=False, - can_run_manually=False, - timeout=300, - run=Mock(), - ), - } - ], + "endpoints.tasks.MANUAL_TASKS", + { + "non_manual_task": Mock( + spec=Task, + task_type=TaskType.CLEANUP, + title="Non-Manual Task", + description="Non-Manual Description", + enabled=True, + manual_run=False, + can_run_manually=False, + timeout=300, + run=Mock(), + ), + }, ) - @patch("endpoints.tasks.scheduled_tasks", []) + @patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ()) def test_run_single_task_non_manual(self, mock_queue, client, access_token): """Test running a task that cannot be run manually""" response = client.post( @@ -515,25 +496,21 @@ def test_build_task_info_structure( } with patch( - "endpoints.tasks.manual_tasks", - [ - { - "name": "test_task", - "type": TaskType.CLEANUP, - "task": Mock( - spec=Task, - title="Test Task", - description="Test Description", - enabled=True, - manual_run=True, - can_run_manually=True, - timeout=300, - cron_string="0 0 * * *", - ), - } - ], + "endpoints.tasks.MANUAL_TASKS", + { + "test_task": Mock( + spec=Task, + title="Test Task", + description="Test Description", + enabled=True, + manual_run=True, + can_run_manually=True, + timeout=300, + cron_string="0 0 * * *", + ), + }, ): - with patch("endpoints.tasks.scheduled_tasks", []): + with patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ()): response = client.get( "/api/tasks", headers={"Authorization": f"Bearer {access_token}"} ) @@ -561,26 +538,22 @@ def test_full_workflow(self, mock_queue, client, access_token): # Then run a specific task (if any exist) with patch( - "endpoints.tasks.manual_tasks", - [ - { - "name": "workflow_task", - "type": TaskType.CLEANUP, - "task": Mock( - spec=Task, - task_type=TaskType.CLEANUP, - title="Workflow Task", - description="Workflow Description", - enabled=True, - manual_run=True, - can_run_manually=True, - timeout=300, - run=Mock(), - ), - } - ], + "endpoints.tasks.MANUAL_TASKS", + { + "workflow_task": Mock( + spec=Task, + task_type=TaskType.CLEANUP, + title="Workflow Task", + description="Workflow Description", + enabled=True, + manual_run=True, + can_run_manually=True, + timeout=300, + run=Mock(), + ), + }, ): - with patch("endpoints.tasks.scheduled_tasks", []): + with patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ()): run_response = client.post( "/api/tasks/run/workflow_task", headers={"Authorization": f"Bearer {access_token}"}, diff --git a/backend/tests/tasks/test_cleanup_orphaned_resources.py b/backend/tests/tasks/test_cleanup_orphaned_resources.py index 0d1d2322d1..bbab91496d 100644 --- a/backend/tests/tasks/test_cleanup_orphaned_resources.py +++ b/backend/tests/tasks/test_cleanup_orphaned_resources.py @@ -41,28 +41,6 @@ def test_cron_string_follows_config_override(self): ): assert CleanupOrphanedResourcesTask().cron_string == "30 2 * * *" - def test_init_unschedules_when_no_cron(self, task): - task.cron_string = None - - with patch.object(task, "unschedule") as mock_unschedule: - assert task.init() is None - mock_unschedule.assert_called_once() - - def test_init_schedules_when_enabled(self, task): - task.enabled = True - task.cron_string = "0 5 * * *" - - with patch.object(task, "_get_existing_job", return_value=None): - with patch.object(task, "schedule") as mock_schedule: - task.init() - mock_schedule.assert_called_once() - - def test_init_does_not_schedule_when_disabled(self, task): - with patch.object(task, "_get_existing_job", return_value=None): - with patch.object(task, "schedule") as mock_schedule: - assert task.init() is None - mock_schedule.assert_not_called() - class TestCleanupOrphanedResourcesRun: @pytest.fixture diff --git a/backend/tests/tasks/test_cleanup_zip_cache.py b/backend/tests/tasks/test_cleanup_zip_cache.py index 75bb9019be..c5cdf637e1 100644 --- a/backend/tests/tasks/test_cleanup_zip_cache.py +++ b/backend/tests/tasks/test_cleanup_zip_cache.py @@ -1,5 +1,3 @@ -from unittest.mock import MagicMock - from tasks.scheduled.cleanup_zip_cache import CleanupZipCacheTask @@ -19,10 +17,9 @@ async def test_run_calls_cleanup(self, mocker): await task.run() mock_cleanup.assert_called_once_with() - async def test_run_disabled_unschedules(self, mocker): + async def test_run_disabled_skips_the_cleanup(self, mocker): task = CleanupZipCacheTask() task.enabled = False - mocker.patch.object(task, "unschedule", MagicMock()) mock_cleanup = mocker.patch( "tasks.scheduled.cleanup_zip_cache.cleanup_stale_zips", ) diff --git a/backend/tests/tasks/test_cron_config.py b/backend/tests/tasks/test_cron_config.py new file mode 100644 index 0000000000..54df7d3617 --- /dev/null +++ b/backend/tests/tasks/test_cron_config.py @@ -0,0 +1,66 @@ +import importlib + +import pytest + +from tasks import cron_config +from tasks.registry import SCHEDULED_TASKS +from tasks.tasks import run_task_by_name + + +@pytest.fixture +def registered(mocker): + """Reload the config and report what it registered with the scheduler.""" + + def _reload(tasks): + register = mocker.patch("rq.cron.register") + mocker.patch.dict(cron_config.SCHEDULED_TASKS, tasks, clear=True) + importlib.reload(cron_config) + return register + + return _reload + + +def _task(mocker, *, enabled=True, cron_string="0 4 * * *"): + return mocker.MagicMock( + enabled=enabled, + cron_string=cron_string, + timeout=100, + title="Test Task", + description="test task", + task_type=mocker.MagicMock(value="cleanup"), + ) + + +class TestCronConfig: + """The cron process registers what this module declares, and nothing else.""" + + def test_registers_an_enabled_task_by_name(self, mocker, registered): + register = registered({"test_task": _task(mocker)}) + + register.assert_called_once() + args, kwargs = register.call_args + assert args[0] is run_task_by_name + assert kwargs["kwargs"] == {"name": "test_task"} + assert kwargs["cron"] == "0 4 * * *" + + def test_skips_a_disabled_task(self, mocker, registered): + assert registered({"off": _task(mocker, enabled=False)}).call_count == 0 + + def test_skips_a_task_with_no_cron_string(self, mocker, registered): + assert registered({"no_cron": _task(mocker, cron_string="")}).call_count == 0 + + def test_registers_the_real_schedule(self, mocker): + # Guards the actual catalog: every enabled task with a cron string is + # registered, because nothing else schedules them any more. + register = mocker.patch("rq.cron.register") + importlib.reload(cron_config) + + expected = [ + name + for name, task in SCHEDULED_TASKS.items() + if task.enabled and task.cron_string + ] + registered_names = [ + call.kwargs["kwargs"]["name"] for call in register.call_args_list + ] + assert registered_names == expected diff --git a/backend/tests/tasks/test_prevent_requeue.py b/backend/tests/tasks/test_prevent_requeue.py deleted file mode 100644 index a7b7676200..0000000000 --- a/backend/tests/tasks/test_prevent_requeue.py +++ /dev/null @@ -1,48 +0,0 @@ -import unittest -from unittest.mock import MagicMock, patch - -from rq.job import Job - -from tasks.tasks import PeriodicTask, TaskType - - -def dummy_task(): - pass - - -class DummyTask(PeriodicTask): - async def run(self, *args, **kwargs): - pass - - -class TestPreventRequeue(unittest.TestCase): - @patch("tasks.tasks.tasks_scheduler") - @patch("tasks.tasks.low_prio_queue") - def test_task_not_scheduled_if_in_queue( - self, mock_low_prio_queue, mock_tasks_scheduler - ): - for method_name in ["init", "schedule"]: - with self.subTest(method=method_name): - task = DummyTask( - title="Test Task", - description="A test task", - task_type=TaskType.GENERIC, - enabled=True, - manual_run=False, - cron_string="* * * * *", - func="backend.tests.tasks.test_prevent_requeue.dummy_task", - ) - - mock_job = MagicMock(spec=Job) - mock_job.func_name = ( - "backend.tests.tasks.test_prevent_requeue.dummy_task" - ) - - mock_low_prio_queue.get_jobs.return_value = [mock_job] - mock_tasks_scheduler.get_jobs.return_value = [] - - method_to_call = getattr(task, method_name) - result = method_to_call() - - self.assertIsNone(result) - mock_tasks_scheduler.cron.assert_not_called() diff --git a/backend/tests/tasks/test_registry.py b/backend/tests/tasks/test_registry.py new file mode 100644 index 0000000000..4a3f98b6e7 --- /dev/null +++ b/backend/tests/tasks/test_registry.py @@ -0,0 +1,22 @@ +import pytest + +from tasks.registry import MANUAL_TASKS, SCHEDULED_TASKS, get_task +from tasks.tasks import PeriodicTask + + +class TestRegistry: + """A job payload carries only a name, so the catalog has to resolve it.""" + + @pytest.mark.parametrize("name", sorted(SCHEDULED_TASKS | MANUAL_TASKS)) + def test_every_name_resolves(self, name: str): + assert get_task(name) is not None + + def test_a_name_is_never_registered_twice(self): + assert not SCHEDULED_TASKS.keys() & MANUAL_TASKS.keys() + + def test_scheduled_tasks_can_be_scheduled(self): + for name, task in SCHEDULED_TASKS.items(): + assert isinstance(task, PeriodicTask), name + + def test_an_unknown_name_resolves_to_nothing(self): + assert get_task("no_such_task") is None diff --git a/backend/tests/tasks/test_scan_library.py b/backend/tests/tasks/test_scan_library.py index 80fdfe6bea..181a9759a7 100644 --- a/backend/tests/tasks/test_scan_library.py +++ b/backend/tests/tasks/test_scan_library.py @@ -68,14 +68,12 @@ async def test_run_disabled(self, task, mocker): "tasks.scheduled.scan_library.scan_platforms" ) mock_log = mocker.patch("tasks.scheduled.scan_library.log") - task.unschedule = MagicMock() await task.run() mock_log.info.assert_called_once_with( - "Scheduled library scan not enabled, unscheduling..." + "Scheduled library scan not enabled, skipping..." ) - task.unschedule.assert_called_once() mock_scan_platforms.assert_not_called() def test_task_instance(self): diff --git a/backend/tests/tasks/test_tasks.py b/backend/tests/tasks/test_tasks.py index 4dd3ea0ef8..b471cf8093 100644 --- a/backend/tests/tasks/test_tasks.py +++ b/backend/tests/tasks/test_tasks.py @@ -1,18 +1,10 @@ -from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest -from rq.exceptions import DeserializationError -from rq.job import Job -from exceptions.task_exceptions import SchedulerException -from tasks.tasks import ( - PeriodicTask, - RemoteFilePullTask, - TaskType, - drop_unreadable_scheduled_jobs, - tasks_scheduler, -) +from exceptions.task_exceptions import TaskNotFoundException +from tasks.tasks import PeriodicTask, RemoteFilePullTask, TaskType, run_task_by_name class ConcretePeriodicTask(PeriodicTask): @@ -53,162 +45,6 @@ def test_init(self, task): assert task.enabled is True assert task.cron_string == "0 0 * * *" - @patch.object(tasks_scheduler, "get_jobs") - def test_get_existing_job_found(self, mock_get_jobs, task): - """Test finding an existing job""" - mock_job = MagicMock(spec=Job) - mock_job.func_name = "test.function" - mock_get_jobs.return_value = [mock_job] - - result = task._get_existing_job() - assert result == mock_job - - @patch.object(tasks_scheduler, "get_jobs") - def test_get_existing_job_not_found(self, mock_get_jobs, task): - """Test when no existing job is found""" - mock_job = MagicMock(spec=Job) - mock_job.func_name = "other.function" - mock_get_jobs.return_value = [mock_job] - - result = task._get_existing_job() - assert result is None - - @patch.object(tasks_scheduler, "get_jobs") - def test_get_existing_job_empty_list(self, mock_get_jobs, task): - """Test when no jobs exist""" - mock_get_jobs.return_value = [] - - result = task._get_existing_job() - assert result is None - - @patch.object(ConcretePeriodicTask, "_get_existing_job") - @patch.object(ConcretePeriodicTask, "schedule") - @patch.object(ConcretePeriodicTask, "unschedule") - def test_init_enabled_no_existing_job( - self, mock_unschedule, mock_schedule, mock_get_existing_job, task - ): - """Test init when task is enabled and no existing job""" - mock_job = MagicMock(spec=Job) - mock_get_existing_job.return_value = None - mock_schedule.return_value = mock_job - - result = task.init() - - mock_schedule.assert_called_once() - mock_unschedule.assert_not_called() - assert result == mock_job - - @patch.object(ConcretePeriodicTask, "_get_existing_job") - @patch.object(ConcretePeriodicTask, "schedule") - @patch.object(ConcretePeriodicTask, "unschedule") - def test_init_disabled_with_existing_job( - self, mock_unschedule, mock_schedule, mock_get_existing_job, disabled_task - ): - """Test init when task is disabled but has existing job""" - mock_job = MagicMock(spec=Job) - mock_get_existing_job.return_value = mock_job - mock_unschedule.return_value = None - - result = disabled_task.init() - - mock_unschedule.assert_called_once() - mock_schedule.assert_not_called() - assert result is None - - @patch.object(ConcretePeriodicTask, "_get_existing_job") - @patch.object(ConcretePeriodicTask, "schedule") - @patch.object(ConcretePeriodicTask, "unschedule") - def test_init_enabled_with_existing_job( - self, mock_unschedule, mock_schedule, mock_get_existing_job, task - ): - """Test init when task is enabled and job already exists""" - mock_job = MagicMock(spec=Job) - mock_get_existing_job.return_value = mock_job - - result = task.init() - - mock_schedule.assert_not_called() - mock_unschedule.assert_not_called() - assert result is None # Should do nothing - - @patch.object(ConcretePeriodicTask, "_get_existing_job") - @patch.object(tasks_scheduler, "cron") - def test_schedule_success(self, mock_cron, mock_get_existing_job, task): - """Test successful scheduling""" - mock_job = MagicMock(spec=Job) - mock_get_existing_job.return_value = None - mock_cron.return_value = mock_job - - result = task.schedule() - - mock_cron.assert_called_once_with( - "0 0 * * *", - func="test.function", - repeat=None, - timeout=5 * 60, - meta={"task_name": "Test Task", "task_type": "generic"}, - ) - assert result == mock_job - - def test_schedule_not_enabled(self, disabled_task): - """Test scheduling when task is not enabled""" - with pytest.raises( - SchedulerException, match="Scheduled disabled task is not enabled." - ): - disabled_task.schedule() - - @patch.object(ConcretePeriodicTask, "_get_existing_job") - @patch("tasks.tasks.log") - def test_schedule_already_scheduled(self, mock_log, mock_get_existing_job, task): - """Test scheduling when job already exists""" - mock_job = MagicMock() - mock_get_existing_job.return_value = mock_job - - result = task.schedule() - - mock_log.info.assert_called_once_with("Test task is already scheduled.") - assert result is None - - def test_schedule_no_cron_string(self): - """Test scheduling with no cron string""" - task = ConcretePeriodicTask( - func="test.function", - title="Test Task", - task_type=TaskType.GENERIC, - description="test task", - enabled=True, - cron_string=None, - ) - - with patch.object(task, "_get_existing_job", return_value=None): - result = task.schedule() - assert result is None - - @patch.object(ConcretePeriodicTask, "_get_existing_job") - @patch.object(tasks_scheduler, "cancel") - @patch("tasks.tasks.log") - def test_unschedule_success( - self, mock_log, mock_cancel, mock_get_existing_job, task - ): - """Test successful unscheduling""" - mock_job = MagicMock(spec=Job) - mock_get_existing_job.return_value = mock_job - - task.unschedule() - - mock_cancel.assert_called_once_with(mock_job) - mock_log.info.assert_called_once_with("Test task unscheduled.") - - @patch.object(ConcretePeriodicTask, "_get_existing_job") - @patch("tasks.tasks.log") - def test_unschedule_not_scheduled(self, mock_log, mock_get_existing_job, task): - """Test unscheduling when no job exists""" - mock_get_existing_job.return_value = None - - task.unschedule() - - mock_log.info.assert_called_once_with("Test task is not scheduled.") - async def test_run_abstract_method(self, task): """Test that run method works in concrete implementation""" result = await task.run() @@ -303,18 +139,14 @@ async def test_run_response_error(self, mock_log, mock_ctx_httpx_client, task): mock_log.error.assert_any_call(http_error) assert result is None - @patch.object(RemoteFilePullTask, "unschedule") @patch("tasks.tasks.log") - async def test_run_disabled_not_forced( - self, mock_log, mock_unschedule, disabled_task - ): + async def test_run_disabled_not_forced(self, mock_log, disabled_task): """Test run when task is disabled and not forced""" result = await disabled_task.run(force=False) mock_log.info.assert_called_once_with( - "Scheduled disabled remote task not enabled, unscheduling..." + "Scheduled disabled remote task not enabled, skipping..." ) - mock_unschedule.assert_called_once() assert result is None @patch("tasks.tasks.ctx_httpx_client") @@ -331,32 +163,28 @@ async def test_run_disabled_but_forced(self, mock_ctx_httpx_client, disabled_tas assert result == b"forced content" -class TestDropUnreadableScheduledJobs: - """One unreadable job crashes the scheduler on every poll until it is gone.""" +class TestRunTaskByName: + """Jobs carry a task's name, so the runner has to resolve it.""" + + async def test_runs_the_registered_task(self, mocker): + task = MagicMock() + task.run = AsyncMock(return_value="ran") + mocker.patch("tasks.registry.get_task", return_value=task) + + assert await run_task_by_name("some_task") == "ran" + task.run.assert_awaited_once_with() - @staticmethod - def _job(func_name: str | None): - job = MagicMock(spec=Job) - job.id = "job-1" - if func_name is None: - type(job).func_name = PropertyMock(side_effect=DeserializationError) - else: - job.func_name = func_name - return job + async def test_forwards_keyword_arguments(self, mocker): + task = MagicMock() + task.run = AsyncMock(return_value=None) + mocker.patch("tasks.registry.get_task", return_value=task) - @patch.object(tasks_scheduler, "cancel") - @patch.object(tasks_scheduler, "get_jobs") - def test_drops_a_job_whose_payload_cannot_be_read(self, get_jobs, cancel): - job = self._job(None) - get_jobs.return_value = [job] + await run_task_by_name("some_task", force=True) - assert drop_unreadable_scheduled_jobs() == 1 - cancel.assert_called_once_with(job) + task.run.assert_awaited_once_with(force=True) - @patch.object(tasks_scheduler, "cancel") - @patch.object(tasks_scheduler, "get_jobs") - def test_leaves_readable_jobs_scheduled(self, get_jobs, cancel): - get_jobs.return_value = [self._job("test.function")] + async def test_raises_for_a_name_that_is_not_registered(self, mocker): + mocker.patch("tasks.registry.get_task", return_value=None) - assert drop_unreadable_scheduled_jobs() == 0 - cancel.assert_not_called() + with pytest.raises(TaskNotFoundException, match="some_task"): + await run_task_by_name("some_task") diff --git a/backend/watcher.py b/backend/watcher.py index df0947f154..cff5f40f89 100644 --- a/backend/watcher.py +++ b/backend/watcher.py @@ -8,8 +8,6 @@ import sentry_sdk from opentelemetry import trace -from rq import Worker -from rq.job import Job, JobStatus from config import ( ENABLE_RESCAN_ON_FILESYSTEM_CHANGE, @@ -20,7 +18,10 @@ TASK_RESULT_TTL, ) from config.config_manager import config_manager as cm -from endpoints.sockets.scan import scan_platforms +from endpoints.sockets.scan import ( + get_pending_scan_jobs, + scan_platforms, +) from handler.database import db_platform_handler from handler.metadata import ( meta_flashpoint_handler, @@ -36,12 +37,12 @@ meta_ss_handler, meta_tgdb_handler, ) -from handler.redis_handler import get_job_func_name, low_prio_queue, redis_client +from handler.redis_handler import get_job_kwargs, low_prio_queue from handler.scan_handler import MetadataSource, ScanType from logger.formatter import CYAN from logger.formatter import highlight as hl from logger.logger import log -from tasks.tasks import TaskType, tasks_scheduler +from tasks.tasks import TaskType from utils import get_version sentry_sdk.init( @@ -70,50 +71,6 @@ class EventType(enum.StrEnum): Change = tuple[EventType, str] -def get_pending_scan_jobs() -> list[Job]: - """Get all pending scan jobs (scheduled, queued, or running) for scan_platforms function. - - Returns: - list[Job]: List of pending scan jobs that are not completed or failed - """ - pending_jobs = [] - - # Get jobs from the scheduler (delayed/scheduled jobs) - scheduled_jobs = tasks_scheduler.get_jobs() - for job in scheduled_jobs: - if ( - isinstance(job, Job) - and get_job_func_name(job) == "endpoints.sockets.scan.scan_platforms" - and job.get_status() - in [JobStatus.SCHEDULED, JobStatus.QUEUED, JobStatus.STARTED] - ): - pending_jobs.append(job) - - # Get jobs from the queue (immediate jobs) - queue_jobs = low_prio_queue.get_jobs() - for job in queue_jobs: - if ( - isinstance(job, Job) - and get_job_func_name(job) == "endpoints.sockets.scan.scan_platforms" - and job.get_status() in [JobStatus.QUEUED, JobStatus.STARTED] - ): - pending_jobs.append(job) - - # Get currently running jobs from workers - workers = Worker.all(connection=redis_client) - for worker in workers: - current_job = worker.get_current_job() - if ( - current_job - and get_job_func_name(current_job) - == "endpoints.sockets.scan.scan_platforms" - and current_job.get_status() == JobStatus.STARTED - ): - pending_jobs.append(current_job) - - return pending_jobs - - def process_changes(changes: Sequence[Change]) -> None: if not ENABLE_RESCAN_ON_FILESYSTEM_CHANGE: return @@ -194,15 +151,16 @@ def _is_excluded(path: str) -> bool: log.warning("No metadata sources enabled, skipping rescan") return - # Get currently pending scan jobs (scheduled, queued, or running) - pending_jobs = get_pending_scan_jobs() - - # If a full rescan is already scheduled, skip further processing - full_rescan_jobs = [ - job for job in pending_jobs if job.args and job.args[0] == [] + # The platforms each pending scan covers. A scan with no platform ids + # covers the whole library, which is also what a task-driven scan does. + pending_scopes = [ + kwargs.get("platform_ids") or [] + for job in get_pending_scan_jobs() + if (kwargs := get_job_kwargs(job)) is not None ] - if full_rescan_jobs: - log.info(f"Full rescan already scheduled ({len(full_rescan_jobs)} job(s))") + + if any(not scope for scope in pending_scopes): + log.info("Full rescan already pending") return time_delta = timedelta(minutes=RESCAN_ON_FILESYSTEM_CHANGE_DELAY) @@ -211,16 +169,16 @@ def _is_excluded(path: str) -> bool: # Any change to a platform directory should trigger a full rescan if changes_platform_directory: log.info(f"Platform directory changed, {rescan_in_msg}") - tasks_scheduler.enqueue_in( + low_prio_queue.enqueue_in( time_delta, scan_platforms, platform_ids=[], metadata_sources=metadata_sources, scan_type=ScanType.UPDATE, - timeout=SCAN_TIMEOUT, - job_result_ttl=TASK_RESULT_TTL, + job_timeout=SCAN_TIMEOUT, + result_ttl=TASK_RESULT_TTL, meta={ - "task_name": "Unidentified Scan", + "task_name": "Update Scan", "task_type": TaskType.SCAN, }, ) @@ -233,27 +191,19 @@ def _is_excluded(path: str) -> bool: if not db_platform: continue - # Skip if a scan is already scheduled for this platform - platform_scan_jobs = [ - job - for job in pending_jobs - if job.args and db_platform.id in job.args[0] - ] - if platform_scan_jobs: - log.info( - f"Scan already scheduled for {hl(fs_slug)} ({len(platform_scan_jobs)} job(s))" - ) + if any(db_platform.id in scope for scope in pending_scopes): + log.info(f"Scan already pending for {hl(fs_slug)}") continue log.info(f"Change detected in {hl(fs_slug)} folder, {rescan_in_msg}") - tasks_scheduler.enqueue_in( + low_prio_queue.enqueue_in( time_delta, scan_platforms, platform_ids=[db_platform.id], metadata_sources=metadata_sources, scan_type=ScanType.QUICK, - timeout=SCAN_TIMEOUT, - job_result_ttl=TASK_RESULT_TTL, + job_timeout=SCAN_TIMEOUT, + result_ttl=TASK_RESULT_TTL, meta={ "task_name": "Quick Scan", "task_type": TaskType.SCAN, diff --git a/docker/init_scripts/init b/docker/init_scripts/init index bc888149ae..9229398633 100755 --- a/docker/init_scripts/init +++ b/docker/init_scripts/init @@ -221,42 +221,48 @@ start_bin_valkey-server() { error_log "Internal valkey did not become ready after $((max_retries * 500))ms" } -# Commands to start RQ scheduler -start_bin_rq_scheduler() { - info_log "Starting RQ scheduler" - - RQ_REDIS_HOST=${REDIS_HOST:-127.0.0.1} \ - RQ_REDIS_PORT=${REDIS_PORT:-6379} \ - RQ_REDIS_USERNAME=${REDIS_USERNAME:-""} \ - RQ_REDIS_PASSWORD=${REDIS_PASSWORD:-""} \ - RQ_REDIS_DB=${REDIS_DB:-0} \ - RQ_REDIS_SSL=${REDIS_SSL:-0} \ - rqscheduler \ +# The URL carries the password, so it goes through RQ_REDIS_URL rather than +# --url, which would put it on a world-readable command line. +build_redis_url() { + if [[ -n ${REDIS_PASSWORD-} ]]; then + echo "redis${REDIS_SSL:+s}://${REDIS_USERNAME-}:${REDIS_PASSWORD}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + elif [[ -n ${REDIS_USERNAME-} ]]; then + echo "redis${REDIS_SSL:+s}://${REDIS_USERNAME}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + else + echo "redis${REDIS_SSL:+s}://${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + fi +} + +# Commands to start the RQ cron scheduler, which runs the periodic tasks listed +# in the cron config. Delayed jobs are released by the worker itself. +start_bin_rq_cron() { + info_log "Starting RQ cron scheduler" + + local redis_url + redis_url="$(build_redis_url)" + + PYTHONPATH="/backend:${PYTHONPATH-}" \ + RQ_REDIS_URL="${redis_url}" \ + rq cron \ --path /backend \ - --pid /tmp/rq_scheduler.pid & + --logging-level "${LOGLEVEL}" \ + tasks.cron_config & + echo "$!" >/tmp/rq_cron.pid } # Commands to start RQ worker start_bin_rq_worker() { info_log "Starting RQ worker" - # Build Redis URL properly local redis_url - if [[ -n ${REDIS_PASSWORD-} ]]; then - redis_url="redis${REDIS_SSL:+s}://${REDIS_USERNAME-}:${REDIS_PASSWORD}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" - elif [[ -n ${REDIS_USERNAME-} ]]; then - redis_url="redis${REDIS_SSL:+s}://${REDIS_USERNAME}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" - else - redis_url="redis${REDIS_SSL:+s}://${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" - fi + redis_url="$(build_redis_url)" # Set PYTHONPATH so RQ can find the tasks module. - # The connection URL goes through RQ_REDIS_URL rather than --url, so the - # embedded password stays out of the worker's world-readable command line. # Use a worker class that drops the noisy per-sweep "cleaning registries for # queue" log line. The maintenance interval keeps its default (~10 min) so - # orphaned STARTED jobs and stale workers are still pruned promptly, which the - # watcher's Worker.all() scan dedupe relies on. + # orphaned STARTED jobs and stale workers are still pruned promptly. + # --with-scheduler releases delayed jobs, which is how the watcher's rescans + # wait out their delay. PYTHONPATH="/backend:${PYTHONPATH-}" \ RQ_REDIS_URL="${redis_url}" \ rq worker \ @@ -265,6 +271,7 @@ start_bin_rq_worker() { --pid /tmp/rq_worker.pid \ --results-ttl "${TASK_RESULT_TTL:-86400}" \ --logging_level "${LOGLEVEL}" \ + --with-scheduler \ high default low & } @@ -325,7 +332,7 @@ stop_process_pid() { shutdown() { # shutdown in reverse order stop_process_pid rq_worker - stop_process_pid rq_scheduler + stop_process_pid rq_cron stop_process_pid sync_watcher stop_process_pid watcher stop_process_pid nginx @@ -380,8 +387,8 @@ run_startup while ! ((exited)); do watchdog_process_pid gunicorn - # always run the scheduler - watchdog_process_pid rq_scheduler + # always run the cron scheduler + watchdog_process_pid rq_cron watchdog_process_pid rq_worker diff --git a/docs/BACKEND_ARCHITECTURE.md b/docs/BACKEND_ARCHITECTURE.md index 854b907ef1..644dd51751 100644 --- a/docs/BACKEND_ARCHITECTURE.md +++ b/docs/BACKEND_ARCHITECTURE.md @@ -298,7 +298,9 @@ backend/ │ └── known_bios_files.json # Verified BIOS hashes │ ├── tasks/ # Background job system -│ ├── tasks.py # Base Task, PeriodicTask classes +│ ├── tasks.py # Base Task, PeriodicTask, run_task_by_name +│ ├── registry.py # Name -> task catalog, the API and cron address +│ ├── cron_config.py # Schedule the `rq cron` process loads │ ├── scheduled/ # Cron-scheduled tasks │ │ ├── scan_library.py # Nightly library rescan │ │ ├── sync_retroachievements_progress.py # Pull RA user progress @@ -352,13 +354,7 @@ backend/ ```text 1. alembic upgrade head # Run database migrations 2. startup.main() # Async startup tasks - ├── Initialize scheduled jobs (RQ Scheduler) - │ ├── cleanup_netplay - │ ├── scan_library (if ENABLE_SCHEDULED_RESCAN) - │ ├── update_switch_titledb - │ ├── update_launchbox_metadata - │ ├── convert_images_to_webp - │ └── sync_retroachievements_progress + ├── Clear stale delayed scans and legacy scheduler keys └── Load fixture caches into Redis ├── mame_index.json ├── scummvm_index.json @@ -1382,7 +1378,13 @@ Redis-backed for horizontal scaling across multiple server instances. ### Scheduled Tasks -Configured via environment variables and managed by RQ Scheduler: +Declared in `tasks/registry.py` and registered with RQ's cron scheduler by +`tasks/cron_config.py`, which the `rq cron` process loads at start. A task is +registered only when it is enabled and has a cron string, so turning one off is +a restart rather than an unschedule. Delayed jobs, which is how the filesystem +watcher defers a rescan, are released by the worker itself (`--with-scheduler`). + +Toggled via environment variables: | Task | Env Toggle | Default Cron | Description | | --------------------------------- | -------------------------------------------------- | ------------------ | ---------------------- | diff --git a/entrypoint.sh b/entrypoint.sh index e815271bb6..780939091d 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -47,18 +47,6 @@ else uv run python main.py & fi -echo "Starting RQ scheduler..." -RQ_REDIS_HOST=${REDIS_HOST:-127.0.0.1} \ - RQ_REDIS_PORT=${REDIS_PORT:-6379} \ - RQ_REDIS_USERNAME=${REDIS_USERNAME:-""} \ - RQ_REDIS_PASSWORD=${REDIS_PASSWORD:-""} \ - RQ_REDIS_DB=${REDIS_DB:-0} \ - RQ_REDIS_SSL=${REDIS_SSL:-0} \ - rqscheduler \ - --path /app/backend \ - --pid /tmp/rq_scheduler.pid & - -echo "Starting RQ worker..." # Build Redis URL properly if [[ -n ${REDIS_PASSWORD-} ]]; then REDIS_URL="redis${REDIS_SSL:+s}://${REDIS_USERNAME-}:${REDIS_PASSWORD}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" @@ -68,13 +56,23 @@ else REDIS_URL="redis${REDIS_SSL:+s}://${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" fi +echo "Starting RQ cron scheduler..." +# The URL carries the password, so it goes through RQ_REDIS_URL rather than +# --url, which would put it on a world-readable command line. +PYTHONPATH="/app/backend:${PYTHONPATH-}" \ + RQ_REDIS_URL="${REDIS_URL}" \ + rq cron \ + --path /app/backend \ + --logging-level "${LOGLEVEL:-INFO}" \ + tasks.cron_config & + +echo "Starting RQ worker..." # Set PYTHONPATH so RQ can find the tasks module. -# The connection URL goes through RQ_REDIS_URL rather than --url, so the -# embedded password stays out of the worker's world-readable command line. # Use a worker class that drops the noisy per-sweep "cleaning registries for # queue" log line. The maintenance interval keeps its default (~10 min) so -# orphaned STARTED jobs and stale workers are still pruned promptly, which the -# watcher's Worker.all() scan dedupe relies on. +# orphaned STARTED jobs and stale workers are still pruned promptly. +# --with-scheduler releases delayed jobs, which is how the watcher's rescans +# wait out their delay. PYTHONPATH="/app/backend:${PYTHONPATH-}" \ RQ_REDIS_URL="${REDIS_URL}" \ rq worker \ @@ -82,6 +80,7 @@ PYTHONPATH="/app/backend:${PYTHONPATH-}" \ --worker-class handler.rq_worker.RomMWorker \ --pid /tmp/rq_worker.pid \ --logging_level "${LOGLEVEL:-INFO}" \ + --with-scheduler \ high default low & echo "Starting watcher..." diff --git a/pyproject.toml b/pyproject.toml index 6fbe370c16..a8a98f2a7d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,9 +48,6 @@ dependencies = [ "python-socketio ~= 5.16", "redis ~= 6.2", "rq ~= 2.11", - # TODO: Move back to upstream `rq-scheduler`, when support for username and SSL settings is added. - # Related PR: https://github.com/rq/rq-scheduler/pull/325 - "rq-scheduler @ git+https://github.com/adamantike/rq-scheduler.git@feat/script-options-username-ssl", "sentry-sdk ~= 2.32", "starlette ~= 1.6.0", "streaming-form-data ~= 1.19", diff --git a/uv.lock b/uv.lock index b9d5cecc6b..cbe9b0b3e1 100644 --- a/uv.lock +++ b/uv.lock @@ -439,12 +439,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d0/39/783980e78cb92c2d7bdb1fc7dbc86e94ccc6d58224d76a7f1f51b6c51e30/croniter-6.2.2-py3-none-any.whl", hash = "sha256:a5d17b1060974d36251ea4faf388233eca8acf0d09cbd92d35f4c4ac8f279960", size = 45422, upload-time = "2026-03-15T08:43:46.626Z" }, ] -[[package]] -name = "crontab" -version = "1.0.4" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/1e/8b/3ea72ac8e26090b63779b4e0074af79b02bbbab7ddd01b36109bc0892d31/crontab-1.0.4.tar.gz", hash = "sha256:715b0e5e105bc62c9683cbb93c1cc5821e07a3e28d17404576d22dba7a896c92", size = 21677, upload-time = "2025-04-09T18:23:59.01Z" } - [[package]] name = "cryptography" version = "49.0.0" @@ -2231,7 +2225,6 @@ dependencies = [ { name = "pyyaml" }, { name = "redis" }, { name = "rq" }, - { name = "rq-scheduler" }, { name = "sentry-sdk" }, { name = "sqlalchemy", extra = ["mariadb-connector", "mysql-connector", "postgresql-psycopg"] }, { name = "starlette" }, @@ -2321,7 +2314,6 @@ requires-dist = [ { name = "pyyaml", specifier = "~=6.0" }, { name = "redis", specifier = "~=6.2" }, { name = "rq", specifier = "~=2.11" }, - { name = "rq-scheduler", git = "https://github.com/adamantike/rq-scheduler.git?rev=feat%2Fscript-options-username-ssl" }, { name = "sentry-sdk", specifier = "~=2.32" }, { name = "sqlalchemy", extras = ["mariadb-connector", "mysql-connector", "postgresql-psycopg"], specifier = "~=2.0" }, { name = "starlette", specifier = "~=1.6.0" }, @@ -2356,16 +2348,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/80/1e/f85cba91fe58c63812b4acce073fc448230381c4785e03832f455305ff82/rq-2.11.0-py3-none-any.whl", hash = "sha256:2f54a31375d7c1b5a1642acef4948809ce6484775b9741fe7edd9399ec9306a9", size = 126654, upload-time = "2026-08-17T02:59:17.112Z" }, ] -[[package]] -name = "rq-scheduler" -version = "0.14.0" -source = { git = "https://github.com/adamantike/rq-scheduler.git?rev=feat%2Fscript-options-username-ssl#39583cb2a00c6faa12ef34c7277893064a83c4de" } -dependencies = [ - { name = "crontab" }, - { name = "python-dateutil" }, - { name = "rq" }, -] - [[package]] name = "sentry-sdk" version = "2.32.0" From bce4a3c2bd4ae407ad4c298a9d7c7c0278f622b0 Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Sun, 23 Aug 2026 15:12:45 -0500 Subject: [PATCH 4/8] fix(tasks): keep a request body from choosing which task runs Spreading the caller's arguments alongside the task name let a body set `name` and redirect the run, so an authorized caller could reach a task the route had just refused: one the API does not surface, or one whose `manual_run` is off. The endpoint checked the name from the path, and the worker then resolved whatever name the payload ended up holding. Nest the caller's arguments under `task_kwargs` instead of spreading them. Ordering the spread so the trusted name wins would also close the hole, but nesting makes the collision impossible rather than order-dependent, and leaves a task free to take an argument called name. Co-Authored-By: Claude Opus 5 --- backend/endpoints/tasks.py | 6 +++-- backend/tasks/tasks.py | 7 +++--- backend/tests/endpoints/test_tasks.py | 36 +++++++++++++++++++++++++++ backend/tests/tasks/test_tasks.py | 14 ++++++++++- 4 files changed, 57 insertions(+), 6 deletions(-) diff --git a/backend/endpoints/tasks.py b/backend/endpoints/tasks.py index 8816092e3e..e27a4ec9fa 100644 --- a/backend/endpoints/tasks.py +++ b/backend/endpoints/tasks.py @@ -319,10 +319,12 @@ async def run_single_task( ) # Enqueued by name, like the scheduled runs, so the payload carries no - # pickled task and the job is readable by whatever version picks it up. + # pickled task and the job is readable by whatever version picks it up. The + # caller's arguments are nested rather than spread, so a body cannot name a + # different task than the one this route just authorized. job = low_prio_queue.enqueue( run_task_by_name, - kwargs={"name": task_name, **(task_kwargs or {})}, + kwargs={"name": task_name, "task_kwargs": task_kwargs or {}}, job_timeout=task_instance.timeout, result_ttl=TASK_RESULT_TTL, meta={ diff --git a/backend/tasks/tasks.py b/backend/tasks/tasks.py index 22218b33b6..0410e28897 100644 --- a/backend/tasks/tasks.py +++ b/backend/tasks/tasks.py @@ -11,7 +11,7 @@ from utils.context import ctx_httpx_client -async def run_task_by_name(name: str, **kwargs: Any) -> Any: +async def run_task_by_name(name: str, task_kwargs: dict[str, Any] | None = None) -> Any: """Run the task registered under ``name``. Every scheduled and manually triggered task is enqueued through here, so a @@ -20,7 +20,8 @@ async def run_task_by_name(name: str, **kwargs: Any) -> Any: Args: name: The key the task is registered under. - kwargs: Forwarded to the task's ``run``. + task_kwargs: Forwarded to the task's ``run``, nested so that they cannot + collide with the name of the task to run. Returns: Whatever the task returns. @@ -33,7 +34,7 @@ async def run_task_by_name(name: str, **kwargs: Any) -> Any: if task is None: raise TaskNotFoundException(name) - return await task.run(**kwargs) + return await task.run(**(task_kwargs or {})) def update_job_meta(metadata: dict[str, Any]) -> None: diff --git a/backend/tests/endpoints/test_tasks.py b/backend/tests/endpoints/test_tasks.py index 96a6ff0b7e..e3095bb40d 100644 --- a/backend/tests/endpoints/test_tasks.py +++ b/backend/tests/endpoints/test_tasks.py @@ -569,3 +569,39 @@ def test_error_handling(self, client, access_token): headers={"Authorization": f"Bearer {access_token}"}, ) assert response.status_code == status.HTTP_404_NOT_FOUND + + +class TestRunSingleTaskArgumentHandling: + """A request body must not be able to choose which task runs.""" + + @patch("endpoints.tasks.low_prio_queue.enqueue", return_value=create_mock_job()) + @patch( + "endpoints.tasks.MANUAL_TASKS", + { + "allowed_task": Mock( + spec=Task, + task_type=TaskType.CLEANUP, + title="Allowed Task", + description="Allowed", + enabled=True, + manual_run=True, + can_run_manually=True, + timeout=300, + ), + }, + ) + @patch("endpoints.tasks.VISIBLE_SCHEDULED_TASKS", ()) + def test_body_cannot_override_the_task_name( + self, mock_enqueue, client, access_token + ): + response = client.post( + "/api/tasks/run/allowed_task", + headers={"Authorization": f"Bearer {access_token}"}, + json={"name": "sync_push_pull"}, + ) + + assert response.status_code == status.HTTP_200_OK + assert mock_enqueue.call_args.kwargs["kwargs"] == { + "name": "allowed_task", + "task_kwargs": {"name": "sync_push_pull"}, + } diff --git a/backend/tests/tasks/test_tasks.py b/backend/tests/tasks/test_tasks.py index b471cf8093..f66c09dedd 100644 --- a/backend/tests/tasks/test_tasks.py +++ b/backend/tests/tasks/test_tasks.py @@ -179,10 +179,22 @@ async def test_forwards_keyword_arguments(self, mocker): task.run = AsyncMock(return_value=None) mocker.patch("tasks.registry.get_task", return_value=task) - await run_task_by_name("some_task", force=True) + await run_task_by_name("some_task", {"force": True}) task.run.assert_awaited_once_with(force=True) + async def test_forwarded_arguments_cannot_name_another_task(self, mocker): + # The arguments reach the task rather than this function's own name, so a + # request body cannot redirect the run to a task it was not allowed. + task = MagicMock() + task.run = AsyncMock(return_value=None) + get_task = mocker.patch("tasks.registry.get_task", return_value=task) + + await run_task_by_name("allowed_task", {"name": "hidden_task"}) + + get_task.assert_called_once_with("allowed_task") + task.run.assert_awaited_once_with(name="hidden_task") + async def test_raises_for_a_name_that_is_not_registered(self, mocker): mocker.patch("tasks.registry.get_task", return_value=None) From 63f2d49545ced4a098e9e3f5e4ed622c3e19434e Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Sun, 23 Aug 2026 15:43:04 -0500 Subject: [PATCH 5/8] fix(watcher): read a pending scan's whole scope, not just its platform ids A scan can be scoped by platform id, by folder slug for a directory with no database row, or by rom. Treating anything without platform ids as a scan of the whole library meant a queued slug-scoped or rom-scoped scan suppressed every rescan the watcher would otherwise schedule. Summarise what the pending scans cover instead, and match a changed folder against the slugs as well as the ids. A scan whose payload cannot be read is left out: its scope is unknowable, and a duplicate scan costs less than a rescan that never happens. Adds the tests this file never had, plus tests for the job accessors that tolerate an unreadable payload. Four of them fail against the previous scope check. Co-Authored-By: Claude Opus 5 --- backend/tests/handler/test_redis_handler.py | 71 ++++++ backend/tests/test_watcher.py | 251 ++++++++++++++++++++ backend/watcher.py | 64 ++++- 3 files changed, 375 insertions(+), 11 deletions(-) create mode 100644 backend/tests/handler/test_redis_handler.py create mode 100644 backend/tests/test_watcher.py diff --git a/backend/tests/handler/test_redis_handler.py b/backend/tests/handler/test_redis_handler.py new file mode 100644 index 0000000000..2a55b38cc1 --- /dev/null +++ b/backend/tests/handler/test_redis_handler.py @@ -0,0 +1,71 @@ +from unittest.mock import MagicMock, PropertyMock + +from rq.exceptions import DeserializationError, InvalidJobOperation +from rq.job import Job, JobStatus + +from handler.redis_handler import get_job_func_name, get_job_kwargs, get_job_status + + +def make_job() -> MagicMock: + job = MagicMock(spec=Job) + job.id = "job-1" + return job + + +def unreadable(attribute: str) -> MagicMock: + """A job whose `attribute` fails the way an undeserializable payload does. + + Each mock gets its own type, so patching the property is instance-local. + """ + job = make_job() + setattr(type(job), attribute, PropertyMock(side_effect=DeserializationError)) + return job + + +class TestGetJobFuncName: + def test_returns_the_recorded_name(self): + job = make_job() + job.func_name = "tasks.tasks.run_task_by_name" + + assert get_job_func_name(job) == "tasks.tasks.run_task_by_name" + + def test_falls_back_when_the_payload_cannot_be_read(self): + assert get_job_func_name(unreadable("func_name")) == "" + + def test_returns_the_given_fallback(self): + job = unreadable("func_name") + + assert get_job_func_name(job, fallback="unknown") == "unknown" + + +class TestGetJobKwargs: + def test_returns_the_enqueued_keyword_arguments(self): + job = make_job() + job.kwargs = {"platform_ids": [1]} + + assert get_job_kwargs(job) == {"platform_ids": [1]} + + def test_returns_empty_for_a_job_enqueued_without_keywords(self): + job = make_job() + job.kwargs = {} + + assert get_job_kwargs(job) == {} + + def test_returns_none_when_the_payload_cannot_be_read(self): + # None rather than empty: the caller has to tell "covers nothing" apart + # from "cannot be known". + assert get_job_kwargs(unreadable("kwargs")) is None + + +class TestGetJobStatus: + def test_returns_the_recorded_status(self): + job = make_job() + job.get_status.return_value = JobStatus.QUEUED + + assert get_job_status(job) == JobStatus.QUEUED + + def test_returns_none_once_the_job_hash_has_expired(self): + job = make_job() + job.get_status.side_effect = InvalidJobOperation + + assert get_job_status(job) is None diff --git a/backend/tests/test_watcher.py b/backend/tests/test_watcher.py new file mode 100644 index 0000000000..e4e7dc049c --- /dev/null +++ b/backend/tests/test_watcher.py @@ -0,0 +1,251 @@ +from itertools import count +from unittest.mock import MagicMock, PropertyMock + +import pytest +import watcher as watcher_module +from rq.exceptions import DeserializationError +from rq.job import Job +from watcher import EventType, get_pending_scan_coverage, process_changes + +from config import LIBRARY_BASE_PATH +from handler.scan_handler import ScanType + +_job_ids = count() + + +def make_job(**kwargs) -> MagicMock: + """An RQ job stub enqueued the way the scan callers enqueue: keywords only.""" + job = MagicMock(spec=Job) + job.id = f"job-{next(_job_ids)}" + job.args = () + job.kwargs = kwargs + return job + + +def watcher_full_rescan_job() -> MagicMock: + return make_job(platform_ids=[], scan_type=ScanType.UPDATE) + + +def watcher_platform_job(platform_id: int) -> MagicMock: + return make_job(platform_ids=[platform_id], scan_type=ScanType.QUICK) + + +def scheduled_rescan_job() -> MagicMock: + """The scheduled rescan goes through the task runner, named by keyword.""" + return make_job(name="scan_library") + + +def patch_pending_jobs(mocker, *jobs): + return mocker.patch.object( + watcher_module, "get_pending_scan_jobs", return_value=list(jobs) + ) + + +class TestPendingScanCoverage: + """Scans are enqueued with keywords, so the scope lives in job.kwargs.""" + + def test_nothing_pending(self, mocker): + patch_pending_jobs(mocker) + + coverage = get_pending_scan_coverage() + + assert coverage.full_library == 0 + assert coverage.platform_ids == frozenset() + assert coverage.platform_fs_slugs == frozenset() + + def test_a_scan_with_no_platform_ids_covers_the_library(self, mocker): + patch_pending_jobs(mocker, watcher_full_rescan_job()) + + assert get_pending_scan_coverage().full_library == 1 + + def test_the_scheduled_rescan_task_covers_the_library(self, mocker): + patch_pending_jobs(mocker, scheduled_rescan_job()) + + assert get_pending_scan_coverage().full_library == 1 + + def test_full_rescans_are_counted(self, mocker): + patch_pending_jobs(mocker, watcher_full_rescan_job(), scheduled_rescan_job()) + + assert get_pending_scan_coverage().full_library == 2 + + def test_scoped_scans_report_their_platforms(self, mocker): + patch_pending_jobs( + mocker, watcher_platform_job(1), make_job(platform_ids=[2, 3]) + ) + + coverage = get_pending_scan_coverage() + + assert coverage.full_library == 0 + assert coverage.platform_ids == frozenset({1, 2, 3}) + + def test_a_socket_scan_scoped_by_slug_reports_its_slugs(self, mocker): + # The socket accepts folders with no database row, which arrive as slugs. + patch_pending_jobs(mocker, make_job(platform_ids=[], platform_fs_slugs=["gba"])) + + coverage = get_pending_scan_coverage() + + assert coverage.full_library == 0 + assert coverage.platform_fs_slugs == frozenset({"gba"}) + + def test_a_rom_scoped_scan_is_not_a_full_rescan(self, mocker): + # A scan of selected roms resolves its platforms from the database, so + # it covers neither the library nor any platform this can name. + patch_pending_jobs(mocker, make_job(platform_ids=[], roms_ids=[7])) + + coverage = get_pending_scan_coverage() + + assert coverage.full_library == 0 + assert coverage.platform_ids == frozenset() + + def test_ignores_positional_arguments(self, mocker): + # job.args is what the old dedupe read, and it is always empty. + job = make_job(platform_ids=[1]) + job.args = ([2],) + patch_pending_jobs(mocker, job) + + assert get_pending_scan_coverage().platform_ids == frozenset({1}) + + def test_an_unreadable_payload_is_ignored(self, mocker): + # Its scope is unknowable, and a duplicate scan costs less than a rescan + # that never happens because of a job nobody can read. + job = make_job() + type(job).kwargs = PropertyMock(side_effect=DeserializationError) + patch_pending_jobs(mocker, job) + + coverage = get_pending_scan_coverage() + + assert coverage.full_library == 0 + assert coverage.platform_ids == frozenset() + + +class TestProcessChanges: + """A filesystem change must not schedule a scan that is already pending.""" + + @pytest.fixture(autouse=True) + def library_layout(self, mocker): + config = MagicMock() + config.has_structure_path_b = False + config.EXCLUDED_SINGLE_FILES = [] + config.EXCLUDED_MULTI_FILES = [] + config.EXCLUDED_MULTI_PARTS_FILES = [] + mocker.patch.object(watcher_module.cm, "get_config", return_value=config) + mocker.patch.object( + watcher_module.meta_igdb_handler, "is_enabled", return_value=True + ) + + @pytest.fixture + def platform(self, mocker): + db_platform = MagicMock(id=1, fs_slug="gba") + mocker.patch.object( + watcher_module.db_platform_handler, + "get_platform_by_fs_slug", + return_value=db_platform, + ) + return db_platform + + @pytest.fixture + def enqueue_in(self, mocker): + return mocker.patch.object(watcher_module.low_prio_queue, "enqueue_in") + + def rom_change(self, fs_slug: str = "gba"): + return (EventType.ADDED, f"{LIBRARY_BASE_PATH}/roms/{fs_slug}/game.gba") + + def platform_dir_change(self, fs_slug: str = "gba"): + return (EventType.ADDED, f"{LIBRARY_BASE_PATH}/roms/{fs_slug}") + + def test_schedules_a_scan_for_the_changed_platform( + self, mocker, platform, enqueue_in + ): + patch_pending_jobs(mocker) + + process_changes([self.rom_change()]) + + enqueue_in.assert_called_once() + assert enqueue_in.call_args.kwargs["platform_ids"] == [platform.id] + + def test_a_platform_directory_change_schedules_a_full_rescan( + self, mocker, platform, enqueue_in + ): + patch_pending_jobs(mocker) + + process_changes([self.platform_dir_change()]) + + enqueue_in.assert_called_once() + assert enqueue_in.call_args.kwargs["platform_ids"] == [] + + def test_a_pending_full_rescan_absorbs_every_change( + self, mocker, platform, enqueue_in + ): + patch_pending_jobs(mocker, watcher_full_rescan_job()) + + process_changes([self.rom_change()]) + + enqueue_in.assert_not_called() + + def test_a_pending_scheduled_rescan_absorbs_every_change( + self, mocker, platform, enqueue_in + ): + patch_pending_jobs(mocker, scheduled_rescan_job()) + + process_changes([self.rom_change()]) + + enqueue_in.assert_not_called() + + def test_a_pending_scan_for_the_platform_is_not_duplicated( + self, mocker, platform, enqueue_in + ): + patch_pending_jobs(mocker, watcher_platform_job(platform.id)) + + process_changes([self.rom_change()]) + + enqueue_in.assert_not_called() + + def test_a_pending_scan_scoped_by_slug_is_not_duplicated( + self, mocker, platform, enqueue_in + ): + patch_pending_jobs(mocker, make_job(platform_fs_slugs=[platform.fs_slug])) + + process_changes([self.rom_change()]) + + enqueue_in.assert_not_called() + + def test_a_pending_scan_for_another_platform_does_not_block( + self, mocker, platform, enqueue_in + ): + patch_pending_jobs(mocker, watcher_platform_job(platform.id + 1)) + + process_changes([self.rom_change()]) + + enqueue_in.assert_called_once() + + def test_a_platform_missing_from_the_database_is_skipped(self, mocker, enqueue_in): + patch_pending_jobs(mocker) + mocker.patch.object( + watcher_module.db_platform_handler, + "get_platform_by_fs_slug", + return_value=None, + ) + + process_changes([self.rom_change()]) + + enqueue_in.assert_not_called() + + def test_a_pending_scan_scoped_to_another_slug_does_not_block( + self, mocker, platform, enqueue_in + ): + # A slug-scoped scan names no platform id, which must not be mistaken + # for a scan of the whole library. + patch_pending_jobs(mocker, make_job(platform_fs_slugs=["snes"])) + + process_changes([self.rom_change()]) + + enqueue_in.assert_called_once() + + def test_a_pending_rom_scoped_scan_does_not_block( + self, mocker, platform, enqueue_in + ): + patch_pending_jobs(mocker, make_job(roms_ids=[7])) + + process_changes([self.rom_change()]) + + enqueue_in.assert_called_once() diff --git a/backend/watcher.py b/backend/watcher.py index cff5f40f89..874dfcbf37 100644 --- a/backend/watcher.py +++ b/backend/watcher.py @@ -3,6 +3,7 @@ import json import os from collections.abc import Sequence +from dataclasses import dataclass from datetime import timedelta from typing import cast @@ -71,6 +72,51 @@ class EventType(enum.StrEnum): Change = tuple[EventType, str] +@dataclass(frozen=True) +class PendingScanCoverage: + """What the scans already in flight cover, so a rescan is not duplicated.""" + + full_library: int + platform_ids: frozenset[int] + platform_fs_slugs: frozenset[str] + + +def get_pending_scan_coverage() -> PendingScanCoverage: + """Summarise what the scans waiting to run already cover. + + Returns: + PendingScanCoverage: How many pending scans cover the whole library, and + the platforms and folders the rest are scoped to. + """ + full_library = 0 + platform_ids: set[int] = set() + platform_fs_slugs: set[str] = set() + + for job in get_pending_scan_jobs(): + kwargs = get_job_kwargs(job) + if kwargs is None: + # Nothing is known about what an unreadable scan covers, and a + # duplicate scan costs less than a rescan that never happens. + continue + + # Scans are enqueued with keywords only. A task-driven scan names no + # scope at all, and covers everything. + job_platform_ids = kwargs.get("platform_ids") or [] + job_platform_fs_slugs = kwargs.get("platform_fs_slugs") or [] + if not (job_platform_ids or job_platform_fs_slugs or kwargs.get("roms_ids")): + full_library += 1 + continue + + platform_ids.update(job_platform_ids) + platform_fs_slugs.update(job_platform_fs_slugs) + + return PendingScanCoverage( + full_library=full_library, + platform_ids=frozenset(platform_ids), + platform_fs_slugs=frozenset(platform_fs_slugs), + ) + + def process_changes(changes: Sequence[Change]) -> None: if not ENABLE_RESCAN_ON_FILESYSTEM_CHANGE: return @@ -151,16 +197,9 @@ def _is_excluded(path: str) -> bool: log.warning("No metadata sources enabled, skipping rescan") return - # The platforms each pending scan covers. A scan with no platform ids - # covers the whole library, which is also what a task-driven scan does. - pending_scopes = [ - kwargs.get("platform_ids") or [] - for job in get_pending_scan_jobs() - if (kwargs := get_job_kwargs(job)) is not None - ] - - if any(not scope for scope in pending_scopes): - log.info("Full rescan already pending") + pending = get_pending_scan_coverage() + if pending.full_library: + log.info(f"Full rescan already pending ({pending.full_library} job(s))") return time_delta = timedelta(minutes=RESCAN_ON_FILESYSTEM_CHANGE_DELAY) @@ -191,7 +230,10 @@ def _is_excluded(path: str) -> bool: if not db_platform: continue - if any(db_platform.id in scope for scope in pending_scopes): + if ( + db_platform.id in pending.platform_ids + or fs_slug in pending.platform_fs_slugs + ): log.info(f"Scan already pending for {hl(fs_slug)}") continue From 965ae34b59602c5a93eb66f63909400ba359dd12 Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Sun, 23 Aug 2026 15:59:07 -0500 Subject: [PATCH 6/8] refactor(startup): let RQ settle the backfill duplicate check Both startup backfills asked whether their job id existed and then enqueued it, which two instances starting together can both get past. RQ does the check, the save and the push in one Lua call and raises DuplicateJobError, so the race closes and the pre-check goes. While here, enqueue them by name like every other task, so neither payload carries a pickled task instance. Co-Authored-By: Claude Opus 5 --- backend/startup.py | 42 +++++++++++++++++------------- backend/tests/test_startup.py | 48 ++++++++++++++++++++++------------- 2 files changed, 55 insertions(+), 35 deletions(-) diff --git a/backend/startup.py b/backend/startup.py index 4ad314ed51..0ae957b0bd 100644 --- a/backend/startup.py +++ b/backend/startup.py @@ -4,7 +4,7 @@ import sentry_sdk from opentelemetry import trace -from rq.job import Job +from rq.exceptions import DuplicateJobError from config import ( ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP, @@ -35,6 +35,7 @@ recompute_save_content_hashes_task, ) from tasks.scheduled.convert_images_to_webp import convert_images_to_webp_task +from tasks.tasks import run_task_by_name from utils import get_version from utils.cache import conditionally_set_cache from utils.context import initialize_context @@ -44,6 +45,11 @@ RECOMPUTE_SAVE_HASHES_JOB_ID = "recompute_save_content_hashes_bootstrap" CONVERT_IMAGES_TO_WEBP_JOB_ID = "convert_images_to_webp_bootstrap" +# The names these backfills are registered under, which is what the payload +# carries rather than the task itself. +RECOMPUTE_SAVE_HASHES_TASK = "recompute_save_content_hashes" +CONVERT_IMAGES_TO_WEBP_TASK = "convert_images_to_webp" + def _enqueue_recompute_save_hashes_if_needed() -> None: """Backfill content_hash for saves uploaded before the path-resolution @@ -66,16 +72,13 @@ def _enqueue_recompute_save_hashes_if_needed() -> None: return try: - if Job.exists(RECOMPUTE_SAVE_HASHES_JOB_ID, low_prio_queue.connection): - log.info( - "recompute_save_content_hashes already queued or running from a " - "previous restart; skipping enqueue" - ) - return - + # A fixed id with unique=True settles it in one round trip, so two + # instances starting together cannot both get past the check. low_prio_queue.enqueue( - recompute_save_content_hashes_task.run, + run_task_by_name, + kwargs={"name": RECOMPUTE_SAVE_HASHES_TASK}, job_id=RECOMPUTE_SAVE_HASHES_JOB_ID, + unique=True, job_timeout=TASK_TIMEOUT, meta={ "task_name": recompute_save_content_hashes_task.title, @@ -86,6 +89,11 @@ def _enqueue_recompute_save_hashes_if_needed() -> None: f"Enqueued recompute_save_content_hashes ({missing} saves with NULL content_hash); " "running on low-priority worker" ) + except DuplicateJobError: + log.info( + "recompute_save_content_hashes already queued or running from a " + "previous restart; skipping enqueue" + ) except Exception: log.exception( "Failed to enqueue recompute_save_content_hashes; admins can run it manually" @@ -101,16 +109,11 @@ def _enqueue_convert_images_to_webp() -> None: enabling. Without a backfill, existing covers have no .webp sibling and every request 404s until the cron eventually runs.""" try: - if Job.exists(CONVERT_IMAGES_TO_WEBP_JOB_ID, low_prio_queue.connection): - log.info( - "convert_images_to_webp already queued or running from a " - "previous restart; skipping enqueue" - ) - return - low_prio_queue.enqueue( - convert_images_to_webp_task.run, + run_task_by_name, + kwargs={"name": CONVERT_IMAGES_TO_WEBP_TASK}, job_id=CONVERT_IMAGES_TO_WEBP_JOB_ID, + unique=True, job_timeout=TASK_TIMEOUT, meta={ "task_name": convert_images_to_webp_task.title, @@ -118,6 +121,11 @@ def _enqueue_convert_images_to_webp() -> None: }, ) log.info("Enqueued convert_images_to_webp backfill on low-priority worker") + except DuplicateJobError: + log.info( + "convert_images_to_webp already queued or running from a previous " + "restart; skipping enqueue" + ) except Exception: log.exception( "Failed to enqueue convert_images_to_webp; admins can run it manually" diff --git a/backend/tests/test_startup.py b/backend/tests/test_startup.py index 273cd16553..34d5c908b6 100644 --- a/backend/tests/test_startup.py +++ b/backend/tests/test_startup.py @@ -1,8 +1,11 @@ """Tests for startup-time auto-enqueue of the recompute task.""" import startup +from rq.exceptions import DuplicateJobError from rq.job import JOB_ID_PATTERN +from tasks.registry import get_task + def test_enqueue_recompute_skips_when_no_missing_hashes(mocker): """Saves all have content_hash -> no enqueue.""" @@ -21,15 +24,17 @@ def test_enqueue_recompute_fires_when_missing_hashes_present(mocker): mocker.patch.object( startup.db_save_handler, "count_saves_missing_content_hash", return_value=42 ) - mocker.patch.object(startup.Job, "exists", return_value=False) enqueue = mocker.patch.object(startup.low_prio_queue, "enqueue") startup._enqueue_recompute_save_hashes_if_needed() enqueue.assert_called_once() args, kwargs = enqueue.call_args - # First positional arg is the bound task.run method - assert args[0].__self__ is startup.recompute_save_content_hashes_task + # Enqueued by name, so the payload survives the code moving underneath it + assert args[0] is startup.run_task_by_name + assert kwargs["kwargs"] == {"name": startup.RECOMPUTE_SAVE_HASHES_TASK} + # RQ settles the duplicate check and the enqueue in one round trip + assert kwargs["unique"] is True # Sanity-check the meta payload routes correctly in the task list UI assert kwargs["meta"]["task_name"] == ( startup.recompute_save_content_hashes_task.title @@ -50,17 +55,26 @@ def test_recompute_job_id_is_valid_rq_id(): assert JOB_ID_PATTERN.fullmatch(startup.RECOMPUTE_SAVE_HASHES_JOB_ID) -def test_enqueue_recompute_skips_when_already_queued(mocker): - """An in-flight job from a previous restart -> skip enqueue, don't double up.""" +def test_enqueue_recompute_tolerates_an_in_flight_job(mocker): + """An in-flight job from a previous restart -> RQ refuses, startup carries on.""" mocker.patch.object( startup.db_save_handler, "count_saves_missing_content_hash", return_value=10 ) - mocker.patch.object(startup.Job, "exists", return_value=True) - enqueue = mocker.patch.object(startup.low_prio_queue, "enqueue") + mocker.patch.object( + startup.low_prio_queue, "enqueue", side_effect=DuplicateJobError("exists") + ) startup._enqueue_recompute_save_hashes_if_needed() - enqueue.assert_not_called() + +def test_both_backfills_name_a_registered_task(mocker): + """The name in the payload is all the runner gets, so it has to resolve.""" + assert get_task(startup.RECOMPUTE_SAVE_HASHES_TASK) is ( + startup.recompute_save_content_hashes_task + ) + assert get_task(startup.CONVERT_IMAGES_TO_WEBP_TASK) is ( + startup.convert_images_to_webp_task + ) def test_enqueue_recompute_swallows_count_error(mocker): @@ -82,7 +96,6 @@ def test_enqueue_recompute_swallows_enqueue_error(mocker): mocker.patch.object( startup.db_save_handler, "count_saves_missing_content_hash", return_value=5 ) - mocker.patch.object(startup.Job, "exists", return_value=False) mocker.patch.object( startup.low_prio_queue, "enqueue", side_effect=RuntimeError("redis gone") ) @@ -92,14 +105,15 @@ def test_enqueue_recompute_swallows_enqueue_error(mocker): def test_enqueue_convert_webp_fires_when_not_queued(mocker): """No in-flight bootstrap job -> enqueue the backfill exactly once.""" - mocker.patch.object(startup.Job, "exists", return_value=False) enqueue = mocker.patch.object(startup.low_prio_queue, "enqueue") startup._enqueue_convert_images_to_webp() enqueue.assert_called_once() args, kwargs = enqueue.call_args - assert args[0].__self__ is startup.convert_images_to_webp_task + assert args[0] is startup.run_task_by_name + assert kwargs["kwargs"] == {"name": startup.CONVERT_IMAGES_TO_WEBP_TASK} + assert kwargs["unique"] is True assert kwargs["meta"]["task_name"] == startup.convert_images_to_webp_task.title assert kwargs["meta"]["task_type"] == ( startup.convert_images_to_webp_task.task_type.value @@ -114,19 +128,17 @@ def test_convert_webp_job_id_is_valid_rq_id(): assert JOB_ID_PATTERN.fullmatch(startup.CONVERT_IMAGES_TO_WEBP_JOB_ID) -def test_enqueue_convert_webp_skips_when_already_queued(mocker): - """An in-flight job from a previous restart -> skip enqueue, don't double up.""" - mocker.patch.object(startup.Job, "exists", return_value=True) - enqueue = mocker.patch.object(startup.low_prio_queue, "enqueue") +def test_enqueue_convert_webp_tolerates_an_in_flight_job(mocker): + """An in-flight job from a previous restart -> RQ refuses, startup carries on.""" + mocker.patch.object( + startup.low_prio_queue, "enqueue", side_effect=DuplicateJobError("exists") + ) startup._enqueue_convert_images_to_webp() - enqueue.assert_not_called() - def test_enqueue_convert_webp_swallows_enqueue_error(mocker): """A failed enqueue must not crash startup.""" - mocker.patch.object(startup.Job, "exists", return_value=False) mocker.patch.object( startup.low_prio_queue, "enqueue", side_effect=RuntimeError("redis gone") ) From 3a0a7d444c3673440f4171e4370a6413a5412803 Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Sun, 23 Aug 2026 16:18:22 -0500 Subject: [PATCH 7/8] fix(docker): read REDIS_SSL as the boolean the app reads it as The shell built the connection URL with `redis${REDIS_SSL:+s}`, which only asks whether the variable is set. `env.template` ships `REDIS_SSL=false`, so following the documented value pointed the worker at `rediss://` while the app parsed the same variable through `safe_str_to_bool` and connected in plaintext. Adding the cron process put a second process behind it. Parse the value in both entrypoints, matching the truthy set the app accepts. Also clear the old scheduler's lock and instance keys even when its registry held no jobs: a scheduler that never queued anything still registered itself, and only the orphan-job work needs the registry. Co-Authored-By: Claude Opus 5 --- backend/startup.py | 29 ++++++++++++++-------------- backend/tests/test_startup.py | 36 +++++++++++++++++++++++++++++++++++ docker/init_scripts/init | 20 ++++++++++++++++--- entrypoint.sh | 15 +++++++++++---- 4 files changed, 79 insertions(+), 21 deletions(-) diff --git a/backend/startup.py b/backend/startup.py index 0ae957b0bd..46daecd94e 100644 --- a/backend/startup.py +++ b/backend/startup.py @@ -150,26 +150,27 @@ def _drop_legacy_scheduler_state() -> None: job_id.decode() for job_id in redis_client.zrange(LEGACY_SCHEDULED_JOBS_KEY, 0, -1) } - if not legacy_job_ids: - return - # A cron job the old scheduler had already queued lives in both places, - # and it still has to run, so only the orphans are deleted. - queued = set() - for queue in (high_prio_queue, default_queue, low_prio_queue): - queued.update(queue.get_job_ids()) + if legacy_job_ids: + # A cron job the old scheduler had already queued lives in both + # places, and it still has to run, so only the orphans are deleted. + queued: set[str] = set() + for queue in (high_prio_queue, default_queue, low_prio_queue): + queued.update(queue.get_job_ids()) - orphans = legacy_job_ids - queued - if orphans: - redis_client.delete(*(f"rq:job:{job_id}" for job_id in orphans)) + orphans = legacy_job_ids - queued + if orphans: + redis_client.delete(*(f"rq:job:{job_id}" for job_id in orphans)) + log.info( + f"Cleared {len(legacy_job_ids)} job(s) left behind by the old scheduler" + ) + + # The registry, the lock and the scheduler's own keys go regardless: an + # old scheduler that never held a job still registered itself. redis_client.delete(*LEGACY_SCHEDULER_KEYS) for key in redis_client.scan_iter("rq:scheduler_instance:*"): redis_client.delete(key) - - log.info( - f"Cleared {len(legacy_job_ids)} job(s) left behind by the old scheduler" - ) except Exception: log.exception("Failed to clear the old scheduler's leftovers") diff --git a/backend/tests/test_startup.py b/backend/tests/test_startup.py index 34d5c908b6..a4dcacea0a 100644 --- a/backend/tests/test_startup.py +++ b/backend/tests/test_startup.py @@ -1,5 +1,6 @@ """Tests for startup-time auto-enqueue of the recompute task.""" +import pytest import startup from rq.exceptions import DuplicateJobError from rq.job import JOB_ID_PATTERN @@ -144,3 +145,38 @@ def test_enqueue_convert_webp_swallows_enqueue_error(mocker): ) startup._enqueue_convert_images_to_webp() + + +class TestDropLegacySchedulerState: + """The old scheduler's keys go on the first start after the migration.""" + + @pytest.fixture + def redis(self, mocker): + redis = mocker.patch.object(startup, "redis_client") + redis.scan_iter.return_value = [] + return redis + + def test_removes_the_scheduler_keys_with_no_jobs_left_behind(self, redis): + redis.zrange.return_value = [] + + startup._drop_legacy_scheduler_state() + + redis.delete.assert_called_once_with(*startup.LEGACY_SCHEDULER_KEYS) + + def test_deletes_orphaned_jobs_but_not_queued_ones(self, mocker, redis): + redis.zrange.return_value = [b"orphan", b"queued"] + for queue in (startup.high_prio_queue, startup.default_queue): + mocker.patch.object(queue, "get_job_ids", return_value=[]) + mocker.patch.object( + startup.low_prio_queue, "get_job_ids", return_value=["queued"] + ) + + startup._drop_legacy_scheduler_state() + + assert redis.delete.call_args_list[0].args == ("rq:job:orphan",) + assert redis.delete.call_args_list[-1].args == startup.LEGACY_SCHEDULER_KEYS + + def test_survives_a_redis_failure(self, redis): + redis.zrange.side_effect = RuntimeError("redis gone") + + startup._drop_legacy_scheduler_state() diff --git a/docker/init_scripts/init b/docker/init_scripts/init index 9229398633..af22f0c07f 100755 --- a/docker/init_scripts/init +++ b/docker/init_scripts/init @@ -221,15 +221,29 @@ start_bin_valkey-server() { error_log "Internal valkey did not become ready after $((max_retries * 500))ms" } +# The app reads REDIS_SSL as a boolean, so "false" and "0" mean plaintext. +# Testing the variable for non-emptiness would make every documented value TLS. +redis_scheme() { + local ssl + ssl="$(printf '%s' "${REDIS_SSL-}" | tr '[:upper:]' '[:lower:]')" + case "${ssl}" in + 1 | true | yes | on) echo "rediss" ;; + *) echo "redis" ;; + esac +} + # The URL carries the password, so it goes through RQ_REDIS_URL rather than # --url, which would put it on a world-readable command line. build_redis_url() { + local scheme + scheme="$(redis_scheme)" + if [[ -n ${REDIS_PASSWORD-} ]]; then - echo "redis${REDIS_SSL:+s}://${REDIS_USERNAME-}:${REDIS_PASSWORD}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + echo "${scheme}://${REDIS_USERNAME-}:${REDIS_PASSWORD}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" elif [[ -n ${REDIS_USERNAME-} ]]; then - echo "redis${REDIS_SSL:+s}://${REDIS_USERNAME}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + echo "${scheme}://${REDIS_USERNAME}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" else - echo "redis${REDIS_SSL:+s}://${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + echo "${scheme}://${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" fi } diff --git a/entrypoint.sh b/entrypoint.sh index 780939091d..7bfd17d41a 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -47,13 +47,20 @@ else uv run python main.py & fi -# Build Redis URL properly +# Build Redis URL properly. The app reads REDIS_SSL as a boolean, so "false" +# and "0" mean plaintext; testing it for non-emptiness would make every +# documented value TLS. +REDIS_SCHEME="redis" +case "$(printf '%s' "${REDIS_SSL-}" | tr '[:upper:]' '[:lower:]')" in +1 | true | yes | on) REDIS_SCHEME="rediss" ;; +*) ;; +esac if [[ -n ${REDIS_PASSWORD-} ]]; then - REDIS_URL="redis${REDIS_SSL:+s}://${REDIS_USERNAME-}:${REDIS_PASSWORD}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + REDIS_URL="${REDIS_SCHEME}://${REDIS_USERNAME-}:${REDIS_PASSWORD}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" elif [[ -n ${REDIS_USERNAME-} ]]; then - REDIS_URL="redis${REDIS_SSL:+s}://${REDIS_USERNAME}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + REDIS_URL="${REDIS_SCHEME}://${REDIS_USERNAME}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" else - REDIS_URL="redis${REDIS_SSL:+s}://${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + REDIS_URL="${REDIS_SCHEME}://${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" fi echo "Starting RQ cron scheduler..." From e42e9dc4b08422bd9984b53ae69768c2db8c9fa7 Mon Sep 17 00:00:00 2001 From: Sam Dornan Date: Mon, 24 Aug 2026 10:29:46 -0500 Subject: [PATCH 8/8] chore(deps): drop the rq freshness-window exclusion The rolling seven-day window has reached rq 2.11.0, so `rq ~= 2.11` resolves without an exclusion of its own. Co-Authored-By: Claude Opus 5 --- pyproject.toml | 5 +---- uv.lock | 1 - 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index a8a98f2a7d..aa4032fa15 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -143,10 +143,7 @@ package = false exclude-newer = "7 days" # vcrpy >= 8.2.0 is required for aiohttp 3.14 compatibility (the removal of # `AsyncStreamReaderMixin`); allow it past the rolling 7-day window. -# rq 2.11.0 carries the scheduler locking and cron job history the scheduler -# migration builds on. The rolling window reaches it on 2026-08-24, after which -# this entry can go. -exclude-newer-package = { vcrpy = "2026-06-17", rq = "2026-08-17" } +exclude-newer-package = { vcrpy = "2026-06-17" } [tool.ty.environment] root = ["./backend"] diff --git a/uv.lock b/uv.lock index cbe9b0b3e1..f918c33cb4 100644 --- a/uv.lock +++ b/uv.lock @@ -11,7 +11,6 @@ exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for exclude-newer-span = "P7D" [options.exclude-newer-package] -rq = "2026-08-18T05:00:00Z" vcrpy = "2026-06-18T05:00:00Z" [[package]]