Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-17 16:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix(m3u): enhance custom properties handling and DB connection management

Browse source 
- 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)
This commit is contained in:
SergeantPanda 2026-06-18 14:59:09 -05:00
parent cfe58db222
commit 46695588f7
13 changed files with 446 additions and 119 deletions
Download patch file Download diff file Expand all files Collapse all files

2
CHANGELOG.md
View file

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

11
apps/channels/compact_numbering.py
View file

@ -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"))

9
apps/channels/models.py
View file

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

11
apps/channels/serializers.py
View file

@ -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

10
apps/m3u/api_views.py
View file

@ -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")

8
apps/m3u/forms.py
View file

@ -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

11
apps/m3u/models.py
View file

@ -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

20
apps/m3u/serializers.py
View file

@ -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

273
apps/m3u/tasks.py
View file

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

74
apps/m3u/tests/test_refresh_db_recovery.py Normal file
View file

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

42
core/utils.py
View file

@ -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

20
dispatcharr/celery.py
View file

@ -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

74
tests/test_custom_properties_as_dict.py Normal file
View file

@ -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})