enhancement(epg): optimize EPG export performance and database interactions

- Streamlined `generate_epg()` to incrementally stream EPG data without loading the entire guide, improving memory efficiency.
- Reduced database load by deferring the fetching of large `programme_index` blobs, enhancing response times for EPG generation.
- Introduced a composite index on `(epg_id, id)` in `ProgramData` to optimize query performance during EPG exports.
- Updated tests to ensure proper functionality and performance of new features.

CHANGELOG.md

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

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/models.py

@@ -153,6 +153,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/output/epg_chunk_cache.py

@@ -0,0 +1,230 @@
+"""Single-flight Redis chunk cache for XMLTV EPG streaming 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)  # legacy monolithic entry
+        redis.delete(chunks_key, _ready_key(base_key))
+        redis.set(status_key, STATUS_BUILDING, ex=lock_ttl)
+        for chunk in source():
+            redis.rpush(chunks_key, _encode_chunk(chunk))
+            _refresh_build_ttl(redis, base_key, lock_ttl)
+            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 EPG in %s chunks", redis.llen(chunks_key))
+    except Exception:
+        logger.exception("EPG 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("EPG cache build failed")
+
+        if time.monotonic() >= deadline:
+            if offset == 0 and _try_acquire_lock(redis, base_key, lock_ttl):
+                logger.warning("EPG cache follower timed out; rebuilding %s", base_key)
+                yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl)
+                return
+            logger.warning("EPG 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("EPG 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_epg_response(
+    cache_key,
+    source,
+    *,
+    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 XMLTV EPG output with single-flight Redis chunk caching.
+
+    ``source`` must be a callable returning a chunk iterator. Only the leader
+    invokes it; followers read chunks already written to Redis.
+    """
+    if redis is None:
+        redis = _get_redis()
+
+    if redis.get(_ready_key(cache_key)):
+        logger.debug("Serving EPG 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 EPG (cache leader)")
+            stream = _stream_build(redis, cache_key, source, cache_ttl, lock_ttl)
+        else:
+            logger.debug("Following in-flight EPG build")
+            stream = _stream_follow(
+                redis,
+                cache_key,
+                source,
+                cache_ttl,
+                lock_ttl,
+                poll_interval,
+                max_follower_wait,
+            )
+
+    response = StreamingHttpResponse(stream, content_type="application/xml")
+    response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"'
+    response["Cache-Control"] = "no-cache"
+    return response

apps/output/test_epg_chunk_cache.py

@@ -0,0 +1,187 @@
+import threading
+import time
+from unittest import TestCase
+
+from apps.output.epg_chunk_cache import (
+    STATUS_BUILDING,
+    STATUS_READY,
+    _chunks_key,
+    _lock_key,
+    _ready_key,
+    _status_key,
+    stream_epg_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 EPGChunkCacheTests(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_epg_response("epg:test", source, redis=redis))
+
+        self.assertEqual(body, "<tv></tv>")
+        self.assertEqual(calls, [1])
+        self.assertEqual(redis.get(_ready_key("epg:test")), "1")
+        self.assertEqual(redis.get(_status_key("epg:test")), STATUS_READY)
+        self.assertEqual(redis.llen(_chunks_key("epg:test")), 2)
+        self.assertFalse(redis.exists(_lock_key("epg:test")))
+
+    def test_cache_hit_skips_source(self):
+        redis = FakeRedis()
+        calls = []
+
+        def source():
+            calls.append(1)
+            yield "<tv>"
+            yield "</tv>"
+
+        _consume(stream_epg_response("epg:test", source, redis=redis))
+        calls.clear()
+        body = _consume(stream_epg_response("epg:test", source, redis=redis))
+
+        self.assertEqual(body, "<tv></tv>")
+        self.assertEqual(calls, [])
+
+    def test_follower_reads_leader_chunks_without_rebuilding(self):
+        redis = FakeRedis()
+        base = "epg: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_epg_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_epg_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_epg_response(
+                    "epg: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,114 @@
-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.views._epg_export_teardown")
+        self._log_event_patch = patch("apps.output.views.log_system_event")
+        self._close_db_patch = patch("django.db.close_old_connections")
+        self._epg_cache_patch = patch(
+            "apps.output.views.stream_epg_response",
+            side_effect=_epg_response_without_redis,
+        )
+        self._network_patch.start()
+        self._epg_teardown_patch.start()
+        self._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._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 +117,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 +164,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 +182,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 +212,144 @@ 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)
+
+
+class OutputEPGHelperTest(SimpleTestCase):
+    def test_ceil_to_half_hour_on_boundary(self):
+        from django.utils import timezone
+        from apps.output.views 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.views 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.assertGreater(aligned, dt.replace(second=0, microsecond=0))

apps/output/views.py

@@ -25,9 +25,55 @@ from apps.proxy.utils import get_user_active_connections
 import regex
 from core.utils import log_system_event, build_absolute_uri_with_port
 import hashlib
+from apps.output.epg_chunk_cache import stream_epg_response
 
 logger = logging.getLogger(__name__)
 
+_EPG_CHANNEL_XML_BATCH_SIZE = 200
+_EPG_PROGRAM_YIELD_BATCH_SIZE = 1000
+_EPG_PROGRAM_DB_CHUNK_SIZE = 20000
+
+
+def _programme_overlaps_export_window(start_time, end_time, lookback_cutoff, cutoff_date):
+    if end_time < lookback_cutoff:
+        return False
+    if cutoff_date is not None and start_time >= cutoff_date:
+        return False
+    return True
+
+
+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)
+
+
+def _epg_export_teardown():
+    from django.db import close_old_connections
+
+    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()
+
+
 def get_client_identifier(request):
     """Get client information including IP, user agent, and a unique hash identifier
 
@@ -112,6 +158,10 @@ def generate_m3u(request, profile_name=None, user=None):
     # Check if this is a POST request and the body is not empty (which we don't want to allow)
     logger.debug("Generating M3U for profile: %s, user: %s, method: %s", profile_name, user.username if user else "Anonymous", request.method)
 
+    if request.method == "POST" and request.body:
+        if request.body.decode() != '{}':
+            return HttpResponseForbidden("POST requests with body are not allowed.")
+
     # Check cache for recent identical request (helps with double-GET from browsers)
     from django.core.cache import cache
     cache_params = f"{profile_name or 'all'}:{user.username if user else 'anonymous'}:{request.GET.urlencode()}"
@@ -123,10 +173,6 @@ def generate_m3u(request, profile_name=None, user=None):
         response = HttpResponse(cached_content, content_type="audio/x-mpegurl")
         response["Content-Disposition"] = 'attachment; filename="channels.m3u"'
         return response
-    # Check if this is a POST request with data (which we don't want to allow)
-    if request.method == "POST" and request.body:
-        if request.body.decode() != '{}':
-            return HttpResponseForbidden("POST requests with body are not allowed.")
 
     if user is not None:
         if user.user_level < 10:
@@ -409,7 +455,15 @@ def generate_fallback_programs(channel_id, channel_name, now, num_days, program_
     return programs
 
 
-def generate_dummy_programs(channel_id, channel_name, num_days=1, program_length_hours=4, epg_source=None):
+def generate_dummy_programs(
+    channel_id,
+    channel_name,
+    num_days=1,
+    program_length_hours=4,
+    epg_source=None,
+    export_lookback=None,
+    export_cutoff=None,
+):
     """
     Generate dummy EPG programs for channels.
 
@@ -435,29 +489,26 @@ def generate_dummy_programs(channel_id, channel_name, num_days=1, program_length
     if epg_source and epg_source.source_type == 'dummy' and epg_source.custom_properties:
         custom_programs = generate_custom_dummy_programs(
             channel_id, channel_name, now, num_days,
-            epg_source.custom_properties
+            epg_source.custom_properties,
+            export_lookback=export_lookback,
+            export_cutoff=export_cutoff,
         )
-        # If custom generation succeeded, return those programs
-        # If it returned empty (pattern didn't match), check for custom fallback templates
-        if custom_programs:
+        if custom_programs is not None:
             return custom_programs
-        else:
-            logger.info(f"Custom pattern didn't match for '{channel_name}', checking for custom fallback templates")
 
-            # Check if custom fallback templates are provided
-            custom_props = epg_source.custom_properties
-            fallback_title = custom_props.get('fallback_title_template', '').strip()
-            fallback_description = custom_props.get('fallback_description_template', '').strip()
+        logger.info(f"Custom pattern didn't match for '{channel_name}', checking for custom fallback templates")
 
-            # If custom fallback templates exist, use them instead of default
-            if fallback_title or fallback_description:
-                logger.info(f"Using custom fallback templates for '{channel_name}'")
-                return generate_fallback_programs(
-                    channel_id, channel_name, now, num_days,
-                    program_length_hours, fallback_title, fallback_description
-                )
-            else:
-                logger.info(f"No custom fallback templates found, using default dummy EPG")
+        custom_props = epg_source.custom_properties
+        fallback_title = custom_props.get('fallback_title_template', '').strip()
+        fallback_description = custom_props.get('fallback_description_template', '').strip()
+
+        if fallback_title or fallback_description:
+            logger.info(f"Using custom fallback templates for '{channel_name}'")
+            return generate_fallback_programs(
+                channel_id, channel_name, now, num_days,
+                program_length_hours, fallback_title, fallback_description
+            )
+        logger.info(f"No custom fallback templates found, using default dummy EPG")
 
     # Default humorous program descriptions based on time of day
     time_descriptions = {
@@ -531,7 +582,15 @@ def generate_dummy_programs(channel_id, channel_name, num_days=1, program_length
     return programs
 
 
-def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, custom_properties):
+def generate_custom_dummy_programs(
+    channel_id,
+    channel_name,
+    now,
+    num_days,
+    custom_properties,
+    export_lookback=None,
+    export_cutoff=None,
+):
     """
     Generate programs using custom dummy EPG regex patterns.
 
@@ -616,7 +675,7 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust
 
     if not title_pattern:
         logger.warning(f"No title_pattern in custom_properties, falling back to default")
-        return []  # Return empty, will use default
+        return None
 
     logger.debug(f"Title pattern from DB: {repr(title_pattern)}")
 
@@ -633,7 +692,7 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust
     except Exception as e:
         logger.error(f"Invalid title regex pattern after conversion: {e}")
         logger.error(f"Pattern was: {repr(title_pattern)}")
-        return []
+        return None
 
     time_regex = None
     if time_pattern:
@@ -665,7 +724,7 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust
     title_match = title_regex.search(channel_name)
     if not title_match:
         logger.debug(f"Channel name '{channel_name}' doesn't match title pattern")
-        return []  # Return empty, will use default
+        return None
 
     groups = title_match.groupdict()
     logger.debug(f"Title pattern matched. Groups: {groups}")
@@ -917,57 +976,93 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust
         iterations = num_days
 
     for day in range(iterations):
-        # Start from current time (like standard dummy) instead of midnight
-        # This ensures programs appear in the guide's current viewing window
-        day_start = now + timedelta(days=day)
-        day_end = day_start + timedelta(days=1)
-
-        if time_info:
-            # We have an extracted event time - this is when the MAIN event starts
-            # The extracted time is in the SOURCE timezone (e.g., 8PM ET)
-            # We need to convert it to UTC for storage
-
-            # Determine which date to use
-            if date_info:
-                # Use the extracted date from the channel title
-                current_date = datetime(
-                    date_info['year'],
-                    date_info['month'],
-                    date_info['day']
-                ).date()
-                logger.debug(f"Using extracted date: {current_date}")
-            else:
-                # No date extracted, use day offset from current time in SOURCE timezone
-                # This ensures we calculate "today" in the event's timezone, not UTC
-                # For example: 8:30 PM Central (1:30 AM UTC next day) for a 10 PM ET event
-                # should use today's date in ET, not tomorrow's date in UTC
-                now_in_source_tz = now.astimezone(source_tz)
-                current_date = (now_in_source_tz + timedelta(days=day)).date()
-                logger.debug(f"No date extracted, using day offset in {source_tz}: {current_date}")
-
-            # Create a naive datetime (no timezone info) representing the event in source timezone
+        event_overlaps_window = True
+        if date_info and time_info:
+            current_date = datetime(
+                date_info['year'],
+                date_info['month'],
+                date_info['day'],
+            ).date()
             event_start_naive = datetime.combine(
                 current_date,
                 datetime.min.time().replace(
                     hour=time_info['hour'],
-                    minute=time_info['minute']
-                )
+                    minute=time_info['minute'],
+                ),
             )
-
-            # Use pytz to localize the naive datetime to the source timezone
-            # This automatically handles DST!
             try:
-                event_start_local = source_tz.localize(event_start_naive)
-                # Convert to UTC
-                event_start_utc = event_start_local.astimezone(pytz.utc)
-                logger.debug(f"Converted {event_start_local} to UTC: {event_start_utc}")
+                event_start_utc = source_tz.localize(event_start_naive).astimezone(pytz.utc)
             except Exception as e:
                 logger.error(f"Error localizing time to {source_tz}: {e}")
-                # Fallback: treat as UTC
                 event_start_utc = django_timezone.make_aware(event_start_naive, pytz.utc)
-
             event_end_utc = event_start_utc + timedelta(minutes=program_duration)
 
+            lookback = export_lookback if export_lookback is not None else now
+            event_overlaps_window = _programme_overlaps_export_window(
+                event_start_utc, event_end_utc, lookback, export_cutoff
+            )
+            if not event_overlaps_window:
+                logger.debug(
+                    "Custom dummy event outside export window; filling window only: %s",
+                    channel_name,
+                )
+                event_happened = event_end_utc < lookback
+                day_start = _ceil_to_half_hour(lookback)
+                if export_cutoff is not None:
+                    day_end = export_cutoff
+                else:
+                    day_end = now + timedelta(days=num_days if num_days > 0 else 3)
+            else:
+                day_start = source_tz.localize(
+                    datetime.combine(current_date, datetime.min.time())
+                ).astimezone(pytz.utc)
+                day_end = day_start + timedelta(days=1)
+                if export_lookback is not None:
+                    day_start = max(day_start, export_lookback)
+                if export_cutoff is not None:
+                    day_end = min(day_end, export_cutoff)
+        else:
+            day_start = now + timedelta(days=day)
+            day_end = day_start + timedelta(days=1)
+            if export_lookback is not None:
+                day_start = max(day_start, export_lookback)
+            if export_cutoff is not None:
+                day_end = min(day_end, export_cutoff)
+
+        if day_start >= day_end:
+            continue
+
+        if time_info:
+            if not date_info:
+                now_in_source_tz = now.astimezone(source_tz)
+                current_date = (now_in_source_tz + timedelta(days=day)).date()
+                logger.debug(f"No date extracted, using day offset in {source_tz}: {current_date}")
+
+                event_start_naive = datetime.combine(
+                    current_date,
+                    datetime.min.time().replace(
+                        hour=time_info['hour'],
+                        minute=time_info['minute'],
+                    ),
+                )
+                try:
+                    event_start_local = source_tz.localize(event_start_naive)
+                    event_start_utc = event_start_local.astimezone(pytz.utc)
+                    logger.debug(f"Converted {event_start_local} to UTC: {event_start_utc}")
+                except Exception as e:
+                    logger.error(f"Error localizing time to {source_tz}: {e}")
+                    event_start_utc = django_timezone.make_aware(event_start_naive, pytz.utc)
+
+                event_end_utc = event_start_utc + timedelta(minutes=program_duration)
+
+                lookback = export_lookback if export_lookback is not None else now
+                if not _programme_overlaps_export_window(
+                    event_start_utc, event_end_utc, lookback, export_cutoff
+                ):
+                    continue
+            else:
+                logger.debug(f"Using extracted date: {current_date}")
+
             # Pre-generate the main event title and description for reuse
             if title_template:
                 main_event_title = format_template(title_template, all_groups)
@@ -994,15 +1089,13 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust
 
 
             # Determine if this day is before, during, or after the event
-            # Event only happens on day 0 (first day)
-            is_event_day = (day == 0)
+            # Event only happens on day 0 (first day) when it falls inside the window
+            is_event_day = (day == 0) and event_overlaps_window
 
             if is_event_day and not event_happened:
-                # This is THE day the event happens
-                # Fill programs BEFORE the event
                 current_time = day_start
 
-                while current_time < event_start_utc:
+                while current_time < event_start_utc and current_time < day_end:
                     program_start_utc = current_time
                     program_end_utc = min(current_time + timedelta(minutes=program_duration), event_start_utc)
 
@@ -1084,8 +1177,8 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust
 
                 event_happened = True
 
-                # Fill programs AFTER the event until end of day
-                current_time = event_end_utc
+                # Fill programs AFTER the event until end of export day window
+                current_time = max(event_end_utc, day_start)
 
                 while current_time < day_end:
                     program_start_utc = current_time
@@ -1325,18 +1418,10 @@ def generate_dummy_epg(
 
 def generate_epg(request, profile_name=None, user=None):
     """
-    Dynamically generate an XMLTV (EPG) file using streaming response to handle keep-alives.
+    Dynamically generate an XMLTV (EPG) file using a streaming response.
     Since the EPG data is stored independently of Channels, we group programmes
     by their associated EPGData record.
-    This version filters data based on the 'days' parameter and sends keep-alives during processing.
     """
-    # Check cache for recent identical request (helps with double-GET from browsers)
-    from django.core.cache import cache
-    # Resolve all effective parameter values once here so they are reused for both
-    # the cache key and inside epg_generator() via closure.
-    # The cache key is built from resolved values only — not from the raw query string —
-    # so equivalent requests (e.g. days=7 via URL param vs. user default of 7) share
-    # the same cache entry regardless of how the value was supplied.
     user_custom = (user.custom_properties or {}) if user else {}
     try:
         num_days = int(request.GET.get('days', user_custom.get('epg_days', 0)))
@@ -1356,21 +1441,13 @@ def generate_epg(request, profile_name=None, user=None):
     )
     content_cache_key = f"epg_content:{cache_params}"
 
-    cached_content = cache.get(content_cache_key)
-    if cached_content:
-        logger.debug("Serving EPG from cache")
-        response = HttpResponse(cached_content, content_type="application/xml")
-        response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"'
-        response["Cache-Control"] = "no-cache"
-        return response
-
     def epg_generator():
         """Generator function that yields EPG data with keep-alives during processing."""
 
-        xml_lines = []
-        xml_lines.append('<?xml version="1.0" encoding="UTF-8"?>')
-        xml_lines.append(
-            '<tv generator-info-name="Dispatcharr" generator-info-url="https://github.com/Dispatcharr/Dispatcharr">'
+        yield '<?xml version="1.0" encoding="UTF-8"?>\n'
+        yield (
+            '<tv generator-info-name="Dispatcharr" '
+            'generator-info-url="https://github.com/Dispatcharr/Dispatcharr">\n'
         )
 
         # Get channels based on user/profile
@@ -1416,14 +1493,19 @@ def generate_epg(request, profile_name=None, user=None):
         # Resolve effective values at SQL level and exclude hidden channels
         # so output ordering/display honors user overrides.
         from apps.channels.managers import with_effective_values
-        channels = (
+        channels = list(
             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.order_by('channelstream__order'))
             )
         )
+        channel_count = len(channels)
 
         # For dummy EPG, use either the specified value or default to 3 days
         dummy_days = num_days if num_days > 0 else 3
@@ -1465,12 +1547,14 @@ def generate_epg(request, profile_name=None, user=None):
         _logo_url_prefix = _base_url + _logo_prefix_raw + "/"
         _logo_url_suffix = "/" + _logo_suffix_raw
 
-        dummy_epg_ids_for_program_check = set()
+        dummy_program_list = []
+        real_epg_map = {}
+        channel_xml_batch = []
 
-        # Process channels for the <channel> section
         for channel in channels:
             effective_name = channel.effective_name
             effective_epg_data = channel.effective_epg_data_obj
+            effective_epg_data_id = channel.effective_epg_data_id
             effective_logo = channel.effective_logo_obj
             effective_number = channel.effective_channel_number
 
@@ -1492,8 +1576,6 @@ def generate_epg(request, profile_name=None, user=None):
 
             # Check if this is a custom dummy EPG with channel logo URL template
             if effective_epg_data and effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy':
-                if channel.effective_epg_data_id:
-                    dummy_epg_ids_for_program_check.add(channel.effective_epg_data_id)
                 epg_source = effective_epg_data.epg_source
                 if epg_source.custom_properties:
                     custom_props = epg_source.custom_properties
@@ -1548,51 +1630,16 @@ def generate_epg(request, profile_name=None, user=None):
                         tvg_logo = direct_logo
                     else:
                         tvg_logo = f"{_logo_url_prefix}{effective_logo.id}{_logo_url_suffix}"
-            display_name = effective_name
-            xml_lines.append(f'  <channel id="{html.escape(channel_id)}">')
-            xml_lines.append(f'    <display-name>{html.escape(display_name)}</display-name>')
-            xml_lines.append(f'    <icon src="{html.escape(tvg_logo)}" />')
-            xml_lines.append("  </channel>")
+            channel_xml_batch.append(f'  <channel id="{html.escape(channel_id)}">')
+            channel_xml_batch.append(f'    <display-name>{html.escape(effective_name)}</display-name>')
+            channel_xml_batch.append(f'    <icon src="{html.escape(tvg_logo)}" />')
+            channel_xml_batch.append("  </channel>")
 
-        # Send all channel definitions
-        channel_xml = '\n'.join(xml_lines) + '\n'
-        yield channel_xml
-        xml_lines = []  # Clear to save memory
+            if len(channel_xml_batch) >= _EPG_CHANNEL_XML_BATCH_SIZE * 4:
+                yield '\n'.join(channel_xml_batch) + '\n'
+                channel_xml_batch = []
 
-        dummy_epg_with_programs = set()
-        if dummy_epg_ids_for_program_check:
-            dummy_epg_with_programs = set(
-                ProgramData.objects.filter(epg_id__in=dummy_epg_ids_for_program_check)
-                .values_list('epg_id', flat=True)
-                .distinct()
-            )
-
-        # Pre-pass: categorize channels into dummy and real EPG groups
-        dummy_program_list = []  # (channel_id, pattern_match_name, epg_source_or_None)
-        real_epg_map = {}  # epg_data_id -> [channel_id, ...]
-
-        for channel in channels:
-            effective_name = channel.effective_name
-            effective_epg_data = channel.effective_epg_data_obj
-            effective_epg_data_id = channel.effective_epg_data_id
-            effective_number = channel.effective_channel_number
-
-            # Determine channel_id (same logic as channel section)
-            if tvg_id_source == 'tvg_id' and channel.effective_tvg_id:
-                channel_id = channel.effective_tvg_id
-            elif tvg_id_source == 'gracenote' and channel.effective_tvc_guide_stationid:
-                channel_id = channel.effective_tvc_guide_stationid
-            else:
-                if user is not None:
-                    formatted_channel_number = channel_num_map[channel.id]
-                else:
-                    formatted_channel_number = format_channel_number(effective_number)
-                channel_id = str(formatted_channel_number) if formatted_channel_number != "" else str(channel.id)
-
-            display_name = effective_epg_data.name if effective_epg_data else effective_name
             pattern_match_name = effective_name
-
-            # Check if we should use stream name instead of channel name
             if effective_epg_data and effective_epg_data.epg_source:
                 epg_source = effective_epg_data.epg_source
                 if epg_source.custom_properties:
@@ -1619,48 +1666,19 @@ def generate_epg(request, profile_name=None, user=None):
 
             if not effective_epg_data:
                 dummy_program_list.append((channel_id, pattern_match_name, None))
+            elif effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy':
+                dummy_program_list.append((channel_id, pattern_match_name, effective_epg_data.epg_source))
             else:
-                if effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy':
-                    if effective_epg_data_id in dummy_epg_with_programs:
-                        real_epg_map.setdefault(effective_epg_data_id, []).append(channel_id)
-                    else:
-                        dummy_program_list.append((channel_id, pattern_match_name, effective_epg_data.epg_source))
-                    continue
-
                 real_epg_map.setdefault(effective_epg_data_id, []).append(channel_id)
 
-        # Emit dummy programmes
-        for channel_id, pattern_match_name, epg_source in dummy_program_list:
-            program_length_hours = 4
-            dummy_programs = generate_dummy_programs(
-                channel_id, pattern_match_name,
-                num_days=dummy_days,
-                program_length_hours=program_length_hours,
-                epg_source=epg_source
-            )
-            for program in dummy_programs:
-                start_str = program['start_time'].strftime("%Y%m%d%H%M%S %z")
-                stop_str = program['end_time'].strftime("%Y%m%d%H%M%S %z")
-                yield f'  <programme start="{start_str}" stop="{stop_str}" channel="{html.escape(channel_id)}">\n'
-                yield f"    <title>{html.escape(program['title'])}</title>\n"
-                if program.get('sub_title'):
-                    yield f"    <sub-title>{html.escape(program['sub_title'])}</sub-title>\n"
-                yield f"    <desc>{html.escape(program['description'])}</desc>\n"
-                custom_data = program.get('custom_properties', {})
-                if 'categories' in custom_data:
-                    for cat in custom_data['categories']:
-                        yield f"    <category>{html.escape(cat)}</category>\n"
-                if 'date' in custom_data:
-                    yield f"    <date>{html.escape(custom_data['date'])}</date>\n"
-                if custom_data.get('live', False):
-                    yield f"    <live />\n"
-                if custom_data.get('new', False):
-                    yield f"    <new />\n"
-                if 'icon' in custom_data:
-                    yield f'    <icon src="{html.escape(custom_data["icon"])}" />\n'
-                yield f"  </programme>\n"
+        if channel_xml_batch:
+            yield '\n'.join(channel_xml_batch) + '\n'
+
+        del channels
+        del channel_num_map
+
+        batch_size = _EPG_PROGRAM_YIELD_BATCH_SIZE
 
-        # Emit real programmes: single bulk query, chunked to avoid server-side cursor issues.
         all_epg_ids = list(real_epg_map.keys())
         if all_epg_ids:
             if num_days > 0:
@@ -1682,27 +1700,44 @@ def generate_epg(request, profile_name=None, user=None):
 
             current_epg_id = None
             channel_ids_for_epg = None
-            is_multi = False
-            multi_buffer = []
+            pending = []
             program_batch = []
-            batch_size = 1000
-            chunk_size = 5000
-            # Keyset pagination: track last (epg_id, id) instead of OFFSET
-            # to avoid skipping/duplicating rows if the table changes mid-stream.
+            chunk_size = _EPG_PROGRAM_DB_CHUNK_SIZE
             last_epg_id = 0
             last_id = 0
             _poster_url_base = build_absolute_uri_with_port(request, "/api/epg/programs/")
 
+            def flush_pending():
+                nonlocal program_batch, pending
+                if not pending:
+                    return
+                pending.sort(key=lambda row: (row[0], row[1]))
+                escaped_primary = (
+                    html.escape(channel_ids_for_epg[0])
+                    if len(channel_ids_for_epg) > 1 else None
+                )
+                for _, _, xml_text in pending:
+                    program_batch.append(xml_text)
+                    if escaped_primary:
+                        for cid in channel_ids_for_epg[1:]:
+                            program_batch.append(xml_text.replace(
+                                f'channel="{escaped_primary}"',
+                                f'channel="{html.escape(cid)}"',
+                                1,
+                            ))
+                    if len(program_batch) >= batch_size:
+                        yield '\n'.join(program_batch) + '\n'
+                        program_batch = []
+                pending.clear()
+
             while True:
                 program_chunk = list(
                     programs_base_qs.filter(epg_id__gte=last_epg_id)
                     .exclude(epg_id=last_epg_id, id__lte=last_id)[:chunk_size]
                 )
-
                 if not program_chunk:
                     break
 
-                # Advance keyset cursor to last row in this chunk
                 last_row = program_chunk[-1]
                 last_epg_id = last_row['epg_id']
                 last_id = last_row['id']
@@ -1710,31 +1745,19 @@ def generate_epg(request, profile_name=None, user=None):
                 for prog in program_chunk:
                     epg_id = prog['epg_id']
 
-                    # When epg_id changes, flush multi-channel buffer for previous group
                     if epg_id != current_epg_id:
-                        if is_multi and multi_buffer:
-                            escaped_primary = html.escape(channel_ids_for_epg[0])
-                            for extra_cid in channel_ids_for_epg[1:]:
-                                escaped_extra = html.escape(extra_cid)
-                                for xml_text in multi_buffer:
-                                    program_batch.append(xml_text.replace(
-                                        f'channel="{escaped_primary}"',
-                                        f'channel="{escaped_extra}"',
-                                        1,
-                                    ))
-                                    if len(program_batch) >= batch_size:
-                                        yield '\n'.join(program_batch) + '\n'
-                                        program_batch = []
-                            multi_buffer = []
-
+                        yield from flush_pending()
                         current_epg_id = epg_id
                         channel_ids_for_epg = real_epg_map[epg_id]
-                        is_multi = len(channel_ids_for_epg) > 1
 
-                    # Build programme XML for primary channel_id
                     primary_cid = channel_ids_for_epg[0]
-                    start_str = prog['start_time'].strftime("%Y%m%d%H%M%S %z")
-                    stop_str = prog['end_time'].strftime("%Y%m%d%H%M%S %z")
+                    # DB datetimes are UTC (USE_TZ=True, TIME_ZONE=UTC); format
+                    # directly instead of strftime("%Y%m%d%H%M%S %z"), which is
+                    # ~10x slower and dominates XML build over 750k rows.
+                    st = prog['start_time']
+                    et = prog['end_time']
+                    start_str = f"{st.year:04d}{st.month:02d}{st.day:02d}{st.hour:02d}{st.minute:02d}{st.second:02d} +0000"
+                    stop_str = f"{et.year:04d}{et.month:02d}{et.day:02d}{et.hour:02d}{et.minute:02d}{et.second:02d} +0000"
 
                     program_xml = [f'  <programme start="{start_str}" stop="{stop_str}" channel="{html.escape(primary_cid)}">']
                     program_xml.append(f'    <title>{html.escape(prog["title"])}</title>')
@@ -1932,67 +1955,101 @@ def generate_epg(request, profile_name=None, user=None):
                     program_xml.append("  </programme>")
 
                     xml_text = '\n'.join(program_xml)
-                    program_batch.append(xml_text)
+                    pending.append((prog['start_time'], prog['id'], xml_text))
 
-                    if is_multi:
-                        multi_buffer.append(xml_text)
+                del program_chunk
 
-                    if len(program_batch) >= batch_size:
-                        yield '\n'.join(program_batch) + '\n'
-                        program_batch = []
-
-            # Final flush of multi-channel buffer
-            if is_multi and multi_buffer:
-                escaped_primary = html.escape(channel_ids_for_epg[0])
-                for extra_cid in channel_ids_for_epg[1:]:
-                    escaped_extra = html.escape(extra_cid)
-                    for xml_text in multi_buffer:
-                        program_batch.append(xml_text.replace(
-                            f'channel="{escaped_primary}"',
-                            f'channel="{escaped_extra}"',
-                            1,
-                        ))
+            yield from flush_pending()
 
             if program_batch:
                 yield '\n'.join(program_batch) + '\n'
 
-        # Send final closing tag and completion message
+        del real_epg_map
+
+        for channel_id, pattern_match_name, epg_source in dummy_program_list:
+            program_length_hours = 4
+            dummy_programs = generate_dummy_programs(
+                channel_id, pattern_match_name,
+                num_days=dummy_days,
+                program_length_hours=program_length_hours,
+                epg_source=epg_source,
+                export_lookback=lookback_cutoff,
+                export_cutoff=cutoff_date,
+            )
+            if not dummy_programs:
+                continue
+            dummy_batch = []
+            for program in dummy_programs:
+                start_str = program['start_time'].strftime("%Y%m%d%H%M%S %z")
+                stop_str = program['end_time'].strftime("%Y%m%d%H%M%S %z")
+                lines = [
+                    f'  <programme start="{start_str}" stop="{stop_str}" channel="{html.escape(channel_id)}">',
+                    f"    <title>{html.escape(program['title'])}</title>",
+                ]
+                if program.get('sub_title'):
+                    lines.append(f"    <sub-title>{html.escape(program['sub_title'])}</sub-title>")
+                lines.append(f"    <desc>{html.escape(program['description'])}</desc>")
+                custom_data = program.get('custom_properties', {})
+                if 'categories' in custom_data:
+                    for cat in custom_data['categories']:
+                        lines.append(f"    <category>{html.escape(cat)}</category>")
+                if 'date' in custom_data:
+                    lines.append(f"    <date>{html.escape(custom_data['date'])}</date>")
+                if custom_data.get('live', False):
+                    lines.append("    <live />")
+                if custom_data.get('new', False):
+                    lines.append("    <new />")
+                if 'icon' in custom_data:
+                    lines.append(f'    <icon src="{html.escape(custom_data["icon"])}" />')
+                lines.append("  </programme>")
+                dummy_batch.append('\n'.join(lines))
+                if len(dummy_batch) >= batch_size:
+                    yield '\n'.join(dummy_batch) + '\n'
+                    dummy_batch = []
+            del dummy_programs
+            if dummy_batch:
+                yield '\n'.join(dummy_batch) + '\n'
+
+        del dummy_program_list
+
         yield "</tv>\n"
 
-        # Log system event for EPG download after streaming completes (with deduplication based on client)
         client_id, client_ip, user_agent = get_client_identifier(request)
         event_cache_key = f"epg_download:{user.username if user else 'anonymous'}:{profile_name or 'all'}:{client_id}"
-        if not cache.get(event_cache_key):
-            # `len()` reuses the queryset's iteration cache populated above;
-            # `count()` would issue a separate SELECT COUNT(*).
-            log_system_event(
-                event_type='epg_download',
-                profile=profile_name or 'all',
-                user=user.username if user else 'anonymous',
-                channels=len(channels),
-                client_ip=client_ip,
-                user_agent=user_agent,
-            )
-            cache.set(event_cache_key, True, 2)  # Prevent duplicate events for 2 seconds
 
-    # Wrapper generator that collects content for caching
-    def caching_generator():
-        collected_content = []
-        for chunk in epg_generator():
-            collected_content.append(chunk)
-            yield chunk
-        # After streaming completes, cache the full content
-        full_content = ''.join(collected_content)
-        cache.set(content_cache_key, full_content, 300)
-        logger.debug("Cached EPG content (%d bytes)", len(full_content))
+        def _log_epg_download():
+            from django.core.cache import cache as event_cache
 
-    response = StreamingHttpResponse(
-        streaming_content=caching_generator(),
-        content_type="application/xml"
-    )
-    response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"'
-    response["Cache-Control"] = "no-cache"
-    return response
+            if not event_cache.get(event_cache_key):
+                log_system_event(
+                    event_type='epg_download',
+                    profile=profile_name or 'all',
+                    user=user.username if user else 'anonymous',
+                    channels=channel_count,
+                    client_ip=client_ip,
+                    user_agent=user_agent,
+                )
+                event_cache.set(event_cache_key, True, 2)
+
+        try:
+            from core.utils import _is_gevent_monkey_patched
+
+            if _is_gevent_monkey_patched():
+                import gevent
+
+                gevent.spawn(_log_epg_download)
+            else:
+                _log_epg_download()
+        except Exception:
+            _log_epg_download()
+
+    def build_epg_stream():
+        try:
+            yield from epg_generator()
+        finally:
+            _epg_export_teardown()
+
+    return stream_epg_response(content_cache_key, build_epg_stream)
 
 
 def xc_get_user(request):

core/tests.py

@@ -1,6 +1,6 @@
 from unittest.mock import patch, MagicMock
 
-from django.test import TestCase
+from django.test import TestCase, SimpleTestCase
 
 from apps.epg.models import EPGSource
 from core.models import CoreSettings, DVR_SETTINGS_KEY, EPG_SETTINGS_KEY
@@ -301,3 +301,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