From 49b7b9e21b0dd896764d9d12abb61983bd7896e9 Mon Sep 17 00:00:00 2001 From: None Date: Thu, 5 Feb 2026 23:12:30 -0600 Subject: [PATCH 01/25] 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 | 60 +++++++++++++++++--------------- apps/proxy/ts_proxy/views.py | 23 ++++++++++-- 2 files changed, 51 insertions(+), 32 deletions(-) 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) From 16309cd9b455daf98fd1bc5af213c83531111ea9 Mon Sep 17 00:00:00 2001 From: None Date: Tue, 10 Feb 2026 18:15:48 -0600 Subject: [PATCH 02/25] Remove if logic from scheduled shutdown - stream_generator.py: _schedule_channel_shutdown_if_needed() already guards itself, the if logic created conflict between release_stream() and stop_channel(), so if the stream was already released the shutdown would not be scheduled --- apps/proxy/ts_proxy/stream_generator.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 50404f1d..771b0aa9 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -472,8 +472,7 @@ class StreamGenerator: logger.error(f"Could not log client disconnect event: {e}") # Schedule channel shutdown if no clients left - if not stream_released: # Only if we haven't already released the stream - self._schedule_channel_shutdown_if_needed(local_clients) + self._schedule_channel_shutdown_if_needed(local_clients) def _schedule_channel_shutdown_if_needed(self, local_clients): """ From 15d67c7f6518cd599823c57d070562da792f23d2 Mon Sep 17 00:00:00 2001 From: None Date: Tue, 10 Feb 2026 18:30:43 -0600 Subject: [PATCH 03/25] Atomize profile counter swap in update_stream_profile to prevent counter drift DECR/SET/INCR during stream profile switches are now executed in a Redis pipeline, preventing partial updates if an exception occurs mid-operation --- apps/channels/models.py | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/apps/channels/models.py b/apps/channels/models.py index ec5e135b..dd67095c 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -566,18 +566,18 @@ class Channel(models.Model): if current_profile_id == new_profile_id: return True - # Decrement connection count for old profile + # Use pipeline for atomic profile switch to prevent counter drift + # if an exception occurs between DECR and INCR old_profile_connections_key = f"profile_connections:{current_profile_id}" - old_count = int(redis_client.get(old_profile_connections_key) or 0) - if old_count > 0: - redis_client.decr(old_profile_connections_key) - - # Update the profile mapping - redis_client.set(f"stream_profile:{stream_id}", new_profile_id) - - # Increment connection count for new profile new_profile_connections_key = f"profile_connections:{new_profile_id}" - redis_client.incr(new_profile_connections_key) + old_count = int(redis_client.get(old_profile_connections_key) or 0) + + pipe = redis_client.pipeline() + if old_count > 0: + pipe.decr(old_profile_connections_key) + pipe.set(f"stream_profile:{stream_id}", new_profile_id) + pipe.incr(new_profile_connections_key) + pipe.execute() logger.info( f"Updated stream {stream_id} profile from {current_profile_id} to {new_profile_id}" ) From 9e8f227f494eb0d1aa27b1359d8b470221a56e3a Mon Sep 17 00:00:00 2001 From: None Date: Tue, 10 Feb 2026 18:58:17 -0600 Subject: [PATCH 04/25] Fix: VOD proxy double-DECR of profile connection counter during cleanup - multi_worker_connection_manager.py: Error handler at line 1042: removed connection_manager=self from cleanup() call since the DECR was already done at line 1036. Prevents double-DECR. - connection_manager.py: Added connection_key_exists check before calling remove_connection(). The fallback DECR from session data now only fires when the connection tracking key has already expired. Prevents double-DECR. --- apps/proxy/vod_proxy/connection_manager.py | 22 +++++++++++++------ .../multi_worker_connection_manager.py | 3 ++- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/apps/proxy/vod_proxy/connection_manager.py b/apps/proxy/vod_proxy/connection_manager.py index aafc75cd..42d69a11 100644 --- a/apps/proxy/vod_proxy/connection_manager.py +++ b/apps/proxy/vod_proxy/connection_manager.py @@ -1363,17 +1363,25 @@ class VODConnectionManager: client_id = session_id # Remove individual connection tracking keys created during streaming + # Check if connection key exists first - remove_connection() + # handles the profile DECR when the key exists, but skips it + # if the key already expired by TTL + connection_key_exists = False if content_type and content_uuid: + connection_key = self._get_connection_key(content_type, content_uuid, client_id) + connection_key_exists = bool(self.redis_client.exists(connection_key)) logger.info(f"[{session_id}] Cleaning up connection tracking keys") self.remove_connection(content_type, content_uuid, client_id) - # Remove from profile connections if counted (additional safety check) - if session_data.get(b'connection_counted') == b'True' and profile_id: - profile_key = self._get_profile_connections_key(int(profile_id.decode('utf-8'))) - current_count = int(self.redis_client.get(profile_key) or 0) - if current_count > 0: - self.redis_client.decr(profile_key) - logger.info(f"[{session_id}] Decremented profile {profile_id.decode('utf-8')} connections") + # Fallback DECR: only if the connection tracking key had already + # expired (remove_connection skips DECR in that case) + if not connection_key_exists: + if session_data.get(b'connection_counted') == b'True' and profile_id: + profile_key = self._get_profile_connections_key(int(profile_id.decode('utf-8'))) + current_count = int(self.redis_client.get(profile_key) or 0) + if current_count > 0: + self.redis_client.decr(profile_key) + logger.info(f"[{session_id}] Decremented profile {profile_id.decode('utf-8')} connections (fallback)") # Remove session tracking key self.redis_client.delete(session_key) diff --git a/apps/proxy/vod_proxy/multi_worker_connection_manager.py b/apps/proxy/vod_proxy/multi_worker_connection_manager.py index 1534f761..58d6bd04 100644 --- a/apps/proxy/vod_proxy/multi_worker_connection_manager.py +++ b/apps/proxy/vod_proxy/multi_worker_connection_manager.py @@ -1036,9 +1036,10 @@ class MultiWorkerVODConnectionManager: self._decrement_profile_connections(m3u_profile.id) # Also clean up the Redis connection state since we won't be using it + # Pass connection_manager=None since we already decremented above if redis_connection: try: - redis_connection.cleanup(connection_manager=self, current_worker_id=self.worker_id) + redis_connection.cleanup(current_worker_id=self.worker_id) except Exception as cleanup_error: logger.error(f"[{client_id}] Error during cleanup after connection failure: {cleanup_error}") From cf96287c90a32120ad22aa7a17973ddae96a627a Mon Sep 17 00:00:00 2001 From: None Date: Tue, 10 Feb 2026 19:06:51 -0600 Subject: [PATCH 05/25] Add guard to remove_connection() - multi_worker_connection_manager.py: Add missing count > 0 guard to multi-worker remove_connection() This method doesn't appear to be called anywhere but added the guard in case it is put back into service --- apps/proxy/vod_proxy/multi_worker_connection_manager.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/proxy/vod_proxy/multi_worker_connection_manager.py b/apps/proxy/vod_proxy/multi_worker_connection_manager.py index 58d6bd04..63eeae00 100644 --- a/apps/proxy/vod_proxy/multi_worker_connection_manager.py +++ b/apps/proxy/vod_proxy/multi_worker_connection_manager.py @@ -1263,12 +1263,14 @@ class MultiWorkerVODConnectionManager: profile_id = connection_data.get('m3u_profile_id') if profile_id: profile_connections_key = f"profile_connections:{profile_id}" + current_count = int(self.redis_client.get(profile_connections_key) or 0) # Use pipeline for atomic operations pipe = self.redis_client.pipeline() pipe.delete(connection_key) pipe.srem(content_connections_key, client_id) - pipe.decr(profile_connections_key) + if current_count > 0: + pipe.decr(profile_connections_key) pipe.execute() logger.info(f"Removed Redis-backed connection: {client_id}") From 5a4b51b60a90d64d11a56df273607cc0be6ac72d Mon Sep 17 00:00:00 2001 From: None Date: Wed, 11 Feb 2026 18:08:24 -0600 Subject: [PATCH 06/25] Fix TOCTOU race condition in Channel.get_stream() with atomic slot reservation Add _check_and_reserve_profile_slot() method that uses an INCR-first-then-check pattern to atomically reserve connection slots. This eliminates the TOCTOU race condition where separate GET > check > INCR operations allowed concurrent requests to both pass the capacity check, exceeding max_streams. For profiles with max_streams=0 (unlimited), INCR is skipped entirely. If over capacity, the increment is rolled back with DECR. Based on PR #838 by patchy8736, adapted from Lua script to native Redis commands for consistency with project style and easier debugging. --- apps/channels/models.py | 53 ++++++++++++++++++++++++++++++----------- 1 file changed, 39 insertions(+), 14 deletions(-) diff --git a/apps/channels/models.py b/apps/channels/models.py index dd67095c..50538238 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -391,6 +391,38 @@ class Channel(models.Model): return stream_profile + def _check_and_reserve_profile_slot(self, profile, redis_client): + """ + Atomically check and reserve a connection slot for the given profile. + + Uses an INCR-first-then-check pattern to eliminate the TOCTOU race + condition where separate GET + check + INCR operations could allow + concurrent requests to both pass the capacity check. + + For profiles with max_streams=0 (unlimited), no reservation is needed. + + Args: + profile: M3UAccountProfile instance + redis_client: Redis client instance + + Returns: + tuple: (reserved: bool, current_count: int) + """ + if profile.max_streams == 0: + return (True, 0) + + profile_connections_key = f"profile_connections:{profile.id}" + + # Atomically increment first — this is a single Redis command + new_count = redis_client.incr(profile_connections_key) + + if new_count <= profile.max_streams: + return (True, new_count) + + # Over capacity — roll back the increment + redis_client.decr(profile_connections_key) + return (False, new_count - 1) + def get_stream(self): """ Finds an available stream for the requested channel and returns the selected stream and profile. @@ -456,24 +488,16 @@ class Channel(models.Model): for profile in profiles: has_active_profiles = True - profile_connections_key = f"profile_connections:{profile.id}" - current_connections = int( - redis_client.get(profile_connections_key) or 0 + # Atomically check and reserve a slot (INCR-first pattern) + reserved, current_count = self._check_and_reserve_profile_slot( + profile, redis_client ) - # Check if profile has available slots (or unlimited connections) - if ( - profile.max_streams == 0 - or current_connections < profile.max_streams - ): - # Start a new stream + if reserved: + # Slot reserved — assign stream to this channel redis_client.set(f"channel_stream:{self.id}", stream.id) redis_client.set(f"stream_profile:{stream.id}", profile.id) - # Increment connection count for profiles with limits - if profile.max_streams > 0: - redis_client.incr(profile_connections_key) - return ( stream.id, profile.id, @@ -483,7 +507,8 @@ class Channel(models.Model): # This profile is at max connections has_streams_but_maxed_out = True logger.debug( - f"Profile {profile.id} at max connections: {current_connections}/{profile.max_streams}" + f"Profile {profile.id} at max connections: " + f"{current_count}/{profile.max_streams}" ) # No available streams - determine specific reason From 33f68a9808a1a29eed16afb6d85ab8b04ff58c85 Mon Sep 17 00:00:00 2001 From: None Date: Wed, 11 Feb 2026 19:11:28 -0600 Subject: [PATCH 07/25] Remove unused Stream.get_stream() with TOCTOU vulnerability Remove the Stream.get_stream() method which had the same TOCTOU race condition as Channel.get_stream() All stream acquisition goes through Channel.get_stream(), making this dead code. Removing it eliminates a potential source of connection capacity leaks Based on PR #838 by patchy8736 --- apps/channels/models.py | 48 ----------------------------------------- 1 file changed, 48 deletions(-) diff --git a/apps/channels/models.py b/apps/channels/models.py index 50538238..998b7aca 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -202,54 +202,6 @@ class Stream(models.Model): return stream_profile - def get_stream(self): - """ - Finds an available stream for the requested channel and returns the selected stream and profile. - """ - redis_client = RedisClient.get_client() - profile_id = redis_client.get(f"stream_profile:{self.id}") - if profile_id: - profile_id = int(profile_id) - return self.id, profile_id, None - - # Retrieve the M3U account associated with the stream. - m3u_account = self.m3u_account - m3u_profiles = m3u_account.profiles.all() - default_profile = next((obj for obj in m3u_profiles if obj.is_default), None) - profiles = [default_profile] + [ - obj for obj in m3u_profiles if not obj.is_default - ] - - for profile in profiles: - logger.info(profile) - # Skip inactive profiles - if profile.is_active == False: - continue - - profile_connections_key = f"profile_connections:{profile.id}" - current_connections = int(redis_client.get(profile_connections_key) or 0) - - # Check if profile has available slots (or unlimited connections) - if profile.max_streams == 0 or current_connections < profile.max_streams: - # Start a new stream - redis_client.set(f"channel_stream:{self.id}", self.id) - redis_client.set( - f"stream_profile:{self.id}", profile.id - ) # Store only the matched profile - - # Increment connection count for profiles with limits - if profile.max_streams > 0: - redis_client.incr(profile_connections_key) - - return ( - self.id, - profile.id, - None, - ) # Return newly assigned stream and matched profile - - # 4. No available streams - return None, None, None - def release_stream(self): """ Called when a stream is finished to release the lock. From 56719d93fe53445226a8c2dd812e9a23287f2ced Mon Sep 17 00:00:00 2001 From: None Date: Wed, 11 Feb 2026 19:16:01 -0600 Subject: [PATCH 08/25] Add boolean return and metadata fallback to Channel.release_stream() Channel.release_stream() now returns True on success and False when no stream/profile info could be found for cleanup. This allows callers to detect and handle failed releases. When the primary Redis keys (channel_stream/stream_profile) are already cleaned up by the proxy, the method falls back to the channel metadata hash (ts_proxy:channel:{uuid}:metadata) to recover stream_id and profile_id for proper connection counter cleanup. Uses str(self.uuid) for metadata key lookup - fixes a bug in PR #838 which used self.id (integer PK), but metadata is keyed by UUID. Enhanced logging now includes channel UUID for better traceability. Based on PR #838 by patchy8736 --- apps/channels/models.py | 58 ++++++++++++++++++++++++++++++++++++----- 1 file changed, 52 insertions(+), 6 deletions(-) diff --git a/apps/channels/models.py b/apps/channels/models.py index 998b7aca..9e466cc7 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -3,6 +3,8 @@ from django.core.exceptions import ValidationError from django.conf import settings from core.models import StreamProfile, CoreSettings from core.utils import RedisClient +from apps.proxy.ts_proxy.redis_keys import RedisKeys +from apps.proxy.ts_proxy.constants import ChannelMetadataField import logging import uuid from datetime import datetime @@ -476,32 +478,74 @@ class Channel(models.Model): def release_stream(self): """ Called when a stream is finished to release the lock. + + Returns: + bool: True if stream was successfully released, False if + no stream/profile info could be found for cleanup. """ redis_client = RedisClient.get_client() stream_id = redis_client.get(f"channel_stream:{self.id}") if not stream_id: - logger.debug("Invalid stream ID pulled from channel index") - return + # Primary key missing — try metadata hash fallback. + # The proxy may have already cleaned up channel_stream/stream_profile + # keys, but the metadata hash can still have the stream_id and profile. + metadata_key = RedisKeys.channel_metadata(str(self.uuid)) + meta_stream_id = redis_client.hget( + metadata_key, ChannelMetadataField.STREAM_ID + ) + meta_profile_id = redis_client.hget( + metadata_key, ChannelMetadataField.M3U_PROFILE + ) + + if meta_stream_id and meta_profile_id: + stream_id = int(meta_stream_id) + profile_id = int(meta_profile_id) + logger.debug( + f"Channel {self.uuid}: recovered stream_id={stream_id}, " + f"profile_id={profile_id} from metadata fallback" + ) + # Clean up any remaining keys + redis_client.delete(f"channel_stream:{self.id}") + redis_client.delete(f"stream_profile:{stream_id}") + + profile_connections_key = f"profile_connections:{profile_id}" + current_count = int( + redis_client.get(profile_connections_key) or 0 + ) + if current_count > 0: + redis_client.decr(profile_connections_key) + return True + + logger.debug( + f"Channel {self.uuid}: no stream info found in primary keys " + f"or metadata fallback" + ) + return False redis_client.delete(f"channel_stream:{self.id}") # Remove active stream stream_id = int(stream_id) logger.debug( - f"Found stream ID {stream_id} associated with channel stream {self.id}" + f"Channel {self.uuid}: found stream_id={stream_id} for " + f"channel_stream:{self.id}" ) # Get the matched profile for cleanup profile_id = redis_client.get(f"stream_profile:{stream_id}") if not profile_id: - logger.debug("Invalid profile ID pulled from stream index") - return + logger.debug( + f"Channel {self.uuid}: no profile found for " + f"stream_profile:{stream_id}" + ) + return False redis_client.delete(f"stream_profile:{stream_id}") # Remove profile association profile_id = int(profile_id) logger.debug( - f"Found profile ID {profile_id} associated with stream {stream_id}" + f"Channel {self.uuid}: found profile_id={profile_id} for " + f"stream {stream_id}" ) profile_connections_key = f"profile_connections:{profile_id}" @@ -511,6 +555,8 @@ class Channel(models.Model): if current_count > 0: redis_client.decr(profile_connections_key) + return True + def update_stream_profile(self, new_profile_id): """ Updates the profile for the current stream and adjusts connection counts. From 6826e94921b0d7fa49a032d161059c014ce2ddde Mon Sep 17 00:00:00 2001 From: None Date: Wed, 11 Feb 2026 19:27:45 -0600 Subject: [PATCH 09/25] Add boolean return to Stream.release_stream() Stream.release_stream() now returns True on success and False when no profile info could be found in stream_profile:{stream_id}. This allows callers to detect and handle failed releases. Enhanced logging to include stream ID for better traceability. Based on PR #838 by patchy8736 --- apps/channels/models.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/apps/channels/models.py b/apps/channels/models.py index 9e466cc7..b6cb6ab0 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -207,6 +207,10 @@ class Stream(models.Model): def release_stream(self): """ Called when a stream is finished to release the lock. + + Returns: + bool: True if stream was successfully released, False if + no profile info could be found for cleanup. """ redis_client = RedisClient.get_client() @@ -214,14 +218,17 @@ class Stream(models.Model): # Get the matched profile for cleanup profile_id = redis_client.get(f"stream_profile:{stream_id}") if not profile_id: - logger.debug("Invalid profile ID pulled from stream index") - return + logger.debug( + f"Stream {stream_id}: no profile found in " + f"stream_profile:{stream_id}" + ) + return False redis_client.delete(f"stream_profile:{stream_id}") # Remove profile association profile_id = int(profile_id) logger.debug( - f"Found profile ID {profile_id} associated with stream {stream_id}" + f"Stream {stream_id}: found profile_id={profile_id}" ) profile_connections_key = f"profile_connections:{profile_id}" @@ -231,6 +238,8 @@ class Stream(models.Model): if current_count > 0: redis_client.decr(profile_connections_key) + return True + class ChannelManager(models.Manager): def active(self): From fe17d0d5850c0b58e2cea150d97202e8e713afdb Mon Sep 17 00:00:00 2001 From: None Date: Wed, 11 Feb 2026 19:49:11 -0600 Subject: [PATCH 10/25] Check release_stream() boolean return at all call sites All 12 call sites for Channel.release_stream() and Stream.release_stream() now check the boolean return value and log warnings when release fails. Files updated: - server.py: 4 call sites (zombie cleanup + _clean_redis_keys) - channel_service.py: 2 call sites (stop_channel) - stream_generator.py: 1 call site (last client disconnect) - views.py: 4 call sites (failed init, redirect, init failure, exception) - url_utils.py: 1 call site (URL generation error) Also fixed bare except clause in _clean_redis_keys and uncaught Stream.DoesNotExist in channel_service.stop_channel. Based on PR #838 by patchy8736 --- apps/proxy/ts_proxy/server.py | 25 +++++++++++++------ .../ts_proxy/services/channel_service.py | 22 ++++++++++------ apps/proxy/ts_proxy/stream_generator.py | 8 +++--- apps/proxy/ts_proxy/url_utils.py | 3 ++- apps/proxy/ts_proxy/views.py | 12 ++++++--- 5 files changed, 47 insertions(+), 23 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index df51744d..feef4957 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -788,13 +788,17 @@ class ProxyServer: # Force release resources in the Channel model try: channel = Channel.objects.get(uuid=channel_id) - channel.release_stream() - logger.info(f"Released stream allocation for zombie channel {channel_id}") + if not channel.release_stream(): + logger.warning(f"Failed to release stream for zombie channel {channel_id}") + else: + logger.info(f"Released stream allocation for zombie channel {channel_id}") except Exception as e: try: stream = Stream.objects.get(stream_hash=channel_id) - stream.release_stream() - logger.info(f"Released stream allocation for zombie channel {channel_id}") + if not stream.release_stream(): + logger.warning(f"Failed to release stream for zombie channel {channel_id}") + else: + logger.info(f"Released stream allocation for zombie channel {channel_id}") except Exception as e: logger.error(f"Error releasing stream for zombie channel {channel_id}: {e}") @@ -1339,10 +1343,15 @@ class ProxyServer: # Release the channel, stream, and profile keys from the channel try: channel = Channel.objects.get(uuid=channel_id) - channel.release_stream() - except: - stream = Stream.objects.get(stream_hash=channel_id) - stream.release_stream() + if not channel.release_stream(): + logger.debug(f"Channel {channel_id}: release_stream found no keys to clean") + except Channel.DoesNotExist: + try: + stream = Stream.objects.get(stream_hash=channel_id) + if not stream.release_stream(): + logger.debug(f"Stream {channel_id}: release_stream found no keys to clean") + except Stream.DoesNotExist: + logger.debug(f"No Channel or Stream found for {channel_id}") if not self.redis_client: return 0 diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index 4c4a73ac..efe07c7e 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -265,15 +265,23 @@ class ChannelService: # Release the channel in the channel model if applicable try: channel = Channel.objects.get(uuid=channel_id) - channel.release_stream() - logger.info(f"Released channel {channel_id} stream allocation") - model_released = True + model_released = channel.release_stream() + if model_released: + logger.info(f"Released channel {channel_id} stream allocation") + else: + logger.warning(f"Channel {channel_id}: release_stream found no keys to clean") except Channel.DoesNotExist: logger.warning(f"Could not find Channel model for UUID {channel_id}, attempting stream hash") - stream = Stream.objects.get(stream_hash=channel_id) - stream.release_stream() - logger.info(f"Released stream {channel_id} stream allocation") - model_released = True + try: + stream = Stream.objects.get(stream_hash=channel_id) + model_released = stream.release_stream() + if model_released: + logger.info(f"Released stream {channel_id} stream allocation") + else: + logger.warning(f"Stream {channel_id}: release_stream found no keys to clean") + except Stream.DoesNotExist: + logger.error(f"No Channel or Stream found for {channel_id}") + model_released = False except Exception as e: logger.error(f"Error releasing channel stream: {e}") model_released = False diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 771b0aa9..3c2a66fe 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -441,9 +441,11 @@ class StreamGenerator: try: # Get the channel by UUID channel = Channel.objects.get(uuid=self.channel_id) - channel.release_stream() - stream_released = True - logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}") + stream_released = channel.release_stream() + if stream_released: + logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}") + else: + logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}") except Exception as e: logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}") except Exception as e: diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py index 9daaf4ef..fe06ad48 100644 --- a/apps/proxy/ts_proxy/url_utils.py +++ b/apps/proxy/ts_proxy/url_utils.py @@ -162,7 +162,8 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]] 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() + if not channel.release_stream(): + logger.warning(f"Failed to release stream for channel {channel_id} after URL generation error") return None, None, False, None except Exception as e: logger.error(f"Error generating stream URL: {e}") diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 6980f957..5060a3a7 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -222,7 +222,8 @@ def stream_ts(request, channel_id): if stream_url is None: # Release any connection slot that may have been allocated # by the error-checking get_stream() call during retries - channel.release_stream() + if not channel.release_stream(): + logger.debug(f"[{client_id}] release_stream found no keys during failed init cleanup") # Get the specific error message if available wait_duration = f"{int(time.time() - wait_start_time)}s" @@ -314,7 +315,8 @@ def stream_ts(request, channel_id): f"[{client_id}] Alternate stream #{alt['stream_id']} failed validation: {message}" ) # Release stream lock before redirecting - channel.release_stream() + if not channel.release_stream(): + logger.warning(f"[{client_id}] Failed to release stream before redirect") # Final decision based on validation results if is_valid: logger.info( @@ -351,7 +353,8 @@ def stream_ts(request, channel_id): if not success: if connection_allocated: - channel.release_stream() + if not channel.release_stream(): + logger.warning(f"[{client_id}] Failed to release stream after init failure") connection_allocated = False return JsonResponse( {"error": "Failed to initialize channel"}, status=500 @@ -521,7 +524,8 @@ def stream_ts(request, channel_id): logger.error(f"Error in stream_ts: {e}", exc_info=True) if connection_allocated: try: - channel.release_stream() + if not channel.release_stream(): + logger.warning(f"[{client_id}] Failed to release stream in exception handler") except Exception: pass return JsonResponse({"error": str(e)}, status=500) From 6360271e2830831fbfca9bf5c938e29dd25d7ef5 Mon Sep 17 00:00:00 2001 From: None Date: Wed, 11 Feb 2026 20:09:17 -0600 Subject: [PATCH 11/25] Fix unprotected _clean_redis_keys() call in check_if_channel_exists Wrap _clean_redis_keys() call in check_if_channel_exists() with try/except to prevent exception propagation when Channel/Stream lookup or release_stream() encounters DB or Redis errors. --- apps/proxy/ts_proxy/server.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index feef4957..3a6e8ae3 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -766,7 +766,10 @@ class ProxyServer: if self.redis_client.exists(key): # Found orphaned keys without metadata - clean them up logger.warning(f"Found orphaned keys for channel {channel_id} without metadata - cleaning up") - self._clean_redis_keys(channel_id) + try: + self._clean_redis_keys(channel_id) + except Exception as e: + logger.error(f"Error cleaning redis keys for channel {channel_id}: {e}") return False return False From 8bb8a0fca904ccfc3c554ac6702e88ab94d3f094 Mon Sep 17 00:00:00 2001 From: None Date: Wed, 11 Feb 2026 20:52:30 -0600 Subject: [PATCH 12/25] fix(ts-proxy): Remove redundant local import causing UnboundLocalError The local `from apps.channels.models import Channel` inside a conditional block in _handle_client_disconnect() shadowed the top-level import. When the condition was False, Python's function-level scoping caused line 462's Channel.objects.get() to raise UnboundLocalError. Channel is already imported at module level (line 11), making the local import unnecessary. Discovered when testing proxy changes --- apps/proxy/ts_proxy/stream_generator.py | 1 - 1 file changed, 1 deletion(-) diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 3c2a66fe..875d9d0e 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -437,7 +437,6 @@ class StreamGenerator: client_count = proxy_server.client_managers[self.channel_id].get_total_client_count() # Only the last client or owner should release the stream if client_count <= 1 and proxy_server.am_i_owner(self.channel_id): - from apps.channels.models import Channel try: # Get the channel by UUID channel = Channel.objects.get(uuid=self.channel_id) From 837ba847685bb93f6d9351720e68c3ca19c6d04e Mon Sep 17 00:00:00 2001 From: None Date: Wed, 11 Feb 2026 23:57:10 -0600 Subject: [PATCH 13/25] fix(ts-proxy): Restore Stream.get_stream() with atomic slot reservation Stream.get_stream() was incorrectly removed as dead code in commit 33f68a98. It is actually called get_stream_object() in views.py when a stream hash (not channel UUID) is used, such as during stream preview. The variable is named channel but holds a Stream instance, which masked the dependency during analysis. Restored with the INCR-first-then-check pattern applied, fixing the same TOCTOU vulnerability the original method had. --- apps/channels/models.py | 46 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/apps/channels/models.py b/apps/channels/models.py index b6cb6ab0..b9c1dde1 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -204,6 +204,52 @@ class Stream(models.Model): return stream_profile + def get_stream(self): + """ + Finds an available profile for this stream and reserves a connection slot. + + Returns: + Tuple[Optional[int], Optional[int], Optional[str]]: (stream_id, profile_id, error_reason) + """ + redis_client = RedisClient.get_client() + profile_id = redis_client.get(f"stream_profile:{self.id}") + if profile_id: + profile_id = int(profile_id) + return self.id, profile_id, None + + # Retrieve the M3U account associated with the stream. + m3u_account = self.m3u_account + m3u_profiles = m3u_account.profiles.all() + default_profile = next((obj for obj in m3u_profiles if obj.is_default), None) + profiles = [default_profile] + [ + obj for obj in m3u_profiles if not obj.is_default + ] + + for profile in profiles: + logger.info(profile) + # Skip inactive profiles + if profile.is_active == False: + continue + + # Atomic slot reservation: INCR first, check, rollback if over capacity + if profile.max_streams == 0: + reserved = True + else: + profile_connections_key = f"profile_connections:{profile.id}" + new_count = redis_client.incr(profile_connections_key) + if new_count <= profile.max_streams: + reserved = True + else: + redis_client.decr(profile_connections_key) + reserved = False + + if reserved: + redis_client.set(f"channel_stream:{self.id}", self.id) + redis_client.set(f"stream_profile:{self.id}", profile.id) + return self.id, profile.id, None + + return None, None, "All active M3U profiles have reached maximum connection limits" + def release_stream(self): """ Called when a stream is finished to release the lock. From 20ce45f7adfd7a0f959207f09125accec2e93806 Mon Sep 17 00:00:00 2001 From: None Date: Thu, 12 Feb 2026 00:02:40 -0600 Subject: [PATCH 14/25] fix(ts-proxy): Add Stream fallback in cleanup path to prevent counter leak on stream preview When previewing a stream directly (via stream hash), the cleanup in stream_generator.py tried Channel.objects.get(uuid=stream_hash) which failed with ValidationError since the hash is not a UUID. This prevented release_stream() from running, permanently leaking the connection counter. With max_streams=1, a single preview permanently blocked all subsequent streams. Added Stream.objects.get(stream_hash=...) fallback so the cleanup path works for both Channel (UUID) and Stream (hash) identifiers. Also added Stream import to stream_generator.py. --- apps/proxy/ts_proxy/stream_generator.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 875d9d0e..f43eb528 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -8,7 +8,7 @@ import logging import threading import gevent # Add this import at the top of your file from apps.proxy.config import TSConfig as Config -from apps.channels.models import Channel +from apps.channels.models import Channel, Stream from core.utils import log_system_event from .server import ProxyServer from .utils import create_ts_packet, get_logger @@ -438,9 +438,12 @@ class StreamGenerator: # Only the last client or owner should release the stream if client_count <= 1 and proxy_server.am_i_owner(self.channel_id): try: - # Get the channel by UUID - channel = Channel.objects.get(uuid=self.channel_id) - stream_released = channel.release_stream() + # Try Channel first (normal flow), fall back to Stream (preview flow) + try: + obj = Channel.objects.get(uuid=self.channel_id) + except (Channel.DoesNotExist, Exception): + obj = Stream.objects.get(stream_hash=self.channel_id) + stream_released = obj.release_stream() if stream_released: logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}") else: From 51b58247378a63de9adfa0dd075adb85f8cb5e44 Mon Sep 17 00:00:00 2001 From: None Date: Thu, 12 Feb 2026 00:09:29 -0600 Subject: [PATCH 15/25] Fix cleanup paths for stream preview Fix exception handling in _clean_redis_keys() and stop_channel() where `except Channel.DoesNotExist` didn't catch ValidationError from invalid UUID format, bypassing the Stream fallback path. --- apps/proxy/ts_proxy/server.py | 4 ++-- apps/proxy/ts_proxy/services/channel_service.py | 9 +++------ 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 3a6e8ae3..d036bc2e 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -1348,12 +1348,12 @@ class ProxyServer: channel = Channel.objects.get(uuid=channel_id) if not channel.release_stream(): logger.debug(f"Channel {channel_id}: release_stream found no keys to clean") - except Channel.DoesNotExist: + except (Channel.DoesNotExist, Exception): try: stream = Stream.objects.get(stream_hash=channel_id) if not stream.release_stream(): logger.debug(f"Stream {channel_id}: release_stream found no keys to clean") - except Stream.DoesNotExist: + except (Stream.DoesNotExist, Exception): logger.debug(f"No Channel or Stream found for {channel_id}") if not self.redis_client: diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index efe07c7e..e765ece3 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -270,7 +270,7 @@ class ChannelService: logger.info(f"Released channel {channel_id} stream allocation") else: logger.warning(f"Channel {channel_id}: release_stream found no keys to clean") - except Channel.DoesNotExist: + except (Channel.DoesNotExist, Exception): logger.warning(f"Could not find Channel model for UUID {channel_id}, attempting stream hash") try: stream = Stream.objects.get(stream_hash=channel_id) @@ -279,12 +279,9 @@ class ChannelService: logger.info(f"Released stream {channel_id} stream allocation") else: logger.warning(f"Stream {channel_id}: release_stream found no keys to clean") - except Stream.DoesNotExist: - logger.error(f"No Channel or Stream found for {channel_id}") + except (Stream.DoesNotExist, Exception) as e: + logger.error(f"No Channel or Stream found for {channel_id}: {e}") model_released = False - except Exception as e: - logger.error(f"Error releasing channel stream: {e}") - model_released = False return { 'status': 'success', From c69718b55987362403d468913dbe6cd7dd040954 Mon Sep 17 00:00:00 2001 From: None Date: Tue, 17 Feb 2026 21:18:20 -0600 Subject: [PATCH 16/25] Fix: Make release_stream() idempotent to prevent double-DECR on rapid channel switching Multiple code paths call release_stream() during channel shutdown (stream generator cleanup, _clean_redis_keys, ChannelService.stop_channel), causing the profile_connections counter to be decremented multiple times for the same stream. The metadata hash fallback persisted longer than the primary keys, allowing subsequent calls to find stale stream_id/m3u_profile and DECR again. Changes: - Clear STREAM_ID and M3U_PROFILE metadata fields after decrementing in both the primary and fallback paths, making release_stream() safe to call multiple times - Add metadata fallback when stream_profile:{id} key is missing but channel_stream:{id} exists, preventing a counter leak in that edge case - Replace redundant get_stream() call in views.py with direct Redis reads to avoid side-effect INCR from a method that can allocate connection slots --- apps/channels/models.py | 47 +++++++++++++++++++++++++++++------- apps/proxy/ts_proxy/views.py | 14 +++++++++-- 2 files changed, 50 insertions(+), 11 deletions(-) diff --git a/apps/channels/models.py b/apps/channels/models.py index b9c1dde1..217a92c0 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -564,6 +564,14 @@ class Channel(models.Model): redis_client.delete(f"channel_stream:{self.id}") redis_client.delete(f"stream_profile:{stream_id}") + # Clear metadata fields so duplicate release_stream() calls + # won't find them and DECR again + redis_client.hdel( + metadata_key, + ChannelMetadataField.STREAM_ID, + ChannelMetadataField.M3U_PROFILE, + ) + profile_connections_key = f"profile_connections:{profile_id}" current_count = int( redis_client.get(profile_connections_key) or 0 @@ -588,16 +596,27 @@ class Channel(models.Model): # Get the matched profile for cleanup profile_id = redis_client.get(f"stream_profile:{stream_id}") - if not profile_id: - logger.debug( - f"Channel {self.uuid}: no profile found for " - f"stream_profile:{stream_id}" + if profile_id: + redis_client.delete(f"stream_profile:{stream_id}") # Remove profile association + profile_id = int(profile_id) + else: + # stream_profile key missing — try metadata hash fallback + metadata_key = RedisKeys.channel_metadata(str(self.uuid)) + meta_profile_id = redis_client.hget( + metadata_key, ChannelMetadataField.M3U_PROFILE ) - return False - - redis_client.delete(f"stream_profile:{stream_id}") # Remove profile association - - profile_id = int(profile_id) + if meta_profile_id: + profile_id = int(meta_profile_id) + logger.debug( + f"Channel {self.uuid}: recovered profile_id={profile_id} " + f"from metadata fallback (stream_profile:{stream_id} was missing)" + ) + else: + logger.warning( + f"Channel {self.uuid}: no profile found for " + f"stream_profile:{stream_id} or in metadata fallback" + ) + return False logger.debug( f"Channel {self.uuid}: found profile_id={profile_id} for " f"stream {stream_id}" @@ -610,6 +629,16 @@ class Channel(models.Model): if current_count > 0: redis_client.decr(profile_connections_key) + # Clear metadata fields so duplicate release_stream() calls + # (e.g. from _clean_redis_keys or ChannelService.stop_channel) + # won't find them via fallback and DECR again + metadata_key = RedisKeys.channel_metadata(str(self.uuid)) + redis_client.hdel( + metadata_key, + ChannelMetadataField.STREAM_ID, + ChannelMetadataField.M3U_PROFILE, + ) + return True def update_stream_profile(self, new_profile_id): diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 5060a3a7..90ac2673 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -244,8 +244,18 @@ def stream_ts(request, channel_id): if needs_initialization: connection_allocated = True - # Get the stream ID from the channel - stream_id, m3u_profile_id, _ = channel.get_stream() + # Read stream assignment from Redis (already set by generate_stream_url → get_stream). + # Avoid calling get_stream() again — (INCR profile counter) + # It could double-allocate if the keys were cleared by a concurrent release. + stream_id = None + m3u_profile_id = None + if proxy_server.redis_client: + stream_id_bytes = proxy_server.redis_client.get(f"channel_stream:{channel.id}") + if stream_id_bytes: + stream_id = int(stream_id_bytes) + profile_id_bytes = proxy_server.redis_client.get(f"stream_profile:{stream_id}") + if profile_id_bytes: + m3u_profile_id = int(profile_id_bytes) logger.info( f"Channel {channel_id} using stream ID {stream_id}, m3u account profile ID {m3u_profile_id}" ) From 9b6c793b1186da86c2860f74006ee8b42d23ad1c Mon Sep 17 00:00:00 2001 From: Endoze Date: Thu, 26 Feb 2026 09:41:57 -0500 Subject: [PATCH 17/25] fix(wsgi): ensure gevent monkey-patching before Django import chain Celery's Redis connection pool was silently failing in modular deployment mode because Django's setup imports celery.py before uWSGI's gevent plugin reliably patches stdlib sockets. Moving patch_all() to the top of wsgi.py guarantees all broker connections are gevent-aware regardless of uWSGI initialization order. --- dispatcharr/tests/__init__.py | 0 dispatcharr/tests/test_wsgi.py | 39 ++++++++++++++++++++++++++++++++++ dispatcharr/wsgi.py | 15 +++++++++++++ 3 files changed, 54 insertions(+) create mode 100644 dispatcharr/tests/__init__.py create mode 100644 dispatcharr/tests/test_wsgi.py diff --git a/dispatcharr/tests/__init__.py b/dispatcharr/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/dispatcharr/tests/test_wsgi.py b/dispatcharr/tests/test_wsgi.py new file mode 100644 index 00000000..cdf7a2e9 --- /dev/null +++ b/dispatcharr/tests/test_wsgi.py @@ -0,0 +1,39 @@ +"""Tests for dispatcharr/wsgi.py gevent monkey-patching.""" + +import os +import subprocess +import sys +import unittest + + +class TestWSGIGeventPatching(unittest.TestCase): + """Verify that wsgi.py applies gevent monkey-patching before Django imports. + + These tests run in subprocesses because monkey.patch_all() must execute + before any other imports. By the time Django's test runner loads this + file, ssl/socket are already imported, so calling patch_all() in-process + would fail. A subprocess mirrors how uWSGI actually loads wsgi.py. + """ + + def test_socket_is_patched_after_wsgi_import(self): + """Loading wsgi.py first (as uWSGI does) should patch sockets.""" + result = subprocess.run( + [ + sys.executable, "-c", + "import dispatcharr.wsgi; " + "from gevent import monkey; " + "assert monkey.is_module_patched('socket'), " + "'socket module was not patched by wsgi.py'; " + "print('PASS')", + ], + capture_output=True, + text=True, + env={**os.environ, "DJANGO_SETTINGS_MODULE": "dispatcharr.settings"}, + timeout=60, + cwd="/app", + ) + self.assertEqual( + result.returncode, + 0, + f"monkey-patching check failed:\n{result.stderr}", + ) diff --git a/dispatcharr/wsgi.py b/dispatcharr/wsgi.py index 29c2ba01..6e8860af 100644 --- a/dispatcharr/wsgi.py +++ b/dispatcharr/wsgi.py @@ -1,6 +1,21 @@ """ WSGI config for dispatcharr project. """ +# When running under uWSGI with gevent, ensure monkey-patching is fully +# applied before any other imports. Django's setup triggers the import of +# dispatcharr/__init__.py → celery.py, which initialises the Celery broker +# transport. If those imports happen before sockets are patched, Celery's +# Redis connection pool holds unpatched sockets that silently fail under +# gevent's cooperative scheduling. Calling patch_all() here—before the +# Django import chain—guarantees every socket (including pooled broker +# connections) is gevent-aware. The call is idempotent, so it's harmless +# if uWSGI's gevent plugin has already patched. +try: + from gevent import monkey + monkey.patch_all() +except ImportError: + pass + import os from django.core.wsgi import get_wsgi_application From dcb3cf8ea866a9e4928e476bbb5398b2e979bb6f Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 11:56:06 -0600 Subject: [PATCH 18/25] changelog: Update changelog for monkey patch PR. --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index fadc44e8..55fac12c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Celery Redis connections using unpatched sockets under gevent: `dispatcharr/__init__.py` imports `celery.py` as part of `django.setup()`, initialising the Redis connection pool before uWSGI's gevent plugin patches stdlib sockets. Added `gevent.monkey.patch_all()` at the top of `wsgi.py` to guarantee all broker connections are gevent-aware. - Thanks [@endoze](https://github.com/endoze) - TS proxy streams dying after 30–200 seconds in multi-worker uWSGI/Celery deployments, caused by three interrelated bugs. (Fixes #992, #980) - Thanks [@PFalko](https://github.com/PFalko) - **Double ProxyServer instantiation**: `ProxyConfig.ready()` called `TSProxyServer()` directly while `TSProxyConfig.ready()` also called `TSProxyServer.get_instance()`, creating two instances per worker — each with its own cleanup thread. The orphaned thread could not extend ownership because it had no entries in `stream_managers`. Fixed by using `TSProxyServer.get_instance()` in `ProxyConfig.ready()`. - **`flushdb()` on every Redis client init**: `RedisClient.get_client()` called `client.flushdb()` whenever `_client` was `None`. Celery autoscale (`--autoscale=6,1`) spawning new workers mid-stream triggered this path, nuking all Redis keys including active ownership keys, client records, and channel metadata. Removed the `flushdb()` call entirely. From 7653859a9a268ba0d9d5aac56152d79800516f16 Mon Sep 17 00:00:00 2001 From: None Date: Tue, 3 Mar 2026 14:22:45 -0600 Subject: [PATCH 19/25] Reset connection_allocated flag after redirect stream release The redirect path in stream_ts() released the connection slot via release_stream() but did not reset the connection_allocated flag. Could have could caused a redundant no-op call in the exception handler. --- apps/proxy/ts_proxy/views.py | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 90ac2673..ccbb947d 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -327,6 +327,7 @@ def stream_ts(request, channel_id): # Release stream lock before redirecting if not channel.release_stream(): logger.warning(f"[{client_id}] Failed to release stream before redirect") + connection_allocated = False # Final decision based on validation results if is_valid: logger.info( From e78c24aad735952eb6f8939dc2935aa4fe20546d Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 15:13:46 -0600 Subject: [PATCH 20/25] Enhance Redis chunk management in StreamBuffer and StreamGenerator - Register Lua scripts for optimized chunk retrieval in StreamBuffer. - Implement atomic Lua binary search to find the oldest available chunk in Redis. - Update StreamGenerator to handle expired chunks more efficiently, reducing TOCTOU race conditions. --- apps/proxy/ts_proxy/stream_buffer.py | 92 +++++++++++++++++++++++++ apps/proxy/ts_proxy/stream_generator.py | 36 +++++++--- 2 files changed, 119 insertions(+), 9 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_buffer.py b/apps/proxy/ts_proxy/stream_buffer.py index 85feb5dd..2bf99527 100644 --- a/apps/proxy/ts_proxy/stream_buffer.py +++ b/apps/proxy/ts_proxy/stream_buffer.py @@ -45,6 +45,15 @@ class StreamBuffer: self._write_buffer = bytearray() self.target_chunk_size = ConfigHelper.get('BUFFER_CHUNK_SIZE', TS_PACKET_SIZE * 5644) # ~1MB default + # Register Lua scripts once — subsequent calls use EVALSHA (just the + # SHA hash) instead of sending the full script text on every invocation. + if self.redis_client: + self._find_oldest_chunk_sha = self.redis_client.register_script( + self._FIND_OLDEST_CHUNK_LUA + ) + else: + self._find_oldest_chunk_sha = None + # Track timers for proper cleanup self.stopping = False self.fill_timers = [] @@ -328,6 +337,89 @@ class StreamBuffer: return chunks, client_index + chunk_count + # Lua script that runs an atomic binary search on the Redis server. + # Chunks expire in FIFO order (same TTL, sequential writes), so the + # alive range is contiguous: [oldest_surviving .. buffer_head]. + # Binary search finds the boundary in O(log N) EXISTS calls with zero + # round-trips between steps and no TOCTOU races (Lua scripts are atomic). + # + # ARGV[1] = key prefix (e.g. "ts_proxy:channel::buffer:chunk:") + # ARGV[2] = low index (client_index + 1, first chunk the client needs) + # ARGV[3] = high index (buffer head, most recent chunk) + # + # Returns: the index of the oldest existing chunk, or -1 if none exist. + _FIND_OLDEST_CHUNK_LUA = """ + local prefix = ARGV[1] + local low = tonumber(ARGV[2]) + local high = tonumber(ARGV[3]) + + if redis.call('EXISTS', prefix .. high) == 0 then + return -1 + end + + local result = high + while low <= high do + local mid = math.floor((low + high) / 2) + if redis.call('EXISTS', prefix .. mid) == 1 then + result = mid + high = mid - 1 + else + low = mid + 1 + end + end + return result + """ + + def find_oldest_available_chunk(self, client_index): + """Find the oldest (lowest-index) chunk that still exists in Redis. + + Executes an atomic Lua binary search on the Redis server — one + round-trip, ~log2(N) EXISTS calls, no TOCTOU between steps. + + The actual read attempt (get_optimized_client_data) is what + authoritatively detects expiration; this method is best-effort + positioning that self-corrects on the next iteration if the found + chunk also expires before the client can read it. + + Args: + client_index: The client's current local_index (last consumed chunk). + + Returns: + int or None: The local_index value the client should jump to + (one before the first available chunk), or None if no + chunks are available at all. + """ + if not self.redis_client: + return None + + low = client_index + 1 # First chunk the client needs + high = self.index # Latest chunk written + + if low > high: + return None + + try: + # Uses EVALSHA under the hood — sends only the SHA hash, + # not the full script text, on every call after the first. + result = self._find_oldest_chunk_sha( + args=[ + RedisKeys.buffer_chunk_prefix(self.channel_id), + low, + high, + ], + ) + + if result == -1: + return None + + # Return result - 1 so local_index points to one before the + # first available chunk (matching the "last consumed" convention). + return int(result) - 1 + + except Exception as e: + logger.error(f"Error running find_oldest_chunk Lua script for channel {self.channel_id}: {e}") + return None + # Add a new method to safely create timers def schedule_timer(self, delay, callback, *args, **kwargs): """Schedule a timer and track it for proper cleanup""" diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 50404f1d..8f631d53 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -224,17 +224,35 @@ class StreamGenerator: self.empty_reads += 1 self.consecutive_empty += 1 - # Check if we're too far behind (chunks expired from Redis) + # We got no data despite being behind the buffer head. + # The read itself is the authoritative signal — no separate + # existence check needed, avoiding TOCTOU races with Redis TTL. chunks_behind = self.buffer.index - self.local_index - if chunks_behind > 50: # If more than 50 chunks behind, jump forward - # Calculate new position: stay a few chunks behind current buffer - initial_behind = ConfigHelper.initial_behind_chunks() - new_index = max(self.local_index, self.buffer.index - initial_behind) + if chunks_behind > 0: + # Next chunk has expired — find the oldest chunk still in Redis + new_index = self.buffer.find_oldest_available_chunk(self.local_index) - logger.warning(f"[{self.client_id}] Client too far behind ({chunks_behind} chunks), jumping from {self.local_index} to {new_index}") - self.local_index = new_index - self.consecutive_empty = 0 # Reset since we're repositioning - continue # Try again immediately with new position + if new_index is not None: + skipped = new_index - self.local_index + logger.warning( + f"[{self.client_id}] Next chunk expired (index {self.local_index + 1}), " + f"jumping to oldest available: {new_index + 1} " + f"(skipped {skipped} chunks, buffer head at {self.buffer.index})" + ) + self.local_index = new_index + else: + # No chunks available at all — jump to near the buffer head + initial_behind = ConfigHelper.initial_behind_chunks() + new_index = max(self.local_index, self.buffer.index - initial_behind) + logger.warning( + f"[{self.client_id}] No chunks available in buffer, " + f"jumping to near buffer head: {new_index} " + f"(buffer head at {self.buffer.index})" + ) + self.local_index = new_index + + self.consecutive_empty = 0 + continue # Retry immediately with the new position if self._should_send_keepalive(self.local_index): keepalive_packet = create_ts_packet('keepalive') From d487bb5671236877c28dfe1e148e7072c9a15fe2 Mon Sep 17 00:00:00 2001 From: SergeantPanda <61642231+SergeantPanda@users.noreply.github.com> Date: Tue, 3 Mar 2026 15:33:55 -0600 Subject: [PATCH 21/25] Revert "fix(wsgi): ensure gevent monkey-patching before Django import chain" --- CHANGELOG.md | 1 - dispatcharr/tests/__init__.py | 0 dispatcharr/tests/test_wsgi.py | 39 ---------------------------------- dispatcharr/wsgi.py | 15 ------------- 4 files changed, 55 deletions(-) delete mode 100644 dispatcharr/tests/__init__.py delete mode 100644 dispatcharr/tests/test_wsgi.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 55fac12c..fadc44e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Celery Redis connections using unpatched sockets under gevent: `dispatcharr/__init__.py` imports `celery.py` as part of `django.setup()`, initialising the Redis connection pool before uWSGI's gevent plugin patches stdlib sockets. Added `gevent.monkey.patch_all()` at the top of `wsgi.py` to guarantee all broker connections are gevent-aware. - Thanks [@endoze](https://github.com/endoze) - TS proxy streams dying after 30–200 seconds in multi-worker uWSGI/Celery deployments, caused by three interrelated bugs. (Fixes #992, #980) - Thanks [@PFalko](https://github.com/PFalko) - **Double ProxyServer instantiation**: `ProxyConfig.ready()` called `TSProxyServer()` directly while `TSProxyConfig.ready()` also called `TSProxyServer.get_instance()`, creating two instances per worker — each with its own cleanup thread. The orphaned thread could not extend ownership because it had no entries in `stream_managers`. Fixed by using `TSProxyServer.get_instance()` in `ProxyConfig.ready()`. - **`flushdb()` on every Redis client init**: `RedisClient.get_client()` called `client.flushdb()` whenever `_client` was `None`. Celery autoscale (`--autoscale=6,1`) spawning new workers mid-stream triggered this path, nuking all Redis keys including active ownership keys, client records, and channel metadata. Removed the `flushdb()` call entirely. diff --git a/dispatcharr/tests/__init__.py b/dispatcharr/tests/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/dispatcharr/tests/test_wsgi.py b/dispatcharr/tests/test_wsgi.py deleted file mode 100644 index cdf7a2e9..00000000 --- a/dispatcharr/tests/test_wsgi.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Tests for dispatcharr/wsgi.py gevent monkey-patching.""" - -import os -import subprocess -import sys -import unittest - - -class TestWSGIGeventPatching(unittest.TestCase): - """Verify that wsgi.py applies gevent monkey-patching before Django imports. - - These tests run in subprocesses because monkey.patch_all() must execute - before any other imports. By the time Django's test runner loads this - file, ssl/socket are already imported, so calling patch_all() in-process - would fail. A subprocess mirrors how uWSGI actually loads wsgi.py. - """ - - def test_socket_is_patched_after_wsgi_import(self): - """Loading wsgi.py first (as uWSGI does) should patch sockets.""" - result = subprocess.run( - [ - sys.executable, "-c", - "import dispatcharr.wsgi; " - "from gevent import monkey; " - "assert monkey.is_module_patched('socket'), " - "'socket module was not patched by wsgi.py'; " - "print('PASS')", - ], - capture_output=True, - text=True, - env={**os.environ, "DJANGO_SETTINGS_MODULE": "dispatcharr.settings"}, - timeout=60, - cwd="/app", - ) - self.assertEqual( - result.returncode, - 0, - f"monkey-patching check failed:\n{result.stderr}", - ) diff --git a/dispatcharr/wsgi.py b/dispatcharr/wsgi.py index 6e8860af..29c2ba01 100644 --- a/dispatcharr/wsgi.py +++ b/dispatcharr/wsgi.py @@ -1,21 +1,6 @@ """ WSGI config for dispatcharr project. """ -# When running under uWSGI with gevent, ensure monkey-patching is fully -# applied before any other imports. Django's setup triggers the import of -# dispatcharr/__init__.py → celery.py, which initialises the Celery broker -# transport. If those imports happen before sockets are patched, Celery's -# Redis connection pool holds unpatched sockets that silently fail under -# gevent's cooperative scheduling. Calling patch_all() here—before the -# Django import chain—guarantees every socket (including pooled broker -# connections) is gevent-aware. The call is idempotent, so it's harmless -# if uWSGI's gevent plugin has already patched. -try: - from gevent import monkey - monkey.patch_all() -except ImportError: - pass - import os from django.core.wsgi import get_wsgi_application From 24c20a2f4accf486b89c7e664d92571d2cdc845b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 15:59:49 -0600 Subject: [PATCH 22/25] changelog: Update changelog for proxy changes. --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index fadc44e8..49e7be68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- TS proxy client stream lag recovery now only bumps clients forward when their next required chunk has genuinely expired from Redis (TTL), rather than unconditionally jumping if they fell more than 50 chunks behind. Clients are repositioned to the oldest available chunk (minimum data loss) using an atomic server-side Lua binary search, falling back to near the buffer head if nothing is available. - TS proxy streams dying after 30–200 seconds in multi-worker uWSGI/Celery deployments, caused by three interrelated bugs. (Fixes #992, #980) - Thanks [@PFalko](https://github.com/PFalko) - **Double ProxyServer instantiation**: `ProxyConfig.ready()` called `TSProxyServer()` directly while `TSProxyConfig.ready()` also called `TSProxyServer.get_instance()`, creating two instances per worker — each with its own cleanup thread. The orphaned thread could not extend ownership because it had no entries in `stream_managers`. Fixed by using `TSProxyServer.get_instance()` in `ProxyConfig.ready()`. - **`flushdb()` on every Redis client init**: `RedisClient.get_client()` called `client.flushdb()` whenever `_client` was `None`. Celery autoscale (`--autoscale=6,1`) spawning new workers mid-stream triggered this path, nuking all Redis keys including active ownership keys, client records, and channel metadata. Removed the `flushdb()` call entirely. From 086259ef9dfad69c83efe11379a12536b652d29a Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 16:28:30 -0600 Subject: [PATCH 23/25] changelog: Update changelog for stream lock PR. --- CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 49e7be68..be8b88b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- TS proxy connection slot leaks and TOCTOU races in stream initialization (Fixes #947) - Thanks [@CodeBormen](https://github.com/CodeBormen) + - **TOCTOU race in slot reservation**: `get_stream()` previously used a `GET`→check→`INCR` sequence, allowing concurrent requests to both read the same count below the limit and both reserve a slot, silently exceeding `max_streams`. Replaced with an atomic `INCR`-first pattern: increment unconditionally, check the result, roll back with `DECR` if over capacity. + - **Leak on URL generation failure**: `generate_stream_url()` called `get_stream()` (which `INCR`s the counter) but had no cleanup path if subsequent DB lookups or URL construction failed. The post-`get_stream()` block is now wrapped in a `try/except` that calls `release_stream()` on any error. + - **Leak on retry-loop timeout**: the retry loop in `stream_ts()` called a bare `get_stream()` on the first failure to classify the error reason. If a slot was available, this `INCR`'d the counter and set Redis keys that were never released when the loop timed out. A `release_stream()` call is now issued before returning 503. + - **Leak on `initialize_channel()` failure**: when `initialize_channel()` returned `False`, the connection slot allocated by the preceding `get_stream()` was never released. A `connection_allocated` flag now tracks whether this request performed the `INCR` (fresh initialization vs. joining an existing channel), and `release_stream()` is called guarded by that flag to prevent incorrect decrements when attaching to an already-running channel. + - **Safety net for unexpected exceptions**: the outer `except` in `stream_ts()` now checks `connection_allocated` and calls `release_stream()` as a last-resort cleanup for any unhandled exception that escapes before the channel is handed off to the stream lifecycle. + - **`release_stream()` now returns `bool`** and adds a metadata-hash fallback: if the primary `channel_stream` / `stream_profile` Redis keys have already been cleaned up by the proxy, it recovers `stream_id` and `profile_id` from the channel's metadata hash and clears those fields atomically to prevent duplicate `DECR`s on repeated calls. + - **`update_stream_profile()` uses a Redis pipeline** for the old-profile DECR + key update + new-profile INCR sequence, preventing counter drift if the process crashes between operations. + - **`stream_generator._cleanup()`** now falls back to `Stream.objects.get()` when the channel UUID resolves to a preview flow rather than a normal channel, rather than silently skipping the slot release. + - **VOD `cleanup_persistent_connection()`** fallback DECR is now conditional: it only decrements the profile counter when the connection tracking key had already expired by TTL (i.e., `remove_connection()` would have skipped the DECR), preventing double-decrements when the key is still present. - TS proxy client stream lag recovery now only bumps clients forward when their next required chunk has genuinely expired from Redis (TTL), rather than unconditionally jumping if they fell more than 50 chunks behind. Clients are repositioned to the oldest available chunk (minimum data loss) using an atomic server-side Lua binary search, falling back to near the buffer head if nothing is available. - TS proxy streams dying after 30–200 seconds in multi-worker uWSGI/Celery deployments, caused by three interrelated bugs. (Fixes #992, #980) - Thanks [@PFalko](https://github.com/PFalko) - **Double ProxyServer instantiation**: `ProxyConfig.ready()` called `TSProxyServer()` directly while `TSProxyConfig.ready()` also called `TSProxyServer.get_instance()`, creating two instances per worker — each with its own cleanup thread. The orphaned thread could not extend ownership because it had no entries in `stream_managers`. Fixed by using `TSProxyServer.get_instance()` in `ProxyConfig.ready()`. From fc75a68195b59d1f00d1df2994c8741f443a6d76 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 18:12:32 -0600 Subject: [PATCH 24/25] =?UTF-8?q?Enhancement:=20new=20clients=20joining=20?= =?UTF-8?q?an=20active=20channel=20are=20now=20positioned=20a=20configurab?= =?UTF-8?q?le=20number=20of=20seconds=20behind=20live=20rather=20than=20a?= =?UTF-8?q?=20fixed=20chunk=20count.=20The=20start=20position=20is=20deter?= =?UTF-8?q?mined=20by=20wall-clock=20chunk=20receive=20time=20(stored=20as?= =?UTF-8?q?=20a=20Redis=20sorted=20set=20alongside=20the=20buffer),=20so?= =?UTF-8?q?=20the=20buffer=20depth=20is=20consistent=20in=20seconds=20rega?= =?UTF-8?q?rdless=20of=20stream=20bitrate.=20Setting=20the=20value=20to=20?= =?UTF-8?q?`0`=20starts=20clients=20at=20live=20with=20no=20buffer.=20Defa?= =?UTF-8?q?ults=20to=202=20seconds.=20Existing=20chunk-count=20gating=20fo?= =?UTF-8?q?r=20the=20first=20client=20connecting=20to=20a=20channel=20is?= =?UTF-8?q?=20unchanged.=20The=20setting=20is=20exposed=20in=20Settings=20?= =?UTF-8?q?=E2=86=92=20Proxy=20as=20"New=20Client=20Buffer=20(seconds)".?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 1 + apps/proxy/config.py | 2 + apps/proxy/ts_proxy/config_helper.py | 9 +++ apps/proxy/ts_proxy/redis_keys.py | 6 ++ apps/proxy/ts_proxy/stream_buffer.py | 79 +++++++++++++++++++ apps/proxy/ts_proxy/stream_generator.py | 43 +++++++++- core/api_views.py | 1 + core/models.py | 1 + core/serializers.py | 6 ++ .../forms/settings/ProxySettingsForm.jsx | 12 ++- frontend/src/constants.js | 5 ++ .../forms/settings/ProxySettingsFormUtils.js | 1 + 12 files changed, 161 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index be8b88b4..88f6797a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- New Client Buffer proxy setting: new clients joining an active channel are now positioned a configurable number of seconds behind live rather than a fixed chunk count. The start position is determined by wall-clock chunk receive time (stored as a Redis sorted set alongside the buffer), so the buffer depth is consistent in seconds regardless of stream bitrate. Setting the value to `0` starts clients at live with no buffer. Defaults to 2 seconds. Existing chunk-count gating for the first client connecting to a channel is unchanged. The setting is exposed in Settings → Proxy as "New Client Buffer (seconds)". - Channel table filter for channels that have stale streams: A new "Has Stale Streams" filter option in the channel table header menu highlights and filters channels containing at least one stale stream. Channels with stale streams are visually distinguished with an orange tint. The filter is mutually exclusive with "Only Empty Channels". - Thanks [@JCBird1012](https://github.com/JCBird1012) ### Fixed diff --git a/apps/proxy/config.py b/apps/proxy/config.py index 3b1ce967..dbdd8278 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -43,6 +43,7 @@ class BaseConfig: "redis_chunk_ttl": 60, "channel_shutdown_delay": 0, "channel_init_grace_period": 5, + "new_client_behind_seconds": 2, } finally: @@ -81,6 +82,7 @@ class TSConfig(BaseConfig): # Buffer settings INITIAL_BEHIND_CHUNKS = 4 # How many chunks behind to start a client (4 chunks = ~1MB) CHUNK_BATCH_SIZE = 5 # How many chunks to fetch in one batch + NEW_CLIENT_BEHIND_SECONDS = 2 # Start new clients this many seconds behind live (0 = start at live) KEEPALIVE_INTERVAL = 0.5 # Seconds between keepalive packets when at buffer head # Chunk read timeout CHUNK_TIMEOUT = 5 # Seconds to wait for each chunk read diff --git a/apps/proxy/ts_proxy/config_helper.py b/apps/proxy/ts_proxy/config_helper.py index d7d33558..f0f7a5b7 100644 --- a/apps/proxy/ts_proxy/config_helper.py +++ b/apps/proxy/ts_proxy/config_helper.py @@ -41,6 +41,15 @@ class ConfigHelper: """Get number of chunks to start behind""" return ConfigHelper.get('INITIAL_BEHIND_CHUNKS', 4) + @staticmethod + def new_client_behind_seconds(): + """Get number of seconds behind live to start new clients. + 0 means start at live (buffer head). + Loaded from DB proxy_settings so users can change it at runtime.""" + from apps.proxy.config import TSConfig + settings = TSConfig.get_proxy_settings() + return settings.get('new_client_behind_seconds', 2) + @staticmethod def keepalive_interval(): """Get keepalive interval in seconds""" diff --git a/apps/proxy/ts_proxy/redis_keys.py b/apps/proxy/ts_proxy/redis_keys.py index 22b02648..29846f98 100644 --- a/apps/proxy/ts_proxy/redis_keys.py +++ b/apps/proxy/ts_proxy/redis_keys.py @@ -79,6 +79,12 @@ class RedisKeys: """Key for worker heartbeat""" return f"ts_proxy:worker:{worker_id}:heartbeat" + @staticmethod + def chunk_timestamps(channel_id): + """Sorted set mapping chunk receive-timestamps (score) to chunk indices (member). + Used for time-based client positioning.""" + return f"ts_proxy:channel:{channel_id}:buffer:chunk_timestamps" + @staticmethod def transcode_active(channel_id): """Key indicating active transcode process""" diff --git a/apps/proxy/ts_proxy/stream_buffer.py b/apps/proxy/ts_proxy/stream_buffer.py index 2bf99527..111468bf 100644 --- a/apps/proxy/ts_proxy/stream_buffer.py +++ b/apps/proxy/ts_proxy/stream_buffer.py @@ -45,14 +45,21 @@ class StreamBuffer: self._write_buffer = bytearray() self.target_chunk_size = ConfigHelper.get('BUFFER_CHUNK_SIZE', TS_PACKET_SIZE * 5644) # ~1MB default + # Sorted-set key for chunk receive-timestamps (time-based positioning) + self.chunk_timestamps_key = RedisKeys.chunk_timestamps(channel_id) if channel_id else "" + # Register Lua scripts once — subsequent calls use EVALSHA (just the # SHA hash) instead of sending the full script text on every invocation. if self.redis_client: self._find_oldest_chunk_sha = self.redis_client.register_script( self._FIND_OLDEST_CHUNK_LUA ) + self._find_chunk_by_time_sha = self.redis_client.register_script( + self._FIND_CHUNK_BY_TIME_LUA + ) else: self._find_oldest_chunk_sha = None + self._find_chunk_by_time_sha = None # Track timers for proper cleanup self.stopping = False @@ -101,6 +108,14 @@ class StreamBuffer: chunk_key = RedisKeys.buffer_chunk(self.channel_id, chunk_index) self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(chunk_data)) + # Record receive timestamp for time-based client positioning + if self.chunk_timestamps_key: + now = time.time() + self.redis_client.zadd(self.chunk_timestamps_key, {str(chunk_index): now}) + # Prune entries whose chunks have expired from Redis + self.redis_client.zremrangebyscore(self.chunk_timestamps_key, '-inf', now - self.chunk_ttl) + self.redis_client.expire(self.chunk_timestamps_key, self.chunk_ttl) + # Update local tracking self.index = chunk_index writes_done += 1 @@ -284,6 +299,13 @@ class StreamBuffer: if hasattr(self, '_partial_packet'): self._partial_packet = bytearray() + # Clean up the chunk timestamps sorted set + if self.redis_client and self.chunk_timestamps_key: + try: + self.redis_client.delete(self.chunk_timestamps_key) + except Exception as e: + logger.error(f"Error deleting chunk timestamps key: {e}") + except Exception as e: logger.error(f"Error during buffer stop: {e}") @@ -420,6 +442,63 @@ class StreamBuffer: logger.error(f"Error running find_oldest_chunk Lua script for channel {self.channel_id}: {e}") return None + # ------------------------------------------------------------------ + # Lua script: atomic reverse-scan of the chunk_timestamps sorted set. + # Finds the chunk whose receive-timestamp is closest to (but <=) a + # target wall-clock time. Returns the chunk index or -1. + # + # KEYS[1] = chunk_timestamps sorted-set key + # ARGV[1] = target timestamp (time.time() - desired_seconds_behind) + # ------------------------------------------------------------------ + _FIND_CHUNK_BY_TIME_LUA = """ + local ts_key = KEYS[1] + local target = tonumber(ARGV[1]) + + -- ZREVRANGEBYSCORE returns members with score <= target, highest first. + local result = redis.call('ZREVRANGEBYSCORE', ts_key, target, '-inf', 'LIMIT', 0, 1) + if #result == 0 then + return -1 + end + return tonumber(result[1]) + """ + + def find_chunk_index_by_time(self, seconds_behind): + """Find the chunk index that was received approximately *seconds_behind* + seconds ago. + + Uses an atomic Lua script against the chunk_timestamps sorted set so + no data can expire between the lookup and the read. + + Returns: + int or None: The chunk index to position the client at (this is + the *last consumed* convention, so the next read + starts at index+1). None if no suitable chunk + exists. + """ + if not self.redis_client or not self.chunk_timestamps_key: + return None + + target_time = time.time() - seconds_behind + + try: + result = self._find_chunk_by_time_sha( + keys=[self.chunk_timestamps_key], + args=[target_time], + ) + if result is None or int(result) == -1: + # No chunk old enough — fall back to the oldest available chunk + oldest = self.redis_client.zrange(self.chunk_timestamps_key, 0, 0) + if oldest: + return max(0, int(oldest[0]) - 1) # "last consumed" convention + return None + + # Return index - 1 so next read starts at that chunk + return max(0, int(result) - 1) + + except Exception as e: + logger.error(f"Error in find_chunk_index_by_time for channel {self.channel_id}: {e}") + return None + # Add a new method to safely create timers def schedule_timer(self, delay, callback, *args, **kwargs): """Schedule a timer and track it for proper cleanup""" diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index a8ee799f..05892293 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -186,10 +186,47 @@ class StreamGenerator: logger.error(f"[{self.client_id}] No buffer found for channel {self.channel_id}") return False - # Client state tracking - use config for initial position - initial_behind = ConfigHelper.initial_behind_chunks() + # Client state tracking — determine start position + # When behind_seconds > 0, use time-based positioning to start + # the client that many seconds behind live. + # When behind_seconds == 0, start at live (buffer head). + behind_seconds = ConfigHelper.new_client_behind_seconds() current_buffer_index = buffer.index - self.local_index = max(0, current_buffer_index - initial_behind) + + if behind_seconds > 0: + time_index = buffer.find_chunk_index_by_time(behind_seconds) + if time_index is not None: + self.local_index = max(0, time_index) + logger.info( + f"[{self.client_id}] Time-based positioning: " + f"{behind_seconds}s behind -> index {self.local_index} " + f"(buffer head at {current_buffer_index})" + ) + else: + # Not enough buffer for the requested time — start as far + # back as possible (oldest available chunk). + oldest = buffer.find_oldest_available_chunk(0) + if oldest is not None: + self.local_index = max(0, oldest) + logger.info( + f"[{self.client_id}] Buffer shorter than {behind_seconds}s, " + f"starting at oldest available chunk {self.local_index} " + f"(buffer head at {current_buffer_index})" + ) + else: + # No timestamp data at all — start at live + self.local_index = current_buffer_index + logger.info( + f"[{self.client_id}] No timestamp data, starting at live: " + f"index {self.local_index} (buffer head at {current_buffer_index})" + ) + else: + # 0 = start at live (buffer head) + self.local_index = current_buffer_index + logger.info( + f"[{self.client_id}] Starting at live (behind_seconds=0): " + f"index {self.local_index} (buffer head at {current_buffer_index})" + ) # Store important objects as instance variables self.buffer = buffer diff --git a/core/api_views.py b/core/api_views.py index 3d131ab9..43e55d02 100644 --- a/core/api_views.py +++ b/core/api_views.py @@ -183,6 +183,7 @@ class ProxySettingsViewSet(viewsets.ViewSet): "redis_chunk_ttl": 60, "channel_shutdown_delay": 0, "channel_init_grace_period": 5, + "new_client_behind_seconds": 2, } settings_obj, created = CoreSettings.objects.get_or_create( key=PROXY_SETTINGS_KEY, diff --git a/core/models.py b/core/models.py index 10f9073f..7ce7713d 100644 --- a/core/models.py +++ b/core/models.py @@ -329,6 +329,7 @@ class CoreSettings(models.Model): "redis_chunk_ttl": 60, "channel_shutdown_delay": 0, "channel_init_grace_period": 5, + "new_client_behind_seconds": 2, }) # System Settings diff --git a/core/serializers.py b/core/serializers.py index 03ba33e8..b0b730ee 100644 --- a/core/serializers.py +++ b/core/serializers.py @@ -78,6 +78,7 @@ class ProxySettingsSerializer(serializers.Serializer): redis_chunk_ttl = serializers.IntegerField(min_value=10, max_value=3600) channel_shutdown_delay = serializers.IntegerField(min_value=0, max_value=300) channel_init_grace_period = serializers.IntegerField(min_value=0, max_value=60) + new_client_behind_seconds = serializers.IntegerField(min_value=0, max_value=120, required=False, default=2) def validate_buffering_timeout(self, value): if value < 0 or value > 300: @@ -104,6 +105,11 @@ class ProxySettingsSerializer(serializers.Serializer): raise serializers.ValidationError("Channel init grace period must be between 0 and 60 seconds") return value + def validate_new_client_behind_seconds(self, value): + if value < 0 or value > 120: + raise serializers.ValidationError("New client buffer must be between 0 and 120 seconds") + return value + class SystemNotificationSerializer(serializers.ModelSerializer): """Serializer for system notifications.""" diff --git a/frontend/src/components/forms/settings/ProxySettingsForm.jsx b/frontend/src/components/forms/settings/ProxySettingsForm.jsx index a5c8d858..52440769 100644 --- a/frontend/src/components/forms/settings/ProxySettingsForm.jsx +++ b/frontend/src/components/forms/settings/ProxySettingsForm.jsx @@ -24,6 +24,7 @@ const ProxySettingsOptions = React.memo(({ proxySettingsForm }) => { 'redis_chunk_ttl', 'channel_shutdown_delay', 'channel_init_grace_period', + 'new_client_behind_seconds', ].includes(key); }; const isFloatField = (key) => { @@ -36,7 +37,9 @@ const ProxySettingsOptions = React.memo(({ proxySettingsForm }) => { ? 3600 : key === 'channel_shutdown_delay' ? 300 - : 60; + : key === 'new_client_behind_seconds' + ? 120 + : 60; }; return ( <> @@ -97,7 +100,12 @@ const ProxySettingsForm = React.memo(({ active }) => { useEffect(() => { if (settings) { if (settings['proxy_settings']?.value) { - proxySettingsForm.setValues(settings['proxy_settings'].value); + // Merge defaults so any newly-added keys not yet in the stored + // settings object still show their default value rather than blank. + proxySettingsForm.setValues({ + ...getProxySettingDefaults(), + ...settings['proxy_settings'].value, + }); } } }, [settings]); diff --git a/frontend/src/constants.js b/frontend/src/constants.js index d2caa45b..0c19eea1 100644 --- a/frontend/src/constants.js +++ b/frontend/src/constants.js @@ -55,6 +55,11 @@ export const PROXY_SETTINGS_OPTIONS = { label: 'Channel Initialization Grace Period', description: 'Grace period in seconds during channel initialization', }, + new_client_behind_seconds: { + label: 'New Client Buffer (seconds)', + description: + 'Seconds of received buffer to start behind live when a new client connects (0 = start at live). Note: this is chunk receive time, not video duration.', + }, }; export const M3U_FILTER_TYPES = [ diff --git a/frontend/src/utils/forms/settings/ProxySettingsFormUtils.js b/frontend/src/utils/forms/settings/ProxySettingsFormUtils.js index 9ca185e8..e6e0464b 100644 --- a/frontend/src/utils/forms/settings/ProxySettingsFormUtils.js +++ b/frontend/src/utils/forms/settings/ProxySettingsFormUtils.js @@ -14,5 +14,6 @@ export const getProxySettingDefaults = () => { redis_chunk_ttl: 60, channel_shutdown_delay: 0, channel_init_grace_period: 5, + new_client_behind_seconds: 2, }; }; From b9a9d929cee8e642cf1fff62bfbda83cac242cdd Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 18:21:52 -0600 Subject: [PATCH 25/25] tests: Fix proxy settings test. --- .../forms/settings/__tests__/ProxySettingsFormUtils.test.js | 2 ++ 1 file changed, 2 insertions(+) diff --git a/frontend/src/utils/forms/settings/__tests__/ProxySettingsFormUtils.test.js b/frontend/src/utils/forms/settings/__tests__/ProxySettingsFormUtils.test.js index 8feefccb..a48e5a37 100644 --- a/frontend/src/utils/forms/settings/__tests__/ProxySettingsFormUtils.test.js +++ b/frontend/src/utils/forms/settings/__tests__/ProxySettingsFormUtils.test.js @@ -61,6 +61,7 @@ describe('ProxySettingsFormUtils', () => { redis_chunk_ttl: 60, channel_shutdown_delay: 0, channel_init_grace_period: 5, + new_client_behind_seconds: 2, }); }); @@ -80,6 +81,7 @@ describe('ProxySettingsFormUtils', () => { expect(typeof result.redis_chunk_ttl).toBe('number'); expect(typeof result.channel_shutdown_delay).toBe('number'); expect(typeof result.channel_init_grace_period).toBe('number'); + expect(typeof result.new_client_behind_seconds).toBe('number'); }); }); });