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

@@ -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.

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

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"""

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

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

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):
         """

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

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()

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.

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()