From 7989fa3673a58888befa9977d451626fac7d48a4 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 15 Jun 2026 17:32:38 -0500 Subject: [PATCH] fix(proxy): enhance database connection management during VOD playback and stats updates - Added calls to `close_old_connections()` in `stream_vod()` and `build_vod_stats_data()` to prevent connection leaks during long-lived streaming responses and background stats refreshes. - Improved connection handling in various components to ensure proper resource cleanup and prevent blocking issues. --- CHANGELOG.md | 1 + .../tests/test_ts_proxy_ghost_clients.py | 8 + .../channels/tests/test_ts_proxy_keepalive.py | 12 +- apps/channels/tests/test_ts_proxy_teardown.py | 12 +- apps/proxy/live_proxy/input/manager.py | 15 +- apps/proxy/live_proxy/server.py | 17 +- .../vod_proxy/tests/test_vod_db_cleanup.py | 146 ++++++++++++++++++ apps/proxy/vod_proxy/views.py | 6 + 8 files changed, 198 insertions(+), 19 deletions(-) create mode 100644 apps/proxy/vod_proxy/tests/test_vod_db_cleanup.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e8cacfda..acce9b4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **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. +- **VOD proxy could leak geventpool DB checkouts during playback and stats updates.** `stream_vod()` ran ORM lookups for content and M3U profiles, then returned a long-lived `StreamingHttpResponse` without releasing the checkout, so each movie/episode stream could hold a pool slot for its full duration. Background VOD stats refresh (`build_vod_stats_data()`, triggered on start/stop and by the admin stats API) also queried movie, episode, and profile rows from daemon threads with no cleanup. `stream_vod()` now calls `close_old_connections()` before handing off to the streaming generator, and `build_vod_stats_data()` releases its checkout in a `finally` block. - **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/channels/tests/test_ts_proxy_ghost_clients.py b/apps/channels/tests/test_ts_proxy_ghost_clients.py index 6742cf4b..3cc7aa9e 100644 --- a/apps/channels/tests/test_ts_proxy_ghost_clients.py +++ b/apps/channels/tests/test_ts_proxy_ghost_clients.py @@ -247,6 +247,14 @@ class BasicStatsGhostClientTests(TestCase): redis.scard.return_value = len(client_ids) redis.smembers.return_value = client_ids redis.hget.return_value = None # individual field lookups + redis.hmget.return_value = [ + b'VLC/3.0', + b'127.0.0.1', + b'1773500000.0', + None, + b'mpegts', + None, + ] # Pipeline for remove_ghost_clients pipe = MagicMock() diff --git a/apps/channels/tests/test_ts_proxy_keepalive.py b/apps/channels/tests/test_ts_proxy_keepalive.py index b8fb524a..8645a158 100644 --- a/apps/channels/tests/test_ts_proxy_keepalive.py +++ b/apps/channels/tests/test_ts_proxy_keepalive.py @@ -218,7 +218,7 @@ class DoStatsUpdateTests(TestCase): mock_redis.scan.return_value = (0, []) with patch("apps.proxy.live_proxy.client_manager.send_websocket_update") as mock_ws, \ - patch("redis.Redis.from_url", return_value=mock_redis): + patch("core.utils.RedisClient.get_client", return_value=mock_redis): cm._do_stats_update() mock_ws.assert_called_once() @@ -231,25 +231,25 @@ class DoStatsUpdateTests(TestCase): """Redis failure must be swallowed (logged), not propagated.""" cm = self._make_client_manager() - with patch("redis.Redis.from_url", side_effect=Exception("Redis down")): + with patch("core.utils.RedisClient.get_client", side_effect=Exception("Redis down")): try: cm._do_stats_update() except Exception as e: self.fail(f"_do_stats_update raised an exception: {e}") - def test_do_stats_update_scans_channel_client_keys(self): - """Must scan for live:channel:*:clients pattern.""" + def test_do_stats_update_scans_channel_metadata_keys(self): + """Must scan for live:channel:*:metadata pattern.""" cm = self._make_client_manager() mock_redis = MagicMock() mock_redis.scan.return_value = (0, []) with patch("apps.proxy.live_proxy.client_manager.send_websocket_update"), \ - patch("redis.Redis.from_url", return_value=mock_redis): + patch("core.utils.RedisClient.get_client", return_value=mock_redis): cm._do_stats_update() scan_call = mock_redis.scan.call_args - self.assertIn("live:channel:*:clients", str(scan_call)) + self.assertIn("live:channel:*:metadata", str(scan_call)) # --------------------------------------------------------------------------- diff --git a/apps/channels/tests/test_ts_proxy_teardown.py b/apps/channels/tests/test_ts_proxy_teardown.py index 3bc92f18..593df034 100644 --- a/apps/channels/tests/test_ts_proxy_teardown.py +++ b/apps/channels/tests/test_ts_proxy_teardown.py @@ -390,17 +390,21 @@ class CleanRedisKeysOrderTests(TestCase): mock_channel_get.side_effect = channel_get mock_stream_get.side_effect = Stream.DoesNotExist - def scan(*args, **kwargs): + channel_key = f"live:channel:{CHANNEL_ID}:input:buffer:index".encode() + + def scan(cursor, match=None, count=100): call_order.append("redis") - return (0, [b"live:channel:foo:buffer:index"]) + if match == f"live:channel:{CHANNEL_ID}:*": + return (0, [channel_key]) + return (0, []) server.redis_client.scan.side_effect = scan server._clean_redis_keys(CHANNEL_ID) - self.assertEqual(call_order, ["release", "redis"]) + self.assertEqual(call_order, ["release", "redis", "redis"]) channel.release_stream.assert_called_once() - server.redis_client.delete.assert_called_once() + server.redis_client.delete.assert_called_once_with(channel_key) class LocalUpstreamActivityTests(TestCase): diff --git a/apps/proxy/live_proxy/input/manager.py b/apps/proxy/live_proxy/input/manager.py index 7d01a88f..3c7c147d 100644 --- a/apps/proxy/live_proxy/input/manager.py +++ b/apps/proxy/live_proxy/input/manager.py @@ -67,6 +67,7 @@ class StreamManager: # Add to your __init__ method self._buffer_check_timers = [] self.stopping = False + self.stop_requested = False # Add tracking for tried streams and current stream self.current_stream_id = stream_id @@ -572,7 +573,9 @@ class StreamManager: try: metadata_key = RedisKeys.channel_metadata(self.channel_id) owner_key = RedisKeys.channel_owner(self.channel_id) - current_owner = self.buffer.redis_client.get(owner_key) + current_owner = self._decode_redis_value( + self.buffer.redis_client.get(owner_key) + ) is_owner = ( current_owner @@ -583,12 +586,10 @@ class StreamManager: should_update = is_owner if not should_update and no_owner: - current_state_bytes = self.buffer.redis_client.hget( - metadata_key, ChannelMetadataField.STATE - ) - current_state = ( - current_state_bytes - if current_state_bytes else None + current_state = self._decode_redis_value( + self.buffer.redis_client.hget( + metadata_key, ChannelMetadataField.STATE + ) ) should_update = current_state in ChannelState.PRE_ACTIVE if not should_update and current_state: diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py index 3cb1fdae..ddec2d02 100644 --- a/apps/proxy/live_proxy/server.py +++ b/apps/proxy/live_proxy/server.py @@ -1402,6 +1402,15 @@ class ProxyServer: f"Error broadcasting upstream stop for channel {channel_id}: {e}" ) + @staticmethod + def _channel_id_from_metadata_key(key): + if isinstance(key, bytes): + key = key.decode('utf-8', errors='replace') + parts = key.split(':') + if len(parts) >= 3: + return parts[2] + return None + def _stop_upstream_before_redis_cleanup(self, channel_id): """Stop local ffmpeg before deleting Redis keys (prevents delete/recreate loops).""" if self._has_local_upstream_activity(channel_id): @@ -1955,7 +1964,9 @@ class ProxyServer: for key in channel_keys: try: - channel_id = key.split(':')[2] + channel_id = self._channel_id_from_metadata_key(key) + if not channel_id: + continue # Check if this channel has an owner owner = self.get_channel_owner(channel_id) @@ -1995,7 +2006,9 @@ class ProxyServer: for key in channel_keys: try: - channel_id = key.split(':')[2] + channel_id = self._channel_id_from_metadata_key(key) + if not channel_id: + continue # Get metadata first metadata = self.redis_client.hgetall(key) diff --git a/apps/proxy/vod_proxy/tests/test_vod_db_cleanup.py b/apps/proxy/vod_proxy/tests/test_vod_db_cleanup.py new file mode 100644 index 00000000..5d009803 --- /dev/null +++ b/apps/proxy/vod_proxy/tests/test_vod_db_cleanup.py @@ -0,0 +1,146 @@ +"""VOD proxy must release geventpool checkouts after ORM on stream and stats paths.""" + +from unittest.mock import MagicMock, patch + +from django.http import StreamingHttpResponse +from django.test import RequestFactory, SimpleTestCase + + +class StreamVodDbCleanupTests(SimpleTestCase): + def setUp(self): + self.factory = RequestFactory() + + @patch("apps.proxy.vod_proxy.views.close_old_connections") + @patch("apps.proxy.vod_proxy.views.MultiWorkerVODConnectionManager") + @patch("apps.proxy.vod_proxy.views._transform_url", return_value="http://example.com/movie.mp4") + @patch("apps.proxy.vod_proxy.views._get_m3u_profile") + @patch("apps.proxy.vod_proxy.views._get_stream_url_from_relation", return_value="http://upstream/movie.mp4") + @patch("apps.proxy.vod_proxy.views._get_content_and_relation") + @patch("apps.proxy.vod_proxy.views.network_access_allowed", return_value=True) + def test_stream_vod_closes_db_before_streaming_response( + self, + _network_ok, + mock_content, + _stream_url, + mock_profile, + _transform, + mock_manager_cls, + mock_close, + ): + movie = MagicMock() + movie.name = "Test Movie" + relation = MagicMock() + relation.m3u_account.name = "Provider" + mock_content.return_value = (movie, relation) + + profile = MagicMock() + profile.id = 1 + profile.max_streams = 5 + mock_profile.return_value = (profile, 0) + + mock_manager = MagicMock() + mock_manager.stream_content_with_session.return_value = StreamingHttpResponse( + streaming_content=iter([b"data"]), + content_type="video/mp4", + ) + mock_manager_cls.get_instance.return_value = mock_manager + + request = self.factory.get( + "/proxy/vod/movie/uuid/session123/", + HTTP_USER_AGENT="test-agent", + ) + request.user = MagicMock(is_authenticated=False) + + from apps.proxy.vod_proxy.views import stream_vod + + response = stream_vod( + request, + content_type="movie", + content_id="uuid", + session_id="session123", + ) + + self.assertIsInstance(response, StreamingHttpResponse) + mock_close.assert_called_once() + mock_manager.stream_content_with_session.assert_called_once() + + +class BuildVodStatsDbCleanupTests(SimpleTestCase): + @patch("apps.proxy.vod_proxy.views.close_old_connections") + @patch("apps.proxy.vod_proxy.views.Movie") + def test_build_vod_stats_data_closes_db(self, mock_movie, mock_close): + redis_client = MagicMock() + redis_client.scan.side_effect = [ + (0, ["vod_persistent_connection:s1"]), + ] + redis_client.hgetall.return_value = { + "content_obj_type": "movie", + "content_uuid": "movie-uuid", + "content_name": "Test Movie", + "m3u_profile_id": "1", + "client_ip": "127.0.0.1", + "client_user_agent": "agent", + "connected_at": "1000.0", + "last_activity": "1001.0", + "active_streams": "1", + } + + movie_obj = MagicMock( + name="Test Movie", + logo=None, + year=2020, + rating=7.5, + genre="Action", + description="Desc", + tmdb_id="1", + imdb_id="tt1", + ) + mock_movie.objects.select_related.return_value.get.return_value = movie_obj + + with patch("apps.m3u.models.M3UAccountProfile") as mock_profile_model: + mock_profile_model.objects.select_related.return_value.get.return_value = MagicMock( + name="Profile 1", + m3u_account=MagicMock(name="Account", id=1), + ) + + from apps.proxy.vod_proxy.views import build_vod_stats_data + + stats = build_vod_stats_data(redis_client) + + self.assertEqual(stats["total_connections"], 1) + mock_close.assert_called_once() + + @patch("apps.proxy.vod_proxy.views.close_old_connections") + def test_build_vod_stats_data_closes_db_on_error(self, mock_close): + redis_client = MagicMock() + redis_client.scan.side_effect = RuntimeError("redis down") + + from apps.proxy.vod_proxy.views import build_vod_stats_data + + stats = build_vod_stats_data(redis_client) + + self.assertEqual(stats["total_connections"], 0) + mock_close.assert_called_once() + + +class VodStatsUpdateDbCleanupTests(SimpleTestCase): + @patch("core.utils.send_websocket_update") + @patch("apps.proxy.vod_proxy.views.build_vod_stats_data") + def test_do_vod_stats_update_uses_build_vod_stats_data(self, mock_build, mock_ws): + mock_build.return_value = { + "vod_connections": [], + "total_connections": 0, + "timestamp": 0, + } + + from apps.proxy.vod_proxy.multi_worker_connection_manager import ( + MultiWorkerVODConnectionManager, + ) + + manager = MultiWorkerVODConnectionManager.__new__(MultiWorkerVODConnectionManager) + manager.redis_client = MagicMock() + + manager._do_vod_stats_update() + + mock_build.assert_called_once_with(manager.redis_client) + mock_ws.assert_called_once() diff --git a/apps/proxy/vod_proxy/views.py b/apps/proxy/vod_proxy/views.py index f9740aa1..fef8fed0 100644 --- a/apps/proxy/vod_proxy/views.py +++ b/apps/proxy/vod_proxy/views.py @@ -8,6 +8,7 @@ import random import logging import requests from urllib.parse import urlencode +from django.db import close_old_connections from django.http import JsonResponse, Http404, HttpResponse from django.shortcuts import get_object_or_404 from django.views.decorators.csrf import csrf_exempt @@ -601,6 +602,9 @@ def stream_vod(request, content_type, content_id, session_id=None, profile_id=No # Get connection manager (Redis-backed for multi-worker support) connection_manager = MultiWorkerVODConnectionManager.get_instance() + # Release ORM checkout before returning a long-lived StreamingHttpResponse. + close_old_connections() + # Stream the content with session-based connection reuse logger.info("[VOD-STREAM] Calling connection manager to stream content") response = connection_manager.stream_content_with_session( @@ -1063,6 +1067,8 @@ def build_vod_stats_data(redis_client): except Exception as e: logger.error(f"Error building VOD stats: {e}") return {'vod_connections': [], 'total_connections': 0, 'timestamp': time.time()} + finally: + close_old_connections() @api_view(["GET"])