refactor(tasks): schedule with RQ instead of rq-scheduler

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
 

backend/endpoints/sockets/scan.py

@@ -2,13 +2,16 @@
 
 import asyncio
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
 from itertools import batched, chain
 from typing import Any, Final
 
 import pydash
 import socketio  # type: ignore
 from rq import Worker, get_current_job
+from rq.exceptions import 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
@@ -52,6 +55,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,
@@ -71,7 +75,7 @@
 from models.firmware import Firmware
 from models.platform import Platform
 from models.rom import Rom
-from tasks.tasks import SCAN_LIBRARY_TASK_FUNC, tasks_scheduler, update_job_meta
+from tasks.tasks import update_job_meta
 from utils import emoji
 from utils.audio_tags import remove_persisted_cover
 from utils.context import initialize_context
@@ -80,6 +84,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.
@@ -90,14 +98,17 @@ def _scan_platforms_func_name() -> str:
     return f"{scan_platforms.__module__}.{scan_platforms.__name__}"
 
 
-def _scan_job_func_names() -> frozenset[str]:
-    """Every job function name that ends up running a scan.
+def _is_scan_job(job: Job) -> bool:
+    """Whether this job runs a scan.
 
-    Socket and watcher scans enqueue scan_platforms itself, while the scheduled
-    rescan enqueues its own task and calls scan_platforms in process. Both have
-    to be recognised or an in-flight scan goes unseen.
+    Socket and watcher scans enqueue scan_platforms itself. Task-driven scans go
+    through the task runner, which every task shares, so they are recognised by
+    the type their job carries instead.
     """
-    return frozenset((_scan_platforms_func_name(), SCAN_LIBRARY_TASK_FUNC))
+    if get_job_func_name(job) == _scan_platforms_func_name():
+        return True
+
+    return job.meta.get("task_type") == TaskType.SCAN
 
 
 def _get_running_scan_job() -> Job | None:
@@ -106,43 +117,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
@@ -1331,14 +1409,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")
 
@@ -1391,7 +1473,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
@@ -1401,11 +1484,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)"
     )