fix(voice-limit-guard): two simultaneous joins can no longer both pass a full room (#25)
Lint / Shell (shellcheck) (pull_request) Successful in 16s
Lint / JS (eslint) (pull_request) Successful in 13s
Lint / No secrets in webhook configs (pull_request) Successful in 8s
Lint / Landing page is rendered (matrix (pull_request) Successful in 10s
Lint / Python (ruff) (pull_request) Successful in 8s
Lint / Python deps (pip-audit) (pull_request) Successful in 1m25s
Lint / Secret scan (gitleaks) (pull_request) Successful in 13s
Lint / Shell (shellcheck) (push) Successful in 14s
Lint / JS (eslint) (push) Successful in 7s
Lint / No secrets in webhook configs (push) Successful in 4s
Lint / Landing page is rendered (matrix (push) Successful in 4s
Lint / Python (ruff) (push) Successful in 5s
Lint / Python deps (pip-audit) (push) Successful in 40s
Lint / Secret scan (gitleaks) (push) Successful in 6s

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017PCb59uY3UKqDqt7B6hzd5
This commit is contained in:
Lotus CI
2026-10-18 15:24:27 -04:00
co-authored by Claude Opus 5.5
parent b1335a830f
commit fa0351084b
2 changed files with 73 additions and 5 deletions
+28
View File
@@ -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"))