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())); + }); + }); +}