diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c0b1729..5d8aeeb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 0e8eda42..9661c5b9 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -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: diff --git a/apps/epg/tests/test_refresh_db_recovery.py b/apps/epg/tests/test_refresh_db_recovery.py new file mode 100644 index 00000000..11885b34 --- /dev/null +++ b/apps/epg/tests/test_refresh_db_recovery.py @@ -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()