diff --git a/lib/watch_together/services/watch_together_peer_service.dart b/lib/watch_together/services/watch_together_peer_service.dart index 986e40077..e6b00ce0f 100644 --- a/lib/watch_together/services/watch_together_peer_service.dart +++ b/lib/watch_together/services/watch_together_peer_service.dart @@ -36,9 +36,29 @@ class WatchTogetherPeerService with KeepaliveMixin { /// reconnect is published to consumers. final Future Function()? debugReconnectSetupSucceededBarrier; + /// How long initial setup waits for the relay to acknowledge its + /// announcement. + /// + /// This and the two release budgets below are separate knobs so a test can + /// expire exactly the phase it is about. Compressing them together would + /// also put a real WebSocket handshake on a deadline shorter than a loopback + /// round trip, which is a race, not a contract. + final Duration debugInitialSetupTimeout; + + /// How long a release waits for the replacement WebSocket handshake it needs + /// when transport was already lost. + final Duration debugReleaseConnectTimeout; + + /// How long a release waits for the relay to acknowledge its reconnect, + /// endSession, or leave announcement. + final Duration debugReleaseTimeout; + WatchTogetherPeerService({ WatchTogetherRelayEndpoint? endpoint, this.debugReconnectSetupSucceededBarrier, + this.debugInitialSetupTimeout = const Duration(seconds: 10), + this.debugReleaseConnectTimeout = const Duration(seconds: 10), + this.debugReleaseTimeout = const Duration(seconds: 10), WebSocketChannel Function(Uri uri)? debugChannelFactory, }) : endpoint = endpoint ?? WatchTogetherRelayEndpoint.defaultEndpoint, _channelFactory = debugChannelFactory ?? ((uri) => WebSocketChannel.connect(uri)); @@ -635,7 +655,7 @@ class WatchTogetherPeerService with KeepaliveMixin { for (var attempt = 0; attempt < _maxReconnectAttempts; attempt++) { try { final completer = await _connectAndAnnounce(type, epoch); - await completer.future.timeout(const Duration(seconds: 10), onTimeout: () => throw timeoutError); + await completer.future.timeout(debugInitialSetupTimeout, onTimeout: () => throw timeoutError); return; } catch (error) { if (_disposed || epoch != _connectionEpoch) rethrow; @@ -784,18 +804,18 @@ class WatchTogetherPeerService with KeepaliveMixin { final reconnectCompleter = await _connectAndAnnounce( RelayProtocol.join, epoch, - connectTimeout: const Duration(seconds: 10), + connectTimeout: debugReleaseConnectTimeout, connectOperation: 'WatchTogether release reconnect', ); await reconnectCompleter.future.namedTimeout( - const Duration(seconds: 10), + debugReleaseTimeout, operation: 'WatchTogether release reconnect', ); } final releaseCompleter = _announce(_isHost ? RelayProtocol.endSession : RelayProtocol.leave); await releaseCompleter.future.namedTimeout( - const Duration(seconds: 10), + debugReleaseTimeout, operation: _isHost ? 'WatchTogether end session' : 'WatchTogether leave session', ); return; diff --git a/test/watch_together/watch_together_peer_service_test.dart b/test/watch_together/watch_together_peer_service_test.dart index b8fb75fe0..50ce2b01a 100644 --- a/test/watch_together/watch_together_peer_service_test.dart +++ b/test/watch_together/watch_together_peer_service_test.dart @@ -28,18 +28,20 @@ Future _withShortenedTimer({ ); } -Future _withSetupTimersShortened(Future Function() body) { +/// Collapses the initial-setup and release retry backoff so a test that +/// exhausts the retries does not spend the real 250 ms + 500 ms between them. +/// +/// Every relay wait a test needs to *expire* is set through the service's own +/// budgets instead, so nothing here puts a real loopback handshake on a +/// deadline shorter than the round trip it is waiting for. +Future _withRetryBackoffShortened(Future Function() body) { return _withShortenedTimer( - original: const Duration(seconds: 10), - replacement: const Duration(milliseconds: 10), + original: const Duration(milliseconds: 250), + replacement: const Duration(milliseconds: 1), body: () => _withShortenedTimer( - original: const Duration(milliseconds: 250), + original: const Duration(milliseconds: 500), replacement: const Duration(milliseconds: 1), - body: () => _withShortenedTimer( - original: const Duration(milliseconds: 500), - replacement: const Duration(milliseconds: 1), - body: body, - ), + body: body, ), ); } @@ -136,11 +138,17 @@ void main() { WatchTogetherPeerService serviceFor( _RelayServer relay, { Future Function()? debugReconnectSetupSucceededBarrier, + Duration debugInitialSetupTimeout = const Duration(seconds: 10), + Duration debugReleaseConnectTimeout = const Duration(seconds: 10), + Duration debugReleaseTimeout = const Duration(seconds: 10), WebSocketChannel Function(Uri uri)? debugChannelFactory, }) { final service = WatchTogetherPeerService( endpoint: WatchTogetherRelayEndpoint.resolve(relay.baseUrl), debugReconnectSetupSucceededBarrier: debugReconnectSetupSucceededBarrier, + debugInitialSetupTimeout: debugInitialSetupTimeout, + debugReleaseConnectTimeout: debugReleaseConnectTimeout, + debugReleaseTimeout: debugReleaseTimeout, debugChannelFactory: debugChannelFactory, ); services.add(service); @@ -529,10 +537,16 @@ void main() { test('setup preserves typed timeout and relay errors', () async { final timeoutRelay = await relayWith((_, _, _) {}); - final timeoutService = serviceFor(timeoutRelay); + // This relay answers nothing, so the setup announcement and the + // best-effort release that follows both have to give up on their own. + final timeoutService = serviceFor( + timeoutRelay, + debugInitialSetupTimeout: const Duration(milliseconds: 10), + debugReleaseTimeout: const Duration(milliseconds: 10), + ); await expectLater( - _withSetupTimersShortened(() => timeoutService.createSession(sessionId: 'slow1')), + _withRetryBackoffShortened(() => timeoutService.createSession(sessionId: 'slow1')), throwsA( isA() .having((error) => error.type, 'type', PeerErrorType.timeout) @@ -660,10 +674,13 @@ void main() { relay.send(socket, {'type': 'ended', 'sessionId': message['sessionId'], 'protocolVersion': 2}); } }); - final service = serviceFor(relay); + // Only the setup acknowledgement is compressed: the recovery that follows + // keeps its real budgets, so neither its handshake nor its acknowledgements + // are racing a deadline shorter than a loopback round trip. + final service = serviceFor(relay, debugInitialSetupTimeout: const Duration(milliseconds: 10)); await expectLater( - _withSetupTimersShortened(() => service.createSession(sessionId: 'lostc')), + _withRetryBackoffShortened(() => service.createSession(sessionId: 'lostc')), throwsA( isA() .having((error) => error.type, 'type', PeerErrorType.timeout) @@ -705,10 +722,10 @@ void main() { }); } }); - final service = serviceFor(relay); + final service = serviceFor(relay, debugInitialSetupTimeout: const Duration(milliseconds: 10)); await expectLater( - _withSetupTimersShortened(() => service.joinSession('lostj')), + _withRetryBackoffShortened(() => service.joinSession('lostj')), throwsA( isA() .having((error) => error.type, 'type', PeerErrorType.timeout) @@ -896,6 +913,7 @@ void main() { }); final service = serviceFor( relay, + debugReleaseConnectTimeout: const Duration(milliseconds: 10), debugChannelFactory: (uri) { if (channelCalls++ == 0) return WebSocketChannel.connect(uri); final channel = _PendingWebSocketChannel(); @@ -909,7 +927,7 @@ void main() { await relay.sockets.single.close(); await disconnected.timeout(const Duration(seconds: 1)); - await expectLater(_withSetupTimersShortened(service.releaseSession), throwsA(isA())); + await expectLater(_withRetryBackoffShortened(service.releaseSession), throwsA(isA())); expect(pendingChannels, hasLength(3)); expect(pendingChannels.every((channel) => channel.sink.closed), isTrue); @@ -936,10 +954,10 @@ void main() { }); } }); - final service = serviceFor(relay); + final service = serviceFor(relay, debugReleaseTimeout: const Duration(milliseconds: 10)); await service.joinSession('lostl'); - await _withSetupTimersShortened(service.releaseSession); + await _withRetryBackoffShortened(service.releaseSession); expect(relay.messages, hasLength(2)); expect(relay.messages[0].map((message) => message['type']), ['join', 'leave']);