From 46695588f73dc27f17930b52e7d84f4ac73860d7 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 18 Jun 2026 14:59:09 -0500 Subject: [PATCH] fix(m3u): enhance custom properties handling and DB connection management - Implemented `ensure_custom_properties_dict()` to normalize custom properties across various models and serializers, addressing issues with legacy JSON-encoded strings. - Updated M3U account and channel group models to ensure custom properties are consistently stored as dictionaries during save operations. - Enhanced Celery task management by ensuring old DB connections are closed before and after tasks, improving reliability and preventing errors during account refresh operations. (Fixes #1338) --- CHANGELOG.md | 2 + apps/channels/compact_numbering.py | 11 +- apps/channels/models.py | 9 +- apps/channels/serializers.py | 11 - apps/m3u/api_views.py | 10 +- apps/m3u/forms.py | 8 +- apps/m3u/models.py | 11 + apps/m3u/serializers.py | 20 +- apps/m3u/tasks.py | 273 ++++++++++++++------- apps/m3u/tests/test_refresh_db_recovery.py | 74 ++++++ core/utils.py | 42 +++- dispatcharr/celery.py | 20 +- tests/test_custom_properties_as_dict.py | 74 ++++++ 13 files changed, 446 insertions(+), 119 deletions(-) create mode 100644 apps/m3u/tests/test_refresh_db_recovery.py create mode 100644 tests/test_custom_properties_as_dict.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1712d447..74a5f6a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **M3U refresh recovers DB connections and account status after failures.** Celery workers now call `close_old_connections()` before and after every task; `refresh_single_m3u_account` also resets its DB connection at task start and retries account loads once after a connection reset (avoids opaque `list index out of range` errors from poisoned worker connections). Accounts leave `fetching`/`parsing` for a terminal `error` or `success` state when refresh aborts. Error-path status updates use `QuerySet.update()` on a clean connection instead of `save()` on a connection that may already be INTRANS after `refresh_m3u_groups` or batch ORM failures. Failed group downloads now set `error` instead of returning silently. Refresh start replaces stale `last_message` text. `refresh_account_profiles` releases DB connections after each profile failure. (Fixes #1338) +- **M3U `custom_properties` JSON normalization.** Legacy rows and some API writes could store a JSON-encoded string instead of an object in `custom_properties`, causing refresh, profile sync, and auto-sync to fail with `'str' object has no attribute 'get'`. M3U account, profile, and group-relation models normalize on `save()`; group settings bulk writes and read/merge paths use shared helpers in `core.utils` without re-parsing dict values. - **XC live playback URL building now normalizes account server URLs.** On-demand live URLs (introduced for Server Group profile rotation in 0.27.0) rebuild `/live/{user}/{pass}/{stream_id}.ts` from current credentials instead of reusing the synced `stream.url`. That path now runs `normalize_server_url()` so pasted API URLs (e.g. `/player_api.php` with query params) are stripped while sub-paths like `/server1` are preserved. `get_transformed_credentials()` normalizes the base URL at the source; VOD movie and episode relations use the same helper instead of constructing a throwaway `XCClient` (which also avoided per-request HTTP session setup). (Fixes #1363) - **Live proxy now releases geventpool DB connections on more paths.** `stream_ts()` calls `close_old_connections()` before `StreamingHttpResponse` so clients waiting on channel init do not keep a pool slot while the generator polls Redis. `generate_stream_url()`, `get_stream_info_for_switch()`, `get_alternate_streams()`, and `get_connections_left()` release in `finally` blocks after ORM work on stream-manager failover paths that run outside Django's request cycle. TS client disconnect cleanup matches fMP4, and `ChannelService` metadata helpers release after their ORM lookups. diff --git a/apps/channels/compact_numbering.py b/apps/channels/compact_numbering.py index fd29a153..27a26cb6 100644 --- a/apps/channels/compact_numbering.py +++ b/apps/channels/compact_numbering.py @@ -25,7 +25,8 @@ import logging from django.db import transaction from .models import Channel, ChannelOverride, ChannelGroupM3UAccount -from apps.m3u.tasks import _custom_properties_as_dict, _next_available_number +from apps.m3u.tasks import _next_available_number +from core.utils import ensure_custom_properties_dict from core.utils import ( acquire_task_lock, natural_sort_key, @@ -37,7 +38,7 @@ logger = logging.getLogger(__name__) def is_compact_group(group_relation): """Return True if the given ChannelGroupM3UAccount is in compact mode.""" - cp = _custom_properties_as_dict(group_relation.custom_properties) + cp = ensure_custom_properties_dict(group_relation.custom_properties) return bool(cp.get("compact_numbering")) @@ -72,7 +73,7 @@ def get_group_relation_for_channel(channel): for rel in ChannelGroupM3UAccount.objects.filter( m3u_account_id=channel.auto_created_by_id ): - cp = _custom_properties_as_dict(rel.custom_properties) + cp = ensure_custom_properties_dict(rel.custom_properties) if str(cp.get("group_override", "")) == target: return rel return None @@ -179,7 +180,7 @@ def assign_compact_numbers_for_channels(channel_ids): for rel in ChannelGroupM3UAccount.objects.filter( m3u_account_id__in={k[0] for k in unresolved} ): - cp = _custom_properties_as_dict(rel.custom_properties) + cp = ensure_custom_properties_dict(rel.custom_properties) target = cp.get("group_override") if not target: continue @@ -295,7 +296,7 @@ def _repack_inner(group_relation): account_id = group_relation.m3u_account_id group_id = group_relation.channel_group_id - cp = _custom_properties_as_dict(group_relation.custom_properties) + cp = ensure_custom_properties_dict(group_relation.custom_properties) sort_order = cp.get("channel_sort_order") or "" sort_reverse = bool(cp.get("channel_sort_reverse")) diff --git a/apps/channels/models.py b/apps/channels/models.py index ef1d0cff..747d0b89 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -2,7 +2,7 @@ from django.db import models from django.core.exceptions import ValidationError from django.conf import settings from core.models import StreamProfile, CoreSettings -from core.utils import RedisClient +from core.utils import RedisClient, custom_properties_as_dict from apps.proxy.live_proxy.redis_keys import RedisKeys from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState import logging @@ -1119,6 +1119,13 @@ class ChannelGroupM3UAccount(models.Model): def __str__(self): return f"{self.channel_group.name} - {self.m3u_account.name} (Enabled: {self.enabled})" + def save(self, *args, **kwargs): + if self.custom_properties is not None and not isinstance( + self.custom_properties, dict + ): + self.custom_properties = custom_properties_as_dict(self.custom_properties) + super().save(*args, **kwargs) + class Logo(models.Model): name = models.CharField(max_length=255) diff --git a/apps/channels/serializers.py b/apps/channels/serializers.py index 9b7d3a2a..2e80d840 100644 --- a/apps/channels/serializers.py +++ b/apps/channels/serializers.py @@ -221,17 +221,6 @@ class ChannelGroupM3UAccountSerializer(serializers.ModelSerializer): return data - def to_internal_value(self, data): - # Accept both dict and JSON string for custom_properties (for backward compatibility) - val = data.get("custom_properties") - if isinstance(val, str): - try: - data["custom_properties"] = json.loads(val) - except Exception: - pass - - return super().to_internal_value(data) - def validate(self, attrs): # Partial PATCHes only carry submitted fields; fill missing # start/end from the instance so the validator catches a PATCH diff --git a/apps/m3u/api_views.py b/apps/m3u/api_views.py index a49ddd6f..1117a226 100644 --- a/apps/m3u/api_views.py +++ b/apps/m3u/api_views.py @@ -23,7 +23,7 @@ logger = logging.getLogger(__name__) from .models import M3UAccount, M3UFilter, ServerGroup, M3UAccountProfile from core.models import UserAgent -from core.utils import safe_upload_path +from core.utils import safe_upload_path, ensure_custom_properties_dict from apps.channels.models import ChannelGroupM3UAccount from core.serializers import UserAgentSerializer from apps.vod.models import M3UVODCategoryRelation @@ -536,7 +536,9 @@ class M3UAccountViewSet(viewsets.ModelViewSet): auto_channel_sync=setting.get("auto_channel_sync", False), auto_sync_channel_start=setting.get("auto_sync_channel_start"), auto_sync_channel_end=setting.get("auto_sync_channel_end"), - custom_properties=setting.get("custom_properties", {}), + custom_properties=ensure_custom_properties_dict( + setting.get("custom_properties") + ), ) for setting in group_settings if setting.get("channel_group") @@ -561,7 +563,9 @@ class M3UAccountViewSet(viewsets.ModelViewSet): category_id=setting["id"], m3u_account=account, enabled=setting.get("enabled", True), - custom_properties=setting.get("custom_properties", {}), + custom_properties=ensure_custom_properties_dict( + setting.get("custom_properties") + ), ) for setting in category_settings if setting.get("id") diff --git a/apps/m3u/forms.py b/apps/m3u/forms.py index cf6586c3..af4d6072 100644 --- a/apps/m3u/forms.py +++ b/apps/m3u/forms.py @@ -1,6 +1,7 @@ # apps/m3u/forms.py from django import forms from .models import M3UAccount, M3UFilter +from core.utils import ensure_custom_properties_dict import re class M3UAccountForm(forms.ModelForm): @@ -28,7 +29,9 @@ class M3UAccountForm(forms.ModelForm): # Set initial value for enable_vod from custom_properties if self.instance and self.instance.custom_properties: - custom_props = self.instance.custom_properties or {} + custom_props = self.instance.custom_properties + if not isinstance(custom_props, dict): + custom_props = ensure_custom_properties_dict(custom_props) self.fields['enable_vod'].initial = custom_props.get('enable_vod', False) def save(self, commit=True): @@ -37,8 +40,9 @@ class M3UAccountForm(forms.ModelForm): # Handle enable_vod field enable_vod = self.cleaned_data.get('enable_vod', False) - # Parse existing custom_properties custom_props = instance.custom_properties or {} + if not isinstance(custom_props, dict): + custom_props = ensure_custom_properties_dict(custom_props) # Update VOD preference custom_props['enable_vod'] = enable_vod diff --git a/apps/m3u/models.py b/apps/m3u/models.py index 53c18d38..7d4fe65a 100644 --- a/apps/m3u/models.py +++ b/apps/m3u/models.py @@ -7,6 +7,7 @@ from django.dispatch import receiver from apps.channels.models import StreamProfile from django_celery_beat.models import PeriodicTask from core.models import CoreSettings, UserAgent +from core.utils import custom_properties_as_dict CUSTOM_M3U_ACCOUNT_NAME = "custom" @@ -124,6 +125,11 @@ class M3UAccount(models.Model): return user_agent def save(self, *args, **kwargs): + if self.custom_properties is not None and not isinstance( + self.custom_properties, dict + ): + self.custom_properties = custom_properties_as_dict(self.custom_properties) + # Prevent auto_now behavior by handling updated_at manually if "update_fields" in kwargs and "updated_at" not in kwargs["update_fields"]: # Don't modify updated_at for regular updates @@ -263,6 +269,11 @@ class M3UAccountProfile(models.Model): def save(self, *args, **kwargs): """Auto-sync exp_date from custom_properties for XC accounts on every save. For non-XC accounts, exp_date is set directly and left untouched here.""" + if self.custom_properties is not None and not isinstance( + self.custom_properties, dict + ): + self.custom_properties = custom_properties_as_dict(self.custom_properties) + parsed = self._parse_exp_date_from_custom_properties() if parsed is not None: # XC account with exp_date in custom_properties — always sync diff --git a/apps/m3u/serializers.py b/apps/m3u/serializers.py index 485a3687..c1dc6f92 100644 --- a/apps/m3u/serializers.py +++ b/apps/m3u/serializers.py @@ -1,4 +1,4 @@ -from core.utils import validate_flexible_url +from core.utils import validate_flexible_url, ensure_custom_properties_dict from rest_framework import serializers, status from rest_framework.response import Response from .models import M3UAccount, M3UFilter, ServerGroup, M3UAccountProfile @@ -270,11 +270,15 @@ class M3UAccountSerializer(serializers.ModelSerializer): # overwrite their corresponding keys; clients should set those via # the typed top-level fields rather than the custom_properties # payload. - incoming_custom = validated_data.get("custom_properties") or {} - custom_props = { - **(instance.custom_properties or {}), - **incoming_custom, - } + incoming_custom = {} + if "custom_properties" in validated_data: + incoming_custom = validated_data["custom_properties"] or {} + if not isinstance(incoming_custom, dict): + incoming_custom = ensure_custom_properties_dict(incoming_custom) + existing_custom = instance.custom_properties or {} + if not isinstance(existing_custom, dict): + existing_custom = ensure_custom_properties_dict(existing_custom) + custom_props = {**existing_custom, **incoming_custom} if enable_vod is not None: custom_props["enable_vod"] = enable_vod @@ -346,7 +350,9 @@ class M3UAccountSerializer(serializers.ModelSerializer): auto_enable_new_groups_series = validated_data.pop("auto_enable_new_groups_series", True) # Parse existing custom_properties or create new - custom_props = validated_data.get("custom_properties", {}) + custom_props = validated_data.get("custom_properties") or {} + if not isinstance(custom_props, dict): + custom_props = ensure_custom_properties_dict(custom_props) # Set preferences (default to True for auto_enable_new_groups) custom_props["enable_vod"] = enable_vod diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index 2d5ac1ee..be2a612f 100644 --- a/apps/m3u/tasks.py +++ b/apps/m3u/tasks.py @@ -7,29 +7,23 @@ import os import gc import gzip, zipfile from concurrent.futures import ThreadPoolExecutor, as_completed -from celery.app.control import Inspect -from celery.result import AsyncResult -from celery import shared_task, current_app, group +from celery import shared_task from django.conf import settings -from django.core.cache import cache from django.db import models, transaction from .models import M3UAccount from apps.channels.models import Stream, ChannelGroup, ChannelGroupM3UAccount -from asgiref.sync import async_to_sync -from channels.layers import get_channel_layer from django.utils import timezone import time import json from core.utils import ( - RedisClient, acquire_task_lock, release_task_lock, TaskLockRenewer, natural_sort_key, log_system_event, + ensure_custom_properties_dict, ) from core.models import CoreSettings, UserAgent -from asgiref.sync import async_to_sync from core.xtream_codes import Client as XCClient from core.utils import send_websocket_update from .utils import normalize_stream_url @@ -39,6 +33,104 @@ logger = logging.getLogger(__name__) BATCH_SIZE = 1500 # Optimized batch size for threading m3u_dir = os.path.join(settings.MEDIA_ROOT, "cached_m3u") +_NON_TERMINAL_REFRESH_STATUSES = frozenset({ + M3UAccount.Status.FETCHING, + M3UAccount.Status.PARSING, +}) + + +def _release_task_db_connection(): + """Return the Celery worker's DB connection to a clean state after ORM errors.""" + from django.db import close_old_connections + close_old_connections() + + +def _db_query_with_retry(fn, *, label="DB query", max_retries=2): + """ + Run an ORM read with one connection reset + retry on transient failures. + + 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 + + transient_errors = (OperationalError, InterfaceError, IndexError) + for attempt in range(max_retries): + try: + return fn() + except transient_errors as exc: + if attempt + 1 >= max_retries: + raise + logger.warning( + "%s failed (%s), resetting DB connection (%s/%s)", + label, + exc, + attempt + 1, + max_retries, + ) + _release_task_db_connection() + + +def _get_active_m3u_account(account_id): + return _db_query_with_retry( + lambda: M3UAccount.objects.get(id=account_id, is_active=True), + label=f"load active M3U account {account_id}", + ) + + +def _set_m3u_account_status( + account_id, + status, + last_message=None, + *, + notify_error=False, + ws_action="parsing", + ws_error=None, +): + """Update account status using a fresh connection (safe after DB failures).""" + _release_task_db_connection() + update = {"status": status} + if last_message is not None: + update["last_message"] = last_message + try: + M3UAccount.objects.filter(id=account_id).update(**update) + if notify_error: + send_m3u_update( + account_id, + ws_action, + 100, + status="error", + error=ws_error or last_message, + ) + except Exception as e: + logger.error( + f"Failed to set account {account_id} status to {status}: {e}" + ) + + +def _ensure_m3u_refresh_terminal_status(account_id): + """Mark refresh as failed when the task exits while still in progress.""" + _release_task_db_connection() + try: + current_status = ( + M3UAccount.objects.filter(id=account_id) + .values_list("status", flat=True) + .first() + ) + if current_status in _NON_TERMINAL_REFRESH_STATUSES: + message = "Refresh did not complete successfully" + M3UAccount.objects.filter(id=account_id).update( + status=M3UAccount.Status.ERROR, + last_message=message, + ) + send_m3u_update( + account_id, "parsing", 100, status="error", error=message + ) + except Exception as e: + logger.debug( + f"Could not verify terminal refresh status for account {account_id}: {e}" + ) + _EXTINF_ATTR_RE = re.compile(r'([^\s=]+)\s*=\s*(["\'])(.*?)\2') @@ -630,7 +722,7 @@ def process_groups(account, groups, scan_start_time=None): logger.info(f"Currently {len(existing_groups)} existing groups") # Check if we should auto-enable new groups based on account settings - account_custom_props = account.custom_properties or {} + account_custom_props = ensure_custom_properties_dict(account.custom_properties) auto_enable_new_groups_live = account_custom_props.get("auto_enable_new_groups_live", True) # Separate existing groups from groups that need to be created @@ -673,7 +765,9 @@ def process_groups(account, groups, scan_start_time=None): existing_rel = existing_relationships[group.name] # Get existing custom properties (now JSONB, no need to parse) - existing_custom_props = existing_rel.custom_properties or {} + existing_custom_props = ensure_custom_properties_dict( + existing_rel.custom_properties + ) # Get the new xc_id from groups data new_xc_id = custom_props.get("xc_id") @@ -696,6 +790,8 @@ def process_groups(account, groups, scan_start_time=None): logger.debug(f"Updated xc_id for group '{group.name}' from '{existing_xc_id}' to '{new_xc_id}' - account {account.id}") else: # Update last_seen even if xc_id hasn't changed + if isinstance(existing_rel.custom_properties, str): + existing_rel.custom_properties = existing_custom_props existing_rel.last_seen = scan_start_time existing_rel.is_stale = False relations_to_update.append(existing_rel) @@ -1755,33 +1851,6 @@ def _range_exhausted_error(mode, start_number, end_number, fallback_start): return f"Channel number range {range_start}-{int(end_number)} is full" -def _custom_properties_as_dict(value): - """ - Normalize a JSONField-backed custom_properties value into a dict. - - Historical data has rows where the field holds a JSON-encoded string - instead of a dict. Django's JSONField serializes whatever it gets, so - `.get()` on one of those rows raises AttributeError and aborts the - entire sync. Treat string values as JSON to parse, and fall back to an - empty dict for anything that isn't a dict after parsing. - """ - import json - - if isinstance(value, dict): - return value - if isinstance(value, str): - try: - parsed = json.loads(value) - except (ValueError, TypeError): - logger.warning( - "custom_properties stored as non-JSON string; ignoring: %r", - value[:100], - ) - return {} - return parsed if isinstance(parsed, dict) else {} - return {} - - def _classify_sync_failure(exc): """ Map an exception raised during per-stream sync to a coarse typed @@ -1890,7 +1959,7 @@ def sync_auto_channels(account_id, scan_start_time=None): custom_epg_id = None # New option: select specific EPG source (takes priority over force_dummy_epg) channel_numbering_mode = "fixed" # Default mode channel_numbering_fallback = 1 # Default fallback for provider mode - group_custom_props = _custom_properties_as_dict( + group_custom_props = ensure_custom_properties_dict( group_relation.custom_properties ) if group_custom_props: @@ -2734,7 +2803,7 @@ def sync_auto_channels(account_id, scan_start_time=None): # "preserve_customized" keeps those with a ChannelOverride row; # "never" disables cleanup. Hidden channels are preserved across all # modes so event/PPV channels that come and go are not silently lost. - cleanup_mode = (account.custom_properties or {}).get( + cleanup_mode = ensure_custom_properties_dict(account.custom_properties).get( "orphan_channel_cleanup", "always" ) if cleanup_mode != "never": @@ -2948,12 +3017,19 @@ def refresh_account_profiles(account_id): if profile_client.authenticate(): # Get account information specific to this profile's credentials profile_account_info = profile_client.get_account_info() + if not isinstance(profile_account_info, dict): + raise TypeError( + f"Unexpected account info type: {type(profile_account_info).__name__}" + ) # Merge with existing custom_properties if they exist - existing_props = profile.custom_properties or {} - existing_props.update(profile_account_info) - profile.custom_properties = existing_props - profile.save(update_fields=['custom_properties']) + profile.custom_properties = { + **ensure_custom_properties_dict( + profile.custom_properties + ), + **profile_account_info, + } + profile.save(update_fields=['custom_properties', 'exp_date']) profiles_updated += 1 logger.info(f"Updated account information for profile '{profile.name}' ({profiles_updated}/{profiles.count()})") @@ -2964,6 +3040,7 @@ def refresh_account_profiles(account_id): except Exception as profile_error: profiles_failed += 1 logger.error(f"Failed to update account information for profile '{profile.name}': {str(profile_error)}") + _release_task_db_connection() # Continue with other profiles even if one fails result_msg = f"Profile refresh complete for account {account.name}: {profiles_updated} updated, {profiles_failed} failed" @@ -2978,6 +3055,8 @@ def refresh_account_profiles(account_id): error_msg = f"Error refreshing profiles for account {account_id}: {str(e)}" logger.error(error_msg) return error_msg + finally: + _release_task_db_connection() @shared_task @@ -3032,10 +3111,11 @@ def refresh_account_info(profile_id): account_info = client.get_account_info() # Update only this specific profile with the new account info - if not profile.custom_properties: - profile.custom_properties = {} - profile.custom_properties.update(account_info) - profile.save() + profile.custom_properties = { + **ensure_custom_properties_dict(profile.custom_properties), + **account_info, + } + profile.save(update_fields=['custom_properties', 'exp_date']) # Send success notification to frontend via websocket send_websocket_update( @@ -3098,10 +3178,26 @@ def refresh_single_m3u_account(account_id): lock_renewer = TaskLockRenewer("refresh_single_m3u_account", account_id) lock_renewer.start() + _release_task_db_connection() + try: return _refresh_single_m3u_account_impl(account_id) + except Exception as e: + logger.error( + f"refresh_single_m3u_account failed for account {account_id}: {e}", + exc_info=True, + ) + _set_m3u_account_status( + account_id, + M3UAccount.Status.ERROR, + f"Error processing M3U: {str(e)[:500]}", + notify_error=True, + ws_error=str(e)[:500], + ) + raise finally: - # Guaranteed cleanup on all exit paths (success, exception, early return) + _ensure_m3u_refresh_terminal_status(account_id) + _release_task_db_connection() lock_renewer.stop() release_task_lock("refresh_single_m3u_account", account_id) @@ -3116,22 +3212,25 @@ def _refresh_single_m3u_account_impl(account_id): streams_deleted = 0 try: - account = M3UAccount.objects.get(id=account_id, is_active=True) + account = _get_active_m3u_account(account_id) if not account.is_active: logger.debug(f"Account {account_id} is not active, skipping.") return - # Set status to fetching - account.status = M3UAccount.Status.FETCHING - account.save(update_fields=['status']) + # Set status to fetching and replace stale completion messages. + _set_m3u_account_status( + account_id, + M3UAccount.Status.FETCHING, + "Refresh in progress...", + ) + account = _get_active_m3u_account(account_id) filters = list(account.filters.all()) # Check if VOD is enabled for this account - vod_enabled = False - if account.custom_properties: - custom_props = account.custom_properties or {} - vod_enabled = custom_props.get('enable_vod', False) + vod_enabled = ensure_custom_properties_dict(account.custom_properties).get( + 'enable_vod', False + ) except M3UAccount.DoesNotExist: # The M3U account doesn't exist, so delete the periodic task if it exists @@ -3197,6 +3296,16 @@ def _refresh_single_m3u_account_impl(account_id): logger.error( f"Failed to refresh M3U groups for account {account_id}: {result}" ) + error_msg = ( + "Failed to refresh M3U groups - download failed or other error" + ) + _set_m3u_account_status( + account_id, + M3UAccount.Status.ERROR, + error_msg, + notify_error=True, + ws_error=error_msg, + ) return "Failed to update m3u account - download failed or other error" extinf_data, groups = result @@ -3211,23 +3320,23 @@ def _refresh_single_m3u_account_impl(account_id): # For XC accounts, empty extinf_data is normal at this stage if not extinf_data and not is_xc_account: logger.error(f"No streams found for non-XC account {account_id}") - account.status = M3UAccount.Status.ERROR - account.last_message = "No streams found in M3U source" - account.save(update_fields=["status", "last_message"]) - send_m3u_update( - account_id, "parsing", 100, status="error", error="No streams found" + error_msg = "No streams found in M3U source" + _set_m3u_account_status( + account_id, + M3UAccount.Status.ERROR, + error_msg, + notify_error=True, + ws_error=error_msg, ) except Exception as e: logger.error(f"Exception in refresh_m3u_groups: {str(e)}", exc_info=True) - account.status = M3UAccount.Status.ERROR - account.last_message = f"Error refreshing M3U groups: {str(e)}" - account.save(update_fields=["status", "last_message"]) - send_m3u_update( + error_msg = f"Error refreshing M3U groups: {str(e)[:500]}" + _set_m3u_account_status( account_id, - "parsing", - 100, - status="error", - error=f"Error refreshing M3U groups: {str(e)}", + M3UAccount.Status.ERROR, + error_msg, + notify_error=True, + ws_error=error_msg, ) return "Failed to update m3u account" @@ -3241,15 +3350,13 @@ def _refresh_single_m3u_account_impl(account_id): # Modified validation logic for different account types if (not groups) or (not is_xc_account and not extinf_data): logger.error(f"No data to process for account {account_id}") - account.status = M3UAccount.Status.ERROR - account.last_message = "No data available for processing" - account.save(update_fields=["status", "last_message"]) - send_m3u_update( + error_msg = "No data available for processing" + _set_m3u_account_status( account_id, - "parsing", - 100, - status="error", - error="No data available for processing", + M3UAccount.Status.ERROR, + error_msg, + notify_error=True, + ws_error=error_msg, ) return "Failed to update m3u account, no data available" @@ -3365,7 +3472,7 @@ def _refresh_single_m3u_account_impl(account_id): group_id = rel.channel_group.id # Load the custom properties with the xc_id - custom_props = rel.custom_properties or {} + custom_props = ensure_custom_properties_dict(rel.custom_properties) if "xc_id" in custom_props: filtered_groups[group_name] = { "xc_id": custom_props["xc_id"], @@ -3590,13 +3697,7 @@ def _refresh_single_m3u_account_impl(account_id): except Exception as e: logger.error(f"Error processing M3U for account {account_id}: {str(e)}") - 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 + raise finally: # Free large data structures regardless of success or failure if 'existing_groups' in locals(): @@ -3623,6 +3724,8 @@ def _refresh_single_m3u_account_impl(account_id): except OSError: pass + _release_task_db_connection() + return f"Dispatched jobs complete." diff --git a/apps/m3u/tests/test_refresh_db_recovery.py b/apps/m3u/tests/test_refresh_db_recovery.py new file mode 100644 index 00000000..42db8bd5 --- /dev/null +++ b/apps/m3u/tests/test_refresh_db_recovery.py @@ -0,0 +1,74 @@ +from unittest.mock import MagicMock, patch + +from django.test import SimpleTestCase + +from apps.m3u.tasks import ( + _db_query_with_retry, + _get_active_m3u_account, + _release_task_db_connection, + refresh_single_m3u_account, +) + + +class DbQueryWithRetryTests(SimpleTestCase): + def test_retries_after_index_error_from_poisoned_connection(self): + fn = MagicMock(side_effect=[IndexError("list index out of range"), "ok"]) + + with patch( + "apps.m3u.tasks._release_task_db_connection" + ) as mock_release: + result = _db_query_with_retry(fn, label="test query") + + self.assertEqual(result, "ok") + self.assertEqual(fn.call_count, 2) + mock_release.assert_called_once() + + def test_raises_after_exhausting_retries(self): + fn = MagicMock(side_effect=IndexError("list index out of range")) + + with patch("apps.m3u.tasks._release_task_db_connection"): + with self.assertRaises(IndexError): + _db_query_with_retry(fn, label="test query", max_retries=2) + + self.assertEqual(fn.call_count, 2) + + +class RefreshTaskDbStartupTests(SimpleTestCase): + @patch("apps.m3u.tasks._ensure_m3u_refresh_terminal_status") + @patch("apps.m3u.tasks._refresh_single_m3u_account_impl") + @patch("apps.m3u.tasks.release_task_lock") + @patch("apps.m3u.tasks.acquire_task_lock", return_value=True) + @patch("apps.m3u.tasks.TaskLockRenewer") + @patch("apps.m3u.tasks._release_task_db_connection") + def test_refresh_releases_db_connection_before_impl( + self, + mock_release, + _mock_renewer, + _mock_acquire, + _mock_release_lock, + mock_impl, + _mock_ensure_terminal, + ): + call_order = [] + + def track_release(): + call_order.append("release") + + mock_release.side_effect = track_release + mock_impl.side_effect = lambda *_a, **_k: call_order.append("impl") or "done" + + result = refresh_single_m3u_account(140) + + self.assertEqual(result, "done") + self.assertEqual(call_order[:2], ["release", "impl"]) + + @patch("apps.m3u.tasks.M3UAccount") + def test_get_active_m3u_account_uses_retry_helper(self, mock_model): + mock_model.objects.get.return_value = MagicMock(is_active=True) + + with patch("apps.m3u.tasks._db_query_with_retry") as mock_retry: + mock_retry.side_effect = lambda fn, **_: fn() + account = _get_active_m3u_account(140) + + mock_retry.assert_called_once() + self.assertTrue(account.is_active) diff --git a/core/utils.py b/core/utils.py index 188d9cb2..29c613b3 100644 --- a/core/utils.py +++ b/core/utils.py @@ -1,3 +1,4 @@ +import json import redis import logging import time @@ -7,7 +8,6 @@ from pathlib import Path import re from django.conf import settings from redis.exceptions import ConnectionError, TimeoutError -from django.core.cache import cache from asgiref.sync import async_to_sync from channels.layers import get_channel_layer from django.core.validators import URLValidator @@ -71,6 +71,46 @@ def natural_sort_key(text): return [convert(c) for c in re.split('([0-9]+)', text)] + +def custom_properties_as_dict(value): + """ + Normalize a JSONField-backed custom_properties value into a dict. + + Historical rows (TextField era and early JSONField migration) may store a + JSON-encoded string instead of an object. API clients can also submit a + string value because JSONField accepts any JSON type. Call this before + reading or merging custom_properties. + """ + if isinstance(value, dict): + return value + if isinstance(value, str): + try: + parsed = json.loads(value) + except (ValueError, TypeError): + logger.warning( + "custom_properties stored as non-JSON string; ignoring: %r", + value[:100], + ) + return {} + return parsed if isinstance(parsed, dict) else {} + if value is None: + return {} + return {} + + +def ensure_custom_properties_dict(value): + """ + Return a dict for read/merge/bulk-write paths. Dict values pass through + without re-parsing. Use model ``save()`` (not this) as the canonical + normalizer for ORM writes that go through ``save()``. + """ + if isinstance(value, dict): + return value + if value is None: + return {} + return custom_properties_as_dict(value) + + class RedisClient: _client = None _buffer = None diff --git a/dispatcharr/celery.py b/dispatcharr/celery.py index 8154b52e..18be81eb 100644 --- a/dispatcharr/celery.py +++ b/dispatcharr/celery.py @@ -2,7 +2,7 @@ import os from celery import Celery import logging -from celery.signals import task_postrun, worker_ready +from celery.signals import task_postrun, task_prerun, worker_ready logger = logging.getLogger(__name__) @@ -68,18 +68,30 @@ app.conf.task_routes = { 'apps.channels.tasks.run_recording': {'queue': 'dvr'}, } + +@task_prerun.connect +def reset_db_connection_before_task(**kwargs): + """Discard stale DB connections before each task (Celery workers are long-lived).""" + from django.db import close_old_connections + + try: + close_old_connections() + except Exception: + pass + + # Add memory cleanup after task completion @task_postrun.connect # Use the imported signal def cleanup_task_memory(**kwargs): """Clean up memory and database connections after each task completes""" - from django.db import connection + from django.db import close_old_connections # Get task name from kwargs task_name = kwargs.get('task').name if kwargs.get('task') else '' - # Close database connection for this Celery worker process + # Return all DB connections to the pool in a clean state try: - connection.close() + close_old_connections() except Exception: pass diff --git a/tests/test_custom_properties_as_dict.py b/tests/test_custom_properties_as_dict.py new file mode 100644 index 00000000..bd863fdc --- /dev/null +++ b/tests/test_custom_properties_as_dict.py @@ -0,0 +1,74 @@ +import uuid + +from django.test import SimpleTestCase, TestCase + +from core.utils import custom_properties_as_dict, ensure_custom_properties_dict +from apps.m3u.models import M3UAccount, M3UAccountProfile +from apps.channels.models import ChannelGroupM3UAccount, ChannelGroup + + +class CustomPropertiesAsDictTests(SimpleTestCase): + def test_dict_passthrough(self): + value = {"enable_vod": True} + self.assertIs(custom_properties_as_dict(value), value) + + def test_json_string_parsed(self): + self.assertEqual( + custom_properties_as_dict('{"enable_vod": true}'), + {"enable_vod": True}, + ) + + def test_non_json_string_returns_empty_dict(self): + self.assertEqual(custom_properties_as_dict("not-json"), {}) + + def test_json_array_returns_empty_dict(self): + self.assertEqual(custom_properties_as_dict("[1, 2]"), {}) + + def test_none_returns_empty_dict(self): + self.assertEqual(custom_properties_as_dict(None), {}) + + +class EnsureCustomPropertiesDictTests(SimpleTestCase): + def test_dict_passthrough_without_reparse(self): + value = {"enable_vod": True} + self.assertIs(ensure_custom_properties_dict(value), value) + + def test_none_returns_empty_dict(self): + self.assertEqual(ensure_custom_properties_dict(None), {}) + + +class CustomPropertiesSaveNormalizationTests(TestCase): + def test_m3u_account_save_rewrites_string_custom_properties(self): + account = M3UAccount.objects.create( + name=f"Test Account {uuid.uuid4().hex[:8]}", + custom_properties='{"enable_vod": true}', + ) + account.refresh_from_db() + self.assertEqual(account.custom_properties, {"enable_vod": True}) + + def test_profile_save_rewrites_string_custom_properties(self): + account = M3UAccount.objects.create( + name=f"Test Account {uuid.uuid4().hex[:8]}" + ) + profile = M3UAccountProfile.objects.get( + m3u_account=account, is_default=True + ) + profile.custom_properties = '{"notes": "hello"}' + profile.save(update_fields=["custom_properties"]) + profile.refresh_from_db() + self.assertEqual(profile.custom_properties, {"notes": "hello"}) + + def test_group_relation_save_rewrites_string_custom_properties(self): + account = M3UAccount.objects.create( + name=f"Test Account {uuid.uuid4().hex[:8]}" + ) + group = ChannelGroup.objects.create( + name=f"Sports {uuid.uuid4().hex[:8]}" + ) + rel = ChannelGroupM3UAccount.objects.create( + m3u_account=account, + channel_group=group, + custom_properties='{"force_dummy_epg": true}', + ) + rel.refresh_from_db() + self.assertEqual(rel.custom_properties, {"force_dummy_epg": True})