forked from Mirrors/Dispatcharr
Add stream health recovery and reconnection logic
This commit is contained in:
SergeantPanda
parent
6ccc2b9a6d
commit
07485f87a3
3 changed files with 115 additions and 5 deletions
|
|
@ -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
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue