From ce9f7ccbf93fbc745ef8ac95b38d9f6e846ee405 Mon Sep 17 00:00:00 2001 From: None Date: Thu, 5 Mar 2026 13:47:20 -0600 Subject: [PATCH] fix(m3u): fix Celery worker memory leak during M3U/XC refresh Restructure cleanup in refresh_single_m3u_account so del statements run before gc.collect(), enabling cyclic garbage collection. Add worker_max_memory_per_child as a safety net against memory fragmentation. Closes #1012, Closes #1053 --- apps/m3u/tasks.py | 84 +++++++++++--------- apps/m3u/tests/__init__.py | 0 apps/m3u/tests/test_memory_cleanup.py | 106 ++++++++++++++++++++++++++ dispatcharr/settings.py | 4 + 4 files changed, 156 insertions(+), 38 deletions(-) create mode 100644 apps/m3u/tests/__init__.py create mode 100644 apps/m3u/tests/test_memory_cleanup.py diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index a9413143..c0c90c57 100644 --- a/apps/m3u/tasks.py +++ b/apps/m3u/tasks.py @@ -1170,14 +1170,12 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys): retval = f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated." - # Aggressive garbage collection - # del streams_to_create, streams_to_update, stream_hashes, existing_streams - # from core.utils import cleanup_memory - # cleanup_memory(log_usage=True, force_collection=True) - # Clean up database connections for threading connections.close_all() + # Free batch data structures (reference-counted deallocation) + del streams_to_create, streams_to_update, stream_hashes, existing_streams + return retval @@ -2642,6 +2640,16 @@ def refresh_single_m3u_account(account_id): lock_renewer = TaskLockRenewer("refresh_single_m3u_account", account_id) lock_renewer.start() + try: + return _refresh_single_m3u_account_impl(account_id) + finally: + # Guaranteed cleanup on all exit paths (success, exception, early return) + lock_renewer.stop() + release_task_lock("refresh_single_m3u_account", account_id) + + +def _refresh_single_m3u_account_impl(account_id): + """Implementation of M3U account refresh with guaranteed memory cleanup.""" # Record start time refresh_start_timestamp = timezone.now() # For the cleanup function start_time = time.time() # For tracking elapsed time as float @@ -2653,8 +2661,6 @@ def refresh_single_m3u_account(account_id): account = M3UAccount.objects.get(id=account_id, is_active=True) if not account.is_active: logger.debug(f"Account {account_id} is not active, skipping.") - lock_renewer.stop() - release_task_lock("refresh_single_m3u_account", account_id) return # Set status to fetching @@ -2683,8 +2689,6 @@ def refresh_single_m3u_account(account_id): else: logger.debug(f"No orphaned task found for M3U account {account_id}") - lock_renewer.stop() - release_task_lock("refresh_single_m3u_account", account_id) return f"M3UAccount with ID={account_id} not found or inactive, task cleaned up" # Fetch M3U lines and handle potential issues @@ -2699,6 +2703,7 @@ def refresh_single_m3u_account(account_id): extinf_data = data["extinf_data"] groups = data["groups"] + del data # Free top-level dict; extinf_data/groups retain their references except json.JSONDecodeError as e: # Handle corrupted JSON file logger.error( @@ -2734,8 +2739,6 @@ def refresh_single_m3u_account(account_id): logger.error( f"Failed to refresh M3U groups for account {account_id}: {result}" ) - lock_renewer.stop() - release_task_lock("refresh_single_m3u_account", account_id) return "Failed to update m3u account - download failed or other error" extinf_data, groups = result @@ -2768,8 +2771,6 @@ def refresh_single_m3u_account(account_id): status="error", error=f"Error refreshing M3U groups: {str(e)}", ) - lock_renewer.stop() - release_task_lock("refresh_single_m3u_account", account_id) return "Failed to update m3u account" # Only proceed with parsing if we actually have data and no errors were encountered @@ -2792,8 +2793,6 @@ def refresh_single_m3u_account(account_id): status="error", error="No data available for processing", ) - lock_renewer.stop() - release_task_lock("refresh_single_m3u_account", account_id) return "Failed to update m3u account, no data available" hash_keys = CoreSettings.get_m3u_hash_key().split(",") @@ -2939,6 +2938,9 @@ def refresh_single_m3u_account(account_id): logger.info(f"Processing {len(all_xc_streams)} XC streams in {len(batches)} batches") + # Free the original list; batches hold independent sliced copies + del all_xc_streams + # Use threading for XC stream processing - now with consistent batch sizes max_workers = min(4, len(batches)) logger.debug(f"Using {max_workers} threads for XC stream processing") @@ -3099,32 +3101,38 @@ def refresh_single_m3u_account(account_id): except Exception as e: logger.error(f"Error processing M3U for account {account_id}: {str(e)}") - account.status = M3UAccount.Status.ERROR - account.last_message = f"Error processing M3U: {str(e)}" - account.save(update_fields=["status", "last_message"]) + try: + account.status = M3UAccount.Status.ERROR + account.last_message = f"Error processing M3U: {str(e)}" + account.save(update_fields=["status", "last_message"]) + except Exception: + logger.debug(f"Failed to update account {account_id} status during error handling") raise # Re-raise the exception for Celery to handle finally: - lock_renewer.stop() - release_task_lock("refresh_single_m3u_account", account_id) + # Free large data structures regardless of success or failure + if 'existing_groups' in locals(): + del existing_groups + if 'extinf_data' in locals(): + del extinf_data + if 'groups' in locals(): + del groups + if 'batches' in locals(): + del batches + if 'all_xc_streams' in locals(): + del all_xc_streams + if 'data' in locals(): + del data + if 'filtered_groups' in locals(): + del filtered_groups + if 'channel_group_relationships' in locals(): + del channel_group_relationships - # Aggressive garbage collection - # Only delete variables if they exist - if 'existing_groups' in locals(): - del existing_groups - if 'extinf_data' in locals(): - del extinf_data - if 'groups' in locals(): - del groups - if 'batches' in locals(): - del batches - - from core.utils import cleanup_memory - - cleanup_memory(log_usage=True, force_collection=True) - - # Clean up cache file since we've fully processed it - if os.path.exists(cache_path): - os.remove(cache_path) + # Remove cache file after processing (success or failure) + cache_path = os.path.join(m3u_dir, f"{account_id}.json") + try: + os.remove(cache_path) + except OSError: + pass return f"Dispatched jobs complete." diff --git a/apps/m3u/tests/__init__.py b/apps/m3u/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/apps/m3u/tests/test_memory_cleanup.py b/apps/m3u/tests/test_memory_cleanup.py new file mode 100644 index 00000000..bd556e4c --- /dev/null +++ b/apps/m3u/tests/test_memory_cleanup.py @@ -0,0 +1,106 @@ +""" +Tests for memory cleanup behavior in M3U refresh tasks. + +Verifies that database connections are properly closed, task locks are +released on all exit paths, and garbage collection runs where expected. + +""" +from unittest.mock import patch, MagicMock + +from django.test import SimpleTestCase + +from apps.m3u.models import M3UAccount + + +class ProcessM3UBatchCleanupTests(SimpleTestCase): + """Verify process_m3u_batch_direct cleans up after processing.""" + + @patch("apps.m3u.tasks.Stream") + @patch("apps.m3u.tasks.M3UAccount") + def test_connections_closed_after_batch(self, mock_account_cls, mock_stream_cls): + """Database connections must be closed after batch processing (thread safety).""" + from apps.m3u.tasks import process_m3u_batch_direct + + mock_account = MagicMock() + mock_account.filters.order_by.return_value = [] + mock_account_cls.objects.get.return_value = mock_account + mock_stream_cls.objects.filter.return_value.select_related.return_value.only.return_value = ( + [] + ) + mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123") + + with patch("django.db.connections") as mock_connections: + process_m3u_batch_direct(1, [], {}, ["name", "url"]) + mock_connections.close_all.assert_called() + + +class LockReleaseTests(SimpleTestCase): + """Verify task lock is released on all exit paths.""" + + @patch("apps.m3u.tasks.delete_m3u_refresh_task_by_id", return_value=False) + def test_lock_released_on_account_not_found(self, mock_delete): + """release_task_lock must be called when account does not exist.""" + with patch( + "apps.m3u.tasks.acquire_task_lock", return_value=True + ), patch("apps.m3u.tasks.release_task_lock") as mock_release, patch( + "apps.m3u.tasks.TaskLockRenewer" + ): + with patch( + "apps.m3u.tasks.M3UAccount.objects.get", + side_effect=M3UAccount.DoesNotExist, + ): + from apps.m3u.tasks import refresh_single_m3u_account + + refresh_single_m3u_account(99999) + + mock_release.assert_called_once_with( + "refresh_single_m3u_account", 99999 + ) + + def test_lock_released_on_exception(self): + """release_task_lock must be called when an exception is raised.""" + mock_account = MagicMock() + mock_account.is_active = True + mock_account.account_type = "STD" + mock_account.custom_properties = {} + mock_account.filters.all.return_value = [] + mock_account.status = MagicMock() + + with patch( + "apps.m3u.tasks.acquire_task_lock", return_value=True + ), patch("apps.m3u.tasks.release_task_lock") as mock_release, patch( + "apps.m3u.tasks.TaskLockRenewer" + ): + with patch( + "apps.m3u.tasks.M3UAccount.objects.get", return_value=mock_account + ): + with patch("os.path.exists", return_value=False): + with patch( + "apps.m3u.tasks.refresh_m3u_groups", + side_effect=RuntimeError("test"), + ): + from apps.m3u.tasks import refresh_single_m3u_account + + try: + refresh_single_m3u_account(1) + except RuntimeError: + pass + + mock_release.assert_called_once_with("refresh_single_m3u_account", 1) + + +class XCCategoryCleanupTests(SimpleTestCase): + """Regression guard: process_xc_category_direct must continue to clean up.""" + + @patch("apps.m3u.tasks.XCClient") + @patch("apps.m3u.tasks.M3UAccount") + def test_xc_category_calls_gc_collect(self, mock_account_cls, mock_xc_client): + """gc.collect() must be called after XC category processing.""" + from apps.m3u.tasks import process_xc_category_direct + + mock_account = MagicMock() + mock_account_cls.objects.get.return_value = mock_account + + with patch("gc.collect") as mock_gc, patch("django.db.connections"): + process_xc_category_direct(1, {}, {}, ["name", "url"]) + mock_gc.assert_called() diff --git a/dispatcharr/settings.py b/dispatcharr/settings.py index c8350895..50bac5d5 100644 --- a/dispatcharr/settings.py +++ b/dispatcharr/settings.py @@ -240,6 +240,10 @@ CELERY_BROKER_TRANSPORT_OPTIONS = { CELERY_ACCEPT_CONTENT = ["json"] CELERY_TASK_SERIALIZER = "json" +# Worker memory safety net: recycle prefork workers exceeding 512MB RSS. +# Prevents unbounded growth from memory fragmentation or unexpected leaks. +CELERY_WORKER_MAX_MEMORY_PER_CHILD = 524_288 # 512 MB in KB + CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers.DatabaseScheduler" CELERY_BEAT_SCHEDULE = { # Explicitly disable the old fetch-channel-statuses task