How to Build Background Scan Tasks with Redis RQ in RomM

RomM uses Redis RQ (Redis Queue) to run long‑running library scans asynchronously, leveraging a low_prio_queue, reusable PeriodicTask base classes, and a global Scheduler to handle cron‑style job dispatch.

RomM’s background task architecture is designed to keep the web API responsive while handling I/O‑heavy operations like scanning ROM folders. According to the RomM source code, the system centers on a low‑priority RQ queue and abstract task classes that handle serialization, scheduling, and live progress reporting. This guide explains how to implement custom scan tasks using the exact patterns found in backend/tasks/tasks.py and backend/handler/redis_handler.py.

Architecture Overview

RomM delegates heavy lifting to RQ workers through three core components:

• low_prio_queue – Defined in backend/handler/redis_handler.py, this is the shared Redis queue where all scan jobs are enqueued.
• Task / PeriodicTask – Abstract base classes in backend/tasks/tasks.py that enforce a uniform run coroutine and manage job metadata.
• Scheduler – Instantiated in backend/tasks/tasks.py, providing cron‑style scheduling for recurring scans.

The actual scanning logic lives in backend/handler/scan_handler.py, which contains functions like scan_platform and scan_all_platforms that workers execute outside the main web process.

Defining a Scan Task

To create a background scan task, subclass PeriodicTask and implement the asynchronous run method. The func parameter must be a dotted import path to the target function that the RQ worker will resolve and execute.

# backend/tasks/library_scan.py

from tasks import PeriodicTask, TaskType
from handler.scan_handler import scan_all_platforms

class LibraryScanTask(PeriodicTask):
    """Periodic task that triggers a full library scan."""

    def __init__(self) -> None:
        super().__init__(
            title="Library Scan",
            description="Scans all ROM folders for new/updated games",
            task_type=TaskType.SCAN,
            enabled=True,
            manual_run=False,
            cron_string="0 * * * *",  # Every hour

            func="backend.handler.scan_handler.scan_all_platforms",
        )

    async def run(self, *_, **__) -> None:
        await scan_all_platforms()

Key implementation details:

• func – A string import path (e.g., "backend.handler.scan_handler.scan_all_platforms"). The helper get_job_func_name in backend/handler/redis_handler.py safeguards against deserialization errors when the worker resolves this path via importlib.
• cron_string – Optional crontab syntax. When provided, Scheduler.cron registers the job automatically.
• enabled – A boolean flag that the UI toggles. Calling task.init() inspects this flag to either schedule or unschedule the job.

Scheduling the Task at Startup

RomM initializes background tasks during application startup in backend/startup.py. The process creates the global tasks_scheduler (from rq_scheduler.Scheduler), registers the low_prio_queue, and iterates over task subclasses to initialize them.

# backend/startup.py (excerpt)

from tasks.library_scan import LibraryScanTask
from handler.redis_handler import low_prio_queue

def init_background_tasks() -> None:
    # Initialise periodic tasks – they will be scheduled if enabled

    LibraryScanTask().init()

When the web process starts, init() checks the task’s enabled status. If True, it calls schedule(), which registers the job with tasks_scheduler.cron using the specified cron_string.

Running the Job in an RQ Worker

An RQ worker process must run alongside the web application to consume jobs from the low_prio_queue. The worker uses the same Redis connection string defined in the application configuration.

# Example Docker Compose or local development command

rq worker --url redis://redis:6379/0 low_prio_queue

The worker performs the following steps:

1. Dequeues the job from low_prio_queue.
2. Deserializes the func path using get_job_func_name from backend/handler/redis_handler.py.
3. Imports and executes the target function (e.g., scan_all_platforms).
4. Catches and logs exceptions silently while preserving job metadata.

Updating Job Metadata for Live Progress

While a scan runs, the task can report statistics to the frontend via WebSocket. Call update_job_meta (defined in backend/tasks/tasks.py) to store progress information in the job’s Redis metadata.

# Inside backend/handler/scan_handler.py

from tasks import update_job_meta

async def scan_platform(platform, new_roms):
    # ... scanning logic ...

    update_job_meta({
        "platform_scanned": platform.id,
        "roms_found": len(new_roms),
    })

The update_job_meta function writes to current_job.meta, which is then exposed through the /ws endpoint handled by backend/handler/socket_handler.py. This allows the React frontend to display real-time progress bars and status updates.

Complete Execution Flow

Here is the end‑to‑end lifecycle of a scheduled scan task:

1. User enables “Automatic library scan” in the RomM UI, triggering a PATCH to /api/tasks/library-scan.
2. Backend updates the task’s enabled flag in the database and calls LibraryScanTask().init().
3. PeriodicTask.init() detects the enabled state and invokes schedule(), which registers a cron job with tasks_scheduler.cron.
4. At the scheduled time, the RQ worker picks up the job, resolves func="backend.handler.scan_handler.scan_all_platforms", and executes it.
5. During execution, scan_all_platforms iterates over platforms, calling scan_platform and invoking update_job_meta after each platform completes.
6. WebSocket pushes metadata updates to connected clients, rendering live progress in the UI.

Source Code Reference

File	Purpose
backend/tasks/tasks.py	Core Task and PeriodicTask classes, update_job_meta, and the global tasks_scheduler.
backend/handler/redis_handler.py	low_prio_queue definition and get_job_func_name deserialization helper.
backend/handler/scan_handler.py	Heavy scanning functions (scan_platform, scan_all_platforms).
backend/startup.py	Application startup hook that initializes the RQ queue and all background tasks.

Summary

• Use PeriodicTask as the base class for any recurring scan job, supplying a dotted import path to the func parameter.
• Initialize tasks in backend/startup.py by calling .init() on each task instance; this handles scheduling based on the enabled flag.
• Run workers via rq worker --url redis://redis:6379/0 low_prio_queue to process jobs outside the web server.
• Report progress by calling update_job_meta() inside scan functions to enable real-time UI updates via WebSocket.
• Leverage built-in safeguards like get_job_func_name to prevent deserialization errors when RQ resolves function paths.

Frequently Asked Questions

What is the difference between Task and PeriodicTask in RomM?

Task is the abstract base class in backend/tasks/tasks.py that provides the foundation for all background jobs, including metadata handling and the run coroutine. PeriodicTask extends Task to add cron‑style scheduling via the Scheduler and automatically manages enrollment in the low_prio_queue when init() is called.

How does RomM prevent function deserialization errors in RQ workers?

RomM uses the get_job_func_name helper located in backend/handler/redis_handler.py. This function validates and sanitizes the dotted import path stored in the job payload before the worker attempts to resolve it with importlib, preventing common serialization mismatches between the web process and worker processes.

Can I trigger a scan task manually without waiting for the cron schedule?

Yes. While PeriodicTask includes a cron_string for automatic scheduling, you can also invoke the task’s run method directly or enqueue it manually to low_prio_queue using RQ’s standard API. The manual_run flag in the task definition controls whether the UI exposes a manual trigger button.

How does the frontend receive live updates during a background scan?

The update_job_meta function writes progress data (such as platform_scanned and roms_found) to the job’s Redis metadata. The WebSocket handler in backend/handler/socket_handler.py polls this metadata and pushes updates to connected clients via the /ws endpoint, allowing the React frontend to display real-time progress indicators.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}