Method cleanup_task

apps/proxy/live_proxy/server.py:1723–2033

()

    def _start_cleanup_thread(self):
        """Start background thread to maintain ownership and clean up resources"""
        def cleanup_task():
            while True:
                try:
                    close_old_connections()
                    # Send worker heartbeat first
                    if self.redis_client:
                        worker_heartbeat_key = f"live:worker:{self.worker_id}:heartbeat"
                        self._execute_redis_command(
                            lambda: self.redis_client.setex(worker_heartbeat_key, 30, str(time.time()))
                        )

                    # Refresh channel registry
                    self.refresh_channel_registry()

                    # Recover channels whose stop_channel call never returned
                    self._recover_stuck_channel_stops()

                    # Create a unified list of all channels we have locally
                    all_local_channels = (
                        set(self.stream_managers.keys())
                        | set(self.client_managers.keys())
                        | set(self._live_stream_managers.keys())
                    )

                    # Single loop through all channels - process each exactly once
                    for channel_id in list(all_local_channels):
                        if self.am_i_owner(channel_id):
                            # === OWNER CHANNEL HANDLING ===
                            # Extend ownership lease
                            self.extend_ownership(channel_id)

                            # Get channel state from metadata hash
                            channel_state = "unknown"
                            if self.redis_client:
                                metadata_key = RedisKeys.channel_metadata(channel_id)
                                metadata = self.redis_client.hgetall(metadata_key)
                                if metadata and 'state' in metadata:
                                    channel_state = metadata['state']

                            # Check if channel has any clients left
                            total_clients = 0
                            if channel_id in self.client_managers:
                                client_manager = self.client_managers[channel_id]
                                total_clients = client_manager.get_total_client_count()
                            else:
                                # This can happen during reconnection attempts or crashes
                                # Check Redis directly for any connected clients
                                if self.redis_client:
                                    client_set_key = RedisKeys.clients(channel_id)
                                    total_clients = self.redis_client.scard(client_set_key) or 0

                                    if total_clients == 0:
                                        logger.warning(f"Channel {channel_id} is missing client_manager but we're the owner with 0 clients - will trigger cleanup")

                            # Log client count periodically
                            if time.time() % 30 < 1:  # Every ~30 seconds
                                logger.info(f"Channel {channel_id} has {total_clients} clients, state: {channel_state}")