Merge pull request 'fix(voice-limit-guard): two simultaneous joins can no longer both pass a full room (#25)' (#33) from fix/voice-limit-race into main

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"))

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)
+                with RESERVATIONS.room_lock(alias):
-                if should_block(limit, present, requester):
+                    present = livekit_present_users(alias)
-                    log(f"blocked {requester or '?'} from {room_id}: {len(present)}/{limit}")
+                    now = time.monotonic()
-                    self._send_blocked()
+                    counted = present | RESERVATIONS.reserved(alias, present, now)
-                    return
+                    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}")