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.
This commit is contained in:
@@ -438,8 +438,8 @@ class WatchTogetherProvider with ChangeNotifier {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Enter a room by code — joins any reserved room and creates only when the
|
/// Enter a room by code — joins a room that still has someone in it and
|
||||||
/// relay reports that no room reservation exists.
|
/// hosts the code otherwise.
|
||||||
///
|
///
|
||||||
/// Returns `true` if the user became the host.
|
/// Returns `true` if the user became the host.
|
||||||
Future<bool> enterRoom(
|
Future<bool> enterRoom(
|
||||||
@@ -453,13 +453,19 @@ class WatchTogetherProvider with ChangeNotifier {
|
|||||||
final probe = _peerServiceFactory(endpoint: relayEndpoint);
|
final probe = _peerServiceFactory(endpoint: relayEndpoint);
|
||||||
var shouldBeHost = false;
|
var shouldBeHost = false;
|
||||||
try {
|
try {
|
||||||
|
var probeJoined = false;
|
||||||
try {
|
try {
|
||||||
await probe.joinSession(sessionId);
|
await probe.joinSession(sessionId);
|
||||||
|
probeJoined = true;
|
||||||
} on PeerError catch (error) {
|
} on PeerError catch (error) {
|
||||||
if (error.serverCode != RelayProtocol.roomNotFoundCode) rethrow;
|
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();
|
await probe.releaseSession();
|
||||||
}
|
}
|
||||||
} finally {
|
} finally {
|
||||||
|
|||||||
+6
-9
@@ -1883,16 +1883,13 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
|
|||||||
}
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
authorizedLegacyReplacement :=
|
// A room nobody is connected to is an abandoned code, not property.
|
||||||
len(existing.Peers) == 0 &&
|
// Whoever asks for it next takes it, so a host that restarted with a
|
||||||
!existing.closing &&
|
// fresh reconnect token can reuse its own code instead of waiting out
|
||||||
msg.ProtocolVersion == legacyRelayProtocolVersion &&
|
// the cleanup sweep. An occupied room still belongs to its peers.
|
||||||
existing.ProtocolVersion == legacyRelayProtocolVersion &&
|
reclaimable := len(existing.Peers) == 0 && !existing.closing
|
||||||
msg.PeerID == existing.HostPeerID &&
|
|
||||||
msg.ReconnectToken != "" &&
|
|
||||||
reconnectVerifierMatches(existing.hostVerifier, hostVerifier)
|
|
||||||
existing.mu.Unlock()
|
existing.mu.Unlock()
|
||||||
if !authorizedLegacyReplacement {
|
if !reclaimable {
|
||||||
rejection = &serverMsg{Type: relayTypeError, Code: relayErrorRoomExists, Message: "Room already exists"}
|
rejection = &serverMsg{Type: relayTypeError, Code: relayErrorRoomExists, Message: "Room already exists"}
|
||||||
}
|
}
|
||||||
} else if len(s.rooms) >= maxRetainedRooms {
|
} else if len(s.rooms) >= maxRetainedRooms {
|
||||||
|
|||||||
+71
-17
@@ -2273,12 +2273,13 @@ func TestIdempotentCreateReannouncesPreviouslyAbsentHost(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestCreateCannotReclaimReservedEmptyRoom(t *testing.T) {
|
func TestCreateReclaimsAbandonedEmptyRoom(t *testing.T) {
|
||||||
h := newRelayHarness(t)
|
h := newRelayHarness(t)
|
||||||
hostToken, hostVerifier := mustReconnectToken(t)
|
hostToken, hostVerifier := mustReconnectToken(t)
|
||||||
original := &Room{
|
original := &Room{
|
||||||
SessionID: "STALE",
|
SessionID: "STALE",
|
||||||
HostPeerID: "old-host",
|
HostPeerID: "old-host",
|
||||||
|
ProtocolVersion: relayProtocolVersion,
|
||||||
hostVerifier: hostVerifier,
|
hostVerifier: hostVerifier,
|
||||||
peerReservations: make(map[string]peerReservation),
|
peerReservations: make(map[string]peerReservation),
|
||||||
Peers: map[string]*Client{},
|
Peers: map[string]*Client{},
|
||||||
@@ -2289,29 +2290,82 @@ func TestCreateCannotReclaimReservedEmptyRoom(t *testing.T) {
|
|||||||
h.srv.rooms["STALE"] = original
|
h.srv.rooms["STALE"] = original
|
||||||
h.srv.mu.Unlock()
|
h.srv.mu.Unlock()
|
||||||
|
|
||||||
|
creatorToken, _ := mustReconnectToken(t)
|
||||||
creator := h.dial(t, "1.1.1.6")
|
creator := h.dial(t, "1.1.1.6")
|
||||||
creator.send(clientMsg{Type: relayTypeCreate, SessionID: "STALE", PeerID: "new-host"})
|
creator.send(clientMsg{
|
||||||
creator.expectError(relayErrorRoomExists)
|
Type: relayTypeCreate,
|
||||||
|
SessionID: "STALE",
|
||||||
unproved := h.dial(t, "1.1.1.60")
|
PeerID: "new-host",
|
||||||
unproved.send(clientMsg{Type: relayTypeJoin, SessionID: "STALE", PeerID: "old-host"})
|
ReconnectToken: creatorToken,
|
||||||
unproved.expectError(relayErrorPeerIdUnavailable)
|
ProtocolVersion: relayProtocolVersion,
|
||||||
|
|
||||||
reconnected := h.dial(t, "1.1.1.61")
|
|
||||||
reconnected.send(clientMsg{
|
|
||||||
Type: relayTypeJoin,
|
|
||||||
SessionID: "STALE",
|
|
||||||
PeerID: "old-host",
|
|
||||||
ReconnectToken: hostToken,
|
|
||||||
})
|
})
|
||||||
reconnected.expectAuthority(relayTypeJoined, "old-host")
|
creator.expectAuthority(relayTypeCreated, "new-host")
|
||||||
|
|
||||||
h.srv.mu.RLock()
|
h.srv.mu.RLock()
|
||||||
current := h.srv.rooms["STALE"]
|
current := h.srv.rooms["STALE"]
|
||||||
h.srv.mu.RUnlock()
|
h.srv.mu.RUnlock()
|
||||||
if current != original {
|
if current == original {
|
||||||
t.Fatal("reserved room identity was replaced")
|
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) {
|
func TestCreateReclaimsOwnedEmptyRoomWithoutDoubleCharging(t *testing.T) {
|
||||||
|
|||||||
@@ -698,7 +698,7 @@ void main() {
|
|||||||
});
|
});
|
||||||
|
|
||||||
group('WatchTogetherProvider — relay authority', () {
|
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;
|
late final _ProviderRelay relay;
|
||||||
relay = await _ProviderRelay.start((socket, message) {
|
relay = await _ProviderRelay.start((socket, message) {
|
||||||
if (message['type'] == 'join') {
|
if (message['type'] == 'join') {
|
||||||
@@ -708,6 +708,7 @@ void main() {
|
|||||||
'hostPeerId': _providerHostId,
|
'hostPeerId': _providerHostId,
|
||||||
'reconnectToken': message['reconnectToken'],
|
'reconnectToken': message['reconnectToken'],
|
||||||
'protocolVersion': 2,
|
'protocolVersion': 2,
|
||||||
|
'peers': [_providerHostId],
|
||||||
});
|
});
|
||||||
} else if (message['type'] == 'leave') {
|
} else if (message['type'] == 'leave') {
|
||||||
relay.send(socket, {
|
relay.send(socket, {
|
||||||
@@ -726,7 +727,7 @@ void main() {
|
|||||||
provider.dispose();
|
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(becameHost, isFalse);
|
||||||
expect(provider.isHost, isFalse);
|
expect(provider.isHost, isFalse);
|
||||||
@@ -746,6 +747,60 @@ void main() {
|
|||||||
expect(relay.messages.where((message) => message['type'] == 'create'), isEmpty);
|
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 {
|
test('a room-not-found probe creates with relay-declared host authority', () async {
|
||||||
late final _ProviderRelay relay;
|
late final _ProviderRelay relay;
|
||||||
relay = await _ProviderRelay.start((socket, message) {
|
relay = await _ProviderRelay.start((socket, message) {
|
||||||
|
|||||||
Reference in New Issue
Block a user