diff --git a/src/backend/core/services/lobby.py b/src/backend/core/services/lobby.py index cc27da76..34cd157b 100644 --- a/src/backend/core/services/lobby.py +++ b/src/backend/core/services/lobby.py @@ -79,6 +79,14 @@ class LobbyService: Handles participant entry requests, status management, and notifications using cache for state management and LiveKit for real-time updates. + + Participant membership per room is tracked in a native Redis SET (the + "room index") so that listing and clearing a room's lobby never scans + the shared keyspace. The per-participant cache entries remain the source + of truth for state: their TTLs implement liveness (a waiter who stops + polling simply expires). The index only records which participant ids + may exist for a room; a stale id costs one cache miss and is pruned + lazily. """ @staticmethod @@ -86,6 +94,61 @@ class LobbyService: """Generate cache key for participant(s) data.""" return f"{settings.LOBBY_KEY_PREFIX}_{room_id!s}_{participant_id}" + @staticmethod + def _get_index_key(room_id: UUID) -> str: + """Raw Redis key of the per-room participant index (a native SET). + + Built through django-redis' make_key so it lives under the same + KEY_PREFIX/version namespace as the participant entries. + """ + return cache.client.make_key(f"{settings.LOBBY_KEY_PREFIX}-index_{room_id!s}") + + @staticmethod + def _redis(write: bool = True): + """Raw redis-py client. + + SADD/SREM/SMEMBERS are not exposed by the Django cache API; this is + the documented django-redis escape hatch. + """ + return cache.client.get_client(write=write) + + def _index_add(self, room_id: UUID, participant_id: str) -> None: + """Record a participant id in the room index. + + Refreshes a backstop TTL on the index so an abandoned room cannot + leak its set beyond the longest participant timeout. + """ + index_key = self._get_index_key(room_id) + pipe = self._redis().pipeline(transaction=False) + pipe.sadd(index_key, participant_id) + pipe.expire(index_key, settings.LOBBY_ACCEPTED_TIMEOUT) + pipe.execute() + + def _index_members(self, room_id: UUID) -> List[str]: + """All participant ids currently indexed for the room.""" + members = self._redis(write=False).smembers(self._get_index_key(room_id)) + return [ + member.decode() if isinstance(member, bytes) else member + for member in members + ] + + def _index_touch(self, room_id: UUID) -> None: + """Re-arm the room index backstop TTL. + + Called whenever a participant entry is written or refreshed so the + index always outlives every entry it references — including a lone + waiter whose rolling WAITING TTL would otherwise outlast the + backstop set at enter() time. + """ + self._redis().expire( + self._get_index_key(room_id), settings.LOBBY_ACCEPTED_TIMEOUT + ) + + def _index_remove(self, room_id: UUID, *participant_ids: str) -> None: + """Drop participant ids from the room index.""" + if participant_ids: + self._redis().srem(self._get_index_key(room_id), *participant_ids) + @staticmethod def _get_or_create_participant_id(request) -> str: """Extract unique participant identifier from the request.""" @@ -209,14 +272,16 @@ class LobbyService: cache.touch( self._get_cache_key(room_id, participant_id), settings.LOBBY_WAITING_TIMEOUT ) + self._index_touch(room_id) def enter( self, room_id: UUID, participant_id: str, username: str ) -> LobbyParticipant: """Add participant to waiting lobby. - Create a new participant entry in waiting status and notify room - participants of the new entry request. + Create a new participant entry in waiting status, index the + participant id for the room, and notify room participants of the + new entry request. """ color = utils.generate_color(participant_id) @@ -245,6 +310,7 @@ class LobbyService: participant.to_dict(), timeout=settings.LOBBY_WAITING_TIMEOUT, ) + self._index_add(room_id, participant_id) return participant @@ -267,15 +333,31 @@ class LobbyService: return None def list_waiting_participants(self, room_id: UUID) -> List[dict]: - """List all waiting participants for a room.""" + """List all waiting participants for a room. - pattern = self._get_cache_key(room_id, "*") - keys = list(cache.iter_keys(pattern, itersize=utils.CACHE_SCAN_ITERSIZE)) + Reads the per-room index (O(participants of this room)) instead of + scanning the shared keyspace. Indexed ids whose cache entry has + expired are pruned lazily here: the entry TTL is the liveness + protocol, so a missing entry means the participant is gone. + """ - if not keys: + member_ids = self._index_members(room_id) + + if not member_ids: return [] - data = cache.get_many(keys) + keys_by_id = { + participant_id: self._get_cache_key(room_id, participant_id) + for participant_id in member_ids + } + data = cache.get_many(list(keys_by_id.values())) + + dead_ids = [ + participant_id + for participant_id, cache_key in keys_by_id.items() + if cache_key not in data + ] + self._index_remove(room_id, *dead_ids) waiting_participants = [] for cache_key, raw_participant in data.items(): @@ -341,16 +423,28 @@ class LobbyService: participant.status = status cache.set(cache_key, participant.to_dict(), timeout=timeout) + self._index_touch(room_id) def clear_room_cache(self, room_id: UUID) -> None: - """Clear all participant entries from the cache for a specific room.""" + """Clear all participant entries from the cache for a specific room. - cache.delete_pattern( - self._get_cache_key(room_id, "*"), itersize=utils.CACHE_SCAN_ITERSIZE - ) + Deletes the indexed participant entries and the index itself with + targeted commands instead of a full-keyspace pattern scan. + """ + + member_ids = self._index_members(room_id) + if member_ids: + cache.delete_many( + [ + self._get_cache_key(room_id, participant_id) + for participant_id in member_ids + ] + ) + self._redis().delete(self._get_index_key(room_id)) def clear_participant_cache(self, room_id: UUID, participant_id: str) -> None: """Clear a given participant entry from the cache for a specific room.""" cache_key = self._get_cache_key(room_id, participant_id) cache.delete(cache_key) + self._index_remove(room_id, participant_id) diff --git a/src/backend/core/tests/rooms/test_api_rooms_lobby.py b/src/backend/core/tests/rooms/test_api_rooms_lobby.py index aabca484..164603d8 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_lobby.py +++ b/src/backend/core/tests/rooms/test_api_rooms_lobby.py @@ -589,6 +589,9 @@ def test_list_waiting_participants_success(settings): "color": "#654321", }, ) + lobby_service = LobbyService() + lobby_service._index_add(room.id, "2f7f162f-e7d1-421b-90e7-02bfbfbf8def") + lobby_service._index_add(room.id, "f4ca3ab8a6c04ad88097b8da33f60f10") response = client.get(f"/api/v1.0/rooms/{room.id}/waiting-participants/") diff --git a/src/backend/core/tests/services/test_lobby.py b/src/backend/core/tests/services/test_lobby.py index 82c840fd..16bd9017 100644 --- a/src/backend/core/tests/services/test_lobby.py +++ b/src/backend/core/tests/services/test_lobby.py @@ -2,7 +2,7 @@ Test lobby service. """ -# pylint: disable=W0621,W0613, W0212, R0913 +# pylint: disable=W0621,W0613, W0212, R0913, C0302 # ruff: noqa: PLR0913, PLR0917 import uuid @@ -24,7 +24,6 @@ from core.services.lobby import ( LobbyParticipantStatus, LobbyService, ) -from core.services.presence import CACHE_SCAN_ITERSIZE from core.utils import NotificationError pytestmark = pytest.mark.django_db @@ -466,18 +465,22 @@ def test_request_entry_participant_with_role( def test_refresh_waiting_status(mock_cache, lobby_service, participant_id): """Test refreshing waiting status for a participant.""" lobby_service._get_cache_key = mock.Mock(return_value="mocked_cache_key") + lobby_service._index_touch = mock.Mock() room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED) lobby_service.refresh_waiting_status(room.id, participant_id) mock_cache.touch.assert_called_once_with( "mocked_cache_key", settings.LOBBY_WAITING_TIMEOUT ) + lobby_service._index_touch.assert_called_once_with(room.id) # pylint: disable=R0917 @mock.patch("core.services.lobby.cache") @mock.patch("core.utils.generate_color") @mock.patch("core.utils.notify_participants") +@mock.patch("core.services.lobby.LobbyService._index_add") def test_enter_success( + mock_index_add, mock_notify, mock_generate_color, mock_cache, @@ -508,13 +511,16 @@ def test_enter_success( mock_notify.assert_called_once_with( room_name=str(room.pk), notification_data={"type": "participantWaiting"} ) + mock_index_add.assert_called_once_with(room.id, participant_id) # pylint: disable=R0917 @mock.patch("core.services.lobby.cache") @mock.patch("core.utils.generate_color") @mock.patch("core.utils.notify_participants") +@mock.patch("core.services.lobby.LobbyService._index_add") def test_enter_with_notification_error( + mock_index_add, mock_notify, mock_generate_color, mock_cache, @@ -541,6 +547,7 @@ def test_enter_with_notification_error( participant.to_dict(), timeout=settings.LOBBY_WAITING_TIMEOUT, ) + mock_index_add.assert_called_once_with(room.id, participant_id) @mock.patch("core.services.lobby.cache") @@ -579,14 +586,15 @@ def test_get_participant_parsing_error( @mock.patch("core.services.lobby.cache") def test_list_waiting_participants_empty(mock_cache, lobby_service): """Test listing waiting participants when none exist.""" - mock_cache.iter_keys.return_value = [] - room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED) + lobby_service._index_members = mock.Mock(return_value=[]) + lobby_service._index_remove = mock.Mock() + result = lobby_service.list_waiting_participants(room.id) assert result == [] - pattern = f"{settings.LOBBY_KEY_PREFIX}_{room.id!s}_*" - mock_cache.iter_keys.assert_called_once_with(pattern, itersize=CACHE_SCAN_ITERSIZE) + lobby_service._index_members.assert_called_once_with(room.id) + lobby_service._index_remove.assert_not_called() mock_cache.get_many.assert_not_called() @@ -595,7 +603,8 @@ def test_list_waiting_participants(mock_cache, lobby_service, participant_dict): """Test listing waiting participants with valid data.""" room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED) cache_key = f"{settings.LOBBY_KEY_PREFIX}_{room.id!s}_participant1" - mock_cache.iter_keys.return_value = [cache_key] + lobby_service._index_members = mock.Mock(return_value=["participant1"]) + lobby_service._index_remove = mock.Mock() mock_cache.get_many.return_value = {cache_key: participant_dict} result = lobby_service.list_waiting_participants(room.id) @@ -603,8 +612,8 @@ def test_list_waiting_participants(mock_cache, lobby_service, participant_dict): assert len(result) == 1 assert result[0]["status"] == "waiting" assert result[0]["username"] == "test-username" - pattern = f"{settings.LOBBY_KEY_PREFIX}_{room.id!s}_*" - mock_cache.iter_keys.assert_called_once_with(pattern, itersize=CACHE_SCAN_ITERSIZE) + lobby_service._index_members.assert_called_once_with(room.id) + lobby_service._index_remove.assert_called_once_with(room.id) mock_cache.get_many.assert_called_once_with([cache_key]) @@ -629,7 +638,10 @@ def test_list_waiting_participants_multiple(mock_cache, lobby_service): "color": "#654321", } - mock_cache.iter_keys.return_value = [cache_key1, cache_key2] + lobby_service._index_members = mock.Mock( + return_value=["participant1", "participant2"] + ) + lobby_service._index_remove = mock.Mock() mock_cache.get_many.return_value = { cache_key1: participant1, cache_key2: participant2, @@ -646,8 +658,7 @@ def test_list_waiting_participants_multiple(mock_cache, lobby_service): # Verify all participants have waiting status assert all(p["status"] == "waiting" for p in result) - pattern = f"{settings.LOBBY_KEY_PREFIX}_{room.id!s}_*" - mock_cache.iter_keys.assert_called_once_with(pattern, itersize=CACHE_SCAN_ITERSIZE) + lobby_service._index_members.assert_called_once_with(room.id) mock_cache.get_many.assert_called_once_with([cache_key1, cache_key2]) @@ -656,7 +667,8 @@ def test_list_waiting_participants_corrupted_data(mock_cache, lobby_service): """Test listing waiting participants with corrupted data.""" room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED) cache_key = f"{settings.LOBBY_KEY_PREFIX}_{room.id!s}_participant1" - mock_cache.iter_keys.return_value = [cache_key] + lobby_service._index_members = mock.Mock(return_value=["participant1"]) + lobby_service._index_remove = mock.Mock() mock_cache.get_many.return_value = {cache_key: {"invalid": "data"}} result = lobby_service.list_waiting_participants(room.id) @@ -681,7 +693,10 @@ def test_list_waiting_participants_partially_corrupted(mock_cache, lobby_service corrupted_participant = {"invalid": "data"} - mock_cache.iter_keys.return_value = [cache_key1, cache_key2] + lobby_service._index_members = mock.Mock( + return_value=["participant1", "participant2"] + ) + lobby_service._index_remove = mock.Mock() mock_cache.get_many.return_value = { cache_key1: corrupted_participant, cache_key2: valid_participant, @@ -699,8 +714,6 @@ def test_list_waiting_participants_partially_corrupted(mock_cache, lobby_service mock_cache.delete.assert_called_once_with(cache_key1) # Verify both cache keys were queried - pattern = f"{settings.LOBBY_KEY_PREFIX}_{room.id!s}_*" - mock_cache.iter_keys.assert_called_once_with(pattern, itersize=CACHE_SCAN_ITERSIZE) mock_cache.get_many.assert_called_once_with([cache_key1, cache_key2]) @@ -724,7 +737,10 @@ def test_list_waiting_participants_non_waiting(mock_cache, lobby_service): "color": "#654321", } - mock_cache.iter_keys.return_value = [cache_key1, cache_key2] + lobby_service._index_members = mock.Mock( + return_value=["participant1", "participant2"] + ) + lobby_service._index_remove = mock.Mock() mock_cache.get_many.return_value = { cache_key1: participant1, cache_key2: participant2, @@ -820,6 +836,7 @@ def test_update_participant_status_success(mock_cache, lobby_service, participan mock_cache.get.return_value = participant_dict lobby_service._get_cache_key = mock.Mock(return_value="mocked_cache_key") + lobby_service._index_touch = mock.Mock() lobby_service._update_participant_status( room.id, @@ -837,6 +854,7 @@ def test_update_participant_status_success(mock_cache, lobby_service, participan mock_cache.set.assert_called_once_with( "mocked_cache_key", expected_data, timeout=60 ) + lobby_service._index_touch.assert_called_once_with(room.id) lobby_service._get_cache_key.assert_called_once_with(room.id, participant_id) @@ -881,9 +899,13 @@ def test_clear_room_cache(settings, lobby_service): timeout=settings.LOBBY_DENIED_TIMEOUT, ) + for participant_id in ("participant1", "participant2", "participant3"): + lobby_service._index_add(room_id, participant_id) + lobby_service.clear_room_cache(room_id) assert cache.keys(f"test-lobby_{room_id!s}_*") == [] + assert lobby_service._index_members(room_id) == [] def test_clear_room_empty(settings, lobby_service): @@ -910,10 +932,13 @@ def test_clear_participant_cache(lobby_service): "color": "#123456", } cache.set(cache_key, participant_data, timeout=settings.LOBBY_WAITING_TIMEOUT) + lobby_service._index_add(room_id, participant_id) assert cache.get(cache_key) is not None + assert participant_id in lobby_service._index_members(room_id) lobby_service.clear_participant_cache(room_id, participant_id) assert cache.get(cache_key) is None + assert participant_id not in lobby_service._index_members(room_id) def test_clear_participant_cache_nonexistent(lobby_service): @@ -927,3 +952,87 @@ def test_clear_participant_cache_nonexistent(lobby_service): lobby_service.clear_participant_cache(room_id, participant_id) assert cache.get(cache_key) is None + + +# Room index integration tests (real cache backend) + + +def test_index_add_members_remove_roundtrip(lobby_service): + """The room index records, lists and forgets participant ids.""" + room_id = uuid.uuid4() + + assert lobby_service._index_members(room_id) == [] + + lobby_service._index_add(room_id, "participant1") + lobby_service._index_add(room_id, "participant2") + + assert sorted(lobby_service._index_members(room_id)) == [ + "participant1", + "participant2", + ] + + # The index carries a backstop TTL so abandoned rooms cannot leak it. + ttl = lobby_service._redis().ttl(lobby_service._get_index_key(room_id)) + assert 0 < ttl <= settings.LOBBY_ACCEPTED_TIMEOUT + + lobby_service._index_remove(room_id, "participant1") + assert lobby_service._index_members(room_id) == ["participant2"] + + +@mock.patch("core.utils.notify_participants") +def test_enter_registers_participant_in_room_index( + mock_notify, lobby_service, participant_id, username +): + """Entering the lobby must index the participant id for the room.""" + room_id = uuid.uuid4() + + lobby_service.enter(room_id, participant_id, username) + + assert lobby_service._index_members(room_id) == [participant_id] + + +def test_list_waiting_participants_prunes_stale_index_ids(settings, lobby_service): + """Indexed ids whose cache entry expired are pruned and not listed.""" + settings.LOBBY_KEY_PREFIX = "test-lobby-prune" + room_id = uuid.uuid4() + + cache.set( + f"test-lobby-prune_{room_id!s}_participant1", + { + "id": "participant1", + "username": "user1", + "status": "waiting", + "color": "#123456", + }, + timeout=100, + ) + lobby_service._index_add(room_id, "participant1") + # participant2 is indexed but its cache entry has expired. + lobby_service._index_add(room_id, "participant2") + + result = lobby_service.list_waiting_participants(room_id) + + assert [participant["id"] for participant in result] == ["participant1"] + assert lobby_service._index_members(room_id) == ["participant1"] + + +def test_refresh_waiting_status_rearms_room_index_ttl(lobby_service, participant_id): + """A lone waiter's polling must keep the room index alive. + + Regression test: the index backstop TTL is only armed at enter() time, + so a participant whose rolling WAITING refreshes outlast it would keep + their entry alive while silently vanishing from the moderator list. + Refreshing the waiting status must therefore re-arm the index TTL. + """ + room_id = uuid.uuid4() + lobby_service._index_add(room_id, participant_id) + + index_key = lobby_service._get_index_key(room_id) + redis_client = lobby_service._redis() + redis_client.expire(index_key, 10) + assert redis_client.ttl(index_key) <= 10 + + lobby_service.refresh_waiting_status(room_id, participant_id) + + assert redis_client.ttl(index_key) > 10 + assert lobby_service._index_members(room_id) == [participant_id]