diff --git a/lib/watch_together/services/watch_together_peer_service.dart b/lib/watch_together/services/watch_together_peer_service.dart index c225d4ccf..c0d08e586 100644 --- a/lib/watch_together/services/watch_together_peer_service.dart +++ b/lib/watch_together/services/watch_together_peer_service.dart @@ -67,7 +67,7 @@ class WatchTogetherPeerService with KeepaliveMixin { final Set _connectedPeers = {}; String? _sessionId; String? _myPeerId; - bool _isHost = false; + bool _announcedAsHost = false; String? _reconnectToken; String? _hostPeerId; @@ -135,7 +135,15 @@ class WatchTogetherPeerService with KeepaliveMixin { /// Relay-declared peer ID whose messages carry host authority. String? get hostPeerId => _hostPeerId; - /// Whether this peer is the host + /// Whether this peer is the host. + /// + /// Derived, never stored: the relay is the authority on host identity and + /// names it in every admission and every `hostChanged`. Before it has + /// admitted us there is no authority yet, so the role is the one we + /// announced — which is what a release of a possibly-committed setup has to + /// go by. + bool get _isHost => _hostPeerId == null ? _announcedAsHost : _hostPeerId == _myPeerId; + bool get isHost => _isHost; /// Whether currently connected to a session @@ -269,7 +277,7 @@ class WatchTogetherPeerService with KeepaliveMixin { responseSessionId != _sessionId || hostPeerId is! String || !RelayProtocol.isValidPeerId(hostPeerId) || - (_isHost && hostPeerId != _myPeerId) || + (type == RelayProtocol.created && hostPeerId != _myPeerId) || reconnectToken is! String || reconnectToken != _reconnectToken || !RelayProtocol.isValidReconnectToken(reconnectToken) || @@ -453,7 +461,6 @@ class WatchTogetherPeerService with KeepaliveMixin { if (newHostPeerId == _hostPeerId) break; // Duplicate delivery. appLogger.d('WatchTogether: Host authority moved to $newHostPeerId'); _hostPeerId = newHostPeerId; - _isHost = newHostPeerId == _myPeerId; _safeAdd(_hostChangedController, newHostPeerId); case RelayProtocol.error: @@ -664,7 +671,7 @@ class WatchTogetherPeerService with KeepaliveMixin { 'Must be 1–${RelayProtocol.maxSessionIdLength} letters, digits, _ or -', ); } - _isHost = true; + _announcedAsHost = true; _sessionId = resolvedSessionId; _myPeerId = const Uuid().v4(); _reconnectToken = _mintReconnectToken(); @@ -702,7 +709,7 @@ class WatchTogetherPeerService with KeepaliveMixin { 'Must be 1–${RelayProtocol.maxSessionIdLength} letters, digits, _ or -', ); } - _isHost = false; + _announcedAsHost = false; _sessionId = resolvedSessionId; _myPeerId = const Uuid().v4(); _reconnectToken = _mintReconnectToken(); @@ -833,7 +840,7 @@ class WatchTogetherPeerService with KeepaliveMixin { _myPeerId = null; _reconnectToken = null; _hostPeerId = null; - _isHost = false; + _announcedAsHost = false; _reconnectAttempts = 0; _teardownInProgress = false; diff --git a/test/watch_together/watch_together_peer_service_test.dart b/test/watch_together/watch_together_peer_service_test.dart index c829a2c7a..00355f077 100644 --- a/test/watch_together/watch_together_peer_service_test.dart +++ b/test/watch_together/watch_together_peer_service_test.dart @@ -1337,4 +1337,113 @@ void main() { expect(service.isHost, isFalse); expect(errors, isEmpty); }); + + test('a demoted host reconnect adopts the authority the relay names', () async { + // The relay only broadcasts hostChanged to peers connected at the time, so + // a host that handed the room over and then dropped learns of its own + // demotion from its re-admission. Rejecting it locks the former host out + // of a room that is still running. + late final _RelayServer relay; + relay = await relayWith((connection, socket, message) { + if (connection == 0 && message['type'] == 'create') { + relay.send(socket, { + 'type': 'created', + 'sessionId': message['sessionId'], + 'hostPeerId': message['peerId'], + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + }); + } else if (connection >= 1 && message['type'] == 'join') { + relay.send(socket, { + 'type': 'joined', + 'sessionId': message['sessionId'], + 'hostPeerId': 'new-host', + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + 'peers': const ['new-host'], + }); + } + }); + final service = serviceFor(relay); + final errors = []; + final subscription = service.onError.listen(errors.add); + addTearDown(subscription.cancel); + final changed = Completer(); + final hostChanges = service.onHostChanged.listen((peerId) { + if (!changed.isCompleted) changed.complete(peerId); + }); + addTearDown(hostChanges.cancel); + final reconnected = Completer(); + service.onReconnected = reconnected.complete; + + await _withShortenedTimer( + original: const Duration(seconds: 2), + replacement: const Duration(milliseconds: 10), + body: () => service.createSession(sessionId: 'XFER9'), + ); + await relay.sockets.single.close(); + await reconnected.future.timeout(const Duration(seconds: 6)); + + expect(await changed.future.timeout(const Duration(seconds: 1)), 'new-host'); + expect(service.hostPeerId, 'new-host'); + expect(service.isHost, isFalse); + expect(errors, isEmpty); + }); + + test('a promotion learned from re-admission moves the transport role too', () async { + late final _RelayServer relay; + relay = await relayWith((connection, socket, message) { + switch (message['type']) { + case 'join': + relay.send(socket, { + 'type': 'joined', + 'sessionId': message['sessionId'], + // The reconnect: the relay made this peer the host while it was + // offline, so the hostChanged broadcast never reached it. + 'hostPeerId': connection == 0 ? _relayHostId : message['peerId'], + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + 'peers': const ['other-guest'], + }); + case 'endSession': + relay.send(socket, { + 'type': 'ended', + 'sessionId': message['sessionId'], + 'peerId': message['peerId'], + 'protocolVersion': 2, + }); + case 'leave': + relay.send(socket, { + 'type': 'left', + 'sessionId': message['sessionId'], + 'peerId': message['peerId'], + 'protocolVersion': 2, + }); + } + }); + final service = serviceFor(relay); + final changed = Completer(); + final subscription = service.onHostChanged.listen((peerId) { + if (!changed.isCompleted) changed.complete(peerId); + }); + addTearDown(subscription.cancel); + final reconnected = Completer(); + service.onReconnected = reconnected.complete; + + await _withShortenedTimer( + original: const Duration(seconds: 2), + replacement: const Duration(milliseconds: 10), + body: () => service.joinSession('xfer8'), + ); + await relay.sockets.single.close(); + await reconnected.future.timeout(const Duration(seconds: 6)); + + expect(await changed.future.timeout(const Duration(seconds: 1)), service.myPeerId); + expect(service.isHost, isTrue); + + // The role the relay declared is the one the transport acts on: a host + // destroys the room instead of quietly leaving it behind. + await service.releaseSession(); + expect(relay.messages[1].map((message) => message['type']), ['join', 'endSession']); + }); }