diff --git a/lib/main.dart b/lib/main.dart index 78f2125b..9da55330 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -1308,13 +1308,26 @@ class _SetupScreenState extends State with MountedSetStateMixin { /// Wire per-server status updates from [MultiServerManager] into the /// splash list so the user sees check/cross marks land as the binder - /// brings each client online. Best-effort: stops listening when the - /// state goes away. + /// brings each client online. [MultiServerManager.connectProgressStream] + /// fires as each individual server settles; [MultiServerManager.statusStream] + /// emits once per connect pass and back-fills anything the progress stream + /// missed (e.g. servers torn down by the binder's visibility sweep). + /// Best-effort: stops listening when the state goes away. StreamSubscription>? _statusSub; + StreamSubscription<({String serverId, bool online})>? _connectProgressSub; void _bindServerStatusListener(ActiveProfileProvider _, MultiServerManager Function() resolveManager) { _statusSub?.cancel(); + _connectProgressSub?.cancel(); final manager = resolveManager(); + _connectProgressSub = manager.connectProgressStream.listen((progress) { + if (!mounted) return; + final existing = _serverStatus[progress.serverId]; + if (existing == null) return; + setState(() { + _serverStatus[progress.serverId] = (existing.$1, progress.online); + }); + }); _statusSub = manager.statusStream.listen((status) { if (!mounted) return; setState(() { @@ -1333,6 +1346,7 @@ class _SetupScreenState extends State with MountedSetStateMixin { @override void dispose() { _statusSub?.cancel(); + _connectProgressSub?.cancel(); super.dispose(); } diff --git a/lib/profiles/active_profile_binder.dart b/lib/profiles/active_profile_binder.dart index 8efea8ee..7014f89f 100644 --- a/lib/profiles/active_profile_binder.dart +++ b/lib/profiles/active_profile_binder.dart @@ -36,6 +36,11 @@ class _ProfileBindResult { final Set expectedServerIds; } +/// Settled outcome of a `fetchServers` call, so the resource refresh can run +/// alongside an optimistic cached-metadata bind without an early failure +/// surfacing as an unhandled async error. +typedef _FetchOutcome = ({List? servers, Object? error, StackTrace? stackTrace}); + @visibleForTesting bool shouldUsePlexHomeTokenCache({required bool preVerified, required bool hasBoundOnce}) { return preVerified || !hasBoundOnce; @@ -117,14 +122,26 @@ class ActiveProfileBinder { void start() { if (_started) return; _started = true; + // Flip `isBinding` before anything else: callers navigate right after + // start(), and screens (DiscoverScreen's no-servers gate) read the flag + // synchronously during their first build. Deferring the mark to the + // microtask below leaves a started-but-flag-false gap in which the gate + // throws "No servers available" on fresh login. Marking before + // addListener keeps the binder's own listener from reacting to this + // notification — the microtask stays the single initial-rebind entry. + activeProfile.markBindingStarted(); activeProfile.addListener(_onActiveProfileChanged); - // Defer the first rebind: _runRebindOnce calls markBindingStarted → - // notifyListeners on ActiveProfileProvider, and start() is invoked from - // Provider's create callback during the build phase. - // Notifying synchronously there re-enters the widget tree before this - // provider's value has been assigned and crashes the inspector. + // Callers invoke start() from async contexts after the offline decision + // has been made (SetupScreen, MainScreen post-frame, AuthScreen). The + // microtask keeps the initial rebind — and any PIN prompt it pops — out + // of the caller's current frame. scheduleMicrotask(() { - if (!_started) return; + if (!_started) { + // Disposed before the rebind could run — settle the flag we set + // above so awaitBindingSettle callers aren't stranded. + activeProfile.markBindingFinished(success: true); + return; + } unawaited(_rebind()); }); } @@ -196,6 +213,7 @@ class ActiveProfileBinder { Future _runRebindOnce() async { _bindingProfileId = activeProfile.activeId; activeProfile.markBindingStarted(); + final stopwatch = Stopwatch()..start(); var success = false; String? attemptedProfileId; try { @@ -272,6 +290,10 @@ class ActiveProfileBinder { } else if (_lastBoundProfileId == attemptedProfileId) { _lastBoundProfileId = null; } + appLogger.i( + 'ActiveProfileBinder: rebind settled', + error: {'profileId': attemptedProfileId, 'success': success, 'elapsedMs': stopwatch.elapsedMilliseconds}, + ); activeProfile.markBindingFinished(success: success); _bindingProfileId = null; } @@ -337,10 +359,27 @@ class ActiveProfileBinder { 'uuid=$homeUuid, useCache=$useCache, preVerified=$preVerified): ${cachedToken == null ? (useCache ? "MISS" : "BYPASS") : "HIT"}', ); if (cachedToken != null) { + // Fire the resource refresh and the optimistic cached-metadata connect + // together: the plex.tv round-trip no longer gates server probing on + // cold start. The reconcile applies whatever the refresh learns + // (rotated tokens, changed URIs, membership) once it lands. + final fetchOutcome = _settleServerFetch(_fetchServersTimed(auth, cachedToken, profile.displayName)); + final optimistic = await _bindOptimisticallyFromCache( + account: account, + userToken: cachedToken, + profileId: profile.id, + profileLabel: profile.displayName, + fetchOutcome: fetchOutcome, + onAuthRejected: () => profileConnections.recordToken(profile.id, parentId, ''), + ); + if (optimistic != null && optimistic.visibleServerIds.isNotEmpty) { + return optimistic; + } try { - final servers = await auth.fetchServers(cachedToken); + final servers = await _unwrapServerFetch(fetchOutcome); if (servers.isNotEmpty) { appLogger.i('ActiveProfileBinder: using cached token for ${profile.displayName} (${servers.length} servers)'); + unawaited(_persistRefreshedServers(account, servers)); return _connectFromServers(account, cachedToken, servers, profile.displayName); } appLogger.w( @@ -359,6 +398,9 @@ class ActiveProfileBinder { error: e, ); if (e.isTransient) { + // The optimistic pass already probed the cached metadata — + // don't burn another race on the same endpoints. + if (optimistic != null) return optimistic; return _connectFromCachedServers(account, cachedToken, profile.displayName, error: e); } return const _ProfileBindResult.empty(); @@ -461,8 +503,24 @@ class ActiveProfileBinder { List? servers; if (userToken != null && userToken.isNotEmpty) { + final cachedUserToken = userToken; + // Same optimistic shape as the plex_home cached path: probe cached + // metadata while the resource refresh runs alongside. + final fetchOutcome = _settleServerFetch(_fetchServersTimed(auth, cachedUserToken, profile.displayName)); + final optimistic = await _bindOptimisticallyFromCache( + account: conn, + userToken: cachedUserToken, + profileId: profile.id, + profileLabel: profile.displayName, + fetchOutcome: fetchOutcome, + onAuthRejected: () => profileConnections.recordToken(profile.id, conn.id, ''), + ); + if (optimistic != null && optimistic.visibleServerIds.isNotEmpty) { + await profileConnections.markUsed(profile.id, conn.id); + return optimistic; + } try { - servers = await auth.fetchServers(userToken); + servers = await _unwrapServerFetch(fetchOutcome); } on MediaServerHttpException catch (e) { if (e.statusCode == 401 || e.statusCode == 403) { appLogger.w( @@ -473,7 +531,9 @@ class ActiveProfileBinder { } else { appLogger.w('ActiveProfileBinder: fetchServers failed for ${profile.displayName}', error: e); if (e.isTransient) { - final ids = await _connectFromCachedServers(conn, userToken, profile.displayName, error: e); + // The optimistic pass already probed the cached metadata. + if (optimistic != null) return optimistic; + final ids = await _connectFromCachedServers(conn, cachedUserToken, profile.displayName, error: e); if (ids.visibleServerIds.isNotEmpty) await profileConnections.markUsed(profile.id, conn.id); return ids; } @@ -494,7 +554,7 @@ class ActiveProfileBinder { if (minted == null) return const _ProfileBindResult.empty(); userToken = minted; try { - servers = await auth.fetchServers(userToken); + servers = await _fetchServersTimed(auth, userToken, profile.displayName); } on MediaServerHttpException catch (e) { appLogger.w('ActiveProfileBinder: fetchServers failed for ${profile.displayName}', error: e); if (e.statusCode == 401 || e.statusCode == 403) { @@ -514,6 +574,9 @@ class ActiveProfileBinder { } } + if (servers != null && servers.isNotEmpty) { + unawaited(_persistRefreshedServers(conn, servers)); + } final ids = await _connectFromServers(conn, userToken, servers ?? const [], profile.displayName); await profileConnections.markUsed(profile.id, conn.id); return ids; @@ -549,7 +612,7 @@ class ActiveProfileBinder { final auth = await _ensureAuth(); final List servers; try { - servers = await auth.fetchServers(userToken); + servers = await _fetchServersTimed(auth, userToken, profileLabel); } on MediaServerHttpException catch (e, st) { appLogger.w('ActiveProfileBinder: fetchServers failed for $profileLabel', error: e, stackTrace: st); if (e.statusCode == 401 || e.statusCode == 403) { @@ -565,6 +628,9 @@ class ActiveProfileBinder { appLogger.w('ActiveProfileBinder: fetchServers failed for $profileLabel', error: e, stackTrace: st); return const _ProfileBindResult.empty(); } + if (servers.isNotEmpty) { + unawaited(_persistRefreshedServers(account, servers)); + } return _connectFromServers(account, userToken, servers, profileLabel); } @@ -595,9 +661,13 @@ class ActiveProfileBinder { appLogger.w('ActiveProfileBinder: no servers for $profileLabel on ${account.accountLabel}'); return const _ProfileBindResult.empty(); } + final stopwatch = Stopwatch()..start(); final updatedConn = account.copyWith(servers: servers); final boundIds = await serverManager.refreshTokensForProfile(updatedConn); - appLogger.i('ActiveProfileBinder: bound ${boundIds.length}/${servers.length} Plex servers for $profileLabel'); + appLogger.i( + 'ActiveProfileBinder: bound ${boundIds.length}/${servers.length} Plex servers for $profileLabel', + error: {'elapsedMs': stopwatch.elapsedMilliseconds}, + ); // Return only the ids that actually connected — the visibility filter // pushed downstream must not include unreachable servers, otherwise // the UI lists them and downstream calls 404/timeout per interaction. @@ -607,6 +677,180 @@ class ActiveProfileBinder { ); } + /// Run [PlexAuthService.fetchServers] with a timing log so cold-start + /// slowness is attributable from logs. Throws through to the caller. + Future> _fetchServersTimed(PlexAuthService auth, String token, String profileLabel) async { + final stopwatch = Stopwatch()..start(); + try { + final servers = await auth.fetchServers(token); + appLogger.i( + 'ActiveProfileBinder: resource refresh completed for $profileLabel', + error: {'servers': servers.length, 'elapsedMs': stopwatch.elapsedMilliseconds}, + ); + return servers; + } catch (_) { + appLogger.d( + 'ActiveProfileBinder: resource refresh failed for $profileLabel', + error: {'elapsedMs': stopwatch.elapsedMilliseconds}, + ); + rethrow; + } + } + + /// Capture a fetch's outcome as a value so the resource refresh can run + /// alongside the optimistic cached bind — an early failure must not + /// surface as an unhandled async error while nothing is awaiting it yet. + Future<_FetchOutcome> _settleServerFetch(Future> fetch) { + return fetch.then<_FetchOutcome>( + (servers) => (servers: servers, error: null, stackTrace: null), + onError: (Object error, StackTrace stackTrace) => (servers: null, error: error, stackTrace: stackTrace), + ); + } + + /// Rethrow a settled fetch with its original error/stack so existing + /// `on MediaServerHttpException` handlers keep working unchanged. + Future> _unwrapServerFetch(Future<_FetchOutcome> outcome) async { + final settled = await outcome; + final error = settled.error; + if (error != null) { + Error.throwWithStackTrace(error, settled.stackTrace ?? StackTrace.current); + } + return settled.servers!; + } + + /// Persist a freshly fetched resource list onto the stored account row so + /// later cold starts (and the cached-metadata fallbacks) work from current + /// URIs instead of the sign-in-day snapshot. Best-effort. + Future _persistRefreshedServers(PlexAccountConnection account, List servers) async { + try { + await connections.upsert(account.copyWith(servers: servers)); + } catch (e, st) { + appLogger.w( + 'ActiveProfileBinder: failed to persist refreshed servers for ${account.accountLabel}', + error: e, + stackTrace: st, + ); + } + } + + /// Cold-start fast path for cached-token binds: connect from the cached + /// server metadata immediately while the plex.tv resource refresh + /// ([fetchOutcome]) runs alongside, then reconcile in the background once + /// it lands. Returns `null` when there is no cached metadata to connect + /// from, and a 0-bound result when every cached endpoint was unreachable — + /// callers fall back to awaiting the fetch in both cases. + /// + /// Trade-off: when the cached metadata is entirely stale (every URI + /// changed since last launch), the failed optimistic pass delays the + /// fresh connect by up to the race budget. The reconcile persists fresh + /// metadata so the next launch recovers. + Future<_ProfileBindResult?> _bindOptimisticallyFromCache({ + required PlexAccountConnection account, + required String userToken, + required String profileId, + required String profileLabel, + required Future<_FetchOutcome> fetchOutcome, + required Future Function() onAuthRejected, + }) async { + if (account.servers.isEmpty) return null; + appLogger.i( + 'ActiveProfileBinder: connecting $profileLabel from cached server metadata while resources refresh', + error: {'servers': account.servers.length}, + ); + final cachedServers = account.servers.map((server) => server.withAccessToken(userToken)).toList(growable: false); + final result = await _connectFromServers(account, userToken, cachedServers, profileLabel); + if (result.visibleServerIds.isEmpty) return result; + _reconcileWhenFetchLands( + fetchOutcome: fetchOutcome, + account: account, + profileId: profileId, + profileLabel: profileLabel, + onAuthRejected: onAuthRejected, + ); + return result; + } + + /// Apply the background resource refresh after an optimistic cached bind: + /// persist fresh metadata, rotate per-server tokens in place, retry servers + /// the optimistic pass left offline, and pick up membership changes. + /// + /// Waits for the in-flight rebind to settle first (applying mid-rebind + /// would race the visibility sweep in [_runRebindOnce]) and no-ops when the + /// active profile has changed — applying a stale profile's clients would + /// leak its servers into another profile's session. + void _reconcileWhenFetchLands({ + required Future<_FetchOutcome> fetchOutcome, + required PlexAccountConnection account, + required String profileId, + required String profileLabel, + required Future Function() onAuthRejected, + }) { + unawaited( + () async { + final settled = await fetchOutcome; + await activeProfile.awaitBindingSettle(); + final error = settled.error; + if (error != null) { + if (error is MediaServerHttpException && (error.statusCode == 401 || error.statusCode == 403)) { + appLogger.w( + 'ActiveProfileBinder: cached token rejected (${error.statusCode}) during background refresh ' + 'for $profileLabel — flagging re-auth', + ); + // Wipe the bad token regardless of the active profile (DB hygiene), + // but only surface the auth banner while this profile is active. + await onAuthRejected(); + if (activeProfile.activeId == profileId) { + serverManager.markPlexConnectionAuthError(account); + } + } else { + appLogger.w( + 'ActiveProfileBinder: background resource refresh failed for $profileLabel; staying on cached metadata', + error: error, + ); + } + return; + } + final fresh = settled.servers!; + if (fresh.isEmpty) { + // A previously-populated account answering with zero servers is + // almost always transient plex.tv weirdness. Re-minting from a + // background task could pop a PIN prompt out of nowhere — keep the + // cache and let the next explicit bind sort it out. + appLogger.w( + 'ActiveProfileBinder: background refresh returned 0 servers for $profileLabel; keeping cached metadata', + ); + return; + } + await _persistRefreshedServers(account, fresh); + if (activeProfile.activeId != profileId) return; + final freshIds = fresh.map((server) => server.clientIdentifier).toSet(); + final cachedIds = account.servers.map((server) => server.clientIdentifier).toSet(); + if (!setEquals(freshIds, cachedIds)) { + appLogger.i( + 'ActiveProfileBinder: server membership changed for $profileLabel — rebinding', + error: {'cached': cachedIds.length, 'fresh': freshIds.length}, + ); + // The refresh just round-trip-validated this profile's token; let + // the rebind reuse the cache instead of re-prompting through /switch. + markPlexHomePreVerified(profileId); + await rebindIfActive(profileId); + return; + } + // Same membership: rotate tokens/URIs in place and retry anything the + // optimistic pass left offline. Newly-online expected servers are + // promoted into the visibility filter by MultiServerProvider when the + // status emission this triggers lands. + await serverManager.refreshTokensForProfile(account.copyWith(servers: fresh)); + }().catchError((Object error, StackTrace stackTrace) { + appLogger.w( + 'ActiveProfileBinder: background reconcile failed for $profileLabel', + error: error, + stackTrace: stackTrace, + ); + }), + ); + } + Future<_ProfileBindResult> _bindJellyfin(JellyfinConnection conn) async { final ok = await serverManager.addJellyfinConnection(conn); // `addJellyfinConnection` registers the client even when the health probe diff --git a/lib/screens/auth_screen.dart b/lib/screens/auth_screen.dart index 33b6b00c..26d2f6b7 100644 --- a/lib/screens/auth_screen.dart +++ b/lib/screens/auth_screen.dart @@ -5,6 +5,7 @@ import 'package:provider/provider.dart'; import '../connection/connection.dart'; import '../connection/connection_registry.dart'; import '../mixins/controller_disposer_mixin.dart'; +import '../profiles/active_profile_binder.dart'; import '../profiles/active_profile_provider.dart'; import '../profiles/plex_home_service.dart'; import '../profiles/profile.dart'; @@ -139,6 +140,14 @@ class _AuthScreenState extends State { if (!mounted) return; + // Start the binder before the picker/MainScreen, mirroring the + // cold-start SetupScreen ordering. On a fresh install SetupScreen + // routes here without ever starting it, so without this the profile + // activated above is bound only by MainScreen's post-frame start() — + // Discover renders a "No servers available" flash in the gap, and the + // picker's awaitBindingSettle resolves before anything is bound. + context.read().start(); + final settings = await SettingsService.getInstance(); if (!mounted) return; diff --git a/lib/services/multi_server_manager.dart b/lib/services/multi_server_manager.dart index 4d27d557..bc0a4755 100644 --- a/lib/services/multi_server_manager.dart +++ b/lib/services/multi_server_manager.dart @@ -43,6 +43,15 @@ class MultiServerManager { Stream> get statusStream => _statusController.stream; + /// Per-server connect progress during a bind. Unlike [statusStream] — whose + /// first emission means "the binder's first connect pass finished" and which + /// triggers libraries/live-tv work per emission — this fires as each + /// individual server lands so the startup splash can flip its checkmarks + /// incrementally without disturbing those contracts. + final _connectProgressController = StreamController<({String serverId, bool online})>.broadcast(); + + Stream<({String serverId, bool online})> get connectProgressStream => _connectProgressController.stream; + /// Servers whose authentication has failed (token rejected). A re-auth flow /// should be offered for these — they will remain "offline" until the user /// signs in again. Cleared once a probe succeeds. @@ -213,6 +222,7 @@ class MultiServerManager { /// creating config, and building client with failover support. Future _createClientForServer({required PlexServer server, required String clientIdentifier}) async { final serverId = server.clientIdentifier; + final stopwatch = Stopwatch()..start(); // Get storage and load cached endpoint for this server final storage = await StorageService.getInstance(); @@ -238,6 +248,7 @@ class MultiServerManager { final workingConnection = streamIterator.current; final baseUrl = workingConnection.uri; + final firstConnectionMs = stopwatch.elapsedMilliseconds; // Create PlexClient with failover support final prioritizedEndpoints = server.prioritizedEndpointUrls(preferredFirst: baseUrl); @@ -263,6 +274,16 @@ class MultiServerManager { // Save the initial endpoint await storage.saveServerEndpoint(ServerId(serverId), baseUrl); + appLogger.i( + 'Connected ${server.name}', + error: { + 'uri': baseUrl, + 'hadCachedEndpoint': cachedEndpoint != null, + 'firstConnectionMs': firstConnectionMs, + 'totalMs': stopwatch.elapsedMilliseconds, + }, + ); + // Drain remaining stream values in background to apply better connections _drainOptimizationStream(streamIterator, client: client, server: server, storage: storage); @@ -442,6 +463,7 @@ class MultiServerManager { _authErrorServers.remove(serverId); _serverStatus[serverId] = true; bound.add(serverId); + _connectProgressController.add((serverId: serverId, online: true)); return; } try { @@ -455,9 +477,11 @@ class MultiServerManager { _serverStatus[serverId] = true; _authErrorServers.remove(serverId); bound.add(serverId); + _connectProgressController.add((serverId: serverId, online: true)); } catch (e, stackTrace) { appLogger.e('refreshTokensForProfile: failed to connect ${server.name}', error: e, stackTrace: stackTrace); _serverStatus[serverId] = false; + _connectProgressController.add((serverId: serverId, online: false)); } }); await Future.wait(futures); @@ -1025,5 +1049,8 @@ class MultiServerManager { if (!_statusController.isClosed) { _statusController.close(); } + if (!_connectProgressController.isClosed) { + _connectProgressController.close(); + } } } diff --git a/lib/services/plex_auth_service.dart b/lib/services/plex_auth_service.dart index 68fb77e4..be61fb34 100644 --- a/lib/services/plex_auth_service.dart +++ b/lib/services/plex_auth_service.dart @@ -487,24 +487,32 @@ class PlexServer { )) { if (selection.phase == EndpointRacePhase.first) { final firstCandidate = selection.candidate; + // Emit the winner immediately — the HTTPS upgrade probe (up to a full + // race timeout when the HTTPS variant is dead) must not gate startup. + // The StreamIterator consumer pulls lazily, so the upgrade below runs + // off the critical path and lands via the same background-promotion + // drain that applies Phase 2 results. + firstConnection = _updateConnectionUrl(firstCandidate.connection, firstCandidate.url); + yield firstConnection; + appLogger.d( + 'Emitted first working connection, continuing latency tests in background', + error: {'uri': firstConnection.uri}, + ); + final upgradedFirstCandidate = await _upgradeCandidateToHttpsIfPossible( firstCandidate, clientIdentifier: clientIdentifier, ); - final emitCandidate = upgradedFirstCandidate ?? firstCandidate; - - firstConnection = _updateConnectionUrl(emitCandidate.connection, emitCandidate.url); - yield firstConnection; if (upgradedFirstCandidate != null && upgradedFirstCandidate.url != firstCandidate.url) { appLogger.i( 'Phase 1 winner upgraded to HTTPS', error: {'from': firstCandidate.url, 'to': upgradedFirstCandidate.url}, ); + // Track the upgrade as the effective first connection so the Phase 2 + // dedup below doesn't re-emit the same HTTPS endpoint as "better". + firstConnection = _updateConnectionUrl(upgradedFirstCandidate.connection, upgradedFirstCandidate.url); + yield firstConnection; } - appLogger.d( - 'Emitted first working connection, continuing latency tests in background', - error: {'uri': firstConnection.uri}, - ); continue; } diff --git a/lib/utils/endpoint_race.dart b/lib/utils/endpoint_race.dart index 029f4beb..a2664d0e 100644 --- a/lib/utils/endpoint_race.dart +++ b/lib/utils/endpoint_race.dart @@ -23,10 +23,15 @@ class EndpointRaceSelection { /// Shared two-phase endpoint discovery used by Plex and Jellyfin. /// -/// Phase 1 emits the first reachable endpoint quickly, using a cached/preferred -/// endpoint first when available. Phase 2 measures all candidates and emits the -/// selector's best endpoint, letting callers promote a lower-latency URL in the -/// background without blocking initial connection setup. +/// Phase 1 emits the first reachable endpoint quickly. A cached/preferred +/// endpoint is probed first and wins deterministically when it answers within +/// [preferredHeadStart]; otherwise the full candidate race starts with the +/// still-pending cached probe merged in as a participant, so a stale cached +/// endpoint (e.g. a LAN address probed from outside the LAN) costs the head +/// start instead of the full [preferredTimeout] serially. Phase 2 measures all +/// candidates and emits the selector's best endpoint, letting callers promote +/// a lower-latency URL in the background without blocking initial connection +/// setup. Stream> raceEndpointCandidates({ required String label, required List candidates, @@ -41,6 +46,7 @@ Stream> raceEndpointCandidates({ required C? Function(Map successfulResults) selectBestCandidate, void Function(C candidate, R result)? onFirstSuccess, Duration preferredTimeout = MediaServerTimeouts.preferredEndpointProbe, + Duration preferredHeadStart = MediaServerTimeouts.preferredEndpointHeadStart, Duration raceTimeout = MediaServerTimeouts.connectionRace, }) async* { if (candidates.isEmpty) { @@ -48,50 +54,86 @@ Stream> raceEndpointCandidates({ return; } + final stopwatch = Stopwatch()..start(); C? firstCandidate; R? firstResult; var fromPreferred = false; + C? cachedCandidate; + Future? pendingCachedProbe; if (preferredUrl != null && preferredUrl.isNotEmpty) { - final cachedCandidate = candidateForUrl?.call(preferredUrl) ?? _candidateForUrl(candidates, urlOf, preferredUrl); - if (cachedCandidate != null) { - appLogger.d('Testing cached $label endpoint before running full race', error: {'uri': preferredUrl}); - final result = await probe(cachedCandidate, preferredTimeout); + cachedCandidate = candidateForUrl?.call(preferredUrl) ?? _candidateForUrl(candidates, urlOf, preferredUrl); + } + if (cachedCandidate != null) { + final cached = cachedCandidate; + appLogger.d('Testing cached $label endpoint with a head start on the race', error: {'uri': preferredUrl}); + final cachedProbe = probe(cached, preferredTimeout); + final headStartResult = await Future.any([cachedProbe, Future.delayed(preferredHeadStart, () => null)]); - if (isSuccess(result)) { - appLogger.i('Cached $label endpoint succeeded, using immediately', error: {'uri': preferredUrl}); - firstCandidate = cachedCandidate; - firstResult = result; - fromPreferred = true; - onFirstSuccess?.call(cachedCandidate, result); - } else { - appLogger.w('Cached $label endpoint failed, falling back to candidate race', error: {'uri': preferredUrl}); - } + if (headStartResult != null && isSuccess(headStartResult)) { + appLogger.i( + 'Cached $label endpoint succeeded, using immediately', + error: {'uri': preferredUrl, 'elapsedMs': stopwatch.elapsedMilliseconds}, + ); + firstCandidate = cached; + firstResult = headStartResult; + fromPreferred = true; + onFirstSuccess?.call(cached, headStartResult); + } else if (headStartResult != null) { + // Failed within the head start (e.g. connection refused) — run the + // plain race; the cached URL is among the candidates and gets a fresh + // probe like any other. + appLogger.w( + 'Cached $label endpoint failed, falling back to candidate race', + error: {'uri': preferredUrl, 'elapsedMs': stopwatch.elapsedMilliseconds}, + ); + } else { + appLogger.d( + 'Cached $label endpoint still pending after head start, racing all candidates', + error: {'uri': preferredUrl}, + ); + pendingCachedProbe = cachedProbe; } } if (firstCandidate == null || firstResult == null) { + // When the cached probe is still in flight, merge it into the race as a + // participant (reusing its future) instead of probing the same URL twice. + final mergedCached = pendingCachedProbe != null ? cachedCandidate : null; + final raceCandidates = mergedCached == null + ? candidates + : [mergedCached, ...candidates.where((c) => urlOf(c) != preferredUrl)]; final first = await _raceFirstSuccess( label: label, - candidates: candidates, + candidates: raceCandidates, urlOf: urlOf, displayTypeOf: displayTypeOf, failureLogFields: failureLogFields, - probe: probe, + probe: (candidate, timeout) => + mergedCached != null && identical(candidate, mergedCached) ? pendingCachedProbe! : probe(candidate, timeout), isSuccess: isSuccess, onFirstSuccess: onFirstSuccess, timeout: raceTimeout, ); if (first == null) { - appLogger.e('No working $label endpoints after race', error: {'candidateCount': candidates.length}); + appLogger.e( + 'No working $label endpoints after race', + error: {'candidateCount': raceCandidates.length, 'elapsedMs': stopwatch.elapsedMilliseconds}, + ); return; } - appLogger.i( - '$label race found first working endpoint', - error: {'uri': urlOf(first.candidate), 'type': displayTypeOf?.call(first.candidate)}, - ); firstCandidate = first.candidate; firstResult = first.result; + fromPreferred = mergedCached != null && identical(firstCandidate, mergedCached); + appLogger.i( + '$label race found first working endpoint', + error: { + 'uri': urlOf(first.candidate), + 'type': displayTypeOf?.call(first.candidate), + 'fromPreferred': fromPreferred, + 'elapsedMs': stopwatch.elapsedMilliseconds, + }, + ); } final resolvedFirstCandidate = firstCandidate; @@ -123,7 +165,7 @@ Stream> raceEndpointCandidates({ appLogger.d( 'Completed latency sweep for $label endpoints', - error: {'successfulCandidates': successfulResults.length}, + error: {'successfulCandidates': successfulResults.length, 'elapsedMs': stopwatch.elapsedMilliseconds}, ); final bestCandidate = selectBestCandidate(successfulResults); diff --git a/lib/utils/media_server_timeouts.dart b/lib/utils/media_server_timeouts.dart index d9cf7930..0b131f29 100644 --- a/lib/utils/media_server_timeouts.dart +++ b/lib/utils/media_server_timeouts.dart @@ -15,17 +15,26 @@ class MediaServerTimeouts { /// can be slower than the top-level home hub call on remote Plex servers. static const libraryHubAttemptTimeouts = [Duration(seconds: 10), Duration(seconds: 8), Duration(seconds: 5)]; - /// Timeout for probing a cached/preferred endpoint before falling back to - /// the full candidate race (used in [PlexServer.findBestWorkingConnection]). + /// Timeout for probing a cached/preferred endpoint (used in + /// [PlexServer.findBestWorkingConnection]). static const preferredEndpointProbe = Duration(milliseconds: 1500); + /// How long the cached/preferred endpoint probe gets to answer before the + /// full candidate race starts alongside it. A healthy cached endpoint + /// answers well inside this window and wins deterministically; a stale one + /// (e.g. a cached LAN address probed from outside the LAN) only delays + /// discovery by this much instead of the full [preferredEndpointProbe]. + static const preferredEndpointHeadStart = Duration(milliseconds: 300); + /// Timeout for the connection race where all candidates are tested in /// parallel (used in [PlexServer.findBestWorkingConnection]). static const connectionRace = Duration(seconds: 2); - /// Per-server connection budget: preferred probe + race + HTTPS upgrade - /// attempt + 1s buffer. - static const perServerConnect = Duration(milliseconds: 1500 + 2000 + 2000 + 1000); + /// Per-server connection watchdog ceiling. The discovery path is no longer + /// strictly serial (cached probe overlaps the race; the HTTPS upgrade runs + /// off the critical path), so this is a generous upper bound rather than a + /// sum of phases. + static const perServerConnect = Duration(milliseconds: 6500); /// HTTP timeout for the live-TV tune POST. Matches Plex web's value — the /// default 10s connect budget is too tight on Fire-TV cold starts. diff --git a/test/profiles/active_profile_binder_test.dart b/test/profiles/active_profile_binder_test.dart index 5e7576bc..4328ff00 100644 --- a/test/profiles/active_profile_binder_test.dart +++ b/test/profiles/active_profile_binder_test.dart @@ -27,6 +27,18 @@ import 'package:plezy/utils/media_server_timeouts.dart'; import '../test_helpers/prefs.dart'; +/// Poll [condition] until it holds, failing after [timeout]. Used to observe +/// the binder's unawaited background reconcile settling. +Future pumpUntil(Future Function() condition, {Duration timeout = const Duration(seconds: 2)}) async { + final deadline = DateTime.now().add(timeout); + while (!await condition()) { + if (DateTime.now().isAfter(deadline)) { + fail('condition not met within $timeout'); + } + await Future.delayed(const Duration(milliseconds: 10)); + } +} + void main() { late AppDatabase db; late ConnectionRegistry connections; @@ -123,6 +135,33 @@ void main() { expect(notifications, lessThan(8)); }); + test('start() marks binding synchronously so first-frame readers see it', () async { + await createActiveLocalProfile('local-sync-start'); + + binder.start(); + + // Read with no await in between — mirrors DiscoverScreen's no-servers + // gate running during the first build right after a fresh login. + expect(activeProfile.isBinding, isTrue); + + final succeeded = await activeProfile.awaitBindingSettle(); + expect(succeeded, isTrue); + expect(activeProfile.isBinding, isFalse); + expect(binder.debugLastBoundProfileId, 'local-sync-start'); + }); + + test('disposing before the initial rebind microtask clears the binding flag', () async { + await createActiveLocalProfile('local-dispose-early'); + + binder.start(); + expect(activeProfile.isBinding, isTrue); + binder.dispose(); + + // Let the scheduled microtask observe the disposal. + await Future.delayed(Duration.zero); + expect(activeProfile.isBinding, isFalse); + }); + test('initial bind can be deferred until profile selection', () async { final profile = await createActiveLocalProfile('local-deferred'); shouldDeferInitialBind = true; @@ -195,7 +234,9 @@ void main() { await binder.rebindActive(); expect(activeProfile.lastBindingSucceeded, isFalse); - expect(failingManager.refreshCalls, 1); + // Two connect passes: the optimistic cached-metadata pass binds nothing, + // so the bind falls back to the freshly fetched resource list. + expect(failingManager.refreshCalls, 2); expect(multiServerProvider.serverIds, isEmpty); expect(multiServerProvider.expectedServerIds, ['srv-1']); }); @@ -375,7 +416,7 @@ void main() { expect(prepared.manager.lastConnection?.servers.single.clientIdentifier, 'srv-1'); }); - test('does not use cached server metadata after cached token auth failure', () async { + test('binds from cache when plex.tv rejects the token, then flags re-auth from the reconcile', () async { final prepared = await preparePlexHomeBind( protected: true, httpClient: MockClient((request) async { @@ -385,10 +426,89 @@ void main() { await binder.rebindActive(); - expect(activeProfile.lastBindingSucceeded, isFalse); - expect(prepared.manager.refreshCalls, 0); - final pc = await profileConnections.get(prepared.profileId, 'plex.account'); - expect(pc?.userToken, isNull); + // The optimistic cached bind settles first — content stays available + // (mirrors the cold-start cache policy: the cached token was valid + // when minted; the splash must not block on plex.tv's verdict). + expect(activeProfile.lastBindingSucceeded, isTrue); + expect(prepared.manager.refreshCalls, 1); + expect(prepared.manager.lastConnection?.servers.single.accessToken, 'home-user-token'); + + // The background reconcile sees the 401: wipes the cached token and + // flags the account for re-auth — no silent /switch re-mint that + // could pop a PIN prompt from a background task. + await pumpUntil(() async { + final pc = await profileConnections.get(prepared.profileId, 'plex.account'); + return pc?.userToken == null; + }); + await pumpUntil(() async => manager.authErrorServerIds.contains('srv-1')); + // No second connect pass was attempted with the rejected token. + expect(prepared.manager.refreshCalls, 1); + }); + + test('optimistic cached bind settles without waiting for the resource refresh, then reconciles', () async { + final fetchGate = Completer(); + final prepared = await preparePlexHomeBind( + protected: false, + httpClient: MockClient((request) async { + await fetchGate.future; + return http.Response(jsonEncode([_serverJson()]), 200, headers: {'content-type': 'application/json'}); + }), + ); + + // Settles while plex.tv still hasn't answered. + await binder.rebindActive().timeout(const Duration(seconds: 2)); + + expect(activeProfile.lastBindingSucceeded, isTrue); + expect(binder.debugLastBoundProfileId, prepared.profileId); + expect(prepared.manager.refreshCalls, 1); + expect(prepared.manager.lastConnection?.servers.single.accessToken, 'home-user-token'); + + fetchGate.complete(); + + // Same membership → the reconcile rotates in the freshly fetched + // per-server tokens in place. + await pumpUntil(() async => prepared.manager.refreshCalls == 2); + expect(prepared.manager.lastConnection?.servers.single.accessToken, 'server-token'); + + // And the refreshed metadata was persisted onto the stored account row. + final account = await connections.getPlexAccount('plex.account'); + expect(account?.servers.single.accessToken, 'server-token'); + }); + + test('membership change in the background refresh triggers a full rebind', () async { + final fetchGate = Completer(); + final prepared = await preparePlexHomeBind( + protected: false, + httpClient: MockClient((request) async { + await fetchGate.future; + return http.Response( + jsonEncode([_serverJson(clientIdentifier: 'srv-2')]), + 200, + headers: {'content-type': 'application/json'}, + ); + }), + ); + + await binder.rebindActive().timeout(const Duration(seconds: 2)); + expect(prepared.manager.refreshCalls, 1); + expect(prepared.manager.lastConnection?.servers.single.clientIdentifier, 'srv-1'); + + fetchGate.complete(); + + // Fresh membership {srv-2} ≠ cached {srv-1}: the reconcile persists the + // new resource list and triggers a full rebind, which reuses the + // just-validated cached token (pre-verified — no /switch round-trip) + // and binds the new membership. + await pumpUntil(() async => prepared.manager.lastConnection?.servers.single.clientIdentifier == 'srv-2'); + final account = await connections.getPlexAccount('plex.account'); + expect(account?.servers.single.clientIdentifier, 'srv-2'); + expect(binder.debugLastBoundProfileId, prepared.profileId); + + // The rebind's own reconcile sees identical membership and converges + // with one final in-place token pass — no rebind loop. + await pumpUntil(() async => prepared.manager.refreshCalls >= 3); + await Future.delayed(const Duration(milliseconds: 50)); + expect(prepared.manager.refreshCalls, 3); }); }); } @@ -414,9 +534,9 @@ PlexServer _server({required String accessToken}) { ); } -Map _serverJson() => { +Map _serverJson({String clientIdentifier = 'srv-1'}) => { 'name': 'Home Server', - 'clientIdentifier': 'srv-1', + 'clientIdentifier': clientIdentifier, 'accessToken': 'server-token', 'owned': true, 'provides': 'server', diff --git a/test/services/multi_server_manager_progress_test.dart b/test/services/multi_server_manager_progress_test.dart new file mode 100644 index 00000000..d43936f6 --- /dev/null +++ b/test/services/multi_server_manager_progress_test.dart @@ -0,0 +1,95 @@ +import 'package:drift/native.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:plezy/connection/connection.dart'; +import 'package:plezy/database/app_database.dart'; +import 'package:plezy/media/ids.dart'; +import 'package:plezy/models/plex/plex_config.dart'; +import 'package:plezy/services/plex_api_cache.dart'; +import 'package:plezy/services/plex_auth_service.dart'; +import 'package:plezy/services/plex_client.dart'; +import 'package:plezy/services/multi_server_manager.dart'; + +void main() { + // refreshTokensForProfile starts connectivity monitoring after a successful + // bind, which touches platform channels. + TestWidgetsFlutterBinding.ensureInitialized(); + + test('refreshTokensForProfile emits per-server progress and a single status snapshot', () async { + final db = AppDatabase.forTesting(NativeDatabase.memory()); + PlexApiCache.initialize(db); + addTearDown(db.close); + + final manager = MultiServerManager(); + addTearDown(manager.dispose); + + PlexClient buildClient(String serverId) => PlexClient.forTesting( + config: PlexConfig( + baseUrl: 'http://$serverId:32400', + token: 'old-token', + clientIdentifier: 'client-id', + product: 'Plezy', + version: 'test', + ), + serverId: ServerId(serverId), + serverName: serverId, + httpClient: MockClient((_) async => http.Response('{}', 200, headers: {'content-type': 'application/json'})), + ); + + // Both servers already registered and online — refreshTokensForProfile + // takes the in-place token-rotation fast path for them. + manager.debugRegisterClientForTesting(buildClient('srv-1')); + manager.debugRegisterClientForTesting(buildClient('srv-2')); + + final progress = <({String serverId, bool online})>[]; + final statusEmissions = >[]; + final progressSub = manager.connectProgressStream.listen(progress.add); + final statusSub = manager.statusStream.listen(statusEmissions.add); + addTearDown(progressSub.cancel); + addTearDown(statusSub.cancel); + + final connection = PlexAccountConnection( + id: 'plex.account', + accountToken: 'account-token', + clientIdentifier: 'client-id', + accountLabel: 'Owner', + servers: [_server('srv-1'), _server('srv-2')], + createdAt: DateTime(2026, 1, 1), + ); + + final bound = await manager.refreshTokensForProfile(connection); + // Let the broadcast stream deliver its pending events. + await Future.delayed(Duration.zero); + + expect(bound, {'srv-1', 'srv-2'}); + // One progress event per server, as each settles… + expect(progress.map((p) => p.serverId).toSet(), {'srv-1', 'srv-2'}); + expect(progress.every((p) => p.online), isTrue); + // …but the status snapshot keeps its one-emission-per-pass contract + // (OfflineModeProvider treats the first emission as "first connect + // pass finished"). + expect(statusEmissions, hasLength(1)); + }); +} + +PlexServer _server(String id) { + return PlexServer( + name: id, + clientIdentifier: id, + accessToken: 'new-token', + connections: [ + PlexConnection( + protocol: 'http', + address: '192.168.1.10', + port: 32400, + uri: 'http://192.168.1.10:32400', + local: true, + relay: false, + ipv6: false, + ), + ], + owned: true, + presence: true, + ); +} diff --git a/test/utils/endpoint_race_test.dart b/test/utils/endpoint_race_test.dart new file mode 100644 index 00000000..73635356 --- /dev/null +++ b/test/utils/endpoint_race_test.dart @@ -0,0 +1,174 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:plezy/utils/endpoint_race.dart'; + +typedef _Result = ({String url, bool ok}); + +void main() { + const headStart = Duration(milliseconds: 60); + + Stream> race({ + required List candidates, + String? preferred, + required Future<_Result> Function(String url) probe, + Future<_Result> Function(String url)? measure, + String? Function(Map results)? selectBest, + }) { + return raceEndpointCandidates( + label: 'test', + candidates: candidates, + urlOf: (c) => c, + preferredUrl: preferred, + probe: (c, _) => probe(c), + measure: measure ?? (c) async => (url: c, ok: false), + isSuccess: (r) => r.ok, + selectBestCandidate: selectBest ?? (results) => results.keys.first, + preferredTimeout: const Duration(milliseconds: 500), + preferredHeadStart: headStart, + raceTimeout: const Duration(milliseconds: 500), + ); + } + + Future<_Result> resultAfter(String url, Duration delay, {required bool ok}) async { + await Future.delayed(delay); + return (url: url, ok: ok); + } + + test('healthy cached endpoint wins within the head start without racing', () async { + final probeCounts = {}; + final selections = await race( + candidates: ['a', 'cached'], + preferred: 'cached', + probe: (url) { + probeCounts[url] = (probeCounts[url] ?? 0) + 1; + return resultAfter(url, const Duration(milliseconds: 10), ok: true); + }, + ).toList(); + + expect(selections.first.phase, EndpointRacePhase.first); + expect(selections.first.candidate, 'cached'); + expect(selections.first.fromPreferred, isTrue); + expect(probeCounts['cached'], 1); + // The race never started; only the phase-2 measure touches other URLs. + expect(probeCounts.containsKey('a'), isFalse); + }); + + test('stale-slow cached endpoint overlaps the race instead of serially blocking it', () async { + final probeCounts = {}; + final stopwatch = Stopwatch()..start(); + final firstTimes = []; + final selections = >[]; + await for (final selection in race( + candidates: ['fast', 'cached'], + preferred: 'cached', + probe: (url) { + probeCounts[url] = (probeCounts[url] ?? 0) + 1; + return resultAfter( + url, + url == 'cached' ? const Duration(milliseconds: 250) : const Duration(milliseconds: 10), + ok: url != 'cached', + ); + }, + )) { + selections.add(selection); + firstTimes.add(stopwatch.elapsedMilliseconds); + } + + expect(selections.first.candidate, 'fast'); + expect(selections.first.fromPreferred, isFalse); + // Emitted shortly after the head start — not after the cached probe's + // full budget (the pre-change serial behavior). + expect(firstTimes.first, lessThan(200)); + // The pending cached probe was merged into the race, not re-fired. + expect(probeCounts['cached'], 1); + + // Let the still-pending cached probe finish inside the test body. + await Future.delayed(const Duration(milliseconds: 300)); + }); + + test('cached endpoint that answers after the head start still wins when first', () async { + final probeCounts = {}; + final selections = await race( + candidates: ['slow', 'cached'], + preferred: 'cached', + probe: (url) { + probeCounts[url] = (probeCounts[url] ?? 0) + 1; + return resultAfter( + url, + url == 'cached' ? const Duration(milliseconds: 120) : const Duration(milliseconds: 350), + ok: true, + ); + }, + ).toList(); + + expect(selections.first.candidate, 'cached'); + expect(selections.first.fromPreferred, isTrue); + expect(probeCounts['cached'], 1); + + await Future.delayed(const Duration(milliseconds: 400)); + }); + + test('cached endpoint failing within the head start falls back to a fresh race', () async { + final probeCounts = {}; + final selections = await race( + candidates: ['cached', 'alt'], + preferred: 'cached', + probe: (url) { + probeCounts[url] = (probeCounts[url] ?? 0) + 1; + return resultAfter(url, const Duration(milliseconds: 10), ok: url == 'alt'); + }, + ).toList(); + + expect(selections.first.candidate, 'alt'); + expect(selections.first.fromPreferred, isFalse); + // Fast-fail keeps today's semantics: the cached URL re-races as a + // normal candidate (one probe up front, one inside the race). + expect(probeCounts['cached'], 2); + }); + + test('preferred URL not among candidates skips the cached probe entirely', () async { + final probeCounts = {}; + final selections = await race( + candidates: ['a', 'b'], + preferred: 'custom-url', + probe: (url) { + probeCounts[url] = (probeCounts[url] ?? 0) + 1; + return resultAfter(url, const Duration(milliseconds: 10), ok: url == 'a'); + }, + ).toList(); + + expect(selections.first.candidate, 'a'); + expect(selections.first.fromPreferred, isFalse); + expect(probeCounts.containsKey('custom-url'), isFalse); + }); + + test('emits nothing when every candidate fails', () async { + final selections = await race( + candidates: ['a', 'b'], + preferred: 'a', + probe: (url) => resultAfter(url, const Duration(milliseconds: 10), ok: false), + ).toList(); + + expect(selections, isEmpty); + }); + + test('phase 2 still promotes the selector-best endpoint', () async { + final selections = await race( + candidates: ['quick', 'better'], + probe: (url) => resultAfter( + url, + url == 'quick' ? const Duration(milliseconds: 10) : const Duration(milliseconds: 80), + ok: true, + ), + measure: (url) async => (url: url, ok: true), + selectBest: (results) => 'better', + ).toList(); + + expect(selections, hasLength(2)); + expect(selections.first.phase, EndpointRacePhase.first); + expect(selections.first.candidate, 'quick'); + expect(selections.last.phase, EndpointRacePhase.best); + expect(selections.last.candidate, 'better'); + }); +}