fix(proxy): enhance channel shutdown management and client reconnection handling

- Improved the handling of channel shutdown delays to ensure they reset correctly upon client reconnections, preventing premature shutdowns.
- Implemented logic to cancel pending shutdowns when clients reconnect, ensuring proper resource management and preventing client disconnect issues across uWSGI workers.
- Enhanced logging for shutdown processes and client management to provide clearer insights during channel operations.

apps/proxy/live_proxy/client_manager.py

@@ -256,6 +256,14 @@ class ClientManager:
 
                 # Store in Redis
                 if self.redis_client:
+                    from .services.channel_service import ChannelService
+
+                    if ChannelService.cancel_pending_shutdown(self.channel_id):
+                        logger.info(
+                            f"Cancelled pending shutdown for channel {self.channel_id} "
+                            f"(client {client_id} reconnected)"
+                        )
+
                     self.redis_client.hset(client_key, mapping=client_data)
                     self.redis_client.expire(client_key, self.client_ttl)
 
@@ -302,7 +310,10 @@ class ClientManager:
                 return len(self.clients)
 
         except Exception as e:
-            logger.error(f"Error adding client {client_id}: {e}")
+            logger.error(f"Error adding client {client_id}: {e}", exc_info=True)
+            with self.lock:
+                self.clients.discard(client_id)
+            self._registered_clients.discard(client_id)
             return False
 
     def remove_client(self, client_id):
@@ -337,7 +348,8 @@ class ClientManager:
         if remaining == 0:
             logger.warning(f"Last client removed: {client_id} - channel may shut down soon")
             disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
-            self.redis_client.setex(disconnect_key, 60, str(time.time()))
+            ttl = max(int(ConfigHelper.channel_shutdown_delay() * 2), 60)
+            self.redis_client.setex(disconnect_key, ttl, str(time.time()))
 
         self._notify_owner_of_activity()
 

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

@@ -348,7 +348,11 @@ class FMP4StreamGenerator:
                             client_count = proxy_server.client_managers[
                                 self.channel_id
                             ].get_total_client_count()
-                            if client_count <= 1 and proxy_server.am_i_owner(self.channel_id):
+                            if (
+                                client_count <= 1
+                                and proxy_server.am_i_owner(self.channel_id)
+                                and ConfigHelper.channel_shutdown_delay() <= 0
+                            ):
                                 try:
                                     try:
                                         obj = Channel.objects.get(uuid=self.channel_id)

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

@@ -603,7 +603,8 @@ class StreamGenerator:
                 if self.channel_id in proxy_server.client_managers:
                     client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
                     # Pool slots are global; the last client on any worker must release.
-                    if client_count <= 1:
+                    # During shutdown_delay, keep the slot until coordinated stop runs.
+                    if client_count <= 1 and ConfigHelper.channel_shutdown_delay() <= 0:
                         try:
                             try:
                                 obj = Channel.objects.get(uuid=self.channel_id)

apps/proxy/live_proxy/server.py

@@ -905,6 +905,81 @@ class ProxyServer:
             logger.error(f"Error cleaning zombie channel {channel_id}: {e}", exc_info=True)
             return False
 
+    @staticmethod
+    def _shutdown_disconnect_ttl():
+        delay = ConfigHelper.channel_shutdown_delay()
+        return max(int(delay * 2), 60)
+
+    def _wait_for_shutdown_delay(self, channel_id):
+        """
+        Wait until shutdown_delay has elapsed since the Redis disconnect
+        timestamp. Returns False if clients reconnect or the timer is cancelled.
+        Uses Redis state so concurrent disconnect handlers and multi-worker
+        reconnects always honour the latest last-client disconnect time.
+        """
+        if not self.redis_client:
+            return True
+
+        shutdown_delay = ConfigHelper.channel_shutdown_delay()
+        if shutdown_delay <= 0:
+            return True
+
+        disconnect_key = RedisKeys.last_client_disconnect(channel_id)
+        client_set_key = RedisKeys.clients(channel_id)
+        poll_interval = 1.0
+
+        logger.info(
+            f"Waiting up to {shutdown_delay}s before stopping channel {channel_id}..."
+        )
+
+        while True:
+            total = self.redis_client.scard(client_set_key) or 0
+            if total > 0:
+                logger.info(
+                    f"New clients connected during shutdown delay for "
+                    f"{channel_id} - aborting shutdown"
+                )
+                self.redis_client.delete(disconnect_key)
+                return False
+
+            disconnect_value = self.redis_client.get(disconnect_key)
+            if not disconnect_value:
+                logger.info(
+                    f"Shutdown delay cancelled for {channel_id} - aborting shutdown"
+                )
+                return False
+
+            try:
+                if isinstance(disconnect_value, bytes):
+                    disconnect_value = disconnect_value.decode()
+                disconnect_time = float(disconnect_value)
+            except (ValueError, TypeError):
+                logger.warning(
+                    f"Invalid disconnect timestamp for {channel_id}, aborting wait"
+                )
+                return False
+
+            elapsed = time.time() - disconnect_time
+            if elapsed >= shutdown_delay:
+                total = self.redis_client.scard(client_set_key) or 0
+                if total > 0:
+                    logger.info(
+                        f"Clients connected at end of shutdown delay for "
+                        f"{channel_id} - aborting shutdown"
+                    )
+                    self.redis_client.delete(disconnect_key)
+                    return False
+                return True
+
+            elapsed_display = max(0.0, elapsed)
+            remaining = max(0.0, shutdown_delay - elapsed)
+            logger.debug(
+                f"Channel {channel_id[:8]} shutdown timer: "
+                f"{elapsed_display:.1f}s of {shutdown_delay}s elapsed "
+                f"({remaining:.1f}s remaining)"
+            )
+            gevent.sleep(min(poll_interval, remaining))
+
     def handle_client_disconnect(self, channel_id):
         """
         Handle client disconnect event - check if channel should shut down and
@@ -995,18 +1070,20 @@ class ProxyServer:
                 disconnect_key = RedisKeys.last_client_disconnect(channel_id)
 
                 if shutdown_delay > 0:
-                    self.redis_client.setex(disconnect_key, 60, str(time.time()))
-                    logger.info(f"Waiting {shutdown_delay}s before stopping channel...")
-                    gevent.sleep(shutdown_delay)
-
-                    total = self.redis_client.scard(client_set_key) or 0
-                    if total > 0:
-                        logger.info(f"New clients connected during shutdown delay - aborting shutdown")
-                        self.redis_client.delete(disconnect_key)
+                    if not self.redis_client.get(disconnect_key):
+                        self.redis_client.setex(
+                            disconnect_key,
+                            self._shutdown_disconnect_ttl(),
+                            str(time.time()),
+                        )
+                    if not self._wait_for_shutdown_delay(channel_id):
                         return
-
-                if shutdown_delay <= 0:
-                    self.redis_client.setex(disconnect_key, 60, str(time.time()))
+                else:
+                    self.redis_client.setex(
+                        disconnect_key,
+                        self._shutdown_disconnect_ttl(),
+                        str(time.time()),
+                    )
 
                 # Coordinated stop runs local teardown + Redis cleanup once.
                 # Do not call _stop_upstream_before_redis_cleanup here — it races
@@ -1794,7 +1871,11 @@ class ProxyServer:
                                 if not disconnect_time:
                                     # First time seeing zero clients, set timestamp
                                     if self.redis_client:
-                                        self.redis_client.setex(disconnect_key, 60, str(current_time))
+                                        self.redis_client.setex(
+                                            disconnect_key,
+                                            self._shutdown_disconnect_ttl(),
+                                            str(current_time),
+                                        )
                                     logger.warning(f"No clients detected for channel {channel_id}, starting shutdown timer")
                                 elif current_time - disconnect_time > ConfigHelper.channel_shutdown_delay():
                                     # We've had no clients for the shutdown delay period
@@ -1808,7 +1889,9 @@ class ProxyServer:
                             else:
                                 # There are clients or we're still connecting - clear any disconnect timestamp
                                 if self.redis_client:
-                                    self.redis_client.delete(f"live:channel:{channel_id}:last_client_disconnect_time")
+                                    self.redis_client.delete(
+                                        RedisKeys.last_client_disconnect(channel_id)
+                                    )
 
                         else:
                             # === NON-OWNER CHANNEL HANDLING ===

apps/proxy/live_proxy/services/channel_service.py

@@ -93,10 +93,103 @@ class ChannelService:
 
     @staticmethod
     def is_channel_unavailable_for_new_clients(channel_id):
-        """Reject new stream requests while teardown is active or shutdown is pending."""
-        return (
-            ChannelService.is_channel_teardown_active(channel_id)
-            or ChannelService.is_shutdown_pending(channel_id)
+        """Reject new stream requests only while coordinated teardown is active."""
+        return ChannelService.is_channel_teardown_active(channel_id)
+
+    @staticmethod
+    def cancel_pending_shutdown(channel_id):
+        """
+        Abort the post-disconnect grace timer when a client reconnects.
+
+        Clears the disconnect timestamp and any leaked stopping markers. When
+        upstream is still active but the last client released its profile slot
+        during the grace window, re-reserve the slot from Redis metadata.
+
+        Does not run during coordinated stop_channel() — clearing teardown
+        markers mid-stop would leave clients attached to upstream that is
+        about to be torn down.
+        """
+        from django.db import close_old_connections
+
+        proxy_server = ProxyServer.get_instance()
+        if not proxy_server.redis_client:
+            return False
+
+        if channel_id in proxy_server._stopping_channels:
+            return False
+
+        disconnect_key = RedisKeys.last_client_disconnect(channel_id)
+        had_pending = bool(proxy_server.redis_client.exists(disconnect_key))
+        in_grace = had_pending or ChannelService.is_shutdown_pending(channel_id)
+
+        if not in_grace:
+            return False
+
+        try:
+            proxy_server.redis_client.delete(disconnect_key)
+
+            metadata_key = RedisKeys.channel_metadata(channel_id)
+            state = proxy_server.redis_client.hget(
+                metadata_key, ChannelMetadataField.STATE
+            )
+            if state:
+                state_str = state.decode() if isinstance(state, bytes) else state
+                if state_str == ChannelState.STOPPING:
+                    proxy_server.redis_client.hset(metadata_key, mapping={
+                        ChannelMetadataField.STATE: ChannelState.ACTIVE,
+                        ChannelMetadataField.STATE_CHANGED_AT: str(time.time()),
+                    })
+
+            stop_key = RedisKeys.channel_stopping(channel_id)
+            if proxy_server.redis_client.exists(stop_key):
+                proxy_server.redis_client.delete(stop_key)
+
+            if ChannelService._channel_proxy_is_active(
+                proxy_server.redis_client, channel_id
+            ):
+                from apps.channels.models import Channel
+
+                channel = Channel.objects.filter(uuid=channel_id).first()
+                if channel and not proxy_server.redis_client.get(
+                    f"channel_stream:{channel.id}"
+                ):
+                    sid, pid, error, slot_reserved = channel.get_stream()
+                    if error:
+                        logger.warning(
+                            f"Could not re-reserve stream for {channel_id} "
+                            f"after shutdown cancel: {error}"
+                        )
+                    elif slot_reserved and sid and pid:
+                        proxy_server.redis_client.hset(metadata_key, mapping={
+                            ChannelMetadataField.STREAM_ID: str(sid),
+                            ChannelMetadataField.M3U_PROFILE: str(pid),
+                        })
+                        logger.info(
+                            f"Re-reserved profile slot for {channel_id} "
+                            f"(stream={sid}, profile={pid})"
+                        )
+        finally:
+            close_old_connections()
+
+        return True
+
+    @staticmethod
+    def _channel_proxy_is_active(redis_client, channel_id):
+        """True when live proxy metadata shows this channel is still running."""
+        metadata_key = RedisKeys.channel_metadata(channel_id)
+        if not redis_client.exists(metadata_key):
+            return False
+        state = redis_client.hget(metadata_key, ChannelMetadataField.STATE)
+        if state is None:
+            return False
+        if isinstance(state, bytes):
+            state = state.decode()
+        return state in (
+            ChannelState.ACTIVE,
+            ChannelState.WAITING_FOR_CLIENTS,
+            ChannelState.BUFFERING,
+            ChannelState.INITIALIZING,
+            ChannelState.CONNECTING,
         )
 
     @staticmethod