fix(scan): stop refusing scans a stalled scheduler left pending

backend/endpoints/sockets/scan.py

@@ -2,12 +2,14 @@
 
 import asyncio
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
 from itertools import batched, chain
 from typing import Any, Final
 
 import pydash
 import socketio  # type: ignore
 from rq import Worker, get_current_job
+from rq.exceptions import InvalidJobOperation, NoSuchJobError
 from rq.job import Job, JobStatus
 from sqlalchemy.exc import IntegrityError
 
@@ -52,6 +54,7 @@
 from handler.metadata.ss_handler import log_scan_summary as log_ss_scan_summary
 from handler.redis_handler import (
     get_job_func_name,
+    get_job_status,
     high_prio_queue,
     low_prio_queue,
     redis_client,
@@ -80,6 +83,10 @@
 
 STOP_SCAN_FLAG: Final = "scan:stop"
 
+# A delayed watcher scan this far past due was left behind by a scheduler that
+# stopped releasing them, and the change it reacted to has long since settled.
+STALE_SCHEDULED_SCAN_AGE: Final = timedelta(hours=1)
+
 
 def _scan_platforms_func_name() -> str:
     """Fully qualified name RQ records for a directly enqueued scan.
@@ -108,41 +115,123 @@ def _get_running_scan_job() -> Job | None:
     """
     func_names = _scan_job_func_names()
     for worker in Worker.all(connection=redis_client):
-        job = worker.get_current_job()
+        # A worker killed mid-scan keeps pointing at its job until its own
+        # registration expires, and the job can be gone by then.
+        try:
+            job = worker.get_current_job()
+        except NoSuchJobError:
+            continue
+
         if job is not None and get_job_func_name(job) in func_names:
             return job
 
     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 the scheduler releases them.
     """
     func_names = _scan_job_func_names()
     jobs: dict[str, Job] = {}
 
     for job in chain(high_prio_queue.get_jobs(), low_prio_queue.get_jobs()):
-        if isinstance(job, Job) and get_job_func_name(job) in func_names:
+        if (
+            isinstance(job, Job)
+            and get_job_func_name(job) in func_names
+            and get_job_status(job) == JobStatus.QUEUED
+        ):
             jobs[job.id] = job
 
-    # The scheduler registry also holds the standing cron entry for the
-    # scheduled rescan, which is a schedule rather than a pending scan, so only
-    # delayed scan_platforms jobs count as queued here.
+    return list(jobs.values())
+
+
+def _get_scheduled_scan_jobs() -> list[Job]:
+    """Scans waiting out a delay in the scheduler, which only the watcher sets.
+
+    The scheduler is the only thing that releases them, so one that is down
+    leaves them there for good and they cannot stand in for a scan in flight.
+    """
+    # The registry also holds the standing cron entry for the scheduled rescan,
+    # which is a schedule rather than a pending scan, so only delayed
+    # scan_platforms jobs count here.
     scan_platforms_func_name = _scan_platforms_func_name()
+    jobs: dict[str, Job] = {}
+
     for job in tasks_scheduler.get_jobs():
         if (
             isinstance(job, Job)
             and get_job_func_name(job) == scan_platforms_func_name
-            and job.get_status() in (JobStatus.SCHEDULED, JobStatus.QUEUED)
+            and get_job_status(job) in (JobStatus.SCHEDULED, JobStatus.QUEUED)
         ):
             jobs[job.id] = job
 
     return list(jobs.values())
 
 
+def _cancel_scheduled_scan_job(job: Job) -> None:
+    """Drop a delayed scan from the scheduler.
+
+    Cancelling the job on its own leaves the id in the scheduler's registry, and
+    the scheduler queues it anyway once the delay is up, which runs the scan that
+    was just stopped.
+    """
+    tasks_scheduler.cancel(job)
+    try:
+        job.cancel()
+    except InvalidJobOperation:
+        # Already cancelled; the registry entry was the part that mattered.
+        pass
+
+
+def drop_stale_scheduled_scans() -> int:
+    """Drop delayed watcher scans that are long past due.
+
+    Releasing a backlog of them at once, which is what a stalled scheduler does
+    the moment it recovers, would run the same library scan over and over.
+
+    Returns:
+        int: How many scans were dropped.
+    """
+    scan_platforms_func_name = _scan_platforms_func_name()
+    cutoff = datetime.now(timezone.utc) - STALE_SCHEDULED_SCAN_AGE
+    dropped = 0
+
+    for job, scheduled_at in tasks_scheduler.get_jobs(with_times=True):
+        if (
+            not isinstance(job, Job)
+            or get_job_func_name(job) != scan_platforms_func_name
+            # The scheduler records due times as naive UTC.
+            or scheduled_at is None
+            or scheduled_at.replace(tzinfo=timezone.utc) > cutoff
+        ):
+            continue
+
+        _cancel_scheduled_scan_job(job)
+        dropped += 1
+        log.warning(f"Dropped scan scheduled for {scheduled_at} UTC, too long past due")
+
+    return dropped
+
+
+def _scan_job_label(job: Job) -> str:
+    """How to refer to a scan job when reporting it to a client."""
+    return str(job.meta.get("task_name") or "A scan")
+
+
+def _scan_in_flight_message(running: Job | None, queued: list[Job]) -> str:
+    """Say which scan is in the way, so the client knows what to wait on."""
+    if running is None:
+        return f"{_scan_job_label(queued[0])} is already queued"
+
+    if get_job_status(running) in (JobStatus.CANCELED, JobStatus.STOPPED):
+        return f"{_scan_job_label(running)} is still stopping, try again in a moment"
+
+    return f"{_scan_job_label(running)} is already running"
+
+
 @dataclass
 class ScanStats:
     total_platforms: int = 0
@@ -1331,14 +1420,18 @@ async def scan_handler(sid: str, options: dict[str, Any]):
 
     # Without this, every request enqueues another full scan behind the running
     # one, and a client that lost the progress socket has no way to tell.
-    if not DEV_MODE and (_get_running_scan_job() or _get_queued_scan_jobs()):
-        log.info(f"{emoji.EMOJI_STOP_SIGN} Scan already in progress, ignoring request")
-        await socket_handler.socket_server.emit(
-            "scan:done_ko",
-            "A scan is already in progress",
-            to=sid,
-        )
-        return
+    if not DEV_MODE:
+        running_job = _get_running_scan_job()
+        queued_jobs = _get_queued_scan_jobs()
+        if running_job is not None or queued_jobs:
+            message = _scan_in_flight_message(running_job, queued_jobs)
+            log.info(f"{emoji.EMOJI_STOP_SIGN} {message}, ignoring request")
+            await socket_handler.socket_server.emit(
+                "scan:done_ko",
+                message,
+                to=sid,
+            )
+            return
 
     log.info(f"{emoji.EMOJI_MAGNIFYING_GLASS_TILTED_RIGHT} Scanning")
 
@@ -1394,18 +1487,23 @@ async def stop_scan_handler(sid: str):
     for job in queued_jobs:
         job.cancel()
 
+    scheduled_jobs = _get_scheduled_scan_jobs()
+    for job in scheduled_jobs:
+        _cancel_scheduled_scan_job(job)
+
     # A running scan cannot be interrupted from here, it polls the stop flag
     # between platforms and ROMs and unwinds itself.
     running_job = _get_running_scan_job()
     if running_job is not None:
         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)"
     )

backend/endpoints/tasks.py

@@ -278,7 +278,13 @@ async def get_tasks_status(request: Request) -> list[TaskStatusResponse]:
     # Get currently running jobs from workers
     workers = Worker.all(connection=redis_client)
     for worker in workers:
-        current_job = worker.get_current_job()
+        # A worker killed mid-job keeps pointing at it until its own
+        # registration expires, and the job can be gone by then.
+        try:
+            current_job = worker.get_current_job()
+        except NoSuchJobError:
+            continue
+
         if current_job:
             all_tasks.append(_build_task_status_response(current_job))
 

backend/handler/redis_handler.py

@@ -5,8 +5,8 @@
 from redis import Redis
 from redis.asyncio import Redis as AsyncRedis
 from rq import Queue
-from rq.exceptions import DeserializationError
-from rq.job import Job
+from rq.exceptions import DeserializationError, InvalidJobOperation
+from rq.job import Job, JobStatus
 
 from config import IS_PYTEST_RUN, REDIS_URL
 from logger.logger import log
@@ -74,3 +74,18 @@ def get_job_func_name(job: Job, fallback: str = "") -> str:
     except DeserializationError:
         # Job data cannot be deserialized (e.g., function no longer exists)
         return fallback
+
+
+def get_job_status(job: Job) -> JobStatus | None:
+    """Safely get the status of an RQ job, which is gone once its hash expires.
+
+    Args:
+        job: The RQ Job object to get the status of
+
+    Returns:
+        The job status, or None if the job no longer has one
+    """
+    try:
+        return job.get_status()
+    except InvalidJobOperation:
+        return None

backend/startup.py

@@ -16,6 +16,7 @@
     SENTRY_DSN,
     TASK_TIMEOUT,
 )
+from endpoints.sockets.scan import drop_stale_scheduled_scans
 from handler.database import db_save_handler
 from handler.metadata.base_handler import (
     MAME_XML_KEY,
@@ -44,6 +45,7 @@
 from tasks.scheduled.update_launchbox_metadata import update_launchbox_metadata_task
 from tasks.scheduled.update_switch_titledb import update_switch_titledb_task
 from tasks.sync_push_pull_task import sync_push_pull_task
+from tasks.tasks import drop_unreadable_scheduled_jobs
 from utils import get_version
 from utils.cache import conditionally_set_cache
 from utils.context import initialize_context
@@ -140,6 +142,15 @@ async def main() -> None:
     async with initialize_context():
         log.info("Running startup tasks")
 
+        # A job the scheduler cannot read crashes it on every poll, so it has
+        # to go before the scheduler picks it up again, along with the scans that
+        # piled up behind it while nothing was being released.
+        try:
+            drop_unreadable_scheduled_jobs()
+            drop_stale_scheduled_scans()
+        except Exception:
+            log.exception("Failed to clean up the scheduler registry")
+
         # Initialize scheduled tasks
         cleanup_netplay_task.init()
         cleanup_zip_cache_task.init()

backend/tasks/tasks.py

@@ -21,6 +21,29 @@
 SCAN_LIBRARY_TASK_FUNC: Final = "tasks.scheduled.scan_library.scan_library_task.run"
 
 
+def drop_unreadable_scheduled_jobs() -> int:
+    """Remove scheduled jobs whose payload can no longer be deserialized.
+
+    The scheduler reads a job's function name before taking it out of the
+    registry, so one unreadable job left behind by an older version crashes it
+    on every poll and nothing scheduled ever runs again.
+
+    Returns:
+        int: How many jobs were dropped.
+    """
+    dropped = 0
+
+    for job in tasks_scheduler.get_jobs():
+        if not isinstance(job, Job) or get_job_func_name(job):
+            continue
+
+        tasks_scheduler.cancel(job)
+        dropped += 1
+        log.warning(f"Dropped scheduled job {job.id}, its payload cannot be read")
+
+    return dropped
+
+
 def update_job_meta(metadata: dict[str, Any]) -> None:
     """Update the current RQ job's meta data with update stats information"""
     try: