diff --git a/CHANGELOG.md b/CHANGELOG.md index 01ab218f..e8cacfda 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,7 +47,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **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()`. - **Rapid channel switching could leave bare `buffer:index` keys in Redis.** `buffer.stop()` flushed a final partial chunk via `INCR` during teardown, creating a persistent index key (TTL -1) after chunk keys expired; under load overlapping disconnect and cleanup-thread stops could race on `_stop_local_stream_activity`, blocking on ffmpeg stderr join before `_clean_redis_keys` ever ran. Teardown no longer writes to Redis from `buffer.stop()`, disconnect uses coordinated stop only (no duplicate upstream stop), model `release_stream()` runs before Redis keys are scanned/deleted (so `profile_connections` counters are not left stuck), ownership is released after Redis cleanup but before the blocking local ffmpeg stop, and a `finally` block guarantees Redis cleanup if local teardown raises. -- **Live proxy could leak geventpool DB checkouts outside HTTP requests.** With `django-db-geventpool`, `connection.close()` returns handles to the per-worker pool (`MAX_CONNS=8`); without it, greenlets and OS threads keep connections checked out until the pool blocks on `pool.get()`. Proxy paths that touch the ORM outside Django's request cycle now call `close_old_connections()` after work completes: `log_system_event()` (all system events), `_clean_redis_keys()`, channel init/start logging, client disconnect cleanup (TS and fMP4 generators), and profile lookup in `_establish_transcode_connection()` before spawning ffmpeg. +- **Live proxy could leak geventpool DB checkouts outside HTTP requests.** With `django-db-geventpool`, `connection.close()` returns handles to the per-worker pool (`MAX_CONNS=8`); without it, greenlets and OS threads keep connections checked out until the pool blocks on `pool.get()`. Proxy paths that touch the ORM outside Django's request cycle now call `close_old_connections()` after work completes: `_clean_redis_keys()`, profile lookup in `_establish_transcode_connection()` before spawning ffmpeg, and fMP4 client disconnect cleanup when releasing streams (TS disconnect relies on `log_system_event()`). +- **System events could block callers on Connect and plugin handlers.** `log_system_event()` ran Connect subscriptions and plugin `"events"` hooks synchronously after the DB write, so slow integrations could stall live-proxy and streaming paths even when inserting the event row was fast. Connect/plugin dispatch now runs on a separate gevent when the hub is available (synchronously in Celery workers without a hub). `log_system_event()` and `dispatch_event_system()` also call `close_old_connections()` in `finally` blocks so integration work does not leave pool slots checked out on the caller or dispatch greenlet. - **Plugins could leak geventpool DB checkouts after UI or Connect event runs.** Third-party plugins run inline on uWSGI greenlets (manual actions and Connect `"events"` hooks) with no guaranteed connection cleanup at the plugin boundary. `PluginManager.run_action()` and `stop_plugin()` now call `close_old_connections()` in a `finally` block so each action returns its pool slot whether the plugin succeeds or raises. - **Zombie ffmpeg could survive owner-lock expiry and orphan Redis sweeps.** When the 30s owner lock lapsed under single-worker load, disconnect handling treated the worker as non-owner so coordinated stop never ran, while ffmpeg kept writing and the orphan sweeper only deleted Redis keys (recreating them immediately). Disconnect now re-acquires ownership when local upstream is still active, stops locally when the last client leaves without a lock, orphan cleanup stops local ffmpeg before Redis deletion via `_has_local_upstream_activity`, re-init stops lingering upstream before starting a duplicate thread, and `is_channel_teardown_active` includes channels mid-`stop_channel` on this worker so rapid reconnect gets 503 during teardown. - **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. diff --git a/Plugins.md b/Plugins.md index 97e95ed2..4dc15a55 100644 --- a/Plugins.md +++ b/Plugins.md @@ -317,7 +317,7 @@ Still follow these rules: - **Heavy or long work:** dispatch a Celery task (`.delay()`) and return quickly from `run()`. Celery workers close connections after each task; blocking the uWSGI gevent hub with `time.sleep`, sync HTTP, or large CPU work can freeze the whole worker regardless of DB cleanup. - **Background threads or greenlets you spawn:** each thread/greenlet that uses the ORM must call `close_old_connections()` (or `connection.close()`) in its own `finally` block when done. The wrapper only covers the thread/greenlet that called `run_action()`. -- **Connect event hooks:** actions with an `"events"` list run synchronously inside whatever caller triggered the event (for example a live-proxy system event). Keep event handlers short or defer to Celery. +- **Connect event hooks:** actions with an `"events"` list are dispatched from `log_system_event()` on a separate gevent when uWSGI has an active hub (otherwise synchronously, e.g. Celery). Keep handlers short or defer heavy work to Celery. ### Important: Don’t Ask Users for URL/User/Password Dispatcharr plugins run **inside** the Dispatcharr backend process. That means they already have direct access to the app’s models, tasks, and internal utilities. diff --git a/apps/proxy/live_proxy/input/manager.py b/apps/proxy/live_proxy/input/manager.py index e8315006..7d01a88f 100644 --- a/apps/proxy/live_proxy/input/manager.py +++ b/apps/proxy/live_proxy/input/manager.py @@ -650,48 +650,48 @@ class StreamManager: logger.debug(f"Closing existing HTTP connections before establishing transcode connection for channel {self.channel_id}") self._close_connection() - close_old_connections() - channel = get_stream_object(self.channel_id) + try: + channel = get_stream_object(self.channel_id) - # Use FFmpeg specifically for HLS streams - if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg: - from core.models import StreamProfile - try: - stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True) - logger.info("Using FFmpeg stream profile for unsupported proxy content (HLS/RTSP/UDP)") - except StreamProfile.DoesNotExist: - # Fall back to channel's profile if FFmpeg not found + # Use FFmpeg specifically for HLS streams + if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg: + from core.models import StreamProfile + try: + stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True) + logger.info("Using FFmpeg stream profile for unsupported proxy content (HLS/RTSP/UDP)") + except StreamProfile.DoesNotExist: + # Fall back to channel's profile if FFmpeg not found + stream_profile = channel.get_stream_profile() + logger.warning(f"FFmpeg profile not found, using channel default profile for channel: {self.channel_id}") + else: stream_profile = channel.get_stream_profile() - logger.warning(f"FFmpeg profile not found, using channel default profile for channel: {self.channel_id}") - else: - stream_profile = channel.get_stream_profile() - # Build and start transcode command - self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent) + # Build and start transcode command + self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent) - # Store stream command for efficient log parser routing - self.stream_command = stream_profile.command - # Map actual commands to parser types for direct routing - command_to_parser = { - 'ffmpeg': 'ffmpeg', - 'cvlc': 'vlc', - 'vlc': 'vlc', - 'streamlink': 'streamlink' - } - self.parser_type = command_to_parser.get(self.stream_command.lower()) - if self.parser_type: - logger.debug(f"Using {self.parser_type} parser for log parsing (command: {self.stream_command})") - else: - logger.debug(f"Unknown stream command '{self.stream_command}', will use auto-detection for log parsing") + # Store stream command for efficient log parser routing + self.stream_command = stream_profile.command + # Map actual commands to parser types for direct routing + command_to_parser = { + 'ffmpeg': 'ffmpeg', + 'cvlc': 'vlc', + 'vlc': 'vlc', + 'streamlink': 'streamlink' + } + self.parser_type = command_to_parser.get(self.stream_command.lower()) + if self.parser_type: + logger.debug(f"Using {self.parser_type} parser for log parsing (command: {self.stream_command})") + else: + logger.debug(f"Unknown stream command '{self.stream_command}', will use auto-detection for log parsing") - # For UDP streams, remove any user_agent parameters from the command - if hasattr(self, 'stream_type') and self.stream_type == StreamType.UDP: - # Filter out any arguments that contain the user_agent value or related headers - self.transcode_cmd = [arg for arg in self.transcode_cmd if self.user_agent not in arg and 'user-agent' not in arg.lower() and 'user_agent' not in arg.lower()] - logger.debug(f"Removed user_agent parameters from UDP stream command for channel: {self.channel_id}") - - # Profile lookup is done; release the pool slot before a long-lived ffmpeg process. - close_old_connections() + # For UDP streams, remove any user_agent parameters from the command + if hasattr(self, 'stream_type') and self.stream_type == StreamType.UDP: + # Filter out any arguments that contain the user_agent value or related headers + self.transcode_cmd = [arg for arg in self.transcode_cmd if self.user_agent not in arg and 'user-agent' not in arg.lower() and 'user_agent' not in arg.lower()] + logger.debug(f"Removed user_agent parameters from UDP stream command for channel: {self.channel_id}") + finally: + # Release the pool slot before posix_spawn or before returning on profile errors. + close_old_connections() logger.debug(f"Starting transcode process: {self.transcode_cmd} for channel: {self.channel_id}") diff --git a/apps/proxy/live_proxy/output/ts/generator.py b/apps/proxy/live_proxy/output/ts/generator.py index 69dc341f..efde02ad 100644 --- a/apps/proxy/live_proxy/output/ts/generator.py +++ b/apps/proxy/live_proxy/output/ts/generator.py @@ -8,7 +8,6 @@ import gevent from apps.proxy.config import TSConfig as Config from apps.channels.models import Channel, Stream from core.utils import log_system_event -from django.db import close_old_connections from ...server import ProxyServer from ...utils import create_ts_packet, get_logger from ...redis_keys import RedisKeys @@ -591,62 +590,59 @@ class StreamGenerator: def _cleanup(self): """Clean up resources and report final statistics.""" - try: - # Client cleanup - elapsed = time.time() - self.stream_start_time - local_clients = 0 - total_clients = 0 - proxy_server = ProxyServer.get_instance() + # Client cleanup + elapsed = time.time() - self.stream_start_time + local_clients = 0 + total_clients = 0 + proxy_server = ProxyServer.get_instance() - # Release M3U profile stream allocation if this is the last client - stream_released = False - if proxy_server.redis_client: - try: - if self.channel_id in proxy_server.client_managers: - client_count = proxy_server.client_managers[self.channel_id].get_total_client_count() - # Pool slots are global; the last client on any worker must release. - if client_count <= 1: + # Release M3U profile stream allocation if this is the last client + stream_released = False + if proxy_server.redis_client: + try: + if self.channel_id in proxy_server.client_managers: + client_count = proxy_server.client_managers[self.channel_id].get_total_client_count() + # Pool slots are global; the last client on any worker must release. + if client_count <= 1: + try: try: - try: - obj = Channel.objects.get(uuid=self.channel_id) - except (Channel.DoesNotExist, Exception): - obj = Stream.objects.get(stream_hash=self.channel_id) - stream_released = obj.release_stream() - if stream_released: - logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}") - else: - logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}") - except Exception as e: - logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}") - except Exception as e: - logger.error(f"[{self.client_id}] Error checking stream data for release: {e}") + obj = Channel.objects.get(uuid=self.channel_id) + except (Channel.DoesNotExist, Exception): + obj = Stream.objects.get(stream_hash=self.channel_id) + stream_released = obj.release_stream() + if stream_released: + logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}") + else: + logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}") + except Exception as e: + logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}") + except Exception as e: + logger.error(f"[{self.client_id}] Error checking stream data for release: {e}") - if self.channel_id in proxy_server.client_managers: - client_manager = proxy_server.client_managers[self.channel_id] - if self.client_id in client_manager.clients: - local_clients = client_manager.remove_client(self.client_id) - else: - local_clients = client_manager.get_client_count() - total_clients = client_manager.get_total_client_count() - logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})") + if self.channel_id in proxy_server.client_managers: + client_manager = proxy_server.client_managers[self.channel_id] + if self.client_id in client_manager.clients: + local_clients = client_manager.remove_client(self.client_id) + else: + local_clients = client_manager.get_client_count() + total_clients = client_manager.get_total_client_count() + logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})") - # Log client disconnect event - try: - log_system_event( - 'client_disconnect', - channel_id=self.channel_id, - channel_name=self.channel_name, - client_ip=self.client_ip, - client_id=self.client_id, - user_agent=self.client_user_agent[:100] if self.client_user_agent else None, - duration=round(elapsed, 2), - bytes_sent=self.bytes_sent, - username=self.user.username if self.user else None - ) - except Exception as e: - logger.error(f"Could not log client disconnect event: {e}") - finally: - close_old_connections() + # Log client disconnect event + try: + log_system_event( + 'client_disconnect', + channel_id=self.channel_id, + channel_name=self.channel_name, + client_ip=self.client_ip, + client_id=self.client_id, + user_agent=self.client_user_agent[:100] if self.client_user_agent else None, + duration=round(elapsed, 2), + bytes_sent=self.bytes_sent, + username=self.user.username if self.user else None + ) + except Exception as e: + logger.error(f"Could not log client disconnect event: {e}") def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None): diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py index 98346676..3cb1fdae 100644 --- a/apps/proxy/live_proxy/server.py +++ b/apps/proxy/live_proxy/server.py @@ -760,7 +760,6 @@ class ProxyServer: ) except Exception as e: logger.error(f"Could not log channel start event: {e}") - finally: close_old_connections() # Create client manager with channel_id, redis_client AND worker_id (only if not already exists) @@ -1367,11 +1366,7 @@ class ProxyServer: return def _log_stop(): - try: - close_old_connections() - log_system_event('channel_stop', **stop_event_data) - except Exception as e: - logger.error(f"Could not log channel stop event: {e}") + log_system_event('channel_stop', **stop_event_data) gevent.spawn(_log_stop) diff --git a/core/utils.py b/core/utils.py index 5931d985..188d9cb2 100644 --- a/core/utils.py +++ b/core/utils.py @@ -623,6 +623,8 @@ def validate_flexible_url(value): raise ValidationError("Enter a valid URL.") def dispatch_event_system(event_type, channel_id=None, channel_name=None, **details): + from django.db import close_old_connections + try: from apps.connect.utils import trigger_event from apps.channels.models import Channel, Stream @@ -696,9 +698,48 @@ def dispatch_event_system(event_type, channel_id=None, channel_name=None, **deta trigger_event(event_type, payload) - except Exception as e: + except Exception: # Don't fail main path if connect dispatch fails pass + finally: + close_old_connections() + + +def _dispatch_system_event_integrations( + event_type, channel_id=None, channel_name=None, **details +): + """ + Run Connect subscriptions and plugin event hooks without blocking the caller. + + On gevent uWSGI workers, dispatch runs in a spawned greenlet so slow webhooks, + scripts, or plugin handlers cannot stall live-proxy teardown or streaming paths. + Celery prefork workers (gevent patched but no hub) run synchronously instead. + """ + + def _run(): + try: + dispatch_event_system( + event_type, + channel_id=channel_id, + channel_name=channel_name, + **details, + ) + except Exception as e: + logger.error( + "Failed to dispatch Connect/plugin handlers for event %s: %s", + event_type, + e, + ) + + if _should_use_sync_websocket_send(): + _run() + elif _is_gevent_monkey_patched(): + import gevent + + gevent.spawn(_run) + else: + _run() + def log_system_event(event_type, channel_id=None, channel_name=None, **details): """ @@ -726,8 +767,13 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details): details=details ) - # Trigger connect integrations for specific events - dispatch_event_system(event_type, channel_id=channel_id, channel_name=channel_name, **details) + # Connect integrations and plugin event hooks (non-blocking on gevent uWSGI) + _dispatch_system_event_integrations( + event_type, + channel_id=channel_id, + channel_name=channel_name, + **details, + ) # Get max events from settings (default 100) try: diff --git a/tests/test_log_system_event.py b/tests/test_log_system_event.py new file mode 100644 index 00000000..6ff92ff4 --- /dev/null +++ b/tests/test_log_system_event.py @@ -0,0 +1,55 @@ +"""Tests for log_system_event Connect/plugin dispatch and DB cleanup.""" + +from unittest.mock import patch + +from django.test import SimpleTestCase + + +class LogSystemEventDispatchTests(SimpleTestCase): + @patch("django.db.close_old_connections") + @patch("core.utils._dispatch_system_event_integrations") + @patch("core.models.SystemEvent.objects") + @patch("core.models.CoreSettings.objects") + def test_log_system_event_dispatches_integrations_and_closes_db( + self, mock_core_settings, mock_system_event, mock_dispatch, mock_close + ): + mock_system_event.count.return_value = 1 + mock_core_settings.filter.return_value.first.return_value = None + + from core.utils import log_system_event + + log_system_event("channel_start", channel_id="abc", channel_name="Test") + + mock_dispatch.assert_called_once_with( + "channel_start", + channel_id="abc", + channel_name="Test", + ) + mock_close.assert_called_once() + + @patch("django.db.close_old_connections") + @patch("apps.connect.utils.trigger_event") + def test_integration_dispatch_closes_db_on_sync_path( + self, mock_trigger, mock_close + ): + from core.utils import _dispatch_system_event_integrations + + with patch("core.utils._should_use_sync_websocket_send", return_value=True): + _dispatch_system_event_integrations("client_connect", channel_id="abc") + + mock_trigger.assert_called_once() + mock_close.assert_called_once() + + @patch("core.utils.dispatch_event_system") + def test_integration_dispatch_spawns_on_gevent_uwsgi( + self, mock_dispatch + ): + from core.utils import _dispatch_system_event_integrations + + with patch("core.utils._should_use_sync_websocket_send", return_value=False), patch( + "core.utils._is_gevent_monkey_patched", return_value=True + ), patch("gevent.spawn") as mock_spawn: + _dispatch_system_event_integrations("channel_stop", channel_id="abc") + + mock_spawn.assert_called_once() + mock_dispatch.assert_not_called()