fix(plex): keep relay out of the phase-1 race, preferred persistence, and pinned sessions

A relay connection can win the phase-1 race (plex.tv edge answers fast),
gets saved as the preferred endpoint, and the cached URL then wins the
deterministic head-start probe at every bind — pinning future sessions to
relay's 2 Mbps cap even after direct connectivity returns, surfacing as
HTTP 500s on Original quality. A session that lands on relay legitimately
(direct down at connect time) stays there all evening: re-optimization
only fired on connectivity events.

Relay is now a fallback tier in the race: all probes still start at once,
but a relay success is held until every direct candidate has failed, and a
cached relay URL gets no head start. A single persist gate refuses relay
at all four endpoint-save sites, classified against both the live server
and the connect-time capture so rotated relay URIs cannot slip through.
While the active endpoint is relay, a bounded-backoff re-probe (30s/60s/
120s) re-races the candidates so a returning direct endpoint promotes
away without a restart.
This commit is contained in:
edde746
2026-08-18 09:23:04 +02:00
parent 986237491f
commit 304d8a5e33
5 changed files with 520 additions and 22 deletions
+124 -7
View File
@@ -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<String, Timer> _relayEscapeTimers = {};
final Map<String, int> _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<void> _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<bool> _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<void> debugFireRelayEscapeForTesting(ServerId serverId) {
_relayEscapeTimers.remove(serverId)?.cancel();
return _runRelayEscape(serverId);
}
Future<void> _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<void> _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 = <MediaServerClient>{..._clients.values, ..._jellyfinByCompoundId.values};
+5 -1
View File
@@ -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) => {
+85 -11
View File
@@ -33,6 +33,16 @@ class EndpointRaceSelection<C, R> {
/// 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<EndpointRaceSelection<C, R>> raceEndpointCandidates<C, R>({
@@ -43,6 +53,7 @@ Stream<EndpointRaceSelection<C, R>> raceEndpointCandidates<C, R>({
Map<String, Object?> Function(C candidate, R result)? failureLogFields,
String? preferredUrl,
C? Function(String url)? candidateForUrl,
int Function(C candidate)? tierOf,
required Future<R> Function(C candidate, Duration timeout) probe,
required Future<R> Function(C candidate) measure,
required bool Function(R result) isSuccess,
@@ -67,6 +78,14 @@ Stream<EndpointRaceSelection<C, R>> raceEndpointCandidates<C, R>({
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<EndpointRaceSelection<C, R>> raceEndpointCandidates<C, R>({
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<C, R>({
required Future<R> 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<String, Object?> 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 = <int, int>{};
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<C, R>({
);
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<C, R>({
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();
}),
);
}
+184 -3
View File
@@ -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<PlexConnection>(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 = <JellyfinConnection>[];
@@ -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<String>? prioritizedEndpoints;
final bool hasEndpointCallback;
final Future<void> 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,
),
+122
View File
@@ -23,6 +23,7 @@ void main() {
Stream<EndpointRaceSelection<String, _Result>> race({
required List<String> candidates,
String? preferred,
int Function(String url)? tierOf,
required Future<_Result> Function(String url) probe,
Future<_Result> Function(String url)? measure,
String? Function(Map<String, _Result> 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 = <String, int>{};
final selections = <EndpointRaceSelection<String, _Result>>[];
final gates = <String, Completer<_Result>>{};
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 = <EndpointRaceSelection<String, _Result>>[];
final gates = <String, Completer<_Result>>{};
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 = <EndpointRaceSelection<String, _Result>>[];
final gates = <String, Completer<_Result>>{};
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 = <String, int>{};
final selections = <EndpointRaceSelection<String, _Result>>[];
final gates = <String, Completer<_Result>>{};
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);
});
});
}