From fa0351084b64b5bfd42c7b8a15333768bb3ccd01 Mon Sep 17 00:00:00 2001 From: Lotus CI Date: Sun, 18 Oct 2026 15:24:27 -0400 Subject: [PATCH] fix(voice-limit-guard): two simultaneous joins can no longer both pass a full room (#25) The limit was checked against participants LiveKit already had, so tokens issued but not yet connected were not counted: two users pressing Join together could both get into a 1-person channel. Serialise the check per room and give each admitted user a 30 s reservation, counted like a present user until LiveKit lists them or it expires. Refs #25 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017PCb59uY3UKqDqt7B6hzd5 --- livekit/test_voice_limit_guard.py | 28 +++++++++++++++++ livekit/voice-limit-guard.py | 50 +++++++++++++++++++++++++++---- 2 files changed, 73 insertions(+), 5 deletions(-) diff --git a/livekit/test_voice_limit_guard.py b/livekit/test_voice_limit_guard.py index 9ff765f..f743762 100644 --- a/livekit/test_voice_limit_guard.py +++ b/livekit/test_voice_limit_guard.py @@ -389,3 +389,31 @@ class NormalizePermissionTest(unittest.TestCase): guard.livekit_update_participant = lambda *a: None participant = {"identity": "@a:x:DEV", "permission": {"canPublish": True, "canPublishSources": ["CAMERA"]}} self.assertFalse(guard.reconcile_participant("!r:x", participant, {"SCREEN_SHARE"})) + + +class SlotReservationTests(unittest.TestCase): + """[matrix #25] Simultaneous joins can't both pass a full room.""" + + def test_second_simultaneous_join_is_blocked(self): + r = guard.SlotReservations(ttl=30) + present = set() + counted = present | r.reserved("room", present, now=0) + self.assertFalse(guard.should_block(1, counted, "@a:x")) + r.reserve("room", "@a:x", now=0) + # @b asks a moment later, before @a has connected to LiveKit. + counted = present | r.reserved("room", present, now=1) + self.assertTrue(guard.should_block(1, counted, "@b:x")) + # @a asking again (second device) is still allowed. + self.assertFalse(guard.should_block(1, counted, "@a:x")) + + def test_reservation_expires_and_clears_once_present(self): + r = guard.SlotReservations(ttl=30) + r.reserve("room", "@a:x", now=0) + self.assertEqual(r.reserved("room", set(), now=31), set()) + r.reserve("room", "@a:x", now=40) + self.assertEqual(r.reserved("room", {"@a:x"}, now=41), set()) + + def test_room_lock_is_per_room(self): + r = guard.SlotReservations() + self.assertIs(r.room_lock("a"), r.room_lock("a")) + self.assertIsNot(r.room_lock("a"), r.room_lock("b")) diff --git a/livekit/voice-limit-guard.py b/livekit/voice-limit-guard.py index 51326a4..24f12e7 100644 --- a/livekit/voice-limit-guard.py +++ b/livekit/voice-limit-guard.py @@ -145,6 +145,41 @@ def should_block(limit: int, present_users: set, requester: str) -> bool: return len(present_users) >= limit +# [matrix #25] Tokens issued but not yet connected weren't counted, so two +# users pressing Join together could both pass a 1-person limit. Each admitted +# user holds a short reservation (counted like a present user) until LiveKit +# lists them or it expires; checks for a room are serialised with a lock. +RESERVATION_TTL = 30.0 + + +class SlotReservations: + """Per-room reservations for admitted-but-not-yet-connected users.""" + + def __init__(self, ttl: float = RESERVATION_TTL): + self.ttl = ttl + self._by_room = {} # alias -> {user: expiry} + self._guard = threading.Lock() + self._room_locks = {} + + def room_lock(self, alias: str) -> threading.Lock: + with self._guard: + return self._room_locks.setdefault(alias, threading.Lock()) + + def reserved(self, alias: str, present: set, now: float) -> set: + """Unexpired reservations; drops expired ones and users now present.""" + room = self._by_room.get(alias, {}) + for user in [u for u, exp in room.items() if exp <= now or u in present]: + del room[user] + return set(room) + + def reserve(self, alias: str, user: str, now: float) -> None: + if user: + self._by_room.setdefault(alias, {})[user] = now + self.ttl + + +RESERVATIONS = SlotReservations() + + def allowed_sources(policy: dict): """Pure decision: the narrowed `canPublishSources` list, or None. @@ -618,11 +653,16 @@ class Handler(BaseHTTPRequestHandler): # LiveKit-admin outage here cannot skip the publish-source policy below. if limit > 0 and alias: try: - present = livekit_present_users(alias) - if should_block(limit, present, requester): - log(f"blocked {requester or '?'} from {room_id}: {len(present)}/{limit}") - self._send_blocked() - return + with RESERVATIONS.room_lock(alias): + present = livekit_present_users(alias) + now = time.monotonic() + counted = present | RESERVATIONS.reserved(alias, present, now) + if should_block(limit, counted, requester): + log(f"blocked {requester or '?'} from {room_id}: {len(counted)}/{limit}") + self._send_blocked() + return + if requester not in present: + RESERVATIONS.reserve(alias, requester, now) except Exception as exc: # Fail open on the limit only; still apply the policy below. log(f"limit check error for {room_id}: {exc}")