Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-17 16:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

perf(epg): async index build, defer index column, tight build lock, channel-id canonicalization, retry-on-error

Browse source
This commit is contained in:
Five Boroughs 2026-05-31 04:30:35 +02:00
parent 9903f0b75d
commit 10d558f136
No known key found for this signature in database
7 changed files with 123 additions and 118 deletions
Download patch file Download diff file Expand all files Collapse all files

5
apps/epg/api_views.py
View file

@ -777,7 +777,10 @@ class CurrentProgramsAPIView(APIView):
# Limit to 50 IDs per request
epg_data_ids = epg_data_ids[:50]
epg_data_entries = EPGData.objects.select_related('epg_source').filter(id__in=epg_data_ids)
# 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)
# Batch-fetch current programs for all requested EPG entries in one query
db_programs = ProgramData.objects.filter(

40
apps/epg/tasks.py
View file

@ -318,8 +318,8 @@ def refresh_epg_data(source_id):
gc.collect()
return
# Build byte-offset index for preview lookups (runs before full program parse)
build_programme_index(source.id)
# Build byte-offset index for preview lookups in the background so refresh isn't blocked by it
build_programme_index_task.delay(source.id)
parse_programs_for_source(source)
@ -2368,6 +2368,14 @@ _MAX_START_TAG = 4096 # generous upper bound for a start tag with namespaces/ex
_OFFSET_CAP = 10 # max block-starts recorded per channel; exceeding this flags the channel as interleaved
def _decode_channel_id(raw):
"""Match how EPGData.tvg_id is stored: resolve XML entities and strip, so byte-level index keys equal the lxml-parsed channel ids."""
s = raw.decode('utf-8', errors='replace')
if '&' in s:
s = html.unescape(s)
return s.strip()
def _find_programme_tag(buf, start):
"""
Find the next <programme element in *buf* starting from *start*.
@ -2465,9 +2473,7 @@ def build_programme_index(source_id):
buf, idx, tag_end + 1 if tag_end != -1 else idx + _MAX_START_TAG
)
if m:
channel_id = (m.group(1) or m.group(2)).decode(
'utf-8', errors='replace'
)
channel_id = _decode_channel_id(m.group(1) or m.group(2))
if channel_id not in index:
index[channel_id] = [abs_pos]
elif channel_id != prev_channel:
@ -2510,8 +2516,17 @@ def build_programme_index(source_id):
@shared_task
def build_programme_index_task(source_id):
"""Celery wrapper so build_programme_index can be dispatched async."""
build_programme_index(source_id)
"""Celery wrapper. Locks so refresh and preview don't both build the same source. Releases on finish rather than waiting out the TTL."""
from core.utils import RedisClient
redis_client = RedisClient.get_client()
lock_key = f'building_programme_index_{source_id}'
if not redis_client.set(lock_key, '1', nx=True, ex=300):
return
try:
build_programme_index(source_id)
finally:
redis_client.delete(lock_key)
def find_current_program_for_tvg_id(epg_or_id):
@ -2522,8 +2537,6 @@ def find_current_program_for_tvg_id(epg_or_id):
Returns dict, None, or "timeout".
"""
from core.utils import RedisClient
if isinstance(epg_or_id, EPGData):
epg = epg_or_id
else:
@ -2575,10 +2588,7 @@ def find_current_program_for_tvg_id(epg_or_id):
# No index yet: dispatch a background build and let the frontend retry.
# A sync scan can block a worker for ~10s on SMB-hosted EPGs.
redis_client = RedisClient.get_client()
lock_key = f'building_programme_index_{source.id}'
if redis_client.set(lock_key, '1', nx=True, ex=300):
build_programme_index_task.delay(source.id)
build_programme_index_task.delay(source.id)
return 'timeout'
@ -2625,7 +2635,7 @@ def _read_programs_at_offsets(file_path, tvg_id, offsets, now):
)
continue
ch = (m.group(1) or m.group(2)).decode('utf-8', errors='replace')
ch = _decode_channel_id(m.group(1) or m.group(2))
if ch != tvg_id:
done = True # different channel, end of block
break
@ -2718,7 +2728,7 @@ def _scan_from_offset_for_tvg_id(file_path, tvg_id, start_offset, now, timeout_s
)
continue
ch = (m.group(1) or m.group(2)).decode('utf-8', errors='replace')
ch = _decode_channel_id(m.group(1) or m.group(2))
if ch != tvg_id:
search_from = (
tag_end + 1 if tag_end != -1 else tag_start + _PROGRAMME_TAG_LEN

113
apps/epg/tests/test_programme_index.py
View file

@ -6,7 +6,11 @@ from django.test import TestCase
from django.utils import timezone
from apps.epg.models import EPGSource, EPGData
from apps.epg.tasks import find_current_program_for_tvg_id, build_programme_index
from apps.epg.tasks import (
find_current_program_for_tvg_id,
build_programme_index,
build_programme_index_task,
)
FIXTURE_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
FIXTURE_XML = os.path.join(FIXTURE_DIR, "test_epg.xml")
@ -161,6 +165,43 @@ class FindCurrentProgramTests(TestCase):
finally:
os.unlink(tmp_path)
def test_channel_id_entities_and_whitespace_match_tvg_id(self):
# programme@channel carries an XML entity and surrounding whitespace;
# EPGData.tvg_id holds the lxml-decoded, stripped form. The index key
# and lookup must canonicalize to the same value.
xml = (
'<?xml version="1.0" encoding="UTF-8"?>\n'
"<tv>\n"
' <channel id="A&amp;E.us"/>\n'
' <programme start="20000101000000 +0000" '
'stop="20991231235959 +0000" channel=" A&amp;E.us ">\n'
" <title>A and E Now</title>\n"
" </programme>\n"
"</tv>\n"
)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".xml", delete=False
) as f:
f.write(xml)
tmp_path = f.name
try:
src = EPGSource.objects.create(
name="Entities", source_type="xmltv", file_path=tmp_path
)
build_programme_index(src.id)
src.refresh_from_db()
self.assertIn("A&E.us", src.programme_index["channels"])
epg = EPGData.objects.create(
tvg_id="A&E.us", name="A&E", epg_source=src
)
result = find_current_program_for_tvg_id(epg)
self.assertIsNotNone(result)
self.assertEqual(result["title"], "A and E Now")
finally:
os.unlink(tmp_path)
@patch("apps.epg.tasks.build_programme_index_task")
def test_no_index_dispatches_build_and_returns_timeout(self, mock_build_task):
# Source with no index and file on disk
@ -176,44 +217,10 @@ class FindCurrentProgramTests(TestCase):
epg_source=src,
)
mock_redis = MagicMock()
mock_redis.set.return_value = True
with patch(
"core.utils.RedisClient.get_client",
return_value=mock_redis,
):
result = find_current_program_for_tvg_id(epg)
result = find_current_program_for_tvg_id(epg)
self.assertEqual(result, "timeout")
mock_build_task.delay.assert_called_once_with(src.id)
mock_redis.set.assert_called_once()
lock_key = mock_redis.set.call_args.args[0]
self.assertEqual(lock_key, f"building_programme_index_{src.id}")
@patch("apps.epg.tasks.build_programme_index_task")
def test_no_index_skips_dispatch_when_lock_held(self, mock_build_task):
src = EPGSource.objects.create(
name="Lock Held",
source_type="xmltv",
file_path=FIXTURE_XML,
programme_index=None,
)
epg = EPGData.objects.create(
tvg_id="channel.current",
name="Current",
epg_source=src,
)
mock_redis = MagicMock()
mock_redis.set.return_value = False # someone else already building
with patch(
"core.utils.RedisClient.get_client",
return_value=mock_redis,
):
result = find_current_program_for_tvg_id(epg)
self.assertEqual(result, "timeout")
mock_build_task.delay.assert_not_called()
class BuildProgrammeIndexTests(TestCase):
@ -240,6 +247,40 @@ class BuildProgrammeIndexTests(TestCase):
# Should log error but not raise
build_programme_index(99999)
@patch("apps.epg.tasks.build_programme_index")
def test_task_builds_and_releases_lock_when_free(self, mock_build):
mock_redis = MagicMock()
mock_redis.set.return_value = True # lock acquired
with patch("core.utils.RedisClient.get_client", return_value=mock_redis):
build_programme_index_task(42)
mock_build.assert_called_once_with(42)
mock_redis.set.assert_called_once()
self.assertEqual(
mock_redis.set.call_args.args[0], "building_programme_index_42"
)
mock_redis.delete.assert_called_once_with("building_programme_index_42")
@patch("apps.epg.tasks.build_programme_index")
def test_task_skips_when_lock_held(self, mock_build):
mock_redis = MagicMock()
mock_redis.set.return_value = False # another build in flight
with patch("core.utils.RedisClient.get_client", return_value=mock_redis):
build_programme_index_task(42)
mock_build.assert_not_called()
mock_redis.delete.assert_not_called()
@patch("apps.epg.tasks.build_programme_index", side_effect=RuntimeError("boom"))
def test_task_releases_lock_on_failure(self, mock_build):
mock_redis = MagicMock()
mock_redis.set.return_value = True
with patch("core.utils.RedisClient.get_client", return_value=mock_redis):
with self.assertRaises(RuntimeError):
build_programme_index_task(42)
mock_redis.delete.assert_called_once_with("building_programme_index_42")
def test_per_channel_interleaved_marking(self):
xml = (
'<?xml version="1.0" encoding="UTF-8"?>\n'

32
frontend/src/api.js
View file

@ -1530,27 +1530,21 @@ export default class API {
}
static async getCurrentProgramForEpg(epgId) {
try {
const response = await request(
`${host}/api/epg/current-programs/`,
{
method: 'POST',
body: { epg_data_ids: [epgId] },
}
);
// The endpoint returns an array, get the first item
if (response && response.length > 0) {
if (response[0].parsing) {
return { parsing: true };
}
return response[0];
const response = await request(
`${host}/api/epg/current-programs/`,
{
method: 'POST',
body: { epg_data_ids: [epgId] },
}
return null;
} catch (e) {
console.error('Failed to retrieve current program for EPG', e);
return null;
);
if (response && response.length > 0) {
if (response[0].parsing) {
return { parsing: true };
}
return response[0];
}
return null;
}
// Notice there's a duplicated "refreshPlaylist" method above;

44
frontend/src/components/cards/StreamConnectionCard.jsx
View file

@ -54,50 +54,6 @@ import {
import useVideoStore from '../../store/useVideoStore';
import { buildLiveStreamUrl } from '../../utils/components/FloatingVideoUtils.js';
const formatProgramTime = (seconds) => {
const absSeconds = Math.abs(seconds);
const hours = Math.floor(absSeconds / 3600);
const minutes = Math.floor((absSeconds % 3600) / 60);
const secs = Math.floor(absSeconds % 60);
if (hours > 0) {
return `${hours}:${minutes.toString().padStart(2, '0')}:${secs.toString().padStart(2, '0')}`;
}
return `${minutes}:${secs.toString().padStart(2, '0')}`;
};
const ProgramProgress = ({ currentProgram }) => {
const now = new Date();
const startTime = new Date(currentProgram.start_time);
const endTime = new Date(currentProgram.end_time);
const totalDuration = (endTime - startTime) / 1000; // in seconds
const elapsed = (now - startTime) / 1000; // in seconds
const remaining = (endTime - now) / 1000; // in seconds
const percentage = Math.min(
100,
Math.max(0, (elapsed / totalDuration) * 100)
);
return (
<Stack gap="xs" mt={4}>
<Group justify="space-between" align="center">
<Text size="xs" c="dimmed">
{formatProgramTime(elapsed)} elapsed
</Text>
<Text size="xs" c="dimmed">
{formatProgramTime(remaining)} remaining
</Text>
</Group>
<Progress
value={percentage}
size="sm"
color="#3BA882"
style={{
backgroundColor: 'rgba(255, 255, 255, 0.1)',
}}
/>
</Stack>
);
};
// Create a separate component for each channel card to properly handle the hook
const StreamConnectionCard = ({

5
frontend/src/components/forms/__tests__/ChannelEpgPreview.test.jsx
View file

@ -145,9 +145,10 @@ describe('Channel EPG preview hook', () => {
await act(async () => {
await vi.advanceTimersByTimeAsync(0);
});
for (const delay of [3000, 4500, 6750, 10125, 15000]) {
// Backoff ramps to a 15s cap; advance past the 180s deadline that bounds the loop.
for (let elapsed = 0; elapsed <= 190000; elapsed += 15100) {
await act(async () => {
await vi.advanceTimersByTimeAsync(delay + 100);
await vi.advanceTimersByTimeAsync(15100);
});
}

2
frontend/src/hooks/useEpgPreview.jsx
View file

@ -19,7 +19,7 @@ export const useEpgPreview = (epgDataId) => {
setHasFetchedProgram(false);
const fetchWithRetry = async () => {
const maxRetries = 5;
const maxRetries = 20;
const deadlineMs = 3 * 60 * 1000;
const startTime = Date.now();
let delay = 3000;