From 8bd38ad71c455317d9a0753887b223bd3321fee5 Mon Sep 17 00:00:00 2001 From: PFalko Date: Fri, 27 Feb 2026 16:40:28 +0100 Subject: [PATCH 1/4] Fix stream ownership bugs causing streams to die after 30-200s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three interrelated bugs cause TS proxy streams to terminate prematurely in multi-worker uWSGI deployments: 1. Double ProxyServer instantiation: ProxyConfig.ready() in apps/proxy/apps.py calls TSProxyServer() directly, bypassing get_instance(). The subsequent TSProxyConfig.ready() call to get_instance() creates a second instance. Each instance starts its own cleanup thread, but only one holds channel data — the orphaned cleanup thread cannot extend ownership. 2. Redis flushdb() on every client init: RedisClient.get_client() in core/utils.py calls flushdb() whenever a new connection is created. Celery autoscale workers spawning mid-stream nuke all Redis keys including ownership, client records, and channel metadata. 3. No recovery from expired ownership: get_channel_owner() has a TOCTOU bug (two separate GET calls in a lambda). extend_ownership() silently fails when keys expire. Non-owner cleanup unconditionally kills streams even when the worker holds the stream_manager. Fixes: - Use TSProxyServer.get_instance() in ProxyConfig.ready() - Remove flushdb() from Redis client initialization - Use sentinel pattern for gevent-safe singleton (threading.Lock does not work with gevent greenlets) - Single GET in get_channel_owner() to avoid TOCTOU race - Re-acquire expired ownership keys in extend_ownership() - Attempt re-acquisition before cleanup in non-owner path Relates to #992, #980 --- apps/proxy/apps.py | 4 +-- apps/proxy/ts_proxy/server.py | 66 +++++++++++++++++++++++++++++------ core/utils.py | 1 - 3 files changed, 58 insertions(+), 13 deletions(-) diff --git a/apps/proxy/apps.py b/apps/proxy/apps.py index d1c8b966..b33d46cc 100644 --- a/apps/proxy/apps.py +++ b/apps/proxy/apps.py @@ -12,6 +12,6 @@ class ProxyConfig(AppConfig): from .hls_proxy.server import ProxyServer as HLSProxyServer from .ts_proxy.server import ProxyServer as TSProxyServer - # Initialize proxy servers + # Initialize proxy servers (TS uses singleton to prevent duplicate instances) self.hls_proxy = HLSProxyServer() - self.ts_proxy = TSProxyServer() + self.ts_proxy = TSProxyServer.get_instance() diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index df51744d..b28154b3 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -34,18 +34,29 @@ logger = get_logger() class ProxyServer: """Manages TS proxy server instance with worker coordination""" _instance = None + _INITIALIZING = object() # sentinel for gevent-safe singleton @classmethod def get_instance(cls): - if cls._instance is None: + inst = cls._instance + if inst is not None and inst is not cls._INITIALIZING: + return inst + if inst is None: + cls._instance = cls._INITIALIZING from .server import ProxyServer from .stream_manager import StreamManager from .stream_buffer import StreamBuffer from .client_manager import ClientManager - - cls._instance = ProxyServer() - - return cls._instance + real_instance = ProxyServer() + cls._instance = real_instance + return real_instance + # Another greenlet is initializing — wait for completion + import time + while True: + inst = cls._instance + if inst is not None and inst is not cls._INITIALIZING: + return inst + time.sleep(0.05) def __init__(self): """Initialize proxy server with worker identification""" @@ -353,9 +364,16 @@ class ProxyServer: try: lock_key = RedisKeys.channel_owner(channel_id) - return self._execute_redis_command( - lambda: self.redis_client.get(lock_key).decode('utf-8') if self.redis_client.get(lock_key) else None + result = self._execute_redis_command( + lambda: self.redis_client.get(lock_key) ) + if result is None: + return None + try: + return result.decode('utf-8') + except (AttributeError, UnicodeDecodeError) as e: + logger.error(f"Error decoding channel owner for {channel_id}: {e}, raw={result!r}") + return None except Exception as e: logger.error(f"Error getting channel owner: {e}") return None @@ -433,7 +451,7 @@ class ProxyServer: logger.error(f"Error releasing channel ownership: {e}") def extend_ownership(self, channel_id, ttl=30): - """Extend ownership lease with grace period""" + """Extend ownership lease, re-acquiring if key expired""" if not self.redis_client: return False @@ -441,10 +459,24 @@ class ProxyServer: lock_key = RedisKeys.channel_owner(channel_id) current = self.redis_client.get(lock_key) - # Only extend if we're still the owner - if current and current.decode('utf-8') == self.worker_id: + if current is None: + # Key expired — re-acquire if we have the stream_manager + if channel_id in self.stream_managers: + acquired = self.redis_client.setnx(lock_key, self.worker_id) + if acquired: + self.redis_client.expire(lock_key, ttl) + logger.warning(f"Re-acquired expired ownership for channel {channel_id}") + return True + else: + new_owner = self.redis_client.get(lock_key) + logger.warning(f"Could not re-acquire ownership for {channel_id}, new owner: {new_owner}") + return False + return False + + if current.decode('utf-8') == self.worker_id: self.redis_client.expire(lock_key, ttl) return True + return False except Exception as e: logger.error(f"Error extending ownership: {e}") @@ -1173,6 +1205,20 @@ class ProxyServer: else: # === NON-OWNER CHANNEL HANDLING === + # Safety: if we have a stream_manager, we ARE the real owner + # but the Redis key may have expired. Try to re-acquire. + if channel_id in self.stream_managers: + logger.warning( + f"Ownership gap for {channel_id}: this worker has stream_manager " + f"but am_i_owner returned False. Attempting re-acquisition." + ) + reacquired = self.extend_ownership(channel_id) + if reacquired: + logger.info(f"Successfully re-acquired ownership for {channel_id}") + continue + else: + logger.error(f"Failed to re-acquire ownership for {channel_id}, will clean up") + # For channels we don't own, check if they've been stopped/cleaned up in Redis if self.redis_client: # Method 1: Check for stopping key diff --git a/core/utils.py b/core/utils.py index a5ea41df..f0e02bb4 100644 --- a/core/utils.py +++ b/core/utils.py @@ -81,7 +81,6 @@ class RedisClient: # Validate connection with ping client.ping() - client.flushdb() # Disable persistence on first connection - improves performance # Only try to disable if not in a read-only environment From 7fed49a3340e70e739ee7d72954be1ea6629d7cb Mon Sep 17 00:00:00 2001 From: Claude Code Date: Mon, 2 Mar 2026 13:23:45 +0000 Subject: [PATCH 2/4] Address review feedback from CodeBormen - Use atomic SET NX EX instead of separate SETNX + EXPIRE to prevent zombie locks if the process crashes between the two calls - Replace time.sleep() with gevent.sleep() in get_instance() spin-wait to avoid blocking the greenlet hub - Defer cleanup when ownership is lost but clients are still connected, giving the new owner time to establish its stream before we tear down Co-Authored-By: Claude Opus 4.6 --- apps/proxy/ts_proxy/server.py | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index b28154b3..b0b1a08e 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -51,12 +51,11 @@ class ProxyServer: cls._instance = real_instance return real_instance # Another greenlet is initializing — wait for completion - import time while True: inst = cls._instance if inst is not None and inst is not cls._INITIALIZING: return inst - time.sleep(0.05) + gevent.sleep(0.05) def __init__(self): """Initialize proxy server with worker identification""" @@ -462,9 +461,8 @@ class ProxyServer: if current is None: # Key expired — re-acquire if we have the stream_manager if channel_id in self.stream_managers: - acquired = self.redis_client.setnx(lock_key, self.worker_id) + acquired = self.redis_client.set(lock_key, self.worker_id, nx=True, ex=ttl) if acquired: - self.redis_client.expire(lock_key, ttl) logger.warning(f"Re-acquired expired ownership for channel {channel_id}") return True else: @@ -1217,6 +1215,19 @@ class ProxyServer: logger.info(f"Successfully re-acquired ownership for {channel_id}") continue else: + # Defer cleanup if we still have active clients — give the + # new owner time to spin up its own stream before we tear + # ours down, so viewers don't get disconnected. + has_clients = ( + channel_id in self.client_managers + and self.client_managers[channel_id].get_client_count() > 0 + ) + if has_clients: + logger.warning( + f"Ownership lost for {channel_id} but {self.client_managers[channel_id].get_client_count()} " + f"client(s) still connected — deferring cleanup to next cycle" + ) + continue logger.error(f"Failed to re-acquire ownership for {channel_id}, will clean up") # For channels we don't own, check if they've been stopped/cleaned up in Redis From 6d7130d71e4d2f70df8999dbb29ff6cbc4c607a0 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 11:06:52 -0600 Subject: [PATCH 3/4] Bug Fix: fix get_instance deadlock and non-atomic ownership acquisition If ProxyServer() raises during singleton construction, _instance was left as _INITIALIZING permanently, causing all subsequent get_instance() callers to spin in an infinite gevent.sleep() loop. Wrap construction in try/except and reset _instance to None on failure so callers can retry. Also replace the non-atomic setnx() + expire() pair in try_acquire_ownership() with a single atomic SET NX EX call, consistent with the approach already used in extend_ownership() and eliminating the race window where a crash between the two calls could leave a key with no TTL (permanent ownership lock). --- apps/proxy/ts_proxy/server.py | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index b0b1a08e..609a3000 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -43,13 +43,17 @@ class ProxyServer: return inst if inst is None: cls._instance = cls._INITIALIZING - from .server import ProxyServer - from .stream_manager import StreamManager - from .stream_buffer import StreamBuffer - from .client_manager import ClientManager - real_instance = ProxyServer() - cls._instance = real_instance - return real_instance + try: + from .server import ProxyServer + from .stream_manager import StreamManager + from .stream_buffer import StreamBuffer + from .client_manager import ClientManager + real_instance = ProxyServer() + cls._instance = real_instance + return real_instance + except Exception: + cls._instance = None # Reset so next call can retry + raise # Another greenlet is initializing — wait for completion while True: inst = cls._instance @@ -391,20 +395,16 @@ class ProxyServer: # Create a lock key with proper namespace lock_key = RedisKeys.channel_owner(channel_id) - # Use Redis SETNX for atomic locking with error handling + # Use atomic SET NX EX for locking with error handling acquired = self._execute_redis_command( - lambda: self.redis_client.setnx(lock_key, self.worker_id) + lambda: self.redis_client.set(lock_key, self.worker_id, nx=True, ex=ttl) ) if acquired is None: # Redis command failed logger.warning(f"Redis command failed during ownership acquisition - assuming ownership") return True - # If acquired, set expiry to prevent orphaned locks if acquired: - self._execute_redis_command( - lambda: self.redis_client.expire(lock_key, ttl) - ) logger.info(f"Worker {self.worker_id} acquired ownership of channel {channel_id}") return True From d6e4a34f3b78f14a480db00203576b3e826931fb Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 11:11:50 -0600 Subject: [PATCH 4/4] changelog: Update changelog for proxy server pr. --- CHANGELOG.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c6606c9b..fadc44e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Channel table filter for channels that have stale streams: A new "Has Stale Streams" filter option in the channel table header menu highlights and filters channels containing at least one stale stream. Channels with stale streams are visually distinguished with an orange tint. The filter is mutually exclusive with "Only Empty Channels". - Thanks [@JCBird1012](https://github.com/JCBird1012) +### Fixed + +- 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. + - **No recovery from expired ownership**: `get_channel_owner()` called `redis.get()` twice inside a lambda (TOCTOU race — key could expire between calls); `extend_ownership()` silently returned `False` on expiry with no re-acquisition; and the non-owner cleanup path unconditionally killed streams even when the worker held the `stream_manager`. Fixed with a single `GET` in `get_channel_owner()`, re-acquisition via atomic `SET NX EX` in `extend_ownership()`, and a re-acquisition attempt with client-aware cleanup deferral in the cleanup thread. +- `get_instance()` deadlock: if `ProxyServer()` raised an exception during singleton construction, `_instance` was left permanently as the `_INITIALIZING` sentinel, causing all subsequent `get_instance()` callers to spin in an infinite `gevent.sleep()` loop. Construction is now wrapped in `try/except`; on failure `_instance` resets to `None` so the next call can retry. +- Non-atomic ownership acquisition in `try_acquire_ownership()`: replaced the separate `setnx()` + `expire()` calls with a single atomic `SET NX EX`, eliminating the race window where a process crash between the two calls could leave an ownership key with no TTL (permanent ownership lock). + ## [0.20.2] - 2026-03-03 ### Security