diff --git a/README.md b/README.md
index 0d7ec6d87..7ddd8cbac 100644
--- a/README.md
+++ b/README.md
@@ -136,6 +136,10 @@ sudo moss it plezy
- Synchronized playback with friends
- Real-time play / pause / seek sync
- Host handoff when the relay and every connected peer support safe transfers
+- Automatic reconnect authenticates retained room membership; it never silently joins a reused room code.
+
+Self-hosted relays must support `authenticatedResume` for recovery. Deploy the updated relay before updating clients.
+Older relays still accept explicit create/join, but failed recovery requires joining or creating a room again.
### 
Integrations
- Discord Rich Presence[^desktop]
diff --git a/lib/watch_together/services/relay_protocol.g.dart b/lib/watch_together/services/relay_protocol.g.dart
index 693c059c3..af522c138 100644
--- a/lib/watch_together/services/relay_protocol.g.dart
+++ b/lib/watch_together/services/relay_protocol.g.dart
@@ -6,6 +6,7 @@ abstract final class RelayProtocol {
static const String create = 'create';
static const String join = 'join';
+ static const String resume = 'resume';
static const String broadcast = 'broadcast';
static const String sendTo = 'sendTo';
static const String ping = 'ping';
@@ -14,6 +15,7 @@ abstract final class RelayProtocol {
static const String transferHost = 'transferHost';
static const String created = 'created';
static const String joined = 'joined';
+ static const String resumed = 'resumed';
static const String peerJoined = 'peerJoined';
static const String peerLeft = 'peerLeft';
static const String message = 'message';
@@ -36,6 +38,7 @@ abstract final class RelayProtocol {
static const String peerNotFoundCode = 'peer_not_found';
static const String hostTransferUnavailableCode = 'host_transfer_unavailable';
static const String atomicHostTransferFeature = 'atomicHostTransfer';
+ static const String authenticatedResumeFeature = 'authenticatedResume';
static const String hostTransferCapability = 'hostTransfer';
static const int maxRoomSize = 8;
diff --git a/lib/watch_together/services/watch_together_peer_service.dart b/lib/watch_together/services/watch_together_peer_service.dart
index 3e040f2f4..534c455f1 100644
--- a/lib/watch_together/services/watch_together_peer_service.dart
+++ b/lib/watch_together/services/watch_together_peer_service.dart
@@ -69,6 +69,8 @@ class WatchTogetherPeerService with KeepaliveMixin {
StreamSubscription? _channelSubscription;
Completer? _setupCompleter;
String? _setupRequestType;
+ bool _admitted = false;
+ bool _initialRequestMayHaveCommitted = false;
final Set _connectedPeers = {};
String? _sessionId;
String? _myPeerId;
@@ -126,7 +128,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
/// Stream of connection state changes (true = connected, false = disconnected)
Stream get onConnectionStateChanged => _connectionStateController.stream;
- /// Emitted when the host has durably ended the relay room.
+ /// Emitted when this session can no longer be continued safely, for either role.
Stream get onSessionEnded => _sessionEndedController.stream;
/// Emitted with the new host's peer ID when the relay reassigns host
@@ -140,6 +142,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
bool canTransferHostTo(String peerId) =>
!_teardownInProgress &&
+ _admitted &&
_channel != null &&
_isHost &&
_relayEnforcesHostTransfer &&
@@ -163,16 +166,15 @@ class WatchTogetherPeerService with KeepaliveMixin {
/// 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.
+ /// names it in every admission and every `hostChanged`. Before admission,
+ /// the requested role is provisional; disconnected release always resumes
+ /// authenticated membership before choosing a terminal operation.
bool get _isHost => _hostPeerId == null ? _announcedAsHost : _hostPeerId == _myPeerId;
bool get isHost => _isHost;
/// Whether currently connected to a session
- bool get isConnected => _channel != null && _connectedPeers.isNotEmpty;
+ bool get isConnected => _admitted && _channel != null && _connectedPeers.isNotEmpty;
/// List of connected peer IDs
List get connectedPeers => _connectedPeers.toList();
@@ -222,6 +224,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
Duration? connectTimeout,
String connectOperation = 'WatchTogether connect',
}) async {
+ _requireCurrentConnection(epoch);
final channel = await _connectToRelay(timeout: connectTimeout, operation: connectOperation);
if (_disposed || epoch != _connectionEpoch || _sessionId == null) {
unawaited(channel.sink.close());
@@ -239,17 +242,23 @@ class WatchTogetherPeerService with KeepaliveMixin {
final completer = Completer();
_setupCompleter = completer;
_setupRequestType = type;
- if (type == RelayProtocol.create || type == RelayProtocol.join) {
+ final admission = type == RelayProtocol.create || type == RelayProtocol.join || type == RelayProtocol.resume;
+ if (admission) {
+ _admitted = false;
_clearHostTransferEligibility(resetFeature: true);
}
final reconnectToken = _reconnectToken;
+ if (type == RelayProtocol.create || type == RelayProtocol.join) {
+ // Enqueue failures cannot prove that the relay did not commit admission.
+ _initialRequestMayHaveCommitted = true;
+ }
_sendRaw({
'type': type,
'sessionId': _sessionId,
'peerId': _myPeerId,
'reconnectToken': ?reconnectToken,
'protocolVersion': _relayProtocolVersion,
- if (type == RelayProtocol.create || type == RelayProtocol.join) ...{
+ if (admission) ...{
'syncProtocolVersion': SyncMessage.protocolVersion,
'capabilities': [RelayProtocol.hostTransferCapability],
},
@@ -301,6 +310,13 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
List _acceptSetupResponse(Map msg, String type) {
+ final expectedType = switch (_setupRequestType) {
+ RelayProtocol.create => RelayProtocol.created,
+ RelayProtocol.join => RelayProtocol.joined,
+ RelayProtocol.resume => RelayProtocol.resumed,
+ _ => null,
+ };
+ if (_setupCompleter == null || type != expectedType) throw _invalidSetupResponse(type);
final responseSessionId = msg['sessionId'];
final hostPeerId = msg['hostPeerId'];
final reconnectToken = msg['reconnectToken'];
@@ -328,10 +344,15 @@ class WatchTogetherPeerService with KeepaliveMixin {
peers.add(peerId);
}
}
+ final features = msg['features'];
+ if (type == RelayProtocol.resumed &&
+ (features is! List || !features.contains(RelayProtocol.authenticatedResumeFeature))) {
+ throw _invalidSetupResponse(type);
+ }
_hostPeerId = hostPeerId;
_reconnectToken = reconnectToken;
- final features = msg['features'];
+ _admitted = true;
_relayEnforcesHostTransfer = features is List && features.contains(RelayProtocol.atomicHostTransferFeature);
_clearHostTransferEligibility();
return peers;
@@ -347,6 +368,17 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
void _failSetup(PeerError error) {
+ if (_setupRequestType == RelayProtocol.resume && !_isRetryableInitialSetupError(error)) {
+ error = PeerError(
+ type: error.type,
+ message: t.watchTogether.errors.sessionUnavailable,
+ serverCode: error.serverCode,
+ );
+ if (!_initialSetupInProgress && !_teardownInProgress) {
+ _handleSessionEnded(error);
+ return;
+ }
+ }
_safeAdd(_errorController, error);
if (_setupCompleter case final completer? when !completer.isCompleted) {
_setupCompleter = null;
@@ -355,30 +387,20 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
}
- bool _isExhaustedGuestReconnectRoomNotFound(String code) =>
- code == RelayProtocol.roomNotFoundCode &&
- !_isHost &&
- !_initialSetupInProgress &&
- !_teardownInProgress &&
- _hostPeerId != null &&
- _setupRequestType == RelayProtocol.join &&
- _reconnectAttempts >= _maxReconnectAttempts;
-
- void _handleGuestSessionEnded() {
- _teardownInProgress = true;
- ++_connectionEpoch;
- _reconnectTimer?.cancel();
- _reconnectTimer = null;
- stopKeepalive();
- final error = PeerError(type: PeerErrorType.invalidSession, message: t.watchTogether.errors.sessionEnded);
- if (_setupCompleter case final completer? when !completer.isCompleted) {
- _setupCompleter = null;
- _setupRequestType = null;
- _safeAdd(_errorController, error);
- completer.completeError(error);
+ void _handleSessionEnded([PeerError? error]) {
+ final endedError =
+ error ?? PeerError(type: PeerErrorType.invalidSession, message: t.watchTogether.errors.sessionEnded);
+ final completer = _setupCompleter;
+ _setupCompleter = null;
+ _setupRequestType = null;
+ if (completer != null && !completer.isCompleted) {
+ completer.completeError(endedError);
}
+ _safeAdd(_errorController, endedError);
+ // Clear credentials and fence the channel synchronously before notifying
+ // consumers, so terminal cleanup cannot recover into a replacement room.
+ unawaited(disconnect());
_safeAdd(_sessionEndedController, null);
- _handleWebSocketClosed();
}
/// Handle an incoming server message (JSON string).
@@ -388,7 +410,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
final type = msg['type'] as String?;
switch (type) {
- case RelayProtocol.created || RelayProtocol.joined:
+ case RelayProtocol.created || RelayProtocol.joined || RelayProtocol.resumed:
final previousHostPeerId = _hostPeerId;
late final List peers;
try {
@@ -418,6 +440,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
case RelayProtocol.peerJoined:
+ if (!_admitted) break;
final peerId = msg['peerId'] as String;
appLogger.d('WatchTogether: Peer joined: $peerId');
_connectedPeers.add(peerId);
@@ -425,6 +448,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
_safeAdd(_connectionStateController, true);
case RelayProtocol.peerLeft:
+ if (!_admitted) break;
final peerId = msg['peerId'] as String;
appLogger.d('WatchTogether: Peer left: $peerId');
_connectedPeers.remove(peerId);
@@ -434,6 +458,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
case RelayProtocol.message:
+ if (!_admitted) break;
final payload = msg['payload'];
final serverFrom = msg['from'] as String?;
if (payload != null) {
@@ -452,12 +477,18 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
case RelayProtocol.left:
+ if (_setupRequestType != RelayProtocol.leave) {
+ _failSetup(_invalidSetupResponse(RelayProtocol.left));
+ break;
+ }
try {
_acceptTeardownResponse(msg, RelayProtocol.left);
} on PeerError catch (error) {
_failSetup(error);
break;
}
+ _admitted = false;
+ _reconnectToken = null;
if (_setupCompleter case final completer? when !completer.isCompleted) {
_setupCompleter = null;
_setupRequestType = null;
@@ -474,18 +505,23 @@ class WatchTogetherPeerService with KeepaliveMixin {
final expectedTeardown =
_setupRequestType == RelayProtocol.endSession || _setupRequestType == RelayProtocol.leave;
if (expectedTeardown) {
+ _admitted = false;
+ _reconnectToken = null;
if (_setupCompleter case final completer? when !completer.isCompleted) {
_setupCompleter = null;
_setupRequestType = null;
completer.complete();
}
- } else if (!_isHost) {
- _handleGuestSessionEnded();
+ } else if (_admitted ||
+ _setupRequestType == RelayProtocol.resume ||
+ _setupRequestType == RelayProtocol.join) {
+ _handleSessionEnded();
} else {
_failSetup(_invalidSetupResponse(RelayProtocol.ended));
}
case RelayProtocol.hostChanged:
+ if (!_admitted) break;
final newHostPeerId = msg['hostPeerId'];
if (msg['sessionId'] != _sessionId ||
newHostPeerId is! String ||
@@ -500,6 +536,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
_safeAdd(_hostChangedController, newHostPeerId);
case RelayProtocol.hostTransferEligibility:
+ if (!_admitted) break;
final targets = msg['hostTransferTargets'];
if (!_relayEnforcesHostTransfer ||
msg['sessionId'] != _sessionId ||
@@ -517,19 +554,9 @@ class WatchTogetherPeerService with KeepaliveMixin {
case RelayProtocol.error:
final code = msg['code'] as String? ?? 'unknown';
final message = msg['message'] as String? ?? t.common.unknown;
- if (_isExhaustedGuestReconnectRoomNotFound(code)) {
- appLogger.d('WatchTogether: Room gone after reconnect retries; guest session ended');
- _handleGuestSessionEnded();
- break;
- }
appLogger.e('WatchTogether: Server error: $code - $message');
final error = PeerError(type: PeerErrorType.serverError, message: '$code: $message', serverCode: code);
- _safeAdd(_errorController, error);
- if (_setupCompleter case final completer? when !completer.isCompleted) {
- _setupCompleter = null;
- _setupRequestType = null;
- completer.completeError(error);
- }
+ _failSetup(error);
case RelayProtocol.pong:
// Handled by resetPongTimer() already
@@ -571,8 +598,9 @@ class WatchTogetherPeerService with KeepaliveMixin {
/// Handle the WebSocket being closed unexpectedly — attempt reconnection.
void _handleWebSocketClosed() {
final channel = _channel;
- final shouldReconnect = !_initialSetupInProgress && !_teardownInProgress;
+ final shouldReconnect = !_initialSetupInProgress && !_teardownInProgress && _reconnectToken != null;
if (shouldReconnect) ++_connectionEpoch;
+ _admitted = false;
stopKeepalive();
unawaited(_channelSubscription?.cancel());
_channelSubscription = null;
@@ -591,17 +619,13 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
}
- /// Attempt to reconnect to the relay and re-join/re-create the room.
+ /// Recover only the authenticated membership retained by the relay.
void _attemptReconnect(int epoch) {
if (_disposed || epoch != _connectionEpoch || _sessionId == null) return;
if (_reconnectAttempts >= _maxReconnectAttempts) {
appLogger.e('WatchTogether: Max reconnect attempts reached');
- _safeAdd(
- _errorController,
- const PeerError(
- type: PeerErrorType.connectionFailed,
- message: 'Lost connection to relay after multiple reconnect attempts',
- ),
+ _handleSessionEnded(
+ PeerError(type: PeerErrorType.connectionFailed, message: t.watchTogether.errors.sessionUnavailable),
);
return;
}
@@ -614,24 +638,9 @@ class WatchTogetherPeerService with KeepaliveMixin {
_reconnectTimer = Timer(delay, () async {
if (_disposed || epoch != _connectionEpoch || _sessionId == null) return;
try {
- final completer = await _connectAndAnnounce(RelayProtocol.join, epoch);
+ final completer = await _connectAndAnnounce(RelayProtocol.resume, epoch);
if (_disposed || epoch != _connectionEpoch) return;
-
- try {
- await completer.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect');
- } on PeerError catch (e) {
- if (_disposed || epoch != _connectionEpoch) return;
- if (_isHost && e.serverCode == RelayProtocol.roomNotFoundCode) {
- appLogger.d('WatchTogether: Room gone, re-creating as host');
- final createCompleter = _announce(RelayProtocol.create);
- await createCompleter.future.namedTimeout(
- const Duration(seconds: 10),
- operation: 'WatchTogether reconnect create',
- );
- } else {
- rethrow;
- }
- }
+ await completer.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect');
await debugReconnectSetupSucceededBarrier?.call();
@@ -646,6 +655,10 @@ class WatchTogetherPeerService with KeepaliveMixin {
} catch (e) {
if (_disposed || epoch != _connectionEpoch) return;
appLogger.e('WatchTogether: Reconnect failed', error: e);
+ if (!_isRetryableInitialSetupError(e)) {
+ _handleSessionEnded(e is PeerError ? e : null);
+ return;
+ }
_handleWebSocketClosed();
}
});
@@ -659,12 +672,23 @@ class WatchTogetherPeerService with KeepaliveMixin {
error.type == PeerErrorType.timeout)) ||
error is! PeerError;
+ void _requireCurrentConnection(int epoch) {
+ if (_disposed || epoch != _connectionEpoch || _sessionId == null) {
+ throw StateError('Watch Together connection attempt became stale');
+ }
+ }
+
Future _resetTransportForInitialRetry() async {
stopKeepalive();
final subscription = _channelSubscription;
final channel = _channel;
_channelSubscription = null;
_channel = null;
+ _admitted = false;
+ final completer = _setupCompleter;
+ if (completer != null && !completer.isCompleted) {
+ completer.completeError(StateError('Watch Together connection cancelled'));
+ }
_setupCompleter = null;
_setupRequestType = null;
_clearHostTransferEligibility(resetFeature: true);
@@ -681,12 +705,16 @@ class WatchTogetherPeerService with KeepaliveMixin {
try {
for (var attempt = 0; attempt < _maxReconnectAttempts; attempt++) {
try {
- final completer = await _connectAndAnnounce(type, epoch);
+ _requireCurrentConnection(epoch);
+ final requestType = _initialRequestMayHaveCommitted ? RelayProtocol.resume : type;
+ final completer = await _connectAndAnnounce(requestType, epoch);
await completer.future.timeout(debugInitialSetupTimeout, onTimeout: () => throw timeoutError);
+ _requireCurrentConnection(epoch);
return;
} catch (error) {
if (_disposed || epoch != _connectionEpoch) rethrow;
await _resetTransportForInitialRetry();
+ _requireCurrentConnection(epoch);
if (!_isRetryableInitialSetupError(error) || attempt + 1 >= _maxReconnectAttempts) {
rethrow;
}
@@ -694,12 +722,14 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
}
} finally {
- _initialSetupInProgress = false;
+ if (epoch == _connectionEpoch) _initialSetupInProgress = false;
}
}
- Future _bestEffortReleaseFailedSetup(Object setupError) async {
- if (!_isRetryableInitialSetupError(setupError)) return;
+ Future _bestEffortReleaseFailedSetup(Object setupError, int epoch) async {
+ if (epoch != _connectionEpoch || !_initialRequestMayHaveCommitted || !_isRetryableInitialSetupError(setupError)) {
+ return;
+ }
try {
await releaseSession();
} catch (releaseError) {
@@ -730,6 +760,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
_reconnectToken = _mintReconnectToken();
_reconnectAttempts = 0;
final epoch = ++_connectionEpoch;
+ final peerId = _myPeerId;
try {
await _performInitialSetup(
@@ -742,8 +773,13 @@ class WatchTogetherPeerService with KeepaliveMixin {
return _sessionId!;
} catch (e) {
appLogger.e('WatchTogether: Failed to create session', error: e);
- await _bestEffortReleaseFailedSetup(e);
- await disconnect();
+ if (epoch == _connectionEpoch) {
+ await _bestEffortReleaseFailedSetup(e, epoch);
+ // Release owns a new epoch; it must not clear a newer explicit entry.
+ if (_sessionId == resolvedSessionId && _myPeerId == peerId) {
+ await disconnect();
+ }
+ }
rethrow;
}
}
@@ -768,6 +804,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
_reconnectToken = _mintReconnectToken();
_reconnectAttempts = 0;
final epoch = ++_connectionEpoch;
+ final peerId = _myPeerId;
try {
await _performInitialSetup(
@@ -779,14 +816,19 @@ class WatchTogetherPeerService with KeepaliveMixin {
appLogger.d('WatchTogether: Joined session: $_sessionId');
} catch (e) {
appLogger.e('WatchTogether: Failed to join session', error: e);
- await _bestEffortReleaseFailedSetup(e);
- await disconnect();
+ if (epoch == _connectionEpoch) {
+ await _bestEffortReleaseFailedSetup(e, epoch);
+ if (_sessionId == resolvedSessionId && _myPeerId == peerId) {
+ await disconnect();
+ }
+ }
rethrow;
}
}
/// Broadcast a message to all connected peers
void broadcast(SyncMessage message) {
+ if (!_admitted) return;
final payload = message.toJson();
_sendRaw({'type': RelayProtocol.broadcast, 'payload': payload});
}
@@ -796,6 +838,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
if (!RelayProtocol.isValidPeerId(peerId)) {
throw ArgumentError.value(peerId, 'peerId', 'Must be 1–${RelayProtocol.maxPeerIdLength} letters, digits, _ or -');
}
+ if (!_admitted) return;
final payload = message.toJson();
_sendRaw({'type': RelayProtocol.sendTo, 'to': peerId, 'payload': payload});
}
@@ -846,13 +889,16 @@ class WatchTogetherPeerService with KeepaliveMixin {
await _resetTransportForInitialRetry();
}
for (var attempt = 0; attempt < _maxReconnectAttempts; attempt++) {
+ _requireCurrentConnection(epoch);
// Which request a rejection below refers to. Re-admission failures
// are terminal answers about our identity; release failures are not.
var releaseRequested = false;
try {
- if (_channel == null) {
+ if (_channel == null || !_admitted) {
+ if (_channel != null) await _resetTransportForInitialRetry();
+ _requireCurrentConnection(epoch);
final reconnectCompleter = await _connectAndAnnounce(
- RelayProtocol.join,
+ RelayProtocol.resume,
epoch,
connectTimeout: debugReleaseConnectTimeout,
connectOperation: 'WatchTogether release reconnect',
@@ -863,6 +909,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
);
}
+ _requireCurrentConnection(epoch);
releaseRequested = true;
final releaseCompleter = _announce(_isHost ? RelayProtocol.endSession : RelayProtocol.leave);
await releaseCompleter.future.namedTimeout(
@@ -871,27 +918,38 @@ class WatchTogetherPeerService with KeepaliveMixin {
);
return;
} catch (error) {
+ _requireCurrentConnection(epoch);
if (error is PeerError) {
- if (_isRoomAbsentCode(error.serverCode)) return;
+ if (_isRoomAbsentCode(error.serverCode)) {
+ _admitted = false;
+ _reconnectToken = null;
+ return;
+ }
if (error.serverCode == RelayProtocol.peerIdUnavailableCode) {
// From re-admission: the token no longer names an identity
// here (a processed leave whose ACK was lost, an expired
// reservation). Nothing is left to release.
- if (!releaseRequested) return;
+ if (!releaseRequested) {
+ _admitted = false;
+ _reconnectToken = null;
+ return;
+ }
// From the release itself: the relay refused the operation we
// chose for the role we believed we had. A transfer that landed
// while we were tearing down (a guest promoted to host, a host
// demoted) makes exactly this rejection, and a role-mismatch
// rejection is not a release. Reauthenticate through the
- // token: the `joined` admission names the current host, and
+ // token: the `resumed` admission names the current host, and
// the next pass sends the operation that role requires.
appLogger.d('WatchTogether: Release rejected for role host=$_isHost; re-authenticating ownership');
await _resetTransportForInitialRetry();
+ _requireCurrentConnection(epoch);
if (attempt + 1 >= _maxReconnectAttempts) rethrow;
continue;
}
}
await _resetTransportForInitialRetry();
+ _requireCurrentConnection(epoch);
if (!_isRetryableInitialSetupError(error) || attempt + 1 >= _maxReconnectAttempts) {
rethrow;
}
@@ -899,7 +957,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
}
} finally {
- _teardownInProgress = false;
+ if (epoch == _connectionEpoch) _teardownInProgress = false;
}
}
@@ -907,7 +965,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
/// intentional exits call [releaseSession] before this cleanup step.
Future disconnect() async {
appLogger.d('WatchTogether: Disconnecting...');
- ++_connectionEpoch;
+ final epoch = ++_connectionEpoch;
_reconnectTimer?.cancel();
_reconnectTimer = null;
stopKeepalive();
@@ -919,6 +977,9 @@ class WatchTogetherPeerService with KeepaliveMixin {
final setupCompleter = _setupCompleter;
_setupCompleter = null;
_setupRequestType = null;
+ _admitted = false;
+ _initialSetupInProgress = false;
+ _initialRequestMayHaveCommitted = false;
if (setupCompleter != null && !setupCompleter.isCompleted) {
setupCompleter.completeError(StateError('Watch Together connection cancelled'));
}
@@ -938,7 +999,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
} catch (e) {
appLogger.d('WatchTogether: channel close ignored', error: e);
}
- _safeAdd(_connectionStateController, false);
+ if (epoch == _connectionEpoch) _safeAdd(_connectionStateController, false);
}
/// Dispose all resources.
diff --git a/relay_protocol.json b/relay_protocol.json
index 35b41fe1e..fc7af4589 100644
--- a/relay_protocol.json
+++ b/relay_protocol.json
@@ -4,6 +4,7 @@
"clientMessageTypes": {
"create": "create",
"join": "join",
+ "resume": "resume",
"broadcast": "broadcast",
"sendTo": "sendTo",
"ping": "ping",
@@ -14,6 +15,7 @@
"serverMessageTypes": {
"created": "created",
"joined": "joined",
+ "resumed": "resumed",
"peerJoined": "peerJoined",
"peerLeft": "peerLeft",
"message": "message",
@@ -39,7 +41,8 @@
"hostTransferUnavailable": "host_transfer_unavailable"
},
"features": {
- "atomicHostTransfer": "atomicHostTransfer"
+ "atomicHostTransfer": "atomicHostTransfer",
+ "authenticatedResume": "authenticatedResume"
},
"capabilities": {
"hostTransfer": "hostTransfer"
diff --git a/server/main.go b/server/main.go
index 637b9d966..dc26b3347 100644
--- a/server/main.go
+++ b/server/main.go
@@ -1936,7 +1936,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
ReconnectToken: reconnectToken,
ProtocolVersion: relayProtocolVersion,
Peers: existingPeers,
- Features: []string{relayFeatureAtomicHostTransfer},
+ Features: []string{relayFeatureAtomicHostTransfer, relayFeatureAuthenticatedResume},
})
if hostWasAbsent {
existing.broadcastExceptLocked(msg.PeerID, serverMsg{
@@ -2005,17 +2005,19 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
HostPeerID: msg.PeerID,
ReconnectToken: reconnectToken,
ProtocolVersion: msg.ProtocolVersion,
- Features: []string{relayFeatureAtomicHostTransfer},
+ Features: []string{relayFeatureAtomicHostTransfer, relayFeatureAuthenticatedResume},
})
room.publishHostTransferEligibilityLocked()
room.mu.Unlock()
- case relayTypeJoin:
+ case relayTypeJoin, relayTypeResume:
+ resumeOnly := msg.Type == relayTypeResume
if !validRelayID(msg.SessionID, maxSessionIDLength) || !validRelayID(msg.PeerID, maxPeerIDLength) {
client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorInvalidMessage, Message: "Invalid sessionId or peerId"})
continue
}
- if !supportedRelayProtocolVersion(msg.ProtocolVersion) {
+ if !supportedRelayProtocolVersion(msg.ProtocolVersion) ||
+ (resumeOnly && msg.ProtocolVersion != relayProtocolVersion) {
client.sendJSON(serverMsg{
Type: relayTypeError,
Code: relayErrorProtocolMismatch,
@@ -2028,10 +2030,15 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
continue
}
- newToken, newVerifier, err := mintReconnectToken()
- if err != nil {
- client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorInvalidMessage, Message: "Unable to join room"})
- continue
+ var newToken string
+ var newVerifier reconnectVerifier
+ if msg.ProtocolVersion == legacyRelayProtocolVersion {
+ var err error
+ newToken, newVerifier, err = mintReconnectToken()
+ if err != nil {
+ client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorInvalidMessage, Message: "Unable to join room"})
+ continue
+ }
}
presentedVerifier, tokenValid := reconnectVerifierFromToken(msg.ReconnectToken)
@@ -2093,7 +2100,8 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
case occupied:
authorized = false
default:
- authorized = tokenValid
+ // Only explicit initial admission may allocate an identity.
+ authorized = !resumeOnly && tokenValid
responseToken = msg.ReconnectToken
responseVerifier = presentedVerifier
}
@@ -2181,14 +2189,18 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
if s.beforeJoinRoomAck != nil {
s.beforeJoinRoomAck()
}
+ responseType := relayTypeJoined
+ if resumeOnly {
+ responseType = relayTypeResumed
+ }
client.sendJSON(serverMsg{
- Type: relayTypeJoined,
+ Type: responseType,
SessionID: msg.SessionID,
HostPeerID: room.HostPeerID,
ReconnectToken: responseToken,
ProtocolVersion: roomProtocolVersion,
Peers: existingPeers,
- Features: []string{relayFeatureAtomicHostTransfer},
+ Features: []string{relayFeatureAtomicHostTransfer, relayFeatureAuthenticatedResume},
})
room.broadcastExceptLocked(msg.PeerID, serverMsg{Type: relayTypePeerJoined, PeerID: msg.PeerID})
room.publishHostTransferEligibilityLocked()
diff --git a/server/main_test.go b/server/main_test.go
index cbdde990f..2d5909d47 100644
--- a/server/main_test.go
+++ b/server/main_test.go
@@ -2257,7 +2257,7 @@ func TestCreateNegotiatesModernProtocolWithClientKnownToken(t *testing.T) {
}
}
-func TestModernCreateRetryAfterLostSetupResponseIsIdempotent(t *testing.T) {
+func TestModernCreateLostSetupResponseResumesCommittedIdentity(t *testing.T) {
h := newRelayHarness(t)
hostToken, _ := mustReconnectToken(t)
create := clientMsg{
@@ -2282,8 +2282,10 @@ func TestModernCreateRetryAfterLostSetupResponseIsIdempotent(t *testing.T) {
}
retry := h.dial(t, "1.1.1.42")
- retry.send(create)
- created := retry.expectAuthority(relayTypeCreated, "H")
+ resume := create
+ resume.Type = relayTypeResume
+ retry.send(resume)
+ created := retry.expectAuthority(relayTypeResumed, "H")
if created.ReconnectToken != hostToken || created.ProtocolVersion != relayProtocolVersion {
t.Fatalf("retry authority changed: tokenMatch=%v protocol=%d", created.ReconnectToken == hostToken, created.ProtocolVersion)
}
@@ -3070,7 +3072,7 @@ func TestFullRoomAllowsOnlyAuthenticatedLiveReplacements(t *testing.T) {
unprovedHost := h.dial(t, "2.1.0.251")
unprovedHost.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "FULL",
PeerID: "H",
ProtocolVersion: relayProtocolVersion,
@@ -3078,7 +3080,7 @@ func TestFullRoomAllowsOnlyAuthenticatedLiveReplacements(t *testing.T) {
unprovedHost.expectError(relayErrorPeerIdUnavailable)
unprovedGuest := h.dial(t, "2.1.0.252")
unprovedGuest.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "FULL",
PeerID: "G1",
ProtocolVersion: relayProtocolVersion,
@@ -3086,7 +3088,7 @@ func TestFullRoomAllowsOnlyAuthenticatedLiveReplacements(t *testing.T) {
unprovedGuest.expectError(relayErrorPeerIdUnavailable)
wrongGuestToken, _ := mustReconnectToken(t)
unprovedGuest.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "FULL",
PeerID: "G1",
ReconnectToken: wrongGuestToken,
@@ -3096,13 +3098,13 @@ func TestFullRoomAllowsOnlyAuthenticatedLiveReplacements(t *testing.T) {
newHost := h.dial(t, "2.1.0.253")
newHost.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "FULL",
PeerID: "H",
ReconnectToken: created.ReconnectToken,
ProtocolVersion: relayProtocolVersion,
})
- hostJoined := newHost.expectAuthority(relayTypeJoined, "H")
+ hostJoined := newHost.expectAuthority(relayTypeResumed, "H")
if len(hostJoined.Peers) != maxRoomSize-1 {
t.Fatalf("replacement host peers=%v, want %d peers", hostJoined.Peers, maxRoomSize-1)
}
@@ -3116,13 +3118,13 @@ func TestFullRoomAllowsOnlyAuthenticatedLiveReplacements(t *testing.T) {
newGuest := h.dial(t, "2.1.0.254")
newGuest.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "FULL",
PeerID: "G1",
ReconnectToken: guestTokens["G1"],
ProtocolVersion: relayProtocolVersion,
})
- newGuest.expectAuthority(relayTypeJoined, "H")
+ newGuest.expectAuthority(relayTypeResumed, "H")
if messages, err := guests["G1"].recvUntilClosed(2 * time.Second); err != nil {
t.Fatalf("displaced guest did not close: %v (frames=%v)", err, messages)
}
@@ -3269,15 +3271,17 @@ func TestLegacySameSourceHostReconnectAndModernTokenEnforcement(t *testing.T) {
})
}
-func TestJoinAdmissionIsAtomicWithEmptyRoomCleanup(t *testing.T) {
+func TestResumeAdmissionIsAtomicWithEmptyRoomCleanup(t *testing.T) {
h := newRelayHarness(t)
_, hostVerifier := mustReconnectToken(t)
+ guestToken, guestVerifier := mustReconnectToken(t)
now := time.Now()
room := &Room{
SessionID: "ATOMIC_CLEANUP",
HostPeerID: "H",
hostVerifier: hostVerifier,
- peerReservations: make(map[string]peerReservation),
+ ProtocolVersion: relayProtocolVersion,
+ peerReservations: map[string]peerReservation{"G1": {verifier: guestVerifier, absentSince: now}},
Peers: make(map[string]*Client),
CreatedAt: now.Add(-time.Hour),
LastActivityAt: now.Add(-emptyRoomMaxAge - time.Second),
@@ -3297,7 +3301,10 @@ func TestJoinAdmissionIsAtomicWithEmptyRoomCleanup(t *testing.T) {
}
joiner := h.dial(t, "2.2.0.1")
- joiner.send(clientMsg{Type: relayTypeJoin, SessionID: room.SessionID, PeerID: "G1"})
+ joiner.send(clientMsg{
+ Type: relayTypeResume, SessionID: room.SessionID, PeerID: "G1",
+ ReconnectToken: guestToken, ProtocolVersion: relayProtocolVersion,
+ })
<-reached
cleanupDone := make(chan struct{})
@@ -3312,7 +3319,7 @@ func TestJoinAdmissionIsAtomicWithEmptyRoomCleanup(t *testing.T) {
}
close(release)
- joiner.expectAuthority(relayTypeJoined, "H")
+ joiner.expectAuthority(relayTypeResumed, "H")
<-cleanupDone
h.srv.mu.RLock()
@@ -3329,7 +3336,11 @@ func TestJoinAdmissionIsAtomicWithEmptyRoomCleanup(t *testing.T) {
}
second := h.dial(t, "2.2.0.2")
- second.send(clientMsg{Type: relayTypeJoin, SessionID: room.SessionID, PeerID: "G2"})
+ secondToken, _ := mustReconnectToken(t)
+ second.send(clientMsg{
+ Type: relayTypeJoin, SessionID: room.SessionID, PeerID: "G2",
+ ReconnectToken: secondToken, ProtocolVersion: relayProtocolVersion,
+ })
second.expectAuthority(relayTypeJoined, "H")
joiner.expect(relayTypePeerJoined)
second.send(clientMsg{Type: relayTypeBroadcast, Payload: json.RawMessage(`{"atomic":true}`)})
@@ -3338,15 +3349,17 @@ func TestJoinAdmissionIsAtomicWithEmptyRoomCleanup(t *testing.T) {
}
}
-func TestJoinAdmissionIsAtomicWithReservedRoomCreate(t *testing.T) {
+func TestResumeAdmissionIsAtomicWithReservedRoomCreate(t *testing.T) {
h := newRelayHarness(t)
_, hostVerifier := mustReconnectToken(t)
+ guestToken, guestVerifier := mustReconnectToken(t)
now := time.Now()
room := &Room{
SessionID: "ATOMIC_CREATE",
HostPeerID: "H",
hostVerifier: hostVerifier,
- peerReservations: make(map[string]peerReservation),
+ ProtocolVersion: relayProtocolVersion,
+ peerReservations: map[string]peerReservation{"G": {verifier: guestVerifier, absentSince: now}},
Peers: make(map[string]*Client),
CreatedAt: now,
LastActivityAt: now,
@@ -3366,11 +3379,18 @@ func TestJoinAdmissionIsAtomicWithReservedRoomCreate(t *testing.T) {
}
joiner := h.dial(t, "2.3.0.1")
- joiner.send(clientMsg{Type: relayTypeJoin, SessionID: room.SessionID, PeerID: "G"})
+ joiner.send(clientMsg{
+ Type: relayTypeResume, SessionID: room.SessionID, PeerID: "G",
+ ReconnectToken: guestToken, ProtocolVersion: relayProtocolVersion,
+ })
<-reached
creator := h.dial(t, "2.3.0.2")
- creator.send(clientMsg{Type: relayTypeCreate, SessionID: room.SessionID, PeerID: "OTHER"})
+ replacementToken, _ := mustReconnectToken(t)
+ creator.send(clientMsg{
+ Type: relayTypeCreate, SessionID: room.SessionID, PeerID: "OTHER",
+ ReconnectToken: replacementToken, ProtocolVersion: relayProtocolVersion,
+ })
type readResult struct {
message serverMsg
err error
@@ -3395,7 +3415,7 @@ func TestJoinAdmissionIsAtomicWithReservedRoomCreate(t *testing.T) {
}
close(release)
- joiner.expectAuthority(relayTypeJoined, "H")
+ joiner.expectAuthority(relayTypeResumed, "H")
result := <-createResult
if result.err != nil {
t.Fatalf("read create result: %v", result.err)
@@ -3642,13 +3662,13 @@ func TestStalePeerSkipsCleanupBroadcast(t *testing.T) {
g2 := h.dial(t, "6.1.0.3")
g2.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "D2",
PeerID: "G",
ReconnectToken: guestToken,
ProtocolVersion: relayProtocolVersion,
})
- g2.expectAuthority(relayTypeJoined, "H")
+ g2.expectAuthority(relayTypeResumed, "H")
host.expect(relayTypePeerJoined)
if messages, err := g1.recvUntilClosed(2 * time.Second); err != nil {
@@ -3670,6 +3690,195 @@ func TestStalePeerSkipsCleanupBroadcast(t *testing.T) {
}
}
+func TestResumeCannotEnterReplacementRoom(t *testing.T) {
+ for _, replacementHost := range []string{"NEW_HOST", "H"} {
+ t.Run(replacementHost, func(t *testing.T) {
+ h := newRelayHarness(t)
+ host, guest, hostToken, guestToken := createModernRoomWithGuest(t, h, "REUSED", "6.5.0.1", "6.5.0.2")
+ if err := guest.conn.Close(); err != nil {
+ t.Fatal(err)
+ }
+ host.expect(relayTypePeerLeft)
+ host.send(clientMsg{Type: relayTypeEndSession, ReconnectToken: hostToken, ProtocolVersion: relayProtocolVersion})
+ // The old client may lose this terminal ACK; the committed room is gone.
+ host.expect(relayTypeEnded)
+
+ replacementToken, _ := mustReconnectToken(t)
+ replacement := h.dial(t, "6.5.0.3")
+ replacement.send(clientMsg{
+ Type: relayTypeCreate, SessionID: "REUSED", PeerID: replacementHost,
+ ReconnectToken: replacementToken, ProtocolVersion: relayProtocolVersion,
+ })
+ replacement.expectAuthority(relayTypeCreated, replacementHost)
+ for index, identity := range []struct{ peerID, token string }{{"H", hostToken}, {"G", guestToken}} {
+ stale := h.dial(t, fmt.Sprintf("6.5.1.%d", index+1))
+ stale.send(clientMsg{
+ Type: relayTypeResume, SessionID: "REUSED", PeerID: identity.peerID,
+ ReconnectToken: identity.token, ProtocolVersion: relayProtocolVersion,
+ })
+ stale.expectError(relayErrorPeerIdUnavailable)
+ stale.send(clientMsg{Type: relayTypeBroadcast, Payload: json.RawMessage(`{"stale":true}`)})
+ stale.expectError(relayErrorNotInRoom)
+ stale.send(clientMsg{Type: relayTypeEndSession, ReconnectToken: identity.token, ProtocolVersion: relayProtocolVersion})
+ stale.expectError(relayErrorNotInRoom)
+ }
+ // No peerJoined or payload may precede this barrier on the replacement.
+ replacement.send(clientMsg{Type: relayTypePing})
+ replacement.expect(relayTypePong)
+ h.srv.mu.RLock()
+ room := h.srv.rooms["REUSED"]
+ room.mu.RLock()
+ reserved, connected := len(room.peerReservations), len(room.Peers)
+ room.mu.RUnlock()
+ h.srv.mu.RUnlock()
+ if reserved != 0 || connected != 1 {
+ t.Fatalf("rejected resumes allocated membership: reserved=%d connected=%d", reserved, connected)
+ }
+ })
+ }
+}
+
+func TestResumeLostJoinAndLeaveAcknowledgements(t *testing.T) {
+ h := newRelayHarness(t)
+ hostToken, _ := mustReconnectToken(t)
+ host := h.dial(t, "6.5.2.1")
+ host.send(clientMsg{
+ Type: relayTypeCreate, SessionID: "LOST_GUEST_ACK", PeerID: "H",
+ ReconnectToken: hostToken, ProtocolVersion: relayProtocolVersion,
+ })
+ host.expectAuthority(relayTypeCreated, "H")
+ guestToken, _ := mustReconnectToken(t)
+ admission := clientMsg{
+ Type: relayTypeJoin, SessionID: "LOST_GUEST_ACK", PeerID: "G",
+ ReconnectToken: guestToken, ProtocolVersion: relayProtocolVersion,
+ }
+ guest := h.dial(t, "6.5.2.2")
+ guest.send(admission)
+ host.expect(relayTypePeerJoined)
+ // Drop the first transport without reading its committed join ACK.
+ if err := guest.conn.Close(); err != nil {
+ t.Fatal(err)
+ }
+ host.expect(relayTypePeerLeft)
+ resumed := h.dial(t, "6.5.2.3")
+ admission.Type = relayTypeResume
+ resumed.send(admission)
+ ack := resumed.expectAuthority(relayTypeResumed, "H")
+ if !slices.Contains(ack.Features, relayFeatureAuthenticatedResume) {
+ t.Fatal("resume did not advertise authenticated admission")
+ }
+ host.expect(relayTypePeerJoined)
+ resumed.send(clientMsg{Type: relayTypeLeave, ReconnectToken: guestToken, ProtocolVersion: relayProtocolVersion})
+ host.expect(relayTypePeerLeft)
+ // The leave committed, but the departing peer never reads its ACK.
+ if err := resumed.conn.Close(); err != nil {
+ t.Fatal(err)
+ }
+ cleanup := h.dial(t, "6.5.2.4")
+ cleanup.send(admission)
+ cleanup.expectError(relayErrorPeerIdUnavailable)
+ cleanup.send(clientMsg{Type: relayTypeLeave, ReconnectToken: guestToken, ProtocolVersion: relayProtocolVersion})
+ cleanup.expectError(relayErrorNotInRoom)
+ host.send(clientMsg{Type: relayTypePing})
+ host.expect(relayTypePong)
+}
+
+func TestResumeRequiresUnexpiredExistingMembership(t *testing.T) {
+ stateFile := filepath.Join(t.TempDir(), "rooms.json")
+ h := newRelayHarnessAt(t, t.TempDir(), stateFile)
+ host, guest, _, guestToken := createModernRoomWithGuest(t, h, "EXPIRING_RESUME", "6.5.3.1", "6.5.3.2")
+ if err := guest.conn.Close(); err != nil {
+ t.Fatal(err)
+ }
+ host.expect(relayTypePeerLeft)
+ h.srv.mu.RLock()
+ room := h.srv.rooms["EXPIRING_RESUME"]
+ h.srv.mu.RUnlock()
+ expireDisconnectedReservations(room, time.Now())
+
+ resume := clientMsg{
+ Type: relayTypeResume, SessionID: "EXPIRING_RESUME", PeerID: "G",
+ ReconnectToken: guestToken, ProtocolVersion: relayProtocolVersion,
+ }
+ probe := h.dial(t, "6.5.3.3")
+ probe.send(resume)
+ probe.expectError(relayErrorPeerIdUnavailable)
+ resume.PeerID = "NEVER_JOINED"
+ probe.send(resume)
+ probe.expectError(relayErrorPeerIdUnavailable)
+ resume.SessionID = "NEVER_CREATED"
+ probe.send(resume)
+ probe.expectError(relayErrorRoomNotFound)
+ resume.SessionID = "EXPIRING_RESUME"
+ resume.ProtocolVersion = legacyRelayProtocolVersion
+ probe.send(resume)
+ probe.expectError(relayErrorProtocolMismatch)
+ host.send(clientMsg{Type: relayTypePing})
+ host.expect(relayTypePong)
+
+ // Admission-time expiry is persisted; a restart cannot resurrect the guest.
+ if err := h.srv.snap.flushAndStop(2 * time.Second); err != nil {
+ t.Fatalf("flush reservation expiry: %v", err)
+ }
+ restarted := newRelayHarnessAt(t, t.TempDir(), copySnapshotForRestart(t, stateFile))
+ afterRestart := restarted.dial(t, "6.5.3.4")
+ resume.PeerID = "G"
+ resume.ProtocolVersion = relayProtocolVersion
+ afterRestart.send(resume)
+ afterRestart.expectError(relayErrorPeerIdUnavailable)
+
+ // An explicit new join is still allowed to claim the freed identity.
+ freshToken, _ := mustReconnectToken(t)
+ resume.Type = relayTypeJoin
+ resume.ReconnectToken = freshToken
+ probe.send(resume)
+ probe.expectAuthority(relayTypeJoined, "H")
+ host.expect(relayTypePeerJoined)
+}
+
+func TestOfflineGuestResumesAfterHostTransfer(t *testing.T) {
+ h := newRelayHarness(t)
+ host, promoted, _, _ := createTransferRoomWithGuest(t, h, "OFFLINE_TRANSFER", "6.5.4.1", "6.5.4.2")
+ token, _ := mustReconnectToken(t)
+ offline := h.dial(t, "6.5.4.3")
+ admission := currentAdmission(clientMsg{
+ Type: relayTypeJoin, SessionID: "OFFLINE_TRANSFER", PeerID: "OFFLINE",
+ ReconnectToken: token, ProtocolVersion: relayProtocolVersion,
+ })
+ offline.send(admission)
+ offline.expectAuthority(relayTypeJoined, "H")
+ offline.expectEligibility("OFFLINE_TRANSFER", "H", "G", "OFFLINE")
+ host.expect(relayTypePeerJoined)
+ host.expectEligibility("OFFLINE_TRANSFER", "H", "G", "OFFLINE")
+ promoted.expect(relayTypePeerJoined)
+ promoted.expectEligibility("OFFLINE_TRANSFER", "H", "G", "OFFLINE")
+ if err := offline.conn.Close(); err != nil {
+ t.Fatal(err)
+ }
+ host.expect(relayTypePeerLeft)
+ host.expectEligibility("OFFLINE_TRANSFER", "H", "G")
+ promoted.expect(relayTypePeerLeft)
+ promoted.expectEligibility("OFFLINE_TRANSFER", "H", "G")
+ host.send(clientMsg{Type: relayTypeTransferHost, To: "G", ProtocolVersion: relayProtocolVersion})
+ host.expect(relayTypeHostChanged)
+ host.expectEligibility("OFFLINE_TRANSFER", "G", "H")
+ promoted.expect(relayTypeHostChanged)
+ promoted.expectEligibility("OFFLINE_TRANSFER", "G", "H")
+ admission.Type = relayTypeResume
+ resumed := h.dial(t, "6.5.4.4")
+ resumed.send(admission)
+ resumed.expectAuthority(relayTypeResumed, "G")
+ resumed.expectEligibility("OFFLINE_TRANSFER", "G", "H", "OFFLINE")
+ host.expect(relayTypePeerJoined)
+ host.expectEligibility("OFFLINE_TRANSFER", "G", "H", "OFFLINE")
+ promoted.expect(relayTypePeerJoined)
+ promoted.expectEligibility("OFFLINE_TRANSFER", "G", "H", "OFFLINE")
+ resumed.send(clientMsg{Type: relayTypeSendTo, To: "G", Payload: json.RawMessage(`{"resumed":true}`)})
+ if received := promoted.expect(relayTypeMessage); received.From != "OFFLINE" {
+ t.Fatalf("resumed sender=%q, want OFFLINE", received.From)
+ }
+}
+
func TestDisconnectedModernGuestIdentityRejectsTheftAndAcceptsRightfulReconnect(t *testing.T) {
h := newRelayHarness(t)
hostToken, _ := mustReconnectToken(t)
@@ -3718,7 +3927,7 @@ func TestDisconnectedModernGuestIdentityRejectsTheftAndAcceptsRightfulReconnect(
thiefToken, _ := mustReconnectToken(t)
thief := h.dial(t, "6.1.0.12")
thief.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "GUEST_RECONNECT",
PeerID: "G",
ReconnectToken: thiefToken,
@@ -3730,13 +3939,13 @@ func TestDisconnectedModernGuestIdentityRejectsTheftAndAcceptsRightfulReconnect(
rightful := h.dial(t, "6.1.0.13")
rightful.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "GUEST_RECONNECT",
PeerID: "G",
ReconnectToken: guestToken,
ProtocolVersion: relayProtocolVersion,
})
- joined := rightful.expectAuthority(relayTypeJoined, "H")
+ joined := rightful.expectAuthority(relayTypeResumed, "H")
if joined.ReconnectToken != guestToken {
t.Fatal("rightful reconnect rotated the retained guest token")
}
@@ -3921,7 +4130,7 @@ func TestAdmissionPrunePersistsWhenJoinRejected(t *testing.T) {
wrongHostToken, _ := mustReconnectToken(t)
rejected := h.dial(t, "6.2.2.3")
rejected.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "REJECTED_AFTER_PRUNE",
PeerID: "H",
ReconnectToken: wrongHostToken,
@@ -3986,7 +4195,7 @@ func TestGuestReservationSnapshotV4Migration(t *testing.T) {
statePath := filepath.Join(t.TempDir(), "rooms.json")
now := time.Now().UTC()
_, hostVerifier := mustReconnectToken(t)
- _, guestVerifier := mustReconnectToken(t)
+ guestToken, guestVerifier := mustReconnectToken(t)
legacy := stateSnapshot{
Version: 3,
SavedAt: now,
@@ -4039,6 +4248,12 @@ func TestGuestReservationSnapshotV4Migration(t *testing.T) {
if runtimeReservation.absentSince.IsZero() {
t.Fatal("legacy reservation was restored as connected")
}
+ guest := h.dial(t, "6.2.4.1")
+ guest.send(clientMsg{
+ Type: relayTypeResume, SessionID: "V3_MIGRATION", PeerID: "G",
+ ReconnectToken: guestToken, ProtocolVersion: relayProtocolVersion,
+ })
+ guest.expectAuthority(relayTypeResumed, "H")
}
func TestSnapshotV2LoadsAndRewritesV4(t *testing.T) {
@@ -4081,6 +4296,12 @@ func TestSnapshotV2LoadsAndRewritesV4(t *testing.T) {
if snapshot.Version != snapshotFormatVersion {
t.Fatalf("v2 rewrite version=%d, want %d", snapshot.Version, snapshotFormatVersion)
}
+ legacyHost := h.dial(t, "6.2.4.2")
+ legacyHost.send(clientMsg{
+ Type: relayTypeResume, SessionID: "V2_MIGRATION", PeerID: "H",
+ ProtocolVersion: legacyRelayProtocolVersion,
+ })
+ legacyHost.expectError(relayErrorProtocolMismatch)
}
func TestSnapshotV4RetainsGuestAbsenceAcrossRestart(t *testing.T) {
@@ -4722,7 +4943,7 @@ func TestLeavePendingReservationRejectsReplacement(t *testing.T) {
matching := h.dial(t, "6.3.1.3")
matching.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "PENDING_RELEASE",
PeerID: "G",
ReconnectToken: guestToken,
@@ -4732,7 +4953,7 @@ func TestLeavePendingReservationRejectsReplacement(t *testing.T) {
wrongToken, _ := mustReconnectToken(t)
wrong := h.dial(t, "6.3.1.4")
wrong.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "PENDING_RELEASE",
PeerID: "G",
ReconnectToken: wrongToken,
@@ -5587,9 +5808,9 @@ func TestTransferHostAdmissionDoesNotReuseConnectionCapability(t *testing.T) {
typ string
}{
{"host create retry", "H", relayTypeCreate},
- {"host join", "H", relayTypeJoin},
- {"target join", "G", relayTypeJoin},
- {"bystander join", "B", relayTypeJoin},
+ {"host resume", "H", relayTypeResume},
+ {"target resume", "G", relayTypeResume},
+ {"bystander resume", "B", relayTypeResume},
} {
for _, disconnected := range []bool{false, true} {
t.Run(fmt.Sprintf("%s/disconnected=%v", admission.name, disconnected), func(t *testing.T) {
@@ -5637,7 +5858,7 @@ func TestTransferHostAdmissionDoesNotReuseConnectionCapability(t *testing.T) {
Type: admission.typ, SessionID: "XFER_REPLACE", PeerID: admission.peerID,
ReconnectToken: token, ProtocolVersion: relayProtocolVersion,
})
- ackType := relayTypeJoined
+ ackType := relayTypeResumed
if admission.typ == relayTypeCreate {
ackType = relayTypeCreated
}
@@ -5645,7 +5866,7 @@ func TestTransferHostAdmissionDoesNotReuseConnectionCapability(t *testing.T) {
t.Fatalf("replacement changed reconnect identity: %+v", ack)
}
for _, observer := range observers {
- if admission.typ == relayTypeJoin || disconnected {
+ if admission.typ == relayTypeResume || disconnected {
if joined := observer.expect(relayTypePeerJoined); joined.PeerID != admission.peerID {
t.Fatalf("wrong replacement identity: %+v", joined)
}
@@ -6030,7 +6251,7 @@ func TestTransferHostPreservesReconnectAuthority(t *testing.T) {
guest.expect(relayTypeHostChanged)
guest.expectEligibility("XFER_RECONNECT", "G", "H")
- // The old host drops and rejoins with its original token as a guest.
+ // The old host resumes with its original token as a guest.
if err := host.conn.Close(); err != nil {
t.Fatalf("close old host: %v", err)
}
@@ -6041,13 +6262,13 @@ func TestTransferHostPreservesReconnectAuthority(t *testing.T) {
guest.expectEligibility("XFER_RECONNECT", "G")
oldHost := h.dial(t, "6.3.4.3")
oldHost.send(currentAdmission(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "XFER_RECONNECT",
PeerID: "H",
ReconnectToken: hostToken,
ProtocolVersion: relayProtocolVersion,
}))
- rejoinedGuest := oldHost.expectAuthority(relayTypeJoined, "G")
+ rejoinedGuest := oldHost.expectAuthority(relayTypeResumed, "G")
if rejoinedGuest.ReconnectToken != hostToken {
t.Fatalf("old host reconnect token changed: %+v", rejoinedGuest)
}
@@ -6055,7 +6276,7 @@ func TestTransferHostPreservesReconnectAuthority(t *testing.T) {
guest.expect(relayTypePeerJoined)
guest.expectEligibility("XFER_RECONNECT", "G", "H")
- // The new host drops and rejoins with its original token as the host.
+ // The new host resumes with its original token as the host.
if err := guest.conn.Close(); err != nil {
t.Fatalf("close new host: %v", err)
}
@@ -6066,13 +6287,13 @@ func TestTransferHostPreservesReconnectAuthority(t *testing.T) {
oldHost.expectEligibility("XFER_RECONNECT", "G")
newHost := h.dial(t, "6.3.4.4")
newHost.send(currentAdmission(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "XFER_RECONNECT",
PeerID: "G",
ReconnectToken: guestToken,
ProtocolVersion: relayProtocolVersion,
}))
- rejoinedHost := newHost.expectAuthority(relayTypeJoined, "G")
+ rejoinedHost := newHost.expectAuthority(relayTypeResumed, "G")
if rejoinedHost.ReconnectToken != guestToken {
t.Fatalf("new host reconnect token changed: %+v", rejoinedHost)
}
@@ -6127,22 +6348,22 @@ func TestTransferHostSurvivesSnapshotRestart(t *testing.T) {
newHost := hB.dial(t, "6.3.5.3")
newHost.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "XFER_RESTART",
PeerID: "G",
ReconnectToken: guestToken,
ProtocolVersion: relayProtocolVersion,
})
- newHost.expectAuthority(relayTypeJoined, "G")
+ newHost.expectAuthority(relayTypeResumed, "G")
oldHost := hB.dial(t, "6.3.5.4")
oldHost.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "XFER_RESTART",
PeerID: "H",
ReconnectToken: hostToken,
ProtocolVersion: relayProtocolVersion,
})
- oldHost.expectAuthority(relayTypeJoined, "G")
+ oldHost.expectAuthority(relayTypeResumed, "G")
newHost.expect(relayTypePeerJoined)
// Persisted identity does not imply capability on these fresh connections.
newHost.send(clientMsg{Type: relayTypeTransferHost, To: "H", ProtocolVersion: relayProtocolVersion})
@@ -8745,7 +8966,7 @@ func TestSnapshotV4RetainsModernHostAndGuestReservationsAcrossRestart(t *testing
wrongHostToken, _ := mustReconnectToken(t)
hostThief := hB.dial(t, "8.0.1.3")
hostThief.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "V3_RESTART",
PeerID: "H",
ReconnectToken: wrongHostToken,
@@ -8755,13 +8976,13 @@ func TestSnapshotV4RetainsModernHostAndGuestReservationsAcrossRestart(t *testing
restartedHost := hB.dial(t, "8.0.1.4")
restartedHost.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "V3_RESTART",
PeerID: "H",
ReconnectToken: hostToken,
ProtocolVersion: relayProtocolVersion,
})
- hostJoined := restartedHost.expectAuthority(relayTypeJoined, "H")
+ hostJoined := restartedHost.expectAuthority(relayTypeResumed, "H")
if hostJoined.ReconnectToken != hostToken || hostJoined.ProtocolVersion != relayProtocolVersion {
t.Fatalf("restored host authority changed: %+v", hostJoined)
}
@@ -8769,7 +8990,7 @@ func TestSnapshotV4RetainsModernHostAndGuestReservationsAcrossRestart(t *testing
wrongGuestToken, _ := mustReconnectToken(t)
guestThief := hB.dial(t, "8.0.1.5")
guestThief.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "V3_RESTART",
PeerID: "G",
ReconnectToken: wrongGuestToken,
@@ -8779,13 +9000,13 @@ func TestSnapshotV4RetainsModernHostAndGuestReservationsAcrossRestart(t *testing
restartedGuest := hB.dial(t, "8.0.1.6")
restartedGuest.send(clientMsg{
- Type: relayTypeJoin,
+ Type: relayTypeResume,
SessionID: "V3_RESTART",
PeerID: "G",
ReconnectToken: guestToken,
ProtocolVersion: relayProtocolVersion,
})
- guestJoined := restartedGuest.expectAuthority(relayTypeJoined, "H")
+ guestJoined := restartedGuest.expectAuthority(relayTypeResumed, "H")
if guestJoined.ReconnectToken != guestToken || guestJoined.ProtocolVersion != relayProtocolVersion {
t.Fatalf("restored guest authority changed: %+v", guestJoined)
}
diff --git a/server/relay_protocol_gen.go b/server/relay_protocol_gen.go
index 645fb55b7..7abbf55bc 100644
--- a/server/relay_protocol_gen.go
+++ b/server/relay_protocol_gen.go
@@ -8,6 +8,7 @@ const (
relayTypeCreate = "create"
relayTypeJoin = "join"
+ relayTypeResume = "resume"
relayTypeBroadcast = "broadcast"
relayTypeSendTo = "sendTo"
relayTypePing = "ping"
@@ -16,6 +17,7 @@ const (
relayTypeTransferHost = "transferHost"
relayTypeCreated = "created"
relayTypeJoined = "joined"
+ relayTypeResumed = "resumed"
relayTypePeerJoined = "peerJoined"
relayTypePeerLeft = "peerLeft"
relayTypeMessage = "message"
@@ -38,6 +40,7 @@ const (
relayErrorPeerNotFound = "peer_not_found"
relayErrorHostTransferUnavailable = "host_transfer_unavailable"
relayFeatureAtomicHostTransfer = "atomicHostTransfer"
+ relayFeatureAuthenticatedResume = "authenticatedResume"
relayCapabilityHostTransfer = "hostTransfer"
maxRoomSize = 8
diff --git a/test/watch_together/watch_together_peer_service_test.dart b/test/watch_together/watch_together_peer_service_test.dart
index 31b83b42f..a2aa9326c 100644
--- a/test/watch_together/watch_together_peer_service_test.dart
+++ b/test/watch_together/watch_together_peer_service_test.dart
@@ -5,7 +5,6 @@ import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
import 'package:stream_channel/stream_channel.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
-import 'package:plezy/i18n/strings.g.dart';
import 'package:plezy/watch_together/services/watch_together_peer_service.dart';
import 'package:plezy/watch_together/services/watch_together_relay_endpoint.dart';
import 'package:plezy/watch_together/models/sync_message.dart';
@@ -311,9 +310,10 @@ void main() {
test('guest reconnect sends its retained capability', () async {
late final _RelayServer relay;
relay = await relayWith((_, socket, message) {
- if (message['type'] == 'join') {
+ if (message['type'] == 'join' || message['type'] == 'resume') {
relay.send(socket, {
- 'type': 'joined',
+ 'type': message['type'] == 'join' ? 'joined' : 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': _relayHostId,
'reconnectToken': message['reconnectToken'],
@@ -338,7 +338,7 @@ void main() {
final initialToken = relay.messages[0].single['reconnectToken'];
expect(relay.messages[1], [
{
- 'type': 'join',
+ 'type': 'resume',
'sessionId': 'GUEST1',
'peerId': guestPeerId,
'reconnectToken': initialToken,
@@ -357,9 +357,10 @@ void main() {
// mirrors it and surfaces the change through the same onHostChanged path.
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) {
- if (message['type'] == 'join') {
+ if (message['type'] == 'join' || message['type'] == 'resume') {
relay.send(socket, {
- 'type': 'joined',
+ 'type': message['type'] == 'join' ? 'joined' : 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': connection == 0 ? _relayHostId : 'new-host',
'reconnectToken': message['reconnectToken'],
@@ -393,10 +394,10 @@ void main() {
expect(service.hostPeerId, 'new-host');
expect(service.isHost, isFalse);
expect(errors, isEmpty);
- expect(relay.messages[1].map((m) => m['type']), ['join'], reason: 'no leave: the admission stands');
+ expect(relay.messages[1].map((m) => m['type']), ['resume'], reason: 'authenticated continuity is required');
});
- test('host reconnect proves ownership and re-creates with the retained authority', () async {
+ test('host membership loss is terminal and never recreates its room', () async {
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) {
if (connection == 0 && message['type'] == 'create') {
@@ -407,81 +408,44 @@ void main() {
'reconnectToken': message['reconnectToken'],
'protocolVersion': 2,
});
- } else if (connection == 1 && message['type'] == 'join') {
+ } else if (message['type'] == 'resume') {
relay.send(socket, {'type': 'error', 'code': 'room_not_found', 'message': 'Room not found'});
- } else if (connection == 1 && message['type'] == 'create') {
- relay.send(socket, {
- 'type': 'created',
- 'sessionId': message['sessionId'],
- 'hostPeerId': message['peerId'],
- 'reconnectToken': message['reconnectToken'],
- 'protocolVersion': 2,
- });
}
});
final service = serviceFor(relay);
- final reconnected = Completer();
var reconnectCallbacks = 0;
- service.onReconnected = () {
- reconnectCallbacks++;
- reconnected.complete();
- };
+ service.onReconnected = () => reconnectCallbacks++;
+ final ended = service.onSessionEnded.first;
await _withShortenedTimer(
original: const Duration(seconds: 2),
replacement: const Duration(milliseconds: 10),
- body: () => service.createSession(sessionId: 'room2'),
+ body: () async {
+ await service.createSession(sessionId: 'room2');
+ await relay.sockets.single.close();
+ await ended.timeout(const Duration(seconds: 1));
+ await service.releaseSession();
+ },
);
- final hostPeerId = service.myPeerId;
- await relay.sockets.single.close();
- await reconnected.future.timeout(const Duration(seconds: 6));
-
- expect(reconnectCallbacks, 1);
- expect(relay.sockets, hasLength(2));
- final initialCreate = relay.messages[0].single;
- final reconnectToken = initialCreate['reconnectToken'];
- expect(initialCreate, {
- 'type': 'create',
- 'sessionId': 'ROOM2',
- 'peerId': hostPeerId,
- 'reconnectToken': matches(RegExp(r'^[A-Za-z0-9_-]{43}$')),
- 'protocolVersion': 2,
- 'syncProtocolVersion': SyncMessage.protocolVersion,
- 'capabilities': [RelayProtocol.hostTransferCapability],
- });
- expect(relay.messages[1], [
- {
- 'type': 'join',
- 'sessionId': 'ROOM2',
- 'peerId': hostPeerId,
- 'reconnectToken': reconnectToken,
- 'protocolVersion': 2,
- 'syncProtocolVersion': SyncMessage.protocolVersion,
- 'capabilities': [RelayProtocol.hostTransferCapability],
- },
- {
- 'type': 'create',
- 'sessionId': 'ROOM2',
- 'peerId': hostPeerId,
- 'reconnectToken': reconnectToken,
- 'protocolVersion': 2,
- 'syncProtocolVersion': SyncMessage.protocolVersion,
- 'capabilities': [RelayProtocol.hostTransferCapability],
- },
+ expect(reconnectCallbacks, 0);
+ expect(service.sessionId, isNull);
+ expect(relay.messages.map((messages) => messages.map((message) => message['type']).toList()), [
+ ['create'],
+ ['resume'],
]);
- expect(service.hostPeerId, hostPeerId);
});
- test('initial create retry reuses its pre-minted identity and capability', () async {
+ test('lost create ACK resumes the pre-minted identity without another create', () async {
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) async {
- if (message['type'] != 'create') return;
+ if (message['type'] != 'create' && message['type'] != 'resume') return;
if (connection == 0) {
await socket.close();
return;
}
relay.send(socket, {
- 'type': 'created',
+ 'type': 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': message['peerId'],
'reconnectToken': message['reconnectToken'],
@@ -500,7 +464,7 @@ void main() {
final first = relay.messages[0].single;
final retry = relay.messages[1].single;
expect(first['reconnectToken'], matches(RegExp(r'^[A-Za-z0-9_-]{43}$')));
- expect(retry, first);
+ expect(retry, {...first, 'type': 'resume'});
expect(service.myPeerId, first['peerId']);
expect(service.hostPeerId, first['peerId']);
});
@@ -517,11 +481,7 @@ void main() {
await expectLater(
_withRetryBackoffShortened(() => timeoutService.createSession(sessionId: 'slow1')),
- throwsA(
- isA()
- .having((error) => error.type, 'type', PeerErrorType.timeout)
- .having((error) => error.message, 'message', t.watchTogether.errors.timedOut),
- ),
+ throwsA(isA().having((error) => error.type, 'type', PeerErrorType.timeout)),
);
late final _RelayServer errorRelay;
@@ -556,8 +516,7 @@ void main() {
throwsA(
isA()
.having((error) => error.type, 'type', PeerErrorType.serverError)
- .having((error) => error.serverCode, 'serverCode', 'protocol_mismatch')
- .having((error) => error.message, 'message', contains('Relay protocol version 2 is required')),
+ .having((error) => error.serverCode, 'serverCode', 'protocol_mismatch'),
),
);
});
@@ -579,11 +538,7 @@ void main() {
await expectLater(
service.createSession(sessionId: 'token1'),
- throwsA(
- isA()
- .having((error) => error.type, 'type', PeerErrorType.serverError)
- .having((error) => error.message, 'message', t.watchTogether.errors.invalidRelayResponse),
- ),
+ throwsA(isA().having((error) => error.type, 'type', PeerErrorType.serverError)),
);
expect(service.hostPeerId, isNull);
});
@@ -597,11 +552,7 @@ void main() {
await expectLater(
oldRelayService.createSession(sessionId: 'old01'),
- throwsA(
- isA()
- .having((error) => error.type, 'type', PeerErrorType.serverError)
- .having((error) => error.message, 'message', t.watchTogether.errors.invalidRelayResponse),
- ),
+ throwsA(isA().having((error) => error.type, 'type', PeerErrorType.serverError)),
);
expect(oldRelayService.hostPeerId, isNull);
@@ -619,11 +570,7 @@ void main() {
await expectLater(
malformedService.joinSession('bad01'),
- throwsA(
- isA()
- .having((error) => error.type, 'type', PeerErrorType.serverError)
- .having((error) => error.message, 'message', t.watchTogether.errors.invalidRelayResponse),
- ),
+ throwsA(isA().having((error) => error.type, 'type', PeerErrorType.serverError)),
);
expect(malformedService.hostPeerId, isNull);
});
@@ -631,9 +578,10 @@ void main() {
test('exhausted create retries end a possibly committed room before clearing credentials', () async {
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) {
- if (connection >= 3 && message['type'] == 'join') {
+ if (connection >= 3 && message['type'] == 'resume') {
relay.send(socket, {
- 'type': 'joined',
+ 'type': 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': message['peerId'],
'reconnectToken': message['reconnectToken'],
@@ -651,18 +599,14 @@ void main() {
await expectLater(
_withRetryBackoffShortened(() => service.createSession(sessionId: 'lostc')),
- throwsA(
- isA()
- .having((error) => error.type, 'type', PeerErrorType.timeout)
- .having((error) => error.message, 'message', t.watchTogether.errors.timedOut),
- ),
+ throwsA(isA().having((error) => error.type, 'type', PeerErrorType.timeout)),
);
expect(relay.messages.map((messages) => messages.map((message) => message['type']).toList()).toList(), [
['create'],
- ['create'],
- ['create'],
- ['join', 'endSession'],
+ ['resume'],
+ ['resume'],
+ ['resume', 'endSession'],
]);
final announcements = relay.messages.map((messages) => messages.first).toList();
expect(announcements.map((message) => message['peerId']).toSet(), hasLength(1));
@@ -674,9 +618,10 @@ void main() {
test('exhausted join retries leave a possibly committed guest reservation before clearing credentials', () async {
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) {
- if (connection >= 3 && message['type'] == 'join') {
+ if (connection >= 3 && message['type'] == 'resume') {
relay.send(socket, {
- 'type': 'joined',
+ 'type': 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': _relayHostId,
'reconnectToken': message['reconnectToken'],
@@ -696,18 +641,14 @@ void main() {
await expectLater(
_withRetryBackoffShortened(() => service.joinSession('lostj')),
- throwsA(
- isA()
- .having((error) => error.type, 'type', PeerErrorType.timeout)
- .having((error) => error.message, 'message', isNotEmpty),
- ),
+ throwsA(isA().having((error) => error.type, 'type', PeerErrorType.timeout)),
);
expect(relay.messages.map((messages) => messages.map((message) => message['type']).toList()).toList(), [
['join'],
- ['join'],
- ['join'],
- ['join', 'leave'],
+ ['resume'],
+ ['resume'],
+ ['resume', 'leave'],
]);
final announcements = relay.messages.map((messages) => messages.first).toList();
expect(announcements.map((message) => message['peerId']).toSet(), hasLength(1));
@@ -728,11 +669,7 @@ void main() {
await expectLater(
service.joinSession('ended2'),
- throwsA(
- isA()
- .having((error) => error.type, 'type', PeerErrorType.invalidSession)
- .having((error) => error.message, 'message', t.watchTogether.errors.sessionEnded),
- ),
+ throwsA(isA().having((error) => error.type, 'type', PeerErrorType.invalidSession)),
);
await ended.timeout(const Duration(seconds: 1));
@@ -744,7 +681,7 @@ void main() {
test('guest reconnect treats ended as terminal and cancels further reconnects', () async {
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) {
- if (message['type'] != 'join') return;
+ if (message['type'] != 'join' && message['type'] != 'resume') return;
if (connection == 0) {
relay.send(socket, {
'type': 'joined',
@@ -779,10 +716,10 @@ void main() {
expect(relay.sockets, hasLength(2));
});
- test('guest reconnect converges to session ended after room-not-found retries are exhausted', () async {
+ test('guest membership loss is terminal without retrying initial admission', () async {
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) {
- if (message['type'] != 'join') return;
+ if (message['type'] != 'join' && message['type'] != 'resume') return;
if (connection == 0) {
relay.send(socket, {
'type': 'joined',
@@ -828,10 +765,10 @@ void main() {
expect(reconnectCallbacks, 0);
expect(sessionEndedEvents, 1);
expect(service.isConnected, isFalse);
- expect(relay.sockets, hasLength(4));
+ expect(relay.sockets, hasLength(2));
expect(
relay.messages.skip(1).map((messages) => messages.map((message) => message['type']).toList()),
- everyElement(['join']),
+ everyElement(['resume']),
);
});
@@ -916,7 +853,7 @@ void main() {
'protocolVersion': 2,
'peers': [_relayHostId],
});
- } else if (connection == 1 && message['type'] == 'join') {
+ } else if (connection == 1 && message['type'] == 'resume') {
relay.send(socket, {
'type': 'error',
'code': 'peer_id_unavailable',
@@ -931,7 +868,7 @@ void main() {
expect(relay.messages, hasLength(2));
expect(relay.messages[0].map((message) => message['type']), ['join', 'leave']);
- expect(relay.messages[1].map((message) => message['type']), ['join']);
+ expect(relay.messages[1].map((message) => message['type']), ['resume']);
});
test('a guest promoted mid-teardown ends the room instead of reading the rejected leave as released', () async {
@@ -944,8 +881,10 @@ void main() {
relay = await relayWith((connection, socket, message) {
switch (message['type']) {
case 'join':
+ case 'resume':
relay.send(socket, {
- 'type': 'joined',
+ 'type': message['type'] == 'join' ? 'joined' : 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': connection == 0 ? _relayHostId : message['peerId'],
'reconnectToken': message['reconnectToken'],
@@ -970,7 +909,7 @@ void main() {
expect(relay.messages, hasLength(2));
expect(relay.messages[0].map((message) => message['type']), ['join', 'leave']);
- expect(relay.messages[1].map((message) => message['type']), ['join', 'endSession']);
+ expect(relay.messages[1].map((message) => message['type']), ['resume', 'endSession']);
});
test('a host demoted mid-teardown leaves as the guest it became instead of failing the exit', () async {
@@ -987,9 +926,10 @@ void main() {
});
case 'endSession':
relay.send(socket, {'type': 'error', 'code': 'peer_id_unavailable', 'message': 'Unable to end room'});
- case 'join':
+ case 'resume':
relay.send(socket, {
- 'type': 'joined',
+ 'type': 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': 'new-host',
'reconnectToken': message['reconnectToken'],
@@ -1016,7 +956,7 @@ void main() {
expect(relay.messages, hasLength(2));
expect(relay.messages[0].map((message) => message['type']), ['create', 'endSession']);
- expect(relay.messages[1].map((message) => message['type']), ['join', 'leave']);
+ expect(relay.messages[1].map((message) => message['type']), ['resume', 'leave']);
});
test('a release the relay keeps refusing for an unchanged role stays bounded and surfaces', () async {
@@ -1024,8 +964,10 @@ void main() {
relay = await relayWith((connection, socket, message) {
switch (message['type']) {
case 'join':
+ case 'resume':
relay.send(socket, {
- 'type': 'joined',
+ 'type': message['type'] == 'join' ? 'joined' : 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': _relayHostId,
'reconnectToken': message['reconnectToken'],
@@ -1045,8 +987,8 @@ void main() {
);
expect(relay.messages, hasLength(3));
- for (final connection in relay.messages) {
- expect(connection.map((message) => message['type']), ['join', 'leave']);
+ for (var i = 0; i < relay.messages.length; i++) {
+ expect(relay.messages[i].map((message) => message['type']), [i == 0 ? 'join' : 'resume', 'leave']);
}
});
@@ -1089,7 +1031,7 @@ void main() {
]);
});
- test('sequential guest release accepts not-in-room as idempotent success', () async {
+ test('a concurrent host end completes guest release without requesting membership again', () async {
var leaveRequests = 0;
late final _RelayServer relay;
relay = await relayWith((_, socket, message) {
@@ -1104,16 +1046,12 @@ void main() {
});
} else if (message['type'] == 'leave') {
leaveRequests++;
- if (leaveRequests == 1) {
- relay.send(socket, {
- 'type': 'left',
- 'sessionId': message['sessionId'],
- 'peerId': message['peerId'],
- 'protocolVersion': 2,
- });
- } else {
- relay.send(socket, {'type': 'error', 'code': 'not_in_room', 'message': 'Peer is not in the room'});
- }
+ relay.send(socket, {
+ 'type': 'ended',
+ 'sessionId': message['sessionId'],
+ 'peerId': message['peerId'],
+ 'protocolVersion': 2,
+ });
}
});
final service = serviceFor(relay);
@@ -1122,8 +1060,8 @@ void main() {
await service.releaseSession();
await service.releaseSession();
- expect(leaveRequests, 2);
- expect(relay.messages.single.map((message) => message['type']), ['join', 'leave', 'leave']);
+ expect(leaveRequests, 1);
+ expect(relay.messages.single.map((message) => message['type']), ['join', 'leave']);
});
test('host end waits for a protocol-2 ended acknowledgement', () async {
@@ -1264,10 +1202,11 @@ void main() {
final service = serviceFor(relay);
final pending = service.createSession(sessionId: 'cancel1');
+ final cancelled = expectLater(pending, throwsStateError);
await announcementSeen.future.timeout(const Duration(seconds: 1));
await service.disconnect();
- await expectLater(pending, throwsStateError);
+ await cancelled.timeout(const Duration(seconds: 1));
expect(service.sessionId, isNull);
expect(service.connectedPeers, isEmpty);
});
@@ -1285,9 +1224,10 @@ void main() {
'reconnectToken': message['reconnectToken'],
'protocolVersion': 2,
});
- } else if (connection == 1 && message['type'] == 'join') {
+ } else if (connection == 1 && message['type'] == 'resume') {
relay.send(socket, {
- 'type': 'joined',
+ 'type': 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': message['peerId'],
'reconnectToken': message['reconnectToken'],
@@ -1460,14 +1400,17 @@ void main() {
test('reconnection must reestablish enforcing-feature and roster authorization', () async {
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) {
- if (message['type'] == 'create' || message['type'] == 'join') {
+ if (message['type'] == 'create' || message['type'] == 'resume') {
relay.send(socket, {
- 'type': message['type'] == 'create' ? 'created' : 'joined',
+ 'type': message['type'] == 'create' ? 'created' : 'resumed',
'sessionId': message['sessionId'],
'hostPeerId': message['peerId'],
'reconnectToken': message['reconnectToken'],
'protocolVersion': 2,
- if (connection != 1) 'features': [RelayProtocol.atomicHostTransferFeature],
+ 'features': [
+ RelayProtocol.authenticatedResumeFeature,
+ if (connection != 1) RelayProtocol.atomicHostTransferFeature,
+ ],
});
relay.send(socket, {
'type': 'hostTransferEligibility',
@@ -1489,7 +1432,7 @@ void main() {
service.onReconnected = () => reconnected.complete();
await relay.sockets.single.close();
await reconnected.future.timeout(const Duration(seconds: 5));
- expect(service.canTransferHostTo('guest-1'), isFalse, reason: 'old relay cannot inherit prior ACK');
+ expect(service.canTransferHostTo('guest-1'), isFalse, reason: 'resume cannot inherit prior transfer ACK');
final restored = service.onHostTransferEligibilityChanged.firstWhere(
(_) => service.canTransferHostTo('guest-1'),
);
@@ -1576,9 +1519,10 @@ void main() {
test('guest reconnect accepts the host identity pinned by a transfer', () async {
late final _RelayServer relay;
relay = await relayWith((connection, socket, message) {
- if (message['type'] == 'join') {
+ if (message['type'] == 'join' || message['type'] == 'resume') {
relay.send(socket, {
- 'type': 'joined',
+ 'type': message['type'] == 'join' ? 'joined' : 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': connection == 0 ? _relayHostId : 'guest-2',
'reconnectToken': message['reconnectToken'],
@@ -1630,9 +1574,10 @@ void main() {
'reconnectToken': message['reconnectToken'],
'protocolVersion': 2,
});
- } else if (connection >= 1 && message['type'] == 'join') {
+ } else if (connection >= 1 && message['type'] == 'resume') {
relay.send(socket, {
- 'type': 'joined',
+ 'type': 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
'hostPeerId': 'new-host',
'reconnectToken': message['reconnectToken'],
@@ -1672,8 +1617,10 @@ void main() {
relay = await relayWith((connection, socket, message) {
switch (message['type']) {
case 'join':
+ case 'resume':
relay.send(socket, {
- 'type': 'joined',
+ 'type': message['type'] == 'join' ? 'joined' : 'resumed',
+ 'features': [RelayProtocol.authenticatedResumeFeature],
'sessionId': message['sessionId'],
// The reconnect: the relay made this peer the host while it was
// offline, so the hostChanged broadcast never reached it.
@@ -1721,6 +1668,353 @@ void main() {
// 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']);
+ expect(relay.messages[1].map((message) => message['type']), ['resume', 'endSession']);
+ });
+
+ test('lost initial join ACK resumes a promotion before setup returns', () async {
+ late final _RelayServer relay;
+ relay = await relayWith((_, socket, message) async {
+ if (message['type'] == 'join') {
+ await socket.close();
+ } else if (message['type'] == 'resume') {
+ relay.send(socket, {
+ 'type': 'resumed',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': message['peerId'],
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ 'features': [RelayProtocol.authenticatedResumeFeature],
+ });
+ } else if (message['type'] == 'endSession') {
+ relay.send(socket, {'type': 'ended', 'sessionId': message['sessionId'], 'protocolVersion': 2});
+ }
+ });
+ final service = serviceFor(relay);
+ await _withRetryBackoffShortened(() => service.joinSession('promotedSetup'));
+ expect(service.isHost, isTrue);
+ expect(service.hostPeerId, service.myPeerId);
+ await service.releaseSession();
+ expect(relay.messages.map((messages) => messages.map((message) => message['type']).toList()), [
+ ['join'],
+ ['resume', 'endSession'],
+ ]);
+ });
+
+ test('uncommitted ambiguous setup fails closed and only explicit re-entry mints membership', () async {
+ late final _RelayServer relay;
+ relay = await relayWith((connection, socket, message) async {
+ if (connection == 0 && message['type'] == 'create') {
+ await socket.close();
+ } else if (message['type'] == 'resume') {
+ relay.send(socket, {'type': 'error', 'code': 'room_not_found', 'message': 'Room not found'});
+ } else if (message['type'] == 'create') {
+ relay.send(socket, {
+ 'type': 'created',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': message['peerId'],
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ });
+ }
+ });
+ final service = serviceFor(relay);
+ await expectLater(
+ _withRetryBackoffShortened(() => service.createSession(sessionId: 'ambiguous')),
+ throwsA(isA().having((error) => error.serverCode, 'serverCode', 'room_not_found')),
+ );
+ expect(service.sessionId, isNull);
+ expect(relay.messages.map((messages) => messages.single['type']), ['create', 'resume']);
+ final original = relay.messages.first.single;
+ await service.createSession(sessionId: 'ambiguous');
+ final explicitEntry = relay.messages.last.single;
+ expect(explicitEntry['type'], 'create');
+ expect(explicitEntry['peerId'], isNot(original['peerId']));
+ expect(explicitEntry['reconnectToken'], isNot(original['reconnectToken']));
+ });
+
+ for (final invalid in ['joinedTokenEcho', 'missingFeature', 'malformedPeers', 'wrongSession']) {
+ test('resume rejects $invalid before publishing authority or room traffic', () async {
+ late final _RelayServer relay;
+ relay = await relayWith((_, socket, message) {
+ if (message['type'] == 'join') {
+ relay.send(socket, {
+ 'type': 'joined',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': _relayHostId,
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ 'peers': [_relayHostId],
+ });
+ } else if (message['type'] == 'resume') {
+ relay.send(socket, {'type': 'hostChanged', 'sessionId': message['sessionId'], 'hostPeerId': 'replacement'});
+ relay.send(socket, {
+ 'type': invalid == 'joinedTokenEcho' ? 'joined' : 'resumed',
+ 'sessionId': invalid == 'wrongSession' ? 'OTHER_ROOM' : message['sessionId'],
+ 'hostPeerId': 'replacement',
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ if (invalid != 'missingFeature') 'features': [RelayProtocol.authenticatedResumeFeature],
+ 'peers': invalid == 'malformedPeers' ? ['replacement', 7] : ['replacement'],
+ });
+ relay.send(socket, {'type': 'peerJoined', 'peerId': 'replacement'});
+ relay.send(socket, {
+ 'type': 'message',
+ 'from': 'replacement',
+ 'payload': SyncMessage.requestState().toJson(),
+ });
+ }
+ });
+ final service = serviceFor(relay);
+ final authorities = [];
+ final connected = [];
+ final peers = [];
+ final payloads = [];
+ var reconnectCallbacks = 0;
+ service.onReconnected = () => reconnectCallbacks++;
+ addTearDown(service.onHostChanged.listen(authorities.add).cancel);
+ addTearDown(service.onConnectionStateChanged.listen(connected.add).cancel);
+ addTearDown(service.onPeerConnected.listen(peers.add).cancel);
+ addTearDown(service.onMessageReceived.listen(payloads.add).cancel);
+ final ended = service.onSessionEnded.first;
+ await _withShortenedTimer(
+ original: const Duration(seconds: 2),
+ replacement: const Duration(milliseconds: 10),
+ body: () async {
+ await service.joinSession('validation');
+ await relay.sockets.single.close();
+ await ended.timeout(const Duration(seconds: 1));
+ },
+ );
+ expect(authorities, isEmpty);
+ expect(peers, [_relayHostId]);
+ expect(connected.where((value) => value), [true]);
+ expect(payloads, isEmpty);
+ expect(reconnectCallbacks, 0);
+ expect(service.hostPeerId, isNull);
+ expect(service.sessionId, isNull);
+ expect(relay.messages.map((messages) => messages.single['type']), ['join', 'resume']);
+ });
+ }
+
+ for (final releaseInstead in [false, true]) {
+ test(
+ 'released relay compatibility fails closed on ${releaseInstead ? 'disconnected release' : 'resume'}',
+ () async {
+ late final _RelayServer relay;
+ relay = await relayWith((_, socket, message) {
+ if (message['type'] == 'join') {
+ // Released protocol-2 relays support initial admission without features.
+ relay.send(socket, {
+ 'type': 'joined',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': _relayHostId,
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ 'peers': [_relayHostId],
+ });
+ } else {
+ relay.send(socket, {'type': 'error', 'code': 'invalid_message', 'message': 'Unknown message type'});
+ }
+ });
+ final service = serviceFor(relay);
+ var reconnectCallbacks = 0;
+ service.onReconnected = () => reconnectCallbacks++;
+ await _withShortenedTimer(
+ original: const Duration(seconds: 2),
+ replacement: releaseInstead ? const Duration(seconds: 2) : const Duration(milliseconds: 10),
+ body: () async {
+ await service.joinSession('oldRelay');
+ final disconnected = service.onConnectionStateChanged.firstWhere((connected) => !connected);
+ final ended = releaseInstead ? null : service.onSessionEnded.first;
+ await relay.sockets.single.close();
+ await disconnected.timeout(const Duration(seconds: 1));
+ if (releaseInstead) {
+ await expectLater(
+ service.releaseSession(),
+ throwsA(isA().having((error) => error.serverCode, 'serverCode', 'invalid_message')),
+ );
+ } else {
+ await ended!.timeout(const Duration(seconds: 1));
+ expect(service.sessionId, isNull);
+ }
+ },
+ );
+ expect(reconnectCallbacks, 0);
+ expect(relay.messages.map((messages) => messages.single['type']), ['join', 'resume']);
+ },
+ );
+ }
+
+ test('lost host end ACK cannot recreate or end a replacement room during cleanup', () async {
+ late final _RelayServer relay;
+ relay = await relayWith((_, socket, message) {
+ if (message['type'] == 'create') {
+ relay.send(socket, {
+ 'type': 'created',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': message['peerId'],
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ });
+ } else if (message['type'] == 'resume') {
+ // End committed and this reusable code now names unrelated membership.
+ relay.send(socket, {'type': 'error', 'code': 'peer_id_unavailable', 'message': 'Peer ID is unavailable'});
+ }
+ });
+ final service = serviceFor(relay, debugReleaseTimeout: const Duration(milliseconds: 10));
+ await service.createSession(sessionId: 'reusedCode');
+ await _withRetryBackoffShortened(service.releaseSession);
+ await service.releaseSession();
+ expect(relay.messages.map((messages) => messages.map((message) => message['type']).toList()), [
+ ['create', 'endSession'],
+ ['resume'],
+ ]);
+ });
+
+ test('transport failure before sending admission may retry the initial operation', () async {
+ late final _RelayServer relay;
+ relay = await relayWith((_, socket, message) {
+ if (message['type'] == 'create') {
+ relay.send(socket, {
+ 'type': 'created',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': message['peerId'],
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ });
+ }
+ });
+ var attempts = 0;
+ final service = serviceFor(
+ relay,
+ debugChannelFactory: (uri) {
+ if (attempts++ == 0) throw const SocketException('Connection refused before admission');
+ return WebSocketChannel.connect(uri);
+ },
+ );
+ await _withRetryBackoffShortened(() => service.createSession(sessionId: 'preSend'));
+ expect(service.isHost, isTrue);
+ expect(relay.messages.single.single['type'], 'create');
+ });
+
+ test('network failures exhaust bounded resume attempts without falling back to create', () async {
+ late final _RelayServer relay;
+ relay = await relayWith((_, socket, message) async {
+ if (message['type'] == 'create') {
+ relay.send(socket, {
+ 'type': 'created',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': message['peerId'],
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ });
+ } else if (message['type'] == 'resume') {
+ await socket.close();
+ }
+ });
+ final service = serviceFor(relay);
+ var reconnectCallbacks = 0;
+ service.onReconnected = () => reconnectCallbacks++;
+ final ended = service.onSessionEnded.first;
+ await _withShortenedTimer(
+ original: const Duration(seconds: 2),
+ replacement: const Duration(milliseconds: 10),
+ body: () => _withShortenedTimer(
+ original: const Duration(seconds: 4),
+ replacement: const Duration(milliseconds: 10),
+ body: () => _withShortenedTimer(
+ original: const Duration(seconds: 6),
+ replacement: const Duration(milliseconds: 10),
+ body: () async {
+ await service.createSession(sessionId: 'networkLoss');
+ await relay.sockets.single.close();
+ },
+ ),
+ ),
+ );
+ // Only reconnect backoff belongs to the shortened timer zone.
+ await ended.timeout(const Duration(seconds: 2));
+ expect(reconnectCallbacks, 0);
+ expect(service.sessionId, isNull);
+ expect(relay.messages.map((messages) => messages.single['type']), ['create', 'resume', 'resume', 'resume']);
+ });
+
+ test('host release does not treat a guest left ACK as room destruction', () async {
+ late final _RelayServer relay;
+ relay = await relayWith((_, socket, message) {
+ if (message['type'] == 'create') {
+ relay.send(socket, {
+ 'type': 'created',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': message['peerId'],
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ });
+ } else if (message['type'] == 'endSession') {
+ relay.send(socket, {
+ 'type': 'left',
+ 'sessionId': message['sessionId'],
+ 'peerId': message['peerId'],
+ 'protocolVersion': 2,
+ });
+ }
+ });
+ final service = serviceFor(relay);
+ await service.createSession(sessionId: 'wrongReleaseAck');
+ await expectLater(
+ service.releaseSession(),
+ throwsA(isA().having((error) => error.type, 'type', PeerErrorType.serverError)),
+ );
+ expect(relay.messages.single.map((message) => message['type']), ['create', 'endSession']);
+ });
+
+ test('cancelling pending resume cannot publish or release a later explicit session', () async {
+ final resumeSeen = Completer();
+ late final _RelayServer relay;
+ relay = await relayWith((_, socket, message) {
+ if (message['type'] == 'join') {
+ relay.send(socket, {
+ 'type': 'joined',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': _relayHostId,
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ 'peers': [_relayHostId],
+ });
+ } else if (message['type'] == 'resume') {
+ resumeSeen.complete();
+ } else if (message['type'] == 'create') {
+ relay.send(socket, {
+ 'type': 'created',
+ 'sessionId': message['sessionId'],
+ 'hostPeerId': message['peerId'],
+ 'reconnectToken': message['reconnectToken'],
+ 'protocolVersion': 2,
+ });
+ }
+ });
+ final service = serviceFor(relay);
+ var reconnectCallbacks = 0;
+ service.onReconnected = () => reconnectCallbacks++;
+ await _withShortenedTimer(
+ original: const Duration(seconds: 2),
+ replacement: const Duration(milliseconds: 10),
+ body: () async {
+ await service.joinSession('cancelOld');
+ await relay.sockets.single.close();
+ await resumeSeen.future.timeout(const Duration(seconds: 1));
+ await service.disconnect();
+ await service.createSession(sessionId: 'explicitNew');
+ },
+ );
+ expect(service.sessionId, 'EXPLICITNEW');
+ expect(service.hostPeerId, service.myPeerId);
+ expect(service.isHost, isTrue);
+ expect(reconnectCallbacks, 0);
+ expect(relay.messages.map((messages) => messages.map((message) => message['type']).toList()), [
+ ['join'],
+ ['resume'],
+ ['create'],
+ ]);
});
}