mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 00:55:50 +00:00
fix: resolve decode_responses migration bugs and user-limit regressions
Fix a wave of bugs introduced by the user-priority branch's migration
from manual .decode('utf-8') calls to decode_responses=True on the
metadata Redis client:
core/utils.py
- Rewrite _init_client: fix SyntaxError at line 138, missing
redis_password/redis_user params, cls._client corruption when
decode_responses=False was requested, and dead retry backoff logic
apps/proxy/utils.py
- Fix get_user_limit_settings() → get_user_limits_settings() (2 sites)
- Fix VOD scan key vod_persistent_connections:* → vod_persistent_connection:*
- Fix undefined channel_id in VOD loop (use content_uuid from Redis hash)
- Remove leftover print(active_connections) debug statement
- Refactor manual cursor SCAN while-loops to scan_iter()
apps/channels/tasks.py
- Fix closure bug in _d(): md.get(key) → md.get(bkey) (was reading
the outer loop variable instead of the local bytes key)
apps/proxy/ts_proxy/server.py
- Replace stale b'init_time', b'total_bytes', b'state' byte-string
key lookups and .decode() calls with plain string equivalents
apps/proxy/ts_proxy/services/channel_service.py
- Fix ChannelMetadataField.STATE.encode() / .OWNER.encode() →
.STATE / .OWNER (pre-existing, broke under decoded client)
apps/proxy/ts_proxy/views.py
- Fix ChannelMetadataField.OWNER.encode("utf-8") → .OWNER
apps/proxy/ts_proxy/client_manager.py
- Fix cid.decode('utf-8') in remove_ghost_clients(): smembers()
already returns str with decode_responses=True
This commit is contained in:
SergeantPanda
parent
90a600b48f
commit
b671a72707
8 changed files with 114 additions and 149 deletions
|
|
@ -2494,7 +2494,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
|
|||
md = r.hgetall(metadata_key)
|
||||
if md:
|
||||
def _d(bkey, cast=str):
|
||||
v = md.get(key)
|
||||
v = md.get(bkey)
|
||||
try:
|
||||
if v is None:
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -428,8 +428,7 @@ class ClientManager:
|
|||
client_id_list = list(client_ids)
|
||||
pipe = redis_client.pipeline()
|
||||
for cid in client_id_list:
|
||||
cid_str = cid.decode('utf-8')
|
||||
pipe.exists(RedisKeys.client_metadata(channel_id, cid_str))
|
||||
pipe.exists(RedisKeys.client_metadata(channel_id, cid))
|
||||
results = pipe.execute()
|
||||
|
||||
stale_ids = [
|
||||
|
|
|
|||
|
|
@ -940,14 +940,14 @@ class ProxyServer:
|
|||
metadata = self.redis_client.hgetall(metadata_key)
|
||||
if metadata:
|
||||
# Calculate runtime from init_time
|
||||
if b'init_time' in metadata:
|
||||
if 'init_time' in metadata:
|
||||
try:
|
||||
init_time = float(metadata['init_time'])
|
||||
runtime = round(time.time() - init_time, 2)
|
||||
except Exception:
|
||||
pass
|
||||
# Get total bytes transferred
|
||||
if b'total_bytes' in metadata:
|
||||
if 'total_bytes' in metadata:
|
||||
try:
|
||||
total_bytes = int(metadata['total_bytes'])
|
||||
except Exception:
|
||||
|
|
@ -1113,9 +1113,9 @@ class ProxyServer:
|
|||
|
||||
# Also get init time as a fallback
|
||||
init_time = None
|
||||
if metadata and b'init_time' in metadata:
|
||||
if metadata and 'init_time' in metadata:
|
||||
try:
|
||||
init_time = float(metadata[b'init_time'])
|
||||
init_time = float(metadata['init_time'])
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
|
|
@ -1400,7 +1400,7 @@ class ProxyServer:
|
|||
real_count = max(0, client_count - len(stale_ids))
|
||||
if real_count <= 0:
|
||||
# No real clients remain — safe to clean up.
|
||||
state = metadata.get(b'state', b'unknown').decode('utf-8') if b'state' in metadata else 'unknown'
|
||||
state = metadata.get('state', 'unknown')
|
||||
logger.warning(
|
||||
f"Orphaned channel {channel_id} (state: {state}, "
|
||||
f"owner: {owner}) had {client_count} ghost client(s) "
|
||||
|
|
|
|||
|
|
@ -382,8 +382,8 @@ class ChannelService:
|
|||
metadata = proxy_server.redis_client.hgetall(metadata_key)
|
||||
|
||||
# Extract state and owner
|
||||
state = metadata.get(ChannelMetadataField.STATE.encode(), 'unknown')
|
||||
owner = metadata.get(ChannelMetadataField.OWNER.encode(), 'unknown')
|
||||
state = metadata.get(ChannelMetadataField.STATE, 'unknown')
|
||||
owner = metadata.get(ChannelMetadataField.OWNER, 'unknown')
|
||||
|
||||
# Valid states indicate channel is running properly
|
||||
valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING]
|
||||
|
|
@ -432,13 +432,13 @@ class ChannelService:
|
|||
try:
|
||||
# Use factory to parse the line based on stream type
|
||||
parsed_data = LogParserFactory.parse(stream_type, stream_info_line)
|
||||
|
||||
|
||||
if not parsed_data:
|
||||
return
|
||||
|
||||
# Update Redis and database with parsed data
|
||||
ChannelService._update_stream_info_in_redis(
|
||||
channel_id,
|
||||
channel_id,
|
||||
parsed_data.get('video_codec'),
|
||||
parsed_data.get('resolution'),
|
||||
parsed_data.get('width'),
|
||||
|
|
|
|||
|
|
@ -127,7 +127,7 @@ def stream_ts(request, channel_id, user=None):
|
|||
)
|
||||
# Unknown/empty state - check if owner is alive
|
||||
else:
|
||||
owner_field = ChannelMetadataField.OWNER.encode("utf-8")
|
||||
owner_field = ChannelMetadataField.OWNER
|
||||
if owner_field in metadata:
|
||||
owner = metadata[owner_field]
|
||||
owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ def attempt_stream_termination(user_id, requesting_client_id, active_connections
|
|||
try:
|
||||
logger.info("[stream limits]" f"[{requesting_client_id}] User {user_id} has {len(active_connections)} active connections, checking termination candidates")
|
||||
|
||||
user_limit_settings = CoreSettings.get_user_limit_settings()
|
||||
user_limit_settings = CoreSettings.get_user_limits_settings()
|
||||
terminate_oldest = user_limit_settings.get("terminate_oldest", True)
|
||||
prioritize_single = user_limit_settings.get("prioritize_single_client_channels", True)
|
||||
ignore_same_channel = user_limit_settings.get("ignore_same_channel_connections", False)
|
||||
|
|
@ -79,74 +79,58 @@ def get_user_active_connections(user_id):
|
|||
connections = []
|
||||
|
||||
try:
|
||||
cursor = 0
|
||||
|
||||
# Grab live streams
|
||||
while True:
|
||||
cursor, keys = redis_client.scan(cursor=cursor, match="ts_proxy:channel:*:clients:*", count=1000)
|
||||
for key in redis_client.scan_iter(match="ts_proxy:channel:*:clients:*", count=1000):
|
||||
parts = key.split(':')
|
||||
if len(parts) >= 5:
|
||||
channel_id = parts[2]
|
||||
client_id = parts[4]
|
||||
|
||||
for key in keys:
|
||||
parts = key.split(':')
|
||||
if len(parts) >= 5:
|
||||
channel_id = parts[2]
|
||||
client_id = parts[4]
|
||||
client_user_id = redis_client.hget(key, 'user_id')
|
||||
connected_at = redis_client.hget(key, 'connected_at')
|
||||
|
||||
client_user_id = redis_client.hget(key, 'user_id')
|
||||
connected_at = redis_client.hget(key, 'connected_at')
|
||||
logger.info(f"[stream limits] user_id = {user_id}")
|
||||
logger.info(f"[stream limits] channel_id = {channel_id}")
|
||||
logger.info(f"[stream limits] client_id = {client_id}")
|
||||
|
||||
logger.info(f"[stream limits] user_id = {user_id}")
|
||||
logger.info(f"[stream limits] channel_id = {channel_id}")
|
||||
logger.info(f"[stream limits] client_id = {client_id}")
|
||||
|
||||
if client_user_id and int(client_user_id) == user_id:
|
||||
try:
|
||||
logger.info(f"[stream limits] Found LIVE connection for user {user_id} on channel {channel_id} with client ID {client_id}")
|
||||
connected_at = float(connected_at) if connected_at else 0
|
||||
connections.append({
|
||||
'media_id': channel_id,
|
||||
'client_id': client_id,
|
||||
'connected_at': connected_at,
|
||||
'type': 'live',
|
||||
})
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
if cursor == 0:
|
||||
break
|
||||
|
||||
|
||||
cursor = 0
|
||||
if client_user_id and int(client_user_id) == user_id:
|
||||
try:
|
||||
logger.info(f"[stream limits] Found LIVE connection for user {user_id} on channel {channel_id} with client ID {client_id}")
|
||||
connected_at = float(connected_at) if connected_at else 0
|
||||
connections.append({
|
||||
'media_id': channel_id,
|
||||
'client_id': client_id,
|
||||
'connected_at': connected_at,
|
||||
'type': 'live',
|
||||
})
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
# Grab VOD
|
||||
while True:
|
||||
cursor, keys = redis_client.scan(cursor=cursor, match="vod_persistent_connections:*", count=1000)
|
||||
for key in redis_client.scan_iter(match="vod_persistent_connection:*", count=1000):
|
||||
parts = key.split(':')
|
||||
if len(parts) >= 2:
|
||||
client_id = parts[1]
|
||||
|
||||
for key in keys:
|
||||
parts = key.split(':')
|
||||
if len(parts) >= 2:
|
||||
client_id = parts[1]
|
||||
client_user_id = redis_client.hget(key, 'user_id')
|
||||
connected_at = redis_client.hget(key, 'created_at')
|
||||
content_uuid = redis_client.hget(key, 'content_uuid')
|
||||
|
||||
client_user_id = redis_client.hget(key, 'user_id')
|
||||
connected_at = redis_client.hget(key, 'created_at')
|
||||
logger.info(f"[stream limits] user_id = {user_id}")
|
||||
logger.info(f"[stream limits] client_id = {client_id}")
|
||||
|
||||
logger.info(f"[stream limits] user_id = {user_id}")
|
||||
logger.info(f"[stream limits] client_id = {client_id}")
|
||||
|
||||
if client_user_id and int(client_user_id) == user_id:
|
||||
try:
|
||||
logger.info(f"[stream limits] Found VOD connection for user {user_id} on channel {channel_id} with client ID {client_id}")
|
||||
connected_at = float(connected_at) if connected_at else 0
|
||||
connections.append({
|
||||
'media_id': channel_id,
|
||||
'client_id': client_id,
|
||||
'connected_at': connected_at,
|
||||
'type': 'vod',
|
||||
})
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
if cursor == 0:
|
||||
break
|
||||
if client_user_id and int(client_user_id) == user_id:
|
||||
try:
|
||||
logger.info(f"[stream limits] Found VOD connection for user {user_id} on content {content_uuid} with client ID {client_id}")
|
||||
connected_at = float(connected_at) if connected_at else 0
|
||||
connections.append({
|
||||
'media_id': content_uuid or client_id,
|
||||
'client_id': client_id,
|
||||
'connected_at': connected_at,
|
||||
'type': 'vod',
|
||||
})
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
return connections
|
||||
|
||||
|
|
@ -159,14 +143,12 @@ def check_user_stream_limits(user, client_id):
|
|||
# Check user stream limits
|
||||
if user and user.stream_limit > 0:
|
||||
logger.info("[stream limits]" f"[{client_id}] User {user.username} (ID: {user.id}) is requesting a stream (stream_limit: {user.stream_limit})")
|
||||
user_limit_settings = CoreSettings.get_user_limit_settings()
|
||||
user_limit_settings = CoreSettings.get_user_limits_settings()
|
||||
|
||||
active_connections = get_user_active_connections(user.id)
|
||||
unique_channel_count = set([conn['media_id'] for conn in active_connections])
|
||||
user_stream_count = len(unique_channel_count) if user_limit_settings.get("ignore_same_channel_connections", False) else len(active_connections)
|
||||
|
||||
print(active_connections)
|
||||
|
||||
logger.info(f"[stream limits]" f"[{client_id}] User {user.username} currently has {len(active_connections)} active connections across {len(unique_channel_count)} unique channels (counting method: {'unique channels' if user_limit_settings.get('ignore_same_channel_connections', False) else 'total connections'})")
|
||||
|
||||
if user.stream_limit > 0 and user_stream_count >= user.stream_limit:
|
||||
|
|
|
|||
109
core/utils.py
109
core/utils.py
|
|
@ -57,6 +57,8 @@ class RedisClient:
|
|||
redis_host = os.environ.get("REDIS_HOST", getattr(settings, 'REDIS_HOST', 'localhost'))
|
||||
redis_port = int(os.environ.get("REDIS_PORT", getattr(settings, 'REDIS_PORT', 6379)))
|
||||
redis_db = int(os.environ.get("REDIS_DB", getattr(settings, 'REDIS_DB', 0)))
|
||||
redis_password = os.environ.get("REDIS_PASSWORD", getattr(settings, 'REDIS_PASSWORD', ''))
|
||||
redis_user = os.environ.get("REDIS_USER", getattr(settings, 'REDIS_USER', ''))
|
||||
|
||||
# Use standardized settings
|
||||
socket_timeout = getattr(settings, 'REDIS_SOCKET_TIMEOUT', 5)
|
||||
|
|
@ -65,17 +67,23 @@ class RedisClient:
|
|||
socket_keepalive = getattr(settings, 'REDIS_SOCKET_KEEPALIVE', True)
|
||||
retry_on_timeout = getattr(settings, 'REDIS_RETRY_ON_TIMEOUT', True)
|
||||
|
||||
# TLS params from settings (empty dict when TLS is disabled)
|
||||
ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {})
|
||||
|
||||
# Create Redis client with better defaults
|
||||
client = redis.Redis(
|
||||
host=redis_host,
|
||||
port=redis_port,
|
||||
db=redis_db,
|
||||
password=redis_password if redis_password else None,
|
||||
username=redis_user if redis_user else None,
|
||||
socket_timeout=socket_timeout,
|
||||
socket_connect_timeout=socket_connect_timeout,
|
||||
socket_keepalive=socket_keepalive,
|
||||
health_check_interval=health_check_interval,
|
||||
retry_on_timeout=retry_on_timeout,
|
||||
decode_responses=decode_responses
|
||||
decode_responses=decode_responses,
|
||||
**ssl_params
|
||||
)
|
||||
|
||||
# Validate connection with ping
|
||||
|
|
@ -87,84 +95,34 @@ class RedisClient:
|
|||
client.config_set('save', '') # Disable RDB snapshots
|
||||
client.config_set('appendonly', 'no') # Disable AOF logging
|
||||
|
||||
# Set optimal memory settings with environment variable support
|
||||
# Get max memory from environment or use a larger default (512MB instead of 256MB)
|
||||
#max_memory = os.environ.get('REDIS_MAX_MEMORY', '512mb')
|
||||
#eviction_policy = os.environ.get('REDIS_EVICTION_POLICY', 'allkeys-lru')
|
||||
|
||||
# TLS params from settings (empty dict when TLS is disabled)
|
||||
ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {})
|
||||
|
||||
# Create Redis client with better defaults
|
||||
client = redis.Redis(
|
||||
host=redis_host,
|
||||
port=redis_port,
|
||||
db=redis_db,
|
||||
password=redis_password if redis_password else None,
|
||||
username=redis_user if redis_user else None,
|
||||
socket_timeout=socket_timeout,
|
||||
socket_connect_timeout=socket_connect_timeout,
|
||||
socket_keepalive=socket_keepalive,
|
||||
health_check_interval=health_check_interval,
|
||||
retry_on_timeout=retry_on_timeout,
|
||||
decode_responses=decode_responses,
|
||||
**ssl_params
|
||||
)
|
||||
|
||||
#logger.info(f"Redis configured with maxmemory={max_memory}, policy={eviction_policy}")
|
||||
|
||||
# Disable protected mode when in debug mode
|
||||
if os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true':
|
||||
client.config_set('protected-mode', 'no') # Disable protected mode in debug
|
||||
logger.warning("Redis protected mode disabled for debug environment")
|
||||
|
||||
# Set optimal memory settings with environment variable support
|
||||
# Get max memory from environment or use a larger default (512MB instead of 256MB)
|
||||
#max_memory = os.environ.get('REDIS_MAX_MEMORY', '512mb')
|
||||
#eviction_policy = os.environ.get('REDIS_EVICTION_POLICY', 'allkeys-lru')
|
||||
|
||||
# Apply memory settings
|
||||
#client.config_set('maxmemory-policy', eviction_policy)
|
||||
#client.config_set('maxmemory', max_memory)
|
||||
|
||||
#logger.info(f"Redis configured with maxmemory={max_memory}, policy={eviction_policy}")
|
||||
|
||||
# Disable protected mode when in debug mode
|
||||
if os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true':
|
||||
client.config_set('protected-mode', 'no') # Disable protected mode in debug
|
||||
logger.warning("Redis protected mode disabled for debug environment")
|
||||
|
||||
logger.trace("Redis persistence disabled for better performance")
|
||||
except redis.exceptions.ResponseError as e:
|
||||
# Improve error handling for Redis configuration errors
|
||||
if "OOM" in str(e):
|
||||
logger.error(f"Redis OOM during configuration: {e}")
|
||||
# Try to increase maxmemory as an emergency measure
|
||||
try:
|
||||
client.config_set('maxmemory', '768mb')
|
||||
logger.warning("Applied emergency Redis memory increase to 768MB")
|
||||
except:
|
||||
pass
|
||||
else:
|
||||
logger.error(f"Redis configuration error: {e}")
|
||||
|
||||
logger.info(f"Connected to Redis at {redis_host}:{redis_port}/{redis_db}")
|
||||
|
||||
cls._client = client
|
||||
break
|
||||
|
||||
except (ConnectionError, TimeoutError) as e:
|
||||
retry_count += 1
|
||||
_tls_hint = _REDIS_TLS_HINT if ssl_params else ""
|
||||
if retry_count >= max_retries:
|
||||
logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}{_tls_hint}")
|
||||
return None
|
||||
logger.trace("Redis persistence disabled for better performance")
|
||||
except redis.exceptions.ResponseError as e:
|
||||
# Improve error handling for Redis configuration errors
|
||||
if "OOM" in str(e):
|
||||
logger.error(f"Redis OOM during configuration: {e}")
|
||||
# Try to increase maxmemory as an emergency measure
|
||||
try:
|
||||
client.config_set('maxmemory', '768mb')
|
||||
logger.warning("Applied emergency Redis memory increase to 768MB")
|
||||
except:
|
||||
pass
|
||||
else:
|
||||
logger.error(f"Redis configuration error: {e}")
|
||||
|
||||
except Exception as e:
|
||||
_tls_hint = _REDIS_TLS_HINT if ssl_params else ""
|
||||
logger.error(f"Unexpected error connecting to Redis: {e}{_tls_hint}")
|
||||
logger.info(f"Connected to Redis at {redis_host}:{redis_port}/{redis_db}")
|
||||
|
||||
return client
|
||||
|
||||
except (ConnectionError, TimeoutError) as e:
|
||||
retry_count += 1
|
||||
_tls_hint = _REDIS_TLS_HINT if ssl_params else ""
|
||||
if retry_count >= max_retries:
|
||||
logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}{_tls_hint}")
|
||||
return None
|
||||
else:
|
||||
# Use exponential backoff for retries
|
||||
|
|
@ -173,10 +131,15 @@ class RedisClient:
|
|||
time.sleep(wait_time)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error connecting to Redis: {e}")
|
||||
_tls_hint = ""
|
||||
try:
|
||||
_tls_hint = _REDIS_TLS_HINT if ssl_params else ""
|
||||
except NameError:
|
||||
pass
|
||||
logger.error(f"Unexpected error connecting to Redis: {e}{_tls_hint}")
|
||||
return None
|
||||
|
||||
return client
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_client(cls, max_retries=5, retry_interval=1):
|
||||
|
|
|
|||
21
debug_redis_check.py
Normal file
21
debug_redis_check.py
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
#!/usr/bin/env python
|
||||
import os, sys
|
||||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'dispatcharr.settings')
|
||||
import django
|
||||
django.setup()
|
||||
|
||||
from core.utils import RedisClient
|
||||
|
||||
r = RedisClient.get_client()
|
||||
print(f"Client: {r}")
|
||||
print(f"decode_responses: {r.connection_pool.connection_kwargs.get('decode_responses')}")
|
||||
|
||||
keys = list(r.scan_iter(match='ts_proxy:channel:*:metadata', count=100))
|
||||
print(f"Found {len(keys)} metadata keys")
|
||||
for k in keys[:3]:
|
||||
print(f" Key: {k!r}")
|
||||
data = r.hgetall(k)
|
||||
field_names = list(data.keys())[:8]
|
||||
print(f" Fields ({len(data)} total): {field_names}")
|
||||
state = data.get('state', 'MISSING')
|
||||
print(f" state={state!r}")
|
||||
Loading…
Add table
Add a link
Reference in a new issue