diff --git a/CHANGELOG.md b/CHANGELOG.md index 4de12ded..7e40e04a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,7 @@ 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. - **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. - **EPG auto-match reliability fixes.** diff --git a/apps/proxy/live_proxy/input/manager.py b/apps/proxy/live_proxy/input/manager.py index ae744473..e8315006 100644 --- a/apps/proxy/live_proxy/input/manager.py +++ b/apps/proxy/live_proxy/input/manager.py @@ -690,6 +690,9 @@ class StreamManager: 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() + logger.debug(f"Starting transcode process: {self.transcode_cmd} for channel: {self.channel_id}") import os as _os diff --git a/apps/proxy/live_proxy/output/fmp4/generator.py b/apps/proxy/live_proxy/output/fmp4/generator.py index 65a8c5c7..e979008c 100644 --- a/apps/proxy/live_proxy/output/fmp4/generator.py +++ b/apps/proxy/live_proxy/output/fmp4/generator.py @@ -11,6 +11,7 @@ import time import gevent 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 ...redis_keys import RedisKeys from ...constants import ChannelMetadataField @@ -331,46 +332,49 @@ class FMP4StreamGenerator: # ------------------------------------------------------------------ def _cleanup(self): - elapsed = time.time() - self.stream_start_time - proxy_server = ProxyServer.get_instance() + try: + elapsed = time.time() - self.stream_start_time + proxy_server = ProxyServer.get_instance() - # Release stream allocation if last client (mirrors StreamGenerator) - if proxy_server.redis_client: - try: - meta_key = RedisKeys.channel_metadata(self.channel_id) - stream_id_bytes = proxy_server.redis_client.hget( - meta_key, ChannelMetadataField.STREAM_ID - ) - if stream_id_bytes: - if self.channel_id in proxy_server.client_managers: - client_count = proxy_server.client_managers[ - self.channel_id - ].get_total_client_count() - if client_count <= 1 and proxy_server.am_i_owner(self.channel_id): - try: + # Release stream allocation if last client (mirrors StreamGenerator) + if proxy_server.redis_client: + try: + meta_key = RedisKeys.channel_metadata(self.channel_id) + stream_id_bytes = proxy_server.redis_client.hget( + meta_key, ChannelMetadataField.STREAM_ID + ) + if stream_id_bytes: + if self.channel_id in proxy_server.client_managers: + client_count = proxy_server.client_managers[ + self.channel_id + ].get_total_client_count() + if client_count <= 1 and proxy_server.am_i_owner(self.channel_id): try: - obj = Channel.objects.get(uuid=self.channel_id) - except (Channel.DoesNotExist, Exception): - obj = Stream.objects.get(stream_hash=self.channel_id) - obj.release_stream() - except Exception as e: - logger.error( - f"[{self.client_id}] Error releasing stream: {e}" - ) - except Exception as e: - logger.error(f"[{self.client_id}] Error in stream release check: {e}") + try: + obj = Channel.objects.get(uuid=self.channel_id) + except (Channel.DoesNotExist, Exception): + obj = Stream.objects.get(stream_hash=self.channel_id) + obj.release_stream() + except Exception as e: + logger.error( + f"[{self.client_id}] Error releasing stream: {e}" + ) + except Exception as e: + logger.error(f"[{self.client_id}] Error in stream release check: {e}") - # Remove from MAIN ClientManager - this is what triggers handle_client_disconnect - # and the zero-clients → stop_channel path, same as TS clients. - local_clients = 0 - total_clients = 0 - if self.channel_id in proxy_server.client_managers: - client_manager = proxy_server.client_managers[self.channel_id] - local_clients = client_manager.remove_client(self.client_id) - total_clients = client_manager.get_total_client_count() + # Remove from MAIN ClientManager - this is what triggers handle_client_disconnect + # and the zero-clients → stop_channel path, same as TS clients. + local_clients = 0 + total_clients = 0 + if self.channel_id in proxy_server.client_managers: + client_manager = proxy_server.client_managers[self.channel_id] + local_clients = client_manager.remove_client(self.client_id) + total_clients = client_manager.get_total_client_count() - logger.info( - f"[{self.client_id}] fMP4 client disconnected after {elapsed:.2f}s " - f"({self.bytes_sent / 1024:.1f} KB sent, " - f"local: {local_clients}, total: {total_clients})" - ) + logger.info( + f"[{self.client_id}] fMP4 client disconnected after {elapsed:.2f}s " + f"({self.bytes_sent / 1024:.1f} KB sent, " + f"local: {local_clients}, total: {total_clients})" + ) + finally: + close_old_connections() diff --git a/apps/proxy/live_proxy/output/ts/generator.py b/apps/proxy/live_proxy/output/ts/generator.py index efde02ad..69dc341f 100644 --- a/apps/proxy/live_proxy/output/ts/generator.py +++ b/apps/proxy/live_proxy/output/ts/generator.py @@ -8,6 +8,7 @@ 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 @@ -590,59 +591,62 @@ class StreamGenerator: def _cleanup(self): """Clean up resources and report final statistics.""" - # Client cleanup - elapsed = time.time() - self.stream_start_time - local_clients = 0 - total_clients = 0 - proxy_server = ProxyServer.get_instance() + try: + # 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: - try: + # 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: - 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}") + 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}") - 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}") + # 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() 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 667daf2f..98346676 100644 --- a/apps/proxy/live_proxy/server.py +++ b/apps/proxy/live_proxy/server.py @@ -760,6 +760,8 @@ 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) if channel_id not in self.client_managers: @@ -2072,42 +2074,45 @@ class ProxyServer: """Clean up all Redis keys for a channel more efficiently""" total_deleted = 0 - # Release the M3U profile slot while channel_stream / metadata still exist. - # Scanning live:channel keys first deletes metadata and breaks release_stream() - # fallback, leaving profile_connections counters stuck (e.g. profile_connections:70 = 1). try: - channel = Channel.objects.get(uuid=channel_id) - if not channel.release_stream(): - logger.debug(f"Channel {channel_id}: release_stream found no keys to clean") - except (Channel.DoesNotExist, Exception): + # Release the M3U profile slot while channel_stream / metadata still exist. + # Scanning live:channel keys first deletes metadata and breaks release_stream() + # fallback, leaving profile_connections counters stuck (e.g. profile_connections:70 = 1). try: - stream = Stream.objects.get(stream_hash=channel_id) - if not stream.release_stream(): - logger.debug(f"Stream {channel_id}: release_stream found no keys to clean") - except (Stream.DoesNotExist, Exception): - logger.debug(f"No Channel or Stream found for {channel_id}") + channel = Channel.objects.get(uuid=channel_id) + if not channel.release_stream(): + logger.debug(f"Channel {channel_id}: release_stream found no keys to clean") + except (Channel.DoesNotExist, Exception): + try: + stream = Stream.objects.get(stream_hash=channel_id) + if not stream.release_stream(): + logger.debug(f"Stream {channel_id}: release_stream found no keys to clean") + except (Stream.DoesNotExist, Exception): + logger.debug(f"No Channel or Stream found for {channel_id}") - if self.redis_client: - try: - patterns = [ - f"live:channel:{channel_id}:*", - RedisKeys.events_channel(channel_id), - ] + if self.redis_client: + try: + patterns = [ + f"live:channel:{channel_id}:*", + RedisKeys.events_channel(channel_id), + ] - for pattern in patterns: - cursor = 0 - while True: - cursor, keys = self.redis_client.scan(cursor, match=pattern, count=100) - if keys: - self.redis_client.delete(*keys) - total_deleted += len(keys) + for pattern in patterns: + cursor = 0 + while True: + cursor, keys = self.redis_client.scan(cursor, match=pattern, count=100) + if keys: + self.redis_client.delete(*keys) + total_deleted += len(keys) - if cursor == 0: - break + if cursor == 0: + break - logger.info(f"Cleaned up {total_deleted} Redis keys for channel {channel_id}") - except Exception as e: - logger.error(f"Error cleaning Redis keys for channel {channel_id}: {e}") + logger.info(f"Cleaned up {total_deleted} Redis keys for channel {channel_id}") + except Exception as e: + logger.error(f"Error cleaning Redis keys for channel {channel_id}: {e}") + finally: + close_old_connections() return total_deleted diff --git a/core/utils.py b/core/utils.py index 08dda3dc..5931d985 100644 --- a/core/utils.py +++ b/core/utils.py @@ -715,6 +715,7 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details): stream_url='http://...', user='admin') """ from core.models import SystemEvent, CoreSettings + from django.db import close_old_connections try: # Create the event @@ -750,6 +751,10 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details): except Exception as e: # Don't let event logging break the main application logger.error(f"Failed to log system event {event_type}: {e}") + finally: + # geventpool keeps checked-out connections until close(); release promptly + # when logging from proxy greenlets/threads outside a normal request cycle. + close_old_connections() def _send_async(channel_layer, group, message):