diff --git a/lib/services/multi_server_manager.dart b/lib/services/multi_server_manager.dart index 8aab57caa..0f7dfb205 100644 --- a/lib/services/multi_server_manager.dart +++ b/lib/services/multi_server_manager.dart @@ -161,6 +161,12 @@ class MultiServerManager { /// Debounce timer for connectivity events — collapses rapid network flapping Timer? _connectivityDebounce; + /// Per-server relay-escape probe timers and attempt counts + /// (see [_syncRelayEscape]). + final Map _relayEscapeTimers = {}; + final Map _relayEscapeAttempts = {}; + static const _relayEscapeBaseDelay = Duration(seconds: 30); + /// Get all registered server IDs (Plex + MediaBrowser). /// /// Sourced from [_clients] rather than [_plexServers] because @@ -375,15 +381,16 @@ class MultiServerManager { serverName: server.name, prioritizedEndpoints: prioritizedEndpoints, onEndpointChanged: (newUrl) async { - await storage.saveServerEndpoint(ServerId(serverId), newUrl); - appLogger.i('Updated endpoint for ${server.name} after failover: $newUrl'); + appLogger.i('Endpoint changed for ${server.name} after failover: $newUrl'); + await _savePreferredEndpoint(ServerId(serverId), storage, newUrl, capturedServer: server); + _syncRelayEscape(ServerId(serverId)); }, onAllEndpointsExhausted: () => _onServerEndpointsExhausted(ServerId(serverId)), seedTranscoderVideoSupport: observedTranscoderVideo, ); - // Save the initial endpoint - await storage.saveServerEndpoint(ServerId(serverId), baseUrl); + // Save the initial endpoint (relay is refused — see _savePreferredEndpoint) + await _savePreferredEndpoint(ServerId(serverId), storage, baseUrl, capturedServer: server); appLogger.i( 'Connected ${server.name}', @@ -401,6 +408,31 @@ class MultiServerManager { return client; } + /// Persist [url] as [serverId]'s preferred endpoint unless it is a relay. + /// + /// The preferred endpoint gets a deterministic head start at the next bind, + /// so persisting a relay URL would pin future sessions to plex.tv's + /// bandwidth-capped relay even after direct connectivity returns (#1974). + /// The client may still *use* a relay endpoint — only persistence is + /// refused; [_syncRelayEscape] owns getting off it. Classified against both + /// the live registered server and the connect-time [capturedServer]: after + /// a profile refresh rotates relay URIs only one of the two may still list + /// the URL, and a URL absent from a server's connection list falls through + /// to the custom-hostname classifier, which cannot recognize relays. + Future _savePreferredEndpoint( + ServerId serverId, + StorageService storage, + String url, { + PlexServer? capturedServer, + }) async { + bool isRelay(PlexServer? server) => server?.networkClassForUrl(url) == PlexNetworkClass.relay; + if (isRelay(_plexServers[serverId]) || isRelay(capturedServer)) { + appLogger.d('Refusing to persist relay endpoint as preferred for $serverId'); + return; + } + await storage.saveServerEndpoint(serverId, url); + } + /// Persists a new endpoint, rebuilds the failover list, and switches the /// client only while it is still the registered client for this server. Future _promoteEndpoint({ @@ -416,11 +448,13 @@ class MultiServerManager { } if (!isCurrent()) return false; - await storage.saveServerEndpoint(serverId, newUrl); + await _savePreferredEndpoint(serverId, storage, newUrl, capturedServer: server); if (!isCurrent()) return false; final newEndpoints = server.prioritizedEndpointUrls(preferredFirst: newUrl); await client.updateEndpointPreferences(newEndpoints, switchToFirst: true); - return isCurrent(); + final current = isCurrent(); + if (current) _syncRelayEscape(serverId); + return current; } /// Continues draining the connection optimization stream in the background, @@ -494,6 +528,7 @@ class MultiServerManager { /// deliberately left alone — they are owned by the futures that set them. MediaServerClient? _forgetServer(String serverId) { _reconnectDebounce.remove(serverId)?.cancel(); + _stopRelayEscape(serverId); final client = _clients.remove(serverId); _activeJellyfinMachine.remove(serverId); _plexServers.remove(serverId); @@ -589,6 +624,7 @@ class MultiServerManager { _serverStatus[serverId] = true; _authErrorServers.remove(serverId); bound.add(serverId); + _syncRelayEscape(ServerId(serverId)); _connectProgressController.add((serverId: serverId, online: true)); } catch (e, stackTrace) { if (isStale() || !identical(_plexServers[serverId], server)) return; @@ -1065,7 +1101,7 @@ class MultiServerManager { appLogger.i('Switched ${server.name} to better endpoint: $newUrl', error: {'type': connection.displayType}); } else { if (_plexServers[serverId] != server) return; - await storage.saveServerEndpoint(serverId, newUrl); + await _savePreferredEndpoint(serverId, storage, newUrl, capturedServer: server); if (_plexServers[serverId] != server) return; appLogger.i('Updated optimal endpoint for ${server.name}: $newUrl', error: {'type': connection.displayType}); } @@ -1075,6 +1111,81 @@ class MultiServerManager { } } + /// Whether [serverId]'s registered client is currently talking to a Plex + /// relay endpoint. + bool _isOnRelay(ServerId serverId) { + final client = _clients[serverId]; + final server = _plexServers[serverId]; + return client is PlexClient && + server != null && + server.networkClassForUrl(client.config.baseUrl) == PlexNetworkClass.relay; + } + + /// Reconcile the relay-escape prober with [serverId]'s current endpoint. + /// + /// A session can land on relay legitimately (direct connectivity was down + /// at connect time) or transiently (a failover walked onto it). plex.tv + /// caps relay bandwidth, and the only other re-optimization trigger is a + /// connectivity event — which never fires on a stable network — so a relay + /// session would otherwise stay capped until restart. While the active + /// endpoint classifies as relay, re-race the candidates on a bounded + /// backoff; the phase-2 selector prefers any working direct endpoint, so + /// the first successful direct probe promotes away and stops the prober. + void _syncRelayEscape(ServerId serverId) { + if (!_isOnRelay(serverId)) { + _stopRelayEscape(serverId); + return; + } + if (_relayEscapeTimers.containsKey(serverId)) return; + final attempt = _relayEscapeAttempts[serverId] ?? 0; + final delay = _relayEscapeDelay(attempt); + appLogger.i( + 'Connected via relay, scheduling direct-endpoint re-probe', + error: {'serverId': serverId, 'attempt': attempt, 'delaySeconds': delay.inSeconds}, + ); + _relayEscapeTimers[serverId] = Timer(delay, () { + _relayEscapeTimers.remove(serverId); + unawaited(_runRelayEscape(serverId)); + }); + } + + void _stopRelayEscape(String serverId) { + _relayEscapeTimers.remove(serverId)?.cancel(); + _relayEscapeAttempts.remove(serverId); + } + + /// 30s, 60s, then every 120s — a returning direct endpoint is picked up + /// quickly without re-racing a genuinely relay-only server forever at a + /// tight cadence. + Duration _relayEscapeDelay(int attempt) => _relayEscapeBaseDelay * (1 << attempt.clamp(0, 2)); + + /// Whether a relay-escape re-probe is scheduled for [serverId]. + @visibleForTesting + bool debugHasPendingRelayEscapeForTesting(ServerId serverId) => _relayEscapeTimers.containsKey(serverId); + + /// Fire [serverId]'s pending relay-escape probe immediately instead of + /// waiting out its backoff — timers armed during a real-async bind are + /// unreachable from a test's fakeAsync zone. + @visibleForTesting + Future debugFireRelayEscapeForTesting(ServerId serverId) { + _relayEscapeTimers.remove(serverId)?.cancel(); + return _runRelayEscape(serverId); + } + + Future _runRelayEscape(ServerId serverId) async { + final server = _plexServers[serverId]; + if (server == null || !_isOnRelay(serverId) || !isServerOnline(serverId)) { + // Gone, promoted away, or offline (the reconnect path owns offline + // servers and re-syncs on success). + _stopRelayEscape(serverId); + return; + } + _relayEscapeAttempts[serverId] = (_relayEscapeAttempts[serverId] ?? 0) + 1; + await _runServerTask(serverId, () => _reoptimizeServer(serverId: serverId, server: server, reason: 'relay-escape')); + // Still on relay (or the optimize slot was busy): keep probing. + _syncRelayEscape(serverId); + } + /// Attempt full reconnection for a single offline server Future _reconnectServer(ServerId serverId, PlexServer server) async { final clientId = _resolveClientIdentifier(serverId); @@ -1106,6 +1217,7 @@ class MultiServerManager { final oldClient = _clients[serverId]; if (oldClient != null) unawaited(_closeClientGracefully(oldClient)); _clients[serverId] = client; + _syncRelayEscape(serverId); updateServerStatus(serverId, true); appLogger.i('Successfully reconnected to ${server.name}'); } catch (e) { @@ -1311,6 +1423,11 @@ class MultiServerManager { timer.cancel(); } _reconnectDebounce.clear(); + for (final timer in _relayEscapeTimers.values) { + timer.cancel(); + } + _relayEscapeTimers.clear(); + _relayEscapeAttempts.clear(); _activeHealthCheck = null; _activeReconnect = null; final clients = {..._clients.values, ..._jellyfinByCompoundId.values}; diff --git a/lib/services/plex_auth_service.dart b/lib/services/plex_auth_service.dart index f83ba8322..5163f091b 100644 --- a/lib/services/plex_auth_service.dart +++ b/lib/services/plex_auth_service.dart @@ -426,7 +426,10 @@ class PlexServer { /// Find the best working connection by testing them /// Returns a Stream that emits connections progressively: - /// 1. First emission: The first connection that responds successfully + /// 1. First emission: The first connection that responds successfully. + /// Relay is a fallback tier: a relay success is held until every direct + /// candidate has failed, and a cached relay URL gets no head start, so + /// a fast plex.tv relay edge can never beat a working direct endpoint. /// 2. Second emission (optional): The best connection after latency testing /// Priority: local > remote > relay, then HTTPS > HTTP, then lowest latency /// Tests both plex.direct URI and direct IP for each connection @@ -473,6 +476,7 @@ class PlexServer { candidates: candidates, preferredUrl: preferredUri, candidateForUrl: _candidateForUrl, + tierOf: (candidate) => candidate.connection.relay ? 1 : 0, urlOf: (candidate) => candidate.url, displayTypeOf: (candidate) => candidate.connection.displayType, failureLogFields: (candidate, result) => { diff --git a/lib/utils/endpoint_race.dart b/lib/utils/endpoint_race.dart index ae814d125..1293e82e5 100644 --- a/lib/utils/endpoint_race.dart +++ b/lib/utils/endpoint_race.dart @@ -33,6 +33,16 @@ class EndpointRaceSelection { /// a lower-latency URL in the background without blocking initial connection /// setup. /// +/// [tierOf] makes phase 1 class-aware without delaying any probe: candidates +/// map to an integer tier (0 = preferred). Every probe still starts +/// immediately, but a success from a higher (fallback) tier — e.g. a Plex +/// relay — is held and only accepted once every lower-tier candidate has +/// failed, so a marginally slower direct endpoint always beats a fast relay +/// edge. The hold is bounded by [raceTimeout] because that already bounds +/// each probe. A cached/preferred URL resolving to a fallback-tier candidate +/// also gets no head start; it competes as an ordinary participant. Omitting +/// [tierOf] keeps the plain first-success race. +/// /// Diagnostic events intentionally omit candidate URLs. Backends register /// endpoints separately for unavoidable network-layer diagnostics. Stream> raceEndpointCandidates({ @@ -43,6 +53,7 @@ Stream> raceEndpointCandidates({ Map Function(C candidate, R result)? failureLogFields, String? preferredUrl, C? Function(String url)? candidateForUrl, + int Function(C candidate)? tierOf, required Future Function(C candidate, Duration timeout) probe, required Future Function(C candidate) measure, required bool Function(R result) isSuccess, @@ -67,6 +78,14 @@ Stream> raceEndpointCandidates({ if (preferredUrl != null && preferredUrl.isNotEmpty) { cachedCandidate = candidateForUrl?.call(preferredUrl) ?? _candidateForUrl(candidates, urlOf, preferredUrl); } + if (cachedCandidate != null && tierOf != null && tierOf(cachedCandidate) > 0) { + // A fallback-tier endpoint must never win the deterministic head start: + // it stays an ordinary race participant so preferred-tier candidates can + // beat it. Guarded here rather than at bind sites so every binder + // self-heals a fallback URL persisted by older builds. + appLogger.d('Cached $label endpoint is fallback-tier, racing without a head start'); + cachedCandidate = null; + } if (cachedCandidate != null) { final cached = cachedCandidate; appLogger.d('Testing cached $label endpoint with a head start on the race'); @@ -109,6 +128,7 @@ Stream> raceEndpointCandidates({ urlOf: urlOf, displayTypeOf: displayTypeOf, failureLogFields: failureLogFields, + tierOf: tierOf, probe: (candidate, timeout) => mergedCached != null && identical(candidate, mergedCached) ? pendingCachedProbe! : probe(candidate, timeout), isSuccess: isSuccess, @@ -194,12 +214,54 @@ Future<({C candidate, R result})?> _raceFirstSuccess({ required Future Function(C candidate, Duration timeout) probe, required bool Function(R result) isSuccess, required Duration timeout, + int Function(C candidate)? tierOf, String Function(C candidate)? displayTypeOf, Map Function(C candidate, R result)? failureLogFields, void Function(C candidate, R result)? onFirstSuccess, }) async { final completer = Completer<({C candidate, R result})?>(); - var completedTests = 0; + + int tierFor(C candidate) => tierOf?.call(candidate) ?? 0; + final pendingByTier = {}; + for (final candidate in candidates) { + pendingByTier.update(tierFor(candidate), (count) => count + 1, ifAbsent: () => 1); + } + + // Best fallback-tier success so far: only accepted once no lower-tier + // probe is still pending, so a preferred-tier endpoint that answers later + // (but within its probe timeout) still wins. + ({C candidate, R result, int tier})? held; + + bool lowerTierPending(int tier) { + for (final entry in pendingByTier.entries) { + if (entry.key < tier && entry.value > 0) return true; + } + return false; + } + + void win(C candidate, R result) { + onFirstSuccess?.call(candidate, result); + completer.complete((candidate: candidate, result: result)); + } + + // Runs after a probe resolves without an immediate win: releases the held + // fallback success once every lower tier has drained, and completes with + // null when all candidates resolved without any success. + void settle() { + if (completer.isCompleted) return; + final fallback = held; + if (fallback != null && !lowerTierPending(fallback.tier)) { + appLogger.i( + '$label fallback-tier endpoint accepted after preferred tiers failed', + error: {'type': displayTypeOf?.call(fallback.candidate)}, + ); + win(fallback.candidate, fallback.result); + return; + } + if (pendingByTier.values.every((count) => count == 0)) { + completer.complete(null); + } + } appLogger.d( 'Running $label endpoint race to find first working endpoint', @@ -207,10 +269,11 @@ Future<({C candidate, R result})?> _raceFirstSuccess({ ); for (final candidate in candidates) { + final tier = tierFor(candidate); unawaited( probe(candidate, timeout) .then((result) { - completedTests++; + pendingByTier[tier] = pendingByTier[tier]! - 1; if (!isSuccess(result)) { final failureFields = failureLogFields?.call(candidate, result); @@ -221,27 +284,38 @@ Future<({C candidate, R result})?> _raceFirstSuccess({ if (failureFields != null) ..._sanitizeEndpointFields(failureFields, urlOf(candidate)), }, ); + settle(); + return; } - if (isSuccess(result) && !completer.isCompleted) { - onFirstSuccess?.call(candidate, result); - completer.complete((candidate: candidate, result: result)); + if (completer.isCompleted) return; + + if (!lowerTierPending(tier)) { + win(candidate, result); + return; } - if (completedTests == candidates.length && !completer.isCompleted) { - completer.complete(null); + // A fallback-tier endpoint answered while a preferred tier is + // still probing (e.g. Plex relay vs direct). Hold it: it wins + // only if every preferred-tier candidate fails, so a slower + // direct endpoint always beats a fast relay edge (#1974). + final current = held; + if (current == null || tier < current.tier) { + appLogger.d( + '$label fallback-tier endpoint succeeded, holding while preferred tiers race', + error: {'type': displayTypeOf?.call(candidate)}, + ); + held = (candidate: candidate, result: result, tier: tier); } }) .catchError((Object error, StackTrace stackTrace) { - completedTests++; + pendingByTier[tier] = pendingByTier[tier]! - 1; appLogger.w( '$label endpoint candidate threw during race', error: {'errorType': error.runtimeType.toString()}, stackTrace: stackTrace, ); - if (completedTests == candidates.length && !completer.isCompleted) { - completer.complete(null); - } + settle(); }), ); } diff --git a/test/services/multi_server_manager_test.dart b/test/services/multi_server_manager_test.dart index 39c8dcf3f..1972a361c 100644 --- a/test/services/multi_server_manager_test.dart +++ b/test/services/multi_server_manager_test.dart @@ -968,6 +968,174 @@ void main() { }); }); + group('relay endpoint handling', () { + test('a relay phase-1 winner is never persisted; a later direct promotion is', () async { + final storage = await _prepareFreshPlexManagerTest(); + final direct = _plexEndpoint('direct'); + final relay = _plexRelayEndpoint('relay'); + final discoveries = StreamController(sync: true); + final server = _ControlledPlexServer( + serverId: 'relay-server', + endpoints: [direct, relay], + discoveryStreams: [() => discoveries.stream], + ); + final factory = _RecordingPlexFactory(); + final manager = MultiServerManager( + plexClientFactory: factory.create, + connectivityChanges: () => const Stream.empty(), + ); + addTearDown(manager.dispose); + addTearDown(discoveries.close); + + final refresh = manager.refreshTokensForProfile(_plexAccount('relay-account', [server]), profileId: 'profile-a'); + await pumpEventQueue(); + discoveries.add(relay); + expect(await refresh, {'relay-server'}); + await pumpEventQueue(); + + final client = factory.clients['relay-server']!; + expect(client.config.baseUrl, relay.uri); + expect(storage.getServerEndpoint(ServerId('relay-server')), isNull); + expect(manager.debugHasPendingRelayEscapeForTesting(ServerId('relay-server')), isTrue); + + discoveries.add(direct); + await discoveries.close(); + await pumpEventQueue(times: 20); + + expect(client.config.baseUrl, direct.uri); + expect(storage.getServerEndpoint(ServerId('relay-server')), direct.uri); + expect(manager.debugHasPendingRelayEscapeForTesting(ServerId('relay-server')), isFalse); + }); + + test('failover onto a relay endpoint never overwrites the preferred endpoint', () async { + final storage = await _prepareFreshPlexManagerTest(); + final direct = _plexEndpoint('failover-direct'); + final relay = _plexRelayEndpoint('failover-relay'); + final server = _ControlledPlexServer( + serverId: 'failover-server', + endpoints: [direct, relay], + discoveryStreams: [() => Stream.value(direct)], + ); + final factory = _RecordingPlexFactory(); + final manager = MultiServerManager( + plexClientFactory: factory.create, + connectivityChanges: () => const Stream.empty(), + ); + addTearDown(manager.dispose); + + expect( + await manager.refreshTokensForProfile(_plexAccount('failover-account', [server]), profileId: 'profile-a'), + {'failover-server'}, + ); + await pumpEventQueue(times: 20); + expect(storage.getServerEndpoint(ServerId('failover-server')), direct.uri); + + // Mid-session failover walks onto the relay candidate: the client may + // use it, but it must not become the next bind's head-start endpoint. + final onEndpointChanged = factory.calls.single.endpointChanged!; + await onEndpointChanged(relay.uri); + expect(storage.getServerEndpoint(ServerId('failover-server')), direct.uri); + + // Reverse rotation (PR #1974 review): a profile refresh rotates the + // relay URI away, then the old relay URL reports a late failover + // switch. The live server no longer lists it — the custom-hostname + // fallback alone would classify it as remote — but the connect-time + // capture still recognizes it as relay. + final rotated = _ControlledPlexServer( + serverId: 'failover-server', + endpoints: [direct, _plexRelayEndpoint('rotated-relay')], + discoveryStreams: [() => Stream.value(direct)], + ); + expect( + await manager.refreshTokensForProfile(_plexAccount('failover-account', [rotated]), profileId: 'profile-a'), + {'failover-server'}, + ); + await onEndpointChanged(relay.uri); + expect(storage.getServerEndpoint(ServerId('failover-server')), direct.uri); + }); + + test('a relay session re-probes on backoff and promotes a returning direct endpoint', () async { + final storage = await _prepareFreshPlexManagerTest(); + final direct = _plexEndpoint('escape-direct'); + final relay = _plexRelayEndpoint('escape-relay'); + final server = _ControlledPlexServer( + serverId: 'escape-server', + endpoints: [direct, relay], + discoveryStreams: [ + () => Stream.value(relay), // bind: only relay answers + () => Stream.value(relay), // first escape probe: still relay-only + () => Stream.value(direct), // second escape probe: direct is back + ], + ); + final factory = _RecordingPlexFactory(); + final manager = MultiServerManager( + plexClientFactory: factory.create, + connectivityChanges: () => const Stream.empty(), + ); + addTearDown(manager.dispose); + + expect(await manager.refreshTokensForProfile(_plexAccount('escape-account', [server]), profileId: 'profile-a'), { + 'escape-server', + }); + await pumpEventQueue(times: 20); + final client = factory.clients['escape-server']!; + expect(client.config.baseUrl, relay.uri); + expect(server.discoveryCalls, 1); + expect(manager.debugHasPendingRelayEscapeForTesting(ServerId('escape-server')), isTrue); + + fakeAsync((async) { + // Fire the bind-scheduled probe (its real timer lives outside this zone). + unawaited(manager.debugFireRelayEscapeForTesting(ServerId('escape-server'))); + async.flushMicrotasks(); + expect(server.discoveryCalls, 2); + expect(client.config.baseUrl, relay.uri); + + // Still relay-only: rescheduled with a doubled backoff (30s -> 60s). + async.elapse(const Duration(seconds: 60) - const Duration(milliseconds: 1)); + async.flushMicrotasks(); + expect(server.discoveryCalls, 2); + async.elapse(const Duration(milliseconds: 1)); + async.flushMicrotasks(); + + expect(server.discoveryCalls, 3); + expect(client.config.baseUrl, direct.uri); + expect(storage.getServerEndpoint(ServerId('escape-server')), direct.uri); + expect(manager.debugHasPendingRelayEscapeForTesting(ServerId('escape-server')), isFalse); + + // Off relay: the prober stays stopped. + async.elapse(const Duration(minutes: 10)); + async.flushMicrotasks(); + expect(server.discoveryCalls, 3); + }); + }); + + test('removing a relay-connected server cancels its escape prober', () async { + await _prepareFreshPlexManagerTest(); + final relay = _plexRelayEndpoint('removed-relay'); + final server = _ControlledPlexServer( + serverId: 'removed-server', + endpoints: [_plexEndpoint('removed-direct'), relay], + discoveryStreams: [() => Stream.value(relay)], + ); + final factory = _RecordingPlexFactory(); + final manager = MultiServerManager( + plexClientFactory: factory.create, + connectivityChanges: () => const Stream.empty(), + ); + addTearDown(manager.dispose); + + expect(await manager.refreshTokensForProfile(_plexAccount('removed-account', [server]), profileId: 'profile-a'), { + 'removed-server', + }); + await pumpEventQueue(times: 20); + expect(manager.debugHasPendingRelayEscapeForTesting(ServerId('removed-server')), isTrue); + + manager.removeServer(ServerId('removed-server')); + expect(manager.debugHasPendingRelayEscapeForTesting(ServerId('removed-server')), isFalse); + expect(server.discoveryCalls, 1); + }); + }); + group('Jellyfin connection updates', () { test('persists refreshed admin status discovered during health checks', () async { final persisted = []; @@ -1648,6 +1816,17 @@ PlexConnection _plexEndpoint(String label) => PlexConnection( ipv6: false, ); +/// Mirrors a plex.tv relay connection: remote, `relay: true`. +PlexConnection _plexRelayEndpoint(String label) => PlexConnection( + protocol: 'https', + address: '$label.invalid', + port: 8443, + uri: 'https://$label.invalid:8443', + local: false, + relay: true, + ipv6: false, +); + class _ControlledPlexServer extends PlexServer { _ControlledPlexServer({ required String serverId, @@ -1677,7 +1856,7 @@ class _PlexFactoryCall { required this.serverId, required this.profileScopeId, required this.prioritizedEndpoints, - required this.hasEndpointCallback, + required this.endpointChanged, required this.hasExhaustionCallback, required this.seedTranscoderVideoSupport, }); @@ -1686,9 +1865,11 @@ class _PlexFactoryCall { final ServerId serverId; final PlexProfileScopeId profileScopeId; final List? prioritizedEndpoints; - final bool hasEndpointCallback; + final Future Function(String newBaseUrl)? endpointChanged; final bool hasExhaustionCallback; final bool? seedTranscoderVideoSupport; + + bool get hasEndpointCallback => endpointChanged != null; } class _RecordingPlexFactory { @@ -1715,7 +1896,7 @@ class _RecordingPlexFactory { serverId: serverId, profileScopeId: profileScopeId, prioritizedEndpoints: prioritizedEndpoints, - hasEndpointCallback: onEndpointChanged != null, + endpointChanged: onEndpointChanged, hasExhaustionCallback: onAllEndpointsExhausted != null, seedTranscoderVideoSupport: seedTranscoderVideoSupport, ), diff --git a/test/utils/endpoint_race_test.dart b/test/utils/endpoint_race_test.dart index 75a49e438..c1ba0d94f 100644 --- a/test/utils/endpoint_race_test.dart +++ b/test/utils/endpoint_race_test.dart @@ -23,6 +23,7 @@ void main() { Stream> race({ required List candidates, String? preferred, + int Function(String url)? tierOf, required Future<_Result> Function(String url) probe, Future<_Result> Function(String url)? measure, String? Function(Map results)? selectBest, @@ -33,6 +34,7 @@ void main() { candidates: candidates, urlOf: (c) => c, preferredUrl: preferred, + tierOf: tierOf, failureLogFields: failureLogFields, probe: (c, _) => probe(c), measure: measure ?? (c) async => (url: c, ok: false), @@ -301,4 +303,124 @@ void main() { expect(selections.last.candidate, 'better'); }); }); + + group('fallback tiers', () { + int relayTier(String url) => url == 'relay' ? 1 : 0; + + test('fallback-tier success is held and accepted only after preferred tiers fail', () { + fakeAsync((async) { + final probeCounts = {}; + final selections = >[]; + final gates = >{}; + + race( + candidates: ['direct', 'relay'], + tierOf: relayTier, + probe: (url) { + probeCounts[url] = (probeCounts[url] ?? 0) + 1; + return (gates[url] = Completer<_Result>()).future; + }, + ).listen(selections.add); + async.flushMicrotasks(); + + // The hold delays acceptance, never the probe: both start at once. + expect(probeCounts, {'direct': 1, 'relay': 1}); + + gates['relay']!.complete((url: 'relay', ok: true)); + async.flushMicrotasks(); + expect(selections, isEmpty); + + gates['direct']!.complete((url: 'direct', ok: false)); + async.flushMicrotasks(); + expect(selections.first.phase, EndpointRacePhase.first); + expect(selections.first.candidate, 'relay'); + }); + }); + + test('a slower preferred-tier success beats an earlier fallback-tier success', () { + fakeAsync((async) { + final selections = >[]; + final gates = >{}; + + race( + candidates: ['direct', 'relay'], + tierOf: relayTier, + probe: (url) => (gates[url] = Completer<_Result>()).future, + ).listen(selections.add); + async.flushMicrotasks(); + + gates['relay']!.complete((url: 'relay', ok: true)); + async.flushMicrotasks(); + expect(selections, isEmpty); + + gates['direct']!.complete((url: 'direct', ok: true)); + async.flushMicrotasks(); + expect(selections.first.candidate, 'direct'); + }); + }); + + test('fallback tier wins without delay once every preferred-tier candidate has failed', () { + fakeAsync((async) { + final selections = >[]; + final gates = >{}; + + race( + candidates: ['direct', 'relay'], + tierOf: relayTier, + probe: (url) => (gates[url] = Completer<_Result>()).future, + ).listen(selections.add); + async.flushMicrotasks(); + + gates['direct']!.complete((url: 'direct', ok: false)); + async.flushMicrotasks(); + gates['relay']!.complete((url: 'relay', ok: true)); + async.flushMicrotasks(); + + // No hold window remains — accepted on the spot, no timer elapse. + expect(selections.first.candidate, 'relay'); + }); + }); + + test('cached fallback-tier URL gets no head start and still loses to a preferred tier', () { + fakeAsync((async) { + final probeCounts = {}; + final selections = >[]; + final gates = >{}; + + race( + candidates: ['direct', 'relay'], + preferred: 'relay', + tierOf: relayTier, + probe: (url) { + probeCounts[url] = (probeCounts[url] ?? 0) + 1; + return (gates[url] = Completer<_Result>()).future; + }, + ).listen(selections.add); + async.flushMicrotasks(); + + // A cached relay URL (persisted by an older build) is demoted to an + // ordinary participant: no head-start window where only it probes. + expect(probeCounts, {'direct': 1, 'relay': 1}); + + gates['relay']!.complete((url: 'relay', ok: true)); + async.flushMicrotasks(); + expect(selections, isEmpty); + + gates['direct']!.complete((url: 'direct', ok: true)); + async.flushMicrotasks(); + expect(selections.first.candidate, 'direct'); + expect(selections.first.fromPreferred, isFalse); + }); + }); + + test('emits nothing when every tier fails', () async { + final selections = await race( + candidates: ['direct', 'relay'], + tierOf: relayTier, + probe: (url) async => (url: url, ok: false), + ).toList(); + + expect(selections, isEmpty); + }); + }); }