fix(vod-proxy): eliminate profile_connections counter stuck-at-nonzero races

Three race conditions in multi_worker_connection_manager could leave
the Redis profile_connections:<id> counter permanently elevated with no
active streams, causing all VOD requests to 503 "All profiles at capacity".

Bug 1 — decrement_active_streams() return value was ignored
All three generator exit paths (normal, GeneratorExit, Exception) called
decrement_active_streams() and unconditionally set decremented = True
regardless of whether the lock was acquired. On lock contention the decrement
was silently skipped, active_streams remained > 0, the subsequent
has_active_streams() check returned True, and _decrement_profile_connections()
was never called. The counter was then stuck until manual DEL.

Fix: add decrement_active_streams_and_check() which performs the decrement and
the "are there remaining streams?" check atomically under a single lock,
eliminating the race window. All three exit paths and the finally block now
use this method and propagate its success/remaining return values.

Bug 2 — non-atomic GET-then-DECR in _decrement_profile_connections()
The previous implementation read the counter with GET then conditionally
called DECR. Two concurrent decrements could both pass the > 0 guard and
both fire, driving the counter to -1. A subsequent _check_and_reserve_profile_slot()
INCR would then produce 0 which passes the <= max_streams check, allowing
an extra stream to bypass the limit on the next request.

Fix: replace GET-then-DECR with a direct DECR (matching the INCR-first
pattern already used by _check_and_reserve_profile_slot) and clamp the
result to 0 if it goes negative.

Bug 3 — has_active_streams() read state without holding the lock
The separate has_active_streams() call after decrement_active_streams()
released its lock left a window where another concurrent stream could
increment active_streams back to 1, causing the profile decrement to be
skipped. This is resolved as a consequence of Bug 1's fix: the new
decrement_active_streams_and_check() method reads active_streams while
the lock is still held, eliminating the window entirely.

Adds tests covering all three scenarios in
apps/proxy/vod_proxy/tests/test_profile_connections.py.

Fixes #1121.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/proxy/vod_proxy/multi_worker_connection_manager.py

@@ -522,6 +522,38 @@ class RedisBackedVODConnection:
         finally:
             self._release_lock()
 
+    def decrement_active_streams_and_check(self):
+        """Atomically decrement active streams and return (success, has_remaining_streams).
+
+        Combines decrement + check under a single lock to eliminate the race window
+        between separate decrement_active_streams() and has_active_streams() calls.
+
+        Returns:
+            (True, False)  - decremented successfully, no streams remain
+            (True, True)   - decremented successfully, other streams still active
+            (False, True)  - lock contention, assume streams remain (safe default)
+        """
+        if not self._acquire_lock():
+            logger.warning(f"[{self.session_id}] DECR-AS-CHECK failed: could not acquire lock")
+            return False, True  # Assume remaining to avoid skipping profile decrement
+
+        try:
+            state = self._get_connection_state()
+            if state and state.active_streams > 0:
+                old = state.active_streams
+                state.active_streams -= 1
+                state.last_activity = time.time()
+                self._save_connection_state(state)
+                logger.debug(f"[{self.session_id}] DECR-AS {old} -> {state.active_streams}")
+                return True, state.active_streams > 0
+            if not state:
+                logger.warning(f"[{self.session_id}] DECR-AS-CHECK failed: no state")
+                return False, False
+            logger.warning(f"[{self.session_id}] DECR-AS-CHECK failed: active_streams already {state.active_streams}")
+            return False, False
+        finally:
+            self._release_lock()
+
     def has_active_streams(self) -> bool:
         """Check if connection has any active streams"""
         state = self._get_connection_state()
@@ -744,17 +776,22 @@ class MultiWorkerVODConnectionManager:
             return None
 
     def _decrement_profile_connections(self, m3u_profile_id: int):
-        """Decrement profile connection count"""
+        """Decrement profile connection count.
+
+        Uses a single atomic DECR (no GET-before-DECR) to avoid the race condition
+        where two concurrent decrements both pass a >0 guard and both fire, sending
+        the counter negative. If the counter would go below zero it is clamped to 0.
+        """
         try:
             profile_connections_key = self._get_profile_connections_key(m3u_profile_id)
-            current_count = int(self.redis_client.get(profile_connections_key) or 0)
-            if current_count > 0:
-                new_count = self.redis_client.decr(profile_connections_key)
-                logger.info(f"[PROFILE-DECR] Profile {m3u_profile_id} connections: {new_count}")
-                return new_count
+            new_count = self.redis_client.decr(profile_connections_key)
+            if new_count < 0:
+                self.redis_client.set(profile_connections_key, 0)
+                new_count = 0
+                logger.warning(f"[PROFILE-DECR] Profile {m3u_profile_id} counter went negative, clamped to 0")
             else:
-                logger.warning(f"[PROFILE-DECR] Profile {m3u_profile_id} already at 0 connections")
-                return 0
+                logger.info(f"[PROFILE-DECR] Profile {m3u_profile_id} connections: {new_count}")
+            return new_count
         except Exception as e:
             logger.error(f"Error decrementing profile connections: {e}")
             return None
@@ -986,11 +1023,10 @@ class MultiWorkerVODConnectionManager:
                         logger.info(f"[{client_id}] Worker {self.worker_id} - Stream stopped by signal: {bytes_sent} bytes sent")
                     else:
                         logger.info(f"[{client_id}] Worker {self.worker_id} - Redis-backed stream completed: {bytes_sent} bytes sent")
-                    redis_connection.decrement_active_streams()
-                    decremented = True
+                    decremented, has_remaining = redis_connection.decrement_active_streams_and_check()
 
                     # Schedule smart cleanup if no active streams after normal completion
-                    if not redis_connection.has_active_streams():
+                    if decremented and not has_remaining:
                         # Decrement profile counter immediately — don't defer to daemon thread
                         state = redis_connection._get_connection_state()
                         profile_id = state.m3u_profile_id if state else m3u_profile.id
@@ -1013,11 +1049,12 @@ class MultiWorkerVODConnectionManager:
                 except GeneratorExit:
                     logger.info(f"[{client_id}] Worker {self.worker_id} - Client disconnected from Redis-backed stream")
                     if not decremented:
-                        redis_connection.decrement_active_streams()
-                        decremented = True
+                        decremented, has_remaining = redis_connection.decrement_active_streams_and_check()
+                    else:
+                        has_remaining = redis_connection.has_active_streams()
 
                     # Schedule smart cleanup if no active streams
-                    if not redis_connection.has_active_streams():
+                    if not has_remaining:
                         # Decrement profile counter immediately — don't defer to daemon thread
                         state = redis_connection._get_connection_state()
                         profile_id = state.m3u_profile_id if state else m3u_profile.id
@@ -1040,11 +1077,12 @@ class MultiWorkerVODConnectionManager:
                 except Exception as e:
                     logger.error(f"[{client_id}] Worker {self.worker_id} - Error in Redis-backed stream: {e}")
                     if not decremented:
-                        redis_connection.decrement_active_streams()
-                        decremented = True
+                        decremented, has_remaining = redis_connection.decrement_active_streams_and_check()
+                    else:
+                        has_remaining = redis_connection.has_active_streams()
 
                     # Decrement profile counter immediately if no other active streams
-                    if not redis_connection.has_active_streams():
+                    if not has_remaining:
                         state = redis_connection._get_connection_state()
                         profile_id = state.m3u_profile_id if state else m3u_profile.id
                         if profile_id:
@@ -1057,7 +1095,13 @@ class MultiWorkerVODConnectionManager:
 
                 finally:
                     if not decremented:
-                        redis_connection.decrement_active_streams()
+                        decremented, has_remaining = redis_connection.decrement_active_streams_and_check()
+                        if decremented and not has_remaining:
+                            state = redis_connection._get_connection_state()
+                            profile_id = state.m3u_profile_id if state else m3u_profile.id
+                            if profile_id:
+                                self._decrement_profile_connections(profile_id)
+                                logger.info(f"[{client_id}] Profile counter decremented for profile {profile_id} in finally block")
 
             # Create streaming response
             response = StreamingHttpResponse(

apps/proxy/vod_proxy/tests/test_profile_connections.py

@@ -0,0 +1,253 @@
+"""
+Tests for VOD proxy profile connection counter fixes.
+
+Covers three race conditions in multi_worker_connection_manager:
+  1. decrement_active_streams() return value was ignored — counter stuck on lock contention
+  2. Non-atomic GET-then-DECR in _decrement_profile_connections() — counter could go negative
+  3. has_active_streams() read without lock — race between decrement and check
+"""
+
+from unittest.mock import MagicMock, patch, call
+from django.test import TestCase
+
+
+class FakeRedis:
+    """Minimal in-memory Redis stand-in for counter tests."""
+
+    def __init__(self):
+        self._data = {}
+
+    def get(self, key):
+        val = self._data.get(key)
+        return str(val).encode() if val is not None else None
+
+    def set(self, key, value, ex=None):
+        self._data[key] = int(value)
+
+    def incr(self, key):
+        self._data[key] = self._data.get(key, 0) + 1
+        return self._data[key]
+
+    def decr(self, key):
+        self._data[key] = self._data.get(key, 0) - 1
+        return self._data[key]
+
+    def delete(self, key):
+        self._data.pop(key, None)
+
+    def exists(self, key):
+        return key in self._data
+
+    def pipeline(self):
+        return FakePipeline(self)
+
+
+class FakePipeline:
+    def __init__(self, redis):
+        self._redis = redis
+        self._cmds = []
+
+    def incr(self, key):
+        self._cmds.append(('incr', key))
+        return self
+
+    def decr(self, key):
+        self._cmds.append(('decr', key))
+        return self
+
+    def execute(self):
+        results = []
+        for cmd, key in self._cmds:
+            results.append(getattr(self._redis, cmd)(key))
+        self._cmds = []
+        return results
+
+
+class MultiWorkerManagerImportMixin:
+    """Mixin to import the manager class with patched Django/Redis deps."""
+
+    @classmethod
+    def get_manager_class(cls):
+        import importlib
+        import sys
+
+        # Stub out heavy Django deps so we can import the module standalone
+        for mod in ['apps.vod.models', 'apps.m3u.models', 'core.utils']:
+            if mod not in sys.modules:
+                sys.modules[mod] = MagicMock()
+
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import (
+            MultiWorkerVODConnectionManager,
+            RedisBackedVODConnection,
+        )
+        return MultiWorkerVODConnectionManager, RedisBackedVODConnection
+
+
+class TestDecrementProfileConnectionsAtomic(TestCase):
+    """Bug 2: _decrement_profile_connections must be atomic (no GET-then-DECR)."""
+
+    def _make_manager(self, redis):
+        _, _ = MultiWorkerManagerImportMixin.get_manager_class()
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import MultiWorkerVODConnectionManager
+        mgr = MultiWorkerVODConnectionManager.__new__(MultiWorkerVODConnectionManager)
+        mgr.redis_client = redis
+        mgr.worker_id = 'test-worker'
+        return mgr
+
+    def test_decrement_does_not_go_negative(self):
+        """Counter must be clamped to 0, never go negative."""
+        redis = FakeRedis()
+        redis.set('profile_connections:1', 0)
+        mgr = self._make_manager(redis)
+
+        result = mgr._decrement_profile_connections(1)
+
+        self.assertEqual(result, 0)
+        self.assertEqual(int(redis._data.get('profile_connections:1', 0)), 0)
+
+    def test_decrement_from_one_reaches_zero(self):
+        """Normal single decrement should reach 0."""
+        redis = FakeRedis()
+        redis.set('profile_connections:1', 1)
+        mgr = self._make_manager(redis)
+
+        result = mgr._decrement_profile_connections(1)
+
+        self.assertEqual(result, 0)
+
+    def test_concurrent_decrements_clamp_to_zero(self):
+        """Two concurrent decrements of a counter at 1 must not leave it at -1."""
+        redis = FakeRedis()
+        redis.set('profile_connections:1', 1)
+        mgr = self._make_manager(redis)
+
+        # Simulate two concurrent decrements (both fire before either reads back)
+        mgr._decrement_profile_connections(1)
+        mgr._decrement_profile_connections(1)
+
+        final = int(redis._data.get('profile_connections:1', 0))
+        self.assertGreaterEqual(final, 0, "Counter must not go negative after concurrent decrements")
+
+
+class TestDecrementActiveStreamsAndCheck(TestCase):
+    """Bug 1 & 3: decrement_active_streams_and_check() must be atomic."""
+
+    def _make_connection(self, redis, session_id='test-session'):
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import RedisBackedVODConnection
+        conn = RedisBackedVODConnection.__new__(RedisBackedVODConnection)
+        conn.session_id = session_id
+        conn.redis_client = redis
+        conn.connection_key = f'vod_connection:{session_id}'
+        conn.lock_key = f'vod_lock:{session_id}'
+        conn.local_session = None
+        conn._lock_acquired = False
+        return conn
+
+    def _make_state(self, active_streams=1, profile_id=7):
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import SerializableConnectionState
+        state = SerializableConnectionState.__new__(SerializableConnectionState)
+        state.session_id = 'test-session'
+        state.stream_url = 'http://example.com/stream.mkv'
+        state.headers = {}
+        state.m3u_profile_id = profile_id
+        state.active_streams = active_streams
+        state.last_activity = 0
+        state.worker_id = 'test-worker'
+        state.content_type = None
+        state.content_length = None
+        state.final_url = None
+        state.request_count = 0
+        state.bytes_sent = 0
+        state.content_obj_type = None
+        state.content_uuid = None
+        state.content_name = None
+        state.client_ip = None
+        state.client_user_agent = None
+        state.utc_start = None
+        state.utc_end = None
+        state.offset = None
+        state.connection_type = 'redis'
+        state.created_at = 0
+        return state
+
+    def test_returns_success_and_no_remaining_when_last_stream(self):
+        """When active_streams goes 1->0, should return (True, False)."""
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import RedisBackedVODConnection
+        conn = MagicMock(spec=RedisBackedVODConnection)
+        conn.session_id = 'test'
+
+        state = MagicMock()
+        state.active_streams = 1
+
+        conn._acquire_lock.return_value = True
+        conn._get_connection_state.return_value = state
+        conn._save_connection_state.return_value = True
+        conn._release_lock.return_value = None
+
+        # Call the real method on the mock instance
+        result = RedisBackedVODConnection.decrement_active_streams_and_check(conn)
+
+        self.assertEqual(result, (True, False))
+        self.assertEqual(state.active_streams, 0)
+
+    def test_returns_success_and_remaining_when_other_streams_active(self):
+        """When active_streams goes 2->1, should return (True, True)."""
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import RedisBackedVODConnection
+        conn = MagicMock(spec=RedisBackedVODConnection)
+        conn.session_id = 'test'
+
+        state = MagicMock()
+        state.active_streams = 2
+
+        conn._acquire_lock.return_value = True
+        conn._get_connection_state.return_value = state
+        conn._save_connection_state.return_value = True
+        conn._release_lock.return_value = None
+
+        result = RedisBackedVODConnection.decrement_active_streams_and_check(conn)
+
+        self.assertEqual(result, (True, True))
+        self.assertEqual(state.active_streams, 1)
+
+    def test_returns_failure_and_assumes_remaining_on_lock_contention(self):
+        """Lock contention must return (False, True) — assume streams remain to be safe."""
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import RedisBackedVODConnection
+        conn = MagicMock(spec=RedisBackedVODConnection)
+        conn.session_id = 'test'
+        conn._acquire_lock.return_value = False
+
+        result = RedisBackedVODConnection.decrement_active_streams_and_check(conn)
+
+        self.assertEqual(result, (False, True))
+        conn._get_connection_state.assert_not_called()
+
+    def test_returns_failure_when_already_at_zero(self):
+        """When active_streams is already 0, should return (False, False)."""
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import RedisBackedVODConnection
+        conn = MagicMock(spec=RedisBackedVODConnection)
+        conn.session_id = 'test'
+
+        state = MagicMock()
+        state.active_streams = 0
+
+        conn._acquire_lock.return_value = True
+        conn._get_connection_state.return_value = state
+        conn._release_lock.return_value = None
+
+        result = RedisBackedVODConnection.decrement_active_streams_and_check(conn)
+
+        self.assertEqual(result, (False, False))
+        conn._save_connection_state.assert_not_called()
+
+    def test_lock_always_released_even_on_exception(self):
+        """Lock must be released even if an exception occurs inside."""
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import RedisBackedVODConnection
+        conn = MagicMock(spec=RedisBackedVODConnection)
+        conn.session_id = 'test'
+        conn._acquire_lock.return_value = True
+        conn._get_connection_state.side_effect = RuntimeError("Redis exploded")
+
+        with self.assertRaises(RuntimeError):
+            RedisBackedVODConnection.decrement_active_streams_and_check(conn)
+
+        conn._release_lock.assert_called_once()