diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ad83500..ce80b8e5 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index aa67335f..7ef71dda 100644 --- a/apps/m3u/tasks.py +++ b/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." diff --git a/apps/timeshift/tests/test_views.py b/apps/timeshift/tests/test_views.py index 535e455e..270dfbe5 100644 --- a/apps/timeshift/tests/test_views.py +++ b/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 diff --git a/core/utils.py b/core/utils.py index 44504b6a..16d9ed85 100644 --- a/core/utils.py +++ b/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 diff --git a/dispatcharr/celery.py b/dispatcharr/celery.py index 18be81eb..16a393dc 100644 --- a/dispatcharr/celery.py +++ b/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: