From 1be982f43da768ea8cdf549517e26cd8bbe57f59 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Tue, 28 Jul 2026 03:12:11 +0200 Subject: [PATCH] fix(watch-together): re-host an abandoned room code instead of joining it A room whose peers have all left is a code nobody is using, but the relay kept it bound to the creator's reconnect capability and rejected every other create with room_exists. The app compounded it: enterRoom only promoted to host on room_not_found, so tapping a recent code landed the user in the retained room as a guest of a host that was never coming back, until the cleanup sweep finally dropped the room. Create now replaces a room with no connected peers, and enterRoom hosts the code when its probe join finds an empty room. An occupied room still rejects create, including from its previous owner, and a host that is merely disconnected still reclaims its peer ID through join with the matching token. --- .../providers/watch_together_provider.dart | 14 ++- server/main.go | 15 ++-- server/main_test.go | 88 +++++++++++++++---- .../watch_together_provider_test.dart | 59 ++++++++++++- 4 files changed, 144 insertions(+), 32 deletions(-) diff --git a/lib/watch_together/providers/watch_together_provider.dart b/lib/watch_together/providers/watch_together_provider.dart index 79df6f7c..f39d8996 100644 --- a/lib/watch_together/providers/watch_together_provider.dart +++ b/lib/watch_together/providers/watch_together_provider.dart @@ -438,8 +438,8 @@ class WatchTogetherProvider with ChangeNotifier { } } - /// Enter a room by code — joins any reserved room and creates only when the - /// relay reports that no room reservation exists. + /// Enter a room by code — joins a room that still has someone in it and + /// hosts the code otherwise. /// /// Returns `true` if the user became the host. Future enterRoom( @@ -453,13 +453,19 @@ class WatchTogetherProvider with ChangeNotifier { final probe = _peerServiceFactory(endpoint: relayEndpoint); var shouldBeHost = false; try { + var probeJoined = false; try { await probe.joinSession(sessionId); + probeJoined = true; } on PeerError catch (error) { if (error.serverCode != RelayProtocol.roomNotFoundCode) rethrow; - shouldBeHost = true; } - if (!shouldBeHost) { + // A room the relay still holds but nobody is connected to is an + // abandoned code, not a session: its declared host is gone and the + // reservation only survives until the next cleanup sweep. Take the code + // over instead of waiting on a host that is never coming back. + shouldBeHost = !probeJoined || probe.connectedPeers.isEmpty; + if (probeJoined) { await probe.releaseSession(); } } finally { diff --git a/server/main.go b/server/main.go index 5fdbd933..3e1bf1cb 100644 --- a/server/main.go +++ b/server/main.go @@ -1883,16 +1883,13 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { } continue } - authorizedLegacyReplacement := - len(existing.Peers) == 0 && - !existing.closing && - msg.ProtocolVersion == legacyRelayProtocolVersion && - existing.ProtocolVersion == legacyRelayProtocolVersion && - msg.PeerID == existing.HostPeerID && - msg.ReconnectToken != "" && - reconnectVerifierMatches(existing.hostVerifier, hostVerifier) + // A room nobody is connected to is an abandoned code, not property. + // Whoever asks for it next takes it, so a host that restarted with a + // fresh reconnect token can reuse its own code instead of waiting out + // the cleanup sweep. An occupied room still belongs to its peers. + reclaimable := len(existing.Peers) == 0 && !existing.closing existing.mu.Unlock() - if !authorizedLegacyReplacement { + if !reclaimable { rejection = &serverMsg{Type: relayTypeError, Code: relayErrorRoomExists, Message: "Room already exists"} } } else if len(s.rooms) >= maxRetainedRooms { diff --git a/server/main_test.go b/server/main_test.go index b94ab282..da049ed4 100644 --- a/server/main_test.go +++ b/server/main_test.go @@ -2273,12 +2273,13 @@ func TestIdempotentCreateReannouncesPreviouslyAbsentHost(t *testing.T) { } } -func TestCreateCannotReclaimReservedEmptyRoom(t *testing.T) { +func TestCreateReclaimsAbandonedEmptyRoom(t *testing.T) { h := newRelayHarness(t) hostToken, hostVerifier := mustReconnectToken(t) original := &Room{ SessionID: "STALE", HostPeerID: "old-host", + ProtocolVersion: relayProtocolVersion, hostVerifier: hostVerifier, peerReservations: make(map[string]peerReservation), Peers: map[string]*Client{}, @@ -2289,29 +2290,82 @@ func TestCreateCannotReclaimReservedEmptyRoom(t *testing.T) { h.srv.rooms["STALE"] = original h.srv.mu.Unlock() + creatorToken, _ := mustReconnectToken(t) creator := h.dial(t, "1.1.1.6") - creator.send(clientMsg{Type: relayTypeCreate, SessionID: "STALE", PeerID: "new-host"}) - creator.expectError(relayErrorRoomExists) - - unproved := h.dial(t, "1.1.1.60") - unproved.send(clientMsg{Type: relayTypeJoin, SessionID: "STALE", PeerID: "old-host"}) - unproved.expectError(relayErrorPeerIdUnavailable) - - reconnected := h.dial(t, "1.1.1.61") - reconnected.send(clientMsg{ - Type: relayTypeJoin, - SessionID: "STALE", - PeerID: "old-host", - ReconnectToken: hostToken, + creator.send(clientMsg{ + Type: relayTypeCreate, + SessionID: "STALE", + PeerID: "new-host", + ReconnectToken: creatorToken, + ProtocolVersion: relayProtocolVersion, }) - reconnected.expectAuthority(relayTypeJoined, "old-host") + creator.expectAuthority(relayTypeCreated, "new-host") h.srv.mu.RLock() current := h.srv.rooms["STALE"] h.srv.mu.RUnlock() - if current != original { - t.Fatal("reserved room identity was replaced") + if current == original { + t.Fatal("abandoned room identity survived the reclaim") } + + // The previous owner's capability died with the room it belonged to, and + // the live replacement is not reclaimable by anyone, owner included. + former := h.dial(t, "1.1.1.60") + former.send(clientMsg{ + Type: relayTypeCreate, + SessionID: "STALE", + PeerID: "old-host", + ReconnectToken: hostToken, + ProtocolVersion: relayProtocolVersion, + }) + former.expectError(relayErrorRoomExists) +} + +// The recent-rooms flow: a host restarts its app, so it presents a fresh +// reconnect capability for a code the relay still holds. The abandoned code +// must come back as a hosted room instead of a ghost room with no host. +func TestAbandonedCodeIsRecreatableByARestartedHost(t *testing.T) { + h := newRelayHarness(t) + firstToken, _ := mustReconnectToken(t) + host := h.dial(t, "6.4.0.1") + host.send(clientMsg{ + Type: relayTypeCreate, + SessionID: "REUSE", + PeerID: "H", + ReconnectToken: firstToken, + ProtocolVersion: relayProtocolVersion, + }) + host.expectAuthority(relayTypeCreated, "H") + host.conn.Close() + h.waitRoomPeers(t, "REUSE", 0) + + // A restarted app mints a new capability, so it cannot prove the previous + // ownership even when it reuses its own peer ID. + restartToken, _ := mustReconnectToken(t) + restarted := h.dial(t, "6.4.0.2") + restarted.send(clientMsg{ + Type: relayTypeCreate, + SessionID: "REUSE", + PeerID: "H", + ReconnectToken: restartToken, + ProtocolVersion: relayProtocolVersion, + }) + recreated := restarted.expectAuthority(relayTypeCreated, "H") + if recreated.ReconnectToken != restartToken { + t.Fatalf("recreated room token=%q, want the presented capability", recreated.ReconnectToken) + } + + guestToken, _ := mustReconnectToken(t) + guest := h.dial(t, "6.4.0.3") + guest.send(clientMsg{ + Type: relayTypeJoin, + SessionID: "REUSE", + PeerID: "G", + ReconnectToken: guestToken, + ProtocolVersion: relayProtocolVersion, + }) + guest.expectAuthority(relayTypeJoined, "H") + restarted.expect(relayTypePeerJoined) } func TestCreateReclaimsOwnedEmptyRoomWithoutDoubleCharging(t *testing.T) { diff --git a/test/watch_together/watch_together_provider_test.dart b/test/watch_together/watch_together_provider_test.dart index d28d368e..044b72bc 100644 --- a/test/watch_together/watch_together_provider_test.dart +++ b/test/watch_together/watch_together_provider_test.dart @@ -698,7 +698,7 @@ void main() { }); group('WatchTogetherProvider — relay authority', () { - test('an empty successful probe joins the reserved room without creating', () async { + test('an occupied room is joined as a guest without creating', () async { late final _ProviderRelay relay; relay = await _ProviderRelay.start((socket, message) { if (message['type'] == 'join') { @@ -708,6 +708,7 @@ void main() { 'hostPeerId': _providerHostId, 'reconnectToken': message['reconnectToken'], 'protocolVersion': 2, + 'peers': [_providerHostId], }); } else if (message['type'] == 'leave') { relay.send(socket, { @@ -726,7 +727,7 @@ void main() { provider.dispose(); }); - final becameHost = await provider.enterRoom('empty1', relayEndpoint: endpoint, displayName: 'Guest'); + final becameHost = await provider.enterRoom('busy01', relayEndpoint: endpoint, displayName: 'Guest'); expect(becameHost, isFalse); expect(provider.isHost, isFalse); @@ -746,6 +747,60 @@ void main() { expect(relay.messages.where((message) => message['type'] == 'create'), isEmpty); }); + test('an abandoned room with no peers is hosted instead of joined', () async { + late final _ProviderRelay relay; + relay = await _ProviderRelay.start((socket, message) { + switch (message['type']) { + case 'join': + relay.send(socket, { + 'type': 'joined', + 'sessionId': message['sessionId'], + 'hostPeerId': _providerHostId, + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + }); + case 'leave': + relay.send(socket, { + 'type': 'left', + 'sessionId': message['sessionId'], + 'peerId': message['peerId'], + 'protocolVersion': 2, + }); + case 'create': + relay.send(socket, { + 'type': 'created', + 'sessionId': message['sessionId'], + 'hostPeerId': message['peerId'], + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + }); + case 'endSession': + relay.send(socket, {'type': 'ended', 'sessionId': message['sessionId'], 'protocolVersion': 2}); + } + }); + addTearDown(relay.close); + final endpoint = WatchTogetherRelayEndpoint.resolve(relay.baseUrl); + final provider = WatchTogetherProvider(); + addTearDown(() async { + await provider.leaveSession(); + provider.dispose(); + }); + + final becameHost = await provider.enterRoom('empty1', relayEndpoint: endpoint, displayName: 'Host'); + + expect(becameHost, isTrue); + expect(provider.isHost, isTrue); + // The probe identity is released before the code is taken over, so the + // relay sees an empty room when the create lands. + expect(relay.messages.map((message) => message['type']).take(3), ['join', 'leave', 'create']); + final create = relay.messages.singleWhere((message) => message['type'] == 'create'); + expect(create['sessionId'], 'EMPTY1'); + expect(provider.session?.hostPeerId, create['peerId']); + final probeJoin = relay.messages.firstWhere((message) => message['type'] == 'join'); + expect(create['peerId'], isNot(probeJoin['peerId'])); + expect(provider.session?.hostPeerId, isNot(_providerHostId)); + }); + test('a room-not-found probe creates with relay-declared host authority', () async { late final _ProviderRelay relay; relay = await _ProviderRelay.start((socket, message) {