From 9393f72cc120200537709ec6996483cd846db170 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 14 Jun 2026 18:59:59 -0500 Subject: [PATCH] fix(proxy): enhance channel teardown and client management - Improved handling of channel teardown to prevent client reconnect issues across uWSGI workers, ensuring proper resource cleanup and ownership management. - Added functionality to clear all client entries from Redis for a channel, enhancing client management during shutdown. - Updated logging and response mechanisms to inform clients of channel availability during teardown,. - Enhanced stream manager behavior to ensure upstream connections are properly managed during ownership transitions. --- CHANGELOG.md | 2 + apps/channels/tests/test_ts_proxy_teardown.py | 232 ++++++++++++++ apps/proxy/live_proxy/client_manager.py | 16 + apps/proxy/live_proxy/input/manager.py | 60 +++- apps/proxy/live_proxy/server.py | 294 +++++++++++++----- .../live_proxy/services/channel_service.py | 99 +++++- apps/proxy/live_proxy/views.py | 21 +- 7 files changed, 625 insertions(+), 99 deletions(-) create mode 100644 apps/channels/tests/test_ts_proxy_teardown.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c5aa324a..2622dfc2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - **Live proxy channels could remain running with no clients after disconnect.** `stop_channel` called `log_system_event('channel_stop')` synchronously before local cleanup and Redis key deletion; Connect integrations or DB event writes on that path could block teardown indefinitely while the cleanup thread kept refreshing metadata TTL for a stale `stream_buffers` entry. Teardown now stops ffmpeg/output managers, releases ownership, removes local buffers and managers, deletes Redis keys, and only then logs the stop event asynchronously. Metadata TTL refresh skips channels mid-shutdown and non-owned channels; a stuck-stop watchdog forces cleanup if `stop_channel` does not return within ~30s. Stream buffers also ignore late `add_chunk()` writes once shutdown begins (`buffer.stopping` set at the start of `stream_manager.stop()`). +- **Client reconnect during channel teardown could wedge the channel across uWSGI workers.** The disconnect path called `proxy_server.stop_channel()` directly, so other workers were not notified and reconnecting clients could attach while ownership was released on the owner worker. The cleanup thread then re-acquired expired ownership and looped on orphaned clients while the upstream connection stayed open. All lifecycle stops now go through coordinated Redis teardown (`channel_stopping` flag, metadata `stopping` state, and `CHANNEL_STOP` pubsub). New stream requests are rejected with `503 Retry-After` during pending shutdown or active teardown; `initialize_channel()` and ownership re-acquisition refuse to proceed in those states; non-owner workers clean local resources only on pubsub; and orphaned-client sweeps force cleanup when teardown is already active. (Fixes #1342) +- **Orphaned upstream threads kept writing Redis after teardown.** Partial cleanup could remove a channel from `stream_managers` while the `stream-{uuid}` OS thread kept running, so orphan metadata sweeps only deleted Redis keys and chunks/metadata were immediately recreated (bare `total_bytes` hash with TTL -1). Stream managers now stop upstream when ownership is lost, orphan cleanup stops local ffmpeg/stream threads even when registry entries are gone (`_live_stream_managers` tracks managers until the thread exits), and forced recovery stops processes before Redis deletion instead of dropping dict entries without calling `stop()`. - **Stale `channel_stream` Redis keys after a channel stopped could skip connection accounting on retune.** On `dev`, `get_stream()` reused any existing `channel_stream` / `stream_profile` assignment without reserving a new slot. If those keys were left behind after a stop, the next tune-in could reach the provider without incrementing Redis counters. `get_stream()` now releases stale assignments when proxy metadata shows the channel is inactive, then reserves fresh slots. - **EPG auto-match reliability fixes.** - Memory could spike to multiple GB on large EPG sources when building a full in-memory catalog before fuzzy matching; single-channel matching now streams rows and bounds ML work to a small candidate set. diff --git a/apps/channels/tests/test_ts_proxy_teardown.py b/apps/channels/tests/test_ts_proxy_teardown.py new file mode 100644 index 00000000..14b4e898 --- /dev/null +++ b/apps/channels/tests/test_ts_proxy_teardown.py @@ -0,0 +1,232 @@ +"""Tests for multi-worker channel teardown coordination.""" +import time +from unittest.mock import MagicMock, patch + +from django.test import TestCase + +from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState +from apps.proxy.live_proxy.input.manager import StreamManager +from apps.proxy.live_proxy.redis_keys import RedisKeys +from apps.proxy.live_proxy.server import ProxyServer +from apps.proxy.live_proxy.services.channel_service import ChannelService + + +CHANNEL_ID = "00000000-0000-0000-0000-000000000099" + + +def _mock_proxy_server(redis_client=None): + server = MagicMock() + server.redis_client = redis_client or MagicMock() + return server + + +class ChannelTeardownAvailabilityTests(TestCase): + @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance") + def test_teardown_active_when_stopping_key_exists(self, mock_get_instance): + redis = MagicMock() + redis.exists.side_effect = lambda key: key == RedisKeys.channel_stopping(CHANNEL_ID) + mock_get_instance.return_value = _mock_proxy_server(redis) + + self.assertTrue(ChannelService.is_channel_teardown_active(CHANNEL_ID)) + + @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance") + def test_teardown_active_when_metadata_state_is_stopping(self, mock_get_instance): + redis = MagicMock() + redis.exists.return_value = False + redis.hget.return_value = ChannelState.STOPPING.encode() + mock_get_instance.return_value = _mock_proxy_server(redis) + + self.assertTrue(ChannelService.is_channel_teardown_active(CHANNEL_ID)) + + @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.channel_shutdown_delay") + @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance") + def test_shutdown_pending_within_delay_window(self, mock_get_instance, mock_delay): + mock_delay.return_value = 5 + redis = MagicMock() + redis.exists.return_value = False + redis.get.return_value = str(time.time() - 2).encode() + mock_get_instance.return_value = _mock_proxy_server(redis) + + self.assertTrue(ChannelService.is_shutdown_pending(CHANNEL_ID)) + self.assertTrue(ChannelService.is_channel_unavailable_for_new_clients(CHANNEL_ID)) + + @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.channel_shutdown_delay") + @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance") + def test_shutdown_pending_expired_after_delay(self, mock_get_instance, mock_delay): + mock_delay.return_value = 5 + redis = MagicMock() + redis.exists.return_value = False + redis.get.return_value = str(time.time() - 10).encode() + mock_get_instance.return_value = _mock_proxy_server(redis) + + self.assertFalse(ChannelService.is_shutdown_pending(CHANNEL_ID)) + + +class LocalStreamActivityTests(TestCase): + def _make_server(self): + with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()): + server = ProxyServer() + server.worker_id = "testhost:1" + server.stream_managers = {} + server.stream_buffers = {} + server.client_managers = {} + server.profile_managers = {} + server.profile_buffers = {} + server._live_stream_managers = {} + server._stopping_channels = set() + server.redis_client = MagicMock() + server.stop_all_output_formats = MagicMock() + server.stop_all_output_profiles = MagicMock() + return server + + def test_has_local_stream_activity_from_live_registry(self): + server = self._make_server() + server._live_stream_managers[CHANNEL_ID] = MagicMock() + self.assertTrue(server._has_local_stream_activity(CHANNEL_ID)) + + @patch.object(ProxyServer, "_join_stream_thread") + def test_stop_local_stream_activity_stops_live_manager(self, mock_join): + server = self._make_server() + manager = MagicMock() + server._live_stream_managers[CHANNEL_ID] = manager + + server._stop_local_stream_activity(CHANNEL_ID) + + manager.stop.assert_called_once() + mock_join.assert_called_once_with(CHANNEL_ID) + self.assertNotIn(CHANNEL_ID, server._live_stream_managers) + + +class OrphanMetadataCleanupTests(TestCase): + def _make_server(self): + with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()): + server = ProxyServer() + server.worker_id = "testhost:1" + server.stream_managers = {} + server.stream_buffers = {} + server.client_managers = {} + server.profile_managers = {} + server.profile_buffers = {} + server._live_stream_managers = {} + server._stopping_channels = set() + server.redis_client = MagicMock() + return server + + @patch.object(ProxyServer, "_clean_redis_keys") + @patch.object(ProxyServer, "_stop_local_stream_activity") + @patch.object(ProxyServer, "_has_local_stream_activity", return_value=True) + def test_orphan_metadata_stops_local_processes_before_redis( + self, mock_has_local, mock_stop_local, mock_clean_redis + ): + server = self._make_server() + metadata_key = RedisKeys.channel_metadata(CHANNEL_ID) + server.redis_client.keys.return_value = [metadata_key.encode()] + server.redis_client.hgetall.return_value = { + b"owner": b"", + b"state": b"unknown", + } + server.redis_client.exists.return_value = False + server.redis_client.scard.return_value = 0 + + server._check_orphaned_metadata() + + mock_has_local.assert_called_with(CHANNEL_ID) + mock_stop_local.assert_called_once_with(CHANNEL_ID) + mock_clean_redis.assert_called_once_with(CHANNEL_ID) + + @patch.object(ProxyServer, "_clean_redis_keys") + @patch.object(ProxyServer, "_stop_local_stream_activity") + @patch.object(ProxyServer, "_has_local_stream_activity", return_value=False) + def test_orphan_metadata_remote_channel_only_cleans_redis( + self, mock_has_local, mock_stop_local, mock_clean_redis + ): + server = self._make_server() + metadata_key = RedisKeys.channel_metadata(CHANNEL_ID) + server.redis_client.keys.return_value = [metadata_key.encode()] + server.redis_client.hgetall.return_value = {b"owner": b"", b"state": b"unknown"} + server.redis_client.exists.return_value = False + server.redis_client.scard.return_value = 0 + + server._check_orphaned_metadata() + + mock_stop_local.assert_not_called() + mock_clean_redis.assert_called_once_with(CHANNEL_ID) + + +class OrphanChannelCleanupTests(TestCase): + def _make_server(self): + with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()): + server = ProxyServer() + server.worker_id = "testhost:1" + server.stream_managers = {} + server.stream_buffers = {} + server.client_managers = {} + server.profile_managers = {} + server.profile_buffers = {} + server._live_stream_managers = {} + server._stopping_channels = set() + server.redis_client = MagicMock() + server.get_channel_owner = MagicMock(return_value=None) + return server + + @patch.object(ProxyServer, "_clean_redis_keys") + @patch.object(ProxyServer, "_stop_local_stream_activity") + @patch.object(ProxyServer, "_has_local_stream_activity", return_value=True) + def test_orphan_channel_stops_local_before_redis( + self, mock_has_local, mock_stop_local, mock_clean_redis + ): + server = self._make_server() + metadata_key = RedisKeys.channel_metadata(CHANNEL_ID) + server.redis_client.keys.return_value = [metadata_key.encode()] + server.redis_client.scard.return_value = 0 + + server._check_orphaned_channels() + + mock_stop_local.assert_called_once_with(CHANNEL_ID) + mock_clean_redis.assert_called_once_with(CHANNEL_ID) + + +class StreamManagerOwnershipTests(TestCase): + def test_still_owner_false_when_different_worker(self): + buffer = MagicMock() + buffer.redis_client.get.side_effect = lambda key: ( + None if "stopping" in key else b"worker-b" + ) + buffer.redis_client.exists.return_value = False + manager = StreamManager( + CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a" + ) + + self.assertFalse(manager._still_owner()) + + def test_still_owner_true_when_owner_lock_expired_but_not_stopping(self): + buffer = MagicMock() + buffer.redis_client.get.return_value = None + buffer.redis_client.exists.return_value = False + manager = StreamManager( + CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a" + ) + + self.assertTrue(manager._still_owner()) + + def test_still_owner_false_when_channel_stopping_key_set(self): + buffer = MagicMock() + buffer.redis_client.exists.return_value = True + manager = StreamManager( + CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a" + ) + + self.assertFalse(manager._still_owner()) + + def test_update_bytes_skipped_after_ownership_lost(self): + buffer = MagicMock() + buffer.redis_client.get.return_value = b"other-worker" + buffer.redis_client.exists.return_value = False + manager = StreamManager( + CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a" + ) + manager.bytes_processed = 1000 + + manager._update_bytes_processed(500) + + buffer.redis_client.hincrby.assert_not_called() diff --git a/apps/proxy/live_proxy/client_manager.py b/apps/proxy/live_proxy/client_manager.py index 3ab1861c..3771d55c 100644 --- a/apps/proxy/live_proxy/client_manager.py +++ b/apps/proxy/live_proxy/client_manager.py @@ -442,3 +442,19 @@ class ClientManager: ) return stale_ids + + @staticmethod + def clear_all_clients(redis_client, channel_id): + """Remove every client entry from Redis for a channel.""" + client_set_key = RedisKeys.clients(channel_id) + client_ids = redis_client.smembers(client_set_key) + if not client_ids: + return 0 + + pipe = redis_client.pipeline(transaction=False) + for cid in client_ids: + cid_str = cid.decode() if isinstance(cid, bytes) else cid + pipe.delete(RedisKeys.client_metadata(channel_id, cid_str)) + pipe.delete(client_set_key) + pipe.execute() + return len(client_ids) diff --git a/apps/proxy/live_proxy/input/manager.py b/apps/proxy/live_proxy/input/manager.py index b560923b..06b12644 100644 --- a/apps/proxy/live_proxy/input/manager.py +++ b/apps/proxy/live_proxy/input/manager.py @@ -182,6 +182,47 @@ class StreamManager: logger.warning(f"Timeout waiting for existing processes to close for channel {self.channel_id} after {timeout}s") return False + def _still_owner(self): + """Return True while this worker should keep the upstream connection open.""" + if self.stopping or self.stop_requested: + return False + + if not self.worker_id: + return True + + redis_client = getattr(self.buffer, 'redis_client', None) + if not redis_client: + return True + + try: + stop_key = RedisKeys.channel_stopping(self.channel_id) + if redis_client.exists(stop_key): + return False + + owner_key = RedisKeys.channel_owner(self.channel_id) + current_owner = redis_client.get(owner_key) + if not current_owner: + # Owner lock TTL may lapse briefly between cleanup-thread renewals. + # Keep running unless coordinated teardown has started (checked above). + return True + if isinstance(current_owner, bytes): + current_owner = current_owner.decode() + return current_owner == self.worker_id + except Exception as e: + logger.debug(f"Ownership check failed for channel {self.channel_id}: {e}") + return False + + def _ensure_owner_or_stop(self): + if self._still_owner(): + return True + + logger.warning( + f"Stream manager for channel {self.channel_id} lost ownership " + f"(worker {self.worker_id}) - stopping upstream" + ) + self.stop() + return False + def run(self): """Main execution loop using HTTP streaming with improved connection handling and stream switching""" # Add a stop flag to the class properties @@ -203,6 +244,8 @@ class StreamManager: # Main stream switching loop - we'll try different streams if needed while self.running and stream_switch_attempts <= max_stream_switches: close_old_connections() + if not self._ensure_owner_or_stop(): + break # Check for stuck switching state if self.url_switching and time.time() - self.url_switch_start_time > self.url_switch_timeout: logger.warning(f"URL switching state appears stuck for channel {self.channel_id} " @@ -255,6 +298,8 @@ class StreamManager: continue # Connection retry loop for current URL while self.running and self.retry_count < self.max_retries and not url_failed and not self.needs_stream_switch: + if not self._ensure_owner_or_stop(): + break logger.info(f"Connection attempt {self.retry_count + 1}/{self.max_retries} for URL: {self.url} for channel {self.channel_id}") @@ -382,6 +427,12 @@ class StreamManager: except Exception as e: logger.error(f"Stream error: {e}", exc_info=True) finally: + try: + from ..server import ProxyServer + ProxyServer.get_instance()._live_stream_managers.pop(self.channel_id, None) + except Exception: + pass + # Enhanced cleanup in the finally block self.connected = False @@ -1020,6 +1071,9 @@ class StreamManager: def _update_bytes_processed(self, chunk_size): """Update the total bytes processed in Redis metadata""" + if not self._still_owner(): + return + try: # Update local counter self.bytes_processed += chunk_size @@ -1584,6 +1638,10 @@ class StreamManager: self.connected = False return False + if not self._still_owner(): + self.stop() + return False + # Track chunk size before adding to buffer chunk_size = len(chunk) self._update_bytes_processed(chunk_size) @@ -1592,7 +1650,7 @@ class StreamManager: success = self.buffer.add_chunk(chunk) # Update last data timestamp in Redis if successful - if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: + if success and self._still_owner() and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: last_data_key = RedisKeys.last_data(self.buffer.channel_id) self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py index 7145fcd4..a348c876 100644 --- a/apps/proxy/live_proxy/server.py +++ b/apps/proxy/live_proxy/server.py @@ -72,6 +72,8 @@ class ProxyServer: self._channel_names = {} self._stopping_channels = set() # channels with an active stop_channel call in progress self._stopping_since = {} # channel_id -> time.time() when stop_channel began + # Managers kept until the stream OS thread exits (may outlive stream_managers dict) + self._live_stream_managers = {} # Generate a unique worker ID pid = os.getpid() @@ -295,22 +297,36 @@ class ProxyServer: json.dumps(switch_result) ) elif event_type == EventType.CHANNEL_STOP: - logger.info(f"Received {EventType.CHANNEL_STOP} event for channel {channel_id}") - # First mark channel as stopping in Redis - if self.redis_client: - # Set stopping state in metadata - metadata_key = RedisKeys.channel_metadata(channel_id) - if self.redis_client.exists(metadata_key): - self.redis_client.hset(metadata_key, mapping={ - "state": ChannelState.STOPPING, - "state_changed_at": str(time.time()) - }) + requester_worker_id = data.get("requester_worker_id") + logger.info( + f"Received {EventType.CHANNEL_STOP} event for channel {channel_id} " + f"from worker {requester_worker_id}" + ) - # If we have local resources for this channel, clean them up - if channel_id in self.stream_buffers or channel_id in self.client_managers: - # Use existing stop_channel method - logger.info(f"Stopping local resources for channel {channel_id}") + # Initiating worker already runs local stop_channel() + if requester_worker_id and requester_worker_id == self.worker_id: + logger.debug( + f"Ignoring {EventType.CHANNEL_STOP} for {channel_id}; " + f"this worker initiated teardown" + ) + elif ( + channel_id in self.stream_managers + or channel_id in self._live_stream_managers + ): + logger.info( + f"Owner worker stopping local upstream for channel {channel_id}" + ) self.stop_channel(channel_id) + elif ( + channel_id in self.stream_buffers + or channel_id in self.client_managers + or channel_id in self.profile_managers + or channel_id in self.output_managers + ): + logger.info( + f"Non-owner worker cleaning local resources for channel {channel_id}" + ) + self._cleanup_local_resources(channel_id) # Acknowledge stop by publishing a response stop_response = { @@ -491,8 +507,16 @@ class ProxyServer: current = self.redis_client.get(lock_key) if current is None: - # Key expired — re-acquire if we have the stream_manager + # Key expired, re-acquire if we have the stream_manager, but never + # during coordinated teardown (multi-worker reconnect-during-stop race). if channel_id in self.stream_managers: + if self._channel_unavailable_for_new_clients(channel_id): + logger.info( + f"Refusing ownership re-acquisition for {channel_id}; " + f"teardown or pending shutdown active" + ) + return False + 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}") @@ -515,6 +539,13 @@ class ProxyServer: def initialize_channel(self, url, channel_id, user_agent=None, transcode=False, stream_id=None): """Initialize a channel without redundant active key""" try: + if self._channel_unavailable_for_new_clients(channel_id): + logger.warning( + f"Refusing to initialize channel {channel_id}; " + f"teardown or pending shutdown active" + ) + return False + if self.redis_client: metadata_key = RedisKeys.channel_metadata(channel_id) if self.redis_client.exists(metadata_key): @@ -721,6 +752,7 @@ class ProxyServer: self.client_managers[channel_id] = client_manager # Start stream manager thread only for the owner + self._live_stream_managers[channel_id] = stream_manager thread = threading.Thread(target=stream_manager.run, daemon=True) thread.name = f"stream-{channel_id}" thread.start() @@ -970,10 +1002,34 @@ class ProxyServer: self.redis_client.delete(disconnect_key) return - self.stop_channel(channel_id) + self._coordinated_stop_channel(channel_id) except Exception as e: logger.error(f"Error handling client disconnect for channel {channel_id}: {e}") + def _channel_unavailable_for_new_clients(self, channel_id): + """True when new clients or ownership re-acquisition should be refused.""" + from .services.channel_service import ChannelService + return ChannelService.is_channel_unavailable_for_new_clients(channel_id) + + def _channel_teardown_active(self, channel_id): + """True when a coordinated stop is in progress (Redis-visible to all workers).""" + from .services.channel_service import ChannelService + return ChannelService.is_channel_teardown_active(channel_id) + + def _coordinated_stop_channel(self, channel_id): + """Stop a channel with Redis markers and pubsub visible to all uWSGI workers.""" + from .services.channel_service import ChannelService + return ChannelService.stop_channel(channel_id) + + def _force_clear_channel_clients(self, channel_id): + """Remove all client entries from Redis for a wedged channel.""" + if not self.redis_client: + return 0 + removed = ClientManager.clear_all_clients(self.redis_client, channel_id) + if removed: + logger.warning(f"Force-cleared {removed} client(s) from channel {channel_id}") + return removed + # ------------------------------------------------------------------ # Output format lifecycle # ------------------------------------------------------------------ @@ -1308,40 +1364,90 @@ class ProxyServer: gevent.spawn(_log_stop) - def _cleanup_stopped_channel_local(self, channel_id): - """Remove local managers/buffers for a channel that is shutting down.""" + def _get_stream_thread(self, channel_id): + thread_name = f"stream-{channel_id}" + for thread in threading.enumerate(): + if thread.name == thread_name: + return thread + return None + + def _has_local_stream_activity(self, channel_id): if channel_id in self.stream_managers: - del self.stream_managers[channel_id] - logger.info(f"Removed stream manager for channel {channel_id}") + return True + if channel_id in self._live_stream_managers: + return True + if channel_id in self.stream_buffers: + return True + if channel_id in self.client_managers: + return True + thread = self._get_stream_thread(channel_id) + return thread is not None and thread.is_alive() + + def _join_stream_thread(self, channel_id, timeout=2.0): + thread = self._get_stream_thread(channel_id) + if thread and thread.is_alive(): + logger.info(f"Waiting for stream thread to terminate for channel {channel_id}") + try: + thread.join(timeout=timeout) + if thread.is_alive(): + logger.warning( + f"Stream thread for channel {channel_id} did not terminate within {timeout}s" + ) + except RuntimeError: + logger.debug(f"Could not join stream thread for channel {channel_id}") + + def _resolve_stream_manager(self, channel_id): + manager = self.stream_managers.pop(channel_id, None) + if manager is None: + manager = self._live_stream_managers.pop(channel_id, None) + return manager + + def _stop_local_stream_activity(self, channel_id): + """Stop local ffmpeg/stream threads regardless of registry state.""" + stream_manager = self._resolve_stream_manager(channel_id) + if stream_manager is not None: + if hasattr(stream_manager, 'stop'): + try: + stream_manager.stop() + except Exception as e: + logger.error(f"Error stopping stream manager for channel {channel_id}: {e}") + logger.info(f"Stopped stream manager for channel {channel_id}") + elif self._get_stream_thread(channel_id): + logger.warning( + f"Found live stream thread for channel {channel_id} without a manager reference" + ) + if channel_id in self.stream_buffers: + self.stream_buffers[channel_id].stopping = True + + self._join_stream_thread(channel_id) if channel_id in self.stream_buffers: - buffer = self.stream_buffers[channel_id] + buffer = self.stream_buffers.pop(channel_id) if hasattr(buffer, 'stop'): try: buffer.stop() - logger.debug(f"Buffer for channel {channel_id} properly stopped") except Exception as e: - logger.error(f"Error stopping buffer: {e}") - - try: - if channel_id in self.stream_buffers: - del self.stream_buffers[channel_id] - logger.info(f"Removed stream buffer for channel {channel_id}") - except KeyError: - logger.debug(f"Buffer for channel {channel_id} already removed") + logger.error(f"Error stopping buffer for channel {channel_id}: {e}") + logger.info(f"Removed stream buffer for channel {channel_id}") if channel_id in self.client_managers: try: - client_manager = self.client_managers[channel_id] + client_manager = self.client_managers.pop(channel_id) if hasattr(client_manager, 'stop'): client_manager.stop() - del self.client_managers[channel_id] logger.info(f"Removed client manager for channel {channel_id}") except KeyError: logger.debug(f"Client manager for channel {channel_id} already removed") + self.stop_all_output_formats(channel_id) + self.stop_all_output_profiles(channel_id) self.profile_managers.pop(channel_id, None) self.profile_buffers.pop(channel_id, None) + self._live_stream_managers.pop(channel_id, None) + + def _cleanup_stopped_channel_local(self, channel_id): + """Remove local managers/buffers for a channel that is shutting down.""" + self._stop_local_stream_activity(channel_id) def _recover_stuck_channel_stops(self): """Force cleanup when stop_channel never finishes (e.g. blocked on logging).""" @@ -1363,14 +1469,11 @@ class ProxyServer: self._stopping_channels.discard(channel_id) self._stopping_since.pop(channel_id, None) - if (channel_id in self.stream_buffers - or channel_id in self.client_managers - or channel_id in self.stream_managers): - try: - self._cleanup_stopped_channel_local(channel_id) - self._clean_redis_keys(channel_id) - except Exception as e: - logger.error(f"Error during forced cleanup for channel {channel_id}: {e}") + try: + self._stop_local_stream_activity(channel_id) + self._clean_redis_keys(channel_id) + except Exception as e: + logger.error(f"Error during forced cleanup for channel {channel_id}: {e}") def stop_channel(self, channel_id): """Stop a channel with proper ownership handling""" @@ -1386,7 +1489,10 @@ class ProxyServer: # First set a stopping key that clients will check if self.redis_client: stop_key = RedisKeys.channel_stopping(channel_id) - self.redis_client.setex(stop_key, 10, "true") + if self.redis_client.exists(stop_key): + self.redis_client.expire(stop_key, 60) + else: + self.redis_client.setex(stop_key, 60, "true") # Only stop the actual stream manager if we're the owner if self.am_i_owner(channel_id): @@ -1456,7 +1562,7 @@ class ProxyServer: for channel_id in channels_to_stop: logger.info(f"Auto-stopping inactive channel {channel_id}") - self.stop_channel(channel_id) + self._coordinated_stop_channel(channel_id) def _cleanup_channel(self, channel_id: str) -> None: """Remove channel resources""" @@ -1467,7 +1573,7 @@ class ProxyServer: def shutdown(self) -> None: """Stop all channels and cleanup""" for channel_id in list(self.stream_managers.keys()): - self.stop_channel(channel_id) + self._coordinated_stop_channel(channel_id) def _start_cleanup_thread(self): """Start background thread to maintain ownership and clean up resources""" @@ -1577,7 +1683,7 @@ class ProxyServer: f"Channel {channel_id} in {channel_state} state with 0 clients for {time_since_ready:.1f}s " f"(after reaching ready, shutdown_delay: {shutdown_delay}s) - stopping channel" ) - self.stop_channel(channel_id) + self._coordinated_stop_channel(channel_id) continue else: # Never reached ready - use grace_period timeout @@ -1589,7 +1695,7 @@ class ProxyServer: f"Channel {channel_id} stuck in {channel_state} state for {time_since_start:.1f}s " f"with no clients (timeout: {connecting_timeout}s) - stopping channel due to upstream issues" ) - self.stop_channel(channel_id) + self._coordinated_stop_channel(channel_id) continue elif connection_ready_time: # We have clients now, but check grace period for state transition @@ -1643,7 +1749,7 @@ class ProxyServer: elif current_time - disconnect_time > ConfigHelper.channel_shutdown_delay(): # We've had no clients for the shutdown delay period logger.warning(f"No clients for {current_time - disconnect_time:.1f}s, stopping channel {channel_id}") - self.stop_channel(channel_id) + self._coordinated_stop_channel(channel_id) else: # Still in shutdown delay period logger.debug(f"Channel {channel_id} shutdown timer: " @@ -1663,6 +1769,15 @@ class ProxyServer: # don't fight the shutdown by trying to re-acquire. if channel_id in self._stopping_channels: continue + + if self._channel_unavailable_for_new_clients(channel_id): + logger.info( + f"Channel {channel_id} teardown active with local stream_manager; " + f"finishing stop instead of re-acquiring ownership" + ) + self.stop_channel(channel_id) + continue + logger.warning( f"Ownership gap for {channel_id}: this worker has stream_manager " f"but am_i_owner returned False. Attempting re-acquisition." @@ -1815,13 +1930,12 @@ class ProxyServer: # Orphaned channel with no clients - clean it up logger.info(f"Cleaning up orphaned channel {channel_id}") - # If we have it locally, stop it properly to clean up processes - if channel_id in self.stream_managers or channel_id in self.client_managers: - logger.info(f"Orphaned channel {channel_id} is local - calling stop_channel") - self.stop_channel(channel_id) - else: - # Just clean up Redis keys for remote channels - self._clean_redis_keys(channel_id) + if self._has_local_stream_activity(channel_id): + logger.info( + f"Orphaned channel {channel_id} has local stream activity - stopping processes" + ) + self._stop_local_stream_activity(channel_id) + self._clean_redis_keys(channel_id) except Exception as e: logger.error(f"Error processing channel key {key}: {e}") @@ -1850,11 +1964,9 @@ class ProxyServer: if not metadata: # Empty metadata - clean it up logger.warning(f"Found empty metadata for channel {channel_id} - cleaning up") - # If we have it locally, stop it properly - if channel_id in self.stream_managers or channel_id in self.client_managers: - self.stop_channel(channel_id) - else: - self._clean_redis_keys(channel_id) + if self._has_local_stream_activity(channel_id): + self._stop_local_stream_activity(channel_id) + self._clean_redis_keys(channel_id) continue # Get owner @@ -1875,13 +1987,12 @@ class ProxyServer: state = metadata.get('state', 'unknown') logger.warning(f"Found orphaned metadata for channel {channel_id} (state: {state}, owner: {owner}, clients: {client_count}) - cleaning up") - # If we have it locally, stop it properly to clean up transcode/proxy processes - if channel_id in self.stream_managers or channel_id in self.client_managers: - logger.info(f"Channel {channel_id} is local - calling stop_channel to clean up processes") - self.stop_channel(channel_id) - else: - # Just clean up Redis keys for remote channels - self._clean_redis_keys(channel_id) + if self._has_local_stream_activity(channel_id): + logger.info( + f"Channel {channel_id} has local stream activity - stopping processes" + ) + self._stop_local_stream_activity(channel_id) + self._clean_redis_keys(channel_id) elif not owner_alive and client_count > 0: # SCARD may include ghost entries from a dead worker's # expired metadata hashes. Validate before deciding. @@ -1897,16 +2008,25 @@ class ProxyServer: f"owner: {owner}) had {client_count} ghost client(s) " f"- cleaning up" ) - if channel_id in self.stream_managers or channel_id in self.client_managers: - self.stop_channel(channel_id) - else: - self._clean_redis_keys(channel_id) + if self._has_local_stream_activity(channel_id): + self._stop_local_stream_activity(channel_id) + self._clean_redis_keys(channel_id) else: - logger.warning( - f"Orphaned channel {channel_id} still has " - f"{real_count} live client(s) after ghost removal " - f"- may need ownership takeover" - ) + if self._channel_teardown_active(channel_id): + logger.warning( + f"Orphaned channel {channel_id} still has " + f"{real_count} client(s) during teardown; forcing cleanup" + ) + self._force_clear_channel_clients(channel_id) + if self._has_local_stream_activity(channel_id): + self._stop_local_stream_activity(channel_id) + self._clean_redis_keys(channel_id) + else: + logger.warning( + f"Orphaned channel {channel_id} still has " + f"{real_count} live client(s) after ghost removal " + f"- may need ownership takeover" + ) except Exception as e: logger.error(f"Error processing metadata key {key}: {e}", exc_info=True) @@ -2023,14 +2143,28 @@ class ProxyServer: def _cleanup_local_resources(self, channel_id): """Clean up local resources for a channel without affecting Redis keys""" try: - if channel_id in self.stream_managers: - if hasattr(self.stream_managers[channel_id], 'stop'): - self.stream_managers[channel_id].stop() - del self.stream_managers[channel_id] - logger.info(f"Non-owner cleanup: Removed stream manager for channel {channel_id}") + stream_manager = self._resolve_stream_manager(channel_id) + if stream_manager is not None: + if hasattr(stream_manager, 'stop'): + stream_manager.stop() + logger.info(f"Non-owner cleanup: Stopped stream manager for channel {channel_id}") + elif self._get_stream_thread(channel_id): + logger.warning( + f"Non-owner cleanup: Found orphaned stream thread for channel {channel_id}" + ) + if channel_id in self.stream_buffers: + self.stream_buffers[channel_id].stopping = True + + self._join_stream_thread(channel_id) + self._live_stream_managers.pop(channel_id, None) if channel_id in self.stream_buffers: - del self.stream_buffers[channel_id] + buffer = self.stream_buffers.pop(channel_id) + if hasattr(buffer, 'stop'): + try: + buffer.stop() + except Exception as e: + logger.error(f"Non-owner cleanup: Error stopping buffer for {channel_id}: {e}") logger.info(f"Non-owner cleanup: Removed stream buffer for channel {channel_id}") if channel_id in self.client_managers: diff --git a/apps/proxy/live_proxy/services/channel_service.py b/apps/proxy/live_proxy/services/channel_service.py index 8cc4a030..6671e444 100644 --- a/apps/proxy/live_proxy/services/channel_service.py +++ b/apps/proxy/live_proxy/services/channel_service.py @@ -10,6 +10,7 @@ from apps.channels.models import Channel, Stream from ..server import ProxyServer from ..redis_keys import RedisKeys from ..constants import EventType, ChannelState, ChannelMetadataField +from ..config_helper import ConfigHelper from ..url_utils import get_stream_info_for_switch from core.utils import log_system_event from .log_parsers import LogParserFactory @@ -19,6 +20,82 @@ logger = logging.getLogger("live_proxy") class ChannelService: """Service class for channel operations""" + @staticmethod + def mark_channel_stopping(channel_id, broadcast=False): + """Mark a channel as stopping in Redis so all uWSGI workers converge on teardown.""" + proxy_server = ProxyServer.get_instance() + if not proxy_server.redis_client: + return False + + try: + metadata_key = RedisKeys.channel_metadata(channel_id) + if proxy_server.redis_client.exists(metadata_key): + proxy_server.redis_client.hset(metadata_key, mapping={ + ChannelMetadataField.STATE: ChannelState.STOPPING, + ChannelMetadataField.STATE_CHANGED_AT: str(time.time()), + }) + + stop_key = RedisKeys.channel_stopping(channel_id) + proxy_server.redis_client.setex(stop_key, 60, "true") + + if broadcast: + ChannelService._publish_channel_stop_event(channel_id) + return True + except Exception as e: + logger.error(f"Error marking channel {channel_id} as stopping: {e}") + return False + + @staticmethod + def is_shutdown_pending(channel_id): + """True while the post-disconnect shutdown delay has started but stop has not run.""" + proxy_server = ProxyServer.get_instance() + if not proxy_server.redis_client: + return False + + delay = ConfigHelper.channel_shutdown_delay() + if delay <= 0: + return False + + disconnect_value = proxy_server.redis_client.get( + RedisKeys.last_client_disconnect(channel_id) + ) + if not disconnect_value: + return False + + try: + disconnect_time = float(disconnect_value) + except (ValueError, TypeError): + return False + + return (time.time() - disconnect_time) < delay + + @staticmethod + def is_channel_teardown_active(channel_id): + """True when a coordinated channel stop is in progress (visible to all workers).""" + proxy_server = ProxyServer.get_instance() + if not proxy_server.redis_client: + return False + + if proxy_server.redis_client.exists(RedisKeys.channel_stopping(channel_id)): + return True + + metadata_key = RedisKeys.channel_metadata(channel_id) + state = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STATE) + if state: + state_str = state.decode() if isinstance(state, bytes) else state + if state_str == ChannelState.STOPPING: + return True + + return False + + @staticmethod + def is_channel_unavailable_for_new_clients(channel_id): + """Reject new stream requests while teardown is active or shutdown is pending.""" + return ( + ChannelService.is_channel_teardown_active(channel_id) + or ChannelService.is_shutdown_pending(channel_id) + ) + @staticmethod def initialize_channel(channel_id, stream_url, user_agent, transcode=False, stream_profile_value=None, stream_id=None, m3u_profile_id=None, channel_name=None, stream_name=None): """ @@ -39,6 +116,7 @@ class ChannelService: bool: Success status """ proxy_server = ProxyServer.get_instance() + if stream_id and proxy_server.redis_client: metadata_key = RedisKeys.channel_metadata(channel_id) # Check if metadata already exists @@ -263,29 +341,16 @@ class ChannelService: try: metadata = proxy_server.redis_client.hgetall(metadata_key) if metadata and 'state' in metadata: - state = metadata['state'] - channel_info = {"state": state} - - # Immediately mark as stopping in metadata so clients detect it faster - proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.STOPPING) - proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STATE_CHANGED_AT, str(time.time())) + channel_info = {"state": metadata['state']} except Exception as e: logger.error(f"Error fetching channel state: {e}") - # Set stopping flag with higher TTL to ensure it persists + # Mark stopping in Redis and notify all workers before local teardown if proxy_server.redis_client: - stop_key = RedisKeys.channel_stopping(channel_id) - proxy_server.redis_client.setex(stop_key, 60, "true") # Higher TTL of 60 seconds - logger.info(f"Set channel stopping flag with 60s TTL for channel {channel_id}") - - # Broadcast stop event to all workers via PubSub - if proxy_server.redis_client: - ChannelService._publish_channel_stop_event(channel_id) - - # Also stop locally to ensure this worker cleans up right away + ChannelService.mark_channel_stopping(channel_id, broadcast=True) + logger.info(f"Marked channel {channel_id} stopping and broadcast stop to all workers") local_result = proxy_server.stop_channel(channel_id) else: - # No Redis, just stop locally local_result = proxy_server.stop_channel(channel_id) # Release the channel in the channel model if applicable diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py index d60d9703..b547cefd 100644 --- a/apps/proxy/live_proxy/views.py +++ b/apps/proxy/live_proxy/views.py @@ -43,6 +43,15 @@ from apps.proxy.utils import check_user_stream_limits logger = get_logger() +def _channel_stopping_response(): + response = JsonResponse( + {"error": "Channel is stopping, retry shortly"}, + status=503, + ) + response["Retry-After"] = "1" + return response + + def _resolve_output_format(user, force=None, request=None): """Return the output format string to use for this client.""" _FORMAT_ALIASES = { @@ -123,6 +132,12 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): status=429 ) + if ChannelService.is_channel_unavailable_for_new_clients(channel_id): + logger.info( + f"[{client_id}] Channel {channel_id} unavailable. Teardown or pending shutdown" + ) + return _channel_stopping_response() + # Check if we need to reinitialize the channel needs_initialization = True channel_state = None @@ -144,7 +159,6 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): ChannelState.BUFFERING, ChannelState.INITIALIZING, ChannelState.CONNECTING, - ChannelState.STOPPING, ]: needs_initialization = False logger.debug( @@ -160,6 +174,11 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): logger.debug( f"[{client_id}] Channel {channel_id} is still initializing, client will wait" ) + elif channel_state == ChannelState.STOPPING: + logger.info( + f"[{client_id}] Channel {channel_id} is stopping, rejecting request" + ) + return _channel_stopping_response() # Terminal states - channel needs cleanup before reinitialization elif channel_state in [ ChannelState.ERROR,