From 0dc8898e8b413764012ef14e80e0a0329c8dd7d1 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 23 Jun 2026 15:10:27 -0500 Subject: [PATCH] refactor(epg): separate programme index into dedicated table for optimized data handling - Moved the `programme_index` from the `EPGSource` model to a new `EPGSourceIndex` table, ensuring that the large JSON blob is only loaded when explicitly accessed, thus improving query performance and memory efficiency. - Updated related queries and API views to utilize the new structure, including adjustments to EPG generation and import logic to prevent unnecessary data loading. - Enhanced memory management in the EPG grid endpoint to reduce worker RSS during response handling. --- CHANGELOG.md | 7 ++- apps/epg/api_views.py | 25 ++++++----- apps/epg/migrations/0026_epgsourceindex.py | 52 ++++++++++++++++++++++ apps/epg/models.py | 37 ++++++++++++--- apps/epg/tasks.py | 17 ++++--- apps/epg/tests/test_epg_import_api.py | 7 ++- apps/epg/tests/test_programme_index.py | 1 - apps/output/epg.py | 38 +++++----------- apps/output/tests.py | 12 ++++- core/tasks.py | 8 +++- core/tests.py | 6 +-- core/utils.py | 24 ++++++++++ 12 files changed, 169 insertions(+), 65 deletions(-) create mode 100644 apps/epg/migrations/0026_epgsourceindex.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 825a06fc..0ad1ef3b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,11 +11,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Lighter EPG export: fewer escape calls and a slimmer channel prefetch.** The primary channel id is escaped once per `epg_id` group instead of once per programme (saves ~750k `html.escape` calls on a large guide). The per-channel `streams` prefetch now loads only `id`/`name` (the only fields the export reads) instead of full `Stream` rows, reducing worker RSS during the channel phase. - **XMLTV EPG export streams without holding the full guide in the worker.** `generate_epg()` streams incrementally; on a cache miss each chunk is pushed to a Redis list as it is yielded (no `''.join()` in the worker). Repeat requests within 300s stream chunks back one at a time from Redis. Post-export `malloc_trim` runs after cold builds. Real programmes use fast `(epg_id, id)` keyset pagination with a per-source `start_time` sort before emit. -- **EPG export no longer fetches the multi-MB `programme_index` per channel.** The channel query `select_related`s `epg_data__epg_source`, which pulled and JSON-parsed each source's byte-offset `programme_index` blob once per channel (~13s of the request on a ~2000-channel guide). It is now `defer()`red since EPG generation never reads it. +- **EPG export no longer fetches the multi-MB `programme_index` per channel.** The channel query `select_related`s `epg_data__epg_source`, which pulled and JSON-parsed each source's byte-offset `programme_index` blob once per channel (~13s of the request on a ~2000-channel guide). The index now lives in its own `EPGSourceIndex` table, so the JOIN never pulls it. - **`ProgramData` composite index `(epg_id, id)`.** The EPG export scans hundreds of thousands of programmes with keyset pagination on `(epg_id, id)`; without a matching index PostgreSQL re-sorted every chunk. A composite index (created `CONCURRENTLY` on PostgreSQL so it does not lock the table) lets each chunk use an ordered index range scan. +- **EPG grid endpoint releases its payload memory back to the OS.** `/api/epg/grid/` drops the redundant full-list copy when appending dummy programmes and runs `malloc_trim` once the response is sent, so worker RSS no longer ratchets up ~20MB per request. ### Changed +- **`programme_index` moved off `EPGSource` into a dedicated `EPGSourceIndex` table.** The multi-MB byte-offset index was repeatedly pulled into web and Celery workers by ordinary `EPGSource` queries and `select_related` JOINs (which ignore manager-level `defer()`). It now lives in a one-to-one `EPGSourceIndex` row, read only when explicitly accessed through the `EPGSource.programme_index` property, so no list, detail, or JOIN query can load it by accident. Migration `0026` copies existing indices across. No API or index-build behavior change. + - **EPG generation extracted into `apps/output/epg.py`.** All XMLTV output logic (`generate_epg`, `generate_dummy_programs`, `generate_custom_dummy_programs`, `generate_dummy_epg`, and supporting helpers) moved from `apps/output/views.py` into a dedicated module. `views.py` retains the thin HTTP endpoint wrappers and auth checks; `epg.py` handles all content generation. No behavior change. - **Channel list with nested streams loads faster.** `GET /api/channels/channels/?include_streams=true` (Channels UI and single-channel fetch) now builds nested stream payloads from the prefetched `channelstream_set` instead of issuing one extra streams M2M query per channel. @@ -29,7 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **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. +- **`POST /api/epg/import/` no longer loads the full EPG source row.** The dummy-source guard uses a narrow existence query instead of `objects.get()`, keeping the import trigger lightweight. 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) - **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/epg/api_views.py b/apps/epg/api_views.py index fbc419c5..1e0bfc41 100644 --- a/apps/epg/api_views.py +++ b/apps/epg/api_views.py @@ -64,8 +64,6 @@ class EPGSourceViewSet(viewsets.ModelViewSet): from apps.channels.models import Channel return EPGSource.objects.select_related( "refresh_task__crontab", "refresh_task__interval" - ).defer( - 'programme_index' ).annotate( has_channels=Exists( Channel.objects.filter(epg_data__epg_source_id=OuterRef('pk')) @@ -1082,13 +1080,19 @@ class EPGGridAPIView(APIView): f"Error creating standard dummy programs for channel {channel.name} (ID: {channel.id}): {str(e)}" ) - # Combine regular and dummy programs - all_programs = list(serialized_programs) + dummy_programs + # Combine regular and dummy programs in place to avoid copying the large list + serialized_programs.extend(dummy_programs) logger.debug( - f"EPGGridAPIView: Returning {len(all_programs)} total programs (including {len(dummy_programs)} dummy programs)." + f"EPGGridAPIView: Returning {len(serialized_programs)} total programs (including {len(dummy_programs)} dummy programs)." ) - return Response({"data": all_programs}, status=status.HTTP_200_OK) + # The grid materializes tens of thousands of program dicts plus the + # rendered JSON; trim once the response is sent so worker RSS does not + # ratchet up per request. + from core.utils import spawn_memory_trim + response = Response({"data": serialized_programs}, status=status.HTTP_200_OK) + response._resource_closers.append(spawn_memory_trim) + return response # ───────────────────────────── @@ -1119,7 +1123,7 @@ class EPGImportAPIView(APIView): epg_id = request.data.get("id", None) force = bool(request.data.get("force", False)) - # Reject dummy sources without loading programme_index (multi-MB JSON). + # Reject dummy sources with a narrow existence query, no full row load. if epg_id is not None: from .models import EPGSource @@ -1236,10 +1240,9 @@ class CurrentProgramsAPIView(APIView): # Limit to 50 IDs per request epg_data_ids = epg_data_ids[:50] - # Defer the multi-MB programme_index the JOIN would pull once per row. The lookup reads it via a targeted refresh_from_db - epg_data_entries = EPGData.objects.select_related('epg_source').defer( - 'epg_source__programme_index' - ).filter(id__in=epg_data_ids) + epg_data_entries = EPGData.objects.select_related('epg_source').filter( + id__in=epg_data_ids + ) # Batch-fetch current programs for all requested EPG entries in one query db_programs = ProgramData.objects.filter( diff --git a/apps/epg/migrations/0026_epgsourceindex.py b/apps/epg/migrations/0026_epgsourceindex.py new file mode 100644 index 00000000..75d9d4eb --- /dev/null +++ b/apps/epg/migrations/0026_epgsourceindex.py @@ -0,0 +1,52 @@ +import django.db.models.deletion +from django.db import migrations, models + + +def copy_index_forward(apps, schema_editor): + EPGSource = apps.get_model('epg', 'EPGSource') + EPGSourceIndex = apps.get_model('epg', 'EPGSourceIndex') + rows = list( + EPGSource.objects.exclude(programme_index__isnull=True).values_list( + 'id', 'programme_index' + ) + ) + for source_id, data in rows: + EPGSourceIndex.objects.update_or_create( + source_id=source_id, defaults={'data': data} + ) + + +def copy_index_backward(apps, schema_editor): + EPGSource = apps.get_model('epg', 'EPGSource') + EPGSourceIndex = apps.get_model('epg', 'EPGSourceIndex') + for source_id, data in EPGSourceIndex.objects.values_list('source_id', 'data'): + EPGSource.objects.filter(id=source_id).update(programme_index=data) + + +class Migration(migrations.Migration): + + dependencies = [ + ('epg', '0025_programdata_epg_id_index'), + ] + + operations = [ + migrations.CreateModel( + name='EPGSourceIndex', + fields=[ + ('source', models.OneToOneField( + on_delete=django.db.models.deletion.CASCADE, + primary_key=True, + related_name='index_record', + serialize=False, + to='epg.epgsource', + )), + ('data', models.JSONField(blank=True, default=None, null=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ], + ), + migrations.RunPython(copy_index_forward, copy_index_backward), + migrations.RemoveField( + model_name='epgsource', + name='programme_index', + ), + ] diff --git a/apps/epg/models.py b/apps/epg/models.py index 04c6072c..b85453af 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -62,12 +62,6 @@ class EPGSource(models.Model): blank=True, help_text="Last status message, including success results or error information" ) - programme_index = models.JSONField( - null=True, - blank=True, - default=None, - help_text="Byte-offset index mapping tvg_id to file positions, built after each EPG refresh" - ) created_at = models.DateTimeField( auto_now_add=True, help_text="Time when this source was created" @@ -124,6 +118,37 @@ class EPGSource(models.Model): kwargs['update_fields'].remove('updated_at') super().save(*args, **kwargs) + @property + def programme_index(self): + """Byte-offset index for this source, read on demand from the separate + EPGSourceIndex table so the multi-MB blob is never pulled into EPGSource + queries or select_related JOINs. Returns the stored dict or None.""" + return ( + EPGSourceIndex.objects.filter(source_id=self.pk) + .values_list('data', flat=True) + .first() + ) + + +class EPGSourceIndex(models.Model): + """Byte-offset programme index for an EPGSource, stored in its own table. + + Kept out of EPGSource so the multi-MB JSON blob is only loaded when read + explicitly, never when querying or joining EPGSource rows. + """ + source = models.OneToOneField( + EPGSource, + on_delete=models.CASCADE, + related_name='index_record', + primary_key=True, + ) + data = models.JSONField(null=True, blank=True, default=None) + updated_at = models.DateTimeField(auto_now=True) + + def __str__(self): + return f"Programme index for source {self.source_id}" + + class EPGData(models.Model): tvg_id = models.CharField(max_length=255, null=True, blank=True, db_index=True) name = models.CharField(max_length=512) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 9661c5b9..83aaa275 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -26,7 +26,7 @@ from core.models import UserAgent, CoreSettings from asgiref.sync import async_to_sync from channels.layers import get_channel_layer -from .models import EPGSource, EPGData, ProgramData, SDScheduleMD5, SDProgramMD5 +from .models import EPGSource, EPGSourceIndex, EPGData, ProgramData, SDScheduleMD5, SDProgramMD5 from core.utils import ( acquire_task_lock, is_task_lock_held, @@ -653,7 +653,9 @@ def _refresh_epg_data_impl(source_id, force=False): 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) + EPGSourceIndex.objects.update_or_create( + source_id=source.id, defaults={'data': None} + ) if not fetch_xmltv(source): logger.error(f"Failed to fetch XMLTV for source {source.name}") return @@ -4484,7 +4486,7 @@ def _programme_to_dict(elem, start_time, end_time): def build_programme_index(source_id): """ Scan the XML file with raw binary I/O to build a {tvg_id: [byte_offset, ...]} map. - Persists the result to EPGSource.programme_index. Most XMLTV files group programmes + Persists the result to the EPGSourceIndex table. Most XMLTV files group programmes by channel, but some split a channel across multiple non-contiguous blocks, so we record block starts up to _OFFSET_CAP and mark only channels that exceed the cap as interleaved. @@ -4573,7 +4575,9 @@ def build_programme_index(source_id): 'channels': index, 'interleaved_channels': sorted(interleaved_channels), } - EPGSource.objects.filter(id=source_id).update(programme_index=result) + EPGSourceIndex.objects.update_or_create( + source_id=source_id, defaults={'data': result} + ) @shared_task @@ -4620,9 +4624,8 @@ def find_current_program_for_tvg_id(epg_or_id): return None now = timezone.now() - # Force a fresh read of the DB-backed index to avoid using stale related-object - # state when an EPG refresh invalidates/rebuilds the index concurrently. - source.refresh_from_db(fields=['programme_index']) + # The property reads the EPGSourceIndex table fresh on each access, so a + # concurrent refresh invalidating/rebuilding the index can't serve stale state. index = source.programme_index if index is not None: diff --git a/apps/epg/tests/test_epg_import_api.py b/apps/epg/tests/test_epg_import_api.py index c58f2a45..cc96c214 100644 --- a/apps/epg/tests/test_epg_import_api.py +++ b/apps/epg/tests/test_epg_import_api.py @@ -7,7 +7,7 @@ from django.test import TestCase from rest_framework import status from rest_framework.test import APIClient -from apps.epg.models import EPGSource +from apps.epg.models import EPGSource, EPGSourceIndex User = get_user_model() @@ -47,7 +47,10 @@ class EPGImportAPITests(TestCase): name="Large Index XMLTV", source_type="xmltv", url="http://example.com/epg.xml", - programme_index={ + ) + EPGSourceIndex.objects.create( + source=source, + data={ "channels": {f"ch.{i}": {"offsets": [0, 100]} for i in range(200)}, "interleaved_channels": [], }, diff --git a/apps/epg/tests/test_programme_index.py b/apps/epg/tests/test_programme_index.py index 5ece0ae1..829afc87 100644 --- a/apps/epg/tests/test_programme_index.py +++ b/apps/epg/tests/test_programme_index.py @@ -519,7 +519,6 @@ class FindCurrentProgramTests(TestCase): name="No Index", source_type="xmltv", file_path=FIXTURE_XML, - programme_index=None, ) epg = EPGData.objects.create( tvg_id="channel.current", diff --git a/apps/output/epg.py b/apps/output/epg.py index 4e11fbe8..626b21aa 100644 --- a/apps/output/epg.py +++ b/apps/output/epg.py @@ -40,34 +40,20 @@ def _programme_overlaps_export_window(start_time, end_time, lookback_cutoff, cut def _ceil_to_half_hour(dt): """Round a datetime up to the next :00 or :30 boundary.""" - dt = dt.replace(second=0, microsecond=0) - remainder = dt.minute % 30 - if remainder == 0: - return dt - return dt + timedelta(minutes=30 - remainder) + original = dt.replace(microsecond=0) + aligned = dt.replace(second=0, microsecond=0) + remainder = aligned.minute % 30 + if remainder != 0: + aligned += timedelta(minutes=30 - remainder) + if aligned < original: + aligned += timedelta(minutes=30) + return aligned def _epg_export_teardown(): - from django.db import close_old_connections + from core.utils import spawn_memory_trim - from core.utils import ( - _is_gevent_monkey_patched, - cleanup_memory, - trim_c_allocator_heap, - ) - - close_old_connections() - - def _run(): - cleanup_memory(force_collection=True) - trim_c_allocator_heap() - - if _is_gevent_monkey_patched(): - import gevent - - gevent.spawn(_run) - else: - _run() + spawn_memory_trim(close_connections=True) def _ordered_channel_streams(channel): @@ -1183,10 +1169,6 @@ def generate_epg(request, profile_name=None, user=None): with_effective_values(base_qs, select_related_fks=True) .exclude(hidden_from_output=True) .order_by("effective_channel_number") - # programme_index is a multi-MB JSON byte-offset index that EPG - # generation never reads; defer it so it isn't fetched and JSON-parsed - # once per channel (was ~13s of the request on large guides). - .defer("epg_data__epg_source__programme_index") .prefetch_related( Prefetch('streams', queryset=Stream.objects.only('id', 'name').order_by('channelstream__order')) ) diff --git a/apps/output/tests.py b/apps/output/tests.py index 5f4fd5f3..870ff765 100644 --- a/apps/output/tests.py +++ b/apps/output/tests.py @@ -355,4 +355,14 @@ class OutputEPGHelperTest(SimpleTestCase): aligned = _ceil_to_half_hour(dt) self.assertEqual(aligned.minute, 30) self.assertEqual(aligned.second, 0) - self.assertGreater(aligned, dt.replace(second=0, microsecond=0)) + self.assertGreaterEqual(aligned, dt.replace(microsecond=0)) + + def test_ceil_to_half_hour_past_boundary_second(self): + from django.utils import timezone + from apps.output.epg import _ceil_to_half_hour + + dt = timezone.now().replace(minute=0, second=52, microsecond=123456) + aligned = _ceil_to_half_hour(dt) + self.assertEqual(aligned.minute, 30) + self.assertEqual(aligned.second, 0) + self.assertGreaterEqual(aligned, dt.replace(microsecond=0)) diff --git a/core/tasks.py b/core/tasks.py index b562f335..a64469b4 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -398,12 +398,16 @@ def scan_and_process_files(): def _rebuild_programme_indices(): """Queue index builds for active EPG sources that are missing their DB index.""" try: + from django.db.models import Q from apps.epg.tasks import build_programme_index_task sources = EPGSource.objects.filter( is_active=True, - programme_index__isnull=True, - ).exclude(source_type__in=('dummy', 'schedules_direct')) + ).exclude( + source_type__in=('dummy', 'schedules_direct') + ).filter( + Q(index_record__isnull=True) | Q(index_record__data__isnull=True) + ) count = 0 for source in sources: diff --git a/core/tests.py b/core/tests.py index 33475ae7..2af52f07 100644 --- a/core/tests.py +++ b/core/tests.py @@ -2,7 +2,7 @@ from unittest.mock import patch, MagicMock from django.test import TestCase, SimpleTestCase -from apps.epg.models import EPGSource +from apps.epg.models import EPGSource, EPGSourceIndex from core.models import CoreSettings, DVR_SETTINGS_KEY, EPG_SETTINGS_KEY @@ -37,14 +37,10 @@ class DispatcharrUserAgentTests(TestCase): class ProgrammeIndexRebuildTests(TestCase): def test_startup_rebuild_does_not_lock_out_queued_build_task(self): - EPGSource.objects.update( - programme_index={"channels": {}, "interleaved_channels": []} - ) source = EPGSource.objects.create( name="Missing Index", source_type="xmltv", is_active=True, - programme_index=None, ) class FakeRedis: diff --git a/core/utils.py b/core/utils.py index 89228088..44504b6a 100644 --- a/core/utils.py +++ b/core/utils.py @@ -608,6 +608,30 @@ def cleanup_memory(log_usage=False, force_collection=True): pass logger.trace("Memory cleanup complete for django") + +def spawn_memory_trim(close_connections=False): + """Reclaim a request's heap pages: GC, then return freed C pages to the OS. + + On gevent uWSGI workers the trim runs in a spawned greenlet so it never + blocks the caller; Celery prefork workers (no gevent hub) run it inline. + Set close_connections=True when called from a streaming generator's teardown + so the pooled DB connection is released first. + """ + def _run(): + cleanup_memory(force_collection=True) + trim_c_allocator_heap() + + if close_connections: + from django.db import close_old_connections + close_old_connections() + + if _is_gevent_monkey_patched(): + import gevent + gevent.spawn(_run) + else: + _run() + + def safe_upload_path(filename: str, base_dir) -> str: """Return a safe absolute path for an uploaded file within base_dir.