Merge pull request #949 from CodeBormen:fix/947-Connection-capacity-leak-failed-stream-initialization-permanently-consumes-connection-slot

Fix/(ts-proxy): Harden connection slot lifecycle — leak prevention, TOCTOU race, and release safety

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

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
@@ -204,7 +206,10 @@ class Stream(models.Model):
 
     def get_stream(self):
         """
-        Finds an available stream for the requested channel and returns the selected stream and profile.
+        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}")
@@ -226,33 +231,32 @@ class Stream(models.Model):
             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)
+            # 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
 
-            # 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:
                 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
+                redis_client.set(f"stream_profile:{self.id}", profile.id)
+                return self.id, profile.id, None
 
-                # 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
+        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.
+
+        Returns:
+            bool: True if stream was successfully released, False if
+                  no profile info could be found for cleanup.
         """
         redis_client = RedisClient.get_client()
 
@@ -260,14 +264,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}"
@@ -277,6 +284,8 @@ class Stream(models.Model):
         if current_count > 0:
             redis_client.decr(profile_connections_key)
 
+        return True
+
 
 class ChannelManager(models.Manager):
     def active(self):
@@ -391,6 +400,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 +497,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 +516,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
@@ -499,32 +533,93 @@ 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}")
+
+                # 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
+                )
+                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
-
-        redis_client.delete(f"stream_profile:{stream_id}")  # Remove profile association
-
-        profile_id = int(profile_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
+            )
+            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"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}"
@@ -534,6 +629,18 @@ 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):
         """
         Updates the profile for the current stream and adjusts connection counts.
@@ -566,18 +673,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}"
         )

apps/proxy/ts_proxy/server.py

@@ -796,7 +796,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
@@ -818,13 +821,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}")
 
@@ -1396,10 +1403,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, 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, Exception):
+                logger.debug(f"No Channel or Stream found for {channel_id}")
 
         if not self.redis_client:
             return 0

apps/proxy/ts_proxy/services/channel_service.py

@@ -265,18 +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
-        except Channel.DoesNotExist:
+            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, Exception):
             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
-        except Exception as e:
-            logger.error(f"Error releasing channel stream: {e}")
-            model_released = False
+            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, Exception) as e:
+                logger.error(f"No Channel or Stream found for {channel_id}: {e}")
+                model_released = False
 
         return {
             'status': 'success',

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
@@ -455,13 +455,17 @@ 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)
-                                    channel.release_stream()
-                                    stream_released = True
-                                    logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
+                                    # 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:
+                                        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:
@@ -490,8 +494,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):
         """

apps/proxy/ts_proxy/url_utils.py

@@ -129,39 +129,42 @@ 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}")
+            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
-
-        # Get the M3U account profile for URL pattern
-        m3u_profile = profile
-
-        # Get the appropriate user agent
-        m3u_account = M3UAccount.objects.get(id=m3u_profile.m3u_account.id)
-        stream_user_agent = m3u_account.get_user_agent().user_agent
-
-        if stream_user_agent is None:
-            stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
-            logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
-
-        # Generate stream URL based on the selected profile
-        input_url = stream.url
-        stream_url = transform_url(input_url, m3u_profile.search_pattern, m3u_profile.replace_pattern)
-
-        # Check if transcoding is needed
-        stream_profile = channel.get_stream_profile()
-        if stream_profile.is_proxy() or stream_profile is None:
-            transcode = False
-        else:
-            transcode = True
-
-        stream_profile_id = stream_profile.id
-
-        return stream_url, stream_user_agent, transcode, stream_profile_id
     except Exception as e:
         logger.error(f"Error generating stream URL: {e}")
         return None, None, False, None

apps/proxy/ts_proxy/views.py

@@ -54,6 +54,7 @@ def stream_ts(request, channel_id):
 
     client_user_agent = None
     proxy_server = ProxyServer.get_instance()
+    connection_allocated = False  # Track if connection slot was allocated via get_stream()
 
     try:
         # Generate a unique client ID
@@ -219,9 +220,10 @@ 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
+                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"
@@ -237,8 +239,23 @@ def stream_ts(request, channel_id):
                     {"error": error_msg, "waited": wait_duration}, status=503
                 )  # 503 Service Unavailable is appropriate here
 
-            # Get the stream ID from the channel
-            stream_id, m3u_profile_id, _ = channel.get_stream()
+            # 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
+
+            # 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}"
             )
@@ -308,7 +325,9 @@ 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")
+                connection_allocated = False
                 # Final decision based on validation results
                 if is_valid:
                     logger.info(
@@ -344,10 +363,17 @@ def stream_ts(request, channel_id):
             )
 
             if not success:
+                if connection_allocated:
+                    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
                 )
 
+            # 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 +533,12 @@ 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:
+                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)
 
 

apps/proxy/vod_proxy/connection_manager.py

@@ -1402,17 +1402,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)

apps/proxy/vod_proxy/multi_worker_connection_manager.py

@@ -1154,9 +1154,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}")
 
@@ -1380,12 +1381,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}")