From a18ecb5c4aee65705b39daef893e97512018a379 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Thu, 27 Aug 2026 02:41:20 +0200 Subject: [PATCH] test(watch-together): stop the relay setup tests racing a real handshake Two relay setup tests failed on CI while the same commit had passed a few minutes earlier. The harness rewrote every 10-second timer in the zone to 10 ms, which also compressed the connect deadline of the recovery that runs once setup retries are exhausted. On a loaded runner that deadline expired after the relay had upgraded the socket but before the client sent its first frame, so the recorded connection carried no messages and shifted the expected sequence; when all three recovery attempts were starved this way, the recovery never announced at all. A reply that never comes is safe to compress, a real loopback handshake is not, and one zone hook matching on duration cannot tell them apart. The service now exposes the setup acknowledgement, the release handshake, and the release acknowledgement as separate budgets, each still 10 seconds in production, and every test sets only the one it means to expire. The blunt helper is gone; what remains collapses the 250 ms and 500 ms retry delays, which are waits rather than races. Verified by reproducing both failure modes under CPU saturation, about one run in ten, then 30 consecutive clean runs of the file under the same load and a full suite at 6332 passing. --- .../services/watch_together_peer_service.dart | 28 ++++++++-- .../watch_together_peer_service_test.dart | 54 ++++++++++++------- 2 files changed, 60 insertions(+), 22 deletions(-) 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']);