From d371bfed3a1b154e945fd235faa5ce7054ec0ffd Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Thu, 3 Sep 2026 07:35:34 +0200 Subject: [PATCH] fix(networking): connect in resolver address order with a Happy Eyeballs race On a dual-stack network Plezy pinned API traffic and the notification websocket to IPv4 while the native Plex apps and the media stream used IPv6. dart:io looks up A before AAAA and connects in arrival order, so the AAAA address was never tried when the IPv4 path answered within 250 ms. Install an HttpClient.connectionFactory that does one lookup (keeping the resolver's RFC 6724 order), interleaves families per RFC 8305 and races candidates 250 ms apart. The task is returned before the lookup starts so connectionTimeout and cancel cover resolution. TLS is done in the factory since the SDK skips it once one is installed; proxied requests stay plain. The library-event and companion-remote websockets get the same client because WebSocket.connect otherwise builds its own. Supersedes #2233. --- .../companion_remote_peer_service.dart | 13 +- .../library_events/library_event_socket.dart | 2 + lib/utils/happy_eyeballs.dart | 166 ++++++++++++ lib/utils/platform_http_client_io.dart | 12 +- test/utils/happy_eyeballs_test.dart | 250 ++++++++++++++++++ 5 files changed, 435 insertions(+), 8 deletions(-) create mode 100644 lib/utils/happy_eyeballs.dart create mode 100644 test/utils/happy_eyeballs_test.dart diff --git a/lib/services/companion_remote/companion_remote_peer_service.dart b/lib/services/companion_remote/companion_remote_peer_service.dart index c5eb1f2aa..eb6358065 100644 --- a/lib/services/companion_remote/companion_remote_peer_service.dart +++ b/lib/services/companion_remote/companion_remote_peer_service.dart @@ -10,6 +10,7 @@ import '../../i18n/strings.g.dart'; import '../../models/companion_remote/remote_command.dart'; import '../../models/companion_remote/remote_session.dart'; import '../../utils/app_logger.dart'; +import '../../utils/happy_eyeballs.dart'; import '../../utils/serial_future_queue.dart'; import '../base_peer_service.dart'; import 'remote_auth_context.dart'; @@ -80,7 +81,11 @@ class CompanionRemotePeerService with KeepaliveMixin { _raceProbeFactory = raceProbeFactory ?? _openRaceProbe; static _RaceProbeConnection _openRaceProbe(Uri uri) { - final channel = IOWebSocketChannel.connect(uri, connectTimeout: const Duration(seconds: 5)); + final channel = IOWebSocketChannel.connect( + uri, + connectTimeout: const Duration(seconds: 5), + customClient: happyEyeballsHttpClient, + ); var connected = false; unawaited( channel.ready.then((_) { @@ -826,7 +831,11 @@ class CompanionRemotePeerService with KeepaliveMixin { _connectionStateController.add(RemoteSessionStatus.connecting); - final channel = IOWebSocketChannel.connect(Uri.parse(url), connectTimeout: _remoteConnectTimeout); + final channel = IOWebSocketChannel.connect( + Uri.parse(url), + connectTimeout: _remoteConnectTimeout, + customClient: happyEyeballsHttpClient, + ); attemptedChannel = channel; _channel = channel; _channelConnected = false; diff --git a/lib/services/library_events/library_event_socket.dart b/lib/services/library_events/library_event_socket.dart index a2c5c6907..df74693b3 100644 --- a/lib/services/library_events/library_event_socket.dart +++ b/lib/services/library_events/library_event_socket.dart @@ -7,6 +7,7 @@ import 'package:web_socket_channel/web_socket_channel.dart'; import '../../media/ids.dart'; import '../../media/library_change_event.dart'; import '../../utils/app_logger.dart'; +import '../../utils/happy_eyeballs.dart'; /// Builds a live channel for [uri]. Injectable so tests can supply a fake or /// point at an ephemeral loopback server. @@ -18,6 +19,7 @@ WebSocketChannel _defaultChannelFactory(Uri uri) => IOWebSocketChannel.connect( // Transport-level pings keep NAT mappings alive and surface a dead peer as // a close event; app-level keepalive (MediaBrowser) rides on top. pingInterval: const Duration(seconds: 30), + customClient: happyEyeballsHttpClient, ); /// Reconnecting websocket base for one server's library-change push channel. diff --git a/lib/utils/happy_eyeballs.dart b/lib/utils/happy_eyeballs.dart new file mode 100644 index 000000000..c715513f5 --- /dev/null +++ b/lib/utils/happy_eyeballs.dart @@ -0,0 +1,166 @@ +import 'dart:async'; +import 'dart:io'; + +/// dart:io resolves A before AAAA and connects in arrival order, so a +/// dual-stack host whose IPv4 path answers within 250 ms is never tried over +/// IPv6, whatever the system resolver ranked first. This does one lookup +/// (which keeps the resolver's RFC 6724 order), interleaves the families per +/// RFC 8305 and races the candidates [defaultAttemptDelay] apart. +/// +/// With a factory installed the SDK no longer secures the socket itself, and +/// `badCertificateCallback`/`SecurityContext` never reach it, so TLS is done +/// here. Proxied requests are handed back plain: the SDK tunnels and secures +/// those on its own. +Future> happyEyeballsConnectionFactory(Uri url, String? proxyHost, int? proxyPort) { + if (proxyHost != null) return Socket.startConnect(proxyHost, proxyPort!); + return Future.value(startHappyEyeballsConnect(url.host, url.port, secure: url.isScheme('https'))); +} + +/// For transports that build their own [HttpClient] when not handed one +/// (`WebSocket.connect`). Never closed. +final HttpClient happyEyeballsHttpClient = HttpClient()..connectionFactory = happyEyeballsConnectionFactory; + +const Duration defaultAttemptDelay = Duration(milliseconds: 250); + +typedef AddressLookup = Future> Function(String host); +typedef AddressConnect = Future> Function(InternetAddress address, int port); + +/// Returns synchronously so `HttpClient.connectionTimeout` and `cancel()` +/// cover the lookup as well as the connect. +ConnectionTask startHappyEyeballsConnect( + String host, + int port, { + bool secure = false, + Duration attemptDelay = defaultAttemptDelay, + AddressLookup lookup = InternetAddress.lookup, + AddressConnect connect = Socket.startConnect, +}) { + final race = _Race(host, port, attemptDelay, lookup, connect)..start(); + final socket = secure ? race.socket.then((raw) => _secure(raw, host)) : race.socket; + return ConnectionTask.fromSocket(socket, race.cancel); +} + +/// Alternates families starting with whichever the resolver ranked first. +List orderCandidates(List addresses) { + if (addresses.length < 2) return addresses; + final lead = addresses.where((a) => a.type == addresses.first.type).toList(); + final other = addresses.where((a) => a.type != addresses.first.type).toList(); + return [ + for (var i = 0; i < lead.length || i < other.length; i++) ...[ + if (i < lead.length) lead[i], + if (i < other.length) other[i], + ], + ]; +} + +Future _secure(Socket socket, String host) async { + try { + return await SecureSocket.secure(socket, host: host); + } catch (_) { + socket.destroy(); + rethrow; + } +} + +class _Race { + _Race(this._host, this._port, this._delay, this._lookup, this._connect); + + final String _host; + final int _port; + final Duration _delay; + final AddressLookup _lookup; + final AddressConnect _connect; + + final _result = Completer(); + final _inFlight = >{}; + List _addresses = const []; + Timer? _timer; + Socket? _winner; + int _next = 0; + int _pending = 0; + Object? _error; + StackTrace? _stackTrace; + bool _cancelled = false; + + Future get socket => _result.future; + + bool get _done => _cancelled || _result.isCompleted; + + Future start() async { + try { + // Uri.host keeps a link-local zone percent-encoded (`fe80::1%25en0`). + final host = _host.replaceFirst('%25', '%'); + final literal = InternetAddress.tryParse(host); + _addresses = literal != null ? [literal] : orderCandidates(await _lookup(host)); + } catch (error, stackTrace) { + if (!_done) _result.completeError(error, stackTrace); + return; + } + _startNext(); + } + + void _startNext() { + _timer?.cancel(); + if (_done) return; + if (_next == _addresses.length) { + if (_pending == 0) { + _result.completeError(_error ?? SocketException("Failed host lookup: '$_host'"), _stackTrace); + } + return; + } + final address = _addresses[_next++]; + if (_next < _addresses.length) _timer = Timer(_delay, _startNext); + _pending++; + unawaited(_attempt(address)); + } + + Future _attempt(InternetAddress address) async { + ConnectionTask? task; + try { + task = await _connect(address, _port); + if (_done) { + task.cancel(); + return; + } + _inFlight.add(task); + final socket = await task.socket; + _inFlight.remove(task); + if (_done) { + socket.destroy(); + return; + } + _timer?.cancel(); + _winner = socket; + _result.complete(socket); + _cancelInFlight(); + return; + } catch (error, stackTrace) { + _error ??= error; + _stackTrace ??= stackTrace; + } finally { + _inFlight.remove(task); + _pending--; + } + _startNext(); + } + + void cancel() { + if (_cancelled) return; + _cancelled = true; + _timer?.cancel(); + _cancelInFlight(); + // A cancel that lands during the TLS handshake would otherwise orphan the + // raw socket: the SDK has already stopped listening for the result. + _winner?.destroy(); + if (!_result.isCompleted) { + _result.completeError(SocketException('Connection attempt cancelled, host: $_host')); + } + } + + void _cancelInFlight() { + for (final task in [..._inFlight]) { + task.cancel(); + } + _inFlight.clear(); + } +} diff --git a/lib/utils/platform_http_client_io.dart b/lib/utils/platform_http_client_io.dart index d9fb12333..21128916f 100644 --- a/lib/utils/platform_http_client_io.dart +++ b/lib/utils/platform_http_client_io.dart @@ -5,6 +5,7 @@ import 'package:http/io_client.dart'; import 'package:win_http/win_http.dart'; import 'app_logger.dart'; +import 'happy_eyeballs.dart'; import 'managed_http_client.dart'; import 'media_server_timeouts.dart'; @@ -31,17 +32,16 @@ ManagedHttpClient _createIoClient(String debugLabel) { final httpClient = HttpClient() ..connectionTimeout = MediaServerTimeouts.connect ..maxConnectionsPerHost = 12 - ..idleTimeout = const Duration(seconds: 90); + ..idleTimeout = const Duration(seconds: 90) + ..connectionFactory = happyEyeballsConnectionFactory; return ManagedHttpClient(IOClient(httpClient), debugLabel: debugLabel, forceCloseOnDrainTimeout: true); } /// Every platform except Windows runs on the tuned dart:io client. /// -/// Windows keeps WinHTTP for `WINHTTP_OPTION_IPV6_FAST_FALLBACK` (Happy -/// Eyeballs, #1128), which dart:io has no equivalent for: it tries a -/// dual-stack host's addresses sequentially, so one unreachable IPv6 address -/// stalls the connection past the endpoint probe budget. WinHTTP also brings -/// the system proxy and the Schannel trust store. +/// Windows keeps WinHTTP for the system proxy and the Schannel trust store. +/// Its `WINHTTP_OPTION_IPV6_FAST_FALLBACK` (#1128) is what +/// [happyEyeballsConnectionFactory] now supplies everywhere else. /// /// Cronet and NSURLSession used to serve Android and Apple here. Both were /// measured slower than this client on the request shapes Plezy actually diff --git a/test/utils/happy_eyeballs_test.dart b/test/utils/happy_eyeballs_test.dart new file mode 100644 index 000000000..25c811efd --- /dev/null +++ b/test/utils/happy_eyeballs_test.dart @@ -0,0 +1,250 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:plezy/utils/happy_eyeballs.dart'; + +void main() { + final v6 = InternetAddress('2001:db8::1'); + final v6Alt = InternetAddress('2001:db8::2'); + final v4 = InternetAddress('192.0.2.1'); + final v4Alt = InternetAddress('192.0.2.2'); + + List literals(List addresses) => addresses.map((a) => a.address).toList(); + + group('orderCandidates', () { + test('keeps a single-family list as resolved', () { + expect(literals(orderCandidates([v6, v6Alt])), ['2001:db8::1', '2001:db8::2']); + expect(orderCandidates(const []), isEmpty); + }); + + test('interleaves families behind the resolver\'s first pick', () { + expect(literals(orderCandidates([v6, v6Alt, v4])), ['2001:db8::1', '192.0.2.1', '2001:db8::2']); + expect(literals(orderCandidates([v4, v4Alt, v6])), ['192.0.2.1', '2001:db8::1', '192.0.2.2']); + }); + }); + + group('startHappyEyeballsConnect', () { + late ServerSocket server; + late StreamSubscription accepted; + final serverSide = []; + + setUp(() async { + server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + accepted = server.listen(serverSide.add); + }); + + tearDown(() async { + await accepted.cancel(); + for (final socket in serverSide) { + socket.destroy(); + } + serverSide.clear(); + await server.close(); + }); + + ConnectionTask connected() => + ConnectionTask.fromSocket(Socket.connect(InternetAddress.loopbackIPv4, server.port), () {}); + + test('tries the resolver\'s first address first', () async { + final attempted = []; + final task = startHappyEyeballsConnect( + 'dual.test', + 443, + lookup: (_) async => [v6, v4], + connect: (address, port) async { + attempted.add(address.address); + return connected(); + }, + ); + addTearDown((await task.socket).destroy); + + expect(attempted, ['2001:db8::1']); + }); + + test('races the next candidate after the stagger and cancels the loser', () async { + final attempted = []; + var stalledCancelled = false; + final task = startHappyEyeballsConnect( + 'dual.test', + 443, + attemptDelay: const Duration(milliseconds: 30), + lookup: (_) async => [v6, v4], + connect: (address, port) async { + attempted.add(address.address); + if (address.type == InternetAddressType.IPv6) { + return ConnectionTask.fromSocket(Completer().future, () => stalledCancelled = true); + } + return connected(); + }, + ); + addTearDown((await task.socket).destroy); + + expect(attempted, ['2001:db8::1', '192.0.2.1']); + expect(stalledCancelled, isTrue); + }); + + test('starts the next candidate immediately when one fails', () async { + final attempted = []; + final stopwatch = Stopwatch()..start(); + final task = startHappyEyeballsConnect( + 'dual.test', + 443, + attemptDelay: const Duration(seconds: 5), + lookup: (_) async => [v6, v4], + connect: (address, port) async { + attempted.add(address.address); + if (address.type == InternetAddressType.IPv6) throw SocketException('no route to ${address.address}'); + return connected(); + }, + ); + addTearDown((await task.socket).destroy); + + expect(attempted, ['2001:db8::1', '192.0.2.1']); + expect(stopwatch.elapsed, lessThan(const Duration(seconds: 1))); + }); + + test('reports the first failure when every candidate fails', () async { + final task = startHappyEyeballsConnect( + 'dual.test', + 443, + attemptDelay: const Duration(seconds: 5), + lookup: (_) async => [v6, v4], + connect: (address, port) async => throw SocketException('no route to ${address.address}'), + ); + + await expectLater( + task.socket, + throwsA(isA().having((e) => e.message, 'message', contains('2001:db8::1'))), + ); + }); + + test('propagates a lookup failure', () async { + final task = startHappyEyeballsConnect( + 'missing.test', + 443, + lookup: (host) async => throw SocketException("Failed host lookup: '$host'"), + connect: (address, port) async => fail('must not connect'), + ); + + await expectLater(task.socket, throwsA(isA())); + }); + + test('cancel tears down every in-flight attempt', () async { + var cancels = 0; + final task = startHappyEyeballsConnect( + 'dual.test', + 443, + attemptDelay: const Duration(milliseconds: 10), + lookup: (_) async => [v6, v4], + connect: (address, port) async => ConnectionTask.fromSocket(Completer().future, () => cancels++), + ); + await Future.delayed(const Duration(milliseconds: 50)); + task.cancel(); + + await expectLater(task.socket, throwsA(isA())); + expect(cancels, 2); + }); + + test('cancel during the lookup never connects', () async { + final lookup = Completer>(); + var connects = 0; + final task = startHappyEyeballsConnect( + 'dual.test', + 443, + lookup: (_) => lookup.future, + connect: (address, port) async { + connects++; + return connected(); + }, + ); + task.cancel(); + await expectLater(task.socket, throwsA(isA())); + + lookup.complete([v4]); + await Future.delayed(Duration.zero); + expect(connects, 0); + }); + + test('skips the resolver for an address literal', () async { + final attempted = []; + final task = startHappyEyeballsConnect( + '192.0.2.1', + 443, + lookup: (_) async => fail('must not resolve'), + connect: (address, port) async { + attempted.add(address.address); + return connected(); + }, + ); + addTearDown((await task.socket).destroy); + + expect(attempted, ['192.0.2.1']); + }); + + test('decodes a percent-encoded link-local zone', () async { + final attempted = []; + final task = startHappyEyeballsConnect( + 'fe80::1%25lo0', + 443, + connect: (address, port) async { + attempted.add(address.address); + return connected(); + }, + ); + addTearDown((await task.socket).destroy); + + expect(attempted, ['fe80::1%lo0']); + }); + + test('secure: true performs a TLS handshake', () async { + final plaintext = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + final greeter = plaintext.listen((socket) => socket.add('HTTP/1.1 200 OK\r\n'.codeUnits)); + addTearDown(() async { + await greeter.cancel(); + await plaintext.close(); + }); + + final task = startHappyEyeballsConnect( + 'plex.invalid', + plaintext.port, + secure: true, + lookup: (_) async => [InternetAddress.loopbackIPv4], + ); + + await expectLater(task.socket, throwsA(anyOf(isA(), isA()))); + }); + }); + + group('happyEyeballsConnectionFactory', () { + test('bounds the lookup by HttpClient.connectionTimeout', () async { + final client = HttpClient() + ..connectionTimeout = const Duration(milliseconds: 100) + ..connectionFactory = (url, proxyHost, proxyPort) => Future.value( + startHappyEyeballsConnect(url.host, url.port, lookup: (_) => Completer>().future), + ); + addTearDown(() => client.close(force: true)); + + await expectLater(client.getUrl(Uri.parse('http://stalled.test/')), throwsA(isA())); + }); + + test('hands a plain socket to a proxy', () async { + final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + final accepted = server.listen((socket) => socket.destroy()); + addTearDown(() async { + await accepted.cancel(); + await server.close(); + }); + + final task = await happyEyeballsConnectionFactory( + Uri.parse('https://plex.invalid/library/sections'), + InternetAddress.loopbackIPv4.address, + server.port, + ); + final socket = await task.socket; + addTearDown(socket.destroy); + + expect(socket, isNot(isA())); + }); + }); +}