48e154d53c
A joining peer learnt the roster by announcing itself and having every member reply with their state. That's one broadcast per member — O(N^2) messages for a room of N — and the in-memory channel layer caps each connection's queue at 100 and drops the overflow silently. Measured with 100 synthetic peers: every peer ended up seeing only 58-96 of the other 100, permanently. Not cosmetic — joinable? and mirrors? both read the roster, so you couldn't join someone you couldn't see, and a mate whose state was dropped wouldn't move you. The consumer now keeps the roster it is already relaying and hands a newcomer a snapshot in one message, so a join costs two broadcasts instead of N. Same 100 peers: roster complete and identical for everyone, 2310 messages sent down to 349, 72.6k deliveries down to 38.6k, server 30% -> 21% of one core, nav latency unchanged at ~14ms p50. This is a cache of relayed gossip, not a source of truth: parties are still worked out entirely in the browsers, and the server still decides nothing about them. It is process-local, like the in-memory channel layer it sits next to — if that ever moves to Redis for multiple workers, this moves too. Measured ceiling for the pathological case (everyone in ONE party, all mirroring each other): 100 peers ~14ms p50 at 21% of a core, 150 still ~14ms at 27%, 250 degrades to ~450ms p50 with the roster incomplete again. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
191 lines
8.7 KiB
Python
191 lines
8.7 KiB
Python
import json
|
||
|
||
from channels.testing import WebsocketCommunicator
|
||
from django.contrib.auth import get_user_model
|
||
from django.contrib.auth.models import AnonymousUser
|
||
from django.test import Client, TestCase, TransactionTestCase
|
||
|
||
from scenes.models import Project, Revision
|
||
from server.asgi import application
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
def ann(name, content="", **extra):
|
||
return {"type": "annotation", "in": ["root"], "name": name,
|
||
"content": content, "marks": [], **extra}
|
||
|
||
|
||
class SceneApiTestCase(TestCase):
|
||
def setUp(self):
|
||
self.alice = User.objects.create_user("alice", password="pw")
|
||
self.bob = User.objects.create_user("bob", password="pw")
|
||
self.project = Project.objects.create(owner=self.alice, name="p")
|
||
self.project.collaborators.add(self.bob)
|
||
self.a = Client(); self.a.login(username="alice", password="pw")
|
||
self.b = Client(); self.b.login(username="bob", password="pw")
|
||
|
||
def put(self, client, **body):
|
||
return client.put(f"/api/projects/{self.project.pk}/scene/",
|
||
json.dumps(body), content_type="application/json")
|
||
|
||
def groups(self):
|
||
self.project.refresh_from_db()
|
||
return self.project.scene.get("groups", {})
|
||
|
||
|
||
class DeltaMergeTests(SceneApiTestCase):
|
||
def test_different_annotations_merge_without_clobber(self):
|
||
self.put(self.a, changed={"a1": ann("opening")})
|
||
self.put(self.b, changed={"a2": ann("beat")})
|
||
self.assertEqual(set(self.groups()), {"a1", "a2"})
|
||
|
||
def test_same_annotation_same_field_is_last_write_wins(self):
|
||
self.put(self.a, changed={"a1": ann("x", "first")})
|
||
self.put(self.b, changed={"a1": {"content": "second"}})
|
||
self.assertEqual(self.groups()["a1"]["content"], "second")
|
||
|
||
def test_same_annotation_different_fields_merge(self):
|
||
self.put(self.a, changed={"a1": ann("x", "first", marks=[{"id": "m0"}])})
|
||
self.put(self.a, changed={"a1": {"marks": [{"id": "m1"}]}})
|
||
self.put(self.b, changed={"a1": {"content": "second"}})
|
||
g = self.groups()["a1"]
|
||
self.assertEqual(g["content"], "second")
|
||
self.assertEqual(g["marks"], [{"id": "m1"}])
|
||
|
||
def test_stale_client_neither_resurrects_nor_wipes(self):
|
||
# bob only ever knew a2; his delta must not erase alice's a1
|
||
self.put(self.a, changed={"a1": ann("x")})
|
||
self.put(self.b, changed={"a2": ann("y")})
|
||
self.assertEqual(set(self.groups()), {"a1", "a2"})
|
||
|
||
def test_delete_is_explicit_and_scoped(self):
|
||
self.put(self.a, changed={"a1": ann("x"), "a2": ann("y")})
|
||
self.put(self.b, deleted=["a1"])
|
||
self.assertEqual(set(self.groups()), {"a2"})
|
||
|
||
|
||
class AttributionTests(SceneApiTestCase):
|
||
def test_server_stamps_creator_ignoring_client(self):
|
||
self.put(self.a, changed={"a1": ann("x", createdBy="hacker", editedBy="hacker")})
|
||
g = self.groups()["a1"]
|
||
self.assertEqual((g["createdBy"], g["editedBy"]), ("alice", "alice"))
|
||
|
||
def test_edit_preserves_creator_updates_editor(self):
|
||
self.put(self.a, changed={"a1": ann("x")})
|
||
self.put(self.b, changed={"a1": ann("x", "edited")})
|
||
g = self.groups()["a1"]
|
||
self.assertEqual(g["createdBy"], "alice")
|
||
self.assertEqual(g["editedBy"], "bob")
|
||
|
||
def test_unchanged_annotation_is_not_restamped(self):
|
||
self.put(self.a, changed={"a1": ann("x")})
|
||
before = self.groups()["a1"]["editedAt"]
|
||
self.put(self.b, changed={"a1": ann("x")}) # identical content
|
||
after = self.groups()["a1"]
|
||
self.assertEqual(after["editedAt"], before)
|
||
self.assertEqual(after["editedBy"], "alice")
|
||
|
||
def test_revision_per_real_change_only(self):
|
||
self.put(self.a, changed={"a1": ann("x")})
|
||
self.put(self.b, changed={"a1": ann("x")}) # no-op → no revision
|
||
self.put(self.a, deleted=["a1"])
|
||
revs = Revision.objects.filter(project=self.project).order_by("created")
|
||
self.assertEqual([r.summary for r in revs],
|
||
["+1 ~0 −0 annotations", "+0 ~0 −1 annotations"])
|
||
self.assertEqual(revs[0].user, self.alice)
|
||
|
||
|
||
class AccessTests(SceneApiTestCase):
|
||
def test_write_requires_authentication(self):
|
||
r = Client().put(f"/api/projects/{self.project.pk}/scene/",
|
||
json.dumps({"changed": {}}), content_type="application/json")
|
||
self.assertEqual(r.status_code, 401)
|
||
|
||
def test_non_collaborator_cannot_edit(self):
|
||
User.objects.create_user("carol", password="pw")
|
||
carol = Client(); carol.login(username="carol", password="pw")
|
||
self.assertEqual(self.put(carol, changed={"a1": ann("x")}).status_code, 404)
|
||
|
||
|
||
class PresenceRelayTests(TransactionTestCase):
|
||
"""The socket hands out connection ids, stamps identity, and hands a joiner
|
||
the room in one message. The party itself is worked out in the browsers, so
|
||
there is nothing here to decide or trust."""
|
||
|
||
def setUp(self):
|
||
self.alice = User.objects.create_user("alice", password="pw")
|
||
self.project = Project.objects.create(owner=self.alice, name="p")
|
||
|
||
async def open(self, user=None):
|
||
"""Connect and drain the handshake, returning (comm, welcome, roster)."""
|
||
comm = WebsocketCommunicator(application, f"/ws/projects/{self.project.pk}/")
|
||
comm.scope["user"] = user or AnonymousUser()
|
||
connected, _ = await comm.connect()
|
||
self.assertTrue(connected)
|
||
welcome = await comm.receive_json_from()
|
||
roster = await comm.receive_json_from()
|
||
await comm.receive_json_from() # our own join, echoed back
|
||
return comm, welcome, roster
|
||
|
||
async def test_welcome_carries_an_id_and_the_signed_in_name(self):
|
||
comm, welcome, _ = await self.open(self.alice)
|
||
self.assertEqual(welcome["kind"], "welcome")
|
||
self.assertEqual(welcome["user"], "alice")
|
||
self.assertTrue(welcome["cid"])
|
||
await comm.disconnect()
|
||
|
||
async def test_a_joiner_is_handed_the_whole_room_in_one_message(self):
|
||
# the alternative — everyone answering a join — costs a broadcast per
|
||
# member, and a room of a hundred loses most of its roster to it
|
||
a, a_hello, a_roster = await self.open(self.alice)
|
||
self.assertEqual(a_roster["peers"], []) # first one in
|
||
await a.send_json_to({"kind": "state", "party": "P", "joinable": False})
|
||
await a.receive_json_from()
|
||
|
||
b, _, b_roster = await self.open()
|
||
self.assertEqual([p["cid"] for p in b_roster["peers"]], [a_hello["cid"]])
|
||
seen = b_roster["peers"][0]
|
||
self.assertEqual((seen["user"], seen["party"], seen["joinable"]),
|
||
("alice", "P", False)) # including what they last said
|
||
self.assertTrue(await b.receive_nothing(timeout=0.2)) # and nobody answers
|
||
await a.disconnect(); await b.disconnect()
|
||
|
||
async def test_a_departure_is_forgotten_not_handed_to_the_next_joiner(self):
|
||
a, _, _ = await self.open(self.alice)
|
||
b, b_hello, _ = await self.open()
|
||
await a.receive_json_from() # b's join
|
||
await b.disconnect()
|
||
await a.receive_json_from() # b's leave
|
||
c, _, c_roster = await self.open()
|
||
self.assertNotIn(b_hello["cid"], [p["cid"] for p in c_roster["peers"]])
|
||
await a.disconnect(); await c.disconnect()
|
||
|
||
async def test_presence_is_stamped_with_the_server_side_identity(self):
|
||
a, a_hello, _ = await self.open(self.alice)
|
||
b, _, _ = await self.open()
|
||
self.assertEqual((await a.receive_json_from())["kind"], "join")
|
||
|
||
await b.send_json_to({"kind": "state", "party": a_hello["cid"],
|
||
"joinable": True, "joining": a_hello["cid"],
|
||
"user": "alice", "cid": "forged"})
|
||
seen = await a.receive_json_from()
|
||
self.assertEqual(seen["party"], a_hello["cid"])
|
||
self.assertIsNone(seen["user"]) # anonymous, not "alice"
|
||
self.assertNotEqual(seen["cid"], "forged")
|
||
await a.disconnect(); await b.disconnect()
|
||
|
||
async def test_scene_edits_are_not_accepted_over_the_socket(self):
|
||
a, _, _ = await self.open(self.alice)
|
||
await a.send_json_to({"kind": "scene", "changed": {"a1": ann("x")}})
|
||
self.assertTrue(await a.receive_nothing(timeout=0.2))
|
||
await a.disconnect()
|
||
|
||
async def test_leaving_tells_the_room(self):
|
||
a, _, _ = await self.open(self.alice)
|
||
b, b_hello, _ = await self.open()
|
||
await a.receive_json_from() # b's join
|
||
await b.disconnect()
|
||
bye = await a.receive_json_from()
|
||
self.assertEqual((bye["kind"], bye["cid"]), ("leave", b_hello["cid"]))
|
||
await a.disconnect()
|