From 64912c61641642f247ad25948da38947dc79fd4d Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 19 Jul 2026 14:06:14 +0000 Subject: [PATCH] fix(live): prevent geventpool DB connection leaks and enhance channel name resolution (Fixes #1418) This update addresses a critical issue where geventpool database connections were not properly released during stream setup and teardown, potentially leading to resource exhaustion. The `close_old_connections()` method is now consistently called in `finally` blocks across various components, ensuring that connections are released even in the event of exceptions. Additionally, the logic for resolving channel and stream names has been improved to prioritize caller-supplied names and Redis values, reducing reliance on ORM during initialization. This change enhances performance and stability, particularly in high-load scenarios. Tests have been updated to validate these improvements. --- CHANGELOG.md | 2 + apps/proxy/live_proxy/channel_status.py | 85 ++++-- apps/proxy/live_proxy/input/manager.py | 26 +- .../proxy/live_proxy/output/fmp4/generator.py | 46 ++- apps/proxy/live_proxy/output/ts/generator.py | 44 ++- apps/proxy/live_proxy/server.py | 69 +++-- .../live_proxy/services/channel_service.py | 107 +++---- .../live_proxy/tests/test_live_db_cleanup.py | 269 +++++++++++++++++- apps/proxy/live_proxy/utils.py | 23 ++ apps/proxy/live_proxy/views.py | 136 +++++---- 10 files changed, 619 insertions(+), 188 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b2fc04e..719e551c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -58,6 +58,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **Live proxy no longer leaks geventpool DB connections across stream setup / teardown.** `stream_ts()` released its checkout only on the happy path before `StreamingHttpResponse`, so early JSON failures, exceptions, 404s, and aborted channel init left slots counted against `MAX_CONNS` until restart (XC/`player_api` hung while `/api/core/version/` stayed fast). Setup now always calls `close_old_connections()` in a `finally` (including re-raised `Http404`); `ProxyServer.initialize_channel`, `ChannelService.initialize_channel`, XC auth, `next_stream`, channel status, and the ffmpeg stderr reader do the same. Channel/stream display names are taken from the caller or Redis during `StreamManager` / TS / fMP4 generator construction so those paths no longer check out the pool just to resolve a name. (Fixes #1418) + - **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. diff --git a/apps/proxy/live_proxy/channel_status.py b/apps/proxy/live_proxy/channel_status.py index 58c66fbc..fb58af0c 100644 --- a/apps/proxy/live_proxy/channel_status.py +++ b/apps/proxy/live_proxy/channel_status.py @@ -6,7 +6,7 @@ from .constants import TS_PACKET_SIZE, ChannelMetadataField from redis.exceptions import ConnectionError, TimeoutError from .utils import get_logger from .client_manager import ClientManager -from django.db import DatabaseError +from django.db import DatabaseError, close_old_connections logger = get_logger() @@ -46,41 +46,64 @@ class ChannelStatus: 'buffer_index': int(buffer_index_value) if buffer_index_value else 0, } - # Add stream ID and name information - stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID) - if stream_id_bytes: - try: - stream_id = int(stream_id_bytes) - info['stream_id'] = stream_id + channel_name = metadata.get(ChannelMetadataField.CHANNEL_NAME) + if channel_name: + info['channel_name'] = ( + channel_name.decode() if isinstance(channel_name, bytes) else channel_name + ) - # Look up stream name from database + # Prefer Redis names (written at channel init). ORM is a fallback only and + # always released in finally so admin status cannot leak pool slots. + try: + stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID) + if stream_id_bytes: try: - from apps.channels.models import Stream - stream = Stream.objects.filter(id=stream_id).first() - if stream: - info['stream_name'] = stream.name - except (ImportError, DatabaseError) as e: - logger.warning(f"Failed to get stream name for ID {stream_id}: {e}") - except ValueError: - logger.warning(f"Invalid stream_id format in Redis: {stream_id_bytes}") + stream_id = int(stream_id_bytes) + info['stream_id'] = stream_id - # Add M3U profile information - m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE) - if m3u_profile_id_bytes: - try: - m3u_profile_id = int(m3u_profile_id_bytes) - info['m3u_profile_id'] = m3u_profile_id + stream_name = metadata.get(ChannelMetadataField.STREAM_NAME) + if stream_name: + info['stream_name'] = ( + stream_name.decode() + if isinstance(stream_name, bytes) + else stream_name + ) + else: + try: + from apps.channels.models import Stream + stream = Stream.objects.filter(id=stream_id).first() + if stream: + info['stream_name'] = stream.name + except (ImportError, DatabaseError) as e: + logger.warning( + f"Failed to get stream name for ID {stream_id}: {e}" + ) + except ValueError: + logger.warning(f"Invalid stream_id format in Redis: {stream_id_bytes}") - # Look up M3U profile name from database + m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE) + if m3u_profile_id_bytes: try: - from apps.m3u.models import M3UAccountProfile - m3u_profile = M3UAccountProfile.objects.filter(id=m3u_profile_id).first() - if m3u_profile: - info['m3u_profile_name'] = m3u_profile.name - except (ImportError, DatabaseError) as e: - logger.warning(f"Failed to get M3U profile name for ID {m3u_profile_id}: {e}") - except ValueError: - logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id_bytes}") + m3u_profile_id = int(m3u_profile_id_bytes) + info['m3u_profile_id'] = m3u_profile_id + + try: + from apps.m3u.models import M3UAccountProfile + m3u_profile = M3UAccountProfile.objects.filter( + id=m3u_profile_id + ).first() + if m3u_profile: + info['m3u_profile_name'] = m3u_profile.name + except (ImportError, DatabaseError) as e: + logger.warning( + f"Failed to get M3U profile name for ID {m3u_profile_id}: {e}" + ) + except ValueError: + logger.warning( + f"Invalid m3u_profile_id format in Redis: {m3u_profile_id_bytes}" + ) + finally: + close_old_connections() # Add timing information state_changed_field = ChannelMetadataField.STATE_CHANGED_AT diff --git a/apps/proxy/live_proxy/input/manager.py b/apps/proxy/live_proxy/input/manager.py index e12a199e..ddc46685 100644 --- a/apps/proxy/live_proxy/input/manager.py +++ b/apps/proxy/live_proxy/input/manager.py @@ -17,21 +17,31 @@ from ..redis_keys import RedisKeys from ..constants import ChannelState, EventType, StreamType, ChannelMetadataField, TS_PACKET_SIZE from ..config_helper import ConfigHelper from ..url_utils import get_alternate_streams, get_stream_info_for_switch, get_stream_object +from ..utils import resolve_channel_display_name logger = get_logger() class StreamManager: """Manages a connection to a TS stream without using raw sockets""" - def __init__(self, channel_id, url, buffer, user_agent=None, transcode=False, stream_id=None, worker_id=None): + def __init__( + self, + channel_id, + url, + buffer, + user_agent=None, + transcode=False, + stream_id=None, + worker_id=None, + channel_name=None, + ): # Basic properties self.channel_id = channel_id - # Cache channel name once to avoid repeated DB queries in hot retry/reconnect loops - try: - _name = Channel.objects.filter(uuid=channel_id).values_list('name', flat=True).first() - self.channel_name = _name if _name else str(channel_id) - except Exception: - self.channel_name = str(channel_id) + # Prefer caller/Redis name so construction never checks out a geventpool slot. + redis_client = getattr(buffer, "redis_client", None) + self.channel_name = resolve_channel_display_name( + channel_id, channel_name=channel_name, redis_client=redis_client + ) self.url = url self.buffer = buffer self.running = True @@ -950,6 +960,8 @@ class StreamManager: logger.error(f"Error in stderr reader thread for channel {self.channel_id}: {e}") except: pass + finally: + close_old_connections() def _log_stderr_content(self, content): """Log stderr content from FFmpeg with appropriate log levels""" diff --git a/apps/proxy/live_proxy/output/fmp4/generator.py b/apps/proxy/live_proxy/output/fmp4/generator.py index d967b5f9..0520f6b7 100644 --- a/apps/proxy/live_proxy/output/fmp4/generator.py +++ b/apps/proxy/live_proxy/output/fmp4/generator.py @@ -18,22 +18,47 @@ from ...constants import ChannelMetadataField from .buffer import FMP4StreamBuffer from .manager import FMP4_STATE_ACTIVE, INIT_SEGMENT_TIMEOUT from ...config_helper import ConfigHelper -from ...utils import get_logger +from ...utils import get_logger, resolve_channel_display_name logger = get_logger() -def create_fmp4_stream_generator(channel_id, client_id, client_ip, client_user_agent, - channel_initializing=False, user=None, fmt='fmp4'): - gen = FMP4StreamGenerator(channel_id, client_id, client_ip, client_user_agent, - channel_initializing, user, fmt=fmt) +def create_fmp4_stream_generator( + channel_id, + client_id, + client_ip, + client_user_agent, + channel_initializing=False, + user=None, + fmt='fmp4', + channel_name=None, +): + gen = FMP4StreamGenerator( + channel_id, + client_id, + client_ip, + client_user_agent, + channel_initializing, + user, + fmt=fmt, + channel_name=channel_name, + ) return gen.generate class FMP4StreamGenerator: - def __init__(self, channel_id, client_id, client_ip, client_user_agent, - channel_initializing=False, user=None, fmt='fmp4'): + def __init__( + self, + channel_id, + client_id, + client_ip, + client_user_agent, + channel_initializing=False, + user=None, + fmt='fmp4', + channel_name=None, + ): self.channel_id = channel_id self.client_id = client_id self.client_ip = client_ip @@ -41,12 +66,7 @@ class FMP4StreamGenerator: self.channel_initializing = channel_initializing self.user = user self.fmt = fmt - - try: - _name = Channel.objects.filter(uuid=channel_id).values_list('name', flat=True).first() - self.channel_name = _name if _name else str(channel_id) - except Exception: - self.channel_name = str(channel_id) + self.channel_name = resolve_channel_display_name(channel_id, channel_name=channel_name) self.stream_start_time = time.time() self.bytes_sent = 0 diff --git a/apps/proxy/live_proxy/output/ts/generator.py b/apps/proxy/live_proxy/output/ts/generator.py index 88175fcd..bc21a925 100644 --- a/apps/proxy/live_proxy/output/ts/generator.py +++ b/apps/proxy/live_proxy/output/ts/generator.py @@ -10,7 +10,7 @@ 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 +from ...utils import create_ts_packet, get_logger, resolve_channel_display_name from ...redis_keys import RedisKeys from ...constants import ChannelMetadataField from ...config_helper import ConfigHelper @@ -23,7 +23,17 @@ class StreamGenerator: data delivery, and cleanup. """ - def __init__(self, channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None): + def __init__( + self, + channel_id, + client_id, + client_ip, + client_user_agent, + channel_initializing=False, + user=None, + buffer=None, + channel_name=None, + ): """ Initialize the stream generator with client and channel details. @@ -36,6 +46,7 @@ class StreamGenerator: user: Authenticated user making the request buffer: Source StreamBuffer to read from. Resolved via ProxyServer.get_buffer() before construction; passed in so the generator is buffer-agnostic. + channel_name: Optional display name (avoids ORM during construction) """ self.channel_id = channel_id self.client_id = client_id @@ -44,12 +55,7 @@ class StreamGenerator: self.channel_initializing = channel_initializing self.user = user self._source_buffer = buffer - # Cache channel name once to avoid repeated DB queries for logging - try: - _name = Channel.objects.filter(uuid=channel_id).values_list('name', flat=True).first() - self.channel_name = _name if _name else str(channel_id) - except Exception: - self.channel_name = str(channel_id) + self.channel_name = resolve_channel_display_name(channel_id, channel_name=channel_name) # Performance and state tracking self.stream_start_time = time.time() @@ -650,10 +656,28 @@ class StreamGenerator: close_old_connections() -def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None): +def create_stream_generator( + channel_id, + client_id, + client_ip, + client_user_agent, + channel_initializing=False, + user=None, + buffer=None, + channel_name=None, +): """ Factory function to create a new stream generator. Returns a function that can be passed to StreamingHttpResponse. """ - generator = StreamGenerator(channel_id, client_id, client_ip, client_user_agent, channel_initializing, user=user, buffer=buffer) + generator = StreamGenerator( + channel_id, + client_id, + client_ip, + client_user_agent, + channel_initializing, + user=user, + buffer=buffer, + channel_name=channel_name, + ) return generator.generate diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py index 004a2d29..8dfc3581 100644 --- a/apps/proxy/live_proxy/server.py +++ b/apps/proxy/live_proxy/server.py @@ -25,7 +25,7 @@ from .client_manager import ClientManager from .output.fmp4.manager import FMP4RemuxManager from .output.profile.manager import OutputProfileManager, PROFILE_STATE_ACTIVE from .redis_keys import RedisKeys -from .constants import ChannelState, EventType, StreamType +from .constants import ChannelState, EventType, StreamType, ChannelMetadataField from .config_helper import ConfigHelper from .utils import get_logger @@ -560,7 +560,16 @@ class ProxyServer: logger.error(f"Error extending ownership: {e}") return False - def initialize_channel(self, url, channel_id, user_agent=None, transcode=False, stream_id=None): + def initialize_channel( + self, + url, + channel_id, + user_agent=None, + transcode=False, + stream_id=None, + channel_name=None, + stream_name=None, + ): """Initialize a channel without redundant active key""" try: if self._channel_unavailable_for_new_clients(channel_id): @@ -699,6 +708,33 @@ class ProxyServer: # We now own the channel - ONLY NOW should we set metadata with initializing state logger.info(f"Worker {self.worker_id} is now the owner of channel {channel_id}") + # Prefer caller-supplied names (no ORM). Fall back to Redis, then UUID string. + if not channel_name and self.redis_client: + try: + raw_name = self.redis_client.hget( + metadata_key, ChannelMetadataField.CHANNEL_NAME + ) + if raw_name: + channel_name = ( + raw_name.decode() if isinstance(raw_name, bytes) else raw_name + ) + except Exception: + pass + channel_name = channel_name or str(channel_id) + self._channel_names[channel_id] = channel_name + + if not stream_name and self.redis_client: + try: + raw_stream = self.redis_client.hget( + metadata_key, ChannelMetadataField.STREAM_NAME + ) + if raw_stream: + stream_name = ( + raw_stream.decode() if isinstance(raw_stream, bytes) else raw_stream + ) + except Exception: + pass + if self.redis_client: # NOW create or update metadata with initializing state metadata = { @@ -718,6 +754,12 @@ class ProxyServer: else: logger.warning(f"No stream_id provided for channel {channel_id} during initialization") + # Persist display names early so other workers/stats skip ORM + if channel_name and channel_name != str(channel_id): + metadata[ChannelMetadataField.CHANNEL_NAME] = channel_name + if stream_name: + metadata[ChannelMetadataField.STREAM_NAME] = stream_name + # Set channel metadata BEFORE creating the StreamManager self.redis_client.hset(metadata_key, mapping=metadata) self.redis_client.expire(metadata_key, 3600) # Increased TTL from 30 seconds to 1 hour @@ -742,26 +784,14 @@ class ProxyServer: user_agent=channel_user_agent, transcode=transcode, stream_id=channel_stream_id, # Pass stream ID to the manager - worker_id=self.worker_id # Pass worker_id explicitly to eliminate circular dependency + worker_id=self.worker_id, # Pass worker_id explicitly to eliminate circular dependency + channel_name=channel_name, ) logger.info(f"Created StreamManager for channel {channel_id} with stream ID {channel_stream_id}") self.stream_managers[channel_id] = stream_manager - # Log channel start event + # Log channel start event (names already resolved without ORM) try: - _name = Channel.objects.filter(uuid=channel_id).values_list('name', flat=True).first() - channel_name = _name if _name else str(channel_id) - self._channel_names[channel_id] = channel_name - - # Get stream name if stream_id is available - stream_name = None - if channel_stream_id: - try: - stream_obj = Stream.objects.get(id=channel_stream_id) - stream_name = stream_obj.name - except Exception: - pass - log_system_event( 'channel_start', channel_id=channel_id, @@ -771,7 +801,6 @@ class ProxyServer: ) except Exception as e: logger.error(f"Could not log channel start event: {e}") - close_old_connections() # Create client manager with channel_id, redis_client AND worker_id (only if not already exists) if channel_id not in self.client_managers: @@ -808,6 +837,10 @@ class ProxyServer: # Release ownership on failure self.release_ownership(channel_id) return False + finally: + # StreamManager.__init__ and channel_start logging check out ORM on this + # greenlet; return the slot on every exit (success, early return, error). + close_old_connections() def check_if_channel_exists(self, channel_id): """ diff --git a/apps/proxy/live_proxy/services/channel_service.py b/apps/proxy/live_proxy/services/channel_service.py index df4509e4..dafa5c8b 100644 --- a/apps/proxy/live_proxy/services/channel_service.py +++ b/apps/proxy/live_proxy/services/channel_service.py @@ -7,6 +7,7 @@ import logging import time import json import gevent +from django.db import close_old_connections from apps.channels.models import Channel, Stream from ..server import ProxyServer from ..redis_keys import RedisKeys @@ -113,8 +114,6 @@ class ChannelService: 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 @@ -289,35 +288,58 @@ class ChannelService: """ proxy_server = ProxyServer.get_instance() - if stream_id and proxy_server.redis_client: - metadata_key = RedisKeys.channel_metadata(channel_id) - # Check if metadata already exists - if proxy_server.redis_client.exists(metadata_key): - # Just update the existing metadata with stream_id - proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STREAM_ID, str(stream_id)) - logger.info(f"Pre-set stream ID {stream_id} in Redis for channel {channel_id}") - else: - # Create initial metadata with essential values - initial_metadata = { - ChannelMetadataField.STREAM_ID: str(stream_id), - "temp_init": str(time.time()) - } - proxy_server.redis_client.hset(metadata_key, mapping=initial_metadata) - logger.info(f"Created initial metadata with stream_id {stream_id} for channel {channel_id}") + try: + if stream_id and proxy_server.redis_client: + metadata_key = RedisKeys.channel_metadata(channel_id) + # Check if metadata already exists + if proxy_server.redis_client.exists(metadata_key): + # Just update the existing metadata with stream_id + proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STREAM_ID, str(stream_id)) + logger.info(f"Pre-set stream ID {stream_id} in Redis for channel {channel_id}") + else: + # Create initial metadata with essential values + initial_metadata = { + ChannelMetadataField.STREAM_ID: str(stream_id), + "temp_init": str(time.time()) + } + proxy_server.redis_client.hset(metadata_key, mapping=initial_metadata) + logger.info(f"Created initial metadata with stream_id {stream_id} for channel {channel_id}") - # Verify the stream_id was set - stream_id_value = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID) - if stream_id_value: - logger.debug(f"Verified stream_id {stream_id_value} is now set in Redis") - else: - logger.error(f"Failed to set stream_id {stream_id} in Redis before initialization") + # Verify the stream_id was set + stream_id_value = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID) + if stream_id_value: + logger.debug(f"Verified stream_id {stream_id_value} is now set in Redis") + else: + logger.error(f"Failed to set stream_id {stream_id} in Redis before initialization") - # Now proceed with channel initialization - success = proxy_server.initialize_channel(stream_url, channel_id, user_agent, transcode, stream_id) + if not channel_name: + try: + channel_name = Channel.objects.filter(uuid=channel_id).values_list( + 'name', flat=True + ).first() + except Exception as e: + logger.warning(f"Failed to load channel name for {channel_id}: {e}") + if not stream_name and stream_id: + try: + stream_name = Stream.objects.filter(id=stream_id).values_list( + 'name', flat=True + ).first() + except Exception as e: + logger.warning(f"Failed to load stream name for {stream_id}: {e}") - # Store additional metadata if initialization was successful - if success and proxy_server.redis_client: - try: + # Now proceed with channel initialization + success = proxy_server.initialize_channel( + stream_url, + channel_id, + user_agent, + transcode, + stream_id, + channel_name=channel_name, + stream_name=stream_name, + ) + + # 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: @@ -326,31 +348,17 @@ class ChannelService: 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}") + if channel_name: + update_data[ChannelMetadataField.CHANNEL_NAME] = channel_name + if stream_name: + update_data[ChannelMetadataField.STREAM_NAME] = stream_name 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 + return success + finally: + close_old_connections() @staticmethod def change_stream_url(channel_id, new_url=None, user_agent=None, target_stream_id=None, m3u_profile_id=None, stream_name=None): @@ -910,7 +918,6 @@ class ChannelService: 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 diff --git a/apps/proxy/live_proxy/tests/test_live_db_cleanup.py b/apps/proxy/live_proxy/tests/test_live_db_cleanup.py index cb6c977b..947bd360 100644 --- a/apps/proxy/live_proxy/tests/test_live_db_cleanup.py +++ b/apps/proxy/live_proxy/tests/test_live_db_cleanup.py @@ -2,7 +2,7 @@ from unittest.mock import MagicMock, patch -from django.http import StreamingHttpResponse +from django.http import Http404, JsonResponse, StreamingHttpResponse from django.test import RequestFactory, SimpleTestCase @@ -10,6 +10,18 @@ class StreamTsDbCleanupTests(SimpleTestCase): def setUp(self): self.factory = RequestFactory() + def _active_proxy(self, client_manager=None): + client_manager = client_manager or 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() + return proxy_server, client_manager + @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") @@ -35,15 +47,7 @@ class StreamTsDbCleanupTests(SimpleTestCase): 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() + proxy_server, client_manager = self._active_proxy() mock_proxy_cls.get_instance.return_value = proxy_server def _generate(): @@ -62,6 +66,86 @@ class StreamTsDbCleanupTests(SimpleTestCase): client_manager.add_client.assert_called_once() mock_close.assert_called_once() + @patch("apps.proxy.live_proxy.views.close_old_connections") + @patch("apps.proxy.live_proxy.views.create_stream_generator", side_effect=RuntimeError("orm blew up")) + @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_on_exception( + self, + mock_proxy_cls, + _network_ok, + mock_get_stream_object, + _unavailable, + _output_profile, + _output_format, + _create_generator, + mock_close, + ): + channel = MagicMock() + channel.id = 1 + channel.uuid = "channel-uuid" + channel.name = "Test Channel" + mock_get_stream_object.return_value = channel + + proxy_server, _ = self._active_proxy() + mock_proxy_cls.get_instance.return_value = proxy_server + + 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, JsonResponse) + self.assertEqual(response.status_code, 500) + mock_close.assert_called_once() + + @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_on_early_client_register_failure( + self, + mock_proxy_cls, + _network_ok, + mock_get_stream_object, + _unavailable, + _output_profile, + _output_format, + _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() + client_manager.add_client.return_value = False + proxy_server, _ = self._active_proxy(client_manager=client_manager) + mock_proxy_cls.get_instance.return_value = proxy_server + + 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, JsonResponse) + self.assertEqual(response.status_code, 503) + mock_close.assert_called_once() + class UrlUtilsDbCleanupTests(SimpleTestCase): @patch("apps.proxy.live_proxy.url_utils.close_old_connections") @@ -141,3 +225,168 @@ class TsGeneratorDbCleanupTests(SimpleTestCase): gen._cleanup() mock_close.assert_called_once() + + +class InitializeChannelDbCleanupTests(SimpleTestCase): + @patch("apps.proxy.live_proxy.services.channel_service.close_old_connections") + @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer") + def test_channel_service_initialize_closes_db_on_failure(self, mock_proxy_cls, mock_close): + proxy_server = MagicMock() + proxy_server.redis_client = None + proxy_server.initialize_channel.return_value = False + mock_proxy_cls.get_instance.return_value = proxy_server + + from apps.proxy.live_proxy.services.channel_service import ChannelService + + result = ChannelService.initialize_channel( + "channel-uuid", + "http://example.com/stream.ts", + "ua", + ) + + self.assertFalse(result) + mock_close.assert_called_once() + + @patch("apps.proxy.live_proxy.server.close_old_connections") + @patch("apps.proxy.live_proxy.server.StreamManager", side_effect=RuntimeError("manager init failed")) + def test_proxy_server_initialize_closes_db_on_stream_manager_failure( + self, _stream_manager, mock_close + ): + from apps.proxy.live_proxy.server import ProxyServer + + proxy = ProxyServer.__new__(ProxyServer) + proxy.redis_client = MagicMock() + proxy.redis_client.exists.return_value = False + proxy.redis_client.hgetall.return_value = {} + proxy.stream_buffers = {} + proxy.client_managers = {} + proxy.stream_managers = {} + proxy._live_stream_managers = {} + proxy._channel_names = {} + proxy.worker_id = "worker-1" + proxy._channel_unavailable_for_new_clients = MagicMock(return_value=False) + proxy._has_local_upstream_activity = MagicMock(return_value=False) + proxy.get_channel_owner = MagicMock(return_value=None) + proxy.try_acquire_ownership = MagicMock(return_value=True) + proxy.release_ownership = MagicMock() + proxy.am_i_owner = MagicMock(return_value=True) + proxy.update_channel_state = MagicMock() + + with patch("apps.proxy.live_proxy.server.StreamBuffer"), patch( + "apps.proxy.live_proxy.server.RedisClient" + ): + result = proxy.initialize_channel( + "http://example.com/stream.ts", + "channel-uuid", + user_agent="ua", + stream_id=1, + ) + + self.assertFalse(result) + proxy.release_ownership.assert_called_once_with("channel-uuid") + self.assertGreaterEqual(mock_close.call_count, 1) + + +class StreamManagerDbCleanupTests(SimpleTestCase): + @patch("apps.proxy.live_proxy.input.manager.Channel.objects") + def test_stream_manager_init_uses_passed_name_without_orm(self, mock_channel_objects): + from apps.proxy.live_proxy.input.manager import StreamManager + + buffer = MagicMock() + buffer.redis_client = None + buffer.channel_id = "channel-uuid" + + manager = StreamManager.__new__(StreamManager) + StreamManager.__init__( + manager, + "channel-uuid", + "http://example.com/stream.ts", + buffer, + user_agent="ua", + channel_name="Test Channel", + ) + + self.assertEqual(manager.channel_name, "Test Channel") + mock_channel_objects.filter.assert_not_called() + + @patch("apps.proxy.live_proxy.input.manager.close_old_connections") + def test_read_stderr_closes_db_on_exit(self, mock_close): + from apps.proxy.live_proxy.input.manager import StreamManager + + manager = StreamManager.__new__(StreamManager) + manager.channel_id = "channel-uuid" + manager.running = True + manager.transcode_process = MagicMock() + manager.transcode_process.stderr = None + + manager._read_stderr() + + mock_close.assert_called_once() + + +class GeneratorAndStatusDbCleanupTests(SimpleTestCase): + def setUp(self): + self.factory = RequestFactory() + + @patch("apps.proxy.live_proxy.output.ts.generator.Channel.objects") + def test_ts_generator_init_uses_passed_name_without_orm(self, mock_channel_objects): + from apps.proxy.live_proxy.output.ts.generator import StreamGenerator + + gen = StreamGenerator( + "channel-uuid", + "client-1", + "127.0.0.1", + "agent", + channel_name="CNN", + ) + + self.assertEqual(gen.channel_name, "CNN") + mock_channel_objects.filter.assert_not_called() + + @patch("apps.proxy.live_proxy.channel_status.close_old_connections") + @patch("apps.proxy.live_proxy.channel_status.ProxyServer") + def test_detailed_channel_info_closes_db(self, mock_proxy_cls, mock_close): + proxy_server = MagicMock() + proxy_server.redis_client = MagicMock() + proxy_server.redis_client.hgetall.return_value = { + "state": "active", + "stream_id": "7", + "stream_name": "Backup Feed", + "m3u_profile": "3", + } + proxy_server.redis_client.get.return_value = "1" + mock_proxy_cls.get_instance.return_value = proxy_server + + from apps.proxy.live_proxy.channel_status import ChannelStatus + + with patch( + "apps.m3u.models.M3UAccountProfile.objects.filter" + ) as mock_profile_filter: + mock_profile_filter.return_value.first.return_value = MagicMock(name="Profile A") + info = ChannelStatus.get_detailed_channel_info("channel-uuid") + + self.assertEqual(info["stream_name"], "Backup Feed") + mock_close.assert_called_once() + + @patch("apps.proxy.live_proxy.views.close_old_connections") + @patch("apps.proxy.live_proxy.views.get_stream_object", side_effect=Http404("missing")) + @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_on_http404( + self, + mock_proxy_cls, + _network_ok, + _get_stream_object, + mock_close, + ): + mock_proxy_cls.get_instance.return_value = MagicMock() + request = self.factory.get("/proxy/live/missing/") + request.user = MagicMock(is_authenticated=False) + + from apps.proxy.live_proxy.views import stream_ts + + # @api_view converts Http404 into a 404 response; finally still releases. + response = stream_ts(request, "missing") + + self.assertEqual(response.status_code, 404) + mock_close.assert_called_once() diff --git a/apps/proxy/live_proxy/utils.py b/apps/proxy/live_proxy/utils.py index c83ce3ba..e69cfbf8 100644 --- a/apps/proxy/live_proxy/utils.py +++ b/apps/proxy/live_proxy/utils.py @@ -5,6 +5,29 @@ import inspect logger = logging.getLogger("live_proxy") + +def resolve_channel_display_name(channel_id, channel_name=None, redis_client=None): + """Resolve a display name without ORM: explicit arg, then Redis, then UUID string.""" + if channel_name: + return channel_name + try: + client = redis_client + if client is None: + from .server import ProxyServer + client = ProxyServer.get_instance().redis_client + if client: + from .redis_keys import RedisKeys + from .constants import ChannelMetadataField + raw = client.hget( + RedisKeys.channel_metadata(channel_id), + ChannelMetadataField.CHANNEL_NAME, + ) + if raw: + return raw.decode() if isinstance(raw, bytes) else raw + except Exception: + pass + return str(channel_id) + def detect_stream_type(url): """ Detect if stream URL is HLS, RTSP/RTP, UDP, or TS format. diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py index d42ffdc0..0bf6d207 100644 --- a/apps/proxy/live_proxy/views.py +++ b/apps/proxy/live_proxy/views.py @@ -4,7 +4,13 @@ import random import re import pathlib from django.db import close_old_connections -from django.http import StreamingHttpResponse, JsonResponse, HttpResponseRedirect, HttpResponse +from django.http import ( + StreamingHttpResponse, + JsonResponse, + HttpResponseRedirect, + HttpResponse, + Http404, +) from django.views.decorators.csrf import csrf_exempt from django.shortcuts import get_object_or_404 from .server import ProxyServer @@ -117,16 +123,20 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): if user is None and hasattr(request, 'user') and request.user.is_authenticated: user = request.user - channel = get_stream_object(channel_id) - client_user_agent = None proxy_server = ProxyServer.get_instance() connection_allocated = False # Track if connection slot was allocated via get_stream() # Initialized before the try so the exception handler can always safely # check/clean it up, regardless of where in the setup a failure occurs. _client_pre_registered = False + channel = None + client_id = None + channel_display_name = None try: + channel = get_stream_object(channel_id) + channel_display_name = getattr(channel, "name", None) + # Generate a unique client ID client_id = f"client_{int(time.time() * 1000)}_{random.randint(1000, 9999)}" client_ip = get_client_ip(request) @@ -571,7 +581,11 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): # Use client_user_agent as fallback if stream_user_agent is None success = proxy_server.initialize_channel( - url, channel_id, stream_user_agent or client_user_agent, use_transcode + url, + channel_id, + stream_user_agent or client_user_agent, + use_transcode, + channel_name=channel_display_name, ) if not success: logger.error( @@ -668,26 +682,33 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): generate = create_fmp4_stream_generator( channel_id, client_id, client_ip, client_user_agent, channel_initializing, user=user, fmt=resolved_format, + channel_name=channel_display_name, ) content_type = "video/mp4" else: generate = create_stream_generator( - channel_id, client_id, client_ip, client_user_agent, channel_initializing, user=user, buffer=source_buffer + channel_id, + client_id, + client_ip, + client_user_agent, + channel_initializing, + user=user, + buffer=source_buffer, + channel_name=channel_display_name, ) content_type = "video/mp2t" - # Release ORM checkout before returning a long-lived StreamingHttpResponse. - close_old_connections() - response = StreamingHttpResponse( streaming_content=generate(), content_type=content_type ) response["Cache-Control"] = "no-cache" return response + except Http404: + raise except Exception as e: logger.error(f"Error in stream_ts: {e}", exc_info=True) - if connection_allocated: + if connection_allocated and channel is not None: try: if not channel.release_stream(): logger.warning(f"[{client_id}] Failed to release stream in exception handler") @@ -703,60 +724,71 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): except Exception: logger.warning(f"[{client_id}] Failed to remove client during exception cleanup") return JsonResponse({"error": str(e)}, status=500) + finally: + # Runs before StreamingHttpResponse is handed to the WSGI server, so the + # request greenlet does not hold a pool slot for the life of the stream. + # Also covers Http404 from get_stream_object (re-raised above). + close_old_connections() @api_view(["GET"]) @permission_classes([AllowAny]) def stream_xc(request, username, password, channel_id): - user = get_object_or_404(User, username=username) + try: + user = get_object_or_404(User, username=username) - extension = pathlib.Path(channel_id).suffix - channel_id = pathlib.Path(channel_id).stem + extension = pathlib.Path(channel_id).suffix + channel_id = pathlib.Path(channel_id).stem - if not network_access_allowed(request, 'STREAMS', user): - return Response({"error": "Forbidden"}, status=403) + if not network_access_allowed(request, 'STREAMS', user): + return Response({"error": "Forbidden"}, status=403) - custom_properties = user.custom_properties or {} + custom_properties = user.custom_properties or {} - if "xc_password" not in custom_properties: - return Response({"error": "Invalid credentials"}, status=401) + if "xc_password" not in custom_properties: + return Response({"error": "Invalid credentials"}, status=401) - if custom_properties["xc_password"] != password: - return Response({"error": "Invalid credentials"}, status=401) + if custom_properties["xc_password"] != password: + return Response({"error": "Invalid credentials"}, status=401) - if user.user_level < 10: - user_profile_count = user.channel_profiles.count() + if user.user_level < 10: + user_profile_count = user.channel_profiles.count() - # If user has ALL profiles or NO profiles, give unrestricted access - if user_profile_count == 0: - # No profile filtering - user sees all channels based on user_level - filters = { - "id": int(channel_id), - "user_level__lte": user.user_level - } - channel = Channel.objects.filter(**filters).first() + # If user has ALL profiles or NO profiles, give unrestricted access + if user_profile_count == 0: + # No profile filtering - user sees all channels based on user_level + filters = { + "id": int(channel_id), + "user_level__lte": user.user_level + } + channel = Channel.objects.filter(**filters).first() + else: + # User has specific limited profiles assigned + filters = { + "id": int(channel_id), + "channelprofilemembership__enabled": True, + "user_level__lte": user.user_level, + "channelprofilemembership__channel_profile__in": user.channel_profiles.all() + } + channel = Channel.objects.filter(**filters).distinct().first() + + if not channel: + return JsonResponse({"error": "Not found"}, status=404) else: - # User has specific limited profiles assigned - filters = { - "id": int(channel_id), - "channelprofilemembership__enabled": True, - "user_level__lte": user.user_level, - "channelprofilemembership__channel_profile__in": user.channel_profiles.all() - } - channel = Channel.objects.filter(**filters).distinct().first() + channel = get_object_or_404(Channel, id=channel_id) - if not channel: - return JsonResponse({"error": "Not found"}, status=404) - else: - channel = get_object_or_404(Channel, id=channel_id) - - if extension.lower() == '.mp4': - force_format = 'fmp4' - elif extension.lower() == '.ts': - force_format = 'mpegts' - else: - force_format = None - return stream_ts(request._request, str(channel.uuid), user, force_output_format=force_format) + if extension.lower() == '.mp4': + force_format = 'fmp4' + elif extension.lower() == '.ts': + force_format = 'mpegts' + else: + force_format = None + return stream_ts(request._request, str(channel.uuid), user, force_output_format=force_format) + except Http404: + raise + finally: + # Auth/channel lookup ORM above; stream_ts also releases on its own paths. + close_old_connections() @csrf_exempt @@ -907,6 +939,8 @@ def channel_status(request, channel_id=None): except Exception as e: logger.error(f"Error in channel_status: {e}", exc_info=True) return JsonResponse({"error": str(e)}, status=500) + finally: + close_old_connections() @csrf_exempt @@ -1123,6 +1157,10 @@ def next_stream(request, channel_id): return JsonResponse(response_data) + except Http404: + raise except Exception as e: logger.error(f"Failed to switch to next stream: {e}", exc_info=True) return JsonResponse({"error": str(e)}, status=500) + finally: + close_old_connections()