fix(vod): prevent stale profile connection reservations in Redis-backed VOD sessions (Fixes #1426)

This update introduces atomic mutations for active_streams using Redis Lua scripts, ensuring that concurrent byte-range requests do not leave stale session metadata. The changes also include improvements to the session state saving mechanism, preventing the recreation of zombie sessions after idle cleanup. Additionally, tests have been added to cover the new atomic behavior and ensure reliability across various scenarios.

CHANGELOG.md

@@ -45,6 +45,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Redis-backed VOD sessions no longer leave stale profile connection reservations after multi-worker range-request teardown.** Jellyfin/FFmpeg clients that issue concurrent byte-range requests for one VOD session could finish teardown while another worker held the session metadata lock (`vod_connection_lock:*`). `decrement_active_streams_and_check()` then failed with `DECR-AS-CHECK failed: could not acquire lock`, skipped the profile-slot release, and left `profile_connections:*` and the session hash occupied until restart or manual Redis cleanup. Per-session `active_streams` is now mutated with Redis Lua (independent of the metadata lock), metadata saves omit and never clobber that counter, idle cleanup deletes the session hash only when still idle, and late metadata writes cannot recreate a deleted session. (Fixes #1426)
 - **EPG programme times no longer shift when PostgreSQL's server default timezone is non-UTC.** After the psycopg3 / Django upgrade, geventpool sessions were no longer pinned to UTC: Django's connection timezone setup is skipped whenever `self.pool` is truthy, and the older `connection_created` `SET TIME ZONE 'UTC0'` receiver was a nested closure registered with a weak reference, so it did not reliably stay registered (in production with `DEBUG=False` it was garbage-collected and never ran; `DEBUG=True` could keep it alive via Django's inspect LRU cache). Sessions that lacked a live pin therefore inherited the server default; on deployments whose default is non-UTC (for example from `/etc/localtime` bind-mounts), psycopg3 decoded `timestamptz` values under that zone and XMLTV output labeled local wall-clock times as `+0000`. The pool backend now sets `TimeZone=UTC` in the libpq startup packet so every pooled connection starts in UTC (and `RESET TimeZone` returns to UTC), and the fragile signal is removed. Stored data was never corrupted, only reads were wrong. (Fixes #651) — Thanks [@nagelm](https://github.com/nagelm)
 - **Frontend date/time unit tests no longer fail outside UTC / en-US.** Three tests encoded the machine timezone or locale into their expectations (`dateTimeUtils.isSame`, `RecordingUtils.toDateString`, `SeriesModalUtils.getEpisodeAirdate`), so the suite failed on boxes east of UTC or non-US locales. Fixtures now use local calendar datetimes and locale-computed expectations so the suite passes in every environment. — Thanks [@nagelm](https://github.com/nagelm)
 - **API error toasts no longer dump raw HTML error pages.** Failed responses that return Django or nginx HTML (for example 500/502/504 when the backend is down) were interpolated into the toast body as markup. Errors are now formatted for display: JSON bodies prefer `detail` / `error` / field messages, HTML and empty bodies collapse to a short status line, and long plain-text bodies are truncated. — Thanks [@nagelm](https://github.com/nagelm)

apps/proxy/vod_proxy/multi_worker_connection_manager.py

@@ -6,13 +6,8 @@ import time
 import json
 import logging
 import threading
-import random
-import re
 import requests
-import pickle
-import base64
 import os
-import socket
 import mimetypes
 from urllib.parse import urlparse
 from typing import Optional, Dict, Any
@@ -23,6 +18,75 @@ from apps.m3u.models import M3UAccountProfile
 
 logger = logging.getLogger("vod_proxy")
 
+# Atomic active_streams mutations. These run as Redis Lua so INCR/DECR never
+# contend with the session metadata lock used for ownership, seek info, and
+# get_stream header updates. One EVALSHA round-trip each.
+_LUA_INCR_ACTIVE_STREAMS = """
+-- vod_incr_as
+local key = KEYS[1]
+local activity = ARGV[1]
+if redis.call('EXISTS', key) == 0 then
+  return 0
+end
+local new_count = redis.call('HINCRBY', key, 'active_streams', 1)
+redis.call('HSET', key, 'last_activity', activity)
+return new_count
+"""
+
+_LUA_DECR_ACTIVE_STREAMS = """
+-- vod_decr_as
+local key = KEYS[1]
+local activity = ARGV[1]
+if redis.call('EXISTS', key) == 0 then
+  return {-1, 0}
+end
+local current = tonumber(redis.call('HGET', key, 'active_streams') or '0')
+if current <= 0 then
+  return {0, 0}
+end
+local new_count = current - 1
+redis.call('HSET', key, 'active_streams', new_count, 'last_activity', activity)
+return {1, new_count}
+"""
+
+_LUA_CLEANUP_IF_IDLE = """
+-- vod_cleanup_idle
+local conn_key = KEYS[1]
+if redis.call('EXISTS', conn_key) == 0 then
+  return 1
+end
+local current = tonumber(redis.call('HGET', conn_key, 'active_streams') or '0')
+if current > 0 then
+  return 0
+end
+-- Delete only the session hash. Do not DEL the metadata lock: another worker
+-- may hold it for ownership/seek updates, and removing it would let a late
+-- HSET recreate a zombie session without a coherent active_streams field.
+redis.call('DEL', conn_key)
+return 1
+"""
+
+# Metadata HSET only when the session hash still exists. Prevents a worker that
+# held the metadata lock across idle cleanup from recreating a zombie hash.
+_LUA_META_SAVE_IF_EXISTS = """
+-- vod_meta_save_if_exists
+local key = KEYS[1]
+local ttl = tonumber(ARGV[1])
+if redis.call('EXISTS', key) == 0 then
+  return 0
+end
+for i = 2, #ARGV, 2 do
+  redis.call('HSET', key, ARGV[i], ARGV[i + 1])
+end
+if ttl and ttl > 0 then
+  redis.call('EXPIRE', key, ttl)
+end
+return 1
+"""
+
+# Cache register_script handles per redis client (EVALSHA thereafter).
+_vod_script_cache: Dict[int, Dict[str, Any]] = {}
+
 
 def get_vod_client_stop_key(client_id):
     """Get the Redis key for signaling a VOD client to stop"""
@@ -234,13 +298,26 @@ class RedisBackedVODConnection:
             logger.error(f"[{self.session_id}] Error getting connection state from Redis: {e}")
             return None
 
-    def _save_connection_state(self, state: SerializableConnectionState):
-        """Save connection state to Redis"""
+    def _save_connection_state(self, state: SerializableConnectionState,
+                               include_active_streams: bool = False):
+        """Save connection state to Redis.
+
+        By default omits ``active_streams`` so metadata writes (ownership,
+        seek info, get_stream headers) cannot clobber the atomic counter
+        maintained by ``increment_active_streams`` / ``decrement_*``.
+        Pass ``include_active_streams=True`` only when creating a new session.
+
+        Metadata saves are applied only if the session hash still exists, so a
+        late write after idle cleanup cannot recreate a zombie session.
+        """
         if not self.redis_client:
             return False
 
         try:
             data = state.to_dict()
+            if not include_active_streams:
+                data.pop('active_streams', None)
+
             # Log the data being saved for debugging
             logger.trace(f"[{self.session_id}] Saving connection state: {data}")
 
@@ -250,20 +327,46 @@ class RedisBackedVODConnection:
                     logger.error(f"[{self.session_id}] None value found for key '{key}' - this should not happen")
                     return False
 
-            self.redis_client.hset(self.connection_key, mapping=data)
-            self.redis_client.expire(self.connection_key, 3600)  # 1 hour TTL
+            if include_active_streams:
+                # Session creation: key may not exist yet.
+                self.redis_client.hset(self.connection_key, mapping=data)
+                self.redis_client.expire(self.connection_key, 3600)
+                return True
+
+            # Flat field/value list for Lua: TTL, then pairs
+            args = ['3600']
+            for field, value in data.items():
+                args.extend([str(field), str(value)])
+
+            saved = int(
+                self._vod_scripts()['meta_save'](
+                    keys=[self.connection_key],
+                    args=args,
+                )
+                or 0
+            )
+            if not saved:
+                logger.debug(
+                    f"[{self.session_id}] Skipped metadata save; session no longer exists"
+                )
+                return False
             return True
         except Exception as e:
             logger.error(f"[{self.session_id}] Error saving connection state to Redis: {e}")
             return False
 
     def _acquire_lock(self, timeout: int = 10) -> bool:
-        """Acquire distributed lock for connection operations"""
+        """Acquire distributed lock for session metadata operations.
+
+        Not used for active_streams accounting; that is Redis-atomic via Lua.
+        """
         if not self.redis_client:
             return False
 
         try:
-            return self.redis_client.set(self.lock_key, "locked", nx=True, ex=timeout)
+            return bool(
+                self.redis_client.set(self.lock_key, "locked", nx=True, ex=timeout)
+            )
         except Exception as e:
             logger.error(f"[{self.session_id}] Error acquiring lock: {e}")
             return False
@@ -278,6 +381,21 @@ class RedisBackedVODConnection:
         except Exception as e:
             logger.error(f"[{self.session_id}] Error releasing lock: {e}")
 
+    def _vod_scripts(self) -> Dict[str, Any]:
+        """Lazily register Lua scripts once per Redis client (EVALSHA thereafter)."""
+        client = self.redis_client
+        cache_key = id(client)
+        cached = _vod_script_cache.get(cache_key)
+        if cached is None:
+            cached = {
+                'incr': client.register_script(_LUA_INCR_ACTIVE_STREAMS),
+                'decr': client.register_script(_LUA_DECR_ACTIVE_STREAMS),
+                'cleanup': client.register_script(_LUA_CLEANUP_IF_IDLE),
+                'meta_save': client.register_script(_LUA_META_SAVE_IF_EXISTS),
+            }
+            _vod_script_cache[cache_key] = cached
+        return cached
+
     def create_connection(self, stream_url: str, headers: dict, m3u_profile_id: int = None,
                          # Session metadata (consolidated from vod_session key)
                          content_obj_type: str = None, content_uuid: str = None,
@@ -315,7 +433,8 @@ class RedisBackedVODConnection:
                 worker_id=worker_id,
                 user_id=user.id if user else "unknown"
             )
-            success = self._save_connection_state(state)
+            # Seed active_streams=0 once; later mutations are Lua HINCRBY only.
+            success = self._save_connection_state(state, include_active_streams=True)
 
             if success:
                 logger.info(f"[{self.session_id}] Created new connection state in Redis with consolidated session metadata")
@@ -438,22 +557,17 @@ class RedisBackedVODConnection:
 
                 logger.info(f"[{self.session_id}] Updated connection state: length={state.content_length}, type={state.content_type}")
 
-            # Save updated state under lock to avoid overwriting concurrent
-            # active_streams changes (e.g., another stream's GeneratorExit decrement)
+            # Metadata-only save: active_streams is omitted so concurrent Lua
+            # INCR/DECR cannot be overwritten by this get_stream header update.
             if self._acquire_lock():
                 try:
-                    current = self._get_connection_state()
-                    if current:
-                        # Preserve the current active_streams value — it may have been
-                        # modified by concurrent increment/decrement operations while
-                        # waiting for the upstream HTTP response.
-                        state.active_streams = current.active_streams
                     self._save_connection_state(state)
                 finally:
                     self._release_lock()
             else:
-                # Fallback: save without lock but skip active_streams to avoid overwrite
+                # Still persist headers without the lock; counter field is excluded.
                 logger.warning(f"[{self.session_id}] Could not acquire lock for get_stream state save")
+                self._save_connection_state(state)
 
             self.local_response = response
             return response
@@ -502,84 +616,96 @@ class RedisBackedVODConnection:
             return range_header
 
     def increment_active_streams(self):
-        """Increment active streams count in Redis. Returns new active_streams count, or 0 on failure."""
-        if not self._acquire_lock():
-            logger.warning(f"[{self.session_id}] INCR-AS failed: could not acquire lock")
+        """Atomically increment active_streams via Redis Lua (no session lock).
+
+        Returns new active_streams count, or 0 on failure / missing session.
+        """
+        if not self.redis_client:
             return 0
 
         try:
-            state = self._get_connection_state()
-            if state:
-                old = state.active_streams
-                state.active_streams += 1
-                state.last_activity = time.time()
-                self._save_connection_state(state)
-                logger.debug(f"[{self.session_id}] INCR-AS {old} -> {state.active_streams}")
-                return state.active_streams
-            logger.warning(f"[{self.session_id}] INCR-AS failed: no state")
+            new_count = int(
+                self._vod_scripts()['incr'](
+                    keys=[self.connection_key],
+                    args=[str(time.time())],
+                )
+                or 0
+            )
+            if new_count <= 0:
+                logger.warning(f"[{self.session_id}] INCR-AS failed: no state")
+                return 0
+            logger.debug(f"[{self.session_id}] INCR-AS -> {new_count}")
+            return new_count
+        except Exception as e:
+            logger.error(f"[{self.session_id}] INCR-AS failed: {e}")
             return 0
-        finally:
-            self._release_lock()
 
     def decrement_active_streams(self):
-        """Decrement active streams count in Redis"""
-        if not self._acquire_lock():
-            logger.warning(f"[{self.session_id}] DECR-AS failed: could not acquire lock")
-            return False
-
-        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
-            if not state:
-                logger.warning(f"[{self.session_id}] DECR-AS failed: no state")
-            else:
-                logger.warning(f"[{self.session_id}] DECR-AS failed: active_streams already {state.active_streams}")
-            return False
-        finally:
-            self._release_lock()
+        """Atomically decrement active_streams via Redis Lua (no session lock)."""
+        success, _has_remaining = self.decrement_active_streams_and_check()
+        return success
 
     def decrement_active_streams_and_check(self):
-        """Atomically decrement active streams and return (success, has_remaining_streams).
+        """Atomically decrement active_streams and report whether any remain.
 
-        Combines decrement + check under a single lock to eliminate the race window
-        between separate decrement_active_streams() and has_active_streams() calls.
+        Uses a single Redis Lua script so the decrement + zero-check cannot
+        race with other workers, and does not take the session metadata lock
+        (so concurrent range-request metadata updates cannot block teardown).
 
         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)
+            (False, False) - no state, or active_streams already 0
         """
-        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
+        if not self.redis_client:
+            return False, False
 
         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:
+            result = self._vod_scripts()['decr'](
+                keys=[self.connection_key],
+                args=[str(time.time())],
+            )
+            # redis-py returns [success_flag, new_count]
+            success_flag = int(result[0])
+            new_count = int(result[1])
+
+            if success_flag < 0:
                 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()
+            if success_flag == 0:
+                logger.warning(
+                    f"[{self.session_id}] DECR-AS-CHECK failed: active_streams already 0"
+                )
+                return False, False
+
+            logger.debug(f"[{self.session_id}] DECR-AS -> {new_count}")
+            return True, new_count > 0
+        except Exception as e:
+            logger.error(f"[{self.session_id}] DECR-AS-CHECK failed: {e}")
+            # Fail closed on remaining=True only for unexpected errors so we do
+            # not double-release the profile slot; callers still log loudly.
+            return False, True
 
     def has_active_streams(self) -> bool:
-        """Check if connection has any active streams"""
-        state = self._get_connection_state()
-        return state.active_streams > 0 if state else False
+        """Check if connection has any active streams (single HGET, not full hash)."""
+        if not self.redis_client:
+            return False
+        try:
+            val = self.redis_client.hget(self.connection_key, 'active_streams')
+            return int(val or 0) > 0
+        except Exception as e:
+            logger.error(f"[{self.session_id}] Error reading active_streams: {e}")
+            return False
+
+    def get_active_streams_count(self) -> int:
+        """Return current active_streams from Redis (0 if missing)."""
+        if not self.redis_client:
+            return 0
+        try:
+            val = self.redis_client.hget(self.connection_key, 'active_streams')
+            return int(val or 0)
+        except Exception:
+            return 0
 
     def get_headers(self):
         """Get headers for response"""
@@ -639,48 +765,35 @@ class RedisBackedVODConnection:
             logger.info(f"[{self.session_id}] No connection state found - local cleanup only")
             return
 
-        # Check if there are active streams
+        # Fast path: active streams present; leave Redis session intact
         if state.active_streams > 0:
-            # There are active streams - check ownership
             if current_worker_id and state.worker_id == current_worker_id:
                 logger.info(f"[{self.session_id}] Active streams present ({state.active_streams}) and we own them - local cleanup only")
             else:
                 logger.info(f"[{self.session_id}] Active streams present ({state.active_streams}) but owned by worker {state.worker_id} - local cleanup only")
             return
 
-        # No active streams - we can clean up Redis state
+        # No active streams: atomically delete only if still idle (Lua).
+        # Avoids racing a reconnect INCR between HGET and DEL without holding
+        # the metadata lock across the whole check.
         if not self.redis_client:
             logger.info(f"[{self.session_id}] No Redis client - local cleanup only")
             return
 
-        # Acquire lock and do final check before cleanup to prevent race conditions
-        if not self._acquire_lock():
-            logger.warning(f"[{self.session_id}] Could not acquire lock for cleanup - skipping")
-            return
-
         try:
-            # Re-check active streams with lock held to prevent race conditions
-            current_state = self._get_connection_state()
-            if not current_state:
-                logger.info(f"[{self.session_id}] Connection state no longer exists - cleanup already done")
+            deleted = int(
+                self._vod_scripts()['cleanup'](
+                    keys=[self.connection_key],
+                    args=[],
+                )
+                or 0
+            )
+            if not deleted:
+                logger.info(
+                    f"[{self.session_id}] Active streams present during cleanup - skipping Redis delete"
+                )
                 return
 
-            if current_state.active_streams > 0:
-                logger.info(f"[{self.session_id}] Active streams now present ({current_state.active_streams}) - skipping cleanup")
-                return
-
-            # Use pipeline for atomic cleanup operations
-            pipe = self.redis_client.pipeline()
-
-            # 1. Remove main connection state (contains consolidated data)
-            pipe.delete(self.connection_key)
-
-            # 2. Remove distributed lock (will be released below anyway)
-            pipe.delete(self.lock_key)
-
-            # Execute all cleanup operations
-            pipe.execute()
-
             logger.info(f"[{self.session_id}] Cleaned up Redis keys (verified no active streams)")
 
             # Decrement profile connections if we have the state and connection manager
@@ -695,9 +808,6 @@ class RedisBackedVODConnection:
 
         except Exception as e:
             logger.error(f"[{self.session_id}] Error cleaning up Redis state: {e}")
-        finally:
-            # Always release the lock
-            self._release_lock()
 
 
 # Modify the VODConnectionManager to use Redis-backed connections
@@ -853,6 +963,7 @@ class MultiWorkerVODConnectionManager:
         # Track whether we incremented profile connections (for cleanup on error)
         profile_connections_incremented = False
         redis_connection = None
+        existing_state = None
 
         logger.info(f"[{client_id}] Worker {self.worker_id} - Redis-backed streaming request for {content_type} {content_name}")
 
@@ -955,7 +1066,7 @@ class MultiWorkerVODConnectionManager:
                 # Without this, stream's GeneratorExit can see active_streams=0
                 # and DECR the profile counter before the new generator starts.
                 if matching_session_id:
-                    # Idle session reuse: active_streams already incremented at line 776
+                    # Idle session reuse: active_streams already incremented above
                     # Always need to re-reserve profile slot (GeneratorExit DECRed it)
                     if not self._check_and_reserve_profile_slot(m3u_profile):
                         logger.warning(f"[{client_id}] Profile {m3u_profile.name} connection limit exceeded on session reuse")
@@ -977,7 +1088,7 @@ class MultiWorkerVODConnectionManager:
                         logger.error(f"[{client_id}] Failed to increment active streams")
                         return HttpResponse("Failed to reserve stream", status=500)
                     # else: new_count > 1, another stream is already active and profile
-                    # counter already reflects it — no INCR needed
+                    # counter already reflects it; no INCR needed
 
                 # Transfer ownership to current worker and update session activity
                 if redis_connection._acquire_lock():
@@ -1082,7 +1193,7 @@ class MultiWorkerVODConnectionManager:
 
                     # Schedule smart cleanup if no active streams after normal completion
                     if stream_decremented and not has_remaining and not profile_decremented:
-                        # Decrement profile counter immediately — don't defer to daemon thread
+                        # 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
                         if profile_id:
@@ -1115,9 +1226,9 @@ class MultiWorkerVODConnectionManager:
                     else:
                         has_remaining = redis_connection.has_active_streams()
 
-                    # Schedule smart cleanup if no active streams
-                    if not has_remaining and not profile_decremented:
-                        # Decrement profile counter immediately — don't defer to daemon thread
+                    # Schedule smart cleanup if this stream's DECR left none remaining
+                    if stream_decremented and not has_remaining and not profile_decremented:
+                        # 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
                         if profile_id:
@@ -1150,8 +1261,8 @@ class MultiWorkerVODConnectionManager:
                     else:
                         has_remaining = redis_connection.has_active_streams()
 
-                    # Decrement profile counter immediately if no other active streams
-                    if not has_remaining and not profile_decremented:
+                    # Decrement profile counter only when this stream's DECR hit zero
+                    if stream_decremented and not has_remaining and not profile_decremented:
                         state = redis_connection._get_connection_state()
                         profile_id = state.m3u_profile_id if state else m3u_profile.id
                         if profile_id:
@@ -1159,7 +1270,7 @@ class MultiWorkerVODConnectionManager:
                             profile_decremented = True
                             logger.info(f"[{client_id}] Profile counter decremented for profile {profile_id} on stream error")
                         # Smart cleanup on error - immediate cleanup since we're in error state
-                        # No connection_manager — profile already decremented above
+                        # No connection_manager; profile already decremented above
                         redis_connection.cleanup(current_worker_id=self.worker_id)
                     yield b"Error: Stream interrupted"
 
@@ -1176,13 +1287,13 @@ class MultiWorkerVODConnectionManager:
 
                             # Delayed cleanup: wait 1s for seeking clients to reconnect
                             # before closing the provider connection and Redis keys.
-                            # cleanup() re-checks active_streams under lock, so a
+                            # cleanup() atomically re-checks active_streams (Lua), so a
                             # reconnecting client that increments active_streams in
                             # time will prevent Redis key deletion.
                             def delayed_cleanup():
                                 time.sleep(1)
                                 logger.info(f"[{client_id}] Worker {self.worker_id} - Checking for smart cleanup in finally block")
-                                # No connection_manager — profile already decremented above
+                                # No connection_manager; profile already decremented above
                                 redis_connection.cleanup(current_worker_id=self.worker_id)
 
                             cleanup_thread = threading.Thread(target=delayed_cleanup)
@@ -1278,6 +1389,13 @@ class MultiWorkerVODConnectionManager:
         except Exception as e:
             logger.error(f"[{client_id}] Worker {self.worker_id} - Error in Redis-backed stream_content_with_session: {e}", exc_info=True)
 
+            # Roll back stream reservation if we incremented before get_stream failed
+            if existing_state and redis_connection:
+                try:
+                    redis_connection.decrement_active_streams()
+                except Exception as decr_error:
+                    logger.error(f"[{client_id}] Error rolling back active_streams after connection failure: {decr_error}")
+
             # Decrement profile connections if we incremented them but failed before streaming started
             if profile_connections_incremented:
                 logger.info(f"[{client_id}] Connection error occurred after profile increment - decrementing profile connections")

apps/proxy/vod_proxy/tests/test_profile_connections.py

@@ -1,10 +1,9 @@
 """
 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
+Covers:
+  1. Atomic active_streams DECR+check via Redis Lua (no session-lock gating)
+  2. Non-atomic GET-then-DECR in _decrement_profile_connections() (counter could go negative)
 """
 
 from unittest.mock import MagicMock, patch, call
@@ -130,124 +129,72 @@ class TestDecrementProfileConnectionsAtomic(TestCase):
 
 
 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
+    """Atomic DECR+check via Redis Lua (no session lock)."""
 
     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'
+        from apps.proxy.vod_proxy.tests.test_vod_lock_contention import (
+            LockAwareFakeRedis,
+            _seed_session,
+            _import_vod,
+            _clear_script_cache,
+        )
 
-        state = MagicMock()
-        state.active_streams = 1
+        RedisBackedVODConnection, _ = _import_vod()
+        redis = LockAwareFakeRedis()
+        _seed_session(redis, "prof-last", active_streams=1)
+        conn = RedisBackedVODConnection("prof-last", redis)
 
-        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)
+        result = conn.decrement_active_streams_and_check()
 
         self.assertEqual(result, (True, False))
-        self.assertEqual(state.active_streams, 0)
+        self.assertEqual(conn.get_active_streams_count(), 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'
+        from apps.proxy.vod_proxy.tests.test_vod_lock_contention import (
+            LockAwareFakeRedis,
+            _seed_session,
+            _import_vod,
+        )
 
-        state = MagicMock()
-        state.active_streams = 2
+        RedisBackedVODConnection, _ = _import_vod()
+        redis = LockAwareFakeRedis()
+        _seed_session(redis, "prof-rem", active_streams=2)
+        conn = RedisBackedVODConnection("prof-rem", redis)
 
-        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)
+        result = conn.decrement_active_streams_and_check()
 
         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()
+        self.assertEqual(conn.get_active_streams_count(), 1)
 
     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'
+        from apps.proxy.vod_proxy.tests.test_vod_lock_contention import (
+            LockAwareFakeRedis,
+            _seed_session,
+            _import_vod,
+        )
 
-        state = MagicMock()
-        state.active_streams = 0
+        RedisBackedVODConnection, _ = _import_vod()
+        redis = LockAwareFakeRedis()
+        _seed_session(redis, "prof-zero", active_streams=0)
+        conn = RedisBackedVODConnection("prof-zero", redis)
 
-        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)
+        result = conn.decrement_active_streams_and_check()
 
         self.assertEqual(result, (False, False))
-        conn._save_connection_state.assert_not_called()
+        self.assertEqual(conn.get_active_streams_count(), 0)
 
-    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")
+    def test_returns_failure_when_no_session(self):
+        from apps.proxy.vod_proxy.tests.test_vod_lock_contention import (
+            LockAwareFakeRedis,
+            _import_vod,
+            _clear_script_cache,
+        )
 
-        with self.assertRaises(RuntimeError):
-            RedisBackedVODConnection.decrement_active_streams_and_check(conn)
+        RedisBackedVODConnection, _ = _import_vod()
+        _clear_script_cache()
+        redis = LockAwareFakeRedis()
+        conn = RedisBackedVODConnection("missing", redis)
 
-        conn._release_lock.assert_called_once()
+        result = conn.decrement_active_streams_and_check()
+
+        self.assertEqual(result, (False, False))

apps/proxy/vod_proxy/tests/test_vod_lock_contention.py

@@ -0,0 +1,604 @@
+"""
+Tests for atomic VOD active_streams accounting.
+
+active_streams is mutated via Redis Lua (HINCRBY / conditional DECR), not the
+session metadata lock. These tests simulate multi-worker Jellyfin-style range
+request churn: metadata lock held while another worker tears down, concurrent
+DECRs, and metadata saves that must not clobber the counter.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from unittest.mock import MagicMock
+
+from django.test import SimpleTestCase
+
+
+class LockAwareFakeRedis:
+    """In-memory Redis with SET NX, hashes, and VOD Lua script shims."""
+
+    def __init__(self):
+        self._data = {}
+        self._hashes = {}
+        self._lock = threading.Lock()
+
+    def set(self, key, value, nx=False, ex=None):
+        with self._lock:
+            if nx and key in self._data:
+                return False
+            self._data[key] = value
+            return True
+
+    def get(self, key):
+        with self._lock:
+            val = self._data.get(key)
+            if val is None:
+                return None
+            return str(val)
+
+    def delete(self, *keys):
+        with self._lock:
+            deleted = 0
+            for key in keys:
+                if key in self._data:
+                    del self._data[key]
+                    deleted += 1
+                if key in self._hashes:
+                    del self._hashes[key]
+                    deleted += 1
+            return deleted
+
+    def exists(self, key):
+        with self._lock:
+            return int(key in self._data or key in self._hashes)
+
+    def incr(self, key):
+        with self._lock:
+            self._data[key] = int(self._data.get(key, 0)) + 1
+            return self._data[key]
+
+    def decr(self, key):
+        with self._lock:
+            self._data[key] = int(self._data.get(key, 0)) - 1
+            return self._data[key]
+
+    def hset(self, key, mapping=None, **kwargs):
+        with self._lock:
+            if key not in self._hashes:
+                self._hashes[key] = {}
+            if mapping:
+                for k, v in mapping.items():
+                    self._hashes[key][str(k)] = str(v)
+            # field/value form: hset(key, field, value) via redis-py kwargs uncommon;
+            # support positional through mapping only.
+            return True
+
+    def hget(self, key, field):
+        with self._lock:
+            return self._hashes.get(key, {}).get(str(field))
+
+    def hgetall(self, key):
+        with self._lock:
+            return dict(self._hashes.get(key, {}))
+
+    def hincrby(self, key, field, amount=1):
+        with self._lock:
+            if key not in self._hashes:
+                self._hashes[key] = {}
+            cur = int(self._hashes[key].get(str(field), 0))
+            cur += int(amount)
+            self._hashes[key][str(field)] = str(cur)
+            return cur
+
+    def expire(self, key, seconds):
+        return True
+
+    def pipeline(self):
+        return _FakePipeline(self)
+
+    def register_script(self, script):
+        return _FakeVodScript(self, script)
+
+
+class _FakePipeline:
+    def __init__(self, redis):
+        self._redis = redis
+        self._cmds = []
+
+    def delete(self, *keys):
+        self._cmds.append(("delete", keys))
+        return self
+
+    def execute(self):
+        results = []
+        for cmd, args in self._cmds:
+            if cmd == "delete":
+                results.append(self._redis.delete(*args))
+        self._cmds = []
+        return results
+
+
+class _FakeVodScript:
+    """Python stand-in for VOD Lua scripts (atomic under threading.Lock)."""
+
+    def __init__(self, redis: LockAwareFakeRedis, script: str):
+        self._redis = redis
+        self._script = script
+
+    def __call__(self, keys=None, args=None):
+        keys = keys or []
+        args = args or []
+        with self._redis._lock:
+            if "vod_incr_as" in self._script:
+                return self._incr(keys[0], args[0])
+            if "vod_decr_as" in self._script:
+                return self._decr(keys[0], args[0])
+            if "vod_cleanup_idle" in self._script:
+                return self._cleanup(keys[0])
+            if "vod_meta_save_if_exists" in self._script:
+                return self._meta_save(keys[0], args)
+            raise AssertionError(f"Unknown VOD script: {self._script[:80]}")
+
+    def _incr(self, key, activity):
+        if key not in self._redis._hashes:
+            return 0
+        h = self._redis._hashes[key]
+        new_count = int(h.get("active_streams", 0)) + 1
+        h["active_streams"] = str(new_count)
+        h["last_activity"] = str(activity)
+        return new_count
+
+    def _decr(self, key, activity):
+        if key not in self._redis._hashes:
+            return [-1, 0]
+        h = self._redis._hashes[key]
+        current = int(h.get("active_streams", 0))
+        if current <= 0:
+            return [0, 0]
+        new_count = current - 1
+        h["active_streams"] = str(new_count)
+        h["last_activity"] = str(activity)
+        return [1, new_count]
+
+    def _cleanup(self, conn_key, lock_key=None):
+        if conn_key not in self._redis._hashes:
+            return 1
+        current = int(self._redis._hashes[conn_key].get("active_streams", 0))
+        if current > 0:
+            return 0
+        del self._redis._hashes[conn_key]
+        return 1
+
+    def _meta_save(self, key, args):
+        if key not in self._redis._hashes:
+            return 0
+        # args[0] = ttl; args[1..] = field, value pairs
+        h = self._redis._hashes[key]
+        for i in range(1, len(args), 2):
+            if i + 1 >= len(args):
+                break
+            h[str(args[i])] = str(args[i + 1])
+        return 1
+
+
+def _clear_script_cache():
+    from apps.proxy.vod_proxy import multi_worker_connection_manager as mod
+
+    mod._vod_script_cache.clear()
+
+
+def _import_vod():
+    import sys
+
+    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 (
+        RedisBackedVODConnection,
+        SerializableConnectionState,
+    )
+
+    return RedisBackedVODConnection, SerializableConnectionState
+
+
+def _seed_session(redis, session_id, active_streams=1, profile_id=7):
+    RedisBackedVODConnection, SerializableConnectionState = _import_vod()
+    _clear_script_cache()
+    conn = RedisBackedVODConnection(session_id, redis)
+    state = SerializableConnectionState(
+        session_id=session_id,
+        stream_url="http://example.com/movie.mkv",
+        headers={},
+        m3u_profile_id=profile_id,
+    )
+    state.active_streams = active_streams
+    state.content_obj_type = "movie"
+    state.content_uuid = "abc-123"
+    state.content_name = "Test Movie"
+    state.worker_id = "worker-a"
+    assert conn._save_connection_state(state, include_active_streams=True)
+    return conn
+
+
+class TestAtomicActiveStreams(SimpleTestCase):
+    def test_incr_decr_round_trip(self):
+        RedisBackedVODConnection, _ = _import_vod()
+        redis = LockAwareFakeRedis()
+        _seed_session(redis, "vod_sym", active_streams=0)
+        a = RedisBackedVODConnection("vod_sym", redis)
+        b = RedisBackedVODConnection("vod_sym", redis)
+
+        self.assertEqual(a.increment_active_streams(), 1)
+        self.assertEqual(b.increment_active_streams(), 2)
+
+        ok_b, rem_b = b.decrement_active_streams_and_check()
+        ok_a, rem_a = a.decrement_active_streams_and_check()
+        self.assertTrue(ok_b and rem_b)
+        self.assertTrue(ok_a and not rem_a)
+        self.assertEqual(a.get_active_streams_count(), 0)
+
+    def test_decr_succeeds_while_session_metadata_lock_held(self):
+        """Teardown must not depend on the session metadata lock."""
+        RedisBackedVODConnection, _ = _import_vod()
+        redis = LockAwareFakeRedis()
+        holder = _seed_session(redis, "vod_lock_held", active_streams=1)
+        waiter = RedisBackedVODConnection("vod_lock_held", redis)
+
+        self.assertTrue(holder._acquire_lock())
+        try:
+            success, has_remaining = waiter.decrement_active_streams_and_check()
+            self.assertEqual((success, has_remaining), (True, False))
+            self.assertEqual(waiter.get_active_streams_count(), 0)
+        finally:
+            holder._release_lock()
+
+    def test_metadata_save_does_not_clobber_active_streams(self):
+        RedisBackedVODConnection, SerializableConnectionState = _import_vod()
+        redis = LockAwareFakeRedis()
+        conn = _seed_session(redis, "vod_clobber", active_streams=0)
+        self.assertEqual(conn.increment_active_streams(), 1)
+        self.assertEqual(conn.increment_active_streams(), 2)
+
+        # Stale in-memory state still has active_streams=0 from seed
+        stale = conn._get_connection_state()
+        stale.active_streams = 0
+        stale.worker_id = "worker-b"
+        conn._save_connection_state(stale)  # include_active_streams=False
+
+        self.assertEqual(conn.get_active_streams_count(), 2)
+        reloaded = conn._get_connection_state()
+        self.assertEqual(reloaded.worker_id, "worker-b")
+        self.assertEqual(reloaded.active_streams, 2)
+
+    def test_decr_at_zero_does_not_go_negative(self):
+        RedisBackedVODConnection, _ = _import_vod()
+        redis = LockAwareFakeRedis()
+        conn = _seed_session(redis, "vod_zero", active_streams=0)
+
+        success, has_remaining = conn.decrement_active_streams_and_check()
+        self.assertEqual((success, has_remaining), (False, False))
+        self.assertEqual(conn.get_active_streams_count(), 0)
+
+    def test_multi_worker_concurrent_teardown_balances_to_zero(self):
+        """Jellyfin-style: AS=2, two workers DECR concurrently under lock churn."""
+        RedisBackedVODConnection, _ = _import_vod()
+        redis = LockAwareFakeRedis()
+        _seed_session(redis, "vod_multi", active_streams=2)
+
+        worker_a = RedisBackedVODConnection("vod_multi", redis)
+        worker_b = RedisBackedVODConnection("vod_multi", redis)
+        interferer = RedisBackedVODConnection("vod_multi", redis)
+
+        results = []
+        barrier = threading.Barrier(2)
+
+        def churn_metadata_lock():
+            for _ in range(8):
+                if interferer._acquire_lock():
+                    # Simulate ownership transfer / seek metadata write
+                    state = interferer._get_connection_state()
+                    if state:
+                        state.last_activity = time.time()
+                        interferer._save_connection_state(state)
+                    interferer._release_lock()
+                time.sleep(0.001)
+
+        def teardown(worker):
+            barrier.wait()
+            results.append(worker.decrement_active_streams_and_check())
+
+        t_churn = threading.Thread(target=churn_metadata_lock, daemon=True)
+        t_a = threading.Thread(target=teardown, args=(worker_a,))
+        t_b = threading.Thread(target=teardown, args=(worker_b,))
+        t_churn.start()
+        t_a.start()
+        t_b.start()
+        t_a.join(timeout=5)
+        t_b.join(timeout=5)
+        t_churn.join(timeout=5)
+
+        self.assertEqual(len(results), 2)
+        self.assertTrue(all(success for success, _ in results))
+        self.assertEqual(sorted(has for _, has in results), [False, True])
+        self.assertEqual(worker_a.get_active_streams_count(), 0)
+
+    def test_cleanup_skips_when_active_streams_present(self):
+        redis = LockAwareFakeRedis()
+        conn = _seed_session(redis, "vod_clean_busy", active_streams=1)
+        conn.cleanup()
+        self.assertIsNotNone(conn._get_connection_state())
+        self.assertEqual(conn.get_active_streams_count(), 1)
+
+    def test_cleanup_deletes_when_idle(self):
+        redis = LockAwareFakeRedis()
+        conn = _seed_session(redis, "vod_clean_idle", active_streams=0)
+        conn.cleanup()
+        self.assertIsNone(conn._get_connection_state())
+
+    def test_cleanup_does_not_delete_metadata_lock(self):
+        """Idle cleanup must not steal the metadata lock from a holder."""
+        redis = LockAwareFakeRedis()
+        holder = _seed_session(redis, "vod_lock_keep", active_streams=0)
+        self.assertTrue(holder._acquire_lock())
+        try:
+            holder.cleanup()
+            # Session hash gone, but lock still held by this worker
+            self.assertIsNone(holder._get_connection_state())
+            self.assertTrue(redis.exists(holder.lock_key))
+            # Holder can still release cleanly
+        finally:
+            holder._release_lock()
+        self.assertFalse(redis.exists(holder.lock_key))
+
+    def test_unconditional_hset_after_cleanup_recreates_zombie(self):
+        """Reproduce: plain HSET after idle cleanup recreates a session hash."""
+        redis = LockAwareFakeRedis()
+        holder = _seed_session(redis, "vod_zombie_raw", active_streams=0)
+        stale = holder._get_connection_state()
+        self.assertTrue(holder._acquire_lock())
+        try:
+            holder.cleanup()
+            self.assertIsNone(holder._get_connection_state())
+
+            # Bypass the exists-guard (old metadata save behavior)
+            data = stale.to_dict()
+            data.pop("active_streams", None)
+            redis.hset(holder.connection_key, mapping=data)
+
+            zombie = holder._get_connection_state()
+            self.assertIsNotNone(zombie)
+            # Recreated without a reliable counter field from Lua INCR/DECR era
+            self.assertFalse(holder.has_active_streams())
+        finally:
+            holder._release_lock()
+
+    def test_metadata_save_after_cleanup_does_not_recreate_zombie(self):
+        """Fixed path: metadata save is a no-op if cleanup already deleted the hash."""
+        redis = LockAwareFakeRedis()
+        holder = _seed_session(redis, "vod_zombie_fix", active_streams=0)
+        stale = holder._get_connection_state()
+        stale.worker_id = "worker-late"
+        stale.last_activity = time.time()
+
+        self.assertTrue(holder._acquire_lock())
+        try:
+            holder.cleanup()
+            self.assertIsNone(holder._get_connection_state())
+
+            saved = holder._save_connection_state(stale)
+            self.assertFalse(saved)
+            self.assertIsNone(holder._get_connection_state())
+            self.assertNotIn(holder.connection_key, redis._hashes)
+        finally:
+            holder._release_lock()
+
+    def test_create_save_still_creates_new_session(self):
+        """include_active_streams=True must still be able to create a missing key."""
+        RedisBackedVODConnection, SerializableConnectionState = _import_vod()
+        _clear_script_cache()
+        redis = LockAwareFakeRedis()
+        conn = RedisBackedVODConnection("vod_create_new", redis)
+        state = SerializableConnectionState(
+            session_id="vod_create_new",
+            stream_url="http://example.com/movie.mkv",
+            headers={},
+            m3u_profile_id=7,
+        )
+        self.assertTrue(conn._save_connection_state(state, include_active_streams=True))
+        self.assertIsNotNone(conn._get_connection_state())
+        self.assertEqual(conn.get_active_streams_count(), 0)
+
+    def test_cleanup_atomic_vs_reconnect_incr(self):
+        """If INCR wins the race, cleanup must not delete the session."""
+        redis = LockAwareFakeRedis()
+        conn = _seed_session(redis, "vod_race", active_streams=0)
+
+        # Reconnect increments before cleanup Lua runs
+        self.assertEqual(conn.increment_active_streams(), 1)
+        conn.cleanup()
+        self.assertEqual(conn.get_active_streams_count(), 1)
+        self.assertIsNotNone(conn._get_connection_state())
+
+
+class TestLegacyLockContentionBehaviorGone(SimpleTestCase):
+    """Document that the old lock-gated DECR orphan path no longer exists."""
+
+    def test_old_no_retry_orphan_scenario_no_longer_orphans(self):
+        RedisBackedVODConnection, _ = _import_vod()
+        redis = LockAwareFakeRedis()
+        holder = _seed_session(redis, "vod_orphan_old", active_streams=1)
+        other = RedisBackedVODConnection("vod_orphan_old", redis)
+
+        self.assertTrue(holder._acquire_lock())
+        success, has_remaining = other.decrement_active_streams_and_check()
+        holder._release_lock()
+
+        # Previously: (False, True) with active_streams left at 1
+        self.assertEqual((success, has_remaining), (True, False))
+        self.assertEqual(other.get_active_streams_count(), 0)
+
+
+class TestVodActiveStreamsRealRedis(SimpleTestCase):
+    """Integration tests against a live Redis (real Lua EVALSHA).
+
+    FakeRedis covers fast unit cases; this class verifies the registered
+    scripts behave correctly on the Redis server used in CI/dev.
+    """
+
+    redis = None
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        try:
+            from core.utils import RedisClient
+
+            client = RedisClient.get_client(max_retries=1, retry_interval=0)
+            if client is not None and client.ping():
+                cls.redis = client
+        except Exception:
+            cls.redis = None
+
+    def setUp(self):
+        if self.redis is None:
+            self.skipTest("Redis not available")
+        import uuid
+
+        from apps.proxy.vod_proxy import multi_worker_connection_manager as mod
+
+        mod._vod_script_cache.clear()
+        self.session_id = f"test_vod_lua_{uuid.uuid4().hex}"
+        self._keys_to_delete = []
+
+    def tearDown(self):
+        if self.redis is None:
+            return
+        RedisBackedVODConnection, _ = _import_vod()
+        conn = RedisBackedVODConnection(self.session_id, self.redis)
+        keys = [conn.connection_key, conn.lock_key, *self._keys_to_delete]
+        if keys:
+            self.redis.delete(*keys)
+        from apps.proxy.vod_proxy import multi_worker_connection_manager as mod
+
+        mod._vod_script_cache.clear()
+
+    def _seed(self, active_streams=1):
+        RedisBackedVODConnection, SerializableConnectionState = _import_vod()
+        conn = RedisBackedVODConnection(self.session_id, self.redis)
+        state = SerializableConnectionState(
+            session_id=self.session_id,
+            stream_url="http://example.com/movie.mkv",
+            headers={"User-Agent": "test"},
+            m3u_profile_id=7,
+        )
+        state.active_streams = active_streams
+        state.worker_id = "real-redis-worker"
+        self.assertTrue(conn._save_connection_state(state, include_active_streams=True))
+        return conn
+
+    def test_real_lua_incr_decr_round_trip(self):
+        RedisBackedVODConnection, _ = _import_vod()
+        a = self._seed(active_streams=0)
+        b = RedisBackedVODConnection(self.session_id, self.redis)
+
+        self.assertEqual(a.increment_active_streams(), 1)
+        self.assertEqual(b.increment_active_streams(), 2)
+        self.assertEqual(int(self.redis.hget(a.connection_key, "active_streams")), 2)
+
+        ok_b, rem_b = b.decrement_active_streams_and_check()
+        ok_a, rem_a = a.decrement_active_streams_and_check()
+        self.assertTrue(ok_b and rem_b)
+        self.assertTrue(ok_a and not rem_a)
+        self.assertEqual(int(self.redis.hget(a.connection_key, "active_streams")), 0)
+
+    def test_real_lua_decr_while_metadata_lock_held(self):
+        RedisBackedVODConnection, _ = _import_vod()
+        holder = self._seed(active_streams=1)
+        waiter = RedisBackedVODConnection(self.session_id, self.redis)
+
+        self.assertTrue(holder._acquire_lock())
+        try:
+            success, has_remaining = waiter.decrement_active_streams_and_check()
+            self.assertEqual((success, has_remaining), (True, False))
+            self.assertEqual(int(self.redis.hget(holder.connection_key, "active_streams")), 0)
+        finally:
+            holder._release_lock()
+
+    def test_real_lua_metadata_save_does_not_recreate_after_cleanup(self):
+        holder = self._seed(active_streams=0)
+        stale = holder._get_connection_state()
+        stale.worker_id = "late-writer"
+
+        self.assertTrue(holder._acquire_lock())
+        try:
+            holder.cleanup()
+            self.assertFalse(bool(self.redis.exists(holder.connection_key)))
+
+            saved = holder._save_connection_state(stale)
+            self.assertFalse(saved)
+            self.assertFalse(bool(self.redis.exists(holder.connection_key)))
+        finally:
+            holder._release_lock()
+
+    def test_real_lua_cleanup_skips_when_active(self):
+        conn = self._seed(active_streams=1)
+        conn.cleanup()
+        self.assertTrue(bool(self.redis.exists(conn.connection_key)))
+        self.assertEqual(int(self.redis.hget(conn.connection_key, "active_streams")), 1)
+
+    def test_real_lua_concurrent_decr_under_metadata_churn(self):
+        RedisBackedVODConnection, _ = _import_vod()
+        self._seed(active_streams=2)
+        worker_a = RedisBackedVODConnection(self.session_id, self.redis)
+        worker_b = RedisBackedVODConnection(self.session_id, self.redis)
+        interferer = RedisBackedVODConnection(self.session_id, self.redis)
+
+        results = []
+        barrier = threading.Barrier(2)
+
+        def churn():
+            for _ in range(10):
+                if interferer._acquire_lock():
+                    state = interferer._get_connection_state()
+                    if state:
+                        state.last_activity = time.time()
+                        interferer._save_connection_state(state)
+                    interferer._release_lock()
+                time.sleep(0.001)
+
+        def teardown(worker):
+            barrier.wait()
+            results.append(worker.decrement_active_streams_and_check())
+
+        threads = [
+            threading.Thread(target=churn, daemon=True),
+            threading.Thread(target=teardown, args=(worker_a,)),
+            threading.Thread(target=teardown, args=(worker_b,)),
+        ]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=5)
+
+        self.assertEqual(len(results), 2)
+        self.assertTrue(all(success for success, _ in results))
+        self.assertEqual(sorted(has for _, has in results), [False, True])
+        self.assertEqual(int(self.redis.hget(worker_a.connection_key, "active_streams")), 0)
+
+    def test_real_lua_scripts_are_registered_as_evalsha(self):
+        """Smoke-check redis-py Script objects hit the live server."""
+        conn = self._seed(active_streams=0)
+        scripts = conn._vod_scripts()
+        for name in ("incr", "decr", "cleanup", "meta_save"):
+            self.assertIn(name, scripts)
+            # Calling Script triggers SCRIPT LOAD / EVALSHA on first use
+            self.assertTrue(hasattr(scripts[name], "sha") or callable(scripts[name]))
+
+        self.assertEqual(conn.increment_active_streams(), 1)
+        # After first call, sha should be populated on redis-py Script
+        incr_script = scripts["incr"]
+        sha = getattr(incr_script, "sha", None)
+        self.assertTrue(sha, "Expected redis-py Script.sha after EVALSHA")