mirror of
https://github.com/suitenumerique/meet.git
synced 2026-09-07 16:05:39 +00:00
wip use an index approach to avoid O(db_size) complexity
This commit is contained in:
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user