Enhancement: Improve live proxy failover logic by resetting connection retries after a specified period of stability and ensuring backup streams are rotated in channel order. Update changelog to reflect these changes.

CHANGELOG.md

@@ -47,6 +47,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **VOD connection card Content Duration badge** shows human-readable lengths (`45m`, `2h 0m`) again on the Stats page instead of bare minute integers. — Thanks [@nick4810](https://github.com/nick4810)
 - **Frontend unit tests pass on Node 25+ again.** Node 25+ exposes a native `localStorage` stub on `globalThis` that lacks Storage API methods (`getItem`, `clear`, etc.), which shadows jsdom's implementation and breaks tests that touch `localStorage` (e.g. `TypeError: localStorage.clear is not a function`). `setupTests.js` now installs an in-memory shim on `globalThis` and `window` when those methods are missing; the shim is skipped automatically once Node provides a working implementation. — Thanks [@nick4810](https://github.com/nick4810)
 - **Live proxy failover works with the default VLC stream profile.** When `cvlc` could not open an upstream URL it often stayed running in dummy mode without writing TS data, so the stream manager never left its read loop and never exhausted retries to switch streams. The locked VLC profile now passes `--play-and-exit` (migration 0027 for existing installs), and `VLCLogParser` treats `unable to open the mrl` as a fatal input error so the connection is closed and the normal retry → failover cycle can proceed.
+- **Live proxy no longer fails over after isolated provider disconnects.** Connection retries on the same URL used to accumulate indefinitely, so three brief drops spread across many hours of otherwise stable playback could exhaust retries and switch streams even when each reconnect succeeded. Retries now reset after 30 minutes without a failure (`RETRY_WINDOW_SECONDS`), so periodic provider drops (for example every few hours) each get a full retry budget on the current stream. Failover still triggers when three failures occur within that window.
+- **Live proxy failover now walks backup streams in channel order.** After a stable session on a backup stream, `tried_stream_ids` is cleared so rotation continues from the current position (stream 2 → 3 → 4 → 1) instead of jumping back to stream 1. `get_alternate_streams()` returns alternates in that rotated order, matching the manual next-stream API.
 
 ## [0.27.2] - 2026-06-30
 

apps/proxy/config.py

@@ -7,6 +7,8 @@ class BaseConfig:
     CHUNK_SIZE = 8192
     CLIENT_POLL_INTERVAL = 0.1
     MAX_RETRIES = 3
+    RETRY_WINDOW_SECONDS = 1800  # Reset retry counter after this long without a failure
+    STABLE_CONNECTION_THRESHOLD = 30  # Seconds of uptime before switch rotation state resets
     RETRY_WAIT_INTERVAL = 0.5  # seconds to wait between retries
     CONNECTION_TIMEOUT = 10  # seconds to wait for initial connection
     MAX_STREAM_SWITCHES = 10  # Maximum number of stream switch attempts before giving up

apps/proxy/live_proxy/config_helper.py

@@ -75,6 +75,16 @@ class ConfigHelper:
         """Get maximum retry attempts"""
         return ConfigHelper.get('MAX_RETRIES', 3)
 
+    @staticmethod
+    def retry_window_seconds():
+        """Reset the retry counter after this many seconds without a failure."""
+        return ConfigHelper.get('RETRY_WINDOW_SECONDS', 1800)
+
+    @staticmethod
+    def stable_connection_threshold():
+        """Seconds of stable playback before switch rotation state resets."""
+        return ConfigHelper.get('STABLE_CONNECTION_THRESHOLD', 30)
+
     @staticmethod
     def max_stream_switches():
         """Get maximum number of stream switch attempts"""

apps/proxy/live_proxy/input/manager.py

@@ -38,6 +38,9 @@ class StreamManager:
         self.connected = False
         self.retry_count = 0
         self.max_retries = ConfigHelper.max_retries()
+        self._retry_window_seconds = ConfigHelper.retry_window_seconds()
+        self._last_failure_time = None
+        self._stable_connection_threshold = ConfigHelper.stable_connection_threshold()
         self.current_response = None
         self.current_session = None
         self.url_switching = False
@@ -157,6 +160,29 @@ class StreamManager:
 
         return session
 
+    def _record_connection_failure(self):
+        """Record a failure; reset the counter if the last one was long ago."""
+        now = time.time()
+        if (
+            self._last_failure_time is not None
+            and (now - self._last_failure_time) > self._retry_window_seconds
+        ):
+            self.retry_count = 0
+        self._last_failure_time = now
+        self.retry_count += 1
+        return self.retry_count
+
+    def _clear_connection_failure_history(self):
+        self.retry_count = 0
+        self._last_failure_time = None
+
+    def _note_stable_connection(self):
+        """Reset stream-switch bookkeeping after sustained successful playback."""
+        if self.current_stream_id:
+            self.tried_stream_ids = {self.current_stream_id}
+        else:
+            self.tried_stream_ids.clear()
+
     def _wait_for_existing_processes_to_close(self, timeout=5.0):
         """Wait for existing processes/connections to fully close before establishing new ones"""
         start_time = time.time()
@@ -373,6 +399,7 @@ class StreamManager:
                     # Attempt reconnect without changing streams
                     if self._attempt_reconnect():
                         logger.info(f"Health-requested reconnect successful for channel {self.channel_id}")
+                        self._clear_connection_failure_history()
                         continue  # Go back to main loop
                     else:
                         logger.warning(f"Health-requested reconnect failed, will try stream switch for channel {self.channel_id}")
@@ -385,10 +412,11 @@ class StreamManager:
                     if self._try_next_stream():
                         logger.info(f"Health-requested stream switch successful for channel {self.channel_id}")
                         stream_switch_attempts += 1
-                        self.retry_count = 0  # Reset retries for new stream
+                        self._clear_connection_failure_history()
                         continue  # Go back to main loop with new stream
                     else:
                         logger.error(f"Health-requested stream switch failed for channel {self.channel_id}")
+                        self._clear_connection_failure_history()
                         # Continue with normal flow
 
                 # Check stream type before connecting
@@ -401,19 +429,26 @@ class StreamManager:
                     self.transcode = True
                     # We'll override the stream profile selection with ffmpeg in the transcoding section
                     self.force_ffmpeg = True
-                # Reset connection retry count for this specific URL
-                self.retry_count = 0
                 url_failed = False
                 if self.url_switching:
                     logger.debug(f"Skipping connection attempt during URL switch for channel {self.channel_id}")
                     gevent.sleep(0.1)
                     continue
                 # Connection retry loop for current URL
-                while self.running and self.retry_count < self.max_retries and not url_failed and not self.needs_stream_switch:
+                while (
+                    self.running
+                    and self.retry_count < self.max_retries
+                    and not url_failed
+                    and not self.needs_stream_switch
+                ):
                     if not self._ensure_owner_or_stop():
                         break
 
-                    logger.info(f"Connection attempt {self.retry_count + 1}/{self.max_retries} for URL: {self.url} for channel {self.channel_id}")
+                    attempt = self.retry_count + 1
+                    logger.info(
+                        f"Connection attempt {attempt}/{self.max_retries} "
+                        f"for URL: {self.url} for channel {self.channel_id}"
+                    )
 
                     # Handle connection based on whether we transcode or not
                     connection_result = False
@@ -434,7 +469,7 @@ class StreamManager:
                                         'channel_reconnect',
                                         channel_id=self.channel_id,
                                         channel_name=self.channel_name,
-                                        attempt=self.retry_count + 1,
+                                        attempt=attempt,
                                         max_attempts=self.max_retries
                                     )
                                 except Exception as e:
@@ -444,16 +479,18 @@ class StreamManager:
                             self._process_stream_data()
                             # If we get here, the connection was closed/failed
 
-                            # Reset stream switch attempts if the connection lasted longer than threshold
-                            # This indicates we had a stable connection for a while before failing
                             connection_duration = time.time() - connection_start_time
-                            stable_connection_threshold = 30  # 30 seconds threshold
+                            stable_threshold = self._stable_connection_threshold
 
                             if self.needs_stream_switch:
                                 logger.info(f"Stream needs to switch after {connection_duration:.1f} seconds for channel: {self.channel_id}")
                                 break  # Exit to switch streams
-                            if connection_duration > stable_connection_threshold:
-                                logger.info(f"Stream was stable for {connection_duration:.1f} seconds, resetting switch attempts counter for channel: {self.channel_id}")
+                            if connection_duration >= stable_threshold:
+                                logger.info(
+                                    f"Stream was stable for {connection_duration:.1f} seconds, "
+                                    f"resetting switch rotation state for channel: {self.channel_id}"
+                                )
+                                self._note_stable_connection()
                                 stream_switch_attempts = 0
 
                         # Connection failed or ended - decide what to do next
@@ -461,14 +498,15 @@ class StreamManager:
                             # Normal shutdown requested
                             return
 
-                        # Connection failed, increment retry count
-                        self.retry_count += 1
                         self.connected = False
+                        failures = self._record_connection_failure()
 
-                        # If we've reached max retries, mark this URL as failed
-                        if self.retry_count >= self.max_retries:
+                        if failures >= self.max_retries:
                             url_failed = True
-                            logger.warning(f"Maximum retry attempts ({self.max_retries}) reached for URL: {self.url} for channel: {self.channel_id}")
+                            logger.warning(
+                                f"Maximum retry attempts ({self.max_retries}) reached for URL: {self.url} "
+                                f"for channel: {self.channel_id}"
+                            )
 
                             # Log connection error event
                             try:
@@ -484,16 +522,20 @@ class StreamManager:
                                 logger.error(f"Could not log connection error event: {e}")
                         else:
                             # Wait with exponential backoff before retrying
-                            timeout = min(.25 * self.retry_count, 3)  # Cap at 3 seconds
-                            logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count}/{self.max_retries}) for channel: {self.channel_id}")
+                            timeout = min(.25 * failures, 3)  # Cap at 3 seconds
+                            logger.info(
+                                f"Reconnecting in {timeout} seconds... "
+                                f"(attempt {failures}/{self.max_retries}) "
+                                f"for channel: {self.channel_id}"
+                            )
                             gevent.sleep(timeout)
 
                     except Exception as e:
                         logger.error(f"Connection error on channel: {self.channel_id}: {e}", exc_info=True)
-                        self.retry_count += 1
                         self.connected = False
+                        failures = self._record_connection_failure()
 
-                        if self.retry_count >= self.max_retries:
+                        if failures >= self.max_retries:
                             url_failed = True
 
                             # Log connection error event with exception details
@@ -511,8 +553,12 @@ class StreamManager:
                                 logger.error(f"Could not log connection error event: {log_error}")
                         else:
                             # Wait with exponential backoff before retrying
-                            timeout = min(.25 * self.retry_count, 3)  # Cap at 3 seconds
-                            logger.info(f"Reconnecting in {timeout} seconds after error... (attempt {self.retry_count}/{self.max_retries}) for channel: {self.channel_id}")
+                            timeout = min(.25 * failures, 3)  # Cap at 3 seconds
+                            logger.info(
+                                f"Reconnecting in {timeout} seconds after error... "
+                                f"(attempt {failures}/{self.max_retries}) "
+                                f"for channel: {self.channel_id}"
+                            )
                             gevent.sleep(timeout)
 
                 # If URL failed and we're still running, try switching to another stream
@@ -525,8 +571,7 @@ class StreamManager:
                         # Successfully switched to a new stream, continue with the new URL
                         stream_switch_attempts += 1
                         logger.info(f"Successfully switched to new URL: {self.url} (switch attempt {stream_switch_attempts}/{max_stream_switches}) for channel: {self.channel_id}")
-                        # Reset retry count for the new stream - important for the loop to work correctly
-                        self.retry_count = 0
+                        self._clear_connection_failure_history()
                         # Continue outer loop with new URL - DON'T add a break statement here
                     else:
                         # No more streams to try
@@ -1391,7 +1436,7 @@ class StreamManager:
                 logger.info(f"Updated stream ID from {old_stream_id} to {stream_id} for channel {self.channel_id}")
 
             # Reset retry counter to allow immediate reconnect
-            self.retry_count = 0
+            self._clear_connection_failure_history()
 
             # Also reset buffer position to prevent stale data after URL change
             if hasattr(self.buffer, 'reset_buffer_position'):

apps/proxy/live_proxy/tests/test_alternate_stream_order.py

@@ -0,0 +1,53 @@
+"""Tests for failover stream rotation order."""
+from django.test import TestCase
+
+from apps.proxy.live_proxy.url_utils import order_alternates_from_current
+
+
+class AlternateStreamOrderTests(TestCase):
+    def test_rotates_from_current_stream(self):
+        alternates = [
+            {'stream_id': 1, 'profile_id': 10},
+            {'stream_id': 3, 'profile_id': 30},
+            {'stream_id': 4, 'profile_id': 40},
+        ]
+        ordered_ids = [1, 2, 3, 4]
+
+        result = order_alternates_from_current(alternates, ordered_ids, current_stream_id=2)
+
+        self.assertEqual([s['stream_id'] for s in result], [3, 4, 1])
+
+    def test_first_failover_from_primary(self):
+        alternates = [
+            {'stream_id': 2, 'profile_id': 20},
+            {'stream_id': 3, 'profile_id': 30},
+            {'stream_id': 4, 'profile_id': 40},
+        ]
+        ordered_ids = [1, 2, 3, 4]
+
+        result = order_alternates_from_current(alternates, ordered_ids, current_stream_id=1)
+
+        self.assertEqual([s['stream_id'] for s in result], [2, 3, 4])
+
+    def test_wraps_from_last_stream(self):
+        alternates = [
+            {'stream_id': 1, 'profile_id': 10},
+            {'stream_id': 2, 'profile_id': 20},
+            {'stream_id': 3, 'profile_id': 30},
+        ]
+        ordered_ids = [1, 2, 3, 4]
+
+        result = order_alternates_from_current(alternates, ordered_ids, current_stream_id=4)
+
+        self.assertEqual([s['stream_id'] for s in result], [1, 2, 3])
+
+    def test_skips_unavailable_streams_in_rotation(self):
+        alternates = [
+            {'stream_id': 4, 'profile_id': 40},
+            {'stream_id': 1, 'profile_id': 10},
+        ]
+        ordered_ids = [1, 2, 3, 4]
+
+        result = order_alternates_from_current(alternates, ordered_ids, current_stream_id=2)
+
+        self.assertEqual([s['stream_id'] for s in result], [4, 1])

apps/proxy/live_proxy/tests/test_failover_retry_window.py

@@ -0,0 +1,63 @@
+"""Tests for connection retry idle reset and stable-playback failover reset."""
+import time
+
+from django.test import SimpleTestCase, TestCase
+
+from apps.proxy.live_proxy.config_helper import ConfigHelper
+from apps.proxy.live_proxy.input.manager import StreamManager
+
+def _make_manager(**overrides):
+    sm = StreamManager.__new__(StreamManager)
+    sm.channel_id = "test-channel"
+    sm.max_retries = 3
+    sm._retry_window_seconds = 1800
+    sm._stable_connection_threshold = 30
+    sm._last_failure_time = None
+    sm.retry_count = 0
+    sm.current_stream_id = 100
+    sm.tried_stream_ids = {100, 200, 300}
+    for key, value in overrides.items():
+        setattr(sm, key, value)
+    return sm
+
+
+class RetryIdleResetTests(TestCase):
+    def test_counter_resets_after_idle_period(self):
+        sm = _make_manager(_retry_window_seconds=60)
+        sm._last_failure_time = time.time() - 120
+        sm.retry_count = 2
+
+        count = sm._record_connection_failure()
+
+        self.assertEqual(count, 1)
+
+    def test_counter_accumulates_within_idle_period(self):
+        sm = _make_manager(_retry_window_seconds=1800)
+        self.assertEqual(sm._record_connection_failure(), 1)
+        self.assertEqual(sm._record_connection_failure(), 2)
+        self.assertEqual(sm._record_connection_failure(), 3)
+        self.assertFalse(sm.should_retry())
+
+    def test_stable_connection_resets_tried_streams_only(self):
+        sm = _make_manager()
+        sm._record_connection_failure()
+        sm._record_connection_failure()
+        sm._note_stable_connection()
+        self.assertEqual(sm.retry_count, 2)
+        self.assertEqual(sm.tried_stream_ids, {100})
+
+    def test_clear_connection_failure_history(self):
+        sm = _make_manager()
+        sm._record_connection_failure()
+        sm._record_connection_failure()
+        sm._clear_connection_failure_history()
+        self.assertEqual(sm.retry_count, 0)
+        self.assertIsNone(sm._last_failure_time)
+
+
+class FailoverConfigDefaultsTests(SimpleTestCase):
+    def test_retry_window_default(self):
+        self.assertEqual(ConfigHelper.retry_window_seconds(), 1800)
+
+    def test_stable_connection_threshold_default(self):
+        self.assertEqual(ConfigHelper.stable_connection_threshold(), 30)

apps/proxy/live_proxy/url_utils.py

@@ -306,6 +306,33 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
     finally:
         close_old_connections()
 
+def order_alternates_from_current(
+    alternate_streams: List[dict],
+    ordered_stream_ids: List[int],
+    current_stream_id: Optional[int],
+) -> List[dict]:
+    """
+    Reorder failover candidates to start after the current stream in channel order,
+    wrapping around.
+    """
+    if not alternate_streams or not ordered_stream_ids or current_stream_id is None:
+        return alternate_streams
+
+    alt_by_id = {entry['stream_id']: entry for entry in alternate_streams}
+
+    try:
+        current_index = ordered_stream_ids.index(current_stream_id)
+    except ValueError:
+        return alternate_streams
+
+    rotated = []
+    for offset in range(1, len(ordered_stream_ids)):
+        stream_id = ordered_stream_ids[(current_index + offset) % len(ordered_stream_ids)]
+        entry = alt_by_id.get(stream_id)
+        if entry is not None:
+            rotated.append(entry)
+    return rotated
+
 def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = None) -> List[dict]:
     """
     Get alternative streams for a channel when the current stream fails.
@@ -331,9 +358,10 @@ def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = No
 
         # Get all assigned streams for this channel using the correct ordering
         streams = channel.streams.all().order_by('channelstream__order')
-        logger.debug(f"Channel {channel_id} has {streams.count()} total assigned streams")
+        ordered_stream_ids = list(streams.values_list('id', flat=True))
+        logger.debug(f"Channel {channel_id} has {len(ordered_stream_ids)} total assigned streams")
 
-        if not streams.exists():
+        if not ordered_stream_ids:
             logger.warning(f"No streams assigned to channel {channel_id}")
             return []
 
@@ -423,7 +451,9 @@ def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = No
         else:
             logger.warning(f"No alternate streams with available connections found for channel {channel_id}")
 
-        return alternate_streams
+        return order_alternates_from_current(
+            alternate_streams, ordered_stream_ids, current_stream_id
+        )
     except Exception as e:
         logger.error(f"Error getting alternate streams for channel {channel_id}: {e}", exc_info=True)
         return []