mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 00:55:50 +00:00
Merge pull request #1035 from PFalko:fix/stream-ownership-bugs
Fix stream ownership bugs causing streams to die after 30-200s
This commit is contained in:
SergeantPanda
GitHub
commit
0db177b937
4 changed files with 88 additions and 23 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -34,18 +34,32 @@ 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:
|
||||
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
|
||||
inst = cls._instance
|
||||
if inst is not None and inst is not cls._INITIALIZING:
|
||||
return inst
|
||||
if inst is None:
|
||||
cls._instance = cls._INITIALIZING
|
||||
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
|
||||
if inst is not None and inst is not cls._INITIALIZING:
|
||||
return inst
|
||||
gevent.sleep(0.05)
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize proxy server with worker identification"""
|
||||
|
|
@ -353,9 +367,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
|
||||
|
|
@ -374,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
|
||||
|
||||
|
|
@ -433,7 +450,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 +458,23 @@ 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.set(lock_key, self.worker_id, nx=True, ex=ttl)
|
||||
if acquired:
|
||||
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 +1203,33 @@ 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:
|
||||
# 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
|
||||
if self.redis_client:
|
||||
# Method 1: Check for stopping key
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue