From f6aab5bc3ce49b31eedbb00f55030fe048905d48 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 10 Mar 2025 20:07:11 -0500 Subject: [PATCH] Stream switching is working fairly consistently need to investigate some issues with it. --- apps/proxy/ts_proxy/server.py | 143 ++++++++++++++++++++++++---------- apps/proxy/ts_proxy/views.py | 127 +++++++++++++++++++++--------- 2 files changed, 192 insertions(+), 78 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index d10d5876..a0c7a2ac 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -96,27 +96,40 @@ class StreamManager: logging.info("Stream manager resources released") def _process_complete_packets(self): - """Process TS packets with detailed logging""" + """Process TS packets with improved resync capability""" try: - # Find sync byte if needed - if not self.sync_found and len(self.recv_buffer) >= 376: - for i in range(min(188, len(self.recv_buffer) - 188)): - # Look for at least two sync bytes (0x47) at 188-byte intervals - if (self.recv_buffer[i] == 0x47 and - self.recv_buffer[i + 188] == 0x47): - - # Trim buffer to start at first sync byte - self.recv_buffer = self.recv_buffer[i:] - self.sync_found = True - logging.debug(f"TS sync found at position {i}") - break + # Enhanced sync byte detection with re-sync capability + if (not self.sync_found or + (len(self.recv_buffer) >= 188 and self.recv_buffer[0] != 0x47)): - # If sync not found, keep last 188 bytes and return - if not self.sync_found: - if len(self.recv_buffer) > 188: - self.recv_buffer = self.recv_buffer[-188:] - return False + # Need to find sync pattern if we haven't found it yet or lost sync + if len(self.recv_buffer) >= 376: # Need at least 2 packet lengths + sync_found = False + # Look for at least two sync bytes (0x47) at 188-byte intervals + for i in range(min(188, len(self.recv_buffer) - 188)): + if (self.recv_buffer[i] == 0x47 and + self.recv_buffer[i + 188] == 0x47): + + # If already had sync but lost it, log the issue + if self.sync_found: + logging.warning(f"Re-syncing TS stream at position {i} (lost sync)") + else: + logging.debug(f"TS sync found at position {i}") + + # Trim buffer to start at first sync byte + self.recv_buffer = self.recv_buffer[i:] + self.sync_found = True + sync_found = True + break + + # If we couldn't find sync in this buffer, discard partial data + if not sync_found: + logging.warning(f"Failed to find sync pattern - discarding {len(self.recv_buffer) - 188} bytes") + if len(self.recv_buffer) > 188: + self.recv_buffer = self.recv_buffer[-188:] # Keep last chunk for next attempt + return False + # If we don't have a complete packet yet, wait for more data if len(self.recv_buffer) < 188: return False @@ -127,8 +140,17 @@ class StreamManager: if packet_count == 0: return False - # Log packet processing - logging.debug(f"Processing {packet_count} TS packets ({packet_count * 188} bytes)") + # Verify all packets have sync bytes + all_synced = True + for i in range(0, packet_count): + if self.recv_buffer[i * 188] != 0x47: + all_synced = False + break + + # If not all packets are synced, re-scan for sync + if not all_synced: + self.sync_found = False # Force re-sync on next call + return False # Extract complete packets packets = self.recv_buffer[:packet_count * 188] @@ -144,6 +166,8 @@ class StreamManager: if first_sync != 0x47 or last_sync != 0x47: logging.warning(f"TS packet alignment issue: first_sync=0x{first_sync:02x}, last_sync=0x{last_sync:02x}") + # Don't process misaligned packets + return False before_index = self.buffer.index success = self.buffer.add_chunk(bytes(packets)) @@ -1123,10 +1147,29 @@ class ProxyServer: if self.redis_client: # Store stream metadata - can be done regardless of ownership metadata_key = f"ts_proxy:channel:{channel_id}:metadata" - if url: # Only update URL if one is provided - self.redis_client.hset(metadata_key, "url", url) + + # FIXED: Use hash operations consistently - don't mix string and hash operations + # Create a dictionary of values to set in the hash + metadata = {} + if url: # Only include URL if provided + metadata["url"] = url if user_agent: - self.redis_client.hset(metadata_key, "user_agent", user_agent) + metadata["user_agent"] = user_agent + + # Initialize state + metadata["state"] = "initializing" + metadata["init_time"] = str(time.time()) + + # Set the hash fields all at once + if metadata: + self.redis_client.hset(metadata_key, mapping=metadata) + + # Set expiration on the hash + self.redis_client.expire(metadata_key, 3600) # 1 hour TTL + + # Set activity key as a separate key + activity_key = f"ts_proxy:active_channel:{channel_id}" + self.redis_client.setex(activity_key, 300, "1") # 5 min TTL # If no url was passed, try to get from Redis if not url: @@ -1222,6 +1265,21 @@ class ProxyServer: logging.info(f"Channel {channel_id} in connecting state - will start grace period after connection") + # SIMPLE CHANNEL REGISTRY - register this channel as active + if self.redis_client: + # Use a simple key for active channel registration + registry_key = f"ts_proxy:active_channels:{channel_id}" + # Store basic info and set a longer TTL (5 minutes) + channel_info = { + "url": url if url else "", + "init_time": str(time.time()), + "owner": self.worker_id + } + self.redis_client.hset(registry_key, mapping=channel_info) + self.redis_client.expire(registry_key, 300) # 5 minute TTL + + logging.info(f"Registered channel {channel_id} in active channels registry") + return True except Exception as e: @@ -1231,27 +1289,16 @@ class ProxyServer: return False def check_if_channel_exists(self, channel_id): - """Check if a channel exists in Redis or locally""" - # Check local memory first + """Simple check if a channel exists using the registry""" + # Check local memory first (quick check) if channel_id in self.stream_managers or channel_id in self.stream_buffers: return True - # Check Redis for channel metadata + # Simple registry check in Redis if self.redis_client: - metadata_key = f"ts_proxy:channel:{channel_id}:metadata" - if self.redis_client.exists(metadata_key): - return True - - # Also check for active channel marker - activity_key = f"ts_proxy:active_channel:{channel_id}" - if self.redis_client.exists(activity_key): - return True - - # Check for any buffer index - buffer_key = f"ts_proxy:buffer:{channel_id}:index" - if self.redis_client.exists(buffer_key): - return True - + registry_key = f"ts_proxy:active_channels:{channel_id}" + return bool(self.redis_client.exists(registry_key)) + return False def stop_channel(self, channel_id): @@ -1412,6 +1459,9 @@ class ProxyServer: # Rest of the cleanup thread... + # Refresh active channel registry + self.refresh_channel_registry() + except Exception as e: logging.error(f"Error in cleanup thread: {e}", exc_info=True) @@ -1491,3 +1541,16 @@ class ProxyServer: except Exception as e: logging.error(f"Error cleaning Redis keys for channel {channel_id}: {e}") + def refresh_channel_registry(self): + """Refresh TTL for active channels in registry""" + if not self.redis_client: + return + + # Refresh registry entries for channels we own + for channel_id in self.stream_managers.keys(): + registry_key = f"ts_proxy:active_channels:{channel_id}" + if self.redis_client.exists(registry_key): + # Update last_active timestamp and extend TTL + self.redis_client.hset(registry_key, "last_active", str(time.time())) + self.redis_client.expire(registry_key, 300) # 5 minute TTL + diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 4fb2b731..271ae380 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -307,61 +307,113 @@ def stream_ts(request, channel_id): @csrf_exempt @require_http_methods(["POST"]) def change_stream(request, channel_id): - """Change stream URL for existing channel with improved multi-worker support""" + """Change stream URL for existing channel with enhanced diagnostics""" try: data = json.loads(request.body) new_url = data.get('url') - user_agent = data.get('user_agent') # Optional user agent + user_agent = data.get('user_agent') if not new_url: return JsonResponse({'error': 'No URL provided'}, status=400) - # Check if channel exists using the proper Redis check - if not proxy_server.check_if_channel_exists(channel_id): - logger.error(f"Channel {channel_id} not found in any worker or Redis") - return JsonResponse({'error': 'Channel not found'}, status=404) - - logger.info(f"Channel {channel_id} found, checking if current worker is owner") + logger.info(f"Attempting to change stream URL for channel {channel_id} to {new_url}") + # Enhanced channel detection + in_local_managers = channel_id in proxy_server.stream_managers + in_local_buffers = channel_id in proxy_server.stream_buffers + + # First check Redis directly before using our wrapper method + redis_keys = None + if proxy_server.redis_client: + try: + redis_keys = proxy_server.redis_client.keys(f"ts_proxy:*:{channel_id}*") + redis_keys = [k.decode('utf-8') for k in redis_keys] if redis_keys else [] + except Exception as e: + logger.error(f"Error checking Redis keys: {e}") + + # Now use our standard check + channel_exists = proxy_server.check_if_channel_exists(channel_id) + + # Log detailed diagnostics + logger.info(f"Channel {channel_id} diagnostics: " + f"in_local_managers={in_local_managers}, " + f"in_local_buffers={in_local_buffers}, " + f"redis_keys_count={len(redis_keys) if redis_keys else 0}, " + f"channel_exists={channel_exists}") + + if not channel_exists: + # If channel doesn't exist but we found Redis keys, force initialize it + if redis_keys: + logger.warning(f"Channel {channel_id} not detected by check_if_channel_exists but Redis keys exist. Forcing initialization.") + proxy_server.initialize_channel(new_url, channel_id, user_agent) + else: + logger.error(f"Channel {channel_id} not found in any worker or Redis") + return JsonResponse({ + 'error': 'Channel not found', + 'diagnostics': { + 'in_local_managers': in_local_managers, + 'in_local_buffers': in_local_buffers, + 'redis_keys': redis_keys, + } + }, status=404) + + # Update metadata in Redis regardless of ownership - this ensures URL is updated + # even if the owner worker is handling another request + if proxy_server.redis_client: + try: + metadata_key = f"ts_proxy:channel:{channel_id}:metadata" + + # First check if the key exists and what type it is + key_type = proxy_server.redis_client.type(metadata_key).decode('utf-8') + logger.debug(f"Redis key {metadata_key} is of type: {key_type}") + + # Use the appropriate method based on the key type + if key_type == 'hash': + proxy_server.redis_client.hset(metadata_key, "url", new_url) + if user_agent: + proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent) + elif key_type == 'none': # Key doesn't exist yet + # Create new hash with all required fields + metadata = {"url": new_url} + if user_agent: + metadata["user_agent"] = user_agent + proxy_server.redis_client.hset(metadata_key, mapping=metadata) + else: + # If key exists with wrong type, delete it and recreate + proxy_server.redis_client.delete(metadata_key) + metadata = {"url": new_url} + if user_agent: + metadata["user_agent"] = user_agent + proxy_server.redis_client.hset(metadata_key, mapping=metadata) + + # Set switch request flag to ensure all workers see it + switch_key = f"ts_proxy:channel:{channel_id}:switch_request" + proxy_server.redis_client.setex(switch_key, 30, new_url) # 30 second TTL + + logger.info(f"Updated metadata for channel {channel_id} in Redis") + except Exception as e: + logger.error(f"Error updating Redis metadata: {e}", exc_info=True) + # If we're the owner, update directly if proxy_server.am_i_owner(channel_id) and channel_id in proxy_server.stream_managers: - logger.info(f"Changing stream URL for channel {channel_id} (owner worker)") + logger.info(f"This worker is the owner, changing stream URL for channel {channel_id}") manager = proxy_server.stream_managers[channel_id] - - # Update metadata in Redis first - if proxy_server.redis_client: - metadata_key = f"ts_proxy:channel:{channel_id}:metadata" - proxy_server.redis_client.hset(metadata_key, "url", new_url) - if user_agent: - proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent) + old_url = manager.url # Update the stream - old_url = manager.url - manager.url = new_url - manager.connected = False - manager._close_socket() # Close existing connection - - logger.info(f"Stream URL changed from {old_url} to {new_url}") + result = manager.update_url(new_url) + logger.info(f"Stream URL changed from {old_url} to {new_url}, result: {result}") return JsonResponse({ 'message': 'Stream URL updated', 'channel': channel_id, 'url': new_url, - 'owner': True + 'owner': True, + 'worker_id': proxy_server.worker_id }) - # If we're not the owner, use Redis to request the change + # If we're not the owner, publish an event for the owner to pick up else: - logger.info(f"Requesting stream URL change for channel {channel_id} (non-owner worker)") - - if not proxy_server.redis_client: - return JsonResponse({'error': 'Redis not available, cannot request stream switch'}, status=500) - - # Update metadata in Redis first (all workers can do this) - metadata_key = f"ts_proxy:channel:{channel_id}:metadata" - proxy_server.redis_client.hset(metadata_key, "url", new_url) - if user_agent: - proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent) - + logger.info(f"This worker is not the owner, requesting URL change via Redis PubSub") # Publish switch request event switch_request = { "event": "stream_switch", @@ -377,13 +429,12 @@ def change_stream(request, channel_id): json.dumps(switch_request) ) - logger.info(f"Published stream switch request to {new_url}") - return JsonResponse({ 'message': 'Stream URL change requested', 'channel': channel_id, 'url': new_url, - 'owner': False + 'owner': False, + 'worker_id': proxy_server.worker_id }) except json.JSONDecodeError: