perf(tasks): batch RetroAchievements progress sync reads (#5144)

* perf(tasks): batch RetroAchievements progress sync reads

Add get_roms_by_ra_ids and get_rom_users_by_rom_ids (single index-backed
IN queries, no eager joins) and fold the per-user feed to {ra_id: status}
so each user costs 2 reads plus writes instead of 2 reads per game.

* fix(tasks): skip joined eager-loads in RA sync batch reads

* style(tasks): trim batch-read comments to house style

* fix(tasks): sync every ROM sharing an ra_id, skip empty rom set early

* fix(tasks): raiseload instead of deprecated noload, fix CI failures

* fix(tasks): apply code-review fixes to RA progress sync batching

pr-ready step 2 (code-review xhigh --fix): load only id/ra_id for the
batched ROM read, flatten the sync loop, isolate per-user status sync
failures, log the ROM id, add a real-row test, and move the batched
RomUser getter next to its siblings.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

* refactor(tasks): simplify RA progress sync batching and its tests

pr-ready step 3 (simplify): select (id, ra_id) columns instead of Rom
entities, drop early returns the getters already short-circuit, and
fold the mocked sync tests into shared helpers and one parametrized
status-transition test.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

* style(tasks): trim batched RomUser getter comment

pr-ready step 4 (review-polish).

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

* docs(tasks): tighten unknown-award test docstring

pr-ready step 5 (humanizer).

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

---------

Co-authored-by: Georges-Antoine Assi <[email protected]>
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
Shreeniwas
2026-10-07 12:43:39 -04:00
committed by GitHub
co-authored by Claude Opus 5.5 Georges-Antoine Assi
parent e053234d74
commit 361e18cc56
4 changed files with 272 additions and 175 deletions
+34
View File
@@ -15,6 +15,7 @@ from sqlalchemy import (
DateTime,
Enum,
Integer,
Row,
SQLColumnExpression,
String,
Text,
@@ -967,6 +968,20 @@ class DBRomsHandler(DBBaseHandler):
return []
return session.scalars(query.filter(Rom.id.in_(ids))).all()
@begin_session
def get_roms_by_ra_ids(
self,
ra_ids: Sequence[int],
*,
session: Session = INJECTED_SESSION,
) -> Sequence[Row[int, int | None]]:
"""Get `(id, ra_id)` for every ROM matching any of `ra_ids`."""
if not ra_ids:
return []
return session.execute(
select(Rom.id, Rom.ra_id).where(Rom.ra_id.in_(ra_ids))
).all()
def get_files_for_roms(
self,
rom_ids: list[int],
@@ -2500,6 +2515,25 @@ class DBRomsHandler(DBBaseHandler):
select(RomUser).filter_by(rom_id=rom_id, user_id=user_id).limit(1)
)
@begin_session
def get_rom_users_by_rom_ids(
self,
user_id: int,
rom_ids: Sequence[int],
*,
session: Session = INJECTED_SESSION,
) -> dict[int, RomUser]:
"""Get `{rom_id: RomUser}` for one user over many ROMs in a single query."""
if not rom_ids:
return {}
# Skips the joined `rom` (with its `platform`/`metadatum`) and `user` loads.
rows = session.scalars(
select(RomUser)
.options(raiseload(RomUser.rom), raiseload(RomUser.user))
.filter(RomUser.user_id == user_id, RomUser.rom_id.in_(rom_ids))
).all()
return {row.rom_id: row for row in rows}
@begin_session
def update_rom_user(
self,
@@ -50,6 +50,7 @@ def _sync_rom_user_statuses(user: User, user_progression: RAUserProgression) ->
changes (e.g. beating a previously-incomplete game) are reflected on every
sync run.
"""
desired: dict[int, RomUserStatus] = {}
for game_progression in user_progression.get("results", []):
rom_ra_id = game_progression.get("rom_ra_id")
if not rom_ra_id:
@@ -61,11 +62,19 @@ def _sync_rom_user_statuses(user: User, user_progression: RAUserProgression) ->
if new_status is None:
continue
rom = db_rom_handler.get_rom_by_metadata_id(ra_id=rom_ra_id)
if not rom:
continue
desired[rom_ra_id] = new_status
rom_user = db_rom_handler.get_rom_user(rom.id, user.id)
# One `ra_id` routinely covers several regional ROM rows; each gets synced.
roms = db_rom_handler.get_roms_by_ra_ids(list(desired))
rom_users = db_rom_handler.get_rom_users_by_rom_ids(
user.id, [rom.id for rom in roms]
)
for rom in roms:
# `IN` never matches a NULL `ra_id`.
new_status = desired[cast(int, rom.ra_id)]
rom_user = rom_users.get(rom.id)
if rom_user is None:
rom_user = db_rom_handler.add_rom_user(rom.id, user.id)
@@ -78,7 +87,7 @@ def _sync_rom_user_statuses(user: User, user_progression: RAUserProgression) ->
db_rom_handler.update_rom_user(rom_user.id, {"status": new_status})
log.debug(
f"Set rom_user status to '{new_status}' for user '{user.username}' "
f"and ROM with RA ID {rom_ra_id}"
f"and ROM {rom.id} (RA ID {rom.ra_id})"
)
@@ -124,7 +133,14 @@ class SyncRetroAchievementsProgressTask(PeriodicTask):
log.debug(
f"Updated RetroAchievements progress for user: {user.username}"
)
_sync_rom_user_statuses(user, user_progression)
# A concurrent rom_user write (e.g. a play session) can fail this
# user's batch; the remaining users still sync.
try:
_sync_rom_user_statuses(user, user_progression)
except Exception as e:
log.error(
f"Failed to sync RetroAchievements statuses for user: {user.username}, error: {e}"
)
processed_users += 1
update_stats.update(processed=processed_users)
@@ -1122,3 +1122,35 @@ class TestRomFileSizeLocking:
for call in session.execute.call_args_list
]
assert locked == sorted({rom.id, second_rom.id})
class TestRaBatchReads:
"""Direct coverage for the RetroAchievements sync's batched getters."""
def test_get_roms_by_ra_ids_returns_every_match(self, platform: Platform):
usa_rom = make_rom(platform, "Game USA", ra_id=12345)
eur_rom = make_rom(platform, "Game Europe", ra_id=12345)
make_rom(platform, "Other Game", ra_id=67890)
rows = db_rom_handler.get_roms_by_ra_ids([12345])
assert {row.id for row in rows} == {usa_rom.id, eur_rom.id}
def test_get_roms_by_ra_ids_empty_input_short_circuits(self):
assert db_rom_handler.get_roms_by_ra_ids([]) == []
def test_get_rom_users_by_rom_ids_scopes_to_user(self, rom: Rom, editor_user: User):
# The `rom` fixture already links this ROM to the admin user.
db_rom_handler.add_rom_user(rom_id=rom.id, user_id=editor_user.id)
rows = db_rom_handler.get_rom_users_by_rom_ids(editor_user.id, [rom.id])
# The session is closed and relations raise, so only columns are read.
assert set(rows) == {rom.id}
assert rows[rom.id].user_id == editor_user.id
assert rows[rom.id].status is None
def test_get_rom_users_by_rom_ids_empty_input_short_circuits(
self, viewer_user: User
):
assert db_rom_handler.get_rom_users_by_rom_ids(viewer_user.id, []) == {}
@@ -1,12 +1,17 @@
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock
import pytest
from tests.factories import make_rom
from adapters.services.retroachievements_types import RAUserCompletionProgressKind
from handler.database import db_rom_handler
from handler.database.roms_handler import DBRomsHandler
from handler.database.users_handler import DBUsersHandler
from handler.metadata.ra_handler import RAHandler
from models.rom import RomUser, RomUserStatus
from models.user import User
from tasks.scheduled.sync_retroachievements_progress import (
SyncRetroAchievementsProgressTask,
_get_rom_user_status_from_ra_award_kind,
@@ -19,6 +24,51 @@ def task() -> SyncRetroAchievementsProgressTask:
return SyncRetroAchievementsProgressTask()
def _progression(*games: tuple[int, str | None]) -> dict[str, Any]:
return {
"total": len(games),
"results": [
{"rom_ra_id": ra_id, "highest_award_kind": award_kind}
for ra_id, award_kind in games
],
}
def _mock_rom_user(rom_user_id: int, status: RomUserStatus | None) -> MagicMock:
rom_user = MagicMock(spec=RomUser)
rom_user.id = rom_user_id
rom_user.status = status
return rom_user
def _mock_sync(
mocker,
user: User,
progression: dict[str, Any],
roms: list[tuple[int, int]],
rom_users: dict[int, MagicMock],
) -> tuple[MagicMock, MagicMock, MagicMock]:
"""Patch one user's RA feed and the sync's DB calls.
Returns:
The `get_roms_by_ra_ids`, `get_rom_users_by_rom_ids` and `update_rom_user` mocks.
"""
mocker.patch.object(DBUsersHandler, "get_users", return_value=[user])
mocker.patch.object(DBUsersHandler, "update_user")
mocker.patch.object(RAHandler, "get_user_progression", return_value=progression)
return (
mocker.patch.object(
DBRomsHandler,
"get_roms_by_ra_ids",
return_value=[SimpleNamespace(id=i, ra_id=r) for i, r in roms],
),
mocker.patch.object(
DBRomsHandler, "get_rom_users_by_rom_ids", return_value=rom_users
),
mocker.patch.object(DBRomsHandler, "update_rom_user"),
)
class TestGetRomUserStatusFromRaAwardKind:
"""Tests for the _get_rom_user_status_from_ra_award_kind helper."""
@@ -138,187 +188,152 @@ class TestSyncRetroAchievementsProgressTask:
{"ra_progression": user_progression},
)
async def test_run_sets_rom_user_status_when_unset(
self, task, viewer_user, rom, mocker
@pytest.mark.parametrize(
("award_kind", "current_status", "expected_status"),
[
(RAUserCompletionProgressKind.MASTERED, None, RomUserStatus.COMPLETED_100),
(
RAUserCompletionProgressKind.MASTERED,
RomUserStatus.INCOMPLETE,
RomUserStatus.COMPLETED_100,
),
(None, None, RomUserStatus.INCOMPLETE),
(RAUserCompletionProgressKind.MASTERED, RomUserStatus.COMPLETED_100, None),
(RAUserCompletionProgressKind.MASTERED, RomUserStatus.RETIRED, None),
(RAUserCompletionProgressKind.MASTERED, RomUserStatus.NEVER_PLAYING, None),
],
)
async def test_run_syncs_rom_user_status(
self, task, viewer_user, mocker, award_kind, current_status, expected_status
):
"""Test that rom_user.status is set from RA award kind when currently unset."""
ra_id = 12345
mocker.patch.object(DBUsersHandler, "get_users", return_value=[viewer_user])
mocker.patch.object(DBUsersHandler, "update_user")
user_progression = {
"total": 1,
"results": [
{
"rom_ra_id": ra_id,
"max_possible": 50,
"num_awarded": 50,
"num_awarded_hardcore": 50,
"highest_award_kind": RAUserCompletionProgressKind.MASTERED,
"earned_achievements": [],
}
],
}
mocker.patch.object(
RAHandler, "get_user_progression", return_value=user_progression
"""The RA award sets the status unless it already matches or is user-owned."""
rom_user = _mock_rom_user(1, current_status)
_, _, mock_update_rom_user = _mock_sync(
mocker,
viewer_user,
_progression((12345, award_kind)),
roms=[(1, 12345)],
rom_users={1: rom_user},
)
mock_rom = MagicMock()
mock_rom.id = rom.id
mocker.patch.object(
DBRomsHandler, "get_rom_by_metadata_id", return_value=mock_rom
)
mock_rom_user = MagicMock(spec=RomUser)
mock_rom_user.id = 1
mock_rom_user.status = None
mocker.patch.object(DBRomsHandler, "get_rom_user", return_value=mock_rom_user)
mock_update_rom_user = mocker.patch.object(DBRomsHandler, "update_rom_user")
await task.run()
mock_update_rom_user.assert_called_once_with(
mock_rom_user.id, {"status": RomUserStatus.COMPLETED_100}
)
async def test_run_updates_existing_rom_user_status_when_changed(
self, task, viewer_user, rom, mocker
):
"""Test that rom_user.status is updated when the RA award kind changes."""
ra_id = 12345
mocker.patch.object(DBUsersHandler, "get_users", return_value=[viewer_user])
mocker.patch.object(DBUsersHandler, "update_user")
user_progression = {
"total": 1,
"results": [
{
"rom_ra_id": ra_id,
"max_possible": 50,
"num_awarded": 50,
"num_awarded_hardcore": 50,
"highest_award_kind": RAUserCompletionProgressKind.MASTERED,
"earned_achievements": [],
}
],
}
mocker.patch.object(
RAHandler, "get_user_progression", return_value=user_progression
)
mock_rom = MagicMock()
mock_rom.id = rom.id
mocker.patch.object(
DBRomsHandler, "get_rom_by_metadata_id", return_value=mock_rom
)
mock_rom_user = MagicMock(spec=RomUser)
mock_rom_user.id = 1
# Previously INCOMPLETE (set by an earlier sync), now mastered in RA
mock_rom_user.status = RomUserStatus.INCOMPLETE
mocker.patch.object(DBRomsHandler, "get_rom_user", return_value=mock_rom_user)
mock_update_rom_user = mocker.patch.object(DBRomsHandler, "update_rom_user")
await task.run()
mock_update_rom_user.assert_called_once_with(
mock_rom_user.id, {"status": RomUserStatus.COMPLETED_100}
)
async def test_run_skips_update_when_status_already_matches(
self, task, viewer_user, rom, mocker
):
"""Test that rom_user.status is not written again when it already matches RA."""
ra_id = 12345
mocker.patch.object(DBUsersHandler, "get_users", return_value=[viewer_user])
mocker.patch.object(DBUsersHandler, "update_user")
user_progression = {
"total": 1,
"results": [
{
"rom_ra_id": ra_id,
"max_possible": 50,
"num_awarded": 50,
"num_awarded_hardcore": 50,
"highest_award_kind": RAUserCompletionProgressKind.MASTERED,
"earned_achievements": [],
}
],
}
mocker.patch.object(
RAHandler, "get_user_progression", return_value=user_progression
)
mock_rom = MagicMock()
mock_rom.id = rom.id
mocker.patch.object(
DBRomsHandler, "get_rom_by_metadata_id", return_value=mock_rom
)
mock_rom_user = MagicMock(spec=RomUser)
mock_rom_user.id = 1
mock_rom_user.status = RomUserStatus.COMPLETED_100 # Already up-to-date
mocker.patch.object(DBRomsHandler, "get_rom_user", return_value=mock_rom_user)
mock_update_rom_user = mocker.patch.object(DBRomsHandler, "update_rom_user")
await task.run()
mock_update_rom_user.assert_not_called()
async def test_run_sets_incomplete_status_for_started_games(
self, task, viewer_user, rom, mocker
):
"""Test that INCOMPLETE status is set for games started but not beaten in RA."""
ra_id = 99999
mocker.patch.object(DBUsersHandler, "get_users", return_value=[viewer_user])
mocker.patch.object(DBUsersHandler, "update_user")
user_progression = {
"total": 1,
"results": [
{
"rom_ra_id": ra_id,
"max_possible": 100,
"num_awarded": 10,
"num_awarded_hardcore": 0,
"highest_award_kind": None,
"earned_achievements": [],
}
],
}
mocker.patch.object(
RAHandler, "get_user_progression", return_value=user_progression
)
mock_rom = MagicMock()
mock_rom.id = rom.id
mocker.patch.object(
DBRomsHandler, "get_rom_by_metadata_id", return_value=mock_rom
)
mock_rom_user = MagicMock(spec=RomUser)
mock_rom_user.id = 1
mock_rom_user.status = None
mocker.patch.object(DBRomsHandler, "get_rom_user", return_value=mock_rom_user)
mock_update_rom_user = mocker.patch.object(DBRomsHandler, "update_rom_user")
await task.run()
mock_update_rom_user.assert_called_once_with(
mock_rom_user.id, {"status": RomUserStatus.INCOMPLETE}
)
if expected_status is None:
mock_update_rom_user.assert_not_called()
else:
mock_update_rom_user.assert_called_once_with(
rom_user.id, {"status": expected_status}
)
async def test_run_skips_status_update_when_rom_not_found(
self, task, viewer_user, mocker
):
"""Test that status update is skipped when the ROM is not in the database."""
mocker.patch.object(DBUsersHandler, "get_users", return_value=[viewer_user])
mocker.patch.object(DBUsersHandler, "update_user")
user_progression = {
"total": 1,
"results": [
{
"rom_ra_id": 99999,
"highest_award_kind": RAUserCompletionProgressKind.MASTERED,
"earned_achievements": [],
}
],
}
mocker.patch.object(
RAHandler, "get_user_progression", return_value=user_progression
_, _, mock_update_rom_user = _mock_sync(
mocker,
viewer_user,
_progression((99999, RAUserCompletionProgressKind.MASTERED)),
roms=[],
rom_users={},
)
mocker.patch.object(DBRomsHandler, "get_rom_by_metadata_id", return_value=None)
mock_update_rom_user = mocker.patch.object(DBRomsHandler, "update_rom_user")
await task.run()
mock_update_rom_user.assert_not_called()
async def test_run_skips_unknown_award(self, task, viewer_user, mocker):
"""An unrecognised award kind is skipped without a write."""
mock_get_roms, _, mock_update_rom_user = _mock_sync(
mocker,
viewer_user,
_progression((12345, "unknown_award")),
roms=[],
rom_users={},
)
await task.run()
mock_get_roms.assert_called_once_with([])
mock_update_rom_user.assert_not_called()
async def test_run_batches_reads_to_single_calls_per_user(
self, task, viewer_user, mocker
):
"""Two games in one feed issue one ROM read and one rom_user read."""
rom_user = _mock_rom_user(1, None)
mock_get_roms, mock_get_rom_users, mock_update_rom_user = _mock_sync(
mocker,
viewer_user,
_progression(
(12345, RAUserCompletionProgressKind.MASTERED),
(67890, RAUserCompletionProgressKind.MASTERED),
),
roms=[(1, 12345), (2, 67890)],
rom_users={1: rom_user},
)
mock_add_rom_user = mocker.patch.object(
DBRomsHandler, "add_rom_user", return_value=rom_user
)
await task.run()
mock_get_roms.assert_called_once_with([12345, 67890])
mock_get_rom_users.assert_called_once_with(viewer_user.id, [1, 2])
mock_add_rom_user.assert_called_once_with(2, viewer_user.id)
assert mock_update_rom_user.call_count == 2
async def test_run_continues_after_status_sync_error(
self, task, viewer_user, editor_user, mocker
):
"""A status-sync failure for one user still syncs the next user."""
rom_user = _mock_rom_user(1, None)
mock_get_roms, _, mock_update_rom_user = _mock_sync(
mocker,
viewer_user,
_progression((12345, RAUserCompletionProgressKind.MASTERED)),
roms=[],
rom_users={1: rom_user},
)
mocker.patch.object(
DBUsersHandler, "get_users", return_value=[viewer_user, editor_user]
)
mock_get_roms.side_effect = [
Exception("DB error"),
[SimpleNamespace(id=1, ra_id=12345)],
]
await task.run()
mock_update_rom_user.assert_called_once_with(
rom_user.id, {"status": RomUserStatus.COMPLETED_100}
)
async def test_run_syncs_real_rows(self, task, viewer_user, platform, mocker):
"""Unmocked getters: every regional ROM is synced and RETIRED is kept."""
usa_rom = make_rom(platform, "Game USA", ra_id=12345)
eur_rom = make_rom(platform, "Game Europe", ra_id=12345)
retired_rom = make_rom(platform, "Retired Game", ra_id=67890)
retired_rom_user = db_rom_handler.add_rom_user(retired_rom.id, viewer_user.id)
db_rom_handler.update_rom_user(
retired_rom_user.id, {"status": RomUserStatus.RETIRED}
)
mocker.patch.object(DBUsersHandler, "get_users", return_value=[viewer_user])
mocker.patch.object(DBUsersHandler, "update_user")
mocker.patch.object(
RAHandler,
"get_user_progression",
return_value=_progression(
(12345, RAUserCompletionProgressKind.MASTERED),
(67890, RAUserCompletionProgressKind.MASTERED),
),
)
await task.run()
for synced_rom in (usa_rom, eur_rom):
rom_user = db_rom_handler.get_rom_user(synced_rom.id, viewer_user.id)
assert rom_user is not None
assert rom_user.status == RomUserStatus.COMPLETED_100
rom_user = db_rom_handler.get_rom_user(retired_rom.id, viewer_user.id)
assert rom_user is not None
assert rom_user.status == RomUserStatus.RETIRED