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

Every scan request has been refused with "A scan is already in progress"
on instances whose rq-scheduler cannot make progress, and waiting does
not help because nothing is running to wait for.

The scheduler reads a job's function name before it takes the job out of
its registry, so one job left behind by an older version, holding an
argument that no longer deserializes, crashes it on every poll and stays
there for the next one to trip on. Nothing scheduled runs again, and the
watcher's delayed rescans pile up behind it. `_get_queued_scan_jobs`
counted those as scans waiting to start, so the guard added in 99e214a8b
refused every manual scan for as long as the scheduler stayed broken.

Clear jobs the scheduler cannot read at startup, before it polls them
again, and split scan discovery so the guard consults only what sits on
a worker queue. Delayed scans still block nothing: the scheduler is the
only thing that releases them, so one that is down must not be able to
refuse scans. A recovered scheduler would release its whole backlog at
once and run the same library scan over and over, so delayed scans more
than an hour past due go too.

Stopping a scan now drops delayed scans out of the scheduler's registry
rather than only cancelling the job, which left the id in the registry
for the scheduler to queue anyway once the delay was up.

A worker killed mid-scan points at a job that can already be gone, which
raised NoSuchJobError out of the socket handler with no reply to the
client, and out of GET /tasks/status as a 500. Reading a job's status has
the same problem once its hash expires, so it goes through a wrapper.

Finally, name the scan in the way when refusing, and say when it is
stopping rather than running, so the message is actionable.

Fixes #4186

Co-Authored-By: Claude Opus 5 <[email protected]>

backend/endpoints/sockets/scan.py

@@ -2,12 +2,14 @@ from __future__ import annotations
 
 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_quota as log_ss_quota
 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 @@ from utils.pegasus_exporter import PegasusExporter
 
 STOP_SCAN_FLAG: Final = "scan:stop"
 
+# A delayed watcher scan this far past due was left behind by a scheduler that
+# stopped releasing them, and the change it reacted to has long since settled.
+STALE_SCHEDULED_SCAN_AGE: Final = timedelta(hours=1)
+
 
 def _scan_platforms_func_name() -> str:
     """Fully qualified name RQ records for a directly enqueued scan.
@@ -108,7 +115,13 @@ def _get_running_scan_job() -> Job | None:
     """
     func_names = _scan_job_func_names()
     for worker in Worker.all(connection=redis_client):
-        job = worker.get_current_job()
+        # A worker killed mid-scan keeps pointing at its job until its own
+        # registration expires, and the job can be gone by then.
+        try:
+            job = worker.get_current_job()
+        except NoSuchJobError:
+            continue
+
         if job is not None and get_job_func_name(job) in func_names:
             return job
 
@@ -116,33 +129,109 @@ def _get_running_scan_job() -> Job | None:
 
 
 def _get_queued_scan_jobs() -> list[Job]:
-    """Scans waiting to run, not yet picked up by a worker.
+    """Scans sitting on a worker queue, waiting to be picked up.
 
-    Socket scans sit in the high priority queue, while watcher scans are delayed
-    through the scheduler before landing in the low priority queue.
+    Socket scans go to the high priority queue and watcher scans to the low
+    priority one, once the scheduler releases them.
     """
     func_names = _scan_job_func_names()
     jobs: dict[str, Job] = {}
 
     for job in chain(high_prio_queue.get_jobs(), low_prio_queue.get_jobs()):
-        if isinstance(job, Job) and get_job_func_name(job) in func_names:
-            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 get_job_func_name(job) in func_names
+            and get_job_status(job) == JobStatus.QUEUED
         ):
             jobs[job.id] = job
 
     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 get_job_status(job) in (JobStatus.SCHEDULED, JobStatus.QUEUED)
+        ):
+            jobs[job.id] = job
+
+    return list(jobs.values())
+
+
+def _cancel_scheduled_scan_job(job: Job) -> None:
+    """Drop a delayed scan from the scheduler.
+
+    Cancelling the job on its own leaves the id in the scheduler's registry, and
+    the scheduler queues it anyway once the delay is up, which runs the scan that
+    was just stopped.
+    """
+    tasks_scheduler.cancel(job)
+    try:
+        job.cancel()
+    except InvalidJobOperation:
+        # Already cancelled; the registry entry was the part that mattered.
+        pass
+
+
+def drop_stale_scheduled_scans() -> int:
+    """Drop delayed watcher scans that are long past due.
+
+    Releasing a backlog of them at once, which is what a stalled scheduler does
+    the moment it recovers, would run the same library scan over and over.
+
+    Returns:
+        int: How many scans were dropped.
+    """
+    scan_platforms_func_name = _scan_platforms_func_name()
+    cutoff = datetime.now(timezone.utc) - STALE_SCHEDULED_SCAN_AGE
+    dropped = 0
+
+    for job, scheduled_at in tasks_scheduler.get_jobs(with_times=True):
+        if (
+            not isinstance(job, Job)
+            or get_job_func_name(job) != scan_platforms_func_name
+            # The scheduler records due times as naive UTC.
+            or scheduled_at is None
+            or scheduled_at.replace(tzinfo=timezone.utc) > cutoff
+        ):
+            continue
+
+        _cancel_scheduled_scan_job(job)
+        dropped += 1
+        log.warning(f"Dropped scan scheduled for {scheduled_at} UTC, too long past due")
+
+    return dropped
+
+
+def _scan_job_label(job: Job) -> str:
+    """How to refer to a scan job when reporting it to a client."""
+    return str(job.meta.get("task_name") or "A scan")
+
+
+def _scan_in_flight_message(running: Job | None, queued: list[Job]) -> str:
+    """Say which scan is in the way, so the client knows what to wait on."""
+    if running is None:
+        return f"{_scan_job_label(queued[0])} is already queued"
+
+    if get_job_status(running) in (JobStatus.CANCELED, JobStatus.STOPPED):
+        return f"{_scan_job_label(running)} is still stopping, try again in a moment"
+
+    return f"{_scan_job_label(running)} is already running"
+
+
 @dataclass
 class ScanStats:
     total_platforms: int = 0
@@ -1331,14 +1420,18 @@ async def scan_handler(sid: str, options: dict[str, Any]):
 
     # Without this, every request enqueues another full scan behind the running
     # one, and a client that lost the progress socket has no way to tell.
-    if not DEV_MODE and (_get_running_scan_job() or _get_queued_scan_jobs()):
-        log.info(f"{emoji.EMOJI_STOP_SIGN} Scan already in progress, ignoring request")
-        await socket_handler.socket_server.emit(
-            "scan:done_ko",
-            "A scan is already in progress",
-            to=sid,
-        )
-        return
+    if not DEV_MODE:
+        running_job = _get_running_scan_job()
+        queued_jobs = _get_queued_scan_jobs()
+        if running_job is not None or queued_jobs:
+            message = _scan_in_flight_message(running_job, queued_jobs)
+            log.info(f"{emoji.EMOJI_STOP_SIGN} {message}, ignoring request")
+            await socket_handler.socket_server.emit(
+                "scan:done_ko",
+                message,
+                to=sid,
+            )
+            return
 
     log.info(f"{emoji.EMOJI_MAGNIFYING_GLASS_TILTED_RIGHT} Scanning")
 
@@ -1394,6 +1487,10 @@ async def stop_scan_handler(sid: str):
     for job in queued_jobs:
         job.cancel()
 
+    scheduled_jobs = _get_scheduled_scan_jobs()
+    for job in scheduled_jobs:
+        _cancel_scheduled_scan_job(job)
+
     # A running scan cannot be interrupted from here, it polls the stop flag
     # between platforms and ROMs and unwinds itself.
     running_job = _get_running_scan_job()
@@ -1401,11 +1498,12 @@ async def stop_scan_handler(sid: str):
         running_job.cancel()
         redis_client.set(STOP_SCAN_FLAG, 1)
 
-    if running_job is None and not queued_jobs:
+    if running_job is None and not queued_jobs and not scheduled_jobs:
         log.info(f"{emoji.EMOJI_STOP_BUTTON} No running scan to stop")
         return
 
     log.info(
         f"{emoji.EMOJI_STOP_BUTTON} Stopping scan "
-        f"({int(running_job is not None)} running, {len(queued_jobs)} queued)"
+        f"({int(running_job is not None)} running, {len(queued_jobs)} queued, "
+        f"{len(scheduled_jobs)} scheduled)"
     )

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 enum import Enum
 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 @@ from config import (
     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.sync_retroachievements_progress import (
 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 @@ tasks_scheduler = Scheduler(queue=low_prio_queue, connection=low_prio_queue.conn
 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:

backend/tests/endpoints/sockets/test_scan.py

@@ -1,9 +1,11 @@
+from datetime import datetime, timedelta, timezone
 from itertools import count
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, Mock
 
 import pytest
 import socketio
+from rq.exceptions import InvalidJobOperation, NoSuchJobError
 from rq.job import Job, JobStatus
 
 from endpoints.sockets import scan as scan_module
@@ -1687,21 +1689,35 @@ 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):
     """An RQ job stub that scan job discovery will accept."""
     job = MagicMock(spec=Job)
     job.id = f"job-{next(_job_ids)}"
     job.func_name = func_name
     job.get_status.return_value = status
+    job.meta = {"task_name": task_name} if task_name else {}
     return job
 
 
 def patch_scan_jobs(
-    mocker, *, running=None, high_queued=(), low_queued=(), scheduled=()
-):
-    """Point every place scan discovery looks at a fixed set of jobs."""
+    mocker,
+    *,
+    running=None,
+    high_queued=(),
+    low_queued=(),
+    scheduled=(),
+    worker_lost=False,
+) -> MagicMock:
+    """Point every place scan discovery looks at a fixed set of jobs.
+
+    Returns the patched scheduler cancel, the only thing that drops a delayed
+    scan out of the scheduler's registry.
+    """
     worker = MagicMock()
-    worker.get_current_job.return_value = running
+    if worker_lost:
+        worker.get_current_job.side_effect = NoSuchJobError
+    else:
+        worker.get_current_job.return_value = running
     mocker.patch.object(scan_module.Worker, "all", return_value=[worker])
     mocker.patch.object(
         scan_module.high_prio_queue, "get_jobs", return_value=list(high_queued)
@@ -1712,6 +1728,7 @@ def patch_scan_jobs(
     mocker.patch.object(
         scan_module.tasks_scheduler, "get_jobs", return_value=list(scheduled)
     )
+    return mocker.patch.object(scan_module.tasks_scheduler, "cancel")
 
 
 class TestScanConcurrency:
@@ -1767,8 +1784,9 @@ class TestScanConcurrency:
 
         enqueue.assert_not_called()
 
-    async def test_refuses_when_a_watcher_scan_is_scheduled(self, mocker, emit):
-        # A watcher scan waits out its delay in the scheduler before it queues.
+    async def test_a_scheduled_watcher_scan_does_not_block(self, mocker, emit):
+        # A delayed watcher scan only moves when the scheduler releases it, so
+        # a scheduler that is down would refuse manual scans for good.
         patch_scan_jobs(
             mocker,
             scheduled=[make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED)],
@@ -1777,7 +1795,79 @@ class TestScanConcurrency:
 
         await scan_handler("sid", {"type": "quick"})
 
-        enqueue.assert_not_called()
+        enqueue.assert_called_once()
+
+    async def test_a_cancelled_queued_scan_does_not_block(self, mocker, emit):
+        patch_scan_jobs(
+            mocker,
+            high_queued=[make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.CANCELED)],
+        )
+        enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue")
+
+        await scan_handler("sid", {"type": "quick"})
+
+        enqueue.assert_called_once()
+
+    async def test_a_queued_scan_whose_status_is_gone_does_not_block(
+        self, mocker, emit
+    ):
+        # RQ raises rather than reporting a status once the job hash expires.
+        job = make_job(SCAN_PLATFORMS_FUNC)
+        job.get_status.side_effect = InvalidJobOperation
+        patch_scan_jobs(mocker, high_queued=[job])
+        enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue")
+
+        await scan_handler("sid", {"type": "quick"})
+
+        enqueue.assert_called_once()
+
+    async def test_a_worker_holding_a_job_that_is_gone_does_not_block(
+        self, mocker, emit
+    ):
+        patch_scan_jobs(mocker, worker_lost=True)
+        enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue")
+
+        await scan_handler("sid", {"type": "quick"})
+
+        enqueue.assert_called_once()
+
+    async def test_refusal_names_the_scan_in_the_way(self, mocker, emit):
+        patch_scan_jobs(
+            mocker,
+            running=make_job(
+                SCAN_PLATFORMS_FUNC, status=JobStatus.STARTED, task_name="Quick Scan"
+            ),
+        )
+        mocker.patch.object(scan_module.high_prio_queue, "enqueue")
+
+        await scan_handler("sid", {"type": "quick"})
+
+        assert emit.await_args.args[1] == "Quick Scan is already running"
+
+    async def test_refusal_says_a_stopping_scan_is_worth_retrying(self, mocker, emit):
+        # A stopped scan holds the worker until it unwinds, which reads as a
+        # scan in progress unless the client is told to come back.
+        patch_scan_jobs(
+            mocker,
+            running=make_job(
+                SCAN_PLATFORMS_FUNC, status=JobStatus.CANCELED, task_name="Quick Scan"
+            ),
+        )
+        mocker.patch.object(scan_module.high_prio_queue, "enqueue")
+
+        await scan_handler("sid", {"type": "quick"})
+
+        assert "still stopping" in emit.await_args.args[1]
+
+    async def test_refusal_reports_a_queued_scan_as_queued(self, mocker, emit):
+        patch_scan_jobs(
+            mocker, high_queued=[make_job(SCAN_PLATFORMS_FUNC, task_name="Full Scan")]
+        )
+        mocker.patch.object(scan_module.high_prio_queue, "enqueue")
+
+        await scan_handler("sid", {"type": "quick"})
+
+        assert emit.await_args.args[1] == "Full Scan is already queued"
 
     async def test_refuses_when_the_scheduled_rescan_is_running(self, mocker, emit):
         # The scheduled rescan runs scan_platforms from inside its own task, so
@@ -1930,12 +2020,28 @@ class TestStopScan:
         # watcher's scan the moment the running one unwinds.
         low_queued = make_job(SCAN_PLATFORMS_FUNC)
         scheduled = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED)
-        patch_scan_jobs(mocker, low_queued=[low_queued], scheduled=[scheduled])
+        scheduler_cancel = patch_scan_jobs(
+            mocker, low_queued=[low_queued], scheduled=[scheduled]
+        )
 
         await stop_scan_handler("sid")
 
         low_queued.cancel.assert_called_once()
         scheduled.cancel.assert_called_once()
+        # Cancelling the job leaves its id in the scheduler, which queues it
+        # anyway once the delay is up.
+        scheduler_cancel.assert_called_once_with(scheduled)
+
+    async def test_drops_a_scheduled_scan_that_was_already_cancelled(
+        self, mocker, emit, redis
+    ):
+        scheduled = make_job(SCAN_PLATFORMS_FUNC, status=JobStatus.SCHEDULED)
+        scheduled.cancel.side_effect = InvalidJobOperation
+        scheduler_cancel = patch_scan_jobs(mocker, scheduled=[scheduled])
+
+        await stop_scan_handler("sid")
+
+        scheduler_cancel.assert_called_once_with(scheduled)
 
     async def test_cancels_queued_scans_with_none_running(self, mocker, emit, redis):
         # Stopping a scan that has not been picked up yet must still drop it,
@@ -1954,3 +2060,45 @@ class TestStopScan:
         await stop_scan_handler("sid")
 
         redis.set.assert_not_called()
+
+
+class TestDropStaleScheduledScans:
+    """A recovered scheduler must not release a backlog of delayed scans."""
+
+    @staticmethod
+    def _scheduled(func_name: str, *, hours_late: float):
+        # The scheduler records due times as naive UTC.
+        due = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(
+            hours=hours_late
+        )
+        return (make_job(func_name, status=JobStatus.SCHEDULED), due)
+
+    def _patch(self, mocker, jobs):
+        mocker.patch.object(
+            scan_module.tasks_scheduler, "get_jobs", return_value=list(jobs)
+        )
+        return mocker.patch.object(scan_module.tasks_scheduler, "cancel")
+
+    def test_drops_a_scan_long_past_due(self, mocker):
+        job, due = self._scheduled(SCAN_PLATFORMS_FUNC, hours_late=2)
+        scheduler_cancel = self._patch(mocker, [(job, due)])
+
+        assert scan_module.drop_stale_scheduled_scans() == 1
+        scheduler_cancel.assert_called_once_with(job)
+
+    def test_keeps_a_scan_still_waiting_out_its_delay(self, mocker):
+        scheduler_cancel = self._patch(
+            mocker, [self._scheduled(SCAN_PLATFORMS_FUNC, hours_late=-1)]
+        )
+
+        assert scan_module.drop_stale_scheduled_scans() == 0
+        scheduler_cancel.assert_not_called()
+
+    def test_leaves_the_standing_rescan_cron_entry_alone(self, mocker):
+        # The cron entry reschedules itself, so a missed run is not a backlog.
+        scheduler_cancel = self._patch(
+            mocker, [self._scheduled(scan_module.SCAN_LIBRARY_TASK_FUNC, hours_late=2)]
+        )
+
+        assert scan_module.drop_stale_scheduled_scans() == 0
+        scheduler_cancel.assert_not_called()

backend/tests/tasks/test_tasks.py

@@ -1,11 +1,18 @@
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
 
 import httpx
 import pytest
+from rq.exceptions import DeserializationError
 from rq.job import Job
 
 from exceptions.task_exceptions import SchedulerException
-from tasks.tasks import PeriodicTask, RemoteFilePullTask, TaskType, tasks_scheduler
+from tasks.tasks import (
+    PeriodicTask,
+    RemoteFilePullTask,
+    TaskType,
+    drop_unreadable_scheduled_jobs,
+    tasks_scheduler,
+)
 
 
 class ConcretePeriodicTask(PeriodicTask):
@@ -322,3 +329,34 @@ class TestRemoteFilePullTask:
         result = await disabled_task.run(force=True)
 
         assert result == b"forced content"
+
+
+class TestDropUnreadableScheduledJobs:
+    """One unreadable job crashes the scheduler on every poll until it is gone."""
+
+    @staticmethod
+    def _job(func_name: str | None):
+        job = MagicMock(spec=Job)
+        job.id = "job-1"
+        if func_name is None:
+            type(job).func_name = PropertyMock(side_effect=DeserializationError)
+        else:
+            job.func_name = func_name
+        return job
+
+    @patch.object(tasks_scheduler, "cancel")
+    @patch.object(tasks_scheduler, "get_jobs")
+    def test_drops_a_job_whose_payload_cannot_be_read(self, get_jobs, cancel):
+        job = self._job(None)
+        get_jobs.return_value = [job]
+
+        assert drop_unreadable_scheduled_jobs() == 1
+        cancel.assert_called_once_with(job)
+
+    @patch.object(tasks_scheduler, "cancel")
+    @patch.object(tasks_scheduler, "get_jobs")
+    def test_leaves_readable_jobs_scheduled(self, get_jobs, cancel):
+        get_jobs.return_value = [self._job("test.function")]
+
+        assert drop_unreadable_scheduled_jobs() == 0
+        cancel.assert_not_called()