diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py index 8b467b7f..9daaf4ef 100644 --- a/apps/proxy/ts_proxy/url_utils.py +++ b/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 diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 91f254a7..6980f957 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/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 - # Note: Only call this if get_stream() actually assigned a stream - # In our case, if stream_url is None, no stream was ever assigned, so don't release + # Release any connection slot that may have been allocated + # by the error-checking get_stream() call during retries + 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)