Merge branch 'dev' of https://github.com/Dispatcharr/Dispatcharr into pr/CodeBormen/949

CHANGELOG.md

@@ -13,7 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
-- Celery Redis connections using unpatched sockets under gevent: `dispatcharr/__init__.py` imports `celery.py` as part of `django.setup()`, initialising the Redis connection pool before uWSGI's gevent plugin patches stdlib sockets. Added `gevent.monkey.patch_all()` at the top of `wsgi.py` to guarantee all broker connections are gevent-aware. - Thanks [@endoze](https://github.com/endoze)
+- TS proxy client stream lag recovery now only bumps clients forward when their next required chunk has genuinely expired from Redis (TTL), rather than unconditionally jumping if they fell more than 50 chunks behind. Clients are repositioned to the oldest available chunk (minimum data loss) using an atomic server-side Lua binary search, falling back to near the buffer head if nothing is available.
 - TS proxy streams dying after 30–200 seconds in multi-worker uWSGI/Celery deployments, caused by three interrelated bugs. (Fixes #992, #980) - Thanks [@PFalko](https://github.com/PFalko)
   - **Double ProxyServer instantiation**: `ProxyConfig.ready()` called `TSProxyServer()` directly while `TSProxyConfig.ready()` also called `TSProxyServer.get_instance()`, creating two instances per worker — each with its own cleanup thread. The orphaned thread could not extend ownership because it had no entries in `stream_managers`. Fixed by using `TSProxyServer.get_instance()` in `ProxyConfig.ready()`.
   - **`flushdb()` on every Redis client init**: `RedisClient.get_client()` called `client.flushdb()` whenever `_client` was `None`. Celery autoscale (`--autoscale=6,1`) spawning new workers mid-stream triggered this path, nuking all Redis keys including active ownership keys, client records, and channel metadata. Removed the `flushdb()` call entirely.

apps/proxy/ts_proxy/stream_buffer.py

@@ -45,6 +45,15 @@ class StreamBuffer:
         self._write_buffer = bytearray()
         self.target_chunk_size = ConfigHelper.get('BUFFER_CHUNK_SIZE', TS_PACKET_SIZE * 5644)  # ~1MB default
 
+        # Register Lua scripts once — subsequent calls use EVALSHA (just the
+        # SHA hash) instead of sending the full script text on every invocation.
+        if self.redis_client:
+            self._find_oldest_chunk_sha = self.redis_client.register_script(
+                self._FIND_OLDEST_CHUNK_LUA
+            )
+        else:
+            self._find_oldest_chunk_sha = None
+
         # Track timers for proper cleanup
         self.stopping = False
         self.fill_timers = []
@@ -328,6 +337,89 @@ class StreamBuffer:
 
         return chunks, client_index + chunk_count
 
+    # Lua script that runs an atomic binary search on the Redis server.
+    # Chunks expire in FIFO order (same TTL, sequential writes), so the
+    # alive range is contiguous: [oldest_surviving .. buffer_head].
+    # Binary search finds the boundary in O(log N) EXISTS calls with zero
+    # round-trips between steps and no TOCTOU races (Lua scripts are atomic).
+    #
+    # ARGV[1] = key prefix  (e.g. "ts_proxy:channel:<id>:buffer:chunk:")
+    # ARGV[2] = low index   (client_index + 1, first chunk the client needs)
+    # ARGV[3] = high index  (buffer head, most recent chunk)
+    #
+    # Returns: the index of the oldest existing chunk, or -1 if none exist.
+    _FIND_OLDEST_CHUNK_LUA = """
+    local prefix = ARGV[1]
+    local low    = tonumber(ARGV[2])
+    local high   = tonumber(ARGV[3])
+
+    if redis.call('EXISTS', prefix .. high) == 0 then
+        return -1
+    end
+
+    local result = high
+    while low <= high do
+        local mid = math.floor((low + high) / 2)
+        if redis.call('EXISTS', prefix .. mid) == 1 then
+            result = mid
+            high = mid - 1
+        else
+            low = mid + 1
+        end
+    end
+    return result
+    """
+
+    def find_oldest_available_chunk(self, client_index):
+        """Find the oldest (lowest-index) chunk that still exists in Redis.
+
+        Executes an atomic Lua binary search on the Redis server — one
+        round-trip, ~log2(N) EXISTS calls, no TOCTOU between steps.
+
+        The actual read attempt (get_optimized_client_data) is what
+        authoritatively detects expiration; this method is best-effort
+        positioning that self-corrects on the next iteration if the found
+        chunk also expires before the client can read it.
+
+        Args:
+            client_index: The client's current local_index (last consumed chunk).
+
+        Returns:
+            int or None: The local_index value the client should jump to
+                         (one before the first available chunk), or None if no
+                         chunks are available at all.
+        """
+        if not self.redis_client:
+            return None
+
+        low = client_index + 1   # First chunk the client needs
+        high = self.index        # Latest chunk written
+
+        if low > high:
+            return None
+
+        try:
+            # Uses EVALSHA under the hood — sends only the SHA hash,
+            # not the full script text, on every call after the first.
+            result = self._find_oldest_chunk_sha(
+                args=[
+                    RedisKeys.buffer_chunk_prefix(self.channel_id),
+                    low,
+                    high,
+                ],
+            )
+
+            if result == -1:
+                return None
+
+            # Return result - 1 so local_index points to one before the
+            # first available chunk (matching the "last consumed" convention).
+            return int(result) - 1
+
+        except Exception as e:
+            logger.error(f"Error running find_oldest_chunk Lua script for channel {self.channel_id}: {e}")
+            return None
+
     # Add a new method to safely create timers
     def schedule_timer(self, delay, callback, *args, **kwargs):
         """Schedule a timer and track it for proper cleanup"""

apps/proxy/ts_proxy/stream_generator.py

@@ -224,17 +224,35 @@ class StreamGenerator:
                 self.empty_reads += 1
                 self.consecutive_empty += 1
 
-                # Check if we're too far behind (chunks expired from Redis)
+                # We got no data despite being behind the buffer head.
+                # The read itself is the authoritative signal — no separate
+                # existence check needed, avoiding TOCTOU races with Redis TTL.
                 chunks_behind = self.buffer.index - self.local_index
-                if chunks_behind > 50:  # If more than 50 chunks behind, jump forward
-                    # Calculate new position: stay a few chunks behind current buffer
-                    initial_behind = ConfigHelper.initial_behind_chunks()
-                    new_index = max(self.local_index, self.buffer.index - initial_behind)
+                if chunks_behind > 0:
+                    # Next chunk has expired — find the oldest chunk still in Redis
+                    new_index = self.buffer.find_oldest_available_chunk(self.local_index)
 
-                    logger.warning(f"[{self.client_id}] Client too far behind ({chunks_behind} chunks), jumping from {self.local_index} to {new_index}")
-                    self.local_index = new_index
-                    self.consecutive_empty = 0  # Reset since we're repositioning
-                    continue  # Try again immediately with new position
+                    if new_index is not None:
+                        skipped = new_index - self.local_index
+                        logger.warning(
+                            f"[{self.client_id}] Next chunk expired (index {self.local_index + 1}), "
+                            f"jumping to oldest available: {new_index + 1} "
+                            f"(skipped {skipped} chunks, buffer head at {self.buffer.index})"
+                        )
+                        self.local_index = new_index
+                    else:
+                        # No chunks available at all — jump to near the buffer head
+                        initial_behind = ConfigHelper.initial_behind_chunks()
+                        new_index = max(self.local_index, self.buffer.index - initial_behind)
+                        logger.warning(
+                            f"[{self.client_id}] No chunks available in buffer, "
+                            f"jumping to near buffer head: {new_index} "
+                            f"(buffer head at {self.buffer.index})"
+                        )
+                        self.local_index = new_index
+
+                    self.consecutive_empty = 0
+                    continue  # Retry immediately with the new position
 
                 if self._should_send_keepalive(self.local_index):
                     keepalive_packet = create_ts_packet('keepalive')

dispatcharr/tests/test_wsgi.py

@@ -1,39 +0,0 @@
-"""Tests for dispatcharr/wsgi.py gevent monkey-patching."""
-
-import os
-import subprocess
-import sys
-import unittest
-
-
-class TestWSGIGeventPatching(unittest.TestCase):
-    """Verify that wsgi.py applies gevent monkey-patching before Django imports.
-
-    These tests run in subprocesses because monkey.patch_all() must execute
-    before any other imports.  By the time Django's test runner loads this
-    file, ssl/socket are already imported, so calling patch_all() in-process
-    would fail.  A subprocess mirrors how uWSGI actually loads wsgi.py.
-    """
-
-    def test_socket_is_patched_after_wsgi_import(self):
-        """Loading wsgi.py first (as uWSGI does) should patch sockets."""
-        result = subprocess.run(
-            [
-                sys.executable, "-c",
-                "import dispatcharr.wsgi; "
-                "from gevent import monkey; "
-                "assert monkey.is_module_patched('socket'), "
-                "'socket module was not patched by wsgi.py'; "
-                "print('PASS')",
-            ],
-            capture_output=True,
-            text=True,
-            env={**os.environ, "DJANGO_SETTINGS_MODULE": "dispatcharr.settings"},
-            timeout=60,
-            cwd="/app",
-        )
-        self.assertEqual(
-            result.returncode,
-            0,
-            f"monkey-patching check failed:\n{result.stderr}",
-        )

dispatcharr/wsgi.py

@@ -1,21 +1,6 @@
 """
 WSGI config for dispatcharr project.
 """
-# When running under uWSGI with gevent, ensure monkey-patching is fully
-# applied before any other imports.  Django's setup triggers the import of
-# dispatcharr/__init__.py → celery.py, which initialises the Celery broker
-# transport.  If those imports happen before sockets are patched, Celery's
-# Redis connection pool holds unpatched sockets that silently fail under
-# gevent's cooperative scheduling.  Calling patch_all() here—before the
-# Django import chain—guarantees every socket (including pooled broker
-# connections) is gevent-aware.  The call is idempotent, so it's harmless
-# if uWSGI's gevent plugin has already patched.
-try:
-    from gevent import monkey
-    monkey.patch_all()
-except ImportError:
-    pass
-
 import os
 from django.core.wsgi import get_wsgi_application