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