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(epg): enhance DB connection management and error handling during EPG refreshes

Browse source 
- Implemented robust DB connection management in `refresh_epg_data` to handle transient errors, ensuring reliable status updates for EPG sources.
- Added retry logic for database queries and improved error handling to prevent task failures from unhandled exceptions.
- Updated EPG source status handling to ensure accurate reporting of errors and terminal states during refresh operations.
This commit is contained in:
SergeantPanda 2026-06-18 16:39:52 -05:00
parent c2ac08fdfd
commit 04394b7eac
3 changed files with 291 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

1
CHANGELOG.md
View file

@ -14,6 +14,7 @@ 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)
- **EPG refresh recovers DB connections and source status after failures.** `refresh_epg_data` now mirrors the M3U hardening: connection reset at task start/end, retried source load after a poisoned-worker connection reset, `QuerySet.update()` for error status on a clean connection, and a `finally` guard that moves sources stuck in `fetching`/`parsing` to `error`. Programme index builds are skipped when programme parsing fails. `parse_channels_only` now persists the error status when a missing file has no URL to refetch.
- **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.
- **`POST /api/epg/import/` no longer loads `programme_index`.** The dummy-source guard uses a narrow existence query instead of `objects.get()`, so import triggers avoid pulling the multi-MB byte-offset index into the uWSGI worker. Request/response shape is unchanged for the UI, plugins, and external importers.
- **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)

275
apps/epg/tasks.py
View file

@ -39,6 +39,105 @@ from core.utils import (
logger = logging.getLogger(__name__)
_NON_TERMINAL_REFRESH_STATUSES = frozenset({
EPGSource.STATUS_FETCHING,
EPGSource.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_epg_source(source_id):
return _db_query_with_retry(
lambda: EPGSource.objects.get(id=source_id),
label=f"load EPG source {source_id}",
)
def _set_epg_source_status(
source_id,
status,
last_message=None,
*,
notify_error=False,
ws_action="refresh",
ws_error=None,
):
"""Update source 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:
EPGSource.objects.filter(id=source_id).update(**update)
if notify_error:
send_epg_update(
source_id,
ws_action,
100,
status="error",
error=ws_error or last_message,
)
except Exception as e:
logger.error(
f"Failed to set EPG source {source_id} status to {status}: {e}"
)
def _ensure_epg_refresh_terminal_status(source_id):
"""Mark refresh as failed when the task exits while still in progress."""
_release_task_db_connection()
try:
current_status = (
EPGSource.objects.filter(id=source_id)
.values_list("status", flat=True)
.first()
)
if current_status in _NON_TERMINAL_REFRESH_STATUSES:
message = "Refresh did not complete successfully"
EPGSource.objects.filter(id=source_id).update(
status=EPGSource.STATUS_ERROR,
last_message=message,
)
send_epg_update(
source_id, "refresh", 100, status="error", error=message
)
except Exception as e:
logger.debug(
f"Could not verify terminal refresh status for EPG source {source_id}: {e}"
)
SD_BASE_URL = 'https://json.schedulesdirect.org/20141201'
SD_DAYS_TO_FETCH = 20
SD_PROGRAM_BATCH_SIZE = 5000
@ -498,104 +597,90 @@ def refresh_epg_data(source_id, force=False):
lock_renewer = TaskLockRenewer('refresh_epg_data', source_id)
lock_renewer.start()
source = None
_release_task_db_connection()
try:
# Try to get the EPG source
try:
source = EPGSource.objects.get(id=source_id)
except EPGSource.DoesNotExist:
# The EPG source doesn't exist, so delete the periodic task if it exists
logger.warning(f"EPG source with ID {source_id} not found, but task was triggered. Cleaning up orphaned task.")
# Call the shared function to delete the task
if delete_epg_refresh_task_by_id(source_id):
logger.info(f"Successfully cleaned up orphaned task for EPG source {source_id}")
else:
logger.info(f"No orphaned task found for EPG source {source_id}")
# Release the lock and exit
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
# Force garbage collection before exit
gc.collect()
return f"EPG source {source_id} does not exist, task cleaned up"
# The source exists but is not active, just skip processing
if not source.is_active:
logger.info(f"EPG source {source_id} is not active. Skipping.")
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
# Force garbage collection before exit
gc.collect()
return
# Skip refresh for dummy EPG sources - they don't need refreshing
if source.source_type == 'dummy':
logger.info(f"Skipping refresh for dummy EPG source {source.name} (ID: {source_id})")
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
gc.collect()
return
# Continue with the normal processing...
logger.info(f"Processing EPGSource: {source.name} (type: {source.source_type})")
if source.source_type == 'xmltv':
# Invalidate the byte-offset index before downloading the new file
# so stale offsets are never used during the refresh window.
EPGSource.objects.filter(id=source.id).update(programme_index=None)
fetch_success = fetch_xmltv(source)
if not fetch_success:
logger.error(f"Failed to fetch XMLTV for source {source.name}")
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
# Force garbage collection before exit
gc.collect()
return
parse_channels_success = parse_channels_only(source)
if not parse_channels_success:
logger.error(f"Failed to parse channels for source {source.name}")
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
# Force garbage collection before exit
gc.collect()
return
# Build byte-offset index after programme data is committed so refresh
# does not compete for memory/IO during the programme swap.
parse_programs_for_source(source)
build_programme_index_task.delay(source.id)
elif source.source_type == 'schedules_direct':
fetch_schedules_direct(source, force=force)
source.save(update_fields=['updated_at'])
# After successful EPG refresh, evaluate DVR series rules to schedule new episodes
try:
from apps.channels.tasks import evaluate_series_rules
evaluate_series_rules.delay()
except Exception:
pass
return _refresh_epg_data_impl(source_id, force=force)
except Exception as e:
logger.error(f"Error in refresh_epg_data for source {source_id}: {e}", exc_info=True)
try:
if source:
source.status = 'error'
source.last_message = f"Error refreshing EPG data: {str(e)}"
source.save(update_fields=['status', 'last_message'])
send_epg_update(source_id, "refresh", 100, status="error", error=str(e))
except Exception as inner_e:
logger.error(f"Error updating source status: {inner_e}")
logger.error(
f"Error in refresh_epg_data for source {source_id}: {e}",
exc_info=True,
)
_set_epg_source_status(
source_id,
EPGSource.STATUS_ERROR,
f"Error refreshing EPG data: {str(e)[:500]}",
notify_error=True,
ws_error=str(e)[:500],
)
finally:
# Clear references to ensure proper garbage collection
source = None
# Force garbage collection before releasing the lock
_ensure_epg_refresh_terminal_status(source_id)
_release_task_db_connection()
gc.collect()
connection.close()
lock_renewer.stop()
release_task_lock('refresh_epg_data', source_id)
def _refresh_epg_data_impl(source_id, force=False):
try:
source = _get_epg_source(source_id)
except EPGSource.DoesNotExist:
logger.warning(
f"EPG source with ID {source_id} not found, but task was triggered. "
"Cleaning up orphaned task."
)
if delete_epg_refresh_task_by_id(source_id):
logger.info(
f"Successfully cleaned up orphaned task for EPG source {source_id}"
)
else:
logger.info(f"No orphaned task found for EPG source {source_id}")
return f"EPG source {source_id} does not exist, task cleaned up"
if not source.is_active:
logger.info(f"EPG source {source_id} is not active. Skipping.")
return
if source.source_type == 'dummy':
logger.info(
f"Skipping refresh for dummy EPG source {source.name} (ID: {source_id})"
)
return
logger.info(f"Processing EPGSource: {source.name} (type: {source.source_type})")
if source.source_type == 'xmltv':
# Invalidate the byte-offset index before downloading the new file
# so stale offsets are never used during the refresh window.
EPGSource.objects.filter(id=source.id).update(programme_index=None)
if not fetch_xmltv(source):
logger.error(f"Failed to fetch XMLTV for source {source.name}")
return
if not parse_channels_only(source):
logger.error(f"Failed to parse channels for source {source.name}")
return
# Build byte-offset index after programme data is committed so refresh
# does not compete for memory/IO during the programme swap.
if not parse_programs_for_source(source):
logger.error(f"Failed to parse programs for source {source.name}")
return
build_programme_index_task.delay(source.id)
elif source.source_type == 'schedules_direct':
fetch_schedules_direct(source, force=force)
EPGSource.objects.filter(id=source.id).update(updated_at=timezone.now())
try:
from apps.channels.tasks import evaluate_series_rules
evaluate_series_rules.delay()
except Exception:
pass
def fetch_xmltv(source):
# Handle cases with local file but no URL
if not source.url and source.file_path and os.path.exists(source.file_path):
@ -1167,7 +1252,15 @@ def parse_channels_only(source):
# Update status to error
source.status = 'error'
source.last_message = f"No URL provided, cannot fetch EPG data"
source.save(update_fields=['updated_at'])
source.save(update_fields=['status', 'last_message'])
send_epg_update(
source.id,
"parsing_channels",
100,
status="error",
error="No URL provided",
)
return False
# Initialize process variable for memory tracking only in debug mode
try:

106
apps/epg/tests/test_refresh_db_recovery.py Normal file
View file

@ -0,0 +1,106 @@
from unittest.mock import MagicMock, patch
from django.test import SimpleTestCase
from apps.epg.tasks import (
_db_query_with_retry,
_ensure_epg_refresh_terminal_status,
_get_epg_source,
_release_task_db_connection,
refresh_epg_data,
)
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.epg.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.epg.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.epg.tasks._ensure_epg_refresh_terminal_status")
@patch("apps.epg.tasks._refresh_epg_data_impl")
@patch("apps.epg.tasks.release_task_lock")
@patch("apps.epg.tasks.acquire_task_lock", return_value=True)
@patch("apps.epg.tasks.TaskLockRenewer")
@patch("apps.epg.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_epg_data(42)
self.assertEqual(result, "done")
self.assertEqual(call_order[:2], ["release", "impl"])
@patch("apps.epg.tasks.EPGSource")
def test_get_epg_source_uses_retry_helper(self, mock_model):
mock_source = MagicMock(id=42)
mock_model.objects.get.return_value = mock_source
with patch("apps.epg.tasks._db_query_with_retry") as mock_retry:
mock_retry.side_effect = lambda fn, **_: fn()
source = _get_epg_source(42)
mock_retry.assert_called_once()
mock_model.objects.get.assert_called_once_with(id=42)
self.assertIs(source, mock_source)
class EnsureEpgTerminalStatusTests(SimpleTestCase):
@patch("apps.epg.tasks.send_epg_update")
@patch("apps.epg.tasks._release_task_db_connection")
def test_marks_stuck_fetching_as_error(self, _mock_release, mock_ws):
with patch("apps.epg.tasks.EPGSource") as mock_model:
mock_model.STATUS_ERROR = "error"
qs = MagicMock()
mock_model.objects.filter.return_value = qs
qs.values_list.return_value.first.return_value = "fetching"
_ensure_epg_refresh_terminal_status(7)
qs.update.assert_called_once()
mock_ws.assert_called_once()
@patch("apps.epg.tasks.send_epg_update")
@patch("apps.epg.tasks._release_task_db_connection")
def test_leaves_success_unchanged(self, _mock_release, mock_ws):
with patch("apps.epg.tasks.EPGSource") as mock_model:
qs = MagicMock()
mock_model.objects.filter.return_value = qs
qs.values_list.return_value.first.return_value = "success"
_ensure_epg_refresh_terminal_status(7)
qs.update.assert_not_called()
mock_ws.assert_not_called()