Merge pull request #1063 from CodeBormen:fix/1012-Celery-Work-Memory-Growth

fix(m3u): fix Celery worker memory leak during M3U/XC refresh (#1012, #1053)

CHANGELOG.md

@@ -29,6 +29,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Celery worker memory leak during M3U/XC refresh causing 20–80 MB growth per cycle with no reclamation (Fixes #1012, #1053) - Thanks [@CodeBormen](https://github.com/CodeBormen)
+  - Restructured `refresh_single_m3u_account()` with a `try/finally` that guarantees `del` of large data structures runs before Celery's `gc.collect()`, and lock release on all exit paths (success, exception, early return)
+  - Re-enabled batch data cleanup in `process_m3u_batch_direct()` (was commented out)
+  - Added `CELERY_WORKER_MAX_MEMORY_PER_CHILD = 512 MB` as a safety net against pymalloc arena fragmentation
 - EPG output was filtering programs using `start_time__gte=now` when the `days` parameter was specified, which caused currently-airing programs (started before the request time but not yet ended) to be omitted from the XML output. This produced a gap in clients' guides immediately after an EPG refresh, lasting until the next program started. Fixed by changing the filter to `end_time__gte=now` so any program that has not yet finished is included.
 - TS proxy connection slot leaks and TOCTOU races in stream initialization (Fixes #947) - Thanks [@CodeBormen](https://github.com/CodeBormen)
   - **TOCTOU race in slot reservation**: `get_stream()` previously used a `GET`→check→`INCR` sequence, allowing concurrent requests to both read the same count below the limit and both reserve a slot, silently exceeding `max_streams`. Replaced with an atomic `INCR`-first pattern: increment unconditionally, check the result, roll back with `DECR` if over capacity.

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."
 

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()

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