From e78c24aad735952eb6f8939dc2935aa4fe20546d Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 15:13:46 -0600 Subject: [PATCH 1/3] Enhance Redis chunk management in StreamBuffer and StreamGenerator - Register Lua scripts for optimized chunk retrieval in StreamBuffer. - Implement atomic Lua binary search to find the oldest available chunk in Redis. - Update StreamGenerator to handle expired chunks more efficiently, reducing TOCTOU race conditions. --- apps/proxy/ts_proxy/stream_buffer.py | 92 +++++++++++++++++++++++++ apps/proxy/ts_proxy/stream_generator.py | 36 +++++++--- 2 files changed, 119 insertions(+), 9 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_buffer.py b/apps/proxy/ts_proxy/stream_buffer.py index 85feb5dd..2bf99527 100644 --- a/apps/proxy/ts_proxy/stream_buffer.py +++ b/apps/proxy/ts_proxy/stream_buffer.py @@ -45,6 +45,15 @@ class StreamBuffer: self._write_buffer = bytearray() self.target_chunk_size = ConfigHelper.get('BUFFER_CHUNK_SIZE', TS_PACKET_SIZE * 5644) # ~1MB default + # Register Lua scripts once — subsequent calls use EVALSHA (just the + # SHA hash) instead of sending the full script text on every invocation. + if self.redis_client: + self._find_oldest_chunk_sha = self.redis_client.register_script( + self._FIND_OLDEST_CHUNK_LUA + ) + else: + self._find_oldest_chunk_sha = None + # Track timers for proper cleanup self.stopping = False self.fill_timers = [] @@ -328,6 +337,89 @@ class StreamBuffer: return chunks, client_index + chunk_count + # Lua script that runs an atomic binary search on the Redis server. + # Chunks expire in FIFO order (same TTL, sequential writes), so the + # alive range is contiguous: [oldest_surviving .. buffer_head]. + # Binary search finds the boundary in O(log N) EXISTS calls with zero + # round-trips between steps and no TOCTOU races (Lua scripts are atomic). + # + # ARGV[1] = key prefix (e.g. "ts_proxy:channel::buffer:chunk:") + # ARGV[2] = low index (client_index + 1, first chunk the client needs) + # ARGV[3] = high index (buffer head, most recent chunk) + # + # Returns: the index of the oldest existing chunk, or -1 if none exist. + _FIND_OLDEST_CHUNK_LUA = """ + local prefix = ARGV[1] + local low = tonumber(ARGV[2]) + local high = tonumber(ARGV[3]) + + if redis.call('EXISTS', prefix .. high) == 0 then + return -1 + end + + local result = high + while low <= high do + local mid = math.floor((low + high) / 2) + if redis.call('EXISTS', prefix .. mid) == 1 then + result = mid + high = mid - 1 + else + low = mid + 1 + end + end + return result + """ + + def find_oldest_available_chunk(self, client_index): + """Find the oldest (lowest-index) chunk that still exists in Redis. + + Executes an atomic Lua binary search on the Redis server — one + round-trip, ~log2(N) EXISTS calls, no TOCTOU between steps. + + The actual read attempt (get_optimized_client_data) is what + authoritatively detects expiration; this method is best-effort + positioning that self-corrects on the next iteration if the found + chunk also expires before the client can read it. + + Args: + client_index: The client's current local_index (last consumed chunk). + + Returns: + int or None: The local_index value the client should jump to + (one before the first available chunk), or None if no + chunks are available at all. + """ + if not self.redis_client: + return None + + low = client_index + 1 # First chunk the client needs + high = self.index # Latest chunk written + + if low > high: + return None + + try: + # Uses EVALSHA under the hood — sends only the SHA hash, + # not the full script text, on every call after the first. + result = self._find_oldest_chunk_sha( + args=[ + RedisKeys.buffer_chunk_prefix(self.channel_id), + low, + high, + ], + ) + + if result == -1: + return None + + # Return result - 1 so local_index points to one before the + # first available chunk (matching the "last consumed" convention). + return int(result) - 1 + + except Exception as e: + logger.error(f"Error running find_oldest_chunk Lua script for channel {self.channel_id}: {e}") + return None + # Add a new method to safely create timers def schedule_timer(self, delay, callback, *args, **kwargs): """Schedule a timer and track it for proper cleanup""" diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 50404f1d..8f631d53 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -224,17 +224,35 @@ class StreamGenerator: self.empty_reads += 1 self.consecutive_empty += 1 - # Check if we're too far behind (chunks expired from Redis) + # We got no data despite being behind the buffer head. + # The read itself is the authoritative signal — no separate + # existence check needed, avoiding TOCTOU races with Redis TTL. chunks_behind = self.buffer.index - self.local_index - if chunks_behind > 50: # If more than 50 chunks behind, jump forward - # Calculate new position: stay a few chunks behind current buffer - initial_behind = ConfigHelper.initial_behind_chunks() - new_index = max(self.local_index, self.buffer.index - initial_behind) + if chunks_behind > 0: + # Next chunk has expired — find the oldest chunk still in Redis + new_index = self.buffer.find_oldest_available_chunk(self.local_index) - logger.warning(f"[{self.client_id}] Client too far behind ({chunks_behind} chunks), jumping from {self.local_index} to {new_index}") - self.local_index = new_index - self.consecutive_empty = 0 # Reset since we're repositioning - continue # Try again immediately with new position + if new_index is not None: + skipped = new_index - self.local_index + logger.warning( + f"[{self.client_id}] Next chunk expired (index {self.local_index + 1}), " + f"jumping to oldest available: {new_index + 1} " + f"(skipped {skipped} chunks, buffer head at {self.buffer.index})" + ) + self.local_index = new_index + else: + # No chunks available at all — jump to near the buffer head + initial_behind = ConfigHelper.initial_behind_chunks() + new_index = max(self.local_index, self.buffer.index - initial_behind) + logger.warning( + f"[{self.client_id}] No chunks available in buffer, " + f"jumping to near buffer head: {new_index} " + f"(buffer head at {self.buffer.index})" + ) + self.local_index = new_index + + self.consecutive_empty = 0 + continue # Retry immediately with the new position if self._should_send_keepalive(self.local_index): keepalive_packet = create_ts_packet('keepalive') From d487bb5671236877c28dfe1e148e7072c9a15fe2 Mon Sep 17 00:00:00 2001 From: SergeantPanda <61642231+SergeantPanda@users.noreply.github.com> Date: Tue, 3 Mar 2026 15:33:55 -0600 Subject: [PATCH 2/3] Revert "fix(wsgi): ensure gevent monkey-patching before Django import chain" --- CHANGELOG.md | 1 - dispatcharr/tests/__init__.py | 0 dispatcharr/tests/test_wsgi.py | 39 ---------------------------------- dispatcharr/wsgi.py | 15 ------------- 4 files changed, 55 deletions(-) delete mode 100644 dispatcharr/tests/__init__.py delete mode 100644 dispatcharr/tests/test_wsgi.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 55fac12c..fadc44e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Celery Redis connections using unpatched sockets under gevent: `dispatcharr/__init__.py` imports `celery.py` as part of `django.setup()`, initialising the Redis connection pool before uWSGI's gevent plugin patches stdlib sockets. Added `gevent.monkey.patch_all()` at the top of `wsgi.py` to guarantee all broker connections are gevent-aware. - Thanks [@endoze](https://github.com/endoze) - TS proxy streams dying after 30–200 seconds in multi-worker uWSGI/Celery deployments, caused by three interrelated bugs. (Fixes #992, #980) - Thanks [@PFalko](https://github.com/PFalko) - **Double ProxyServer instantiation**: `ProxyConfig.ready()` called `TSProxyServer()` directly while `TSProxyConfig.ready()` also called `TSProxyServer.get_instance()`, creating two instances per worker — each with its own cleanup thread. The orphaned thread could not extend ownership because it had no entries in `stream_managers`. Fixed by using `TSProxyServer.get_instance()` in `ProxyConfig.ready()`. - **`flushdb()` on every Redis client init**: `RedisClient.get_client()` called `client.flushdb()` whenever `_client` was `None`. Celery autoscale (`--autoscale=6,1`) spawning new workers mid-stream triggered this path, nuking all Redis keys including active ownership keys, client records, and channel metadata. Removed the `flushdb()` call entirely. diff --git a/dispatcharr/tests/__init__.py b/dispatcharr/tests/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/dispatcharr/tests/test_wsgi.py b/dispatcharr/tests/test_wsgi.py deleted file mode 100644 index cdf7a2e9..00000000 --- a/dispatcharr/tests/test_wsgi.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Tests for dispatcharr/wsgi.py gevent monkey-patching.""" - -import os -import subprocess -import sys -import unittest - - -class TestWSGIGeventPatching(unittest.TestCase): - """Verify that wsgi.py applies gevent monkey-patching before Django imports. - - These tests run in subprocesses because monkey.patch_all() must execute - before any other imports. By the time Django's test runner loads this - file, ssl/socket are already imported, so calling patch_all() in-process - would fail. A subprocess mirrors how uWSGI actually loads wsgi.py. - """ - - def test_socket_is_patched_after_wsgi_import(self): - """Loading wsgi.py first (as uWSGI does) should patch sockets.""" - result = subprocess.run( - [ - sys.executable, "-c", - "import dispatcharr.wsgi; " - "from gevent import monkey; " - "assert monkey.is_module_patched('socket'), " - "'socket module was not patched by wsgi.py'; " - "print('PASS')", - ], - capture_output=True, - text=True, - env={**os.environ, "DJANGO_SETTINGS_MODULE": "dispatcharr.settings"}, - timeout=60, - cwd="/app", - ) - self.assertEqual( - result.returncode, - 0, - f"monkey-patching check failed:\n{result.stderr}", - ) diff --git a/dispatcharr/wsgi.py b/dispatcharr/wsgi.py index 6e8860af..29c2ba01 100644 --- a/dispatcharr/wsgi.py +++ b/dispatcharr/wsgi.py @@ -1,21 +1,6 @@ """ WSGI config for dispatcharr project. """ -# When running under uWSGI with gevent, ensure monkey-patching is fully -# applied before any other imports. Django's setup triggers the import of -# dispatcharr/__init__.py → celery.py, which initialises the Celery broker -# transport. If those imports happen before sockets are patched, Celery's -# Redis connection pool holds unpatched sockets that silently fail under -# gevent's cooperative scheduling. Calling patch_all() here—before the -# Django import chain—guarantees every socket (including pooled broker -# connections) is gevent-aware. The call is idempotent, so it's harmless -# if uWSGI's gevent plugin has already patched. -try: - from gevent import monkey - monkey.patch_all() -except ImportError: - pass - import os from django.core.wsgi import get_wsgi_application From 24c20a2f4accf486b89c7e664d92571d2cdc845b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 15:59:49 -0600 Subject: [PATCH 3/3] changelog: Update changelog for proxy changes. --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index fadc44e8..49e7be68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- TS proxy client stream lag recovery now only bumps clients forward when their next required chunk has genuinely expired from Redis (TTL), rather than unconditionally jumping if they fell more than 50 chunks behind. Clients are repositioned to the oldest available chunk (minimum data loss) using an atomic server-side Lua binary search, falling back to near the buffer head if nothing is available. - TS proxy streams dying after 30–200 seconds in multi-worker uWSGI/Celery deployments, caused by three interrelated bugs. (Fixes #992, #980) - Thanks [@PFalko](https://github.com/PFalko) - **Double ProxyServer instantiation**: `ProxyConfig.ready()` called `TSProxyServer()` directly while `TSProxyConfig.ready()` also called `TSProxyServer.get_instance()`, creating two instances per worker — each with its own cleanup thread. The orphaned thread could not extend ownership because it had no entries in `stream_managers`. Fixed by using `TSProxyServer.get_instance()` in `ProxyConfig.ready()`. - **`flushdb()` on every Redis client init**: `RedisClient.get_client()` called `client.flushdb()` whenever `_client` was `None`. Celery autoscale (`--autoscale=6,1`) spawning new workers mid-stream triggered this path, nuking all Redis keys including active ownership keys, client records, and channel metadata. Removed the `flushdb()` call entirely.