Merge pull request #1383 from Dispatcharr/epg-output-refactor

Epg output refactor

CHANGELOG.md

@@ -7,8 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Performance
+
+- **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). 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.
 
 ### Performance
@@ -20,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.
 

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'))
@@ -986,7 +984,9 @@ class EPGGridAPIView(APIView):
                     channel_name=name_to_parse,
                     num_days=1,
                     program_length_hours=4,
-                    epg_source=epg_source
+                    epg_source=epg_source,
+                    export_lookback=one_hour_ago,
+                    export_cutoff=twenty_four_hours_later,
                 )
 
                 # Custom dummy should always return data (either from patterns or fallback)
@@ -1082,13 +1082,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 +1125,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 +1242,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(

apps/epg/migrations/0025_programdata_epg_id_index.py

@@ -0,0 +1,40 @@
+from django.contrib.postgres.operations import AddIndexConcurrently
+from django.db import migrations, models
+
+
+class AddIndexConcurrentlyIfPostgres(AddIndexConcurrently):
+    """Create the index CONCURRENTLY on PostgreSQL (no table lock on large
+    tables), falling back to a normal blocking AddIndex on other backends
+    such as the sqlite dev/test fallback."""
+
+    def database_forwards(self, app_label, schema_editor, from_state, to_state):
+        if schema_editor.connection.vendor == 'postgresql':
+            super().database_forwards(app_label, schema_editor, from_state, to_state)
+        else:
+            migrations.AddIndex.database_forwards(
+                self, app_label, schema_editor, from_state, to_state
+            )
+
+    def database_backwards(self, app_label, schema_editor, from_state, to_state):
+        if schema_editor.connection.vendor == 'postgresql':
+            super().database_backwards(app_label, schema_editor, from_state, to_state)
+        else:
+            migrations.AddIndex.database_backwards(
+                self, app_label, schema_editor, from_state, to_state
+            )
+
+
+class Migration(migrations.Migration):
+    # CREATE INDEX CONCURRENTLY cannot run inside a transaction.
+    atomic = False
+
+    dependencies = [
+        ('epg', '0024_remove_epgsource_api_key_epgsource_password_and_more'),
+    ]
+
+    operations = [
+        AddIndexConcurrentlyIfPostgres(
+            model_name='programdata',
+            index=models.Index(fields=['epg', 'id'], name='epg_prog_epg_id_idx'),
+        ),
+    ]

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',
+        ),
+    ]

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)
@@ -153,6 +178,11 @@ class ProgramData(models.Model):
     program_id = models.CharField(max_length=64, null=True, blank=True, help_text='Schedules Direct programID (e.g. EP123456789). Null for XMLTV sources.')
     custom_properties = models.JSONField(default=dict, blank=True, null=True)
 
+    class Meta:
+        indexes = [
+            models.Index(fields=['epg', 'id'], name='epg_prog_epg_id_idx'),
+        ]
+
     def __str__(self):
         return f"{self.title} ({self.start_time} - {self.end_time})"
 

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:

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": [],
             },

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",

apps/output/streaming_chunk_cache.py

@@ -0,0 +1,239 @@
+"""Single-flight Redis chunk cache for large streaming HTTP responses."""
+
+import logging
+import time
+
+from django.http import StreamingHttpResponse
+
+logger = logging.getLogger(__name__)
+
+STATUS_BUILDING = "building"
+STATUS_READY = "ready"
+STATUS_ERROR = "error"
+
+DEFAULT_CACHE_TTL = 300
+DEFAULT_LOCK_TTL = 120
+DEFAULT_POLL_INTERVAL = 0.05
+DEFAULT_MAX_FOLLOWER_WAIT = 600
+
+
+def _chunks_key(base_key):
+    return f"{base_key}:chunks"
+
+
+def _ready_key(base_key):
+    return f"{base_key}:ready"
+
+
+def _status_key(base_key):
+    return f"{base_key}:status"
+
+
+def _lock_key(base_key):
+    return f"{base_key}:lock"
+
+
+def _decode_chunk(chunk):
+    if chunk is None:
+        return None
+    if isinstance(chunk, bytes):
+        return chunk.decode("utf-8")
+    return chunk
+
+
+def _encode_chunk(chunk):
+    if isinstance(chunk, bytes):
+        return chunk
+    return chunk.encode("utf-8")
+
+
+def _poll_wait(interval):
+    try:
+        from core.utils import _is_gevent_monkey_patched
+
+        if _is_gevent_monkey_patched():
+            import gevent
+
+            gevent.sleep(interval)
+            return
+    except ImportError:
+        pass
+    time.sleep(interval)
+
+
+def _get_redis():
+    from django_redis import get_redis_connection
+
+    return get_redis_connection("default")
+
+
+def _get_status(redis, base_key):
+    raw = redis.get(_status_key(base_key))
+    if raw is None:
+        return None
+    return _decode_chunk(raw)
+
+
+def _clear_build_keys(redis, base_key):
+    redis.delete(
+        _chunks_key(base_key),
+        _status_key(base_key),
+        _ready_key(base_key),
+        _lock_key(base_key),
+    )
+
+
+def _try_acquire_lock(redis, base_key, lock_ttl):
+    return bool(redis.set(_lock_key(base_key), "1", nx=True, ex=lock_ttl))
+
+
+def _refresh_build_ttl(redis, base_key, lock_ttl):
+    redis.expire(_lock_key(base_key), lock_ttl)
+    redis.expire(_status_key(base_key), lock_ttl)
+    redis.expire(_chunks_key(base_key), lock_ttl)
+
+
+def _stream_ready(redis, base_key):
+    offset = 0
+    chunks_key = _chunks_key(base_key)
+    while True:
+        chunk = redis.lindex(chunks_key, offset)
+        if chunk is None:
+            break
+        yield _decode_chunk(chunk)
+        offset += 1
+
+
+def _stream_build(redis, base_key, source, cache_ttl, lock_ttl):
+    """Leader: stream to client and append each chunk to Redis."""
+    chunks_key = _chunks_key(base_key)
+    status_key = _status_key(base_key)
+    try:
+        from django.core.cache import cache as django_cache
+
+        django_cache.delete(base_key)  # clear any non-chunked entry under this key
+        redis.delete(chunks_key, _ready_key(base_key))
+        redis.set(status_key, STATUS_BUILDING, ex=lock_ttl)
+        refresh_interval = max(1, lock_ttl // 4)
+        last_refresh = 0.0
+        for chunk in source():
+            redis.rpush(chunks_key, _encode_chunk(chunk))
+            now = time.monotonic()
+            if now - last_refresh >= refresh_interval:
+                _refresh_build_ttl(redis, base_key, lock_ttl)
+                last_refresh = now
+            yield chunk
+        redis.set(status_key, STATUS_READY)
+        redis.set(_ready_key(base_key), "1")
+        redis.expire(chunks_key, cache_ttl)
+        redis.expire(status_key, cache_ttl)
+        redis.expire(_ready_key(base_key), cache_ttl)
+        logger.debug("Cached response in %s chunks", redis.llen(chunks_key))
+    except Exception:
+        logger.exception("Chunk cache build failed for %s", base_key)
+        redis.delete(chunks_key)
+        redis.set(status_key, STATUS_ERROR, ex=60)
+        raise
+    finally:
+        redis.delete(_lock_key(base_key))
+
+
+def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval, max_follower_wait):
+    """Follower: read chunks as the leader writes them."""
+    offset = 0
+    deadline = time.monotonic() + max_follower_wait
+    idle_polls = 0
+    chunks_key = _chunks_key(base_key)
+    lock_key = _lock_key(base_key)
+
+    while True:
+        chunk = redis.lindex(chunks_key, offset)
+        if chunk is not None:
+            idle_polls = 0
+            yield _decode_chunk(chunk)
+            offset += 1
+            continue
+
+        status = _get_status(redis, base_key)
+        if status == STATUS_READY:
+            break
+
+        if status == STATUS_ERROR:
+            _clear_build_keys(redis, base_key)
+            if offset == 0 and _try_acquire_lock(redis, base_key, lock_ttl):
+                yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl)
+                return
+            raise RuntimeError("Chunk cache build failed")
+
+        if time.monotonic() >= deadline:
+            if offset == 0 and _try_acquire_lock(redis, base_key, lock_ttl):
+                logger.warning("Chunk cache follower timed out; rebuilding %s", base_key)
+                yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl)
+                return
+            logger.warning("Chunk cache follower timed out after partial read for %s", base_key)
+            break
+
+        lock_active = bool(redis.exists(lock_key))
+        if status != STATUS_BUILDING and not lock_active:
+            idle_polls += 1
+            if offset == 0 and idle_polls >= max(1, int(1.0 / poll_interval)):
+                if _try_acquire_lock(redis, base_key, lock_ttl):
+                    logger.warning("Chunk cache leader lost; rebuilding %s", base_key)
+                    yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl)
+                    return
+        else:
+            idle_polls = 0
+
+        _poll_wait(poll_interval)
+
+
+def stream_cached_response(
+    cache_key,
+    source,
+    *,
+    content_type="application/xml",
+    filename=None,
+    cache_ttl=DEFAULT_CACHE_TTL,
+    lock_ttl=DEFAULT_LOCK_TTL,
+    poll_interval=DEFAULT_POLL_INTERVAL,
+    max_follower_wait=DEFAULT_MAX_FOLLOWER_WAIT,
+    redis=None,
+):
+    """
+    Stream a large response with single-flight Redis chunk caching.
+
+    ``source`` must be a callable returning a chunk iterator. Only the leader
+    invokes it; concurrent followers replay chunks already written to Redis, so
+    the expensive ``source`` runs at most once per ``cache_key``.
+    """
+    if redis is None:
+        redis = _get_redis()
+
+    if redis.get(_ready_key(cache_key)):
+        logger.debug("Serving response from chunk cache")
+        stream = _stream_ready(redis, cache_key)
+    else:
+        status = _get_status(redis, cache_key)
+        if status == STATUS_ERROR:
+            _clear_build_keys(redis, cache_key)
+
+        if _try_acquire_lock(redis, cache_key, lock_ttl):
+            logger.debug("Building response (cache leader)")
+            stream = _stream_build(redis, cache_key, source, cache_ttl, lock_ttl)
+        else:
+            logger.debug("Following in-flight cache build")
+            stream = _stream_follow(
+                redis,
+                cache_key,
+                source,
+                cache_ttl,
+                lock_ttl,
+                poll_interval,
+                max_follower_wait,
+            )
+
+    response = StreamingHttpResponse(stream, content_type=content_type)
+    if filename:
+        response["Content-Disposition"] = f'attachment; filename="{filename}"'
+    response["Cache-Control"] = "no-cache"
+    return response

apps/output/test_streaming_chunk_cache.py

@@ -0,0 +1,187 @@
+import threading
+import time
+from unittest import TestCase
+
+from apps.output.streaming_chunk_cache import (
+    STATUS_BUILDING,
+    STATUS_READY,
+    _chunks_key,
+    _lock_key,
+    _ready_key,
+    _status_key,
+    stream_cached_response,
+)
+
+
+class FakeRedis:
+    """Minimal Redis stand-in for chunk-cache unit tests."""
+
+    def __init__(self):
+        self._strings = {}
+        self._lists = {}
+        self._expires_at = {}
+
+    def _purge_expired(self):
+        now = time.monotonic()
+        expired = [key for key, deadline in self._expires_at.items() if deadline <= now]
+        for key in expired:
+            self._strings.pop(key, None)
+            self._lists.pop(key, None)
+            self._expires_at.pop(key, None)
+
+    def get(self, key):
+        self._purge_expired()
+        return self._strings.get(key)
+
+    def set(self, key, value, nx=False, ex=None):
+        self._purge_expired()
+        if nx and key in self._strings:
+            return None
+        self._strings[key] = value
+        if ex is not None:
+            self._expires_at[key] = time.monotonic() + ex
+        return True
+
+    def delete(self, *keys):
+        for key in keys:
+            self._strings.pop(key, None)
+            self._lists.pop(key, None)
+            self._expires_at.pop(key, None)
+
+    def exists(self, key):
+        self._purge_expired()
+        return key in self._strings or key in self._lists
+
+    def expire(self, key, ttl):
+        if key in self._strings or key in self._lists:
+            self._expires_at[key] = time.monotonic() + ttl
+        return True
+
+    def rpush(self, key, value):
+        self._lists.setdefault(key, []).append(value)
+
+    def lindex(self, key, offset):
+        items = self._lists.get(key, [])
+        if offset < len(items):
+            return items[offset]
+        return None
+
+    def llen(self, key):
+        return len(self._lists.get(key, []))
+
+
+def _consume(response):
+    return b"".join(response.streaming_content).decode("utf-8")
+
+
+class StreamingChunkCacheTests(TestCase):
+    def test_leader_caches_chunks_and_sets_ready(self):
+        redis = FakeRedis()
+        calls = []
+
+        def source():
+            calls.append(1)
+            yield "<tv>"
+            yield "</tv>"
+
+        body = _consume(stream_cached_response("cache:test", source, redis=redis))
+
+        self.assertEqual(body, "<tv></tv>")
+        self.assertEqual(calls, [1])
+        self.assertEqual(redis.get(_ready_key("cache:test")), "1")
+        self.assertEqual(redis.get(_status_key("cache:test")), STATUS_READY)
+        self.assertEqual(redis.llen(_chunks_key("cache:test")), 2)
+        self.assertFalse(redis.exists(_lock_key("cache:test")))
+
+    def test_cache_hit_skips_source(self):
+        redis = FakeRedis()
+        calls = []
+
+        def source():
+            calls.append(1)
+            yield "<tv>"
+            yield "</tv>"
+
+        _consume(stream_cached_response("cache:test", source, redis=redis))
+        calls.clear()
+        body = _consume(stream_cached_response("cache:test", source, redis=redis))
+
+        self.assertEqual(body, "<tv></tv>")
+        self.assertEqual(calls, [])
+
+    def test_follower_reads_leader_chunks_without_rebuilding(self):
+        redis = FakeRedis()
+        base = "cache:follow"
+        leader_started = threading.Event()
+        rebuild_calls = []
+
+        def slow_source():
+            rebuild_calls.append(1)
+            leader_started.set()
+            yield "a"
+            time.sleep(0.05)
+            yield "b"
+
+        def forbidden_source():
+            rebuild_calls.append(2)
+            yield "SHOULD_NOT_RUN"
+
+        def leader():
+            _consume(
+                stream_cached_response(
+                    base,
+                    slow_source,
+                    redis=redis,
+                    poll_interval=0.01,
+                )
+            )
+
+        leader_thread = threading.Thread(target=leader)
+        leader_thread.start()
+        leader_started.wait(timeout=5)
+        follower_body = _consume(
+            stream_cached_response(
+                base,
+                forbidden_source,
+                redis=redis,
+                poll_interval=0.01,
+            )
+        )
+        leader_thread.join(timeout=5)
+
+        self.assertEqual(follower_body, "ab")
+        self.assertEqual(rebuild_calls, [1])
+
+    def test_only_one_leader_when_two_clients_start_together(self):
+        redis = FakeRedis()
+        build_calls = []
+        barrier = threading.Barrier(2)
+        results = {}
+
+        def source():
+            build_calls.append(threading.current_thread().name)
+            yield "x"
+
+        def worker():
+            barrier.wait()
+            results[threading.current_thread().name] = _consume(
+                stream_cached_response(
+                    "cache:race",
+                    source,
+                    redis=redis,
+                    poll_interval=0.01,
+                )
+            )
+
+        threads = [
+            threading.Thread(target=worker, name="t1"),
+            threading.Thread(target=worker, name="t2"),
+        ]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(timeout=10)
+
+        self.assertEqual(results["t1"], "x")
+        self.assertEqual(results["t2"], "x")
+        self.assertEqual(len(build_calls), 1)

apps/output/tests.py

@@ -1,31 +1,117 @@
-from django.test import TestCase, Client
+from django.test import TestCase, Client, SimpleTestCase
 from django.urls import reverse
-from apps.channels.models import Channel, ChannelGroup
+from unittest.mock import patch
+from uuid import uuid4
+from apps.channels.models import Channel, ChannelGroup, ChannelProfile, ChannelProfileMembership
 from apps.epg.models import EPGData, EPGSource
 import xml.etree.ElementTree as ET
+from datetime import timedelta
+
+
+def _response_text(response):
+    """Read body from HttpResponse or StreamingHttpResponse."""
+    if getattr(response, "streaming", False):
+        return b"".join(response.streaming_content).decode()
+    return response.content.decode()
+
+
+def _epg_response_without_redis(cache_key, source, **kwargs):
+    """Test helper: stream EPG directly without Redis chunk caching."""
+    from django.http import StreamingHttpResponse
+
+    response = StreamingHttpResponse(source(), content_type="application/xml")
+    response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"'
+    response["Cache-Control"] = "no-cache"
+    return response
+
+
+class OutputEndpointTestMixin:
+    """Isolate HTTP endpoint tests from network ACL, logging, DB teardown, and Redis."""
 
-class OutputM3UTest(TestCase):
     def setUp(self):
+        super().setUp()
+        self._network_patch = patch(
+            "apps.output.views.network_access_allowed",
+            return_value=True,
+        )
+        self._epg_teardown_patch = patch("apps.output.epg._epg_export_teardown")
+        self._log_event_patch = patch("apps.output.views.log_system_event")
+        self._epg_log_event_patch = patch("apps.output.epg.log_system_event")
+        self._close_db_patch = patch("django.db.close_old_connections")
+        self._epg_cache_patch = patch(
+            "apps.output.epg.stream_cached_response",
+            side_effect=_epg_response_without_redis,
+        )
+        self._network_patch.start()
+        self._epg_teardown_patch.start()
+        self._log_event_patch.start()
+        self._epg_log_event_patch.start()
+        self._close_db_patch.start()
+        self._epg_cache_patch.start()
+
+    def tearDown(self):
+        from django.core.cache import cache
+
+        cache.clear()
+        self._epg_cache_patch.stop()
+        self._close_db_patch.stop()
+        self._epg_log_event_patch.stop()
+        self._log_event_patch.stop()
+        self._epg_teardown_patch.stop()
+        self._network_patch.stop()
+        super().tearDown()
+
+    def _create_isolated_profile(self, prefix):
+        """New profiles auto-include every channel via signal; clear that for tests."""
+        profile = ChannelProfile.objects.create(name=f"{prefix}-{uuid4().hex[:8]}")
+        ChannelProfileMembership.objects.filter(channel_profile=profile).delete()
+        return profile
+
+    def _add_channel_to_profile(self, profile, group, **kwargs):
+        channel = Channel.objects.create(channel_group=group, **kwargs)
+        ChannelProfileMembership.objects.create(
+            channel_profile=profile,
+            channel=channel,
+            enabled=True,
+        )
+        return channel
+
+
+class OutputM3UTest(OutputEndpointTestMixin, TestCase):
+    def setUp(self):
+        super().setUp()
         self.client = Client()
-    
+        self.group = ChannelGroup.objects.create(name=f"M3U Group {uuid4().hex[:8]}")
+        self.profile = self._create_isolated_profile("m3u")
+        self._add_channel_to_profile(
+            self.profile,
+            self.group,
+            channel_number=1.0,
+            name="Test M3U Channel",
+        )
+
+    def _m3u_url(self):
+        return reverse("output:m3u_endpoint", kwargs={"profile_name": self.profile.name})
+
     def test_generate_m3u_response(self):
         """
         Test that the M3U endpoint returns a valid M3U file.
         """
-        url = reverse('output:generate_m3u')
-        response = self.client.get(url)
+        response = self.client.get(self._m3u_url())
         self.assertEqual(response.status_code, 200)
-        content = response.content.decode()
+        content = _response_text(response)
         self.assertIn("#EXTM3U", content)
 
     def test_generate_m3u_response_post_empty_body(self):
         """
         Test that a POST request with an empty body returns 200 OK.
         """
-        url = reverse('output:generate_m3u')
-
-        response = self.client.post(url, data=None, content_type='application/x-www-form-urlencoded')
-        content = response.content.decode()
+        response = self.client.post(
+            self._m3u_url(),
+            data=None,
+            content_type="application/x-www-form-urlencoded",
+        )
+        content = _response_text(response)
 
         self.assertEqual(response.status_code, 200, "POST with empty body should return 200 OK")
         self.assertIn("#EXTM3U", content)
@@ -34,35 +120,40 @@ class OutputM3UTest(TestCase):
         """
         Test that a POST request with a non-empty body returns 403 Forbidden.
         """
-        url = reverse('output:generate_m3u')
-
-        response = self.client.post(url, data={'evilstring': 'muhahaha'})
+        response = self.client.post(self._m3u_url(), data={"evilstring": "muhahaha"})
 
         self.assertEqual(response.status_code, 403, "POST with body should return 403 Forbidden")
-        self.assertIn("POST requests with body are not allowed, body is:", response.content.decode())
+        self.assertIn("POST requests with body are not allowed", _response_text(response))
 
 
-class OutputEPGXMLEscapingTest(TestCase):
+class OutputEPGXMLEscapingTest(OutputEndpointTestMixin, TestCase):
     """Test XML escaping of channel_id attributes in EPG generation"""
 
     def setUp(self):
+        super().setUp()
         self.client = Client()
-        self.group = ChannelGroup.objects.create(name="Test Group")
+        self.group = ChannelGroup.objects.create(name=f"Test Group {uuid4().hex[:8]}")
+        self.profile = self._create_isolated_profile("epg-xml")
+
+    def _add_channel(self, **kwargs):
+        return self._add_channel_to_profile(self.profile, self.group, **kwargs)
+
+    def _epg_url(self, query="tvg_id_source=tvg_id&days=0&prev_days=0"):
+        base = reverse("output:epg_endpoint", kwargs={"profile_name": self.profile.name})
+        return f"{base}?{query}"
 
     def test_channel_id_with_ampersand(self):
         """Test channel ID with ampersand is properly escaped"""
-        channel = Channel.objects.create(
+        self._add_channel(
             channel_number=1.0,
             name="Test Channel",
             tvg_id="News & Sports",
-            channel_group=self.group
         )
 
-        url = reverse('output:generate_epg') + '?tvg_id_source=tvg_id'
-        response = self.client.get(url)
+        response = self.client.get(self._epg_url())
 
         self.assertEqual(response.status_code, 200)
-        content = response.content.decode()
+        content = _response_text(response)
 
         # Should contain escaped ampersand
         self.assertIn('id="News & Sports"', content)
@@ -76,17 +167,15 @@ class OutputEPGXMLEscapingTest(TestCase):
 
     def test_channel_id_with_angle_brackets(self):
         """Test channel ID with < and > characters"""
-        channel = Channel.objects.create(
+        self._add_channel(
             channel_number=2.0,
             name="HD Channel",
             tvg_id="Channel <HD>",
-            channel_group=self.group
         )
 
-        url = reverse('output:generate_epg') + '?tvg_id_source=tvg_id'
-        response = self.client.get(url)
+        response = self.client.get(self._epg_url())
 
-        content = response.content.decode()
+        content = _response_text(response)
         self.assertIn('id="Channel &lt;HD&gt;"', content)
 
         try:
@@ -96,23 +185,28 @@ class OutputEPGXMLEscapingTest(TestCase):
 
     def test_channel_id_with_all_special_chars(self):
         """Test channel ID with all XML special characters"""
-        channel = Channel.objects.create(
+        expected_id = 'Test & "Special" <Chars>'
+        self._add_channel(
             channel_number=3.0,
             name="Complex Channel",
-            tvg_id='Test & "Special" <Chars>',
-            channel_group=self.group
+            tvg_id=expected_id,
         )
 
-        url = reverse('output:generate_epg') + '?tvg_id_source=tvg_id'
-        response = self.client.get(url)
+        response = self.client.get(self._epg_url())
 
-        content = response.content.decode()
+        content = _response_text(response)
         self.assertIn('id="Test &amp; &quot;Special&quot; &lt;Chars&gt;"', content)
 
         try:
             tree = ET.fromstring(content)
-            # Verify we can find the channel with correct ID in parsed tree
-            channel_elem = tree.find('.//channel[@id="Test & \\"Special\\" <Chars>"]')
+            channel_elem = next(
+                (
+                    elem
+                    for elem in tree.findall(".//channel")
+                    if elem.get("id") == expected_id
+                ),
+                None,
+            )
             self.assertIsNotNone(channel_elem)
         except ET.ParseError as e:
             self.fail(f"Generated EPG with all special chars is not valid XML: {e}")
@@ -121,25 +215,203 @@ class OutputEPGXMLEscapingTest(TestCase):
         """Test that programme elements also have escaped channel attributes"""
         epg_source = EPGSource.objects.create(name="Test EPG", source_type="dummy")
         epg_data = EPGData.objects.create(name="Test EPG Data", epg_source=epg_source)
-        channel = Channel.objects.create(
+        self._add_channel(
             channel_number=4.0,
             name="Program Test",
             tvg_id="News & Sports",
             epg_data=epg_data,
-            channel_group=self.group
         )
 
-        url = reverse('output:generate_epg') + '?tvg_id_source=tvg_id'
-        response = self.client.get(url)
+        response = self.client.get(self._epg_url())
 
-        content = response.content.decode()
+        content = _response_text(response)
 
         # Check programme elements have escaped channel attributes
         self.assertIn('channel="News &amp; Sports"', content)
 
         try:
             tree = ET.fromstring(content)
-            programmes = tree.findall('.//programme[@channel="News & Sports"]')
+            programmes = [
+                programme
+                for programme in tree.findall(".//programme")
+                if programme.get("channel") == "News & Sports"
+            ]
             self.assertGreater(len(programmes), 0)
         except ET.ParseError as e:
             self.fail(f"Generated EPG with programme elements is not valid XML: {e}")
+
+    def test_programmes_emitted_in_start_time_order(self):
+        """Programmes for a channel are emitted in start_time order, not insert order."""
+        from django.utils import timezone
+        from apps.epg.models import ProgramData
+
+        epg_source = EPGSource.objects.create(name="Real EPG", source_type="xmltv")
+        epg_data = EPGData.objects.create(name="Station", epg_source=epg_source, tvg_id="station1")
+        self._add_channel(
+            channel_number=149.0,
+            name="Food Network",
+            tvg_id="station1",
+            epg_data=epg_data,
+        )
+        now = timezone.now()
+        # Insert out of chronological order so id order != start_time order.
+        ProgramData.objects.create(
+            epg=epg_data,
+            start_time=now + timedelta(days=3),
+            end_time=now + timedelta(days=3, hours=1),
+            title="Third",
+            tvg_id="station1",
+        )
+        ProgramData.objects.create(
+            epg=epg_data,
+            start_time=now + timedelta(days=1),
+            end_time=now + timedelta(days=1, hours=1),
+            title="First",
+            tvg_id="station1",
+        )
+        ProgramData.objects.create(
+            epg=epg_data,
+            start_time=now + timedelta(days=2),
+            end_time=now + timedelta(days=2, hours=1),
+            title="Second",
+            tvg_id="station1",
+        )
+
+        content = _response_text(self.client.get(self._epg_url("tvg_id_source=tvg_id&days=7")))
+
+        self.assertLess(content.find('<title>First</title>'), content.find('<title>Second</title>'))
+        self.assertLess(content.find('<title>Second</title>'), content.find('<title>Third</title>'))
+
+
+class OutputEPGCustomDummyTest(TestCase):
+    """Custom dummy EPG must not fall back to default when pattern matched but event is outside window."""
+
+    def setUp(self):
+        self.group = ChannelGroup.objects.create(name="Sports Group")
+
+    def test_custom_dummy_outside_window_fills_with_ended_programmes(self):
+        from django.utils import timezone
+        from apps.output.views import generate_dummy_programs
+
+        epg_source = EPGSource.objects.create(
+            name="NHL Dummy",
+            source_type="dummy",
+            custom_properties={
+                "title_pattern": r"(?<league>.*)\s\d+:\s(?<team1>.*?)(?:\s+vs\s+)(?<team2>.*?)\s*@.*",
+                "time_pattern": r"(?<hour>\d{1,2}):(?<minute>\d{2})\s*(?<ampm>AM|PM)",
+                "date_pattern": r"@ (?<month>[A-Za-z]+)\s+(?<day>\d{1,2})",
+                "timezone": "US/Eastern",
+                "program_duration": 180,
+            },
+        )
+        channel_name = (
+            "NHL 01: Washington Capitals vs Philadelphia Flyers @ April 16 07:30 PM ET"
+        )
+        now = timezone.now()
+        lookback = now - timedelta(days=7)
+
+        programs = generate_dummy_programs(
+            channel_id="nhl01",
+            channel_name=channel_name,
+            num_days=7,
+            epg_source=epg_source,
+            export_lookback=lookback,
+            export_cutoff=now + timedelta(days=7),
+        )
+
+        self.assertGreater(len(programs), 0)
+        self.assertTrue(
+            all(p['end_time'] >= lookback for p in programs),
+            "All programmes should fall inside the export window",
+        )
+        self.assertTrue(
+            any('Ended' in p['description'] for p in programs),
+            "Past events outside the window should still show ended filler",
+        )
+        for program in programs:
+            start = program['start_time']
+            self.assertEqual(start.second, 0)
+            self.assertEqual(start.microsecond, 0)
+            self.assertIn(
+                start.minute, (0, 30),
+                "Filler programmes should start on half-hour boundaries",
+            )
+        self.assertGreaterEqual(programs[0]['start_time'], lookback)
+
+    def test_custom_dummy_future_event_fills_grid_window_with_upcoming(self):
+        """Grid-style window: future event should show upcoming filler, not empty."""
+        from django.utils import timezone
+        from apps.output.epg import _programme_overlaps_export_window, generate_dummy_programs
+
+        epg_source = EPGSource.objects.create(
+            name="NHL Dummy Future",
+            source_type="dummy",
+            custom_properties={
+                "title_pattern": r"(?<league>.*)\s\d+:\s(?<team1>.*?)(?:\s+vs\s+)(?<team2>.*?)\s*@.*",
+                "time_pattern": r"(?<hour>\d{1,2}):(?<minute>\d{2})\s*(?<ampm>AM|PM)",
+                "date_pattern": r"@ (?<month>[A-Za-z]+)\s+(?<day>\d{1,2})",
+                "timezone": "US/Eastern",
+                "program_duration": 180,
+            },
+        )
+        now = timezone.now()
+        grid_start = now - timedelta(hours=1)
+        grid_end = now + timedelta(hours=24)
+        future = now + timedelta(days=3)
+        channel_name = (
+            f"NHL 01: Washington Capitals vs Philadelphia Flyers @ "
+            f"{future.strftime('%B')} {future.day} 07:30 PM ET"
+        )
+
+        programs = generate_dummy_programs(
+            channel_id="nhl01",
+            channel_name=channel_name,
+            num_days=1,
+            epg_source=epg_source,
+            export_lookback=grid_start,
+            export_cutoff=grid_end,
+        )
+
+        self.assertGreater(len(programs), 0)
+        self.assertTrue(
+            all(
+                _programme_overlaps_export_window(
+                    p["start_time"], p["end_time"], grid_start, grid_end
+                )
+                for p in programs
+            ),
+            "All programmes should overlap the grid query window",
+        )
+        self.assertTrue(
+            any("Upcoming" in p.get("description", "") for p in programs),
+            "Future events outside the window should show upcoming filler",
+        )
+
+
+class OutputEPGHelperTest(SimpleTestCase):
+    def test_ceil_to_half_hour_on_boundary(self):
+        from django.utils import timezone
+        from apps.output.epg import _ceil_to_half_hour
+
+        dt = timezone.now().replace(minute=30, second=0, microsecond=0)
+        self.assertEqual(_ceil_to_half_hour(dt), dt)
+
+    def test_ceil_to_half_hour_rounds_up(self):
+        from django.utils import timezone
+        from apps.output.epg import _ceil_to_half_hour
+
+        dt = timezone.now().replace(minute=17, second=42, 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))
+
+    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))

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:

core/tests.py

@@ -1,8 +1,8 @@
 from unittest.mock import patch, MagicMock
 
-from django.test import TestCase
+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:
@@ -301,3 +297,14 @@ class DropDBCommandTlsTest(TestCase):
             host='localhost', port=5432,
             autocommit=True,
         )
+
+
+class MallocTrimTests(SimpleTestCase):
+    def test_trim_is_noop_when_libc_has_no_malloc_trim(self):
+        from core.utils import trim_c_allocator_heap
+
+        fake_libc = MagicMock(spec=[])
+        with patch('ctypes.util.find_library', return_value='libc.so.6'), patch(
+            'ctypes.CDLL', return_value=fake_libc
+        ):
+            self.assertFalse(trim_c_allocator_heap())

core/utils.py

@@ -546,6 +546,25 @@ def monitor_memory_usage(func):
         return result
     return wrapper
 
+def trim_c_allocator_heap():
+    """Return unused C heap pages to the OS where supported (glibc malloc_trim)."""
+    try:
+        import ctypes
+        import ctypes.util
+
+        libc_name = ctypes.util.find_library("c")
+        if not libc_name:
+            return False
+        libc = ctypes.CDLL(libc_name)
+        if not hasattr(libc, "malloc_trim"):
+            return False
+        libc.malloc_trim(0)
+        return True
+    except Exception:
+        logger.debug("malloc_trim unavailable or failed", exc_info=True)
+        return False
+
+
 def cleanup_memory(log_usage=False, force_collection=True):
     """
     Comprehensive memory cleanup function to reduce memory footprint
@@ -589,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.