Background Workers and Tasks
Relevant source files
The following files were used as context for generating this wiki page:
The background worker system in Athena is designed to offload heavy processing tasks from the primary application process to ensure immediate responses for user-facing actions like authentication and chat. Athena utilizes a two-process model where the app process handles synchronous I/O and the worker process manages resource-intensive operations such as performance points (PP) calculation, leaderboard updates, and media fetching.
Sources: AGENTS.md:329-333, README.md:36-39
System Architecture
The task system is built on taskiq with taskiq-redis (Valkey) as the message broker. The architecture follows a hexagonal adapter pattern where background jobs act as thin wrappers around core business use-cases.
The diagram shows the flow of a background task from the application process, through the Valkey broker, to the worker process where it is executed by a job adapter and serviced by the core domain logic.
Sources: AGENTS.md:39, AGENTS.md:214-220, README.md:52-53
Components and Responsibilities
Worker Entry Point
The worker is initialized via the osu_server.worker:broker module. It is responsible for setting up the taskiq broker and managing the lifecycle of the worker process.
Sources: AGENTS.md:73, README.md:113
Job Adapters
Job adapters live in the src/osu_server/jobs/ directory. Their primary responsibilities are:
- Validating primitive task payloads.
- Mapping primitives to domain-specific command/query inputs.
- Resolving use-cases via the Dishka dependency injection framework.
- Invoking business logic and reporting success or failure.
Crucially, job adapters must remain "thin." They are prohibited from containing business rules, idempotency logic, persistence consistency checks, or direct database access.
Sources: AGENTS.md:214-220, tests/unit/test_job_boundaries.py:59-69
Job Registry
The JobRegistry manages the names and metadata of all background tasks available to the worker. This registry ensures that tasks are correctly mapped to their adapter functions.
Sources: src/osu_server/infrastructure/jobs/registry.py, tests/unit/test_job_boundaries.py:72-76
Task Inventory
The following table summarizes key background tasks identified in the system:
| Task Name | Category | Description |
|---|---|---|
calculate_score_performance | Scores | Calculates PP and performance metrics for a submitted score. |
fetch_beatmap_metadata | Beatmaps | Retrieves metadata for a beatmap from external sources. |
fetch_beatmap_file | Beatmaps | Downloads the .osu file for a specific beatmap. |
rebuild_beatmap_leaderboards_for_user | Leaderboards | Updates all relevant leaderboards after a user's performance changes. |
persist_channel_message | Chat | Asynchronously saves chat messages to durable storage. |
account_replay_download | Stats | Tracks and records replay download metrics. |
Sources: tests/unit/composition/test_common_provider_graph.py:112-120, tests/unit/jobs/test_beatmap_fetch_tasks.py:61-63
Dependency Injection and Boundary Rules
Tasks resolve their dependencies through the Dishka integration. The worker process uses a specific WorkerProviderGraph that excludes application-only dependencies (like web endpoints) to maintain clean architectural boundaries.
Strict Boundary Rules
To maintain the integrity of the modular monolith, the job layer is subject to strict import rules:
- No Transports: Jobs must not import from
osu_server.transports. - Persistence Independence: Jobs must not import concrete SQLAlchemy models or database sessions directly. They must use repository interfaces.
- Thin Adapters: Jobs stay persistence-adapter independent and do not include low-level infrastructure access.
Sources: AGENTS.md:218-220, pyproject.toml:238-260, tests/unit/test_job_boundaries.py:59-69
Execution Flow Example: Beatmap Fetching
When a beatmap fetch task is triggered, the worker executes the following sequence:
The task adapter resolves the required use-case from the taskiq state (provided by Dishka) and delegates the actual work to the domain service. If the required service is not in the state, the task raises a RuntimeError and logs a runtime_unavailable event.
Sources: src/osu_server/jobs/beatmap_fetch.py, tests/unit/jobs/test_beatmap_fetch_tasks.py:84-120
Summary
The Background Worker and Task system in Athena provides a robust framework for asynchronous processing. By utilizing taskiq for task management and Dishka for dependency injection, the system ensures that heavy processing does not block user interactions. Strict architectural boundaries ensure that job adapters remain thin, delegating all business logic to the core domain services while maintaining process isolation between the web application and background workers.