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 a6c32ca766..d062ee3846 100644 --- a/backend/endpoints/sockets/scan.py +++ b/backend/endpoints/sockets/scan.py @@ -2,12 +2,15 @@ import asyncio from dataclasses import dataclass +from datetime import datetime, timedelta, timezone from itertools import batched, chain from typing import Any, Final import socketio # type: ignore from rq import Worker, get_current_job +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 @@ -49,6 +52,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, @@ -70,7 +74,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 @@ -79,6 +83,10 @@ STOP_SCAN_FLAG: Final = "scan:stop" +# 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) + def _scan_platforms_func_name() -> str: """Fully qualified name RQ records for a directly enqueued scan. @@ -89,14 +97,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: @@ -105,43 +116,110 @@ 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): - job = worker.get_current_job() - if job is not None and get_job_func_name(job) in func_names: + # 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 _is_scan_job(job): return job return 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 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: - 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. - scan_platforms_func_name = _scan_platforms_func_name() - 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 _is_scan_job(job) + and get_job_status(job) == JobStatus.QUEUED ): jobs[job.id] = 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, which only the watcher sets. + + 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. + """ + registry = _scheduled_scan_registry() + jobs = Job.fetch_many(registry.get_job_ids(), connection=redis_client) + + return [job for job in jobs if job is not None and _is_scan_job(job)] + + +def get_pending_scan_jobs() -> list[Job]: + """Scans that have not started yet: queued, or waiting out a delay. + + 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. + """ + 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 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. + """ + registry = _scheduled_scan_registry() + cutoff = datetime.now(timezone.utc) - STALE_SCHEDULED_SCAN_AGE + dropped = 0 + + for job in _get_scheduled_scan_jobs(): + try: + scheduled_at = registry.get_scheduled_time(job) + except NoSuchJobError: + continue + + if scheduled_at > cutoff: + continue + + job.cancel() + dropped += 1 + log.warning(f"Dropped scan scheduled for {scheduled_at}, 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 @@ -1250,14 +1328,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") @@ -1310,7 +1392,8 @@ 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: + scheduled_jobs = _get_scheduled_scan_jobs() + for job in queued_jobs + scheduled_jobs: job.cancel() # A running scan cannot be interrupted from here, it polls the stop flag @@ -1320,11 +1403,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 038c5785a6..e27a4ec9fa 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,22 +33,8 @@ low_prio_queue, redis_client, ) -from tasks.manual.cleanup_missing_firmware import cleanup_missing_firmware_task -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( @@ -57,91 +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": "cleanup_missing_firmware", - "type": TaskType.CLEANUP, - "task": cleanup_missing_firmware_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: @@ -250,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( @@ -286,7 +197,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)) @@ -382,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()) @@ -398,9 +318,13 @@ 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. 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( - task_instance.run, - kwargs=task_kwargs or {}, + run_task_by_name, + kwargs={"name": task_name, "task_kwargs": 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 9102439604..fa2761bed7 100644 --- a/backend/handler/redis_handler.py +++ b/backend/handler/redis_handler.py @@ -1,12 +1,13 @@ import os import sys from enum import Enum +from typing import Any 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 +75,33 @@ 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 + + +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 4d58a41cf3..46daecd94e 100644 --- a/backend/startup.py +++ b/backend/startup.py @@ -4,18 +4,14 @@ 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, - 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, ) +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, @@ -26,24 +22,20 @@ 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 run_task_by_name from utils import get_version from utils.cache import conditionally_set_cache from utils.context import initialize_context @@ -53,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 @@ -75,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, @@ -95,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" @@ -110,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, @@ -127,12 +121,60 @@ 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" ) +# 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 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)) + + 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) + 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.""" @@ -140,31 +182,18 @@ async def main() -> None: async with initialize_context(): log.info("Running startup tasks") - # 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() + # 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_stale_scheduled_scans() + except Exception: + 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..2c2fd0aed7 --- /dev/null +++ b/backend/tasks/registry.py @@ -0,0 +1,52 @@ +"""The catalog of tasks an admin can see, run, or have run on a schedule.""" + +from typing import Final + +from tasks.manual.cleanup_missing_firmware import cleanup_missing_firmware_task +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, + "cleanup_missing_firmware": cleanup_missing_firmware_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 57d526264e..0410e28897 100644 --- a/backend/tasks/tasks.py +++ b/backend/tasks/tasks.py @@ -1,24 +1,40 @@ 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, 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 + job payload holds a name rather than a pickled task, and nothing in Redis + depends on where the code that runs it lives. + + Args: + name: The key the task is registered under. + 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. + """ + # Imported here because the registry imports every task module, and those + # modules import this one. + from tasks.registry import get_task + + task = get_task(name) + if task is None: + raise TaskNotFoundException(name) + + return await task.run(**(task_kwargs or {})) def update_job_meta(metadata: dict[str, Any]) -> None: @@ -84,75 +100,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.""" @@ -163,8 +116,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 fe93ec1d65..4ff1164bbc 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 @@ -33,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(): @@ -1676,26 +1679,55 @@ 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): +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 = {} + 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, *, 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 scheduled-scan 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) @@ -1703,9 +1735,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) - ) + + 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: @@ -1761,8 +1797,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)], @@ -1771,32 +1808,89 @@ 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_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)) + 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_not_called() + enqueue.assert_called_once() - 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. + 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, - scheduled=[ - make_job(scan_module.SCAN_LIBRARY_TASK_FUNC, status=JobStatus.SCHEDULED) - ], + 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): + # 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_called_once() + enqueue.assert_not_called() async def test_ignores_unrelated_jobs(self, mocker, emit): # Only scans block scans; a cleanup or metadata task must not. @@ -1901,7 +1995,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") @@ -1929,6 +2023,8 @@ async def test_cancels_watcher_scans(self, mocker, emit, redis): 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() async def test_cancels_queued_scans_with_none_running(self, mocker, emit, redis): @@ -1948,3 +2044,43 @@ async def test_no_scan_to_stop(self, mocker, emit, redis): await stop_scan_handler("sid") redis.set.assert_not_called() + + +class TestDropStaleScheduledScans: + """A worker that starts after downtime must not release a backlog.""" + + @staticmethod + 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 = 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 + job.cancel.assert_called_once() + + def test_keeps_a_scan_still_waiting_out_its_delay(self, mocker): + 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 + job.cancel.assert_not_called() + + 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 + job.cancel.assert_not_called() diff --git a/backend/tests/endpoints/test_tasks.py b/backend/tests/endpoints/test_tasks.py index aebf205d2e..03097dde7e 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( @@ -229,26 +222,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( @@ -267,8 +256,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( @@ -282,26 +271,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( @@ -315,26 +300,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( @@ -528,25 +509,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}"} ) @@ -574,26 +551,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}"}, @@ -609,3 +582,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/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/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 3ff7f69ab1..f66c09dedd 100644 --- a/backend/tests/tasks/test_tasks.py +++ b/backend/tests/tasks/test_tasks.py @@ -2,10 +2,9 @@ import httpx import pytest -from rq.job import Job -from exceptions.task_exceptions import SchedulerException -from tasks.tasks import PeriodicTask, RemoteFilePullTask, TaskType, tasks_scheduler +from exceptions.task_exceptions import TaskNotFoundException +from tasks.tasks import PeriodicTask, RemoteFilePullTask, TaskType, run_task_by_name class ConcretePeriodicTask(PeriodicTask): @@ -46,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() @@ -296,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") @@ -322,3 +161,42 @@ 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 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() + + 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) + + 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) + + with pytest.raises(TaskNotFoundException, match="some_task"): + await run_task_by_name("some_task") diff --git a/backend/tests/test_startup.py b/backend/tests/test_startup.py index 273cd16553..a4dcacea0a 100644 --- a/backend/tests/test_startup.py +++ b/backend/tests/test_startup.py @@ -1,8 +1,12 @@ """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 +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 +25,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 +56,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 +97,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 +106,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,21 +129,54 @@ 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") ) 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/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 df0947f154..874dfcbf37 100644 --- a/backend/watcher.py +++ b/backend/watcher.py @@ -3,13 +3,12 @@ import json import os from collections.abc import Sequence +from dataclasses import dataclass from datetime import timedelta from typing import cast 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 +19,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 +38,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,48 +72,49 @@ 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. +@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: - list[Job]: List of pending scan jobs that are not completed or failed + PendingScanCoverage: How many pending scans cover the whole library, and + the platforms and folders the rest are scoped to. """ - 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 + 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: @@ -194,15 +197,9 @@ 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] == [] - ] - if full_rescan_jobs: - log.info(f"Full rescan already scheduled ({len(full_rescan_jobs)} job(s))") + 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) @@ -211,16 +208,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 +230,22 @@ 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 ( + 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 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..af22f0c07f 100755 --- a/docker/init_scripts/init +++ b/docker/init_scripts/init @@ -221,42 +221,62 @@ 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 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 "${scheme}://${REDIS_USERNAME-}:${REDIS_PASSWORD}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + elif [[ -n ${REDIS_USERNAME-} ]]; then + echo "${scheme}://${REDIS_USERNAME}@${REDIS_HOST:-127.0.0.1}:${REDIS_PORT:-6379}/${REDIS_DB:-0}" + else + echo "${scheme}://${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 +285,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 +346,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 +401,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 c92636dad8..7cc82ae353 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..7bfd17d41a 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -47,34 +47,39 @@ 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 +# 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..." +# 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 +87,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 c50fb843d7..aa4032fa15 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,10 +47,7 @@ dependencies = [ "python-magic ~= 0.4", "python-socketio ~= 5.16", "redis ~= 6.2", - "rq ~= 2.7", - # 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", + "rq ~= 2.11", "sentry-sdk ~= 2.32", "starlette ~= 1.6.0", "streaming-form-data ~= 1.19", diff --git a/uv.lock b/uv.lock index 14bca9983d..7ac4793d2a 100644 --- a/uv.lock +++ b/uv.lock @@ -438,12 +438,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" @@ -2230,7 +2224,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" }, @@ -2319,8 +2312,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-scheduler", git = "https://github.com/adamantike/rq-scheduler.git?rev=feat%2Fscript-options-username-ssl" }, + { name = "rq", specifier = "~=2.11" }, { name = "sentry-sdk", specifier = "~=2.32" }, { name = "sqlalchemy", extras = ["mariadb-connector", "mysql-connector", "postgresql-psycopg"], specifier = "~=2.0" }, { name = "starlette", specifier = "~=1.6.0" }, @@ -2343,26 +2335,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" }, -] - -[[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" }, + { 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]]