diff --git a/CHANGELOG.md b/CHANGELOG.md index 1da516a1..1f1234c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **Live proxy now releases geventpool DB connections on more paths.** `stream_ts()` calls `close_old_connections()` before `StreamingHttpResponse` so clients waiting on channel init do not keep a pool slot while the generator polls Redis. `generate_stream_url()`, `get_stream_info_for_switch()`, `get_alternate_streams()`, and `get_connections_left()` release in `finally` blocks after ORM work on stream-manager failover paths that run outside Django's request cycle. TS client disconnect cleanup matches fMP4, and `ChannelService` metadata helpers release after their ORM lookups. + ## [0.27.0] - 2026-06-16 ### Added diff --git a/apps/proxy/live_proxy/output/ts/generator.py b/apps/proxy/live_proxy/output/ts/generator.py index 2bd88725..3524a2a6 100644 --- a/apps/proxy/live_proxy/output/ts/generator.py +++ b/apps/proxy/live_proxy/output/ts/generator.py @@ -7,6 +7,7 @@ import time import gevent from apps.proxy.config import TSConfig as Config from apps.channels.models import Channel, Stream +from django.db import close_old_connections from core.utils import log_system_event from ...server import ProxyServer from ...utils import create_ts_packet, get_logger @@ -590,60 +591,63 @@ class StreamGenerator: def _cleanup(self): """Clean up resources and report final statistics.""" - # Client cleanup - elapsed = time.time() - self.stream_start_time - local_clients = 0 - total_clients = 0 - proxy_server = ProxyServer.get_instance() + try: + # Client cleanup + elapsed = time.time() - self.stream_start_time + local_clients = 0 + total_clients = 0 + proxy_server = ProxyServer.get_instance() - # Release M3U profile stream allocation if this is the last client - stream_released = False - if proxy_server.redis_client: - try: - 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. - # During shutdown_delay, keep the slot until coordinated stop runs. - if client_count <= 1 and ConfigHelper.channel_shutdown_delay() <= 0: - try: + # Release M3U profile stream allocation if this is the last client + stream_released = False + if proxy_server.redis_client: + try: + 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. + # During shutdown_delay, keep the slot until coordinated stop runs. + if client_count <= 1 and ConfigHelper.channel_shutdown_delay() <= 0: try: - obj = Channel.objects.get(uuid=self.channel_id) - except (Channel.DoesNotExist, Exception): - obj = Stream.objects.get(stream_hash=self.channel_id) - stream_released = obj.release_stream() - if stream_released: - logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}") - else: - logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}") - except Exception as e: - logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}") - except Exception as e: - logger.error(f"[{self.client_id}] Error checking stream data for release: {e}") + try: + obj = Channel.objects.get(uuid=self.channel_id) + except (Channel.DoesNotExist, Exception): + obj = Stream.objects.get(stream_hash=self.channel_id) + stream_released = obj.release_stream() + if stream_released: + logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}") + else: + logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}") + except Exception as e: + logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}") + except Exception as e: + logger.error(f"[{self.client_id}] Error checking stream data for release: {e}") - if self.channel_id in proxy_server.client_managers: - client_manager = proxy_server.client_managers[self.channel_id] - if self.client_id in client_manager.clients: - local_clients = client_manager.remove_client(self.client_id) - else: - local_clients = client_manager.get_client_count() - total_clients = client_manager.get_total_client_count() - logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})") + if self.channel_id in proxy_server.client_managers: + client_manager = proxy_server.client_managers[self.channel_id] + if self.client_id in client_manager.clients: + local_clients = client_manager.remove_client(self.client_id) + else: + local_clients = client_manager.get_client_count() + total_clients = client_manager.get_total_client_count() + logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})") - # Log client disconnect event - try: - log_system_event( - 'client_disconnect', - channel_id=self.channel_id, - channel_name=self.channel_name, - client_ip=self.client_ip, - client_id=self.client_id, - user_agent=self.client_user_agent[:100] if self.client_user_agent else None, - duration=round(elapsed, 2), - bytes_sent=self.bytes_sent, - username=self.user.username if self.user else None - ) - except Exception as e: - logger.error(f"Could not log client disconnect event: {e}") + # Log client disconnect event + try: + log_system_event( + 'client_disconnect', + channel_id=self.channel_id, + channel_name=self.channel_name, + client_ip=self.client_ip, + client_id=self.client_id, + user_agent=self.client_user_agent[:100] if self.client_user_agent else None, + duration=round(elapsed, 2), + bytes_sent=self.bytes_sent, + username=self.user.username if self.user else None + ) + except Exception as e: + logger.error(f"Could not log client disconnect event: {e}") + finally: + close_old_connections() def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None): diff --git a/apps/proxy/live_proxy/services/channel_service.py b/apps/proxy/live_proxy/services/channel_service.py index 400f55f4..7ad4e71b 100644 --- a/apps/proxy/live_proxy/services/channel_service.py +++ b/apps/proxy/live_proxy/services/channel_service.py @@ -241,34 +241,38 @@ class ChannelService: # Store additional metadata if initialization was successful if success and proxy_server.redis_client: - metadata_key = RedisKeys.channel_metadata(channel_id) - update_data = {} - if stream_profile_value: - update_data[ChannelMetadataField.STREAM_PROFILE] = stream_profile_value - if stream_id: - update_data[ChannelMetadataField.STREAM_ID] = str(stream_id) - if m3u_profile_id: - update_data[ChannelMetadataField.M3U_PROFILE] = str(m3u_profile_id) - - # Store channel name and stream name so stats workers don't need DB calls try: - if not channel_name: - from apps.channels.models import Channel - channel_name = Channel.objects.filter(uuid=channel_id).values_list('name', flat=True).first() - if channel_name: - update_data[ChannelMetadataField.CHANNEL_NAME] = channel_name - else: - # No channel name means stream preview mode, use stream name as display fallback - if stream_id and not stream_name: - from apps.channels.models import Stream - stream_name = Stream.objects.filter(id=stream_id).values_list('name', flat=True).first() - if stream_name: - update_data[ChannelMetadataField.STREAM_NAME] = stream_name - except Exception as e: - logger.warning(f"Failed to store channel/stream names in Redis for {channel_id}: {e}") + metadata_key = RedisKeys.channel_metadata(channel_id) + update_data = {} + if stream_profile_value: + update_data[ChannelMetadataField.STREAM_PROFILE] = stream_profile_value + if stream_id: + update_data[ChannelMetadataField.STREAM_ID] = str(stream_id) + if m3u_profile_id: + update_data[ChannelMetadataField.M3U_PROFILE] = str(m3u_profile_id) - if update_data: - proxy_server.redis_client.hset(metadata_key, mapping=update_data) + # Store channel name and stream name so stats workers don't need DB calls + try: + if not channel_name: + from apps.channels.models import Channel + channel_name = Channel.objects.filter(uuid=channel_id).values_list('name', flat=True).first() + if channel_name: + update_data[ChannelMetadataField.CHANNEL_NAME] = channel_name + else: + # No channel name means stream preview mode, use stream name as display fallback + if stream_id and not stream_name: + from apps.channels.models import Stream + stream_name = Stream.objects.filter(id=stream_id).values_list('name', flat=True).first() + if stream_name: + update_data[ChannelMetadataField.STREAM_NAME] = stream_name + except Exception as e: + logger.warning(f"Failed to store channel/stream names in Redis for {channel_id}: {e}") + + if update_data: + proxy_server.redis_client.hset(metadata_key, mapping=update_data) + finally: + from django.db import close_old_connections + close_old_connections() return success @@ -736,53 +740,57 @@ class ChannelService: @staticmethod def _update_channel_metadata(channel_id, url, user_agent=None, stream_id=None, m3u_profile_id=None, stream_name=None): """Update channel metadata in Redis""" - proxy_server = ProxyServer.get_instance() + try: + proxy_server = ProxyServer.get_instance() - if not proxy_server.redis_client: - return False + if not proxy_server.redis_client: + return False - metadata_key = RedisKeys.channel_metadata(channel_id) + metadata_key = RedisKeys.channel_metadata(channel_id) - # First check if the key exists and what type it is - key_type = proxy_server.redis_client.type(metadata_key) - logger.debug(f"Redis key {metadata_key} is of type: {key_type}") + # First check if the key exists and what type it is + key_type = proxy_server.redis_client.type(metadata_key) + logger.debug(f"Redis key {metadata_key} is of type: {key_type}") - # Build metadata update dict - metadata = {ChannelMetadataField.URL: url} - if user_agent: - metadata[ChannelMetadataField.USER_AGENT] = user_agent - if stream_id: - metadata[ChannelMetadataField.STREAM_ID] = str(stream_id) - if not stream_name: - try: - from apps.channels.models import Stream - stream_name = Stream.objects.filter(id=stream_id).values_list('name', flat=True).first() - except Exception as e: - logger.warning(f"Failed to update stream name in Redis for stream {stream_id}: {e}") - if stream_name: - metadata[ChannelMetadataField.STREAM_NAME] = stream_name - if m3u_profile_id: - metadata[ChannelMetadataField.M3U_PROFILE] = str(m3u_profile_id) + # Build metadata update dict + metadata = {ChannelMetadataField.URL: url} + if user_agent: + metadata[ChannelMetadataField.USER_AGENT] = user_agent + if stream_id: + metadata[ChannelMetadataField.STREAM_ID] = str(stream_id) + if not stream_name: + try: + from apps.channels.models import Stream + stream_name = Stream.objects.filter(id=stream_id).values_list('name', flat=True).first() + except Exception as e: + logger.warning(f"Failed to update stream name in Redis for stream {stream_id}: {e}") + if stream_name: + metadata[ChannelMetadataField.STREAM_NAME] = stream_name + if m3u_profile_id: + metadata[ChannelMetadataField.M3U_PROFILE] = str(m3u_profile_id) - # Also update the stream switch time field - metadata[ChannelMetadataField.STREAM_SWITCH_TIME] = str(time.time()) + # Also update the stream switch time field + metadata[ChannelMetadataField.STREAM_SWITCH_TIME] = str(time.time()) - # Use the appropriate method based on the key type - if key_type == 'hash': - proxy_server.redis_client.hset(metadata_key, mapping=metadata) - elif key_type == 'none': # Key doesn't exist yet - proxy_server.redis_client.hset(metadata_key, mapping=metadata) - else: - # If key exists with wrong type, delete it and recreate - proxy_server.redis_client.delete(metadata_key) - proxy_server.redis_client.hset(metadata_key, mapping=metadata) + # Use the appropriate method based on the key type + if key_type == 'hash': + proxy_server.redis_client.hset(metadata_key, mapping=metadata) + elif key_type == 'none': # Key doesn't exist yet + proxy_server.redis_client.hset(metadata_key, mapping=metadata) + else: + # If key exists with wrong type, delete it and recreate + proxy_server.redis_client.delete(metadata_key) + proxy_server.redis_client.hset(metadata_key, mapping=metadata) - # Set switch request flag to ensure all workers see it - switch_key = RedisKeys.switch_request(channel_id) - proxy_server.redis_client.setex(switch_key, 30, url) # 30 second TTL + # Set switch request flag to ensure all workers see it + switch_key = RedisKeys.switch_request(channel_id) + proxy_server.redis_client.setex(switch_key, 30, url) # 30 second TTL - logger.debug(f"Updated metadata for channel {channel_id} in Redis") - return True + logger.debug(f"Updated metadata for channel {channel_id} in Redis") + return True + finally: + from django.db import close_old_connections + close_old_connections() @staticmethod def _publish_stream_switch_event(channel_id, new_url, user_agent=None, stream_id=None, m3u_profile_id=None): diff --git a/apps/proxy/live_proxy/tests/test_live_db_cleanup.py b/apps/proxy/live_proxy/tests/test_live_db_cleanup.py new file mode 100644 index 00000000..f2015a41 --- /dev/null +++ b/apps/proxy/live_proxy/tests/test_live_db_cleanup.py @@ -0,0 +1,142 @@ +"""Live proxy must release geventpool checkouts after ORM on stream and URL paths.""" + +from unittest.mock import MagicMock, patch + +from django.http import StreamingHttpResponse +from django.test import RequestFactory, SimpleTestCase + + +class StreamTsDbCleanupTests(SimpleTestCase): + def setUp(self): + self.factory = RequestFactory() + + @patch("apps.proxy.live_proxy.views.close_old_connections") + @patch("apps.proxy.live_proxy.views.create_stream_generator") + @patch("apps.proxy.live_proxy.views._resolve_output_format", return_value="mpegts") + @patch("apps.proxy.live_proxy.views._resolve_output_profile", return_value=None) + @patch("apps.proxy.live_proxy.views.ChannelService.is_channel_unavailable_for_new_clients", return_value=False) + @patch("apps.proxy.live_proxy.views.get_stream_object") + @patch("apps.proxy.live_proxy.views.network_access_allowed", return_value=True) + @patch("apps.proxy.live_proxy.views.ProxyServer") + def test_stream_ts_closes_db_before_streaming_response( + self, + mock_proxy_cls, + _network_ok, + mock_get_stream_object, + _unavailable, + _output_profile, + _output_format, + mock_create_generator, + mock_close, + ): + channel = MagicMock() + channel.id = 1 + channel.uuid = "channel-uuid" + channel.name = "Test Channel" + mock_get_stream_object.return_value = channel + + client_manager = MagicMock() + proxy_server = MagicMock() + proxy_server.redis_client = MagicMock() + proxy_server.redis_client.exists.return_value = True + proxy_server.redis_client.hgetall.return_value = {"state": "active"} + proxy_server.stream_buffers = {"channel-uuid": MagicMock()} + proxy_server.client_managers = {"channel-uuid": client_manager} + proxy_server.check_if_channel_exists.return_value = True + proxy_server.get_buffer.return_value = MagicMock() + mock_proxy_cls.get_instance.return_value = proxy_server + + def _generate(): + yield b"chunk" + + mock_create_generator.return_value = lambda: _generate() + + request = self.factory.get("/proxy/live/channel-uuid/") + request.user = MagicMock(is_authenticated=False) + + from apps.proxy.live_proxy.views import stream_ts + + response = stream_ts(request, "channel-uuid") + + self.assertIsInstance(response, StreamingHttpResponse) + mock_close.assert_called_once() + + +class UrlUtilsDbCleanupTests(SimpleTestCase): + @patch("apps.proxy.live_proxy.url_utils.close_old_connections") + @patch("apps.proxy.live_proxy.url_utils.get_stream_object") + def test_generate_stream_url_closes_db(self, mock_get_object, mock_close): + channel = MagicMock() + channel.get_stream.return_value = (None, None, "no streams", False) + mock_get_object.return_value = channel + + from apps.proxy.live_proxy.url_utils import generate_stream_url + + result = generate_stream_url("channel-uuid") + + self.assertIsNone(result[0]) + mock_close.assert_called_once() + + @patch("apps.proxy.live_proxy.url_utils.close_old_connections") + @patch("apps.proxy.live_proxy.url_utils.get_stream_object") + def test_get_alternate_streams_closes_db(self, mock_get_object, mock_close): + channel = MagicMock() + channel.streams.all.return_value.order_by.return_value.exists.return_value = False + mock_get_object.return_value = channel + + from apps.proxy.live_proxy.url_utils import get_alternate_streams + + result = get_alternate_streams("channel-uuid", current_stream_id=1) + + self.assertEqual(result, []) + mock_close.assert_called_once() + + @patch("apps.proxy.live_proxy.url_utils.close_old_connections") + @patch("apps.proxy.live_proxy.url_utils.get_object_or_404") + def test_get_stream_info_for_switch_closes_db_on_error(self, mock_get_404, mock_close): + mock_get_404.side_effect = RuntimeError("db error") + + from apps.proxy.live_proxy.url_utils import get_stream_info_for_switch + + result = get_stream_info_for_switch("channel-uuid", target_stream_id=99) + + self.assertIn("error", result) + mock_close.assert_called_once() + + @patch("apps.proxy.live_proxy.url_utils.close_old_connections") + @patch("apps.proxy.live_proxy.url_utils.M3UAccountProfile.objects.get") + def test_get_connections_left_closes_db(self, mock_get, mock_close): + mock_get.side_effect = Exception("not found") + + from apps.proxy.live_proxy.url_utils import get_connections_left + + result = get_connections_left(999) + + self.assertEqual(result, 0) + mock_close.assert_called_once() + + +class TsGeneratorDbCleanupTests(SimpleTestCase): + @patch("apps.proxy.live_proxy.output.ts.generator.close_old_connections") + @patch("apps.proxy.live_proxy.output.ts.generator.ProxyServer.get_instance") + def test_ts_cleanup_closes_db(self, mock_proxy_cls, mock_close): + proxy_server = MagicMock() + proxy_server.redis_client = None + proxy_server.client_managers = {} + mock_proxy_cls.return_value = proxy_server + + from apps.proxy.live_proxy.output.ts.generator import StreamGenerator + + gen = StreamGenerator.__new__(StreamGenerator) + gen.channel_id = "channel-uuid" + gen.client_id = "client-1" + gen.stream_start_time = 0 + gen.channel_name = "Test" + gen.client_ip = "127.0.0.1" + gen.client_user_agent = "agent" + gen.bytes_sent = 0 + gen.user = None + + gen._cleanup() + + mock_close.assert_called_once() diff --git a/apps/proxy/live_proxy/url_utils.py b/apps/proxy/live_proxy/url_utils.py index 7cdc92ef..2b32f7e6 100644 --- a/apps/proxy/live_proxy/url_utils.py +++ b/apps/proxy/live_proxy/url_utils.py @@ -5,6 +5,7 @@ Utilities for handling stream URLs and transformations. import logging import regex from typing import Optional, Tuple, List +from django.db import close_old_connections from django.shortcuts import get_object_or_404 from apps.channels.models import Channel, Stream from apps.m3u.models import M3UAccount, M3UAccountProfile @@ -156,6 +157,8 @@ def generate_stream_url( except Exception as e: logger.error(f"Error generating stream URL: {e}") return None, None, False, None, False, str(e) + finally: + close_old_connections() def transform_url(input_url: str, search_pattern: str, replace_pattern: str) -> str: """ @@ -300,6 +303,8 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] channel.release_stream() logger.error(f"Error getting stream info for switch: {e}", exc_info=True) return {'error': f'Error: {str(e)}'} + finally: + close_old_connections() def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = None) -> List[dict]: """ @@ -422,6 +427,8 @@ def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = No except Exception as e: logger.error(f"Error getting alternate streams for channel {channel_id}: {e}", exc_info=True) return [] + finally: + close_old_connections() def validate_stream_url(url, user_agent=None, timeout=(5, 5)): """ @@ -592,3 +599,5 @@ def get_connections_left(m3u_profile_id: int) -> int: except Exception as e: logger.error(f"Error getting connections left for M3U profile {m3u_profile_id}: {e}") return 0 + finally: + close_old_connections() diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py index 2119b4bd..001cfc14 100644 --- a/apps/proxy/live_proxy/views.py +++ b/apps/proxy/live_proxy/views.py @@ -3,6 +3,7 @@ import time import random import re import pathlib +from django.db import close_old_connections from django.http import StreamingHttpResponse, JsonResponse, HttpResponseRedirect, HttpResponse from django.views.decorators.csrf import csrf_exempt from django.shortcuts import get_object_or_404 @@ -617,7 +618,9 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): ) content_type = "video/mp2t" - # Return the StreamingHttpResponse from the main function + # Release ORM checkout before returning a long-lived StreamingHttpResponse. + close_old_connections() + response = StreamingHttpResponse( streaming_content=generate(), content_type=content_type )