From 0490de5b5bfe7341bf409abde2a548abb608a924 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Sun, 6 Sep 2026 00:53:40 +0200 Subject: [PATCH] fix(relay): publish admission authority under the room lock A guest joining while the host handed the room to someone else could end up following the host everyone else had just left behind. join installs the client into the room and snapshots HostPeerID under room.mu, then released the lock before enqueueing joined. The client is a hostChanged recipient and a legal transfer target from the moment it is installed, so a transfer committing in that gap enqueued the newer authority first and the stale joined frame overwrote it on the client. transferHost already enqueues while holding the lock for exactly this reason; joined carries the same authority field and did not. The admission frame is now built and enqueued while the room is still locked, and enqueueFrame never blocks, so nothing waits on network I/O under the lock. A beforeJoinRoomAck barrier alongside the two existing test hooks pins the ordering: a transfer cannot commit while an admission is unpublished. --- server/main.go | 42 ++++++++++++++++++------------ server/main_test.go | 63 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 16 deletions(-) diff --git a/server/main.go b/server/main.go index 1724a7b76..bef331c6d 100644 --- a/server/main.go +++ b/server/main.go @@ -1094,6 +1094,7 @@ type Server struct { oauth *oauthProxy // nil when OAUTH_BASE_URL is unset removalErrors removalErrorThrottle beforeJoinRoomLock func() // test-only deterministic admission barrier + beforeJoinRoomAck func() // test-only authority-publication ordering barrier beforeLeaveRoomLock func() // test-only mutation/capture ordering barrier beforeTerminalDelivery func() // test-only post-persistence, pre-delivery barrier mu sync.RWMutex @@ -2075,9 +2076,33 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { } room.LastActivityAt = time.Now() peers := room.peerIDs() - hostPeerID := room.HostPeerID roomProtocolVersion := room.ProtocolVersion + existingPeers := make([]string, 0, len(peers)-1) + for _, peerID := range peers { + if peerID != msg.PeerID { + existingPeers = append(existingPeers, peerID) + } + } s.snap.recordMutation() + // The admission carries host authority, so like hostChanged it is + // enqueued while the room is still locked. This client became a + // hostChanged recipient at room.Peers[msg.PeerID] above, and its + // reservation makes it a legal transfer target immediately; a + // transfer committing in the gap would otherwise enqueue the newer + // authority first and leave this stale one to overwrite it. + // enqueueFrame never blocks (a full queue closes the client), so + // holding room.mu across it is safe. + if s.beforeJoinRoomAck != nil { + s.beforeJoinRoomAck() + } + client.sendJSON(serverMsg{ + Type: relayTypeJoined, + SessionID: msg.SessionID, + HostPeerID: room.HostPeerID, + ReconnectToken: responseToken, + ProtocolVersion: roomProtocolVersion, + Peers: existingPeers, + }) room.mu.Unlock() currentRoom = room @@ -2086,21 +2111,6 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { existingClient.close() } log.Printf("peer %s joined room %s", msg.PeerID, msg.SessionID) - - existingPeers := make([]string, 0, len(peers)-1) - for _, peerID := range peers { - if peerID != msg.PeerID { - existingPeers = append(existingPeers, peerID) - } - } - client.sendJSON(serverMsg{ - Type: relayTypeJoined, - SessionID: msg.SessionID, - HostPeerID: hostPeerID, - ReconnectToken: responseToken, - ProtocolVersion: roomProtocolVersion, - Peers: existingPeers, - }) room.broadcastExcept(msg.PeerID, serverMsg{Type: relayTypePeerJoined, PeerID: msg.PeerID}) case relayTypeLeave: diff --git a/server/main_test.go b/server/main_test.go index 53dd7d366..b87a85288 100644 --- a/server/main_test.go +++ b/server/main_test.go @@ -5385,6 +5385,69 @@ func TestTransferHostSwapsAuthorityAndBroadcastsHostChanged(t *testing.T) { } } +// An admission and a host transfer both publish room authority, and a joining +// client is a broadcast recipient — and a legal transfer target — from the +// moment it is installed. So the admission must publish its authority while it +// still holds room.mu: a transfer committing in the gap would enqueue the newer +// authority first and let the stale joined frame re-pin the client to the +// demoted host. +func TestJoinPublishesHostAuthorityUnderRoomLock(t *testing.T) { + h := newRelayHarness(t) + host, _, _, _ := createModernRoomWithGuest(t, h, "XFER_ORDER", "6.3.4.1", "6.3.4.2") + + atAck := make(chan struct{}) + releaseAck := make(chan struct{}) + var once sync.Once + h.srv.beforeJoinRoomAck = func() { + once.Do(func() { + close(atAck) + <-releaseAck + }) + } + + newcomerToken, _ := mustReconnectToken(t) + newcomer := h.dial(t, "6.3.4.3") + newcomer.send(clientMsg{ + Type: relayTypeJoin, + SessionID: "XFER_ORDER", + PeerID: "N", + ReconnectToken: newcomerToken, + ProtocolVersion: relayProtocolVersion, + }) + + select { + case <-atAck: + case <-time.After(2 * time.Second): + t.Fatal("admission never reached the authority-publication barrier") + } + + h.srv.mu.RLock() + room := h.srv.rooms["XFER_ORDER"] + h.srv.mu.RUnlock() + if room == nil { + close(releaseAck) + t.Fatal("room XFER_ORDER disappeared") + } + unlocked := room.mu.TryLock() + if unlocked { + room.mu.Unlock() + } + close(releaseAck) + if unlocked { + t.Fatal("join published host authority without holding room.mu") + } + + // End to end: the transfer lands strictly after the admission, so the + // newcomer's last authority frame is the room's real host. + host.send(clientMsg{Type: relayTypeTransferHost, To: "G", ProtocolVersion: relayProtocolVersion}) + if first := newcomer.recv(); first.Type != relayTypeJoined || first.HostPeerID != "H" { + t.Fatalf("first newcomer frame=%+v, want joined{H}", first) + } + if second := newcomer.recv(); second.Type != relayTypeHostChanged || second.HostPeerID != "G" { + t.Fatalf("second newcomer frame=%+v, want hostChanged{G}", second) + } +} + func TestTransferHostRejectsNonHostSender(t *testing.T) { h := newRelayHarness(t) host, guest, hostToken, _ := createModernRoomWithGuest(t, h, "XFER_NOT_HOST", "6.3.1.1", "6.3.1.2")