Fix: Connection capacity leak - failed stream initialization permanently consumes connection slot

Fixed four leak paths in the TS proxy streaming initialization:

1. url_utils.py - generate_stream_url(): Wrapped post-get_stream() code in try/except that calls release_stream() on any failure

2. views.py - stream_ts() retry loop: Added release_stream() before returning 503 to clean up the error-checking get_stream() call at line 181 which could INCR without a matching release when the retry loop times out

3. views.py - stream_ts() init failure: Added guarded release_stream() when initialize_channel() returns False, using a connection_allocated flag to prevent incorrect decrements when joining an existing channel

4. views.py - stream_ts() outer except: Added guarded release_stream() as a safety net for unexpected exceptions, only releasing when the flag confirms the slot was allocated and the channel hasn't been initialized yet

The connection_allocated flag prevents double-decrements by tracking whether request flow actually INCR'd the counter (fresh initialization) vs
returned cached results (joining an existing channel on another worker)

apps/proxy/ts_proxy/url_utils.py

@@ -129,39 +129,41 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]
             logger.error(f"No stream available for channel {channel_id}: {error_reason}")
             return None, None, False, None
 
-        # Look up the Stream and Profile objects
+        # get_stream() allocated a connection slot - ensure it's released on any error
         try:
+            # Look up the Stream and Profile objects
             stream = Stream.objects.get(id=stream_id)
             profile = M3UAccountProfile.objects.get(id=profile_id)
-        except (Stream.DoesNotExist, M3UAccountProfile.DoesNotExist) as e:
+
-            logger.error(f"Error getting stream or profile: {e}")
+            # Get the M3U account profile for URL pattern
+            m3u_profile = profile
+
+            # Get the appropriate user agent
+            m3u_account = M3UAccount.objects.get(id=m3u_profile.m3u_account.id)
+            stream_user_agent = m3u_account.get_user_agent().user_agent
+
+            if stream_user_agent is None:
+                stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
+                logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
+
+            # Generate stream URL based on the selected profile
+            input_url = stream.url
+            stream_url = transform_url(input_url, m3u_profile.search_pattern, m3u_profile.replace_pattern)
+
+            # Check if transcoding is needed
+            stream_profile = channel.get_stream_profile()
+            if stream_profile.is_proxy() or stream_profile is None:
+                transcode = False
+            else:
+                transcode = True
+
+            stream_profile_id = stream_profile.id
+
+            return stream_url, stream_user_agent, transcode, stream_profile_id
+        except Exception as e:
+            logger.error(f"Error generating stream URL for channel {channel_id}: {e}")
+            channel.release_stream()
             return None, None, False, None
-
-        # Get the M3U account profile for URL pattern
-        m3u_profile = profile
-
-        # Get the appropriate user agent
-        m3u_account = M3UAccount.objects.get(id=m3u_profile.m3u_account.id)
-        stream_user_agent = m3u_account.get_user_agent().user_agent
-
-        if stream_user_agent is None:
-            stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
-            logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
-
-        # Generate stream URL based on the selected profile
-        input_url = stream.url
-        stream_url = transform_url(input_url, m3u_profile.search_pattern, m3u_profile.replace_pattern)
-
-        # Check if transcoding is needed
-        stream_profile = channel.get_stream_profile()
-        if stream_profile.is_proxy() or stream_profile is None:
-            transcode = False
-        else:
-            transcode = True
-
-        stream_profile_id = stream_profile.id
-
-        return stream_url, stream_user_agent, transcode, stream_profile_id
     except Exception as e:
         logger.error(f"Error generating stream URL: {e}")
         return None, None, False, None

apps/proxy/ts_proxy/views.py

@@ -54,6 +54,7 @@ def stream_ts(request, channel_id):
 
     client_user_agent = None
     proxy_server = ProxyServer.get_instance()
+    connection_allocated = False  # Track if connection slot was allocated via get_stream()
 
     try:
         # Generate a unique client ID
@@ -219,9 +220,9 @@ def stream_ts(request, channel_id):
                     )
 
             if stream_url is None:
-                # Release the channel's stream lock if one was acquired
+                # Release any connection slot that may have been allocated
-                # Note: Only call this if get_stream() actually assigned a stream
+                # by the error-checking get_stream() call during retries
-                # In our case, if stream_url is None, no stream was ever assigned, so don't release
+                channel.release_stream()
 
                 # Get the specific error message if available
                 wait_duration = f"{int(time.time() - wait_start_time)}s"
@@ -237,6 +238,11 @@ def stream_ts(request, channel_id):
                     {"error": error_msg, "waited": wait_duration}, status=503
                 )  # 503 Service Unavailable is appropriate here
 
+            # generate_stream_url() called get_stream() which allocated a connection
+            # slot (INCR'd profile_connections) - track this for cleanup on error
+            if needs_initialization:
+                connection_allocated = True
+
             # Get the stream ID from the channel
             stream_id, m3u_profile_id, _ = channel.get_stream()
             logger.info(
@@ -344,10 +350,16 @@ def stream_ts(request, channel_id):
             )
 
             if not success:
+                if connection_allocated:
+                    channel.release_stream()
+                    connection_allocated = False
                 return JsonResponse(
                     {"error": "Failed to initialize channel"}, status=500
                 )
 
+            # Channel initialized - cleanup lifecycle now owns the connection release
+            connection_allocated = False
+
             # If we're the owner, wait for connection to establish
             if proxy_server.am_i_owner(channel_id):
                 manager = proxy_server.stream_managers.get(channel_id)
@@ -507,6 +519,11 @@ def stream_ts(request, channel_id):
 
     except Exception as e:
         logger.error(f"Error in stream_ts: {e}", exc_info=True)
+        if connection_allocated:
+            try:
+                channel.release_stream()
+            except Exception:
+                pass
         return JsonResponse({"error": str(e)}, status=500)