haeaeae/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(watch-together): derive the host role from the relay's authority

Browse Source 
A host that handed the room to someone else and then lost its connection
could not get back in, and a guest promoted while it was offline came back
as a guest.

The relay names the host in every admission and every hostChanged, but the
client also kept its own _isHost flag beside that identity. Reconnect adopted
the relay's host id without recomputing the flag, so the two disagreed: a
demoted host still required the response to name itself and rejected its own
legitimate re-admission as an invalid response, retry after retry; a promoted
guest updated the session but left the transport a guest, so ending the
session sent leave instead of endSession and room re-creation stayed off.

The role is now derived from the host identity rather than stored next to it.
Before the relay has admitted us there is no authority yet, so the role is
the one we announced, which is what releasing a possibly-committed setup has
to go by. The identity assertion moves from mutable role state to message
semantics: a created response must name us, a joined response is adopted.
This commit is contained in:
edde746
2026-09-06 00:53:28 +02:00
parent 65af57dd93
commit f738e74994
2 changed files with 123 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/watch_together/services/watch_together_peer_service.dart
+14 -7
View File
@@ -67,7 +67,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
final Set<String> _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;
test/watch_together/watch_together_peer_service_test.dart
+109
View File
@@ -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 = <PeerError>[];
final subscription = service.onError.listen(errors.add);
addTearDown(subscription.cancel);
final changed = Completer<String>();
final hostChanges = service.onHostChanged.listen((peerId) {
if (!changed.isCompleted) changed.complete(peerId);
});
addTearDown(hostChanges.cancel);
final reconnected = Completer<void>();
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<String>();
final subscription = service.onHostChanged.listen((peerId) {
if (!changed.isCompleted) changed.complete(peerId);
});
addTearDown(subscription.cancel);
final reconnected = Completer<void>();
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']);
});
}