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

test(watch-together): stop the relay setup tests racing a real handshake

Browse Source 
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.
This commit is contained in:
edde746
2026-08-27 09:21:48 +02:00
parent f7060478d1
commit a18ecb5c4a
2 changed files with 60 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/watch_together/services/watch_together_peer_service.dart
+24 -4
View File
@@ -36,9 +36,29 @@ class WatchTogetherPeerService with KeepaliveMixin {
/// reconnect is published to consumers.
final Future<void> 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;
test/watch_together/watch_together_peer_service_test.dart
+36 -18
View File
@@ -28,18 +28,20 @@ Future<T> _withShortenedTimer<T>({
);
}
Future<T> _withSetupTimersShortened<T>(Future<T> 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<T> _withRetryBackoffShortened<T>(Future<T> 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<void> 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<PeerError>()
.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<PeerError>()
.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<PeerError>()
.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<TimeoutException>()));
await expectLater(_withRetryBackoffShortened(service.releaseSession), throwsA(isA<TimeoutException>()));
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']);