rommapp/romm
1
0
Fork 0
mirror of https://github.com/rommapp/romm.git synced 2026-10-12 02:21:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scan): refuse concurrent scans and stop queued ones

Browse Source 
`scan_handler` enqueued a scan on every request without checking whether
one was already running. A client that lost the progress socket, which
happens routinely on mobile when the tab is backgrounded, has no way to
tell a scan is still in flight, so pressing scan again looks reasonable
and silently queues a second full library scan behind the first. Three
scans of the same library ran back to back on a real instance this way.

Reject the request instead, reusing the `scan:done_ko` channel that
`reject_unauthorized_scan` already uses to report a refused scan.

`stop_scan_handler` only ever cancelled the running job, so the queued
duplicates it left behind started as soon as it stopped. It now cancels
queued scans as well.

Its first lookup loop was also dead: it searched `high_prio_queue` for a
job named `scan_platform` with `is_started` set, but RQ records the name
as `endpoints.sockets.scan.scan_platforms`, and a started job is no
longer in the queue. The second loop hardcoded the full path, which is
what the first one drifted away from, so both lookups now derive the
name from the function object and are shared with the new guard.

Finally, clear the stop flag when a scan starts. It is only cleared by
the scan that observes it, so a stop requested against a scan that ended
first would leave the flag set and abort the next scan before it began.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Ryan McClelland
2026-07-27 17:53:51 -07:00
co-authored by Claude Opus 5
parent 4b83bf0e5d
commit 99e214a8b4
2 changed files with 210 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/endpoints/sockets/scan.py
+64 -18
View File
@@ -63,6 +63,38 @@ from utils.pegasus_exporter import PegasusExporter
STOP_SCAN_FLAG: Final = "scan:stop"
def _scan_job_func_name() -> str:
"""Fully qualified name RQ records for an enqueued scan.
Derived from the function itself so it cannot drift out of sync with the
name RQ stores when the job is enqueued.
"""
return f"{scan_platforms.__module__}.{scan_platforms.__name__}"
def _get_running_scan_job() -> Job | None:
"""The scan currently executing on a worker, if any.
A started job is no longer in the queue, so it can only be found by asking
the workers what they are holding.
"""
for worker in Worker.all(connection=redis_client):
job = worker.get_current_job()
if job is not None and get_job_func_name(job) == _scan_job_func_name():
return job
return None
def _get_queued_scan_jobs() -> list[Job]:
"""Scans waiting in the queue, not yet picked up by a worker."""
return [
job
for job in high_prio_queue.get_jobs()
if get_job_func_name(job) == _scan_job_func_name()
]
@dataclass
class ScanStats:
total_platforms: int = 0
@@ -770,6 +802,10 @@ async def scan_platforms(
roms_ids (list[int], optional): List of selected roms to be scanned.
platform_fs_slugs (list[str], optional): Folders to scan with no database row.
"""
# The flag is cleared by the scan that observes it, so one set against a
# scan that ended first would otherwise stop this one before it began.
redis_client.delete(STOP_SCAN_FLAG)
if not roms_ids:
roms_ids = []
@@ -971,6 +1007,17 @@ async def scan_handler(sid: str, options: dict[str, Any]):
if await reject_unauthorized_scan(sid):
return
# 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
log.info(f"{emoji.EMOJI_MAGNIFYING_GLASS_TILTED_RIGHT} Scanning")
platform_ids = options.get("platforms", [])
@@ -1019,25 +1066,24 @@ async def stop_scan_handler(sid: str):
log.info(f"{emoji.EMOJI_STOP_BUTTON} Stop scan requested...")
async def cancel_job(job: Job):
# Queued scans have not started, so cancelling them is enough. They have to
# go too: stopping only the running scan would hand the worker the next one.
queued_jobs = _get_queued_scan_jobs()
for job in queued_jobs:
job.cancel()
# 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)
log.info(f"{emoji.EMOJI_STOP_BUTTON} Job found, stopping scan...")
existing_jobs = high_prio_queue.get_jobs()
for job in existing_jobs:
if get_job_func_name(job) == "scan_platform" and job.is_started:
return await cancel_job(job)
if running_job is None and not queued_jobs:
log.info(f"{emoji.EMOJI_STOP_BUTTON} No running scan to stop")
return
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.is_started
):
return await cancel_job(current_job)
log.info(f"{emoji.EMOJI_STOP_BUTTON} No running scan to stop")
log.info(
f"{emoji.EMOJI_STOP_BUTTON} Stopping scan "
f"({int(running_job is not None)} running, {len(queued_jobs)} queued)"
)
backend/tests/endpoints/sockets/test_scan.py
+146
View File
@@ -828,3 +828,149 @@ class TestGetPico8CoverUrl:
assert url is not None
assert fs_path in url
assert fs_name in url
class TestScanConcurrency:
"""A scan already in flight must block another from being enqueued."""
@pytest.fixture
def emit(self, mocker):
emit = AsyncMock()
mocker.patch.object(scan_module.socket_handler.socket_server, "emit", emit)
return emit
@pytest.fixture(autouse=True)
def authorized(self, mocker):
user = MagicMock()
user.oauth_scopes = [Scope.TASKS_RUN]
mocker.patch.object(
scan_module, "get_authenticated_user", AsyncMock(return_value=user)
)
mocker.patch.object(scan_module, "DEV_MODE", False)
@staticmethod
def _scan_job():
job = MagicMock()
job.func_name = "endpoints.sockets.scan.scan_platforms"
return job
@staticmethod
def _other_job():
job = MagicMock()
job.func_name = "tasks.scheduled.scan_library.scan_library_task.run"
return job
def _patch_jobs(self, mocker, *, running=None, queued=()):
worker = MagicMock()
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(queued)
)
async def test_enqueues_when_nothing_running(self, mocker, emit):
self._patch_jobs(mocker)
enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue")
await scan_handler("sid", {"type": "quick"})
enqueue.assert_called_once()
async def test_refuses_when_a_scan_is_running(self, mocker, emit):
self._patch_jobs(mocker, running=self._scan_job())
enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue")
await scan_handler("sid", {"type": "quick"})
enqueue.assert_not_called()
emit.assert_awaited_once()
assert emit.await_args.args[0] == "scan:done_ko"
async def test_refuses_when_a_scan_is_queued(self, mocker, emit):
self._patch_jobs(mocker, queued=[self._scan_job()])
enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue")
await scan_handler("sid", {"type": "quick"})
enqueue.assert_not_called()
async def test_ignores_unrelated_jobs(self, mocker, emit):
# Only scans block scans; a cleanup or metadata task must not.
self._patch_jobs(mocker, running=self._other_job(), queued=[self._other_job()])
enqueue = mocker.patch.object(scan_module.high_prio_queue, "enqueue")
await scan_handler("sid", {"type": "quick"})
enqueue.assert_called_once()
class TestStopScan:
"""Stopping must clear queued scans as well as the running one."""
@pytest.fixture
def emit(self, mocker):
emit = AsyncMock()
mocker.patch.object(scan_module.socket_handler.socket_server, "emit", emit)
return emit
@pytest.fixture(autouse=True)
def authorized(self, mocker):
user = MagicMock()
user.oauth_scopes = [Scope.TASKS_RUN]
mocker.patch.object(
scan_module, "get_authenticated_user", AsyncMock(return_value=user)
)
@pytest.fixture
def redis(self, mocker):
return mocker.patch.object(scan_module, "redis_client")
@staticmethod
def _scan_job():
job = MagicMock()
job.func_name = "endpoints.sockets.scan.scan_platforms"
return job
def _patch_jobs(self, mocker, *, running=None, queued=()):
worker = MagicMock()
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(queued)
)
async def test_sets_stop_flag_for_running_scan(self, mocker, emit, redis):
running = self._scan_job()
self._patch_jobs(mocker, running=running)
await stop_scan_handler("sid")
running.cancel.assert_called_once()
redis.set.assert_called_once_with(scan_module.STOP_SCAN_FLAG, 1)
async def test_cancels_queued_scans(self, mocker, emit, redis):
queued = [self._scan_job(), self._scan_job()]
self._patch_jobs(mocker, running=self._scan_job(), queued=queued)
await stop_scan_handler("sid")
for job in queued:
job.cancel.assert_called_once()
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,
# and must not leave a stop flag behind for the next scan to trip on.
queued = [self._scan_job()]
self._patch_jobs(mocker, queued=queued)
await stop_scan_handler("sid")
queued[0].cancel.assert_called_once()
redis.set.assert_not_called()
async def test_no_scan_to_stop(self, mocker, emit, redis):
self._patch_jobs(mocker)
await stop_scan_handler("sid")
redis.set.assert_not_called()