fix(live): resolve Redis state latching issue during buffering-timeout failover (Fixes #1449)

This update ensures that the Redis state for live channels is correctly cleared when a buffering-timeout failover occurs. Previously, the in-memory buffering flag could be cleared without updating Redis, leading to incorrect state reporting. The fix modifies the logic to clear the Redis state to 'active' when the buffering flag is reset, preventing stale 'buffering' states from persisting after recovery. Tests have been updated to validate this behavior.

CHANGELOG.md

@@ -45,6 +45,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Live channel Redis `state` no longer stays latched at `buffering` after a buffering-timeout failover.** When FFmpeg speed dipped below the buffering threshold and the timeout triggered a stream switch, the in-memory buffering flag was cleared without rewriting Redis, and the same stats sample could re-write `buffering`. After the new stream recovered, the speed-good path only cleared Redis when that flag was still set, so a healthy session could report `buffering` until restart or retune. A successful buffering-timeout switch now writes `active` and skips the fallthrough `buffering` write. (Fixes #1449)
 - **Plugin discovery no longer force-reloads on every connect event in multi-worker setups.** Reacting to a newer `.reload_token` (after install/update/reload) previously re-touched the token, so each uWSGI worker's next discovery re-staled every other worker and never converged. Streaming connect/disconnect events then re-imported every enabled plugin on the request path, leaking plugin background threads and degrading workers until restart. Stale-token reactions now reload locally without bumping the shared token; only an explicit `force_reload=True` (install/update/reload API) broadcasts. (Fixes #1452)
 - **Channels table "Copy URL" no longer includes the web player's output profile.** The kebab-menu action was reusing the in-app preview URL builder, so copied links could include `output_format=mpegts` and `output_profile=...` from web-player prefs. Copy now emits a plain `/proxy/ts/stream/{uuid}` URL suitable for external players; Watch still applies player prefs.
 - **Redis-backed VOD sessions no longer leave stale profile connection reservations after multi-worker range-request teardown.** Jellyfin/FFmpeg clients that issue concurrent byte-range requests for one VOD session could finish teardown while another worker held the session metadata lock (`vod_connection_lock:*`). `decrement_active_streams_and_check()` then failed with `DECR-AS-CHECK failed: could not acquire lock`, skipped the profile-slot release, and left `profile_connections:*` and the session hash occupied until restart or manual Redis cleanup. Per-session `active_streams` is now mutated with Redis Lua (independent of the metadata lock), metadata saves omit and never clobber that counter, idle cleanup deletes the session hash only when still idle, and late metadata writes cannot recreate a deleted session. (Fixes #1426)

apps/proxy/live_proxy/input/manager.py

@@ -1087,6 +1087,11 @@ class StreamManager:
                         f"Output Bitrate: {ffmpeg_output_bitrate_str} kbps")
             # If we have a valid speed, check for buffering
             if ffmpeg_speed is not None and ffmpeg_speed < self.buffering_speed:
+                # When a buffering-timeout failover clears the in-memory flag, also
+                # clear Redis and skip the BUFFERING write below. Otherwise the same
+                # stats sample re-writes buffering after self.buffering is False, and
+                # the speed-good recovery path can never clear the Redis label again.
+                switched_after_buffering_timeout = False
                 if self.buffering:
                     # Buffering is still ongoing, check for how long
                     if self.buffering_start_time is None:
@@ -1102,6 +1107,16 @@ class StreamManager:
                                 # Reset buffering state
                                 self.buffering = False
                                 self.buffering_start_time = None
+                                switched_after_buffering_timeout = True
+
+                                # Clear the Redis buffering label.
+                                if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
+                                    metadata_key = RedisKeys.channel_metadata(self.channel_id)
+                                    self.buffer.redis_client.hset(
+                                        metadata_key,
+                                        ChannelMetadataField.STATE,
+                                        ChannelState.ACTIVE,
+                                    )
 
                                 # Log failover event
                                 try:
@@ -1133,12 +1148,13 @@ class StreamManager:
                     except Exception as e:
                         logger.error(f"Could not log buffering event: {e}")
 
-                # Log buffering warning
-                logger.debug(f"FFmpeg speed on channel {self.channel_id} is below {self.buffering_speed} ({ffmpeg_speed}x) - buffering detected")
-                # Set channel state to buffering
-                if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
-                    metadata_key = RedisKeys.channel_metadata(self.channel_id)
-                    self.buffer.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.BUFFERING)
+                if not switched_after_buffering_timeout:
+                    # Log buffering warning
+                    logger.debug(f"FFmpeg speed on channel {self.channel_id} is below {self.buffering_speed} ({ffmpeg_speed}x) - buffering detected")
+                    # Set channel state to buffering
+                    if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
+                        metadata_key = RedisKeys.channel_metadata(self.channel_id)
+                        self.buffer.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.BUFFERING)
             elif ffmpeg_speed is not None and ffmpeg_speed >= self.buffering_speed:
                 # Speed is good, check if we were buffering
                 if self.buffering:

apps/proxy/live_proxy/tests/test_buffering_state_recovery.py

@@ -0,0 +1,130 @@
+"""Tests for Redis channel state after buffering-timeout failover.
+
+After a buffering timeout switch, Redis must not stay latched at buffering
+once the in-memory buffering flag is cleared. The switch path writes ACTIVE
+(same as normal buffering recovery) so mid-session clients are not forced
+through the connecting init-wait path.
+"""
+from unittest.mock import MagicMock, patch
+
+from django.test import TestCase
+
+from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
+from apps.proxy.live_proxy.input.manager import StreamManager
+from apps.proxy.live_proxy.redis_keys import RedisKeys
+
+
+CHANNEL_ID = "00000000-0000-0000-0000-000000000149"
+
+
+class _DictRedis:
+    """Minimal Redis stand-in that records hash field writes."""
+
+    def __init__(self):
+        self.hashes = {}
+
+    def hset(self, key, field=None, value=None, mapping=None):
+        bucket = self.hashes.setdefault(key, {})
+        if mapping:
+            bucket.update(mapping)
+        elif field is not None:
+            bucket[field] = value
+
+
+def _make_stream_manager(redis_client, buffering_timeout=1.0, buffering_speed=1.0):
+    sm = StreamManager.__new__(StreamManager)
+    sm.channel_id = CHANNEL_ID
+    sm.channel_name = "BBC News"
+    sm.buffering = True
+    sm.buffering_timeout = buffering_timeout
+    sm.buffering_speed = buffering_speed
+    sm.buffering_start_time = 0.0
+    # Avoid the bitrate-to-DB flush path; these tests only care about state.
+    sm.current_stream_id = None
+    sm._bitrate_warmup_samples = 10
+    sm._smoothed_output_bitrate = None
+    sm._last_bitrate_db_save_time = 0
+    sm._bitrate_db_save_interval = 60
+
+    buffer = MagicMock()
+    buffer.redis_client = redis_client
+    buffer.channel_id = CHANNEL_ID
+    sm.buffer = buffer
+    return sm
+
+
+class BufferingTimeoutFailoverStateTests(TestCase):
+    @patch.object(StreamManager, "_update_ffmpeg_stats_in_redis")
+    @patch.object(StreamManager, "_try_next_stream", return_value=True)
+    def test_successful_switch_clears_redis_buffering(self, _try_next, _update_stats):
+        redis = _DictRedis()
+        metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
+        redis.hashes[metadata_key] = {
+            ChannelMetadataField.STATE: ChannelState.BUFFERING,
+        }
+        sm = _make_stream_manager(redis)
+
+        with patch("apps.proxy.live_proxy.input.manager.time") as mock_time:
+            mock_time.time.return_value = 10.0
+            with patch("apps.proxy.live_proxy.input.manager.log_system_event"):
+                sm._parse_ffmpeg_stats(
+                    "frame=100 fps=30 q=28.0 size=1024kB time=00:00:03.00 "
+                    "bitrate=500.0kbits/s speed=0.5x"
+                )
+
+        self.assertFalse(sm.buffering)
+        self.assertIsNone(sm.buffering_start_time)
+        self.assertEqual(
+            redis.hashes[metadata_key][ChannelMetadataField.STATE],
+            ChannelState.ACTIVE,
+        )
+
+    @patch.object(StreamManager, "_update_ffmpeg_stats_in_redis")
+    @patch.object(StreamManager, "_try_next_stream", return_value=True)
+    def test_good_speed_after_switch_does_not_re_latch_buffering(
+        self, _try_next, _update_stats
+    ):
+        redis = _DictRedis()
+        metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
+        sm = _make_stream_manager(redis)
+
+        with patch("apps.proxy.live_proxy.input.manager.time") as mock_time:
+            mock_time.time.return_value = 10.0
+            with patch("apps.proxy.live_proxy.input.manager.log_system_event"):
+                sm._parse_ffmpeg_stats(
+                    "frame=100 fps=30 q=28.0 size=1024kB time=00:00:03.00 "
+                    "bitrate=500.0kbits/s speed=0.5x"
+                )
+
+        # New stream is healthy; recovery must not depend on self.buffering.
+        sm._parse_ffmpeg_stats(
+            "frame=200 fps=30 q=28.0 size=2048kB time=00:00:06.00 "
+            "bitrate=700.0kbits/s speed=1.05x"
+        )
+
+        self.assertFalse(sm.buffering)
+        self.assertEqual(
+            redis.hashes[metadata_key][ChannelMetadataField.STATE],
+            ChannelState.ACTIVE,
+        )
+
+    @patch.object(StreamManager, "_update_ffmpeg_stats_in_redis")
+    @patch.object(StreamManager, "_try_next_stream", return_value=False)
+    def test_failed_switch_keeps_buffering_state(self, _try_next, _update_stats):
+        redis = _DictRedis()
+        metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
+        sm = _make_stream_manager(redis)
+
+        with patch("apps.proxy.live_proxy.input.manager.time") as mock_time:
+            mock_time.time.return_value = 10.0
+            with patch("apps.proxy.live_proxy.input.manager.log_system_event"):
+                sm._parse_ffmpeg_stats(
+                    "frame=100 fps=30 q=28.0 size=1024kB time=00:00:03.00 "
+                    "bitrate=500.0kbits/s speed=0.5x"
+                )
+
+        self.assertTrue(sm.buffering)
+        self.assertEqual(
+            redis.hashes[metadata_key][ChannelMetadataField.STATE],
+            ChannelState.BUFFERING,
+        )