Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-18 09:06:06 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

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

Browse source 
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)
This commit is contained in:
None 2026-02-05 23:12:30 -06:00
parent b8e1785d0e
commit 49b7b9e21b
2 changed files with 51 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

60
apps/proxy/ts_proxy/url_utils.py
View file

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

23
apps/proxy/ts_proxy/views.py
View file

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