diff --git a/apps/proxy/config.py b/apps/proxy/config.py index f8151ad8..a465b5d7 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -53,6 +53,12 @@ class TSConfig(BaseConfig): # Make chunk size a multiple of TS packet size for perfect alignment # ~1MB is ideal for streaming (matches typical media buffer sizes) + # Stream health and recovery settings + MAX_HEALTH_RECOVERY_ATTEMPTS = 2 # Maximum times to attempt recovery for a single stream + MAX_RECONNECT_ATTEMPTS = 3 # Maximum reconnects to try before switching streams + MIN_STABLE_TIME_BEFORE_RECONNECT = 30 # Minimum seconds a stream must be stable to try reconnect + FAILOVER_GRACE_PERIOD = 20 # Extra time (seconds) to allow for stream switching before disconnecting clients + diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 8f1c09d9..9377a079 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -335,14 +335,24 @@ class StreamGenerator: def _is_timeout(self): """Check if the stream has timed out.""" + # Get a more generous timeout for stream switching + stream_timeout = getattr(Config, 'STREAM_TIMEOUT', 10) + failover_grace_period = getattr(Config, 'FAILOVER_GRACE_PERIOD', 20) + total_timeout = stream_timeout + failover_grace_period + # Disconnect after long inactivity - if time.time() - self.last_yield_time > Config.STREAM_TIMEOUT: + if time.time() - self.last_yield_time > total_timeout: if self.stream_manager and not self.stream_manager.healthy: - logger.warning(f"[{self.client_id}] No data for {Config.STREAM_TIMEOUT}s and stream unhealthy, disconnecting") + # Check if stream manager is actively switching or reconnecting + if (hasattr(self.stream_manager, 'url_switching') and self.stream_manager.url_switching): + logger.info(f"[{self.client_id}] Stream switching in progress, giving more time") + return False + + logger.warning(f"[{self.client_id}] No data for {total_timeout}s and stream unhealthy, disconnecting") return True elif not self.is_owner_worker and self.consecutive_empty > 100: # Non-owner worker without data for too long - logger.warning(f"[{self.client_id}] Non-owner worker with no data for {Config.STREAM_TIMEOUT}s, disconnecting") + logger.warning(f"[{self.client_id}] Non-owner worker with no data for {total_timeout}s, disconnecting") return True return False diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index 5d06c079..ac5f801b 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -336,6 +336,9 @@ class StreamManager: self.socket = self.transcode_process.stdout # Read from std output self.connected = True + # Set connection start time for stability tracking + self.connection_start_time = time.time() + # Set channel state to waiting for clients self._set_waiting_for_clients() @@ -367,6 +370,9 @@ class StreamManager: self.healthy = True logger.info(f"Successfully connected to stream source") + # Store connection start time for stability tracking + self.connection_start_time = time.time() + # Set channel state to waiting for clients self._set_waiting_for_clients() @@ -588,24 +594,112 @@ class StreamManager: def _monitor_health(self): """Monitor stream health and attempt recovery if needed""" + consecutive_unhealthy_checks = 0 + health_recovery_attempts = 0 + reconnect_attempts = 0 + max_health_recovery_attempts = ConfigHelper.get('MAX_HEALTH_RECOVERY_ATTEMPTS', 2) + max_reconnect_attempts = ConfigHelper.get('MAX_RECONNECT_ATTEMPTS', 3) + min_stable_time = ConfigHelper.get('MIN_STABLE_TIME_BEFORE_RECONNECT', 30) # seconds + while self.running: try: now = time.time() - if now - self.last_data_time > getattr(Config, 'CONNECTION_TIMEOUT', 10) and self.connected: + inactivity_duration = now - self.last_data_time + timeout_threshold = getattr(Config, 'CONNECTION_TIMEOUT', 10) + + if inactivity_duration > timeout_threshold and self.connected: # Mark unhealthy if no data for too long if self.healthy: - logger.warning(f"Stream unhealthy - no data for {now - self.last_data_time:.1f}s") + logger.warning(f"Stream unhealthy - no data for {inactivity_duration:.1f}s") self.healthy = False + + # Track consecutive unhealthy checks + consecutive_unhealthy_checks += 1 + + # After several unhealthy checks in a row, try recovery + if consecutive_unhealthy_checks >= 3 and health_recovery_attempts < max_health_recovery_attempts: + # Calculate how long the stream was stable before failing + connection_start_time = getattr(self, 'connection_start_time', 0) + stable_time = self.last_data_time - connection_start_time if connection_start_time > 0 else 0 + + if stable_time >= min_stable_time and reconnect_attempts < max_reconnect_attempts: + # Stream was stable for a while, try reconnecting first + logger.warning(f"Stream was stable for {stable_time:.1f}s before failing. " + f"Attempting reconnect {reconnect_attempts + 1}/{max_reconnect_attempts}") + reconnect_attempts += 1 + threading.Thread(target=self._attempt_reconnect, daemon=True).start() + else: + # Stream was not stable long enough, or reconnects failed too many times + # Try switching to another stream + if reconnect_attempts > 0: + logger.warning(f"Reconnect attempts exhausted ({reconnect_attempts}/{max_reconnect_attempts}). " + f"Attempting stream switch recovery") + else: + logger.warning(f"Stream was only stable for {stable_time:.1f}s (<{min_stable_time}s). " + f"Skipping reconnect, attempting stream switch") + + health_recovery_attempts += 1 + reconnect_attempts = 0 # Reset for next time + threading.Thread(target=self._attempt_health_recovery, daemon=True).start() elif self.connected and not self.healthy: # Auto-recover health when data resumes logger.info(f"Stream health restored") self.healthy = True + consecutive_unhealthy_checks = 0 + health_recovery_attempts = 0 + reconnect_attempts = 0 + + # If healthy, reset unhealthy counter (but keep other state) + if self.healthy: + consecutive_unhealthy_checks = 0 except Exception as e: logger.error(f"Error in health monitor: {e}") time.sleep(self.health_check_interval) + def _attempt_reconnect(self): + """Attempt to reconnect to the current stream""" + try: + logger.info(f"Attempting reconnect to current stream for channel {self.channel_id}") + + # Don't try to reconnect if we're already switching URLs + if self.url_switching: + logger.info("URL switching already in progress, skipping reconnect") + return + + # Close existing connection + if self.transcode or self.socket: + self._close_socket() + else: + self._close_connection() + + self.connected = False + + # Attempt to establish a new connection using the same URL + connection_result = False + try: + if self.transcode: + connection_result = self._establish_transcode_connection() + else: + connection_result = self._establish_http_connection() + + if connection_result: + # Store connection start time to measure stability + self.connection_start_time = time.time() + logger.info(f"Reconnect successful for channel {self.channel_id}") + return True + else: + logger.warning(f"Reconnect failed for channel {self.channel_id}") + return False + except Exception as e: + logger.error(f"Error during reconnect: {e}", exc_info=True) + return False + + except Exception as e: + logger.error(f"Error in reconnect attempt: {e}", exc_info=True) + return False + def _close_connection(self): """Close HTTP connection resources""" # Close response if it exists