From e3fbe6d92542229dfbebb22de9f5fca35faaca94 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Sun, 25 Jan 2026 18:47:48 +0100 Subject: [PATCH] fix(watch-together): improve sync and buffering handling --- lib/watch_together/models/sync_message.dart | 11 +- .../services/watch_together_peer_service.dart | 106 ++++++++++++++++- .../services/watch_together_sync_manager.dart | 111 ++++++++++++++---- 3 files changed, 203 insertions(+), 25 deletions(-) diff --git a/lib/watch_together/models/sync_message.dart b/lib/watch_together/models/sync_message.dart index dd4888df..cf2a23c6 100644 --- a/lib/watch_together/models/sync_message.dart +++ b/lib/watch_together/models/sync_message.dart @@ -88,6 +88,9 @@ class SyncMessage { /// Title of the media (for mediaSwitch message) final String? mediaTitle; + /// Whether playback is currently playing (for positionSync heartbeat) + final bool? isPlaying; + const SyncMessage({ required this.type, required this.timestamp, @@ -102,6 +105,7 @@ class SyncMessage { this.ratingKey, this.serverId, this.mediaTitle, + this.isPlaying, }); /// Create a PLAY message @@ -139,13 +143,14 @@ class SyncMessage { ); } - /// Create a POSITION_SYNC message - factory SyncMessage.positionSync(Duration position, {String? peerId}) { + /// Create a POSITION_SYNC message (heartbeat with optional play/pause state) + factory SyncMessage.positionSync(Duration position, {String? peerId, bool? isPlaying}) { return SyncMessage( type: SyncMessageType.positionSync, timestamp: DateTime.now().millisecondsSinceEpoch, positionMs: position.inMilliseconds, peerId: peerId, + isPlaying: isPlaying, ); } @@ -268,6 +273,7 @@ class SyncMessage { if (ratingKey != null) map['rk'] = ratingKey; if (serverId != null) map['sid'] = serverId; if (mediaTitle != null) map['title'] = mediaTitle; + if (isPlaying != null) map['pl'] = isPlaying; return jsonEncode(map); } @@ -296,6 +302,7 @@ class SyncMessage { ratingKey: map['rk'] as String?, serverId: map['sid'] as String?, mediaTitle: map['title'] as String?, + isPlaying: map['pl'] as bool?, ); } diff --git a/lib/watch_together/services/watch_together_peer_service.dart b/lib/watch_together/services/watch_together_peer_service.dart index 3d7eb03f..f9fff5f4 100644 --- a/lib/watch_together/services/watch_together_peer_service.dart +++ b/lib/watch_together/services/watch_together_peer_service.dart @@ -42,11 +42,19 @@ class WatchTogetherPeerService { final _errorController = StreamController.broadcast(); final _connectionStateController = StreamController.broadcast(); - // Reconnection state + // Reconnection state (signaling server) int _reconnectAttempts = 0; static const int _maxReconnectAttempts = 3; Timer? _reconnectTimer; + // Peer health monitoring (data channel) + final Map _lastPeerActivity = {}; + final Map _peerReconnectAttempts = {}; + Timer? _peerHealthCheckTimer; + static const Duration _peerTimeout = Duration(seconds: 30); + static const Duration _peerHealthCheckInterval = Duration(seconds: 10); + static const int _maxPeerReconnectAttempts = 3; + /// Stream of peer IDs when a new peer connects Stream get onPeerConnected => _peerConnectedController.stream; @@ -218,9 +226,15 @@ class WatchTogetherPeerService { conn.on('open').listen((_) { appLogger.d('WatchTogether: Data channel opened with: $peerId'); _connections[peerId] = conn; + _updatePeerActivity(peerId); // Track initial activity _peerConnectedController.add(peerId); _connectionStateController.add(true); + // Start health monitoring if not already running + if (_peerHealthCheckTimer == null) { + _startPeerHealthCheck(); + } + if (completer != null && !completer.isCompleted) { completer.complete(); } @@ -228,6 +242,7 @@ class WatchTogetherPeerService { conn.on('data').listen((data) { try { + _updatePeerActivity(peerId); // Track activity on each message final message = SyncMessage.fromJson(data as String); appLogger.d('WatchTogether: Received message: ${message.type} from $peerId'); _messageReceivedController.add(message); @@ -239,9 +254,12 @@ class WatchTogetherPeerService { conn.on('close').listen((_) { appLogger.d('WatchTogether: Connection closed with: $peerId'); _connections.remove(peerId); + _lastPeerActivity.remove(peerId); + _peerReconnectAttempts.remove(peerId); _peerDisconnectedController.add(peerId); if (_connections.isEmpty) { + _stopPeerHealthCheck(); _connectionStateController.add(false); } }); @@ -255,6 +273,9 @@ class WatchTogetherPeerService { originalError: error, ), ); + + // Attempt reconnection on data channel error + _attemptPeerReconnect(peerId); }); } @@ -283,6 +304,84 @@ class WatchTogetherPeerService { } } + /// Start peer health monitoring + void _startPeerHealthCheck() { + _peerHealthCheckTimer?.cancel(); + _peerHealthCheckTimer = Timer.periodic(_peerHealthCheckInterval, (_) { + _checkPeerHealth(); + }); + } + + /// Stop peer health monitoring + void _stopPeerHealthCheck() { + _peerHealthCheckTimer?.cancel(); + _peerHealthCheckTimer = null; + } + + /// Check health of all peer connections + void _checkPeerHealth() { + final now = DateTime.now(); + final peersToReconnect = []; + + for (final peerId in _connections.keys.toList()) { + final lastActivity = _lastPeerActivity[peerId]; + if (lastActivity != null && now.difference(lastActivity) > _peerTimeout) { + appLogger.w('WatchTogether: Peer $peerId timed out (no activity for ${_peerTimeout.inSeconds}s)'); + peersToReconnect.add(peerId); + } + } + + for (final peerId in peersToReconnect) { + _attemptPeerReconnect(peerId); + } + } + + /// Update peer activity timestamp (called on each message received) + void _updatePeerActivity(String peerId) { + _lastPeerActivity[peerId] = DateTime.now(); + // Reset reconnect attempts on successful activity + _peerReconnectAttempts[peerId] = 0; + } + + /// Attempt to reconnect to a peer with exponential backoff + void _attemptPeerReconnect(String peerId) { + final attempts = _peerReconnectAttempts[peerId] ?? 0; + + if (attempts >= _maxPeerReconnectAttempts) { + appLogger.e('WatchTogether: Max reconnect attempts reached for peer $peerId'); + // Remove the dead connection and notify + _connections.remove(peerId); + _lastPeerActivity.remove(peerId); + _peerReconnectAttempts.remove(peerId); + _peerDisconnectedController.add(peerId); + + if (_connections.isEmpty) { + _connectionStateController.add(false); + } + return; + } + + _peerReconnectAttempts[peerId] = attempts + 1; + final delay = Duration(seconds: (attempts + 1) * 2); // Exponential backoff + + appLogger.d( + 'WatchTogether: Attempting peer reconnect to $peerId (${attempts + 1}/$_maxPeerReconnectAttempts) in ${delay.inSeconds}s', + ); + + // Close existing connection if any + _connections[peerId]?.close(); + _connections.remove(peerId); + + // Schedule reconnection attempt + Timer(delay, () { + if (_peer != null && !_connections.containsKey(peerId)) { + appLogger.d('WatchTogether: Reconnecting to peer $peerId'); + final conn = _peer!.connect(peerId, options: PeerConnectOption(reliable: true)); + _handleNewConnection(conn, isOutgoing: true); + } + }); + } + /// Broadcast a message to all connected peers void broadcast(SyncMessage message) { final json = message.toJson(); @@ -318,6 +417,11 @@ class WatchTogetherPeerService { _reconnectTimer?.cancel(); _reconnectTimer = null; + // Stop peer health monitoring + _stopPeerHealthCheck(); + _lastPeerActivity.clear(); + _peerReconnectAttempts.clear(); + // Close all data connections for (final conn in _connections.values) { conn.close(); diff --git a/lib/watch_together/services/watch_together_sync_manager.dart b/lib/watch_together/services/watch_together_sync_manager.dart index aaed301a..37e8247c 100644 --- a/lib/watch_together/services/watch_together_sync_manager.dart +++ b/lib/watch_together/services/watch_together_sync_manager.dart @@ -39,9 +39,18 @@ class WatchTogetherSyncManager { // Drift correction constants static const Duration maxAllowedDrift = Duration(seconds: 2); - static const Duration positionSyncInterval = Duration(seconds: 5); + static const Duration positionSyncInterval = Duration(seconds: 3); static const Duration excessiveDrift = Duration(seconds: 10); + // Buffering debounce constants + static const Duration _bufferingDebounceDelay = Duration(milliseconds: 500); + + // Buffering debounce timer - prevents false pauses from brief buffering events + Timer? _bufferingDebounceTimer; + + // Peers with pending (debounced) buffering state + final Map _pendingBufferingState = {}; + // Track last known state to avoid duplicate broadcasts bool _lastKnownPlaying = false; double _lastKnownRate = 1.0; @@ -159,13 +168,16 @@ class WatchTogetherSyncManager { _rateSubscription?.cancel(); _messageSubscription?.cancel(); _positionSyncTimer?.cancel(); + _bufferingDebounceTimer?.cancel(); _playingSubscription = null; _bufferingSubscription = null; _rateSubscription = null; _messageSubscription = null; _positionSyncTimer = null; + _bufferingDebounceTimer = null; + _pendingBufferingState.clear(); _player = null; appLogger.d('WatchTogether: Player detached'); } @@ -201,7 +213,7 @@ class WatchTogetherSyncManager { }); // Listen to buffering state changes - _bufferingSubscription = _player!.streams.buffering.listen((isBuffering) { + _bufferingSubscription = _player!.streams.buffering.listen((isBuffering) async { if (_isRemoteAction) return; // Announce ready when we stop buffering for the first time (video loaded) @@ -218,6 +230,13 @@ class WatchTogetherSyncManager { } _peerService.broadcast(SyncMessage.buffering(isBuffering, peerId: _peerService.myPeerId)); + + // Check for auto-resume when LOCAL buffering stops + // This fixes the bug where subtitle loading would pause both users but + // only resume would trigger from remote buffering messages, not local + if (!isBuffering) { + await _checkAutoResume(); + } }); // Listen to rate changes @@ -239,11 +258,18 @@ class WatchTogetherSyncManager { } /// Start periodic position sync (host only) + /// Includes play/pause state for eventual consistency void _startPositionSync() { _positionSyncTimer?.cancel(); _positionSyncTimer = Timer.periodic(positionSyncInterval, (_) { if (_player != null && _session.isHost) { - _peerService.broadcast(SyncMessage.positionSync(_player!.state.position, peerId: _peerService.myPeerId)); + _peerService.broadcast( + SyncMessage.positionSync( + _player!.state.position, + peerId: _peerService.myPeerId, + isPlaying: _player!.state.playing, + ), + ); } }); } @@ -364,20 +390,36 @@ class WatchTogetherSyncManager { break; } if (message.peerId != null && message.bufferingState != null) { - _participantBuffering[message.peerId!] = message.bufferingState!; + final peerId = message.peerId!; + final isBuffering = message.bufferingState!; - // Auto-pause when any peer starts buffering - if (isAnyBuffering && _player!.state.playing) { - _wasPlayingBeforeBuffering = true; - appLogger.d('WatchTogether: Peer buffering, pausing playback'); - await _applyRemotePause(); - } - // Auto-resume when all peers stop buffering AND all ready (if we were playing before) - else if (isAllReady && !isAnyBuffering && !_player!.state.playing && _wasPlayingBeforeBuffering) { - _wasPlayingBeforeBuffering = false; - appLogger.d('WatchTogether: All peers done buffering, resuming playback'); - await _applyRemotePlay(position: _pendingPlayPosition); - _pendingPlayPosition = null; + if (isBuffering) { + // Peer started buffering - use debounce to avoid false pauses + _pendingBufferingState[peerId] = true; + + // Cancel existing debounce timer if any + _bufferingDebounceTimer?.cancel(); + _bufferingDebounceTimer = Timer(_bufferingDebounceDelay, () async { + // Check if still pending after debounce delay + if (_pendingBufferingState[peerId] == true) { + _participantBuffering[peerId] = true; + _pendingBufferingState.remove(peerId); + + // Auto-pause when any peer starts buffering (sustained) + if (isAnyBuffering && _player != null && _player!.state.playing) { + _wasPlayingBeforeBuffering = true; + appLogger.d('WatchTogether: Peer buffering (sustained), pausing playback'); + await _applyRemotePause(); + } + } + }); + } else { + // Peer stopped buffering - cancel pending debounce and update immediately + _pendingBufferingState.remove(peerId); + _participantBuffering[peerId] = false; + + // Auto-resume when all peers stop buffering AND all ready + await _checkAutoResume(); } } break; @@ -386,6 +428,20 @@ class WatchTogetherSyncManager { if (message.position != null) { _checkAndCorrectDrift(message.position!, message.timestamp); } + // Reconcile play/pause state if host sent it and we diverged + // This provides eventual consistency for play/pause state + if (message.isPlaying != null && _player != null && !_session.isHost) { + final localPlaying = _player!.state.playing; + if (message.isPlaying! && !localPlaying && !isAnyBuffering && isAllReady) { + // Host is playing but we're paused - sync up + appLogger.d('WatchTogether: Play/pause state diverged, syncing to host (playing)'); + await _applyRemotePlay(position: message.position); + } else if (!message.isPlaying! && localPlaying) { + // Host is paused but we're playing - sync up + appLogger.d('WatchTogether: Play/pause state diverged, syncing to host (paused)'); + await _applyRemotePause(); + } + } break; case SyncMessageType.rate: @@ -437,12 +493,7 @@ class WatchTogetherSyncManager { appLogger.d('WatchTogether: Peer ${message.peerId} player ready: ${message.bufferingState}'); // If we were waiting to play and all are now ready, start playback - if (isAllReady && !isAnyBuffering && _wasPlayingBeforeBuffering) { - _wasPlayingBeforeBuffering = false; - appLogger.d('WatchTogether: All players ready, starting playback'); - await _applyRemotePlay(position: _pendingPlayPosition); - _pendingPlayPosition = null; - } + await _checkAutoResume(); } break; } @@ -518,6 +569,21 @@ class WatchTogetherSyncManager { } } + /// Check if conditions are met to auto-resume playback + /// Called when local or remote buffering stops, or when a peer becomes ready + Future _checkAutoResume() async { + if (_player == null) return; + if (!isAllReady) return; + if (isAnyBuffering) return; + if (_player!.state.playing) return; + if (!_wasPlayingBeforeBuffering) return; + + _wasPlayingBeforeBuffering = false; + appLogger.d('WatchTogether: All conditions met, auto-resuming playback'); + await _applyRemotePlay(position: _pendingPlayPosition); + _pendingPlayPosition = null; + } + /// Apply remote rate change Future _applyRemoteRate(double rate) async { if (_player == null) return; @@ -691,6 +757,7 @@ class WatchTogetherSyncManager { detachPlayer(); _participantBuffering.clear(); _participantReady.clear(); + _pendingBufferingState.clear(); _hasAnnouncedReady = false; } }