From 86abf3e9da7f6472f12443a023c65bb4be070528 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Thu, 2 Jul 2026 12:44:15 +0200 Subject: [PATCH] fix(watch-together): retry guest media switches until they commit Guest switch dispatch pre-marked its dedup key and fired-and-forgot, so any failure (fetch error, reload busy with an auto-advance, navigation race with the host exiting) silently stranded the guest on the old media. A CurrentPlaybackDispatcher now marks a key handled only after the sink reports success against the committed identity, with a serialized in-flight slot, timeout, and generation reset; the reconciler re-offers unattached media on every host heartbeat, making the heartbeat the retry channel. Fetches that outlive their dispatch are re-validated against the current snapshot so a stale switch can't override the live one. hostExitedPlayer now rides the controller's ordered message queue with host authentication instead of racing state handling in the provider. --- lib/screens/main_screen.dart | 15 +- .../parts/episode_navigation.dart | 3 +- .../video_player/parts/watch_together.dart | 84 ++++++++++-- lib/screens/video_player_screen.dart | 4 + lib/utils/video_player_navigation.dart | 18 ++- .../providers/watch_together_provider.dart | 68 ++++++--- .../services/current_playback_dispatcher.dart | 50 +++++++ .../services/guest_playback_reconciler.dart | 14 +- .../services/watch_together_controller.dart | 8 +- .../current_playback_dispatcher_test.dart | 103 ++++++++++++++ .../guest_playback_reconciler_test.dart | 38 ++++++ .../watch_together_controller_test.dart | 52 +++++++ .../watch_together_provider_test.dart | 129 ++++++++++++++++++ 13 files changed, 535 insertions(+), 51 deletions(-) create mode 100644 lib/watch_together/services/current_playback_dispatcher.dart create mode 100644 test/watch_together/current_playback_dispatcher_test.dart diff --git a/lib/screens/main_screen.dart b/lib/screens/main_screen.dart index e067bc94..4eaa8ad0 100644 --- a/lib/screens/main_screen.dart +++ b/lib/screens/main_screen.dart @@ -617,9 +617,9 @@ class _MainScreenState extends State void _setupWatchTogetherCallback() { try { final watchTogether = context.read(); - watchTogether.onMediaSwitched = (ratingKey, serverId, mediaTitle) async { + watchTogether.onMediaSwitched = (ratingKey, serverId, mediaTitle) { appLogger.d('WatchTogether: Media switch received - navigating to $mediaTitle'); - await _navigateToWatchTogetherMedia(ratingKey, serverId); + return _navigateToWatchTogetherMedia(ratingKey, serverId); }; watchTogether.onHostExitedPlayer = () { appLogger.d('WatchTogether: Host exited player - exiting player for guest'); @@ -696,14 +696,17 @@ class _MainScreenState extends State } } - /// Navigate to media when host switches content in Watch Together session - Future _navigateToWatchTogetherMedia(String ratingKey, ServerId serverId) async { - if (!mounted) return; // Check before any context usage + /// Navigate to media when host switches content in Watch Together session. + /// Returns whether navigation was initiated; failures are re-dispatched on + /// the host's next state heartbeat. + Future _navigateToWatchTogetherMedia(String ratingKey, ServerId serverId) async { + if (!mounted) return false; // Check before any context usage try { - await navigateToWatchTogetherPlayback(context, ratingKey: ratingKey, serverId: serverId); + return await navigateToWatchTogetherPlayback(context, ratingKey: ratingKey, serverId: serverId); } catch (e) { appLogger.e('WatchTogether: Failed to navigate to media', error: e); + return false; } } diff --git a/lib/screens/video_player/parts/episode_navigation.dart b/lib/screens/video_player/parts/episode_navigation.dart index d0f7c627..fcacae9b 100644 --- a/lib/screens/video_player/parts/episode_navigation.dart +++ b/lib/screens/video_player/parts/episode_navigation.dart @@ -211,6 +211,7 @@ extension _VideoPlayerEpisodeNavigationMethods on VideoPlayerScreenState { Duration? resumePosition, bool preserveCurrentTrackSelection = false, bool useCurrentAudioStreamSelection = true, + bool showErrorUi = true, String reason = 'media reload', }) async { if (widget.isLive) { @@ -536,7 +537,7 @@ extension _VideoPlayerEpisodeNavigationMethods on VideoPlayerScreenState { }); if (isItemChange) _showChromeForSwappedItem(); appLogger.e('Failed to reload media in-place during $reason', error: e); - if (mounted) { + if (mounted && showErrorUi) { showErrorSnackBar(context, t.messages.errorLoading(error: e.toString())); } return true; diff --git a/lib/screens/video_player/parts/watch_together.dart b/lib/screens/video_player/parts/watch_together.dart index 5894af61..2e150c25 100644 --- a/lib/screens/video_player/parts/watch_together.dart +++ b/lib/screens/video_player/parts/watch_together.dart @@ -108,9 +108,23 @@ extension _VideoPlayerWatchTogetherMethods on VideoPlayerScreenState { } } - /// Handle media switch from host (guest only) using the in-place reload path. - Future _handlePlayerMediaSwitch(String ratingKey, ServerId serverId, String title) async { - if (!mounted) return; + /// Handle media switch from host (guest only) using the in-place reload + /// path. Returns whether the switch was handled; unhandled switches are + /// re-dispatched on the host's next state heartbeat. + Future _handlePlayerMediaSwitch(String ratingKey, ServerId serverId, String title) async { + if (!mounted) return false; + final switchKey = '$serverId:$ratingKey'; + + // Idempotent retry: already on the target with a settled player. Don't + // test identity mid-transition — _currentMetadata is set eagerly at + // reload start and can roll back on failure. + if (_playbackTransition == _PlaybackTransition.idle && + player != null && + _currentMetadata.id == ratingKey && + _currentMetadata.serverId == serverId) { + _wtSwitchToastShownForKey = null; + return true; + } appLogger.d('WatchTogether: Guest handling media switch to $title'); @@ -122,21 +136,40 @@ extension _VideoPlayerWatchTogetherMethods on VideoPlayerScreenState { final client = multiServer.getClientForServer(serverId); if (client == null) { appLogger.w('WatchTogether: Server $serverId not found for media switch'); - if (mounted) showAppSnackBar(context, t.watchTogether.guestSwitchUnavailable); - return; + _showSwitchFailureToastOnce(switchKey, t.watchTogether.guestSwitchUnavailable); + return false; } - final metadata = await client.fetchItem(ratingKey); - if (!mounted) return; + MediaItem? metadata; + try { + metadata = await client.fetchItem(ratingKey); + } catch (e, stackTrace) { + appLogger.w('WatchTogether: Could not fetch metadata for $ratingKey', error: e, stackTrace: stackTrace); + } + if (!mounted) return false; if (metadata == null) { appLogger.w('WatchTogether: Could not fetch metadata for $ratingKey'); - showAppSnackBar(context, t.watchTogether.guestSwitchFailed); - return; + _showSwitchFailureToastOnce(switchKey, t.watchTogether.guestSwitchFailed); + return false; + } + + // The fetch can outlive the dispatch that requested it (slow server, + // host switching again, dispatcher timeout); reloading then would swap + // the live screen to stale media. Unhandled: the current key rides the + // next heartbeat. + final watchTogether = _activeWatchTogetherSession(); + if (watchTogether == null || + watchTogether.currentMediaRatingKey != ratingKey || + watchTogether.currentMediaServerId != serverId) { + appLogger.d('WatchTogether: Skipping stale media switch to $ratingKey'); + return false; } if (player == null || widget.isLive) { + // Route replacement: report handled at initiation — the navigation + // future only completes when the pushed route pops. unawaited(_replaceScreenWithPlayer(metadata)); - return; + return true; } final handled = await _reloadMediaInPlace( @@ -146,10 +179,37 @@ extension _VideoPlayerWatchTogetherMethods on VideoPlayerScreenState { qualityPreset: _selectedQualityPreset, preserveCurrentTrackSelection: false, useCurrentAudioStreamSelection: false, + showErrorUi: false, // the retry loop owns user feedback (once per key) reason: 'watch together media switch', ); - if (!handled && mounted && player == null) { - unawaited(_replaceScreenWithPlayer(metadata)); + if (!mounted) return false; + if (!handled) { + if (player == null) { + unawaited(_replaceScreenWithPlayer(metadata)); + return true; + } + // Busy transition (e.g. auto-advance racing the host switch) — not an + // error; the next heartbeat re-dispatches and converges once idle. + return false; } + // handled==true also covers "reload failed after rollback" and + // "superseded by a newer attempt" — trust only the committed identity. + final onTarget = _currentMetadata.id == ratingKey && _currentMetadata.serverId == serverId; + if (onTarget) { + // A success ends the failure episode for this key; a later failure to + // switch back here must toast again. + _wtSwitchToastShownForKey = null; + } else { + _showSwitchFailureToastOnce(switchKey, t.watchTogether.guestSwitchFailed); + } + return onTarget; + } + + /// Toast a Watch Together switch failure at most once per media key (the + /// heartbeat retry loop calls the handler every few seconds). + void _showSwitchFailureToastOnce(String switchKey, String message) { + if (_wtSwitchToastShownForKey == switchKey) return; + _wtSwitchToastShownForKey = switchKey; + if (mounted) showAppSnackBar(context, message); } } diff --git a/lib/screens/video_player_screen.dart b/lib/screens/video_player_screen.dart index 6ccd0c01..0285ce8d 100644 --- a/lib/screens/video_player_screen.dart +++ b/lib/screens/video_player_screen.dart @@ -256,6 +256,10 @@ class VideoPlayerScreenState extends State with WidgetsBindin _PlaybackTransition _playbackTransition = _PlaybackTransition.idle; bool _playbackIntentShouldPlay = true; + /// Media key of the last Watch Together switch failure the user was + /// toasted about — the heartbeat retry loop must not re-toast every 2s. + String? _wtSwitchToastShownForKey; + bool _showPlayNextDialog = false; bool _isPhone = false; late int _effectiveSelectedMediaIndex; diff --git a/lib/utils/video_player_navigation.dart b/lib/utils/video_player_navigation.dart index bd2897a4..9e36efa0 100644 --- a/lib/utils/video_player_navigation.dart +++ b/lib/utils/video_player_navigation.dart @@ -11,6 +11,7 @@ import '../models/transcode_quality_preset.dart'; import '../providers/download_provider.dart'; import '../providers/multi_server_provider.dart'; import '../providers/watch_state_store.dart'; +import '../watch_together/providers/watch_together_provider.dart'; import '../screens/video_player_screen.dart'; import '../services/external_player_service.dart'; import '../services/offline_watch_sync_service.dart'; @@ -313,7 +314,13 @@ Future navigateToVideoPlayerWithRefresh( } /// Resolves the current Watch Together media and opens the video player. -Future navigateToWatchTogetherPlayback( +/// +/// Returns whether navigation was initiated. The fetch can outlive the +/// dispatch that requested it (slow server, host switching again, dispatcher +/// timeout); navigating then would stack a stale player route on top of the +/// live one, so the key is re-validated against the session's current +/// playback snapshot before the push. +Future navigateToWatchTogetherPlayback( BuildContext context, { required String ratingKey, required ServerId serverId, @@ -331,8 +338,15 @@ Future navigateToWatchTogetherPlayback( throw const WatchTogetherPlaybackNavigationException('Current Watch Together media is unavailable'); } - if (!context.mounted) return; + if (!context.mounted) return false; + + final watchTogether = context.read(); + if (watchTogether.currentMediaRatingKey != ratingKey || watchTogether.currentMediaServerId != serverId) { + appLogger.d('WatchTogether: Skipping stale navigation to $ratingKey'); + return false; + } onBeforeNavigate?.call(); unawaited(navigateToVideoPlayer(context, metadata: metadata)); + return true; } diff --git a/lib/watch_together/providers/watch_together_provider.dart b/lib/watch_together/providers/watch_together_provider.dart index 54202118..f5732418 100644 --- a/lib/watch_together/providers/watch_together_provider.dart +++ b/lib/watch_together/providers/watch_together_provider.dart @@ -10,11 +10,14 @@ import '../../utils/app_logger.dart'; import '../models/playback_state.dart'; import '../models/sync_message.dart'; import '../models/watch_session.dart'; +import '../services/current_playback_dispatcher.dart'; import '../services/watch_together_controller.dart'; import '../services/watch_together_peer_service.dart'; -/// Callback type for when media switches (for guest navigation) -typedef MediaSwitchCallback = void Function(String ratingKey, ServerId serverId, String mediaTitle); +/// Callback type for when media switches (for guest navigation). Returns +/// whether the switch was handled; unhandled keys are re-dispatched on the +/// host's next state heartbeat. +typedef MediaSwitchCallback = Future Function(String ratingKey, ServerId serverId, String mediaTitle); /// Provider for Watch Together functionality /// @@ -34,7 +37,7 @@ class WatchTogetherProvider with ChangeNotifier { List _waitingOnPeerIds = const []; PlaybackPhase? _playbackPhase; String _displayName = 'User'; - String? _lastHandledCurrentPlaybackKey; + final CurrentPlaybackDispatcher _playbackDispatcher = CurrentPlaybackDispatcher(); // Coalesce rapid-fire notifyListeners() calls into a single rebuild per frame. // During Watch Together join, 4-5 notifications fire within milliseconds; @@ -163,7 +166,7 @@ class WatchTogetherProvider with ChangeNotifier { errorMessage: session.errorMessage, hostPeerId: session.hostPeerId, ); - _lastHandledCurrentPlaybackKey = null; + _playbackDispatcher.reset(); } void _dispatchCurrentPlayback({ @@ -178,13 +181,19 @@ class WatchTogetherProvider with ChangeNotifier { return; } - _lastHandledCurrentPlaybackKey = _buildPlaybackKey(ratingKey, ServerId(serverId)); appLogger.d('WatchTogether: Dispatching current playback from $source: $mediaTitle'); - callback(ratingKey, ServerId(serverId), mediaTitle); + // The key is only marked handled if the callback reports success; a + // failed switch is retried on the host's next state heartbeat. + unawaited( + _playbackDispatcher.dispatch( + _buildPlaybackKey(ratingKey, serverId)!, + () => callback(ratingKey, serverId, mediaTitle), + ), + ); } void markCurrentPlaybackHandled({required String ratingKey, required ServerId serverId}) { - _lastHandledCurrentPlaybackKey = _buildPlaybackKey(ratingKey, serverId); + _playbackDispatcher.markHandled(_buildPlaybackKey(ratingKey, serverId)!); } void requestCurrentPlaybackSnapshot() { @@ -239,6 +248,8 @@ class WatchTogetherProvider with ChangeNotifier { controller.onMediaStateReceived = _handleMediaStateReceived; + controller.onHostExitedPlayer = _handleHostExitedPlayer; + controller.onRemoteAction = (peerId, hint) { final type = switch (hint) { PlaybackActionHint.play => ParticipantEventType.resumed, @@ -291,7 +302,7 @@ class WatchTogetherProvider with ChangeNotifier { }) async { // Clean up any existing session await leaveSession(); - _lastHandledCurrentPlaybackKey = null; + _playbackDispatcher.reset(); appLogger.d('WatchTogether: Creating session with control mode: $controlMode'); @@ -335,7 +346,7 @@ class WatchTogetherProvider with ChangeNotifier { Future joinSession(String sessionId, {String? displayName}) async { // Clean up any existing session await leaveSession(); - _lastHandledCurrentPlaybackKey = null; + _playbackDispatcher.reset(); appLogger.d('WatchTogether: Joining session: $sessionId'); @@ -446,7 +457,7 @@ class WatchTogetherProvider with ChangeNotifier { _isWaitingForPeers = false; _waitingOnPeerIds = const []; _playbackPhase = null; - _lastHandledCurrentPlaybackKey = null; + _playbackDispatcher.reset(); _lastActionEventMs.clear(); _hostIntentionallyLeft = false; @@ -606,7 +617,7 @@ class WatchTogetherProvider with ChangeNotifier { // If the host deliberately left, end the session for everyone. if (!isHost && message.peerId == _session?.hostPeerId) { _hostIntentionallyLeft = true; - _handleHostExitedPlayer(message); + _handleHostExitedPlayer(); leaveSession(); } @@ -614,9 +625,8 @@ class WatchTogetherProvider with ChangeNotifier { } break; - case SyncMessageType.hostExitedPlayer: - _handleHostExitedPlayer(message); - break; + // hostExitedPlayer is routed through the controller's ordered message + // queue so it can't overtake (or be overtaken by) state messages. default: // Playback sync messages (state/status/control/...) are handled by @@ -647,22 +657,38 @@ class WatchTogetherProvider with ChangeNotifier { void _handleMediaStateReceived(String ratingKey, String serverId, String? mediaTitle) { if (isHost) return; - final playbackKey = _buildPlaybackKey(ratingKey, serverIdOrNull(serverId)); - final shouldDispatch = playbackKey != _lastHandledCurrentPlaybackKey; + final typedServerId = serverIdOrNull(serverId); + if (typedServerId == null) { + appLogger.w('WatchTogether: Ignoring playback state with blank serverId'); + return; + } + final playbackKey = _buildPlaybackKey(ratingKey, typedServerId); - _updateCurrentPlaybackSnapshot(ratingKey: ratingKey, serverId: ServerId(serverId), mediaTitle: mediaTitle ?? ''); - notifyListeners(); + // Detached guests receive this on every heartbeat; only rebuild when the + // snapshot actually changes. + final session = _session; + final snapshotChanged = + session == null || + session.mediaRatingKey != ratingKey || + session.mediaServerId != typedServerId || + session.mediaTitle != (mediaTitle ?? ''); + _updateCurrentPlaybackSnapshot(ratingKey: ratingKey, serverId: typedServerId, mediaTitle: mediaTitle ?? ''); + if (snapshotChanged) notifyListeners(); - if (shouldDispatch) { + if (_playbackDispatcher.shouldDispatch(playbackKey)) { _dispatchCurrentPlayback( ratingKey: ratingKey, - serverId: ServerId(serverId), + serverId: typedServerId, mediaTitle: mediaTitle ?? '', source: 'playback state', ); } } + @visibleForTesting + void debugHandleMediaState(String ratingKey, String serverId, String? mediaTitle) => + _handleMediaStateReceived(ratingKey, serverId, mediaTitle); + /// Called when user seeks locally (to sync with peers) void onLocalSeek(Duration position) { _controller?.onLocalSeek(position); @@ -710,7 +736,7 @@ class WatchTogetherProvider with ChangeNotifier { } /// Handle host exited player message (guest only) - void _handleHostExitedPlayer(SyncMessage _) { + void _handleHostExitedPlayer() { if (isHost) return; // Host doesn't need to handle their own exit appLogger.d('WatchTogether: Host exited player, callback set: ${onHostExitedPlayer != null}'); diff --git a/lib/watch_together/services/current_playback_dispatcher.dart b/lib/watch_together/services/current_playback_dispatcher.dart new file mode 100644 index 00000000..83c51a55 --- /dev/null +++ b/lib/watch_together/services/current_playback_dispatcher.dart @@ -0,0 +1,50 @@ +import '../../utils/app_logger.dart'; + +/// Serializes guest media-switch dispatches and provides heartbeat-driven +/// retry: a key is only marked handled after its callback reports success, +/// so a failed switch is re-dispatched by the host's next state heartbeat. +class CurrentPlaybackDispatcher { + static const dispatchTimeout = Duration(seconds: 30); + + String? _lastHandledKey; + String? _inFlightKey; + int _generation = 0; + + String? get inFlightKey => _inFlightKey; + + /// Whether [key] should be dispatched now. A single in-flight slot + /// serializes dispatches (concurrent navigations would stack player + /// routes); once it frees, the next heartbeat carries the latest key. + bool shouldDispatch(String? key) => key != null && key != _lastHandledKey && _inFlightKey == null; + + /// Suppress future dispatches of [key] (e.g. a user-initiated join already + /// navigating to it). + void markHandled(String key) => _lastHandledKey = key; + + /// Session left / host exited player: clears state and invalidates any + /// in-flight completion so a stale success can't suppress a later re-join + /// of the same media. + void reset() { + _generation++; + _inFlightKey = null; + _lastHandledKey = null; + } + + Future dispatch(String key, Future Function() invoke, {Duration timeout = dispatchTimeout}) async { + // Synchronous — claims the slot before the first await so a + // same-microtask second state can't double-dispatch. + _inFlightKey = key; + final generation = _generation; + var handled = false; + try { + // then re-types the future: a throwing async callback is + // reified as Future, whose timeout() rejects a bool onTimeout. + handled = await invoke().then((value) => value).timeout(timeout, onTimeout: () => false); + } catch (e, stackTrace) { + appLogger.w('WatchTogether: media switch dispatch failed for $key', error: e, stackTrace: stackTrace); + } + if (generation != _generation) return; // reset() happened mid-flight + _inFlightKey = null; + if (handled) _lastHandledKey = key; // else: next heartbeat retries + } +} diff --git a/lib/watch_together/services/guest_playback_reconciler.dart b/lib/watch_together/services/guest_playback_reconciler.dart index 968d1c02..b3bb56d1 100644 --- a/lib/watch_together/services/guest_playback_reconciler.dart +++ b/lib/watch_together/services/guest_playback_reconciler.dart @@ -10,6 +10,8 @@ import 'clock_sync.dart'; /// Callbacks the reconciler surfaces to the provider/UI layer. class GuestReconcilerCallbacks { /// The host's state names media we don't have loaded — navigate/reload. + /// Fires on EVERY such state (the heartbeat is the retry channel for + /// failed switches); the provider's dispatcher dedups. final void Function(String ratingKey, String serverId, String? mediaTitle)? onMediaSwitchNeeded; final void Function(ControlMode mode)? onControlModeChanged; @@ -226,7 +228,6 @@ class GuestPlaybackReconciler { void onState(PlaybackState state) { if (state.seq <= _lastSeq) return; // Stale or reordered. _lastSeq = state.seq; - final previous = _latestState; _latestState = state; if (state.controlMode != _reportedControlMode) { @@ -259,13 +260,10 @@ class GuestPlaybackReconciler { _sendStatus(force: true); } - // The host moved to media we don't have — hand off to the switch flow. - if (_attachedMediaKey != null && state.mediaKey != _attachedMediaKey) { - _callbacks.onMediaSwitchNeeded?.call(state.ratingKey, state.serverId, state.mediaTitle); - return; - } - if (previous?.mediaKey != state.mediaKey && _attachedMediaKey == null) { - // Not in the player yet — let the provider navigate. + // Not attached to the host's media (detached, or attached to something + // else) — hand off to the switch flow on every state so a failed switch + // retries on the next heartbeat. The provider's dispatcher dedups. + if (_attachedMediaKey == null || state.mediaKey != _attachedMediaKey) { _callbacks.onMediaSwitchNeeded?.call(state.ratingKey, state.serverId, state.mediaTitle); return; } diff --git a/lib/watch_together/services/watch_together_controller.dart b/lib/watch_together/services/watch_together_controller.dart index 5d78451c..57fb5e15 100644 --- a/lib/watch_together/services/watch_together_controller.dart +++ b/lib/watch_together/services/watch_together_controller.dart @@ -90,6 +90,7 @@ class WatchTogetherController { void Function(bool correcting)? onCorrectingChanged; void Function(ControlMode mode)? onControlModeReceived; void Function(String ratingKey, String serverId, String? mediaTitle)? onMediaStateReceived; + void Function()? onHostExitedPlayer; void Function(String peerId, PlaybackActionHint hint)? onRemoteAction; void Function(String peerId)? onPeerNeedsUpdate; void Function(List peerIds)? onResumedWithout; @@ -339,7 +340,12 @@ class WatchTogetherController { break; case SyncMessageType.hostExitedPlayer: - // Handled at the provider level. + // Rides the ordered queue so it can't locally overtake state + // messages that preceded it on the wire. Only the host may end the + // media epoch. + if (!_session.isHost && senderId == _session.hostPeerId) { + onHostExitedPlayer?.call(); + } break; } } diff --git a/test/watch_together/current_playback_dispatcher_test.dart b/test/watch_together/current_playback_dispatcher_test.dart new file mode 100644 index 00000000..5da6e392 --- /dev/null +++ b/test/watch_together/current_playback_dispatcher_test.dart @@ -0,0 +1,103 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:plezy/watch_together/services/current_playback_dispatcher.dart'; + +void main() { + group('CurrentPlaybackDispatcher', () { + test('success marks the key handled and suppresses re-dispatch', () async { + final d = CurrentPlaybackDispatcher(); + expect(d.shouldDispatch('a'), isTrue); + + await d.dispatch('a', () async => true); + expect(d.shouldDispatch('a'), isFalse); // Handled. + expect(d.shouldDispatch('b'), isTrue); // Other keys unaffected. + }); + + test('failure frees the slot without marking handled (heartbeat retry)', () async { + final d = CurrentPlaybackDispatcher(); + await d.dispatch('a', () async => false); + expect(d.inFlightKey, isNull); + expect(d.shouldDispatch('a'), isTrue); // Retryable. + }); + + test('a throwing callback is a failure, not an unhandled error', () async { + final d = CurrentPlaybackDispatcher(); + await expectLater(d.dispatch('a', () async => throw StateError('boom')), completes); + expect(d.shouldDispatch('a'), isTrue); + }); + + test('serializes: nothing dispatches while a key is in flight', () async { + final d = CurrentPlaybackDispatcher(); + final gate = Completer(); + final dispatch = d.dispatch('a', () => gate.future); + + // The slot is claimed synchronously, before the first await. + expect(d.inFlightKey, 'a'); + expect(d.shouldDispatch('a'), isFalse); + expect(d.shouldDispatch('b'), isFalse); + + gate.complete(true); + await dispatch; + expect(d.inFlightKey, isNull); + expect(d.shouldDispatch('b'), isTrue); + }); + + test('a hung callback times out as a failure and frees the slot', () { + fakeAsync((async) { + final d = CurrentPlaybackDispatcher(); + final never = Completer(); + unawaited(d.dispatch('a', () => never.future)); + + async.elapse(CurrentPlaybackDispatcher.dispatchTimeout - const Duration(seconds: 1)); + expect(d.inFlightKey, 'a'); + + async.elapse(const Duration(seconds: 2)); + expect(d.inFlightKey, isNull); + expect(d.shouldDispatch('a'), isTrue); // Timed out ⇒ unhandled. + + // A late success from the original callback changes nothing. + never.complete(true); + async.flushMicrotasks(); + expect(d.shouldDispatch('a'), isTrue); + }); + }); + + test('reset() mid-flight discards the stale completion', () async { + final d = CurrentPlaybackDispatcher(); + final gateA = Completer(); + final dispatchA = d.dispatch('a', () => gateA.future); + + d.reset(); // Host exited / session left. + expect(d.inFlightKey, isNull); + + // A newer dispatch claims the slot under the new generation. + final gateB = Completer(); + final dispatchB = d.dispatch('b', () => gateB.future); + + // The stale completion must neither mark 'a' handled nor free 'b'. + gateA.complete(true); + await dispatchA; + expect(d.shouldDispatch('a'), isFalse); // 'b' still occupies the slot... + expect(d.inFlightKey, 'b'); // ...untouched by the stale completion. + + gateB.complete(true); + await dispatchB; + expect(d.shouldDispatch('a'), isTrue); // 'a' was never marked handled. + expect(d.shouldDispatch('b'), isFalse); + }); + + test('markHandled suppresses a key without a dispatch (user-initiated join)', () { + final d = CurrentPlaybackDispatcher(); + d.markHandled('a'); + expect(d.shouldDispatch('a'), isFalse); + expect(d.shouldDispatch('b'), isTrue); + }); + + test('null keys never dispatch', () { + final d = CurrentPlaybackDispatcher(); + expect(d.shouldDispatch(null), isFalse); + }); + }); +} diff --git a/test/watch_together/guest_playback_reconciler_test.dart b/test/watch_together/guest_playback_reconciler_test.dart index 8b3ebb9e..fb82b0ad 100644 --- a/test/watch_together/guest_playback_reconciler_test.dart +++ b/test/watch_together/guest_playback_reconciler_test.dart @@ -364,6 +364,44 @@ void main() { }); }); + test('detached guest is re-notified on every state (heartbeat retry channel)', () { + fakeAsync((async) { + final switches = []; + final h = _Harness( + async, + callbacks: GuestReconcilerCallbacks(onMediaSwitchNeeded: (rk, sid, title) => switches.add(rk)), + ); + + // Never attached: every heartbeat re-offers the switch so a failed + // navigation can retry (the provider's dispatcher dedups). + h.reconciler.onState(h.state()); + h.reconciler.onState(h.state()); + h.reconciler.onState(h.state()); + async.flushMicrotasks(); + + expect(switches, ['rk1', 'rk1', 'rk1']); + expect(h.player.commandLog, isEmpty); + h.dispose(); + }); + }); + + test('attached to matching media never fires the switch callback', () { + fakeAsync((async) { + final switches = []; + final h = _Harness( + async, + callbacks: GuestReconcilerCallbacks(onMediaSwitchNeeded: (rk, sid, title) => switches.add(rk)), + ); + h.attachReady(); + + h.reconciler.onState(h.state()); + async.elapse(const Duration(seconds: 2)); + + expect(switches, isEmpty); + h.dispose(); + }); + }); + test('attach reconciles to the latest state received while detached', () { fakeAsync((async) { final h = _Harness(async); diff --git a/test/watch_together/watch_together_controller_test.dart b/test/watch_together/watch_together_controller_test.dart index 25e4de10..0acff94f 100644 --- a/test/watch_together/watch_together_controller_test.dart +++ b/test/watch_together/watch_together_controller_test.dart @@ -273,4 +273,56 @@ void main() { room.dispose(); }); }); + + group('hostExitedPlayer routing', () { + test('rides the ordered queue: never overtakes states sent before it', () { + fakeAsync((async) { + final room = _Room(async); + final log = []; + room.guest.onMediaStateReceived = (rk, sid, title) => log.add('state:$rk'); + room.guest.onHostExitedPlayer = () => log.add('hostExit'); + + // Host starts media, then exits the player — wire order matters. + room.hostStartsMedia(); + room.hostService.broadcast(SyncMessage.hostExitedPlayer(peerId: 'host')); + async.flushMicrotasks(); + + expect(log, isNotEmpty); + expect(log.first, 'state:rk1'); + expect(log.last, 'hostExit'); + room.dispose(); + }); + }); + + test('is ignored when forged by a non-host peer', () { + fakeAsync((async) { + final room = _Room(async); + var hostExits = 0; + room.guest.onHostExitedPlayer = () => hostExits++; + + final evil = room.hub.register('evil'); + evil.broadcast(SyncMessage.hostExitedPlayer(peerId: 'evil')); + async.flushMicrotasks(); + + expect(hostExits, 0); + room.dispose(); + }); + }); + + test('the host itself never reacts to a hostExitedPlayer echo', () { + fakeAsync((async) { + final room = _Room(async); + var hostExits = 0; + room.host.onHostExitedPlayer = () => hostExits++; + + // A confused/malicious guest sends the message; the host must not + // tear down its own epoch. + room.guestService.broadcast(SyncMessage.hostExitedPlayer(peerId: 'guest')); + async.flushMicrotasks(); + + expect(hostExits, 0); + room.dispose(); + }); + }); + }); } diff --git a/test/watch_together/watch_together_provider_test.dart b/test/watch_together/watch_together_provider_test.dart index 01a6e989..925fd4ac 100644 --- a/test/watch_together/watch_together_provider_test.dart +++ b/test/watch_together/watch_together_provider_test.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter_test/flutter_test.dart'; import 'package:plezy/media/ids.dart'; import 'package:plezy/watch_together/models/watch_session.dart'; @@ -142,6 +144,133 @@ void main() { }); }); + group('WatchTogetherProvider — media switch dispatch', () { + test('dispatches once with typed args and suppresses the key after success', () async { + final p = WatchTogetherProvider(); + final calls = <(String, String, String)>[]; + p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async { + calls.add((ratingKey, serverId, mediaTitle)); + return true; + }; + + p.debugHandleMediaState('rk1', 's1', 'Ep 1'); + await Future.delayed(Duration.zero); + expect(calls, [('rk1', 's1', 'Ep 1')]); + + // Heartbeat repeat of the handled key: no re-dispatch. + p.debugHandleMediaState('rk1', 's1', 'Ep 1'); + await Future.delayed(Duration.zero); + expect(calls.length, 1); + p.dispose(); + }); + + test('a false result is retried on the next heartbeat state', () async { + final p = WatchTogetherProvider(); + var calls = 0; + p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async { + calls++; + return calls > 1; // Fail once, then succeed. + }; + + p.debugHandleMediaState('rk1', 's1', null); + await Future.delayed(Duration.zero); + p.debugHandleMediaState('rk1', 's1', null); + await Future.delayed(Duration.zero); + expect(calls, 2); + + p.debugHandleMediaState('rk1', 's1', null); + await Future.delayed(Duration.zero); + expect(calls, 2); // Second attempt succeeded; key now handled. + p.dispose(); + }); + + test('a throwing callback is contained and retried', () async { + final p = WatchTogetherProvider(); + var calls = 0; + p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async { + calls++; + throw StateError('network down'); + }; + + expect(() => p.debugHandleMediaState('rk1', 's1', null), returnsNormally); + await Future.delayed(Duration.zero); + p.debugHandleMediaState('rk1', 's1', null); + await Future.delayed(Duration.zero); + expect(calls, 2); + p.dispose(); + }); + + test('no double dispatch while a switch is pending, even for another key', () async { + final p = WatchTogetherProvider(); + final pending = Completer(); + final calls = []; + p.onMediaSwitched = (ratingKey, serverId, mediaTitle) { + calls.add(ratingKey); + return pending.future; + }; + + p.debugHandleMediaState('rk1', 's1', null); + p.debugHandleMediaState('rk1', 's1', null); + p.debugHandleMediaState('rk2', 's1', null); // Serialized behind rk1. + await Future.delayed(Duration.zero); + expect(calls, ['rk1']); + + pending.complete(false); + await Future.delayed(Duration.zero); + // The slot is free again; the next heartbeat re-dispatches. + p.debugHandleMediaState('rk2', 's1', null); + await Future.delayed(Duration.zero); + expect(calls, ['rk1', 'rk2']); + p.dispose(); + }); + + test('onPlayerMediaSwitched takes priority over onMediaSwitched', () async { + final p = WatchTogetherProvider(); + final calls = []; + p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async { + calls.add('main'); + return true; + }; + p.onPlayerMediaSwitched = (ratingKey, serverId, mediaTitle) async { + calls.add('player'); + return true; + }; + + p.debugHandleMediaState('rk1', 's1', null); + await Future.delayed(Duration.zero); + expect(calls, ['player']); + p.dispose(); + }); + + test('markCurrentPlaybackHandled suppresses the marked key', () async { + final p = WatchTogetherProvider(); + var calls = 0; + p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async { + calls++; + return true; + }; + + p.markCurrentPlaybackHandled(ratingKey: 'rk1', serverId: ServerId('s1')); + p.debugHandleMediaState('rk1', 's1', null); + await Future.delayed(Duration.zero); + expect(calls, 0); + p.dispose(); + }); + + test('a blank serverId is ignored without throwing', () { + final p = WatchTogetherProvider(); + var calls = 0; + p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async { + calls++; + return true; + }; + + expect(() => p.debugHandleMediaState('rk1', '', null), returnsNormally); + expect(calls, 0); + p.dispose(); + }); + }); + group('WatchTogetherProvider — leaveSession safety', () { test('leaveSession on a fresh provider is a no-op (no notify)', () async { final p = WatchTogetherProvider();