fix(channel management): Implement channel initialization grace period handling to ensure proper state transitions during live stream startup. Channels now honor the configured grace period for buffer filling, promoting to active state when conditions are met. Updated related tests and documentation for clarity. (Fixes #1380)

apps/channels/tests/test_ts_proxy_teardown.py

@@ -1,6 +1,6 @@
 """Tests for multi-worker channel teardown coordination."""
 import time
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 from django.test import TestCase
 
@@ -692,9 +692,8 @@ class InitWaitAbortTests(TestCase):
 
         self.assertEqual(generator._init_wait_abort_reason(server, time.time()), "client_gone")
 
-    def test_abort_when_connect_stalled_without_buffer(self):
-        from apps.proxy.config import TSConfig as Config
-
+    @patch("apps.proxy.live_proxy.output.ts.generator.ConfigHelper.channel_init_grace_period", return_value=10)
+    def test_abort_when_connect_stalled_without_buffer(self, _mock_grace):
         generator = self._make_generator()
         server = MagicMock()
         client_manager = MagicMock()
@@ -707,9 +706,128 @@ class InitWaitAbortTests(TestCase):
         server.redis_client.hget.return_value = b"connecting"
         server.redis_client.get.return_value = None
 
-        started = time.time() - (getattr(Config, "CONNECTION_TIMEOUT", 10) + 1)
+        started = time.time() - 11
         self.assertEqual(generator._init_wait_abort_reason(server, started), "stalled")
 
+    @patch("apps.proxy.live_proxy.output.ts.generator.ConfigHelper.channel_init_grace_period", return_value=30)
+    def test_no_stall_abort_within_init_grace_period(self, _mock_grace):
+        generator = self._make_generator()
+        server = MagicMock()
+        client_manager = MagicMock()
+        client_manager.clients = {"client-1"}
+        server.client_managers = {CHANNEL_ID: client_manager}
+        buffer = MagicMock()
+        buffer.index = 0
+        server.stream_buffers = {CHANNEL_ID: buffer}
+        server.redis_client = MagicMock()
+        server.redis_client.hget.return_value = b"connecting"
+        server.redis_client.get.return_value = None
+
+        started = time.time() - 15
+        self.assertIsNone(generator._init_wait_abort_reason(server, started))
+
+    @patch("apps.proxy.live_proxy.output.ts.generator.ConfigHelper.channel_init_grace_period", return_value=30)
+    def test_stall_abort_after_init_grace_period(self, _mock_grace):
+        generator = self._make_generator()
+        server = MagicMock()
+        client_manager = MagicMock()
+        client_manager.clients = {"client-1"}
+        server.client_managers = {CHANNEL_ID: client_manager}
+        buffer = MagicMock()
+        buffer.index = 0
+        server.stream_buffers = {CHANNEL_ID: buffer}
+        server.redis_client = MagicMock()
+        server.redis_client.hget.return_value = b"connecting"
+        server.redis_client.get.return_value = None
+
+        started = time.time() - 31
+        self.assertEqual(generator._init_wait_abort_reason(server, started), "stalled")
+
+
+class PromoteChannelWhenBufferReadyTests(TestCase):
+    def _mock_proxy(self, redis_client):
+        proxy_server = MagicMock()
+        proxy_server.redis_client = redis_client
+        return patch(
+            "apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance",
+            return_value=proxy_server,
+        ), proxy_server
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.initial_behind_chunks", return_value=4)
+    def test_buffer_ready_with_clients_becomes_active(self, _mock_chunks):
+        redis = MagicMock()
+        redis.hget.return_value = ChannelState.CONNECTING.encode()
+        redis.get.return_value = b"4"
+        redis.scard.return_value = 2
+
+        ctx, proxy_server = self._mock_proxy(redis)
+        with ctx:
+            result = ChannelService.promote_channel_when_buffer_ready(CHANNEL_ID)
+
+        self.assertEqual(result, ChannelState.ACTIVE)
+        proxy_server.update_channel_state.assert_called_once()
+        args = proxy_server.update_channel_state.call_args[0]
+        self.assertEqual(args[1], ChannelState.ACTIVE)
+        self.assertEqual(args[2]["clients_at_activation"], "2")
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.initial_behind_chunks", return_value=4)
+    def test_buffer_ready_without_clients_becomes_waiting(self, _mock_chunks):
+        redis = MagicMock()
+        redis.hget.return_value = ChannelState.CONNECTING.encode()
+        redis.get.return_value = b"5"
+        redis.scard.return_value = 0
+
+        ctx, proxy_server = self._mock_proxy(redis)
+        with ctx:
+            result = ChannelService.promote_channel_when_buffer_ready(CHANNEL_ID)
+
+        self.assertEqual(result, ChannelState.WAITING_FOR_CLIENTS)
+        proxy_server.update_channel_state.assert_called_once_with(
+            CHANNEL_ID,
+            ChannelState.WAITING_FOR_CLIENTS,
+            {
+                ChannelMetadataField.CONNECTION_READY_TIME: ANY,
+                ChannelMetadataField.BUFFER_CHUNKS: "5",
+            },
+        )
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.initial_behind_chunks", return_value=4)
+    def test_buffer_not_ready_does_not_promote(self, _mock_chunks):
+        redis = MagicMock()
+        redis.hget.return_value = ChannelState.CONNECTING.encode()
+        redis.get.return_value = b"2"
+
+        ctx, proxy_server = self._mock_proxy(redis)
+        with ctx:
+            result = ChannelService.promote_channel_when_buffer_ready(CHANNEL_ID)
+
+        self.assertIsNone(result)
+        proxy_server.update_channel_state.assert_not_called()
+
+    def test_waiting_for_clients_with_clients_becomes_active(self):
+        redis = MagicMock()
+
+        def hget_side_effect(key, field):
+            if field == ChannelMetadataField.STATE:
+                return ChannelState.WAITING_FOR_CLIENTS.encode()
+            if field == ChannelMetadataField.CONNECTION_READY_TIME:
+                return b"1700000000.0"
+            return None
+
+        redis.hget.side_effect = hget_side_effect
+        redis.scard.return_value = 1
+
+        ctx, proxy_server = self._mock_proxy(redis)
+        with ctx:
+            result = ChannelService.promote_channel_when_buffer_ready(CHANNEL_ID)
+
+        self.assertEqual(result, ChannelState.ACTIVE)
+        proxy_server.update_channel_state.assert_called_once_with(
+            CHANNEL_ID,
+            ChannelState.ACTIVE,
+            {"clients_at_activation": "1"},
+        )
+
 
 class UpstreamStopBroadcastTests(TestCase):
     def _make_server(self):

apps/proxy/config.py

@@ -133,7 +133,7 @@ class TSConfig(BaseConfig):
 
     @classmethod
     def get_channel_init_grace_period(cls):
-        """Get channel init grace period from database or default"""
+        """Max seconds to wait for initial buffer fill during channel startup."""
         settings = cls.get_proxy_settings()
         return settings.get("channel_init_grace_period", 5)
 

apps/proxy/live_proxy/client_manager.py

@@ -301,6 +301,8 @@ class ClientManager:
                     # Trigger channel stats update via WebSocket
                     self._trigger_stats_update()
 
+                    ChannelService.promote_channel_when_buffer_ready(self.channel_id)
+
                 # Get total clients across all workers
                 total_clients = self.get_total_client_count()
                 logger.info(f"New client connected: {client_id} (local: {len(self.clients)}, total: {total_clients})")

apps/proxy/live_proxy/config_helper.py

@@ -107,7 +107,7 @@ class ConfigHelper:
 
     @staticmethod
     def channel_init_grace_period():
-        """Get channel initialization grace period in seconds"""
+        """Max seconds to wait for initial buffer fill during channel startup."""
         return Config.get_channel_init_grace_period()
 
     @staticmethod

apps/proxy/live_proxy/input/manager.py

@@ -1420,6 +1420,12 @@ class StreamManager:
         """Check if connection retry is allowed"""
         return self.retry_count < self.max_retries
 
+    def _health_inactivity_threshold(self):
+        """How long without data before marking the stream unhealthy."""
+        if self.connected and getattr(self.buffer, 'index', 0) == 0:
+            return ConfigHelper.channel_init_grace_period()
+        return getattr(Config, 'CONNECTION_TIMEOUT', 10)
+
     def _monitor_health(self):
         """Monitor stream health and set flags for the main loop to handle recovery"""
         consecutive_unhealthy_checks = 0
@@ -1435,7 +1441,7 @@ class StreamManager:
             try:
                 now = time.time()
                 inactivity_duration = now - self.last_data_time
-                timeout_threshold = getattr(Config, 'CONNECTION_TIMEOUT', 10)
+                timeout_threshold = self._health_inactivity_threshold()
 
                 if inactivity_duration > timeout_threshold and self.connected:
                     if self.healthy:
@@ -1830,19 +1836,9 @@ class StreamManager:
                             timer.start()
                             return False
 
-                        # We have enough buffer, proceed with state change
-                        update_data = {
-                            ChannelMetadataField.STATE: ChannelState.WAITING_FOR_CLIENTS,
-                            ChannelMetadataField.CONNECTION_READY_TIME: current_time,
-                            ChannelMetadataField.STATE_CHANGED_AT: current_time,
-                            ChannelMetadataField.BUFFER_CHUNKS: str(current_buffer_index)
-                        }
-                        redis_client.hset(metadata_key, mapping=update_data)
+                        from ..services.channel_service import ChannelService
 
-                        # Get configured grace period or default
-                        grace_period = ConfigHelper.channel_init_grace_period()
-                        logger.info(f"STREAM MANAGER: Updated channel {channel_id} state: {current_state or 'None'} -> {ChannelState.WAITING_FOR_CLIENTS} with {current_buffer_index} buffer chunks")
-                        logger.info(f"Started initial connection grace period ({grace_period}s) for channel {channel_id}")
+                        ChannelService.promote_channel_when_buffer_ready(channel_id)
                     else:
                         logger.debug(f"Not changing state: channel {channel_id} already in {current_state} state")
         except Exception as e:

apps/proxy/live_proxy/output/ts/generator.py

@@ -159,8 +159,8 @@ class StreamGenerator:
                     return "client_gone"
 
         elapsed = time.time() - initialization_start
-        connection_timeout = getattr(Config, 'CONNECTION_TIMEOUT', 10)
-        if elapsed < connection_timeout or not proxy_server.redis_client:
+        init_grace_period = ConfigHelper.channel_init_grace_period()
+        if elapsed < init_grace_period or not proxy_server.redis_client:
             return None
 
         metadata_key = RedisKeys.channel_metadata(self.channel_id)
@@ -204,7 +204,7 @@ class StreamGenerator:
                 logger.warning(
                     f"[{self.client_id}] Channel {self.channel_id} stalled in connecting state "
                     f"with no buffer data after "
-                    f"{getattr(Config, 'CONNECTION_TIMEOUT', 10)}s, aborting init wait"
+                    f"{ConfigHelper.channel_init_grace_period()}s, aborting init wait"
                 )
                 yield create_ts_packet('error', "Error: Connection stalled")
                 return False

apps/proxy/live_proxy/server.py

@@ -789,7 +789,7 @@ class ProxyServer:
                 attempt_key = RedisKeys.connection_attempt(channel_id)
                 self.redis_client.setex(attempt_key, 60, str(time.time()))
 
-                logger.info(f"Channel {channel_id} in {ChannelState.CONNECTING} state - will start grace period after connection")
+                logger.info(f"Channel {channel_id} in {ChannelState.CONNECTING} state - waiting for buffer to fill")
             return True
 
         except Exception as e:
@@ -1758,7 +1758,7 @@ class ProxyServer:
                             if time.time() % 30 < 1:  # Every ~30 seconds
                                 logger.info(f"Channel {channel_id} has {total_clients} clients, state: {channel_state}")
 
-                            # If in connecting or waiting_for_clients state, check grace period
+                            # Pre-active channels: init timeouts and buffer-ready promotion
                             if channel_state in [ChannelState.INITIALIZING, ChannelState.CONNECTING, ChannelState.WAITING_FOR_CLIENTS]:
                                 # Check if channel is already stopping
                                 if self.redis_client:
@@ -1824,27 +1824,13 @@ class ProxyServer:
                                                 )
                                                 self._coordinated_stop_channel(channel_id)
                                                 continue
-                                elif connection_ready_time:
-                                    # We have clients now, but check grace period for state transition
-                                    grace_period = ConfigHelper.channel_init_grace_period()
-                                    time_since_ready = time.time() - connection_ready_time
+                                elif (
+                                    channel_state == ChannelState.WAITING_FOR_CLIENTS
+                                    and total_clients > 0
+                                ):
+                                    from .services.channel_service import ChannelService
 
-                                    logger.debug(f"GRACE PERIOD CHECK: Channel {channel_id} in {channel_state} state, "
-                                                 f"time_since_ready={time_since_ready:.1f}s, grace_period={grace_period}s, "
-                                                 f"total_clients={total_clients}")
-
-                                    if time_since_ready <= grace_period:
-                                        # Still within grace period
-                                        logger.debug(f"Channel {channel_id} in grace period - {time_since_ready:.1f}s of {grace_period}s elapsed")
-                                        continue
-                                    else:
-                                        # Grace period expired with clients - mark channel as active
-                                        logger.info(f"Grace period expired with {total_clients} clients - marking channel {channel_id} as active")
-                                        if self.update_channel_state(channel_id, ChannelState.ACTIVE, {
-                                            "grace_period_ended_at": str(time.time()),
-                                            "clients_at_activation": str(total_clients)
-                                        }):
-                                            logger.info(f"Channel {channel_id} activated with {total_clients} clients after grace period")
+                                    ChannelService.promote_channel_when_buffer_ready(channel_id)
                             # If active and no clients, start normal shutdown procedure
                             elif channel_state not in [ChannelState.CONNECTING, ChannelState.WAITING_FOR_CLIENTS] and total_clients == 0:
                                 # Check if channel is already stopping

apps/proxy/live_proxy/services/channel_service.py

@@ -192,6 +192,78 @@ class ChannelService:
             ChannelState.CONNECTING,
         )
 
+    @staticmethod
+    def promote_channel_when_buffer_ready(channel_id):
+        """
+        Promote channel state once the initial buffer threshold is met.
+
+        - connecting/initializing + buffer ready + clients -> active
+        - connecting/initializing + buffer ready + no clients -> waiting_for_clients
+        - waiting_for_clients + clients -> active
+
+        Returns the resulting state, or None when no promotion applies.
+        """
+        proxy_server = ProxyServer.get_instance()
+        redis_client = proxy_server.redis_client
+        if not redis_client:
+            return None
+
+        metadata_key = RedisKeys.channel_metadata(channel_id)
+        state_raw = redis_client.hget(metadata_key, ChannelMetadataField.STATE)
+        if not state_raw:
+            return None
+
+        state = state_raw.decode() if isinstance(state_raw, bytes) else state_raw
+        if state == ChannelState.ACTIVE:
+            return ChannelState.ACTIVE
+
+        if state == ChannelState.WAITING_FOR_CLIENTS:
+            ready_raw = redis_client.hget(
+                metadata_key, ChannelMetadataField.CONNECTION_READY_TIME
+            )
+            if not ready_raw:
+                return None
+            client_count = redis_client.scard(RedisKeys.clients(channel_id)) or 0
+            if client_count <= 0:
+                return ChannelState.WAITING_FOR_CLIENTS
+            proxy_server.update_channel_state(
+                channel_id,
+                ChannelState.ACTIVE,
+                {"clients_at_activation": str(client_count)},
+            )
+            return ChannelState.ACTIVE
+
+        if state not in (ChannelState.INITIALIZING, ChannelState.CONNECTING):
+            return None
+
+        try:
+            buffer_index = int(redis_client.get(RedisKeys.buffer_index(channel_id)) or 0)
+        except (TypeError, ValueError):
+            buffer_index = 0
+
+        chunks_needed = ConfigHelper.initial_behind_chunks()
+        if buffer_index < chunks_needed:
+            return None
+
+        client_count = redis_client.scard(RedisKeys.clients(channel_id)) or 0
+        new_state = (
+            ChannelState.ACTIVE if client_count > 0 else ChannelState.WAITING_FOR_CLIENTS
+        )
+        current_time = str(time.time())
+        extra = {
+            ChannelMetadataField.CONNECTION_READY_TIME: current_time,
+            ChannelMetadataField.BUFFER_CHUNKS: str(buffer_index),
+        }
+        if new_state == ChannelState.ACTIVE:
+            extra["clients_at_activation"] = str(client_count)
+
+        proxy_server.update_channel_state(channel_id, new_state, extra)
+        logger.info(
+            f"Channel {channel_id} buffer ready ({buffer_index}/{chunks_needed} chunks) "
+            f"-> {new_state} (clients={client_count})"
+        )
+        return new_state
+
     @staticmethod
     def initialize_channel(channel_id, stream_url, user_agent, transcode=False, stream_profile_value=None, stream_id=None, m3u_profile_id=None, channel_name=None, stream_name=None):
         """