refactor(tasks): Improve stream processing logic by adding bulk update functionality for unchanged streams. Enhance memory management and error handling in database queries, and update cleanup functions to optimize memory usage.

CHANGELOG.md

@@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - **XC catch-up (timeshift) support**. Adds a native `/timeshift/{user}/{pass}/{stream_id}/{timestamp}/{duration}.ts` endpoint that proxies replay sessions from an Xtream Codes provider to clients such as iPlayTV (Apple TV) and TiviMate. Auth reuses the same `xc_password` custom-property the live `/live/.../{id}.ts` endpoint already uses. Catch-up sessions appear on `/stats` with a violet `TIMESHIFT` badge alongside live sessions and respect per-channel access rules. — Thanks [@cedric-marcoux](https://github.com/cedric-marcoux) (#1242)
-  - **Catch-up flags** (`tv_archive`, `tv_archive_duration`) denormalized at import time for zero-cost output queries; the per-account rollup also self-heals channels left flagged with no remaining catch-up stream (e.g. after stale-stream cleanup on manual/multi-provider channels).
+  - **Catch-up flags** (`tv_archive`, `tv_archive_duration`) denormalized at import time for zero-cost output queries; end-of-refresh SQL rollup updates channels linked to the refreshed account and self-heals stale flags on those channels only (e.g. after catch-up streams are removed or an account is deactivated).
   - **Strictly-UTC API surface, automatic per-provider timezone.** `server_info.timezone` is always `UTC` and the XC EPG `start`/`end` strings are emitted in UTC; the proxy converts the requested instant to the serving provider's own reported timezone (`server_info.timezone` captured on account refresh) at request time. No timezone to configure.
   - **Multi-provider failover.** Catch-up walks the channel's catch-up streams in order (like live playback): if one provider cannot serve the archive the next is tried, each attempt with its own account context (credentials, reported timezone, user-agent). Accounts that fail with an auth/ban-class status are not retried via their other streams.
   - **Provider pool accounting.** Catch-up reserves a provider profile slot (`connection_pool`) before connecting upstream — same contract as live and VOD — and releases it when the session ends. When the default profile is at capacity it walks the account's alternate profiles with their own resolved credentials, and answers `503` when every profile is full.
@@ -19,6 +19,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - **EPG XMLTV `prev_days` auto-detection** from the provider's largest `tv_archive_duration` (capped at 30 days), overridable via setting or `?prev_days=` URL parameter.
   - **Byte path streams directly via `iter_content` + generator yield** (same pattern as the VOD proxy), with an upfront MPEG-TS sync peek that strips any PHP-warning preamble before bytes reach strict demuxers. Throughput stays comfortably above the typical 5 Mbps FHD bitrate so clients don't buffer.
 
+### Performance
+
+- **M3U/XC stream refresh is faster on large accounts.** Steady-state refreshes split `bulk_update` into a lightweight touch pass (`last_seen` / `is_stale` only) for unchanged streams and a full column update only when provider metadata or catch-up fields actually change.
+- **Celery workers return RSS after memory-intensive tasks.** `cleanup_memory()` accepts an optional `trim_heap` flag (glibc `malloc_trim`); Celery `task_postrun` enables it after `close_old_connections()` for M3U account/group refresh, EPG, VOD, and channel-matching tasks so worker memory drops back toward baseline instead of ratcheting across successive large jobs.
+
+### Fixed
+
+- **M3U group processing retries on poisoned DB connections.** `_db_query_with_retry` now treats psycopg desync errors (`DatabaseError`, e.g. `lost synchronization with server`) as transient; `process_groups` relationship loading uses it so a stale Celery worker connection resets once instead of failing the whole refresh.
+
 ## [0.27.1] - 2026-06-25
 
 ### Security

apps/m3u/tasks.py

@@ -52,9 +52,9 @@ def _db_query_with_retry(fn, *, label="DB query", max_retries=2):
     Poisoned Celery worker connections often surface as OperationalError or as
     ``IndexError: list index out of range`` inside Django's row converters.
     """
-    from django.db import InterfaceError, OperationalError
+    from django.db import DatabaseError, InterfaceError, OperationalError
 
-    transient_errors = (OperationalError, InterfaceError, IndexError)
+    transient_errors = (OperationalError, InterfaceError, IndexError, DatabaseError)
     for attempt in range(max_retries):
         try:
             return fn()
@@ -746,13 +746,16 @@ def process_groups(account, groups, scan_start_time=None):
     all_group_objs = existing_group_objs + newly_created_group_objs
 
     # Get existing relationships for this account
-    existing_relationships = {
-        rel.channel_group.name: rel
-        for rel in ChannelGroupM3UAccount.objects.filter(
-            m3u_account=account,
-            channel_group__name__in=groups.keys()
-        ).select_related('channel_group')
-    }
+    existing_relationships = _db_query_with_retry(
+        lambda: {
+            rel.channel_group.name: rel
+            for rel in ChannelGroupM3UAccount.objects.filter(
+                m3u_account=account,
+                channel_group__name__in=groups.keys(),
+            ).select_related("channel_group")
+        },
+        label=f"process_groups relationships for account {account.id}",
+    )
 
     relations_to_create = []
     relations_to_update = []
@@ -971,6 +974,27 @@ def collect_xc_streams(account_id, enabled_groups):
     )
     return all_streams
 
+
+_STREAM_TOUCH_FIELDS = ("last_seen", "is_stale")
+_STREAM_CHANGED_FIELDS = (
+    "name", "url", "logo_url", "tvg_id", "custom_properties", "is_adult",
+    "last_seen", "updated_at", "is_stale", "stream_id", "stream_chno",
+    "channel_group_id", "is_catchup", "catchup_days",
+)
+
+
+def _bulk_update_stream_refresh_batches(changed_streams, touch_streams, *, batch_size):
+    """Unchanged streams only need last_seen/is_stale; changed rows get the full set."""
+    if touch_streams:
+        Stream.objects.bulk_update(
+            touch_streams, list(_STREAM_TOUCH_FIELDS), batch_size=batch_size,
+        )
+    if changed_streams:
+        Stream.objects.bulk_update(
+            changed_streams, list(_STREAM_CHANGED_FIELDS), batch_size=batch_size,
+        )
+
+
 def process_xc_category_direct(account_id, batch, groups, hash_keys):
     from django.db import connections
 
@@ -981,6 +1005,7 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
 
     streams_to_create = []
     streams_to_update = []
+    streams_to_touch = []
     stream_hashes = {}
 
     try:
@@ -1111,10 +1136,6 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
                     obj.catchup_days != stream_props["catchup_days"]
                 )
 
-                # Keep denormalized catch-up columns in memory (bulk_update reads them).
-                obj.is_catchup = stream_props["is_catchup"]
-                obj.catchup_days = stream_props["catchup_days"]
-
                 if changed:
                     for key, value in stream_props.items():
                         setattr(obj, key, value)
@@ -1123,11 +1144,9 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
                     obj.is_stale = False
                     streams_to_update.append(obj)
                 else:
-                    # Always update last_seen, even if nothing else changed
                     obj.last_seen = timezone.now()
                     obj.is_stale = False
-                    # Don't update updated_at for unchanged streams
-                    streams_to_update.append(obj)
+                    streams_to_touch.append(obj)
 
                 # Remove from existing_streams since we've processed it
                 del existing_streams[stream_hash]
@@ -1144,13 +1163,9 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
                 if streams_to_create:
                     Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True)
 
-                if streams_to_update:
-                    # Simplified bulk update for better performance
-                    Stream.objects.bulk_update(
-                        streams_to_update,
-                        ['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale', 'stream_id', 'stream_chno', 'channel_group_id', 'is_catchup', 'catchup_days'],
-                        batch_size=150  # Smaller batch size for XC processing
-                    )
+                _bulk_update_stream_refresh_batches(
+                    streams_to_update, streams_to_touch, batch_size=150,
+                )
 
                 # Update last_seen for any remaining existing streams that weren't processed
                 if len(existing_streams.keys()) > 0:
@@ -1158,7 +1173,10 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
         except Exception as e:
             logger.error(f"Bulk operation failed for XC streams: {str(e)}")
 
-        retval = f"Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated."
+        retval = (
+            f"Batch processed: {len(streams_to_create)} created, "
+            f"{len(streams_to_update) + len(streams_to_touch)} updated."
+        )
 
     except Exception as e:
         logger.error(f"XC category processing error: {str(e)}")
@@ -1168,7 +1186,7 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
         connections.close_all()
 
     # Aggressive garbage collection
-    del streams_to_create, streams_to_update, stream_hashes, existing_streams
+    del streams_to_create, streams_to_update, streams_to_touch, stream_hashes, existing_streams
     gc.collect()
 
     return retval
@@ -1203,6 +1221,7 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
 
     streams_to_create = []
     streams_to_update = []
+    streams_to_touch = []
     stream_hashes = {}
 
     name_max_length = Stream._meta.get_field('name').max_length
@@ -1350,15 +1369,10 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
                 obj.catchup_days != stream_props["catchup_days"]
             )
 
-            # Keep denormalized catch-up columns in memory (bulk_update reads them).
-            obj.is_catchup = stream_props["is_catchup"]
-            obj.catchup_days = stream_props["catchup_days"]
-
-            # Always update last_seen
             obj.last_seen = timezone.now()
+            obj.is_stale = False
 
             if changed:
-                # Only update fields that changed and set updated_at
                 obj.name = stream_props["name"]
                 obj.url = stream_props["url"]
                 obj.logo_url = stream_props["logo_url"]
@@ -1368,12 +1382,12 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
                 obj.stream_id = stream_props["stream_id"]
                 obj.stream_chno = stream_props["stream_chno"]
                 obj.channel_group_id = stream_props["channel_group_id"]
+                obj.is_catchup = stream_props["is_catchup"]
+                obj.catchup_days = stream_props["catchup_days"]
                 obj.updated_at = timezone.now()
-
-            # Always mark as not stale since we saw it in this refresh
-            obj.is_stale = False
-
-            streams_to_update.append(obj)
+                streams_to_update.append(obj)
+            else:
+                streams_to_touch.append(obj)
         else:
             # New stream
             stream_props["last_seen"] = timezone.now()
@@ -1386,23 +1400,22 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
             if streams_to_create:
                 Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True)
 
-            if streams_to_update:
-                # Update all streams in a single bulk operation
-                Stream.objects.bulk_update(
-                    streams_to_update,
-                    ['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale', 'stream_id', 'stream_chno', 'channel_group_id', 'is_catchup', 'catchup_days'],
-                    batch_size=200
-                )
+            _bulk_update_stream_refresh_batches(
+                streams_to_update, streams_to_touch, batch_size=200,
+            )
     except Exception as e:
         logger.error(f"Bulk operation failed: {str(e)}")
 
-    retval = f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated."
+    retval = (
+        f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, "
+        f"{len(streams_to_update) + len(streams_to_touch)} updated."
+    )
 
     # 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
+    del streams_to_create, streams_to_update, streams_to_touch, stream_hashes, existing_streams
     gc.collect()
 
     return retval
@@ -3814,8 +3827,6 @@ def _refresh_single_m3u_account_impl(account_id):
         except OSError:
             pass
 
-        _release_task_db_connection()
-
     return f"Dispatched jobs complete."
 
 

apps/timeshift/tests/test_views.py

@@ -1901,9 +1901,12 @@ class TimeshiftScrubPreemptTests(TestCase):
 
 
 class RollupSelfHealDbTests(TestCase):
-    """Catch-up flag consistency after stream removal — the ChannelStream signal
-    handles bulk deletes (locked by a regression test) and the account-scoped
-    rollup self-heals stale flags on channels still linked to that account."""
+    """Catch-up flag consistency after stream removal.
+
+    The ChannelStream signal handles bulk deletes (locked by a regression test).
+    The account-scoped rollup self-heals stale flags on channels still linked
+    to that account.
+    """
 
     @classmethod
     def setUpTestData(cls):
@@ -1945,7 +1948,7 @@ class RollupSelfHealDbTests(TestCase):
 
     def test_rollup_self_heals_stale_channel_with_non_catchup_stream(self):
         # Channel still linked to the account but no active catch-up streams
-        # (e.g. catch-up flag cleared on import) — rollup must reset stale flags.
+        # (e.g. catch-up flag cleared on import). Rollup must reset stale flags.
         from apps.channels.models import Channel, ChannelStream, Stream
         from apps.m3u.tasks import rollup_channel_catchup_fields
 
@@ -1988,8 +1991,7 @@ class RollupSelfHealDbTests(TestCase):
 
     def test_rollup_keeps_and_corrects_channels_with_catchup_streams(self):
         # The self-heal pass must not touch channels that legitimately have
-        # catch-up streams — and the account-scoped pass still corrects their
-        # values.
+        # catch-up streams. The account-scoped pass still corrects their values.
         from apps.channels.models import Channel
         from apps.m3u.tasks import rollup_channel_catchup_fields
 

core/utils.py

@@ -565,13 +565,15 @@ def trim_c_allocator_heap():
         return False
 
 
-def cleanup_memory(log_usage=False, force_collection=True):
+def cleanup_memory(log_usage=False, force_collection=True, trim_heap=False):
     """
     Comprehensive memory cleanup function to reduce memory footprint
 
     Args:
         log_usage: Whether to log memory usage before and after cleanup
         force_collection: Whether to force garbage collection
+        trim_heap: Return freed C heap pages to the OS. Only use after DB
+            connections are closed (e.g. Celery task_postrun).
     """
     logger.trace("Starting memory cleanup django memory cleanup")
     # Skip logging if log level is not set to debug or more verbose (like trace)
@@ -606,6 +608,8 @@ def cleanup_memory(log_usage=False, force_collection=True):
             logger.debug(f"Memory after cleanup: {after_mem:.2f} MB (change: {after_mem-before_mem:.2f} MB)")
         except (ImportError, Exception):
             pass
+    if trim_heap:
+        trim_c_allocator_heap()
     logger.trace("Memory cleanup complete for django")
 
 
@@ -618,8 +622,7 @@ def spawn_memory_trim(close_connections=False):
     so the pooled DB connection is released first.
     """
     def _run():
-        cleanup_memory(force_collection=True)
-        trim_c_allocator_heap()
+        cleanup_memory(force_collection=True, trim_heap=True)
 
     if close_connections:
         from django.db import close_old_connections

dispatcharr/celery.py

@@ -99,6 +99,7 @@ def cleanup_task_memory(**kwargs):
     memory_intensive_tasks = [
         'apps.m3u.tasks.refresh_single_m3u_account',
         'apps.m3u.tasks.refresh_m3u_accounts',
+        'apps.m3u.tasks.refresh_m3u_groups',
         'apps.m3u.tasks.process_m3u_batch',
         'apps.m3u.tasks.process_xc_category',
         'apps.m3u.tasks.sync_auto_channels',
@@ -121,7 +122,7 @@ def cleanup_task_memory(**kwargs):
         from core.utils import cleanup_memory
 
         # Use the comprehensive cleanup function
-        cleanup_memory(log_usage=True, force_collection=True)
+        cleanup_memory(log_usage=True, force_collection=True, trim_heap=True)
 
         # Log memory usage if psutil is installed
         try: