diff --git a/lib/main.dart b/lib/main.dart index ab0302ca..ca4bb7e2 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -664,7 +664,10 @@ class _MainAppState extends State with WidgetsBindingObserver { List? targetKeys, bool force = false, }) async { - if (_isAutoDeleteRunning) return; + if (_isAutoDeleteRunning) { + if (targetKeys != null) _pendingSyncKeys.addAll(targetKeys); + return; + } _isAutoDeleteRunning = true; try { await downloadProvider.refreshMetadataFromCache(); @@ -697,6 +700,11 @@ class _MainAppState extends State with WidgetsBindingObserver { } } finally { _isAutoDeleteRunning = false; + if (_pendingSyncKeys.isNotEmpty) { + final queuedKeys = _pendingSyncKeys.toList(); + _pendingSyncKeys.clear(); + unawaited(_autoDeleteAndSync(downloadProvider, targetKeys: queuedKeys)); + } } } diff --git a/lib/mixins/paginated_item_loader.dart b/lib/mixins/paginated_item_loader.dart index 610f6257..d1b6eada 100644 --- a/lib/mixins/paginated_item_loader.dart +++ b/lib/mixins/paginated_item_loader.dart @@ -210,6 +210,13 @@ mixin PaginatedItemLoader on State { /// Mirrors the "one item deleted on the server" invariant: decrements /// [totalSize] even if [index] wasn't in the sparse map (evicted). void removeLoadedItemAndShift(int index) { + _requestId++; + _cancelToken?.abort(); + _cancelToken = AbortController(); + _retryTimer?.cancel(); + _retryTimer = null; + _loadingRanges.clear(); + _scheduledRetry = null; loadedItems.remove(index); final shifted = {}; for (final entry in loadedItems.entries) { diff --git a/lib/mpv/player/player_native.dart b/lib/mpv/player/player_native.dart index a03896a8..fcde802d 100644 --- a/lib/mpv/player/player_native.dart +++ b/lib/mpv/player/player_native.dart @@ -174,6 +174,7 @@ class PlayerNative extends PlayerBase { // Two concurrent invokes on Android caused MpvPlayerPlugin.handleInitialize // to dispose-and-recreate the in-flight core, hanging playback (#930). Future? _initFuture; + Future _rateChangeTail = Future.value(); Future _ensureInitialized() async { if (initialized) return; @@ -554,10 +555,16 @@ class PlayerNative extends PlayerBase { } @override - Future setRate(double rate) async { - // mpv cannot scaletempo compressed (spdif) audio and silently keeps - // playing at 1x, so suspend passthrough while the rate is not 1.0. + Future setRate(double rate) { _currentRate = rate; + final operation = _rateChangeTail.then((_) => _applyRateChange(rate)); + _rateChangeTail = operation.catchError((Object _, StackTrace _) {}); + return operation; + } + + Future _applyRateChange(double rate) async { + // mpv cannot scaletempo compressed (spdif) audio and silently keeps + // playing at 1x, so serialize passthrough and speed transitions. if (_passthroughActive && rate != 1.0) { await _applyPassthrough(false); } diff --git a/lib/providers/companion_remote_provider.dart b/lib/providers/companion_remote_provider.dart index 458031b0..5a40648d 100644 --- a/lib/providers/companion_remote_provider.dart +++ b/lib/providers/companion_remote_provider.dart @@ -43,6 +43,7 @@ class CompanionRemoteProvider with ChangeNotifier, DisposableChangeNotifierMixin Timer? _reconnectTimer; int _reconnectAttempts = 0; + Future? _activeReconnect; bool _intentionalDisconnect = false; // Reconnection context (only hostAddresses and hostClientId are connection-specific) @@ -759,7 +760,18 @@ class CompanionRemoteProvider with ChangeNotifier, DisposableChangeNotifierMixin _reconnectTimer = Timer(delay, _attemptReconnect); } - Future _attemptReconnect() async { + Future _attemptReconnect() { + final active = _activeReconnect; + if (active != null) return active; + late final Future attempt; + attempt = _runReconnectAttempt().whenComplete(() { + if (identical(_activeReconnect, attempt)) _activeReconnect = null; + }); + _activeReconnect = attempt; + return attempt; + } + + Future _runReconnectAttempt() async { if (_lastHostAddresses == null || !isCryptoReady) { appLogger.w('CompanionRemote: No stored context for reconnect'); _session = _session?.copyWith( diff --git a/lib/providers/discover_provider.dart b/lib/providers/discover_provider.dart index 277800fa..6bc6b731 100644 --- a/lib/providers/discover_provider.dart +++ b/lib/providers/discover_provider.dart @@ -90,6 +90,9 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin DiscoverLoadState _hubsState = DiscoverLoadState.initial; String? _errorMessage; int _loadGeneration = 0; + int _contentRevision = 0; + Future? _continueWatchingRefreshFuture; + bool _continueWatchingRefreshQueued = false; Set _lastSeenHiddenKeys = {}; List _lastSeenLibraryOrderKeys = const []; @@ -161,6 +164,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin // listening widgets dirty mid-build. await null; if (isDisposed) return; + ++_contentRevision; appLogger.d('DiscoverProvider: loading content from all servers'); _onDeckState = DiscoverLoadState.loading; _hubsState = DiscoverLoadState.loading; @@ -262,6 +266,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin /// Failures keep the loaded state and leave the ids un-loaded, so the next /// status emission retries them. Future _loadDeltaOnce(Set serverIds) async { + ++_contentRevision; // A full pass may have covered these ids while they sat in the queue. final ids = serverIds.difference(_fullyLoadedServerIds); final onDeckIds = ids.difference(_loadedOnDeckServerIds); @@ -350,17 +355,45 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin }).toList(); } - /// Background refresh of Continue Watching only — never flips load states - /// or surfaces errors (a stale row beats an error flash), never refetches - /// hubs. - Future refreshContinueWatching() async { + /// Background refresh of Continue Watching only. Concurrent events coalesce + /// into the active request plus at most one trailing fresh request. + Future refreshContinueWatching() { + final active = _continueWatchingRefreshFuture; + if (active != null) { + _continueWatchingRefreshQueued = true; + return active; + } + late final Future refresh; + refresh = _runContinueWatchingRefreshes().whenComplete(() { + if (identical(_continueWatchingRefreshFuture, refresh)) { + _continueWatchingRefreshFuture = null; + } + }); + _continueWatchingRefreshFuture = refresh; + return refresh; + } + + Future _runContinueWatchingRefreshes() async { + do { + _continueWatchingRefreshQueued = false; + await _refreshContinueWatchingOnce(); + } while (_continueWatchingRefreshQueued && !isDisposed); + } + + Future _refreshContinueWatchingOnce() async { try { if (!_multiServer.hasConnectedServers) return; + final revision = _contentRevision; + final hiddenKeys = Set.of(_hiddenLibraries.hiddenLibraryKeys); final fetched = await _multiServer.aggregationService.getOnDeckFromAllServers( limit: _continueWatchingProbeLimit, - hiddenLibraryKeys: _hiddenLibraries.hiddenLibraryKeys, + hiddenLibraryKeys: hiddenKeys, ); if (isDisposed) return; + if (revision != _contentRevision) { + _continueWatchingRefreshQueued = true; + return; + } _applyOnDeck(fetched.items); _loadedOnDeckServerIds = fetched.succeededServerIds; safeNotifyListeners(); diff --git a/lib/providers/multi_server_provider.dart b/lib/providers/multi_server_provider.dart index 879fa4a3..3e09fe52 100644 --- a/lib/providers/multi_server_provider.dart +++ b/lib/providers/multi_server_provider.dart @@ -40,6 +40,7 @@ class MultiServerProvider extends ChangeNotifier with DisposableChangeNotifierMi /// Previously-seen set of online server IDs, used to detect new servers Set _previousOnlineServerIds = {}; + int _liveTvCheckGeneration = 0; /// Invoked with the current visibility-filtered online server ids whenever /// the manager's status stream fires (a server connects, reconnects, drops, @@ -286,8 +287,8 @@ class MultiServerProvider extends ChangeNotifier with DisposableChangeNotifierMi /// uniformly). Future checkLiveTvAvailability() async { if (isDisposed) return; + final generation = ++_liveTvCheckGeneration; final newLiveTvServers = []; - for (final serverId in onlineServerIds) { final genericClient = _serverManager.getClient(ServerId(serverId)); if (genericClient == null) continue; @@ -319,7 +320,7 @@ class MultiServerProvider extends ChangeNotifier with DisposableChangeNotifierMi final hadLiveTv = _hasLiveTv; final oldServerIds = _liveTvServers.map((s) => '${s.serverId}\u0000${s.dvrKey}').toSet(); final newServerIds = visibleLiveTvServers.map((s) => '${s.serverId}\u0000${s.dvrKey}').toSet(); - if (isDisposed) return; + if (isDisposed || generation != _liveTvCheckGeneration) return; _liveTvServers ..clear() ..addAll(visibleLiveTvServers); @@ -333,6 +334,7 @@ class MultiServerProvider extends ChangeNotifier with DisposableChangeNotifierMi @override void dispose() { + ++_liveTvCheckGeneration; _statusSubscription?.cancel(); super.dispose(); } diff --git a/lib/services/companion_remote/lan_discovery_service.dart b/lib/services/companion_remote/lan_discovery_service.dart index c379a410..2bf1184c 100644 --- a/lib/services/companion_remote/lan_discovery_service.dart +++ b/lib/services/companion_remote/lan_discovery_service.dart @@ -48,6 +48,7 @@ class LanDiscoveryService { RawDatagramSocket? _listenSocket; StreamSubscription? _listenSubscription; Timer? _staleCleanupTimer; + int _listenGeneration = 0; final Map _discoveredHosts = {}; final _hostsController = StreamController>.broadcast(); @@ -143,9 +144,9 @@ class LanDiscoveryService { Stream> startListeningForContexts(List contexts) { _stopListeningInternal(); _discoveredHosts.clear(); + final generation = _listenGeneration; - _bindListener(contexts); - + unawaited(_bindListener(contexts, generation)); // Periodically remove stale hosts _staleCleanupTimer = Timer.periodic(const Duration(seconds: 2), (_) { final now = DateTime.now(); @@ -166,23 +167,29 @@ class LanDiscoveryService { return _hostsController.stream; } - Future _bindListener(List contexts) async { + Future _bindListener(List contexts, int generation) async { try { - _listenSocket = await RawDatagramSocket.bind( + final socket = await RawDatagramSocket.bind( InternetAddress.anyIPv4, discoveryPort, reuseAddress: true, reusePort: true, ); - + if (generation != _listenGeneration) { + socket.close(); + return; + } + _listenSocket = socket; appLogger.d('LanDiscovery: Listening on port $discoveryPort'); - _listenSubscription = _listenSocket!.listenDatagrams( + _listenSubscription = socket.listenDatagrams( (datagram) => _handleDatagram(datagram, contexts), debugLabel: 'LanDiscovery listener', ); } catch (e) { - appLogger.e('LanDiscovery: Failed to bind listener', error: e); + if (generation == _listenGeneration) { + appLogger.e('LanDiscovery: Failed to bind listener', error: e); + } } } @@ -274,6 +281,7 @@ class LanDiscoveryService { } void _stopListeningInternal() { + ++_listenGeneration; _staleCleanupTimer?.cancel(); _staleCleanupTimer = null; _listenSubscription?.cancel(); diff --git a/lib/services/multi_server_manager.dart b/lib/services/multi_server_manager.dart index 0ec91eba..deaf5787 100644 --- a/lib/services/multi_server_manager.dart +++ b/lib/services/multi_server_manager.dart @@ -90,6 +90,7 @@ class MultiServerManager { /// Coalescing guard for reconnectOfflineServers — prevents concurrent reconnect sweeps Future? _activeReconnect; + int _profileRefreshGeneration = 0; /// Debounce timer for connectivity events — collapses rapid network flapping Timer? _connectivityDebounce; @@ -326,16 +327,26 @@ class MultiServerManager { return client; } - /// Persists a new endpoint, rebuilds the failover list, and switches the client. - Future _promoteEndpoint({ + /// 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({ required PlexClient client, required PlexServer server, required StorageService storage, required String newUrl, }) async { - await storage.saveServerEndpoint(ServerId(server.clientIdentifier), newUrl); + final serverId = ServerId(server.clientIdentifier); + bool isCurrent() { + final registered = _clients[serverId]; + return identical(_plexServers[serverId], server) && (registered == null || identical(registered, client)); + } + + if (!isCurrent()) return false; + await storage.saveServerEndpoint(serverId, newUrl); + if (!isCurrent()) return false; final newEndpoints = server.prioritizedEndpointUrls(preferredFirst: newUrl); await client.updateEndpointPreferences(newEndpoints, switchToFirst: true); + return isCurrent(); } /// Continues draining the connection optimization stream in the background, @@ -349,6 +360,12 @@ class MultiServerManager { () async { try { while (await streamIterator.moveNext()) { + final serverId = ServerId(server.clientIdentifier); + final registered = _clients[serverId]; + if (!identical(_plexServers[serverId], server) || (registered != null && !identical(registered, client))) { + appLogger.d('Stopping stale endpoint optimization for ${server.name}'); + break; + } final connection = streamIterator.current; final newUrl = connection.uri; @@ -483,6 +500,7 @@ class MultiServerManager { PlexAccountConnection connection, { Duration timeout = MediaServerTimeouts.perServerConnect, }) async { + final generation = ++_profileRefreshGeneration; if (connection.servers.isEmpty) return const {}; final bound = {}; final futures = connection.servers.map((server) async { @@ -491,11 +509,12 @@ class MultiServerManager { _plexServers[serverId] = server; final existing = _clients[serverId]; if (existing is PlexClient && ((_serverStatus[serverId] ?? false) || _authErrorServers.contains(serverId))) { - // Rotate the X-Plex-Token in-place so the server treats requests - // as the new user. `applyTokenUpdate` updates both config and - // _http.defaultHeaders — leaving headers stale would silently - // keep authenticating as the previous user. await existing.applyTokenUpdate(server.accessToken); + if (generation != _profileRefreshGeneration || + !identical(_plexServers[serverId], server) || + !identical(_clients[serverId], existing)) { + return; + } _authErrorServers.remove(serverId); _serverStatus[serverId] = true; bound.add(serverId); @@ -507,6 +526,10 @@ class MultiServerManager { server: server, clientIdentifier: connection.clientIdentifier, ).namedTimeout(timeout, operation: 'connect to ${server.name}'); + if (generation != _profileRefreshGeneration || !identical(_plexServers[serverId], server)) { + _closeClient(client); + return; + } final oldClient = _clients[serverId]; if (oldClient != null) _closeClient(oldClient); _clients[serverId] = client; @@ -515,12 +538,14 @@ class MultiServerManager { bound.add(serverId); _connectProgressController.add((serverId: serverId, online: true)); } catch (e, stackTrace) { + if (generation != _profileRefreshGeneration || !identical(_plexServers[serverId], server)) return; 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); + if (generation != _profileRefreshGeneration) return const {}; _statusController.add(Map.from(_serverStatus)); if (bound.isNotEmpty && _connectivitySubscription == null) { _startNetworkMonitoring(); @@ -937,10 +962,13 @@ class MultiServerManager { } if (client != null) { - await _promoteEndpoint(client: client, server: server, storage: storage, newUrl: newUrl); + final promoted = await _promoteEndpoint(client: client, server: server, storage: storage, newUrl: newUrl); + if (!promoted) return; appLogger.i('Switched ${server.name} to better endpoint: $newUrl', error: {'type': connection.displayType}); } else { + if (_plexServers[serverId] != server) return; await storage.saveServerEndpoint(serverId, newUrl); + if (_plexServers[serverId] != server) return; appLogger.i('Updated optimal endpoint for ${server.name}: $newUrl', error: {'type': connection.displayType}); } } @@ -960,6 +988,11 @@ class MultiServerManager { try { appLogger.d('Attempting reconnection for ${server.name}'); final client = await _createClientForServer(server: server, clientIdentifier: clientId); + if (!identical(_plexServers[serverId], server) || _resolveClientIdentifier(serverId) != clientId) { + _closeClient(client); + appLogger.d('Ignoring stale reconnection result for ${server.name}'); + return; + } final oldClient = _clients[serverId]; if (oldClient != null) _closeClient(oldClient); @@ -1138,6 +1171,7 @@ class MultiServerManager { } Set _detachAllClients() { + ++_profileRefreshGeneration; _stopNetworkMonitoring(); for (final timer in _reconnectDebounce.values) { timer.cancel(); diff --git a/lib/services/trackers/tracker_coordinator.dart b/lib/services/trackers/tracker_coordinator.dart index 163215e6..a3ba1ef0 100644 --- a/lib/services/trackers/tracker_coordinator.dart +++ b/lib/services/trackers/tracker_coordinator.dart @@ -49,15 +49,23 @@ class TrackerCoordinator { /// tracker crossing semantics stay aligned with playback progress reporting. final PlaybackTimeline _timeline = PlaybackTimeline(watchedThreshold: _fallbackWatchedThreshold); bool _thresholdCrossed = false; + int _playbackRevision = 0; Future initialize() async { await Future.wait(_trackers.map((t) => t.initialize())); } Future startPlayback(MediaItem metadata, MediaServerClient client, {bool isLive = false}) async { - if (isLive) return; + final revision = ++_playbackRevision; + if (isLive) { + _reset(); + return; + } final mediaType = metadata.kind; - if (mediaType != MediaKind.movie && mediaType != MediaKind.episode) return; + if (mediaType != MediaKind.movie && mediaType != MediaKind.episode) { + _reset(); + return; + } final libraryGlobalKey = metadata.libraryGlobalKey; if (!_hasActiveTrackerForLibrary(libraryGlobalKey)) { _reset(); @@ -72,6 +80,7 @@ class TrackerCoordinator { _resolverClientKey = clientKey; } final ctx = await _buildContext(metadata, _resolver!); + if (revision != _playbackRevision) return; if (ctx == null) { appLogger.d('Trackers: no external IDs for ${metadata.id}'); _reset(); @@ -287,16 +296,13 @@ class TrackerCoordinator { } Future stopPlayback() async { + ++_playbackRevision; final ctx = _ctx; - if (ctx == null) { - _reset(); - return; - } - // Safety net: fire if we passed the threshold but missed the tick. - if (!_thresholdCrossed && _timeline.watchedThresholdReached) { + final shouldMarkWatched = ctx != null && !_thresholdCrossed && _timeline.watchedThresholdReached; + _reset(); + if (ctx != null && shouldMarkWatched) { await _dispatchMarkWatched(ctx); } - _reset(); } void updatePosition(Duration position) { @@ -315,6 +321,7 @@ class TrackerCoordinator { /// Called on Plex profile switch — drops in-flight state across all /// trackers and invalidates the resolver so a fresh Plex client is used. void cancelInFlight() { + ++_playbackRevision; _reset(); _resolver?.clearCache(); _resolver = null; diff --git a/lib/services/trakt/trakt_scrobble_service.dart b/lib/services/trakt/trakt_scrobble_service.dart index d3c22903..97fdf498 100644 --- a/lib/services/trakt/trakt_scrobble_service.dart +++ b/lib/services/trakt/trakt_scrobble_service.dart @@ -53,6 +53,7 @@ class TraktScrobbleService implements TrackerRatingSource { TraktScrobbleState? _lastSentState; DateTime? _lastSentAt; DateTime? _lastSeekCheckpointAt; + int _playbackRevision = 0; Future initialize() async { if (_isInitialized) return; @@ -124,9 +125,15 @@ class TraktScrobbleService implements TrackerRatingSource { /// Drop the current scrobble state without sending a stop. Called on profile /// switch and when the service is disabled mid-playback. void cancelInFlight() { + ++_playbackRevision; + _clearPlaybackState(); + } + + void _clearPlaybackState() { _currentBody = null; _lastSentState = null; _lastSentAt = null; + _lastSeekCheckpointAt = null; _resolver?.clearCache(); _resolver = null; _timeline.reset(); @@ -209,8 +216,9 @@ class TraktScrobbleService implements TrackerRatingSource { } Future startPlayback(MediaItem metadata, MediaServerClient client, {bool isLive = false}) async { - if (!_canScrobble) return; - if (isLive) return; + final revision = ++_playbackRevision; + _clearPlaybackState(); + if (!_canScrobble || isLive) return; final type = metadata.kind; if (type != MediaKind.movie && type != MediaKind.episode) return; @@ -227,13 +235,13 @@ class TraktScrobbleService implements TrackerRatingSource { position: metadata.viewOffsetMs != null ? Duration(milliseconds: metadata.viewOffsetMs!) : Duration.zero, duration: metadata.durationMs != null ? Duration(milliseconds: metadata.durationMs!) : null, ); - _lastSeekCheckpointAt = null; _resolver = TrackerIdResolver(client, needsFribb: () => false); final body = await _buildBody(metadata); + if (revision != _playbackRevision) return; if (body == null) { appLogger.d('Trakt: skipping scrobble — no usable IDs for ${metadata.id}'); - cancelInFlight(); + _clearPlaybackState(); return; } _currentBody = body; @@ -274,9 +282,13 @@ class TraktScrobbleService implements TrackerRatingSource { } Future stopPlayback() async { - if (_currentBody == null) return; + final revision = ++_playbackRevision; + if (_currentBody == null) { + _clearPlaybackState(); + return; + } await _send(TraktScrobbleState.stop, progress: _progressPercent()); - cancelInFlight(); + if (revision == _playbackRevision) _clearPlaybackState(); } Future _buildBody(MediaItem metadata) async { diff --git a/lib/watch_together/providers/watch_together_provider.dart b/lib/watch_together/providers/watch_together_provider.dart index 9859aa19..ab335dfc 100644 --- a/lib/watch_together/providers/watch_together_provider.dart +++ b/lib/watch_together/providers/watch_together_provider.dart @@ -389,8 +389,7 @@ class WatchTogetherProvider with ChangeNotifier { appLogger.d('WatchTogether: Joined session successfully'); } catch (e) { appLogger.e('WatchTogether: Failed to join session', error: e); - _session = _session?.copyWith(state: SessionState.error, errorMessage: e.toString()); - notifyListeners(); + await leaveSession(); rethrow; } } diff --git a/lib/watch_together/services/guest_playback_reconciler.dart b/lib/watch_together/services/guest_playback_reconciler.dart index 89a0ce56..da907683 100644 --- a/lib/watch_together/services/guest_playback_reconciler.dart +++ b/lib/watch_together/services/guest_playback_reconciler.dart @@ -103,6 +103,7 @@ class GuestPlaybackReconciler { bool _nudgeDisabled = false; bool _nudgeConfirmed = false; Timer? _nudgeConfirmTimer; + double? _nudgeTargetRate; int _lastHardSeekMs = -hardSeekCooldownMs; final List _driftSamples = []; @@ -212,6 +213,9 @@ class GuestPlaybackReconciler { _settleTimer = null; _settling = false; _nudging = false; + _nudgeDisabled = false; + _nudgeConfirmed = false; + _nudgeTargetRate = null; _nudgeConfirmTimer?.cancel(); _nudgeConfirmTimer = null; _scheduledStartTimer?.cancel(); @@ -499,6 +503,7 @@ class GuestPlaybackReconciler { if (_nudging && (player.rate - targetRate).abs() < 0.001) return; _nudging = true; + _nudgeTargetRate = targetRate; unawaited(player.setRate(targetRate)); // Arm the capability check once per (un-confirmed) nudge episode — a @@ -507,11 +512,13 @@ class GuestPlaybackReconciler { _nudgeConfirmTimer = Timer(const Duration(milliseconds: nudgeConfirmMs), () { _nudgeConfirmTimer = null; final currentPlayer = _player; - if (currentPlayer == null || !_nudging) return; - if ((currentPlayer.rate - targetRate).abs() > 0.005) { - appLogger.w('WatchTogether: Rate nudges not taking effect — disabling for this session'); + final expectedRate = _nudgeTargetRate; + if (currentPlayer == null || !_nudging || expectedRate == null) return; + if ((currentPlayer.rate - expectedRate).abs() > 0.005) { + appLogger.w('WatchTogether: Rate nudges not taking effect — disabling for this attachment'); _nudgeDisabled = true; _nudging = false; + _nudgeTargetRate = null; unawaited(currentPlayer.setRate(_latestState?.rate ?? 1.0)); } else { _nudgeConfirmed = true; @@ -523,6 +530,7 @@ class GuestPlaybackReconciler { void _exitNudgeIfNeeded(PlaybackState state) { if (!_nudging) return; _nudging = false; + _nudgeTargetRate = null; final player = _player; if (player != null) { unawaited(player.setRate(state.rate)); diff --git a/lib/watch_together/services/watch_together_peer_service.dart b/lib/watch_together/services/watch_together_peer_service.dart index a0dd4b45..0512da94 100644 --- a/lib/watch_together/services/watch_together_peer_service.dart +++ b/lib/watch_together/services/watch_together_peer_service.dart @@ -62,6 +62,8 @@ class WatchTogetherPeerService with KeepaliveMixin { int _reconnectAttempts = 0; static const int _maxReconnectAttempts = 3; Timer? _reconnectTimer; + int _connectionEpoch = 0; + bool _disposed = false; /// Called after a successful reconnection so the provider can re-announce join. void Function()? onReconnected; @@ -126,8 +128,12 @@ class WatchTogetherPeerService with KeepaliveMixin { } /// Connect, listen, and send a room setup announcement. - Future> _connectAndAnnounce(String type) async { + Future> _connectAndAnnounce(String type, int epoch) async { final channel = await _connectToRelay(); + if (_disposed || epoch != _connectionEpoch || _sessionId == null) { + unawaited(channel.sink.close()); + throw StateError('Watch Together connection attempt became stale'); + } _channel = channel; _listenToChannel(channel); @@ -288,26 +294,28 @@ class WatchTogetherPeerService with KeepaliveMixin { /// Handle the WebSocket being closed unexpectedly — attempt reconnection. void _handleWebSocketClosed() { + final channel = _channel; + ++_connectionEpoch; stopKeepalive(); - _channelSubscription?.cancel(); + unawaited(_channelSubscription?.cancel()); _channelSubscription = null; _channel = null; + if (channel != null) unawaited(channel.sink.close()); - // Notify peers lost for (final peerId in _connectedPeers.toList()) { _safeAdd(_peerDisconnectedController, peerId); } _connectedPeers.clear(); _safeAdd(_connectionStateController, false); - // Attempt to reconnect if we had a session - if (_sessionId != null) { - _attemptReconnect(); + if (!_disposed && _sessionId != null) { + _attemptReconnect(_connectionEpoch); } } /// Attempt to reconnect to the relay and re-join/re-create the room. - void _attemptReconnect() { + void _attemptReconnect(int epoch) { + if (_disposed || epoch != _connectionEpoch || _sessionId == null) return; if (_reconnectAttempts >= _maxReconnectAttempts) { appLogger.e('WatchTogether: Max reconnect attempts reached'); _safeAdd( @@ -326,15 +334,15 @@ class WatchTogetherPeerService with KeepaliveMixin { _reconnectTimer?.cancel(); _reconnectTimer = Timer(delay, () async { + if (_disposed || epoch != _connectionEpoch || _sessionId == null) return; try { - // Always try join first — the room may still have peers (e.g. host - // reconnecting while guests remain). Fall back to create only if - // the room no longer exists and we were the host. - final completer = await _connectAndAnnounce(RelayProtocol.join); + final completer = await _connectAndAnnounce(RelayProtocol.join, epoch); + if (_disposed || epoch != _connectionEpoch) return; try { await completer.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect'); } on PeerError catch (e) { + if (_disposed || epoch != _connectionEpoch) return; if (_isHost && e.serverCode == RelayProtocol.roomNotFoundCode) { appLogger.d('WatchTogether: Room gone, re-creating as host'); final createCompleter = _announce(RelayProtocol.create); @@ -347,6 +355,7 @@ class WatchTogetherPeerService with KeepaliveMixin { } } + if (_disposed || epoch != _connectionEpoch) return; _reconnectAttempts = 0; appLogger.d('WatchTogether: Reconnected successfully'); try { @@ -355,6 +364,7 @@ class WatchTogetherPeerService with KeepaliveMixin { appLogger.e('WatchTogether: Reconnect callback failed', error: e); } } catch (e) { + if (_disposed || epoch != _connectionEpoch) return; appLogger.e('WatchTogether: Reconnect failed', error: e); _handleWebSocketClosed(); } @@ -382,9 +392,10 @@ class WatchTogetherPeerService with KeepaliveMixin { _sessionId = resolvedSessionId; _myPeerId = watchTogetherHostPeerId(resolvedSessionId); _reconnectAttempts = 0; + final epoch = ++_connectionEpoch; try { - final completer = await _connectAndAnnounce(RelayProtocol.create); + final completer = await _connectAndAnnounce(RelayProtocol.create, epoch); await completer.future.timeout( const Duration(seconds: 10), @@ -420,9 +431,10 @@ class WatchTogetherPeerService with KeepaliveMixin { _sessionId = resolvedSessionId; _myPeerId = const Uuid().v4(); _reconnectAttempts = 0; + final epoch = ++_connectionEpoch; try { - final completer = await _connectAndAnnounce(RelayProtocol.join); + final completer = await _connectAndAnnounce(RelayProtocol.join, epoch); await completer.future.timeout( const Duration(seconds: 10), @@ -457,34 +469,40 @@ class WatchTogetherPeerService with KeepaliveMixin { /// Disconnect from all peers and close the session Future disconnect() async { appLogger.d('WatchTogether: Disconnecting...'); - + ++_connectionEpoch; _reconnectTimer?.cancel(); _reconnectTimer = null; stopKeepalive(); - unawaited(_channelSubscription?.cancel()); + final subscription = _channelSubscription; + final channel = _channel; _channelSubscription = null; - - try { - await _channel?.sink.close(); - } catch (e) { - appLogger.d('WatchTogether: channel close ignored', error: e); - } _channel = null; + final setupCompleter = _setupCompleter; _setupCompleter = null; - + if (setupCompleter != null && !setupCompleter.isCompleted) { + setupCompleter.completeError(StateError('Watch Together connection cancelled')); + } _connectedPeers.clear(); _sessionId = null; _myPeerId = null; _isHost = false; _reconnectAttempts = 0; + unawaited(subscription?.cancel()); + try { + await channel?.sink.close(); + } catch (e) { + appLogger.d('WatchTogether: channel close ignored', error: e); + } _safeAdd(_connectionStateController, false); } - /// Dispose all resources + /// Dispose all resources. void dispose() { - disconnect(); + if (_disposed) return; + _disposed = true; + unawaited(disconnect()); _peerConnectedController.close(); _peerDisconnectedController.close(); diff --git a/lib/widgets/video_controls/sheets/subtitle_search_sheet.dart b/lib/widgets/video_controls/sheets/subtitle_search_sheet.dart index 3ccb1eb6..1b4b9bf2 100644 --- a/lib/widgets/video_controls/sheets/subtitle_search_sheet.dart +++ b/lib/widgets/video_controls/sheets/subtitle_search_sheet.dart @@ -60,6 +60,7 @@ class _SubtitleSearchSheetState extends State with Controll bool _isSearching = false; String? _error; String? _downloadingKey; + int _searchGeneration = 0; bool _showLanguagePicker = false; @@ -83,6 +84,7 @@ class _SubtitleSearchSheetState extends State with Controll @override void dispose() { + ++_searchGeneration; _debounceTimer?.cancel(); _languageFocusNode.dispose(); _titleFocusNode.dispose(); @@ -92,37 +94,34 @@ class _SubtitleSearchSheetState extends State with Controll Future _search() async { if (!mounted) return; + final generation = ++_searchGeneration; setState(() { _isSearching = true; _error = null; }); try { - // Defense-in-depth: searchSubtitles is Plex-only. The UI gates this - // sheet on `subtitleSearchSupported` upstream, but if a future caller - // reaches us with a Jellyfin server, fail soft instead of throwing. final neutral = context.tryGetMediaClientForServer(ServerId(widget.serverId)); final client = neutral is PlexClient ? neutral : null; if (client == null) { - if (!mounted) return; - setState(() { - _isSearching = false; - }); + if (!mounted || generation != _searchGeneration) return; + setState(() => _isSearching = false); return; } final title = _titleController.text.trim(); + final language = _languageCode; final results = await client.searchSubtitles( widget.ratingKey, - language: _languageCode, + language: language, title: title.isEmpty ? null : title, ); - if (!mounted) return; + if (!mounted || generation != _searchGeneration) return; setState(() { _results = results; _isSearching = false; }); } catch (e) { - if (!mounted) return; + if (!mounted || generation != _searchGeneration) return; setState(() { _error = e.toString(); _isSearching = false; @@ -149,7 +148,7 @@ class _SubtitleSearchSheetState extends State with Controll Future _submitSearchAndFocusFirstResult() async { _debounceTimer?.cancel(); await _search(); - if (!mounted || !InputModeTracker.isKeyboardMode(context)) return; + if (!mounted || !InputModeTracker.isKeyboardMode(context, listen: false)) return; WidgetsBinding.instance.addPostFrameCallback((_) { if (!mounted) return; @@ -158,6 +157,7 @@ class _SubtitleSearchSheetState extends State with Controll } void _onLanguageSelected(String code, String name) { + _debounceTimer?.cancel(); setState(() { _languageCode = code; _languageName = name; diff --git a/test/mpv/player_native_bridge_test.dart b/test/mpv/player_native_bridge_test.dart index b36630d8..167789ba 100644 --- a/test/mpv/player_native_bridge_test.dart +++ b/test/mpv/player_native_bridge_test.dart @@ -248,4 +248,43 @@ void main() { }, ); }); + + test('overlapping playback-rate changes are serialized in call order', () async { + final releaseFirstSpeed = Completer(); + final firstSpeedStarted = Completer(); + final speedValues = []; + await withMockPlayerChannels( + methodChannelName: 'com.plezy/mpv_player', + eventChannelName: 'com.plezy/mpv_player/events', + methodHandler: (call) async { + if (call.method == 'initialize') return true; + if (call.method == 'setProperty') { + final arguments = call.arguments as Map; + if (arguments['name'] == 'speed') { + speedValues.add(arguments['value'] as String); + if (!firstSpeedStarted.isCompleted) firstSpeedStarted.complete(); + if (speedValues.length == 1) await releaseFirstSpeed.future; + } + } + return null; + }, + testBody: () async { + final player = PlayerNative(); + try { + final first = player.setRate(1.25); + final second = player.setRate(1.5); + await firstSpeedStarted.future; + + expect(speedValues, ['1.25']); + + releaseFirstSpeed.complete(); + await Future.wait([first, second]); + expect(speedValues, ['1.25', '1.5']); + } finally { + if (!releaseFirstSpeed.isCompleted) releaseFirstSpeed.complete(); + await player.dispose(); + } + }, + ); + }); } diff --git a/test/watch_together/watch_together_peer_service_test.dart b/test/watch_together/watch_together_peer_service_test.dart index 314e1371..65ae3d54 100644 --- a/test/watch_together/watch_together_peer_service_test.dart +++ b/test/watch_together/watch_together_peer_service_test.dart @@ -215,4 +215,22 @@ void main() { expect(peerEvents, ['guest-1']); expect(relay.messages.single.where((message) => message['type'] == 'create'), hasLength(1)); }); + + test('disconnect cancels an in-flight room announcement without timeout delay', () async { + final announcementSeen = Completer(); + final relay = await relayWith((_, _, message) { + if (message['type'] == 'create' && !announcementSeen.isCompleted) { + announcementSeen.complete(); + } + }); + final service = serviceFor(relay); + + final pending = service.createSession(sessionId: 'cancel1'); + await announcementSeen.future.timeout(const Duration(seconds: 1)); + await service.disconnect(); + + await expectLater(pending, throwsStateError); + expect(service.sessionId, isNull); + expect(service.connectedPeers, isEmpty); + }); }