diff --git a/lib/watch_together/services/attached_player.dart b/lib/watch_together/services/attached_player.dart index e1c6cf378..42b20e909 100644 --- a/lib/watch_together/services/attached_player.dart +++ b/lib/watch_together/services/attached_player.dart @@ -35,7 +35,10 @@ class _Expectation { /// [Player.currentPosition], not the throttled state). /// /// The session controller creates one instance per attachment and disposes -/// it on detach — instance lifecycle *is* the staleness guard. +/// it on detach. The instance outlives a host-role swap: the engine that +/// issued a command may be gone by the time its acknowledgement arrives, so +/// the ledger stays with the attachment and [playingAcks] lets the engine +/// currently in charge account for what its predecessor set in motion. class AttachedPlayer { AttachedPlayer({required Player player, required this._onLost, this._remoteSeek, int Function()? nowMs}) : _player = player, @@ -65,6 +68,7 @@ class AttachedPlayer { final List<_Expectation> _expectations = []; final _playingIntentsController = StreamController.broadcast(); + final _playingAcksController = StreamController.broadcast(); final _bufferingChangesController = StreamController.broadcast(); final _loadedSignalsController = StreamController.broadcast(); @@ -76,6 +80,10 @@ class AttachedPlayer { /// User-initiated play/pause transitions (command acks are filtered out). Stream get playingIntents => _playingIntentsController.stream; + /// Playing transitions consumed as command acknowledgements — including + /// acks for commands issued by a previous role engine on this attachment. + Stream get playingAcks => _playingAcksController.stream; + /// Raw buffering transitions (`paused-for-cache`). Stream get bufferingChanges => _bufferingChangesController.stream; @@ -248,7 +256,10 @@ class AttachedPlayer { void _onPlayingEvent(bool value) { if (_disposed || value == _lastPlaying) return; _lastPlaying = value; - if (_consumePlayingExpectation(value)) return; + if (_consumePlayingExpectation(value)) { + _playingAcksController.add(value); + return; + } _playingIntentsController.add(value); } @@ -268,6 +279,7 @@ class AttachedPlayer { unawaited(subscription.cancel()); } await _playingIntentsController.close(); + await _playingAcksController.close(); await _bufferingChangesController.close(); await _loadedSignalsController.close(); } diff --git a/lib/watch_together/services/guest_playback_reconciler.dart b/lib/watch_together/services/guest_playback_reconciler.dart index 0b456775b..aeb1b6126 100644 --- a/lib/watch_together/services/guest_playback_reconciler.dart +++ b/lib/watch_together/services/guest_playback_reconciler.dart @@ -94,6 +94,12 @@ class GuestPlaybackReconciler { bool _backgrounded = false; bool _disposed = false; + /// Generation of this engine's authority over its player. Bumped on + /// attach, detach, epoch end, and disposal; every async continuation + /// captures it before its await and bails when stale, so a command issued + /// as a guest never lands on a player that has since changed hands. + int _authority = 0; + // Correction state. bool _settling = false; Timer? _settleTimer; @@ -142,6 +148,7 @@ class GuestPlaybackReconciler { Future? startupHold, }) { detachPlayer(); + final authority = _authority; _player = player; _attachedMediaKey = PlaybackState.mediaKeyFor(ratingKey: ratingKey, serverId: serverId); _firstFrameSeen = hasFirstFrame; @@ -149,7 +156,7 @@ class GuestPlaybackReconciler { if (startupHold != null) { startupHold.then((_) { - if (_disposed || !identical(_player, player)) return; + if (authority != _authority) return; _startupHoldResolved = true; _maybeBecomeReady(); }); @@ -192,6 +199,7 @@ class GuestPlaybackReconciler { } void detachPlayer() { + _authority++; for (final subscription in _playerSubscriptions) { unawaited(subscription.cancel()); } @@ -242,6 +250,22 @@ class GuestPlaybackReconciler { _setCorrecting(false); } + /// The host left the video player: the media epoch is over. The last state + /// no longer describes a room anyone authors, so a promotion after this + /// adopts nothing and starts clean. Sequence numbering is kept — the same + /// host may open the next item. + void endEpoch() { + if (_latestState == null) return; + _authority++; + _latestState = null; + _reportedPhase = null; + _reportedWaitingOn = const []; + _scheduledStartTimer?.cancel(); + _scheduledStartTimer = null; + _scheduledStartSeq = null; + appLogger.d('WatchTogether: Room epoch ended by the host'); + } + /// Latest authoritative state from the host (already host-authenticated). void onState(PlaybackState state) { if (state.seq <= _lastSeq) return; // Stale or reordered. @@ -354,6 +378,9 @@ class GuestPlaybackReconciler { } void dispose() { + // Revoke first: continuations still in flight must find the authority + // gone before the player they were issued on is handed over. + _authority++; _disposed = true; detachPlayer(); } @@ -519,9 +546,12 @@ class GuestPlaybackReconciler { _setCorrecting(true); _beginSettle(); appLogger.d('WatchTogether: Hard sync seek to ${targetMs}ms'); + final authority = _authority; unawaited( player.seek(Duration(milliseconds: targetMs.clamp(0, 1 << 48))).then((didSeek) async { - if (didSeek && thenPlay) await player.play(); + // Authority moved while the seek was in flight (a promotion, a detach): the play it was + // leading into would land on a player this engine no longer speaks for. + if (didSeek && thenPlay && authority == _authority) await player.play(); }), ); } diff --git a/lib/watch_together/services/host_playback_coordinator.dart b/lib/watch_together/services/host_playback_coordinator.dart index 50569f875..f9873a3a4 100644 --- a/lib/watch_together/services/host_playback_coordinator.dart +++ b/lib/watch_together/services/host_playback_coordinator.dart @@ -128,11 +128,18 @@ class HostPlaybackCoordinator { bool _backgrounded = false; // Host-transfer handover: the room position inherited from the previous - // host, held until the promoted player has aligned to it (see [attach]). + // host, held until the promoted player has aligned to it (see [adoptRoom]). int? _transitionAnchorMs; bool _transitionAlignSeekInFlight = false; Timer? _transitionAlignTimer; + /// Generation of this engine's authority over its player and epoch. + /// Bumped whenever what an in-flight command was issued against no longer + /// holds — attach, detach, epoch change, room adoption, disposal. Every + /// async continuation captures it before its await and bails when stale: + /// the one fence for commands outliving the binding they were issued on. + int _authority = 0; + // Peer tracking. final Set _knownPeers = {}; final Set _incompatiblePeers = {}; @@ -180,6 +187,7 @@ class HostPlaybackCoordinator { _ratingKey = ratingKey; _serverId = serverId; _mediaTitle = mediaTitle; + _authority++; _localReady = false; _localStalled = false; _recoveringFromSelfStall = false; @@ -195,46 +203,73 @@ class HostPlaybackCoordinator { appLogger.d('WatchTogether: Host epoch -> $newKey'); } + /// Take over a room where the previous host left it: epoch identity, rate, + /// intent, phase, and the room's position at this moment ([anchorMs], on + /// the old host's clock). Independent of a player binding — a peer promoted + /// while detached still becomes the room's authority and answers for it — + /// and broadcast at once so every guest learns the new author. A later + /// [attach] for the same media rebinds into this epoch; a different media + /// is a deliberate selection and starts a fresh one. + /// + /// The inherited position is held until the promoted player has rendered, + /// passed its startup hold, and aligned to it through the normal seek path + /// (readiness and alignment are different states: the player may sit mid + /// correction or at a stale pre-seek spot). Until then every broadcast + /// carries the anchor and this host gates the room; only then does its + /// local position become the room's. A room still loading had no position + /// worth inheriting. + void adoptRoom(PlaybackState state, {int? anchorMs}) { + _authority++; + _ratingKey = state.ratingKey; + _serverId = state.serverId; + _mediaTitle = state.mediaTitle; + // The room's rate, not this player's: it may be mid-nudge. + _rate = state.rate; + // A paused room must not start playing just because its host changed; + // anything else (playing, a stall, mid-load) carries the intent to + // (re)start. + _intendedPlaying = state.phase != PlaybackPhase.paused; + // Whether the room has started once decides who gates the next start: + // a room caught playing has, and its remaining peers are mid-session + // followers; a room stopped anywhere else re-gates on everyone it knows. + _firstStartCompleted = state.phase == PlaybackPhase.playing; + _clearTransitionAnchor(); + _transitionAnchorMs = state.phase == PlaybackPhase.loading ? null : anchorMs; + _setPhase(switch (state.phase) { + PlaybackPhase.paused => PlaybackPhase.paused, + PlaybackPhase.loading => PlaybackPhase.loading, + PlaybackPhase.waitingForPeers || PlaybackPhase.playing => PlaybackPhase.waitingForPeers, + }); + _broadcast(); + appLogger.d('WatchTogether: Adopted room $_mediaKey (${state.phase.name}, rate $_rate, anchor ${anchorMs}ms)'); + } + /// Attach the host's player for the given media. [hasFirstFrame] is the /// screen's first-frame snapshot (covers attaching to an already-rendering /// player); [startupHold] delays readiness past platform startup gates /// (e.g. the Android frame-rate switch). /// - /// [intendPlaying] optionally adopts another host's room intent. Ordinary - /// attachments omit it: a new media epoch implies play, while rebinding an - /// existing epoch preserves its intent, including commands while detached. + /// A new media epoch implies play, while rebinding an existing epoch + /// (quality/version reload, or a player bound after [adoptRoom]) preserves + /// its intent, including commands accepted while detached. /// /// [rate] seeds the room rate for a new epoch: a fresh host passes its /// resolved default speed (the saved preference for this item, resolved /// by the screen before readiness so no later track-selection pass has to - /// apply it underneath the room), and a promoted host passes the room's - /// last broadcast rate, because its player may be mid-correction and its - /// momentary rate is not what the room agreed on. A same-epoch re-attach - /// (quality/version reload) keeps the room's current rate: the reload - /// must not reset a rate the room has since agreed on. The host player is - /// the room clock, so the seeded rate is also applied to it — a - /// coordinator broadcasting a rate its own player is not running is a - /// permanent drift every guest keeps correcting against. - /// - /// [transitionAnchorMs] is the room's authoritative position at the moment - /// this coordinator took the room over (a promotion), computed on the old - /// host's clock. Readiness and timeline alignment are different states: - /// the promoted player may have rendered long ago and still sit mid - /// correction, drifted, or at a stale pre-seek spot. Until it has rendered, - /// passed its startup hold, and aligned to that position through the - /// normal seek path, every broadcast carries the inherited anchor and this - /// host gates the room; only then does its local position become the - /// room's. Null for every other attachment. + /// apply it underneath the room). A same-epoch re-attach keeps the room's + /// current rate: neither a reload nor a rebind after promotion may reset a + /// rate the room has agreed on. The host player is the room clock, so the + /// room rate is also applied to it — a coordinator broadcasting a rate its + /// own player is not running is a permanent drift every guest keeps + /// correcting against. void attach( AttachedPlayer player, { required String ratingKey, required String serverId, String? mediaTitle, bool hasFirstFrame = false, - bool? intendPlaying, double? rate, Future? startupHold, - int? transitionAnchorMs, }) { // A player replaced mid-handover (a seek-induced reload) inherits the // anchor still being aligned to; the replacement aligns instead. @@ -245,9 +280,8 @@ class HostPlaybackCoordinator { // Seed before the epoch broadcast so the loading state already carries // the room rate guests will be asked to run. if (!sameEpoch) _rate = rate ?? player.rate; - _transitionAnchorMs = transitionAnchorMs ?? (sameEpoch ? heldAnchorMs : null); + _transitionAnchorMs = sameEpoch ? heldAnchorMs : null; setLocalMedia(ratingKey: ratingKey, serverId: serverId, mediaTitle: mediaTitle); - if (intendPlaying != null) _intendedPlaying = intendPlaying; _player = player; if ((player.rate - _rate).abs() > 0.001) { @@ -264,10 +298,11 @@ class HostPlaybackCoordinator { _armSafetyIfGated(); } + final authority = _authority; _startupHoldResolved = startupHold == null; if (startupHold != null) { startupHold.then((_) { - if (_disposed || !identical(_player, player)) return; + if (authority != _authority) return; _startupHoldResolved = true; _maybeLocalLoaded(); }); @@ -276,6 +311,7 @@ class HostPlaybackCoordinator { _playerSubscriptions.add(player.loadedSignals.listen((_) => _onLoadedSignal())); _playerSubscriptions.add(player.bufferingChanges.listen(_onSelfBuffering)); _playerSubscriptions.add(player.playingIntents.listen(_onLocalPlayingIntent)); + _playerSubscriptions.add(player.playingAcks.listen(_onAcknowledgedPlaying)); unawaited(player.setCachePauseWait(hostCachePauseWait)); if (hasFirstFrame) { @@ -288,6 +324,7 @@ class HostPlaybackCoordinator { /// Detach the player (episode switch keeps the session; [exiting] ends the /// epoch because the host left the video player). void detachPlayer({bool exiting = false}) { + _authority++; for (final subscription in _playerSubscriptions) { unawaited(subscription.cancel()); } @@ -448,6 +485,9 @@ class HostPlaybackCoordinator { } void dispose() { + // Revoke first: continuations still in flight must find the authority + // gone before anything below tears down what they would touch. + _authority++; _disposed = true; detachPlayer(exiting: true); _allReadyCheckTimer?.cancel(); @@ -518,9 +558,10 @@ class HostPlaybackCoordinator { _transitionAlignTimer = null; if (_transitionAlignSeekInFlight) _finishTransitionAlignment(reason: 'seek render timeout'); }); + final authority = _authority; unawaited( player.seek(Duration(milliseconds: anchorMs)).then((didSeek) { - if (_disposed || !identical(_player, player) || !_transitionAlignSeekInFlight) return; + if (authority != _authority || !_transitionAlignSeekInFlight) return; if (!didSeek) { // The seek was refused: publishing a position this host is not at // would pin every guest to it. Its real position takes over and the @@ -599,6 +640,19 @@ class HostPlaybackCoordinator { } } + /// A play this attachment commanded was acknowledged. The command may have + /// been issued by the engine this one replaced (a guest reconciler mid + /// correction when authority moved): a host player running under a stopped + /// room would broadcast a moving position every heartbeat, so the room's + /// phase wins and the player is put back. + void _onAcknowledgedPlaying(bool playing) { + if (!playing || _phase == PlaybackPhase.playing) return; + final player = _player; + if (player == null || !player.playing) return; + appLogger.d('WatchTogether: Late play acknowledged while ${_phase.name}; pausing'); + unawaited(player.pause()); + } + /// User rate change on the host (the screen already applied it locally). /// Declared by the screen, never inferred from the player's rate stream. void onLocalRateIntent(double rate) { @@ -670,9 +724,10 @@ class HostPlaybackCoordinator { final durationMs = player.duration.inMilliseconds; if (durationMs <= 0 || targetMs < 0 || targetMs > durationMs) return; onRemoteAction?.call(actor, PlaybackActionHint.seek); + final authority = _authority; unawaited( player.seek(Duration(milliseconds: targetMs)).then((didSeek) { - if (didSeek) _afterHostSeek(targetMs, actor: actor); + if (didSeek && authority == _authority) _afterHostSeek(targetMs, actor: actor); }), ); } @@ -689,9 +744,10 @@ class HostPlaybackCoordinator { void _applyRemoteRate(double rate, {required String actor}) { final player = _player; if (player == null || !rate.isFinite || rate < _minimumRemoteRate || rate > _maximumRemoteRate) return; + final authority = _authority; unawaited( player.setRate(rate).then((didSet) { - if (!didSet) return; + if (!didSet || authority != _authority) return; _rate = rate; appLogger.d('WatchTogether: Room rate $rate applied for $actor'); // Reported after the rate is committed so listeners reading [rate] @@ -849,6 +905,7 @@ class HostPlaybackCoordinator { _pendingStartPositionMs = null; final currentPlayer = _player; if (currentPlayer == null || _phase != PlaybackPhase.playing) return; + final authority = _authority; // A vehicle that requires distraction optimization refuses the play. The room would // otherwise sit in a playing phase the host is not honouring, with nothing to correct it, so // put it back to paused: the host is the authority on what it is actually doing. @@ -856,14 +913,21 @@ class HostPlaybackCoordinator { // Only this attachment's own refusal counts. A reload detaches mid-flight and its disposed // player also answers false, but that pause belongs to the source switch, which deliberately // holds the phase so the replacement can pick the room up again. - if (started || _player != currentPlayer || _phase != PlaybackPhase.playing) return; + if (started || authority != _authority || _phase != PlaybackPhase.playing) return; _intendedPlaying = false; _setPhase(PlaybackPhase.paused); _broadcast(); } if (startPos != null && (currentPlayer.position.inMilliseconds - startPos).abs() > 250) { - unawaited(currentPlayer.seek(Duration(milliseconds: startPos)).then((_) => currentPlayer.play()).then(settle)); + unawaited( + currentPlayer + .seek(Duration(milliseconds: startPos)) + // The seek outlived this engine's authority (a demotion, a reload): the play it was + // leading into belongs to nobody now. + .then((_) => authority == _authority ? currentPlayer.play() : Future.value(false)) + .then(settle), + ); } else { unawaited(currentPlayer.play().then(settle)); } diff --git a/lib/watch_together/services/watch_together_controller.dart b/lib/watch_together/services/watch_together_controller.dart index 5d6cf1af4..ce5eaf4ed 100644 --- a/lib/watch_together/services/watch_together_controller.dart +++ b/lib/watch_together/services/watch_together_controller.dart @@ -151,10 +151,13 @@ class WatchTogetherController { final attached = _attachedPlayer; if (session.isHost) { - // Promotion: adopt the room where the old host left it. The position - // is read on the old host's clock before that clock is discarded — - // the reconciler's own player may be mid-correction, and a stale - // local snapshot would become the room's authoritative anchor. + // Promotion: adopt the room where the old host left it, whether or not + // a player is bound right now (a reload gap, the lobby, an episode + // switch). The position is read on the old host's clock before that + // clock is discarded — the reconciler's own player may be + // mid-correction, and a stale local snapshot would become the room's + // authoritative anchor. A player bound later for the same media rebinds + // into the adopted epoch instead of opening a new one. final lastState = _reconciler?.latestState; final firstFrameSeen = _reconciler?.firstFrameSeen ?? false; final transitionAnchorMs = lastState?.targetPositionMs(_clockSync?.hostNowMs() ?? _nowMs()); @@ -163,6 +166,7 @@ class WatchTogetherController { _reconciler?.dispose(); _reconciler = null; _createCoordinator(); + if (lastState != null) _coordinator!.adoptRoom(lastState, anchorMs: transitionAnchorMs); if (attached != null && _attachedRatingKey != null && _attachedServerId != null) { _coordinator!.attach( attached, @@ -171,17 +175,7 @@ class WatchTogetherController { mediaTitle: _attachedMediaTitle, hasFirstFrame: firstFrameSeen, startupHold: _attachedStartupHold, - // A paused room must not start playing just because its host - // changed; anything else (playing, a stall, mid-load) carries the - // intent to (re)start. - intendPlaying: lastState == null || lastState.phase != PlaybackPhase.paused, - // The room's rate, not this player's: it may be mid-nudge. A room - // that never broadcast one falls back to the rate this attachment - // declared. - rate: lastState?.rate ?? _attachedRate, - // Only a room with a meaningful timeline hands one over: a host - // that was still loading had no position worth inheriting. - transitionAnchorMs: lastState?.phase == PlaybackPhase.loading ? null : transitionAnchorMs, + rate: _attachedRate, ); } } else { @@ -508,6 +502,7 @@ class WatchTogetherController { // messages that preceded it on the wire. Only the host may end the // media epoch. if (!_session.isHost && senderId == _session.hostPeerId) { + _reconciler?.endEpoch(); onHostExitedPlayer?.call(); } break; diff --git a/test/watch_together/watch_together_controller_test.dart b/test/watch_together/watch_together_controller_test.dart index bbb74b7cf..0cf02d84d 100644 --- a/test/watch_together/watch_together_controller_test.dart +++ b/test/watch_together/watch_together_controller_test.dart @@ -826,10 +826,14 @@ void main() { }); group('host transfer', () { - WatchSession sessionAs(SessionRole role, {String hostPeerId = 'guest'}) => WatchSession( + WatchSession sessionAs( + SessionRole role, { + String hostPeerId = 'guest', + ControlMode controlMode = ControlMode.hostOnly, + }) => WatchSession( sessionId: 'ROOM1', role: role, - controlMode: ControlMode.hostOnly, + controlMode: controlMode, state: SessionState.connected, hostPeerId: hostPeerId, ); @@ -1016,7 +1020,12 @@ void main() { async.flushMicrotasks(); async.elapse(const Duration(seconds: 3)); - expect(room.guest.phase, PlaybackPhase.loading); + // The adopted room waits on the promoted host itself until its hold + // releases: readiness is a prerequisite it re-gates the room on. + expect(room.guest.phase, PlaybackPhase.waitingForPeers); + expect(room.guestService.outgoingLog.lastWhere((m) => m.type == SyncMessageType.state).state!.waitingOn, [ + 'guest', + ]); expect(room.hostPlayer.commandLog.where((c) => c == 'play'), isEmpty); expect(room.guestPlayer.commandLog.where((c) => c == 'play'), isEmpty); @@ -1081,7 +1090,10 @@ void main() { room.guest.applyHostChange(sessionAs(SessionRole.host)); async.flushMicrotasks(); async.elapse(const Duration(seconds: 3)); - expect(room.guest.phase, PlaybackPhase.loading); + expect(room.guest.phase, PlaybackPhase.waitingForPeers); + expect(room.guestService.outgoingLog.lastWhere((m) => m.type == SyncMessageType.state).state!.waitingOn, [ + 'guest', + ]); expect(room.guestPlayer.commandLog.where((c) => c == 'play'), isEmpty); expect(room.hostPlayer.commandLog.where((c) => c == 'play'), isEmpty); @@ -1318,5 +1330,163 @@ void main() { room.dispose(); }); }); + + test('promotion while the guest is detached adopts the room epoch', () { + fakeAsync((async) { + final room = _Room(async); + room.hostStartsMedia(); + room.guestJoinsMedia(); + room.bothBecomeReady(); + async.elapse(const Duration(seconds: 3)); + room.host.onLocalRate(1.25); + async.flushMicrotasks(); + async.elapse(const Duration(seconds: 1)); + final roomBefore = room.lastHostState(); + expect(roomBefore.phase, PlaybackPhase.playing); + final expectedMs = roomBefore.targetPositionMs(room.nowMs()); + + // The guest is between players (a reload gap, the lobby) when the + // relay hands it the room. + room.guest.detachPlayer(); + async.flushMicrotasks(); + room.host.applyHostChange(sessionAs(SessionRole.guest)); + room.guest.applyHostChange(sessionAs(SessionRole.host)); + async.flushMicrotasks(); + + // It is the room's authority nonetheless: the epoch, rate, and + // position carry over, and the room waits on the new host to bind. + final adopted = room.guestService.outgoingLog.lastWhere((m) => m.type == SyncMessageType.state).state!; + expect(adopted.mediaKey, roomBefore.mediaKey); + expect(adopted.phase, PlaybackPhase.waitingForPeers); + expect(adopted.waitingOn, ['guest']); + expect((adopted.anchorPositionMs - expectedMs).abs(), lessThanOrEqualTo(100)); + expect(adopted.rate, 1.25); + // The demoted host asked for state and got an answer. + expect(room.host.phase, PlaybackPhase.waitingForPeers); + expect(room.host.roomRate, 1.25); + room.dispose(); + }); + }); + + test('a paused room rebound after a detached promotion stays paused at its position', () { + fakeAsync((async) { + final room = _Room(async); + room.hostStartsMedia(); + room.guestJoinsMedia(); + room.bothBecomeReady(); + async.elapse(const Duration(seconds: 3)); + room.hostPlayer.emitPlaying(false); + async.flushMicrotasks(); + async.elapse(const Duration(seconds: 1)); + final paused = room.lastHostState(); + expect(paused.phase, PlaybackPhase.paused); + final pausedAt = paused.anchorPositionMs; + + room.guest.detachPlayer(); + async.flushMicrotasks(); + room.host.applyHostChange(sessionAs(SessionRole.guest)); + room.guest.applyHostChange(sessionAs(SessionRole.host)); + async.flushMicrotasks(); + + // The player comes back for the same item with this peer's own saved + // speed and somewhere else on the timeline: it rebinds into the + // adopted epoch rather than opening a new one. + room.guestPlayer.setPosition(const Duration(seconds: 5)); + room.guestPlayer.commandLog.clear(); + room.guest.attachPlayer(room.guestPlayer, ratingKey: 'rk1', serverId: 'srv', hasFirstFrame: true, rate: 1.5); + async.flushMicrotasks(); + async.elapse(const Duration(seconds: 6)); + + final rebound = room.guestService.outgoingLog.lastWhere((m) => m.type == SyncMessageType.state).state!; + expect(rebound.phase, PlaybackPhase.paused, reason: 'a rebind is not a play intent'); + expect(rebound.anchorPositionMs, pausedAt); + expect(rebound.rate, 1.0, reason: 'the room rate outranks the saved preference'); + expect(room.guestPlayer.state.position.inMilliseconds, pausedAt); + expect(room.guestPlayer.state.rate, 1.0); + expect(room.guestPlayer.commandLog.where((c) => c == 'play'), isEmpty); + expect(room.guestPlayer.state.playing, isFalse); + expect(room.hostPlayer.state.playing, isFalse); + room.dispose(); + }); + }); + + test('an old-role seek continuation does not play the promoted host\'s player', () { + fakeAsync((async) { + final room = _Room(async); + room.hostStartsMedia(); + room.guestJoinsMedia(); + room.bothBecomeReady(); + async.elapse(const Duration(seconds: 3)); + expect(room.guestPlayer.state.playing, isTrue); + + // The guest rolled off the end while the room plays on: its + // reconciler rejoins with a seek that is to be followed by a play. + // The seek is still outstanding when authority moves. + final pendingSeek = Completer(); + room.guestPlayer.setCompleted(true); + room.guestPlayer.nextCommandFuture = pendingSeek.future; + room.guestPlayer.emitPlaying(false); + async.flushMicrotasks(); + expect(room.guestPlayer.commandLog.last, startsWith('seek:')); + room.guestPlayer.commandLog.clear(); + + room.host.applyHostChange(sessionAs(SessionRole.guest)); + room.guest.applyHostChange(sessionAs(SessionRole.host)); + async.flushMicrotasks(); + pendingSeek.complete(); + async.flushMicrotasks(); + + // The guest-era play never lands: the host player is the room clock + // and only the coordinator's group start may run it. + expect(room.guestPlayer.commandLog.where((c) => c == 'play'), isEmpty); + expect(room.guestPlayer.state.playing, isFalse); + final scheduled = room.guestService.outgoingLog.lastWhere((m) => m.type == SyncMessageType.state).state!; + expect(scheduled.phase == PlaybackPhase.waitingForPeers || scheduled.anchorHostTimeMs > room.nowMs(), isTrue); + + // Not a deadlock: the room starts once, on the coordinator's schedule. + async.elapse(const Duration(seconds: 3)); + expect(room.guestPlayer.commandLog.where((c) => c == 'play'), hasLength(1)); + expect(room.guestPlayer.state.playing, isTrue); + expect(room.hostPlayer.state.playing, isTrue); + room.dispose(); + }); + }); + + test('a demoted host\'s rate continuation reports nothing', () { + fakeAsync((async) { + final room = _Room(async, controlMode: ControlMode.anyone, guestControlMode: ControlMode.anyone); + final oldHostActions = <(String, PlaybackActionHint)>[]; + room.host.onRemoteAction = (peer, hint) => oldHostActions.add((peer, hint)); + room.hostStartsMedia(); + room.guestJoinsMedia(); + room.bothBecomeReady(); + async.elapse(const Duration(seconds: 3)); + + // The guest asks for a rate; the host's player is still applying it + // when the host is demoted. + final pendingRate = Completer(); + room.hostPlayer.nextCommandFuture = pendingRate.future; + room.guest.onLocalRate(1.5); + async.flushMicrotasks(); + expect(room.hostPlayer.commandLog.last, 'rate:1.5'); + final hostStatesBefore = room.hostService.outgoingLog.where((m) => m.type == SyncMessageType.state).length; + + room.host.applyHostChange(sessionAs(SessionRole.guest, controlMode: ControlMode.anyone)); + room.guest.applyHostChange(sessionAs(SessionRole.host, controlMode: ControlMode.anyone)); + async.flushMicrotasks(); + pendingRate.complete(); + async.flushMicrotasks(); + async.elapse(const Duration(seconds: 1)); + + // A rate applied by an engine that no longer speaks for the room is + // neither announced nor published; the room rate is the new host's. + // (The new host's own restart is announced as usual.) + expect(oldHostActions.where((a) => a.$2 == PlaybackActionHint.rate), isEmpty); + expect(room.hostService.outgoingLog.where((m) => m.type == SyncMessageType.state).length, hostStatesBefore); + expect(room.host.roomRate, 1.0); + expect(room.guestService.outgoingLog.lastWhere((m) => m.type == SyncMessageType.state).state!.rate, 1.0); + room.dispose(); + }); + }); }); }