From 14a6bf4473bc4636d679ad2bf7d2d11e8957dc85 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 26 Apr 2026 09:06:48 -0500 Subject: [PATCH 01/55] Initial DVR HLS refactor. --- apps/channels/api_urls.py | 5 + apps/channels/api_views.py | 105 +- apps/channels/tasks.py | 1126 +++++++++-------- .../tests/test_dvr_port_resolution.py | 88 +- docker/docker-compose.yml | 6 +- 5 files changed, 742 insertions(+), 588 deletions(-) diff --git a/apps/channels/api_urls.py b/apps/channels/api_urls.py index bd53ae45..b2af3167 100644 --- a/apps/channels/api_urls.py +++ b/apps/channels/api_urls.py @@ -49,6 +49,11 @@ urlpatterns = [ path('series-rules/bulk-remove/', BulkRemoveSeriesRecordingsAPIView.as_view(), name='bulk_remove_series_recordings'), path('series-rules//', DeleteSeriesRuleAPIView.as_view(), name='delete_series_rule'), path('recordings/bulk-delete-upcoming/', BulkDeleteUpcomingRecordingsAPIView.as_view(), name='bulk_delete_upcoming_recordings'), + path( + 'recordings//hls/', + RecordingViewSet.as_view({'get': 'hls'}), + name='recording-hls', + ), path('dvr/comskip-config/', ComskipConfigAPIView.as_view(), name='comskip_config'), ] diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 3e786ff1..2907c48d 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -2444,7 +2444,7 @@ class RecordingViewSet(viewsets.ModelViewSet): def get_permissions(self): # Allow unauthenticated playback of recording files (like other streaming endpoints) - if self.action == 'file': + if self.action in ('file', 'hls'): return [AllowAny()] try: return [perm() for perm in permission_classes_by_action[self.action]] @@ -2464,14 +2464,29 @@ class RecordingViewSet(viewsets.ModelViewSet): @action(detail=True, methods=["get"], url_path="file") def file(self, request, pk=None): - """Stream a recorded file with HTTP Range support for seeking.""" + """Stream a completed recording file with HTTP Range support for seeking. + + For in-progress recordings, file_url in custom_properties points to + /hls/index.m3u8. If a client hits this endpoint while the recording + is still running (or the MKV is not yet produced), it is redirected to + the HLS playlist endpoint. + """ recording = get_object_or_404(Recording, pk=pk) cp = recording.custom_properties or {} file_path = cp.get("file_path") file_name = cp.get("file_name") or "recording" - if not file_path or not os.path.exists(file_path): - raise Http404("Recording file not found") + if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0: + # Redirect to HLS if recording is still in progress + hls_dir = cp.get("_hls_dir") + if hls_dir and os.path.isdir(hls_dir): + from django.http import HttpResponseRedirect + hls_url = request.build_absolute_uri( + f"/api/channels/recordings/{pk}/hls/index.m3u8" + ) + return HttpResponseRedirect(hls_url) + if not file_path or not os.path.exists(file_path): + raise Http404("Recording file not found") # Guess content type ext = os.path.splitext(file_path)[1].lower() @@ -2532,6 +2547,73 @@ class RecordingViewSet(viewsets.ModelViewSet): response["Content-Disposition"] = f"inline; filename=\"{file_name}\"" return response + @action(detail=True, methods=["get"], url_path="hls/(?P.+)") + def hls(self, request, pk=None, seg_path="index.m3u8"): + """Serve HLS playlist and segment files for an in-progress (or completed) recording. + + Clients connecting during recording should use the m3u8 URL returned in + custom_properties.file_url. Segment URLs inside the playlist are rewritten + to route through this endpoint so authentication and path isolation are + preserved. + """ + recording = get_object_or_404(Recording, pk=pk) + cp = recording.custom_properties or {} + hls_dir = cp.get("_hls_dir") + + if not hls_dir or not os.path.isdir(hls_dir): + # HLS dir is gone, recording is likely complete. Redirect to the + # permanent MKV endpoint for .m3u8 requests so clients that still + # have the HLS URL bookmarked get a useful response. + cp = recording.custom_properties or {} + file_path = cp.get("file_path") + if seg_path.endswith(".m3u8") and file_path and os.path.exists(file_path) and os.path.getsize(file_path) > 0: + from django.http import HttpResponseRedirect + return HttpResponseRedirect( + request.build_absolute_uri(f"/api/channels/recordings/{pk}/file/") + ) + raise Http404("HLS content not available for this recording") + + # Security: prevent path traversal outside the HLS directory + safe_dir = os.path.realpath(hls_dir) + requested = os.path.realpath(os.path.join(hls_dir, seg_path)) + if not requested.startswith(safe_dir + os.sep) and requested != safe_dir: + return Response({"error": "Forbidden"}, status=403) + + if not os.path.isfile(requested): + raise Http404(f"HLS file not found: {seg_path}") + + if seg_path.endswith(".m3u8"): + # Rewrite relative segment lines to absolute URLs through this API + base_url = request.build_absolute_uri( + f"/api/channels/recordings/{pk}/hls/" + ) + lines = [] + with open(requested) as _f: + for line in _f: + stripped = line.strip() + if stripped and not stripped.startswith("#"): + lines.append(f"{base_url}{stripped}\n") + else: + lines.append(line) + from django.http import HttpResponse as _HR + return _HR("".join(lines), content_type="application/x-mpegURL") + + if seg_path.endswith(".ts"): + # Refresh the viewer heartbeat in Redis so the Celery task knows an + # active client is still fetching segments. TTL is 20 s, enough for + # three 4-second segments plus network margin. + try: + from core.utils import RedisClient + _rv = RedisClient.get_client(max_retries=1, retry_interval=0) + if _rv: + _rv.set(f"dvr:hls_viewer:{pk}", "1", ex=20) + except Exception: + pass + from django.http import FileResponse as _FR + return _FR(open(requested, "rb"), content_type="video/mp2t") + + raise Http404("Unsupported HLS file type") + @action(detail=True, methods=["post"], url_path="stop") def stop(self, request, pk=None): """Stop a recording early while retaining the partial content for playback.""" @@ -2838,7 +2920,7 @@ class RecordingViewSet(viewsets.ModelViewSet): cp = instance.custom_properties or {} rec_status = cp.get("status", "") file_path = cp.get("file_path") - temp_ts_path = cp.get("_temp_file_path") + hls_dir = cp.get("_hls_dir") channel_uuid = str(instance.channel.uuid) # 1. Delete the DB record (also fires post_delete → revoke_task_on_delete) @@ -2871,6 +2953,17 @@ class RecordingViewSet(viewsets.ModelViewSet): except Exception as ex: logger.warning(f"Failed to delete recording artifact {path}: {ex}") + def _safe_rmtree(path: str): + if not path or not isinstance(path, str): + return + try: + import shutil as _shutil + if any(path.startswith(root) for root in allowed_roots) and os.path.isdir(path): + _shutil.rmtree(path) + logger.info(f"Deleted recording HLS directory: {path}") + except Exception as ex: + logger.warning(f"Failed to delete HLS directory {path}: {ex}") + def _background_cancel(): # Only stop the DVR client if the recording was actively streaming. # Stopping for completed/upcoming recordings would kill an unrelated @@ -2888,7 +2981,7 @@ class RecordingViewSet(viewsets.ModelViewSet): # Best-effort file cleanup in case run_recording already exited # before the DB delete. _safe_remove(file_path) - _safe_remove(temp_ts_path) + _safe_rmtree(hls_dir) try: from django.db import connection as _conn diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 59122d9f..67bec477 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -8,6 +8,8 @@ import time import json import subprocess import signal +import threading +from collections import deque from zoneinfo import ZoneInfo from datetime import datetime, timedelta import gc @@ -1654,22 +1656,18 @@ def _build_output_paths(channel, program, start_time, end_time): final_path = os.path.normpath(final_path) # Avoid overwriting an existing file from a different recording. - # Check BOTH .mkv and .ts — a pre-restart TS segment may exist at - # the same base name even when the MKV is a 0-byte placeholder. + # Check the MKV file and its associated HLS working directory. base, ext = os.path.splitext(final_path) counter = 1 while True: candidate_base = final_path[:-len(ext)] # strip extension - ts_candidate = candidate_base + '.ts' + hls_dir_candidate = candidate_base + '_hls' try: mkv_occupied = os.stat(final_path).st_size > 0 except OSError: mkv_occupied = False - try: - ts_occupied = os.stat(ts_candidate).st_size > 0 - except OSError: - ts_occupied = False - if not mkv_occupied and not ts_occupied: + hls_occupied = os.path.isdir(hls_dir_candidate) + if not mkv_occupied and not hls_occupied: break counter += 1 final_path = f"{base}_{counter}{ext}" @@ -1677,37 +1675,36 @@ def _build_output_paths(channel, program, start_time, end_time): # Ensure directory exists os.makedirs(os.path.dirname(final_path), exist_ok=True) - # Derive temp TS path in same directory + # Derive HLS working directory alongside the final MKV base_no_ext = os.path.splitext(os.path.basename(final_path))[0] - temp_ts_path = os.path.join(os.path.dirname(final_path), f"{base_no_ext}.ts") - return final_path, temp_ts_path, os.path.basename(final_path) + hls_dir = os.path.join(os.path.dirname(final_path), f"{base_no_ext}_hls") + return final_path, hls_dir, os.path.basename(final_path) -def build_dvr_candidates(): - """Build ordered list of candidate base URLs for DVR TS streaming. +def get_dvr_stream_base_url(): + """Return the single correct base URL for DVR to reach the TS stream proxy. - Reads environment variables to determine which URLs to try: - - DISPATCHARR_INTERNAL_TS_BASE_URL: explicit override (first priority) - - DISPATCHARR_PORT: the external port (default 9191) - - DISPATCHARR_ENV/DISPATCHARR_DEBUG/REDIS_HOST: dev-mode detection - - DISPATCHARR_INTERNAL_API_BASE: override for the docker service URL + Priority: + 1. DISPATCHARR_INTERNAL_TS_BASE_URL — explicit override, always wins. + 2. Modular mode (DISPATCHARR_ENV=modular) — celery runs in a separate container + and must reach the web container by its Docker service name on DISPATCHARR_PORT. + Override the host with DISPATCHARR_WEB_HOST for non-standard compose setups. + 3. AIO / dev / debug — celery shares the container with uwsgi which binds on + port 5656; use 127.0.0.1 to avoid any nginx layer. """ explicit = os.environ.get('DISPATCHARR_INTERNAL_TS_BASE_URL') - dispatcharr_port = os.environ.get('DISPATCHARR_PORT', '9191') - is_dev = (os.environ.get('DISPATCHARR_ENV', '').lower() == 'dev') or \ - (os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true') or \ - (os.environ.get('REDIS_HOST', 'redis') in ('localhost', '127.0.0.1')) - candidates = [] if explicit: - candidates.append(explicit) - if is_dev: - # Debug container typically exposes API on 5656 (uwsgi internal port) - candidates.extend(['http://127.0.0.1:5656', f'http://127.0.0.1:{dispatcharr_port}']) - # Docker service name fallback — use DISPATCHARR_PORT so modular mode works with custom ports - candidates.append(os.environ.get('DISPATCHARR_INTERNAL_API_BASE', f'http://web:{dispatcharr_port}')) - # Last-resort localhost ports - candidates.extend(['http://localhost:5656', f'http://localhost:{dispatcharr_port}']) - return candidates + return explicit.rstrip('/') + + dispatcharr_env = os.environ.get('DISPATCHARR_ENV', 'aio').lower() + + if dispatcharr_env == 'modular': + host = os.environ.get('DISPATCHARR_WEB_HOST', 'web') + port = os.environ.get('DISPATCHARR_PORT', '9191') + return f'http://{host}:{port}' + + # AIO, dev, debug: celery and uwsgi share the container, reach uwsgi directly + return 'http://127.0.0.1:5656' @shared_task @@ -1759,11 +1756,10 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): start_time = datetime.fromisoformat(start_time_str) end_time = datetime.fromisoformat(end_time_str) - duration_seconds = int((end_time - start_time).total_seconds()) # Build output paths from templates (refined after loading Recording cp below) filename = None final_path = None - temp_ts_path = None + hls_dir = None channel_layer = get_channel_layer() @@ -1810,7 +1806,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): "started_at": str(datetime.now()), }) # Provide a predictable playback URL for the frontend - cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/" + cp["file_url"] = f"/api/channels/recordings/{recording_id}/hls/index.m3u8" cp["output_file_url"] = cp["file_url"] # Determine program info (may include id for deeper details) @@ -1825,10 +1821,10 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): program.update(epg_match) cp["program"] = program - final_path, temp_ts_path, filename = _build_output_paths(channel, program, start_time, end_time) + final_path, hls_dir, filename = _build_output_paths(channel, program, start_time, end_time) cp["file_name"] = filename cp["file_path"] = final_path - cp["_temp_file_path"] = temp_ts_path + cp["_hls_dir"] = hls_dir # Resolve poster art via the shared pipeline (EPG → VOD → TMDB/OMDb → # TVMaze/iTunes → direct program fields → Logo table → channel logo). @@ -1863,7 +1859,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # Merge only the keys explicitly set into the fresh copy for key in ("status", "started_at", "file_url", "output_file_url", - "file_name", "file_path", "_temp_file_path", + "file_name", "file_path", "_hls_dir", "program", "poster_logo_id", "poster_url"): if key in cp: fresh_cp[key] = cp[key] @@ -1886,15 +1882,8 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): interrupted_reason = None bytes_written = 0 - from requests.exceptions import ReadTimeout, ConnectionError as ReqConnectionError, ChunkedEncodingError - - candidates = build_dvr_candidates() - - chosen_base = None - last_error = None - bytes_written = 0 - interrupted = False - interrupted_reason = None + base_url = get_dvr_stream_base_url() + logger.info(f"DVR recording {recording_id}: using stream base URL {base_url}") def _check_recording_cancelled(rid): """Check if a recording was stopped by user or deleted. @@ -1912,232 +1901,331 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): pass return False, False, None - # --- Retry / reconnection constants --- - # Stream reconnection: retry the same TS proxy base on transient - # connectivity loss. Counter resets when data resumes. - _dvr_max_reconnects = 5 - _dvr_reconnect_delay = 2.0 # seconds - # DB save retry: exponential backoff (1s, 2s, 4s) for transient errors. + # --- DB retry constants --- _dvr_db_max_retries = 3 _dvr_db_retry_interval = 1 # seconds (base for exponential backoff) - # FFmpeg remux retry: covers transient I/O errors. - _dvr_remux_max_retries = 2 - _dvr_remux_retry_interval = 2 # seconds (base for exponential backoff) - for base in candidates: - test_url = f"{base.rstrip('/')}/proxy/ts/stream/{channel.uuid}" - logger.info(f"DVR recording {recording_id}: trying TS base {base}") + # Redis key used to signal that an HLS client is actively fetching segments. + # The hls endpoint refreshes this on every .ts request; we check it before + # deleting the HLS directory so in-flight downloads are not cut short. + _hls_viewer_key = f"dvr:hls_viewer:{recording_id}" - _reconnects = 0 - _file_mode = 'wb' - _stream_started_at = None - _done = False + # --- HLS recording pipeline --- + # Launch FFmpeg with the deterministic base URL resolved above. Wait up to + # _first_segment_timeout seconds for the first segment to appear before + # treating the stream as unavailable. + from django.utils import timezone as _tz - while True: # Reconnection loop for this base + last_error = None + ffmpeg_proc = None + hls_m3u8 = None + hls_seg_pattern = None + _stream_confirmed = False + + if hls_dir: + os.makedirs(hls_dir, exist_ok=True) + hls_m3u8 = os.path.join(hls_dir, "index.m3u8") + hls_seg_pattern = os.path.join(hls_dir, "seg_%05d.ts") + + base = base_url + + if not interrupted: + # Check for stop/delete before starting + should_exit, is_int, reason = _check_recording_cancelled(recording_id) + if should_exit: + interrupted = is_int + interrupted_reason = reason + + if not interrupted: + stream_url = f"{base}/proxy/ts/stream/{channel.uuid}" + logger.info(f"DVR recording {recording_id}: stream URL: {stream_url}") + + if not interrupted: + + # Continue segment numbering from any previous session (server-restart resume) + existing_segs = sorted( + f for f in os.listdir(hls_dir) if f.startswith("seg_") and f.endswith(".ts") + ) if hls_dir else [] + hls_start_number = len(existing_segs) + + ffmpeg_cmd = [ + "ffmpeg", "-y", + "-reconnect", "1", + "-reconnect_streamed", "1", + "-reconnect_delay_max", "5", + "-user_agent", f"Dispatcharr-DVR/recording-{recording_id}", + # Regenerate monotonic PTS to handle erratic/discontinuous timestamps + # from IPTV sources. + "-fflags", "+genpts", + "-i", stream_url, + "-c", "copy", + # Shift output timestamps so they start from 0, fixing negative PTS + # values that can prevent segment boundary detection in the HLS muxer. + "-avoid_negative_ts", "make_zero", + "-f", "hls", + "-hls_time", "4", + "-hls_list_size", "0", + "-hls_flags", "append_list+omit_endlist+independent_segments", + "-start_number", str(hls_start_number), + "-hls_segment_filename", hls_seg_pattern, + hls_m3u8, + ] + + logger.info(f"DVR recording {recording_id}: starting FFmpeg — stream URL: {stream_url}") + logger.info(f"DVR recording {recording_id}: HLS output dir: {hls_dir}") + logger.debug(f"DVR recording {recording_id}: FFmpeg command: {' '.join(str(a) for a in ffmpeg_cmd)}") + # Rolling tail of FFmpeg stderr lines for post-mortem diagnostics + _ffmpeg_stderr_tail = deque(maxlen=200) + try: + ffmpeg_proc = subprocess.Popen( + ffmpeg_cmd, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + except Exception as _fe: + last_error = str(_fe) + logger.warning(f"DVR recording {recording_id}: failed to launch FFmpeg: {_fe}") + ffmpeg_proc = None + + # Drain FFmpeg stderr in a background thread to prevent the OS pipe + # buffer (~64 KB on Linux) from filling up, which would block FFmpeg's + # writes and cause it to silently stop demuxing/segmenting after a few + # minutes of normal progress output. Lines are logged to the Dispatcharr + # log and tail-buffered for diagnostics. + def _drain_ffmpeg_stderr(proc, rec_id, tail): + # FFmpeg emits progress lines terminated with \r (carriage return) + # so it can rewrite them in place; only non-progress messages end + # with \n. Split on either to capture both kinds. try: - with requests.get( - test_url, - headers={ - 'User-Agent': f'Dispatcharr-DVR/recording-{recording_id}', - }, - stream=True, - timeout=(10, 15), - ) as response: - response.raise_for_status() - - _test_window = 3.0 - _window_start = time.time() - _stop_poll_interval = 2.0 - _last_stop_poll = time.time() - - with open(temp_ts_path, _file_mode) as file: - if _stream_started_at is None: - _stream_started_at = time.time() - - for chunk in response.iter_content(chunk_size=8192): - if not chunk: - if not chosen_base and (time.time() - _window_start) > _test_window: - break - continue - - if not chosen_base: - chosen_base = base - - # Data received after reconnect — connection restored - if _reconnects > 0: - logger.info( - f"DVR recording {recording_id}: " - f"stream resumed after reconnect" - ) - _reconnects = 0 - - file.write(chunk) - bytes_written += len(chunk) - - elapsed = time.time() - _stream_started_at - if elapsed > duration_seconds: - break - - # Periodic DB poll: stop, delete, end_time extension - _now = time.time() - if _now - _last_stop_poll >= _stop_poll_interval: - _last_stop_poll = _now - try: - _sc = Recording.objects.filter( - id=recording_id - ).only("custom_properties", "end_time").first() - if _sc is None: - logger.info( - f"DVR recording {recording_id}: " - f"deleted — exiting stream loop" - ) - interrupted = False - break - if (_sc.custom_properties or {}).get("status") == "stopped": - logger.info( - f"DVR recording {recording_id}: " - f"stop requested — exiting stream loop" - ) - break - try: - new_end = _sc.end_time - if new_end is not None: - from django.utils import timezone as _tz - if _tz.is_naive(new_end): - new_end = _tz.make_aware(new_end) - _ref = start_time - if _tz.is_naive(_ref): - _ref = _tz.make_aware(_ref) - new_duration = int( - (new_end - _ref).total_seconds() - ) - if new_duration > duration_seconds: - logger.info( - f"DVR recording {recording_id}: " - f"end_time extended to {new_end}, " - f"new duration {new_duration}s" - ) - duration_seconds = new_duration - except Exception: - pass - except Exception: - pass - - # iter_content exhausted or loop exited normally - if bytes_written > 0: - logger.info( - f"DVR recording {recording_id}: " - f"stream complete, {bytes_written} bytes written" - ) - _done = True - else: - last_error = f"no_data_from_{base}" - logger.warning( - f"DVR recording {recording_id}: no data from " - f"{base} within {_test_window}s, trying next base" - ) - try: - if os.path.exists(temp_ts_path) and os.path.getsize(temp_ts_path) == 0: - os.remove(temp_ts_path) - except FileNotFoundError: - pass - break # Exit reconnection loop - - except (ReadTimeout, ReqConnectionError, ChunkedEncodingError) as e: - if bytes_written > 0: - # Active stream lost — check cancellation before reconnecting - should_exit, is_int, reason = _check_recording_cancelled(recording_id) - if should_exit: - interrupted = is_int - interrupted_reason = reason - if reason == "stopped_by_user": - logger.info( - f"DVR recording {recording_id}: " - f"stopped by user — ending stream" - ) - _done = True + buf = bytearray() + stream = proc.stderr + while True: + byte = stream.read(1) + if not byte: break - - _reconnects += 1 - if _reconnects <= _dvr_max_reconnects: - logger.warning( - f"DVR recording {recording_id}: connection lost " - f"({type(e).__name__}), reconnecting " - f"({_reconnects}/{_dvr_max_reconnects}) " - f"in {_dvr_reconnect_delay}s..." - ) - time.sleep(_dvr_reconnect_delay) - _file_mode = 'ab' + if byte in (b'\r', b'\n'): + if buf: + line = buf.decode('utf-8', errors='replace').strip() + buf.clear() + if line: + tail.append(line) + low = line.lower() + if 'error' in low or 'failed' in low or 'invalid' in low: + logger.warning(f"DVR recording {rec_id} ffmpeg: {line}") + else: + logger.debug(f"DVR recording {rec_id} ffmpeg: {line}") continue + buf.append(byte[0]) + # Safety net: flush absurdly long lines so a malformed + # stream can't grow the buffer without bound. + if len(buf) > 4096: + line = buf.decode('utf-8', errors='replace').strip() + buf.clear() + if line: + tail.append(line) + logger.debug(f"DVR recording {rec_id} ffmpeg: {line}") + # Flush any trailing content + if buf: + line = buf.decode('utf-8', errors='replace').strip() + if line: + tail.append(line) + logger.debug(f"DVR recording {rec_id} ffmpeg: {line}") + except Exception as _de: + logger.debug(f"DVR recording {rec_id}: stderr drain ended: {_de}") - logger.error( - f"DVR recording {recording_id}: max reconnects " - f"({_dvr_max_reconnects}) exceeded — ending recording" - ) - interrupted = True - interrupted_reason = ( - f"stream_interrupted: max reconnects exceeded ({e})" - ) - _done = True - break + if ffmpeg_proc is not None and ffmpeg_proc.stderr is not None: + _stderr_thread = threading.Thread( + target=_drain_ffmpeg_stderr, + args=(ffmpeg_proc, recording_id, _ffmpeg_stderr_tail), + daemon=True, + name=f"dvr-ffmpeg-stderr-{recording_id}", + ) + _stderr_thread.start() - # No data received yet — retry same base before moving on - should_exit, is_int, reason = _check_recording_cancelled(recording_id) - if should_exit: - interrupted = is_int - interrupted_reason = reason - _done = True - break - _reconnects += 1 - if _reconnects <= _dvr_max_reconnects: + end_timestamp = ( + end_time.timestamp() + if hasattr(end_time, 'timestamp') + else time.mktime(end_time.timetuple()) + ) + _stop_poll_interval = 2.0 + _last_stop_poll = time.time() + _ffmpeg_start = time.time() + _first_segment_timeout = 15.0 + _stall_timeout = 60.0 # seconds without new segments → source stream gone + _stream_confirmed = False + _last_seg_count = hls_start_number + _last_new_seg_time = time.time() + + while ffmpeg_proc.poll() is None: + time.sleep(0.5) + now = time.time() + + segs_now = [ + f for f in os.listdir(hls_dir) + if f.startswith("seg_") and f.endswith(".ts") + ] if hls_dir else [] + + # Wait for the first segment to confirm data is flowing + if not _stream_confirmed: + if segs_now: + _stream_confirmed = True + _last_seg_count = len(segs_now) + _last_new_seg_time = now + logger.info( + f"DVR recording {recording_id}: first HLS segment written, stream confirmed" + ) + elif now - _ffmpeg_start > _first_segment_timeout: logger.warning( - f"DVR recording {recording_id}: initial connection " - f"to {base} failed ({type(e).__name__}), retrying " - f"({_reconnects}/{_dvr_max_reconnects}) " - f"in {_dvr_reconnect_delay}s..." + f"DVR recording {recording_id}: no HLS segments produced after " + f"{_first_segment_timeout}s from {base} — stream unavailable" ) - time.sleep(_dvr_reconnect_delay) - continue - last_error = str(e) - logger.warning( - f"DVR recording {recording_id}: base {base} exhausted " - f"retries ({_dvr_max_reconnects}): {e}" - ) - break - - except Exception as e: - last_error = str(e) - logger.warning(f"DVR recording {recording_id}: base {base} failed: {e}") - if bytes_written > 0: - should_exit, is_int, reason = _check_recording_cancelled(recording_id) - if should_exit and reason == "stopped_by_user": - interrupted = False - logger.info( - f"DVR recording {recording_id}: " - f"stopped by user — ending stream" - ) - else: - interrupted = True - interrupted_reason = f"stream_interrupted: {e}" - _done = True + ffmpeg_proc.kill() + try: + ffmpeg_proc.wait(timeout=5) + except subprocess.TimeoutExpired: + pass + last_error = f"no_segments_in_{_first_segment_timeout}s_from_{base}" + ffmpeg_proc = None break - should_exit, is_int, reason = _check_recording_cancelled(recording_id) - if should_exit: - interrupted = is_int - interrupted_reason = reason - _done = True + else: + # Stream was confirmed, detect source stalls (e.g. proxy ghost-kills the client) + if len(segs_now) > _last_seg_count: + _last_seg_count = len(segs_now) + _last_new_seg_time = now + else: + # Also treat a recently-modified segment file as activity. + # With erratic source timestamps, FFmpeg may buffer data inside + # a partially-written segment for longer than hls_time, so the + # segment count won't increase even though data is flowing. + try: + if segs_now and hls_dir: + _newest_mtime = max( + os.path.getmtime(os.path.join(hls_dir, f)) + for f in segs_now + ) + if _newest_mtime > _last_new_seg_time: + _last_new_seg_time = _newest_mtime + except Exception: + pass + if now - _last_new_seg_time > _stall_timeout: + logger.warning( + f"DVR recording {recording_id}: no new HLS segments for " + f"{_stall_timeout:.0f}s — source stream stalled, stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + break + + # Duration check — SIGINT lets FFmpeg write #EXT-X-ENDLIST cleanly + if now >= end_timestamp: + logger.info( + f"DVR recording {recording_id}: scheduled end time reached, stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() break - if _done: - break + # Periodic DB poll: stop, delete, end_time extension + if now - _last_stop_poll >= _stop_poll_interval: + _last_stop_poll = now + try: + _sc = Recording.objects.filter( + id=recording_id + ).only("custom_properties", "end_time").first() + if _sc is None: + logger.info( + f"DVR recording {recording_id}: deleted — stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + interrupted = False + break + if (_sc.custom_properties or {}).get("status") == "stopped": + logger.info( + f"DVR recording {recording_id}: stop requested — stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + break + # Handle end_time extension — just update the deadline, no restart needed + try: + new_end = _sc.end_time + if new_end is not None: + if _tz.is_naive(new_end): + new_end = _tz.make_aware(new_end) + new_ts = new_end.timestamp() + if new_ts > end_timestamp: + logger.info( + f"DVR recording {recording_id}: end_time extended to {new_end}" + ) + end_timestamp = new_ts + except Exception: + pass + except Exception: + pass - if chosen_base is None and bytes_written == 0: + # If FFmpeg exited after confirming the stream, log the exit code so we know why it stopped. + if _stream_confirmed and ffmpeg_proc is not None: + _exit_code = ffmpeg_proc.poll() + if _exit_code is not None and _exit_code != 0: + logger.warning( + f"DVR recording {recording_id}: FFmpeg exited unexpectedly (rc={_exit_code}) " + f"after stream was confirmed — source stream likely disconnected" + ) + + # If FFmpeg exited without the stream being confirmed, log the tail of + # captured stderr for diagnosis. Lines were already logged live by the + # drain thread; this surfaces the recent context in a single message. + elif not _stream_confirmed and ffmpeg_proc is not None: + try: + _exit_code = ffmpeg_proc.poll() + last_error = f"rc={_exit_code} from {base}" + _tail_text = "\n".join(_ffmpeg_stderr_tail) + if _tail_text: + logger.warning( + f"DVR recording {recording_id}: FFmpeg exited (rc={_exit_code}) " + f"for {base} without producing segments.\nFFmpeg stderr tail:\n{_tail_text[-1000:]}" + ) + else: + logger.warning( + f"DVR recording {recording_id}: FFmpeg exited (rc={_exit_code}) " + f"for {base} without producing segments (no stderr output)" + ) + except Exception: + pass + + # Ensure FFmpeg has fully exited (covers cases where the loop broke early) + if ffmpeg_proc is not None and ffmpeg_proc.poll() is None: + try: + ffmpeg_proc.wait(timeout=5) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + + if not _stream_confirmed and not interrupted: interrupted = True - interrupted_reason = f"no_stream_data: {last_error or 'all_bases_failed'}" + interrupted_reason = f"no_stream_data: {last_error or 'ffmpeg_failed'}" - # If no bytes were written at all, check whether this was a deliberate stop or a - # genuine failure. The exception handler above already sets interrupted=False when - # it detects "stopped" status, but do not override that decision here. + # Measure bytes from HLS segment files + try: + bytes_written = sum( + os.path.getsize(os.path.join(hls_dir, f)) + for f in os.listdir(hls_dir) + if f.startswith("seg_") and f.endswith(".ts") + ) if hls_dir and os.path.isdir(hls_dir) else 0 + except Exception: + bytes_written = 0 if bytes_written == 0 and not interrupted: _deliberately_stopped = False try: @@ -2178,19 +2266,19 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # After the loop, the file and response are closed automatically. logger.info(f"Finished recording for channel {channel.name}") - # Log system event for recording end - try: - from core.utils import log_system_event - log_system_event( - 'recording_end', - channel_id=channel.uuid, - channel_name=channel.name, - recording_id=recording_id, - interrupted=interrupted, - bytes_written=bytes_written - ) - except Exception as e: - logger.error(f"Could not log recording end event: {e}") + # Log system event for recording end + try: + from core.utils import log_system_event + log_system_event( + 'recording_end', + channel_id=channel.uuid, + channel_name=channel.name, + recording_id=recording_id, + interrupted=interrupted, + bytes_written=bytes_written + ) + except Exception as e: + logger.error(f"Could not log recording end event: {e}") # If the Recording was deleted (cancelled by user), skip post-processing recording_cancelled = not Recording.objects.filter(id=recording_id).exists() @@ -2199,261 +2287,164 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # Clean up all artifacts for the cancelled recording, # including any pre-restart .ts segments from server recovery. # Use the in-memory recording_obj since the DB row is already deleted. - _cancel_cleanup = [temp_ts_path, final_path] - _cancel_cp = (recording_obj.custom_properties or {}) if recording_obj else {} - _cancel_cleanup.extend(_cancel_cp.get("_pre_restart_ts_paths", [])) - for _cleanup_path in _cancel_cleanup: - if not _cleanup_path: - continue + import shutil as _shutil + if hls_dir and os.path.isdir(hls_dir): try: - os.remove(_cleanup_path) - logger.info(f"Cleaned up cancelled recording artifact: {_cleanup_path}") - except FileNotFoundError: - pass + _shutil.rmtree(hls_dir) + logger.info(f"Cleaned up cancelled recording HLS directory: {hls_dir}") + except Exception as _e: + logger.warning(f"Failed to remove HLS directory {hls_dir}: {_e}") + if final_path: + try: + if os.path.exists(final_path): + os.remove(final_path) + logger.info(f"Cleaned up cancelled recording MKV placeholder: {final_path}") except Exception: pass return - # Concatenate pre-restart .ts segments with the current segment. - # Instead of creating an intermediate combined.ts and then remuxing to - # MKV (which loses timestamp boundary info and causes playback freezes - # at the splice point), go directly from the concat list → MKV. - # This lets ffmpeg's MKV muxer see each segment boundary and write - # correct cue points / clusters for seamless seeking. - _concat_did_remux = False - try: - _rec_obj_for_concat = Recording.objects.filter(id=recording_id).only("custom_properties").first() - _concat_cp = (_rec_obj_for_concat.custom_properties or {}) if _rec_obj_for_concat else {} - pre_restart_segments = _concat_cp.get("_pre_restart_ts_paths", []) - # Filter to segments that still exist on disk and have data - def _has_data(p): - try: - return os.stat(p).st_size > 0 - except OSError: - return False - pre_restart_segments = [p for p in pre_restart_segments if p and _has_data(p)] - if pre_restart_segments and temp_ts_path and os.path.exists(temp_ts_path): - all_segments = pre_restart_segments + [temp_ts_path] - concat_list_path = temp_ts_path + ".concat.txt" - try: - with open(concat_list_path, "w") as cl: - for seg in all_segments: - cl.write(f"file '{seg}'\n") + # --- Post-processing: concat HLS segments → final MKV --- + remux_success = False + hls_m3u8 = os.path.join(hls_dir, "index.m3u8") if hls_dir else None - # Direct concat → MKV in a single pass. - # -reset_timestamps 1 tells the concat demuxer to reset - # timestamps at each segment boundary, eliminating the - # discontinuity that causes playback to freeze at the - # splice point. - concat_result = subprocess.run( - [ - "ffmpeg", "-y", - "-fflags", "+genpts+igndts+discardcorrupt", - "-err_detect", "ignore_err", - "-f", "concat", "-safe", "0", - "-segment_time_metadata", "1", - "-i", concat_list_path, - "-reset_timestamps", "1", - "-map", "0", - "-c", "copy", - final_path, - ], - stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, - ) - if concat_result.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0: - _concat_did_remux = True - # Clean up individual TS segments (including current) - for seg in all_segments: - try: - os.remove(seg) - except OSError: - pass - logger.info( - f"DVR recording {recording_id}: concat→MKV succeeded — " - f"{len(all_segments)} segments → {os.path.basename(final_path)} " - f"({os.path.getsize(final_path):,} bytes)" - ) - else: - logger.warning( - f"DVR recording {recording_id}: direct concat→MKV failed " - f"(rc={concat_result.returncode}), falling back to " - f"normal remux with current segment only. " - f"stderr: {(concat_result.stderr or '')[:500]}" - ) - finally: - try: - os.remove(concat_list_path) - except OSError: - pass - # Clear the pre-restart paths from custom_properties - if _rec_obj_for_concat: - _ccp = _rec_obj_for_concat.custom_properties or {} - _ccp.pop("_pre_restart_ts_paths", None) - _ccp.pop("interrupted_reason", None) - _rec_obj_for_concat.custom_properties = _ccp - _rec_obj_for_concat.save(update_fields=["custom_properties"]) - except Exception as e: - logger.warning( - f"DVR recording {recording_id}: segment concatenation error " - f"({type(e).__name__}: {e}), proceeding with current segment only." - ) - - # Remux TS to MKV container with retry for transient I/O errors - # (Skip if concat already produced the final MKV directly.) - remux_success = _concat_did_remux - existing_mkv_size = 0 - try: - if final_path and os.path.exists(final_path): - existing_mkv_size = os.path.getsize(final_path) - except OSError: - pass - for _remux_attempt in range(_dvr_remux_max_retries): - if remux_success: - break + def _get_hls_segments(m3u8_path, seg_dir): + """Return ordered segment paths from an HLS m3u8 playlist.""" + segs = [] try: - if temp_ts_path and os.path.exists(temp_ts_path): - # First attempt: Direct TS to MKV remux - result = subprocess.run([ + with open(m3u8_path) as _f: + for _line in _f: + _line = _line.strip() + if _line and not _line.startswith('#'): + sp = os.path.join(seg_dir, _line) if not os.path.isabs(_line) else _line + if os.path.exists(sp): + segs.append(sp) + except Exception as _e: + logger.warning(f"DVR recording {recording_id}: failed to parse m3u8: {_e}") + return segs + + segments = ( + _get_hls_segments(hls_m3u8, hls_dir) + if (hls_m3u8 and os.path.exists(hls_m3u8)) + else [] + ) + if not segments and hls_dir and os.path.isdir(hls_dir): + # Fallback: sort all segment files by name if m3u8 is missing or empty + try: + segments = sorted( + os.path.join(hls_dir, f) + for f in os.listdir(hls_dir) + if f.startswith("seg_") and f.endswith(".ts") + ) + except Exception: + segments = [] + + if segments: + concat_list_path = os.path.join(hls_dir, "concat.txt") + try: + with open(concat_list_path, "w") as _cl: + for seg in segments: + _cl.write(f"file '{seg}'\n") + + concat_result = subprocess.run( + [ "ffmpeg", "-y", - "-fflags", "+genpts+igndts+discardcorrupt", # Regenerate timestamps, ignore DTS - "-err_detect", "ignore_err", # Ignore minor stream errors - "-i", temp_ts_path, - "-map", "0", # Map all streams + "-f", "concat", "-safe", "0", + "-i", concat_list_path, "-c", "copy", - final_path - ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - # Check if FFmpeg succeeded (return code 0) and output file is valid - if result.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0: - remux_success = True - logger.info(f"Direct TS→MKV remux succeeded for {os.path.basename(final_path)}") - else: - # Direct remux failed - try fallback: TS → MP4 → MKV to fix timestamps - logger.warning(f"Direct TS→MKV remux failed (return code: {result.returncode}), trying fallback TS→MP4→MKV") - - # Clean up partial/failed MKV - try: - if os.path.exists(final_path): - os.remove(final_path) - except Exception: - pass - - # Step 1: TS → MP4 (MP4 container handles broken timestamps better) - temp_mp4_path = os.path.splitext(temp_ts_path)[0] + ".mp4" - result_mp4 = subprocess.run([ - "ffmpeg", "-y", - "-fflags", "+genpts+igndts+discardcorrupt", - "-err_detect", "ignore_err", - "-i", temp_ts_path, - "-map", "0", - "-c", "copy", - temp_mp4_path - ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - if result_mp4.returncode == 0 and os.path.exists(temp_mp4_path) and os.path.getsize(temp_mp4_path) > 0: - logger.info(f"TS→MP4 conversion succeeded, now converting MP4→MKV") - - # Step 2: MP4 → MKV (clean timestamps from MP4) - result_mkv = subprocess.run([ - "ffmpeg", "-y", - "-i", temp_mp4_path, - "-map", "0", - "-c", "copy", - final_path - ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - if result_mkv.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0: - remux_success = True - logger.info(f"Fallback TS→MP4→MKV remux succeeded for {os.path.basename(final_path)}") - else: - logger.error(f"MP4→MKV conversion failed (return code: {result_mkv.returncode})") - - # Clean up temp MP4 - try: - if os.path.exists(temp_mp4_path): - os.remove(temp_mp4_path) - except Exception: - pass - else: - logger.error(f"TS→MP4 conversion failed (return code: {result_mp4.returncode})") - - # Sanity-check the remuxed file. Two checks: - # 1. If a pre-existing MKV was overwritten, reject a - # file that is drastically smaller (duplicate-task - # overwrite protection). - # 2. If the MKV is smaller than the .ts source, the - # remux likely produced a corrupt or truncated file. - if remux_success: - try: - new_size = os.path.getsize(final_path) - ts_size = os.path.getsize(temp_ts_path) if temp_ts_path and os.path.exists(temp_ts_path) else 0 - reject = False - if existing_mkv_size > 0 and new_size < existing_mkv_size * 0.5: - logger.error( - f"DVR recording {recording_id}: new MKV " - f"({new_size:,} bytes) is less than 50%% of " - f"the previous MKV ({existing_mkv_size:,} bytes) " - f"— refusing to overwrite. Keeping .ts for " - f"manual recovery." - ) - reject = True - elif ts_size > 0 and new_size < ts_size * 0.1: - logger.error( - f"DVR recording {recording_id}: remuxed MKV " - f"({new_size:,} bytes) is less than 10%% of " - f"the source TS ({ts_size:,} bytes) — likely " - f"corrupt. Keeping .ts for manual recovery." - ) - reject = True - if reject: - remux_success = False - try: - os.remove(final_path) - except OSError: - pass - except OSError: - pass - - # Clean up temp TS file only on successful remux - if remux_success: - try: - os.remove(temp_ts_path) - logger.debug(f"Cleaned up temp TS file: {temp_ts_path}") - except Exception as e: - logger.warning(f"Failed to remove temp TS file: {e}") - else: - # Keep TS file for debugging/manual recovery if remux failed - logger.warning(f"Remux failed - keeping temp TS file for recovery: {temp_ts_path}") - # Clean up any partial MKV - try: - if os.path.exists(final_path): - os.remove(final_path) - logger.debug(f"Cleaned up partial MKV file: {final_path}") - except Exception: - pass - break # Completed (success or deterministic failure) - - except (OSError, subprocess.SubprocessError) as e: - # Clean up partial output before potential retry - try: - if os.path.exists(final_path): - os.remove(final_path) - except Exception: - pass - if _remux_attempt + 1 < _dvr_remux_max_retries: - _wait = _dvr_remux_retry_interval * (2 ** _remux_attempt) - logger.warning( - f"DVR recording {recording_id}: remux failed " - f"({type(e).__name__}), retrying in {_wait}s " - f"({_remux_attempt + 1}/{_dvr_remux_max_retries})..." + final_path, + ], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + if ( + concat_result.returncode == 0 + and os.path.exists(final_path) + and os.path.getsize(final_path) > 0 + ): + remux_success = True + logger.info( + f"DVR recording {recording_id}: HLS→MKV concat succeeded — " + f"{len(segments)} segments → {os.path.basename(final_path)} " + f"({os.path.getsize(final_path):,} bytes)" ) - time.sleep(_wait) + # Update DB so *new* client requests go to /file/ (the final MKV) + # rather than the soon-to-be-removed /hls/ endpoint. Important: + # we do NOT clear _hls_dir here, active viewers still in the + # 20s heartbeat window need it to keep fetching .ts segments + # during the viewer-wait grace period below. _hls_dir is + # cleared only after the directory is actually removed. + try: + _pre_rmtree_rec = Recording.objects.filter(id=recording_id).first() + if _pre_rmtree_rec: + _pre_rmtree_cp = _pre_rmtree_rec.custom_properties or {} + _pre_rmtree_cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/" + _pre_rmtree_cp["output_file_url"] = _pre_rmtree_cp["file_url"] + _pre_rmtree_rec.custom_properties = _pre_rmtree_cp + _pre_rmtree_rec.save(update_fields=["custom_properties"]) + except Exception as _pre_e: + logger.warning(f"DVR recording {recording_id}: pre-rmtree DB update failed: {_pre_e}") + import shutil as _shutil + # Wait for active HLS viewers to finish downloading segments + # before deleting the directory. The hls endpoint refreshes + # _hls_viewer_key on every .ts request with a 20s TTL, so the + # key expiring naturally means no segment was fetched in the + # last 20 seconds (client has stopped). We loop until the key + # is gone with a 4-hour hard safety cap in case Redis gets stuck. + _viewer_wait_start = time.time() + _safety_timeout = 14400 # 4 hours absolute maximum + try: + from core.utils import RedisClient as _RC + _rv = _RC.get_client(max_retries=1, retry_interval=0) + if _rv and _rv.exists(_hls_viewer_key): + logger.info( + f"DVR recording {recording_id}: active HLS viewer detected, " + f"deferring HLS directory cleanup until client disconnects" + ) + while _rv.exists(_hls_viewer_key): + if time.time() - _viewer_wait_start > _safety_timeout: + logger.warning( + f"DVR recording {recording_id}: viewer wait safety timeout " + f"({_safety_timeout}s) reached, proceeding with HLS directory cleanup" + ) + break + time.sleep(2) + logger.info( + f"DVR recording {recording_id}: viewer wait complete after " + f"{time.time() - _viewer_wait_start:.1f}s" + ) + except Exception as _ve: + logger.debug(f"DVR recording {recording_id}: viewer wait check failed: {_ve}") + try: + _shutil.rmtree(hls_dir) + logger.debug(f"Cleaned up HLS directory: {hls_dir}") + except Exception as _e: + logger.warning(f"Failed to remove HLS directory {hls_dir}: {_e}") + # Now that the HLS dir is gone, clear _hls_dir from custom_properties + # so the hls endpoint will redirect (.m3u8) or 404 (.ts) cleanly. + try: + _post_rmtree_rec = Recording.objects.filter(id=recording_id).first() + if _post_rmtree_rec: + _post_rmtree_cp = _post_rmtree_rec.custom_properties or {} + if "_hls_dir" in _post_rmtree_cp: + _post_rmtree_cp.pop("_hls_dir", None) + _post_rmtree_rec.custom_properties = _post_rmtree_cp + _post_rmtree_rec.save(update_fields=["custom_properties"]) + except Exception as _post_e: + logger.warning(f"DVR recording {recording_id}: post-rmtree DB update failed: {_post_e}") else: - logger.warning( - f"DVR recording {recording_id}: remux failed " - f"after {_dvr_remux_max_retries} attempts: {e}. " - f"Keeping .ts for manual recovery: {temp_ts_path}" + logger.error( + f"DVR recording {recording_id}: HLS→MKV concat failed " + f"(rc={concat_result.returncode}). Keeping HLS segments for recovery. " + f"stderr: {(concat_result.stderr or '')[:500]}" ) + except Exception as _ce: + logger.error(f"DVR recording {recording_id}: concat exception: {_ce}") + finally: + try: + os.remove(concat_list_path) + except OSError: + pass + else: + logger.warning(f"DVR recording {recording_id}: no HLS segments found. Nothing to concat") # Persist final metadata to Recording (status, ended_at, and stream stats if available) try: @@ -2465,6 +2456,11 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): cp = recording_obj.custom_properties or {} cp["ended_at"] = str(datetime.now()) + # Restore file_url to the permanent MKV endpoint now that recording has ended. + # During recording it pointed to /hls/index.m3u8; clients should use /file/ hereafter. + cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/" + cp["output_file_url"] = cp["file_url"] + # Final status priority: stopped > completed > interrupted. # "stopped" is set by the stop endpoint before stream teardown, so # refresh_from_db() above guarantees it is visible here. @@ -2591,7 +2587,7 @@ def recover_recordings_on_startup(): from django.utils import timezone from .models import Recording from core.utils import RedisClient - from .signals import schedule_recording_task + from .signals import schedule_recording_task, revoke_task redis = RedisClient.get_client() if redis: @@ -2637,14 +2633,12 @@ def recover_recordings_on_startup(): # Preserve the pre-restart .ts segment path so run_recording # can concatenate it with the resumed segment later. - old_ts = cp.get("_temp_file_path") - if old_ts and os.path.exists(old_ts) and os.path.getsize(old_ts) > 0: - prior_segments = cp.get("_pre_restart_ts_paths", []) - prior_segments.append(old_ts) - cp["_pre_restart_ts_paths"] = prior_segments + hls_dir_path = cp.get("_hls_dir") + if hls_dir_path and os.path.isdir(hls_dir_path): + existing_segs = [f for f in os.listdir(hls_dir_path) if f.endswith(".ts")] logger.info( f"recover_recordings_on_startup: recording {rec.id} — " - f"preserving pre-restart TS segment: {old_ts}" + f"HLS dir has {len(existing_segs)} existing segment(s), will resume" ) rec.custom_properties = cp @@ -2687,38 +2681,86 @@ def recover_recordings_on_startup(): for rec in expired: try: cp = rec.custom_properties or {} - ts_path = cp.get("_temp_file_path") + hls_dir_path = cp.get("_hls_dir") mkv_path = cp.get("file_path") - if ts_path and os.path.exists(ts_path) and os.path.getsize(ts_path) > 0 and mkv_path: - logger.info( - f"recover_recordings_on_startup: recording {rec.id} expired " - f"during downtime — remuxing partial TS ({os.path.getsize(ts_path):,} bytes)" - ) - os.makedirs(os.path.dirname(mkv_path), exist_ok=True) - result = subprocess.run( - [ - "ffmpeg", "-y", - "-fflags", "+genpts+igndts+discardcorrupt", - "-err_detect", "ignore_err", - "-i", ts_path, "-map", "0", "-c", "copy", mkv_path, - ], - stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, - ) - if result.returncode == 0 and os.path.exists(mkv_path) and os.path.getsize(mkv_path) > 0: - cp["status"] = "interrupted" - cp["interrupted_reason"] = "server_restarted_after_end" - cp["remux_success"] = True + if hls_dir_path and os.path.isdir(hls_dir_path) and mkv_path: + import shutil as _shutil + # Parse m3u8 for ordered segment list; fall back to sorted filenames + _m3u8 = os.path.join(hls_dir_path, "index.m3u8") + _segs = [] + if os.path.exists(_m3u8): try: - os.remove(ts_path) - except OSError: + with open(_m3u8) as _f: + for _l in _f: + _l = _l.strip() + if _l and not _l.startswith('#'): + _sp = os.path.join(hls_dir_path, _l) if not os.path.isabs(_l) else _l + if os.path.exists(_sp): + _segs.append(_sp) + except Exception: pass - logger.info(f"recover_recordings_on_startup: recording {rec.id} remuxed successfully") + if not _segs: + _segs = sorted( + os.path.join(hls_dir_path, f) + for f in os.listdir(hls_dir_path) + if f.startswith("seg_") and f.endswith(".ts") + ) + + if _segs: + logger.info( + f"recover_recordings_on_startup: recording {rec.id} expired " + f"during downtime, concat {len(_segs)} HLS segment(s) \u2192 MKV" + ) + os.makedirs(os.path.dirname(mkv_path), exist_ok=True) + _concat_txt = os.path.join(hls_dir_path, "concat.txt") + try: + with open(_concat_txt, "w") as _cl: + for _s in _segs: + _cl.write(f"file '{_s}'\n") + _res = subprocess.run( + [ + "ffmpeg", "-y", + "-f", "concat", "-safe", "0", + "-i", _concat_txt, + "-c", "copy", mkv_path, + ], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + if _res.returncode == 0 and os.path.exists(mkv_path) and os.path.getsize(mkv_path) > 0: + cp["status"] = "interrupted" + cp["interrupted_reason"] = "server_restarted_after_end" + cp["remux_success"] = True + try: + _shutil.rmtree(hls_dir_path) + except OSError: + pass + logger.info( + f"recover_recordings_on_startup: recording {rec.id} HLS\u2192MKV concat succeeded" + ) + else: + cp["status"] = "interrupted" + cp["interrupted_reason"] = "server_restarted_after_end" + cp["remux_success"] = False + logger.warning( + f"recover_recordings_on_startup: recording {rec.id} concat failed, keeping HLS dir" + ) + except Exception as _ce: + cp["status"] = "interrupted" + cp["interrupted_reason"] = "server_restarted_after_end" + cp["remux_success"] = False + logger.warning( + f"recover_recordings_on_startup: recording {rec.id} concat error: {_ce}" + ) + finally: + try: + os.remove(_concat_txt) + except OSError: + pass else: cp["status"] = "interrupted" cp["interrupted_reason"] = "server_restarted_after_end" cp["remux_success"] = False - logger.warning(f"recover_recordings_on_startup: recording {rec.id} remux failed, keeping .ts") else: cp["status"] = "interrupted" cp["interrupted_reason"] = "server_restarted_after_end" diff --git a/apps/channels/tests/test_dvr_port_resolution.py b/apps/channels/tests/test_dvr_port_resolution.py index c7373959..61f74454 100644 --- a/apps/channels/tests/test_dvr_port_resolution.py +++ b/apps/channels/tests/test_dvr_port_resolution.py @@ -2,58 +2,68 @@ import os from django.test import SimpleTestCase from unittest.mock import patch -from apps.channels.tasks import build_dvr_candidates +from apps.channels.tasks import get_dvr_stream_base_url -class DVRPortResolutionTests(SimpleTestCase): +class DVRStreamBaseURLTests(SimpleTestCase): """ - Tests that DVR recording candidate URLs respect the DISPATCHARR_PORT - environment variable instead of hardcoding port 9191. + Tests that get_dvr_stream_base_url() returns the correct single URL + for each deployment mode. """ - @patch.dict(os.environ, {'REDIS_HOST': 'redis'}, clear=True) - def test_default_port_uses_9191(self): - """Without DISPATCHARR_PORT set, candidates default to 9191.""" - candidates = build_dvr_candidates() - self.assertIn('http://web:9191', candidates) - self.assertIn('http://localhost:9191', candidates) + @patch.dict(os.environ, {}, clear=True) + def test_aio_default_uses_localhost_5656(self): + """AIO mode (default) reaches uwsgi directly on loopback port 5656.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://127.0.0.1:5656') - @patch.dict(os.environ, {'DISPATCHARR_PORT': '8080', 'REDIS_HOST': 'redis'}, clear=True) - def test_custom_port_reflected_in_candidates(self): - """DISPATCHARR_PORT=8080 replaces all hardcoded 9191 references.""" - candidates = build_dvr_candidates() - self.assertIn('http://web:8080', candidates) - self.assertIn('http://localhost:8080', candidates) - self.assertNotIn('http://web:9191', candidates) - self.assertNotIn('http://localhost:9191', candidates) + @patch.dict(os.environ, {'DISPATCHARR_ENV': 'aio'}, clear=True) + def test_aio_explicit_uses_localhost_5656(self): + """Explicit DISPATCHARR_ENV=aio also uses loopback port 5656.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://127.0.0.1:5656') + + @patch.dict(os.environ, {'DISPATCHARR_ENV': 'dev'}, clear=True) + def test_dev_mode_uses_localhost_5656(self): + """Dev mode shares the container with uwsgi — uses loopback port 5656.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://127.0.0.1:5656') + + @patch.dict(os.environ, {'DISPATCHARR_ENV': 'modular', 'DISPATCHARR_PORT': '9191'}, clear=True) + def test_modular_uses_web_service_name(self): + """Modular mode uses the 'web' Docker service name by default.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://web:9191') + + @patch.dict(os.environ, {'DISPATCHARR_ENV': 'modular', 'DISPATCHARR_PORT': '8080'}, clear=True) + def test_modular_custom_port(self): + """Modular mode respects DISPATCHARR_PORT.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://web:8080') @patch.dict(os.environ, { - 'DISPATCHARR_PORT': '7777', - 'DISPATCHARR_ENV': 'dev', - 'REDIS_HOST': 'redis', + 'DISPATCHARR_ENV': 'modular', + 'DISPATCHARR_PORT': '9191', + 'DISPATCHARR_WEB_HOST': 'dispatcharr_web', }, clear=True) - def test_dev_mode_includes_5656_and_custom_port(self): - """Dev mode includes both uwsgi internal port (5656) and custom port.""" - candidates = build_dvr_candidates() - self.assertIn('http://127.0.0.1:5656', candidates) - self.assertIn('http://127.0.0.1:7777', candidates) + def test_modular_custom_web_host(self): + """DISPATCHARR_WEB_HOST overrides the default 'web' service name.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://dispatcharr_web:9191') @patch.dict(os.environ, { 'DISPATCHARR_INTERNAL_TS_BASE_URL': 'http://custom:1234', - 'REDIS_HOST': 'redis', + 'DISPATCHARR_ENV': 'modular', }, clear=True) - def test_explicit_override_is_first(self): - """DISPATCHARR_INTERNAL_TS_BASE_URL should be the first candidate.""" - candidates = build_dvr_candidates() - self.assertEqual(candidates[0], 'http://custom:1234') + def test_explicit_override_always_wins(self): + """DISPATCHARR_INTERNAL_TS_BASE_URL takes priority over all other settings.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://custom:1234') @patch.dict(os.environ, { - 'DISPATCHARR_PORT': '3000', - 'DISPATCHARR_INTERNAL_API_BASE': 'http://myhost:4000', - 'REDIS_HOST': 'redis', + 'DISPATCHARR_INTERNAL_TS_BASE_URL': 'http://custom:1234/', }, clear=True) - def test_internal_api_base_overrides_web_fallback(self): - """DISPATCHARR_INTERNAL_API_BASE replaces the http://web:{port} default.""" - candidates = build_dvr_candidates() - self.assertIn('http://myhost:4000', candidates) - self.assertNotIn('http://web:3000', candidates) + def test_explicit_override_strips_trailing_slash(self): + """Trailing slash is stripped from DISPATCHARR_INTERNAL_TS_BASE_URL.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://custom:1234') diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 547ac249..58b6e4f5 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -118,8 +118,12 @@ services: - DISPATCHARR_ENV=modular # Internal Service Communication - # Must match the web service port for DVR recording and internal API calls + # Celery uses these to reach the web container for DVR recording. + # DISPATCHARR_PORT must match the port exposed by the web service. + # DISPATCHARR_WEB_HOST defaults to "web" (the service name above). + # Only set DISPATCHARR_WEB_HOST if you rename the web service in this file. - DISPATCHARR_PORT=9191 + #- DISPATCHARR_WEB_HOST=web # PostgreSQL — must match web service settings - POSTGRES_HOST=db From c9566725870100570059d7f3c91cf06d43a230ae Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 26 Apr 2026 09:34:20 -0500 Subject: [PATCH 02/55] Enhancement: Route long-running DVR recordings to a dedicated queue. --- dispatcharr/celery.py | 5 +++++ docker/entrypoint.celery.sh | 6 +++++- docker/uwsgi.debug.ini | 7 ++++--- docker/uwsgi.dev.ini | 7 ++++--- docker/uwsgi.ini | 7 ++++--- 5 files changed, 22 insertions(+), 10 deletions(-) diff --git a/dispatcharr/celery.py b/dispatcharr/celery.py index 8ccb3c33..0d63f063 100644 --- a/dispatcharr/celery.py +++ b/dispatcharr/celery.py @@ -49,6 +49,11 @@ app.conf.update( worker_task_log_format='%(asctime)s %(levelname)s %(task_name)s: %(message)s', ) +# Route long-running DVR recordings to a dedicated `dvr` queue consumed by a thread-pool worker. +app.conf.task_routes = { + 'apps.channels.tasks.run_recording': {'queue': 'dvr'}, +} + # Add memory cleanup after task completion @task_postrun.connect # Use the imported signal def cleanup_task_memory(**kwargs): diff --git a/docker/entrypoint.celery.sh b/docker/entrypoint.celery.sh index b38b6ac3..c795d781 100644 --- a/docker/entrypoint.celery.sh +++ b/docker/entrypoint.celery.sh @@ -67,4 +67,8 @@ NICE_LEVEL="${CELERY_NICE_LEVEL:-5}" if [ "$NICE_LEVEL" -lt 0 ] 2>/dev/null; then echo "Warning: CELERY_NICE_LEVEL=$NICE_LEVEL is negative, requires SYS_NICE capability" fi -nice -n "$NICE_LEVEL" celery -A dispatcharr worker -l info --autoscale=6,1 + +# DVR worker: thread pool for the long-running, I/O-bound run_recording task. +nice -n "$NICE_LEVEL" celery -A dispatcharr worker -Q dvr -n dvr@%h --pool=threads --concurrency=20 -l info & +# Default prefork worker: every queue except `dvr`. +nice -n "$NICE_LEVEL" celery -A dispatcharr worker -Q celery -n default@%h --autoscale=6,1 -l info diff --git a/docker/uwsgi.debug.ini b/docker/uwsgi.debug.ini index d2847335..bc21275d 100644 --- a/docker/uwsgi.debug.ini +++ b/docker/uwsgi.debug.ini @@ -7,9 +7,10 @@ exec-before = python /app/scripts/wait_for_redis.py ; Start Redis first attach-daemon = redis-server -; Then start other services with configurable nice level (default: 5 for low priority) -; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose -attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1 +; Default prefork worker: every queue except `dvr`. +attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q celery -n default@%%h --autoscale=6,1 +; DVR worker: thread pool for the long-running, I/O-bound run_recording task. +attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q dvr -n dvr@%%h --pool=threads --concurrency=20 attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr beat attach-daemon = daphne -b 0.0.0.0 -p 8001 dispatcharr.asgi:application attach-daemon = cd /app/frontend && npm run dev diff --git a/docker/uwsgi.dev.ini b/docker/uwsgi.dev.ini index 51aae9a5..dd799500 100644 --- a/docker/uwsgi.dev.ini +++ b/docker/uwsgi.dev.ini @@ -9,9 +9,10 @@ exec-pre = python /app/scripts/wait_for_redis.py ; Start Redis first attach-daemon = redis-server --protected-mode no -; Then start other services with configurable nice level (default: 5 for low priority) -; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose -attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1 +; Default prefork worker: every queue except `dvr`. +attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q celery -n default@%%h --autoscale=6,1 +; DVR worker: thread pool for the long-running, I/O-bound run_recording task. +attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q dvr -n dvr@%%h --pool=threads --concurrency=20 attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr beat attach-daemon = daphne -b 0.0.0.0 -p 8001 dispatcharr.asgi:application attach-daemon = cd /app/frontend && npm run dev diff --git a/docker/uwsgi.ini b/docker/uwsgi.ini index 920bac48..3972c3d9 100644 --- a/docker/uwsgi.ini +++ b/docker/uwsgi.ini @@ -9,9 +9,10 @@ exec-pre = python /app/scripts/wait_for_redis.py ; Start Redis first attach-daemon = redis-server -; Then start other services with configurable nice level (default: 5 for low priority) -; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose -attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1 +; Default prefork worker: every queue except `dvr`. +attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q celery -n default@%%h --autoscale=6,1 +; DVR worker: thread pool for the long-running, I/O-bound run_recording task. +attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q dvr -n dvr@%%h --pool=threads --concurrency=20 attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr beat attach-daemon = daphne -b 0.0.0.0 -p 8001 dispatcharr.asgi:application From 76f2f2abc49df6d8eb13d59ff2159ed9f8ba5248 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 26 Apr 2026 10:21:40 -0500 Subject: [PATCH 03/55] changelog: Update changelog for DVR enahncements --- CHANGELOG.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 38c590bd..83cf0f3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Watch DVR recordings live while they are still recording**: `run_recording` has been refactored from a single TS file capture to an FFmpeg HLS segmentation pipeline. While recording is in progress, the `file_url` exposed on the recording points to a new `/api/channels/recordings/{id}/hls/index.m3u8` endpoint instead of the eventual MKV path, so any HLS-capable client (the built-in web player, VLC, infuse, channels DVR clients, etc.) can join the stream at any time and watch from the live edge. When the recording ends, the segments are concatenated into the final MKV, `file_url` is updated to point to the existing `/file/` endpoint, and the HLS working directory is cleaned up. Hitting `/file/` while a recording is still in progress now redirects to the HLS playlist rather than 404, and hitting `/hls/index.m3u8` after the recording has finalized redirects to `/file/`, so URLs cached by clients in either form continue to work across the transition. +- New HLS playback endpoint (`RecordingViewSet.hls`) serves `.m3u8` and `.ts` files out of the recording's working directory with path traversal protection (`os.path.realpath` containment check) and `AllowAny` permissions consistent with the existing `/file/` endpoint. Playlist files are rewritten on the fly so each segment line becomes an absolute URL routed back through this endpoint, preserving authentication and path isolation. +- Explicit `path('recordings//hls/', ...)` route registered in `apps/channels/api_urls.py` _before_ `router.urls`. DRF's `DefaultRouter` appends a mandatory trailing slash to every URL it generates, but HLS players request segments by their natural filenames (`seg_00001.ts`) without a trailing slash, so the explicit route is required for the player to ever reach the view. +- `DISPATCHARR_WEB_HOST` environment variable for modular deployments. The Celery container needs to reach the uWSGI / web container to fetch HLS segments while recording (FFmpeg pulls from the public stream URL, the same way any other client does). `get_dvr_stream_base_url()` resolves the base URL deterministically: AIO / dev / debug containers reach uWSGI on `127.0.0.1:$DISPATCHARR_PORT` since Celery and uWSGI share the container, while modular deployments use `http://$DISPATCHARR_WEB_HOST:$DISPATCHARR_PORT` and default `DISPATCHARR_WEB_HOST` to `web` (the compose service name). Documented as a commented-out override in `docker/docker-compose.yml`. +- HLS viewer heartbeat. Every `.ts` request refreshes a Redis key `dvr:hls_viewer:{id}` with a 20 second TTL. After concat succeeds, the recording task waits for that key to expire naturally before deleting the HLS working directory, so an actively-watching client is never cut off mid-segment when the recording ends. Cleanup happens within 20 seconds of the last client stopping, with a 4 hour safety cap as a guard against a stuck Redis state. - **VOD start/stop notifications**: the frontend now shows a toast notification when a VOD stream starts or stops. `vod_started` and `vod_stopped` WebSocket events are fired from the backend when a new provider connection is opened or the last active stream on a session ends. The 1-second delayed-cleanup window is used as a settle period before firing `vod_stopped`, so seek reconnects that re-establish `active_streams` within that window suppress the notification. A `vod_stats` WebSocket push is sent alongside every event to keep the Stats page connection table in sync in real time. `vod_start` and `vod_stop` system events are also written to the system event log for each transition, and are now selectable as integration trigger events (webhook/script) in the Connect page. - **Channel Profiles column in Users table**: users can now see all channel profiles assigned to each user directly in the Users table. Added tooltips for profile names to handle overflow, adjusted column sizing to prevent overflow between columns, and improved the table layout for better readability. (Closes #819) — Thanks [@damien-alt-sudo](https://github.com/damien-alt-sudo) - **Plugin warning & disclaimer components**: extracted shared plugin warning UI into a new `PluginWarnings.jsx` component and normalized warning/disclaimer usage across all plugin action modals. New reusable components: `PluginSecurityWarning` (untrusted code), `PluginSupportDisclaimer` (community support scope), `PluginDowngradeWarning` (version downgrade), `PluginInfoNote` (informational), and `PluginRestartWarning` (backend restart on import). Updated `Plugins.jsx`, `AvailablePluginCard.jsx`, and `PluginCard.jsx` to use the shared components. — Thanks [@sethwv](https://github.com/sethwv) @@ -42,9 +47,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Eliminated per-tick DB queries from the channel stats system. Previously `get_basic_channel_info()` in `channel_status.py` issued a `Stream.objects.filter()` and an `M3UAccountProfile.objects.filter()` on every stats tick for each active channel to resolve display names. Channel name and stream name are now written into the Redis metadata hash at channel-init time (via `initialize_channel`) and read back directly during stats collection, zero DB queries per tick. `m3u_profile_name` is now resolved on the frontend from the already-loaded playlists store rather than being pushed from the backend. - Eliminated a redundant `Channel` DB lookup inside `ChannelService.initialize_channel()`. `views.py` already fetches the `Channel` object via `get_stream_object()` before calling `initialize_channel()`; `channel.name` is now passed as an optional `channel_name` parameter, so the service uses the caller-supplied value and only falls back to a DB query when the name is not provided (e.g. stream-preview paths). A matching `stream_name` parameter was added for the same reason; the `stream_name` DB query is skipped entirely on normal channel start since a channel name is always available. - Eliminated a redundant `Stream` DB lookup on stream switches. `get_stream_info_for_switch()` already fetches the `Stream` object to build the URL; it now includes `stream_name` in its return dict. `change_stream_url()` captures that value and passes it through to `_update_channel_metadata()`, which skips its own `Stream.objects.filter()` when the name is already known. +- **Dedicated thread-pool Celery worker for DVR recordings**. `run_recording` is long-running and almost entirely I/O-bound: it loops in short ticks polling FFmpeg, the DB, and Redis for the full duration of the recording. Running it on the default prefork worker pool (`--autoscale=6,1`) meant at most 6 concurrent recordings, and only if no other background work was running. M3U refreshes, EPG parsing, channel matching, comskip, etc. all competed for the same 6 slots, so if every worker was busy when a recording's start time arrived, the task would queue in Redis and FFmpeg would not start until a worker freed up, causing the user to miss the beginning of the show. + - A second Celery worker is now started alongside the default one, configured with `--pool=threads --concurrency=20` and bound to a dedicated `dvr` queue. Threads fit this workload because every blocking call in `run_recording` releases the GIL (sleep, subprocess wait, file / DB / Redis I/O) and FFmpeg itself runs in a separate OS process. A `task_routes` entry in `dispatcharr/celery.py` routes `apps.channels.tasks.run_recording` to the `dvr` queue regardless of how it is dispatched (Celery Beat `PeriodicTask`, `.delay()`, etc.). The default prefork worker now subscribes only to the `celery` queue (`-Q celery -n default@%h --autoscale=6,1`), so recordings can no longer starve background tasks and background tasks can no longer delay recordings. + + - Net result: up to 20 concurrent recordings, zero startup delay regardless of EPG / M3U refresh activity, and a memory cost of roughly 80 to 120 MB for the second always-on worker process. Each additional concurrent recording costs nearly nothing on top because all 20 threads share that single process. Applied to AIO (`docker/uwsgi.ini`), dev (`docker/uwsgi.dev.ini`), debug (`docker/uwsgi.debug.ini`), and the modular celery container (`docker/entrypoint.celery.sh`). ### Changed +- DVR recording pipeline rewritten around FFmpeg + HLS. The previous implementation wrote a single growing `.ts` file via the proxy and remuxed it to `.mkv` at the end; the new implementation runs FFmpeg with `-f hls`, regenerates monotonic PTS to handle erratic IPTV source timestamps, shifts output timestamps so they start at 0 (fixing negative-PTS streams that prevented HLS segment boundary detection), and concatenates the resulting segments into the final MKV at the end of the recording. Server-restart resume now continues segment numbering from the previous session rather than overwriting earlier segments. +- Stall detection added to the recording loop. Once the first segment confirms the stream is flowing, the loop watches both segment count and segment file mtime. If neither advances for the configured stall window (e.g. when an upstream proxy ghost-kills the client), FFmpeg is signaled and the recording ends cleanly rather than hanging until `end_time`. Recording duration is honored via SIGINT so FFmpeg writes `#EXT-X-ENDLIST` cleanly into the final playlist. +- `end_time` extension is now picked up by the running recording without restarting FFmpeg. The DB poll loop simply updates the in-memory deadline. +- Recording cancellation cleanup updated for the new layout. `RecordingViewSet.destroy()` now removes the HLS working directory via a `_safe_rmtree` helper (with the same `allowed_roots` containment check used for the existing `_safe_remove`) instead of trying to delete the obsolete `_temp_file_path`. +- HLS working directory lifecycle in `custom_properties` is now two-phase. After concat succeeds, only `file_url` and `output_file_url` are updated so new client requests are redirected to the final MKV via `/file/`; `_hls_dir` is intentionally preserved through the viewer-wait grace period so in-flight `.ts` requests keep resolving. `_hls_dir` is cleared from `custom_properties` only after `shutil.rmtree` actually removes the directory. - **Column resizing in `CustomTable`**: column widths are now propagated to body cells via CSS custom properties (`--header-{id}-size`) injected on the table wrapper, rather than reading `column.getSize()` directly in each cell's React style. This decouples body-cell widths from React renders so that memoized rows (which skip re-renders for performance) still reflect resize changes instantly via CSS cascade. - Decoupled row expansion from row selection in `CustomTable`, expanding a channel row no longer also selects it. Added an `onRowExpansionChange` callback to `useTable` so callers can react to expansion changes independently of selection state. - Fixed a stale-closure bug in memoized channel row checkboxes, `selectedTableIdsRef` now ensures the checkbox `onChange` handler and `handleShiftSelect` always read the current selection set rather than the stale set captured at render time. Without this, clicking any checkbox after the first would silently deselect previously checked rows. From 914c612db98528bab71dfaf7578a1bbc20a3df0d Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 26 Apr 2026 11:40:46 -0500 Subject: [PATCH 04/55] Enhancement: Implement network access checks for streaming and update recording status handling during worker shutdown --- apps/channels/api_views.py | 7 +- apps/channels/tasks.py | 208 ++++++++++++++++-- .../src/components/cards/RecordingCard.jsx | 9 +- .../forms/RecordingDetailsModal.jsx | 3 +- 4 files changed, 198 insertions(+), 29 deletions(-) diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 2907c48d..ae941af9 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -62,13 +62,14 @@ from rest_framework.filters import SearchFilter, OrderingFilter from apps.epg.models import EPGData from apps.vod.models import Movie, Series from django.db.models import Q -from django.http import HttpResponse, StreamingHttpResponse, FileResponse, Http404 +from django.http import HttpResponse, StreamingHttpResponse, FileResponse, Http404, JsonResponse from django.utils import timezone import mimetypes from django.conf import settings from rest_framework.pagination import PageNumberPagination +from dispatcharr.utils import network_access_allowed logger = logging.getLogger(__name__) @@ -2471,6 +2472,8 @@ class RecordingViewSet(viewsets.ModelViewSet): is still running (or the MKV is not yet produced), it is redirected to the HLS playlist endpoint. """ + if not network_access_allowed(request, "STREAMS"): + return JsonResponse({"error": "Forbidden"}, status=403) recording = get_object_or_404(Recording, pk=pk) cp = recording.custom_properties or {} file_path = cp.get("file_path") @@ -2556,6 +2559,8 @@ class RecordingViewSet(viewsets.ModelViewSet): to route through this endpoint so authentication and path isolation are preserved. """ + if not network_access_allowed(request, "STREAMS"): + return JsonResponse({"error": "Forbidden"}, status=403) recording = get_object_or_404(Recording, pk=pk) cp = recording.custom_properties or {} hls_dir = cp.get("_hls_dir") diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 67bec477..dd1fd3d8 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -15,6 +15,7 @@ from datetime import datetime, timedelta import gc from celery import shared_task +from celery.signals import worker_shutting_down from django.utils.text import slugify from rapidfuzz import fuzz @@ -32,6 +33,21 @@ from urllib.parse import quote logger = logging.getLogger(__name__) +# Flag set by the Celery `worker_shutting_down` signal so in-flight +# `run_recording` tasks can distinguish a graceful worker shutdown (e.g. +# `docker stop`, container update) from the natural end of a recording. +# When set, the post-FFmpeg path leaves the HLS working directory intact +# and marks the recording "interrupted" so `recover_recordings_on_startup` +# can resume it on the next boot via the existing path-reuse logic. +_DVR_SHUTTING_DOWN = False + + +@worker_shutting_down.connect +def _dvr_mark_shutting_down(**_kwargs): + global _DVR_SHUTTING_DOWN + _DVR_SHUTTING_DOWN = True + + _url_validation_cache = {} _URL_CACHE_TTL = 300 # seconds @@ -1586,9 +1602,14 @@ def _parse_epg_tv_movie_info(program): return is_movie, season, episode, year, sub_title -def _build_output_paths(channel, program, start_time, end_time): +def _build_output_paths(channel, program, start_time, end_time, recording_id): """ - Build (final_path, temp_ts_path, final_filename) using DVR templates. + Build (final_path, hls_dir, final_filename) using DVR templates. + + The HLS working directory is hidden and tagged with the recording id + (e.g. ``.dvr_71_hls``) so it cannot collide with another recording's + files and so orphan-cleanup utilities can identify internal scratch + directories unambiguously. """ from core.models import CoreSettings # Root for DVR recordings: fixed to /data/recordings inside the container @@ -1655,19 +1676,17 @@ def _build_output_paths(channel, program, start_time, end_time): final_path = rel_path if rel_path.startswith('/') else os.path.join(library_root, rel_path) final_path = os.path.normpath(final_path) - # Avoid overwriting an existing file from a different recording. - # Check the MKV file and its associated HLS working directory. + # Avoid overwriting an existing MKV from a different recording. The HLS + # working directory is keyed by recording id, so it cannot collide and is + # not part of this check. base, ext = os.path.splitext(final_path) counter = 1 while True: - candidate_base = final_path[:-len(ext)] # strip extension - hls_dir_candidate = candidate_base + '_hls' try: mkv_occupied = os.stat(final_path).st_size > 0 except OSError: mkv_occupied = False - hls_occupied = os.path.isdir(hls_dir_candidate) - if not mkv_occupied and not hls_occupied: + if not mkv_occupied: break counter += 1 final_path = f"{base}_{counter}{ext}" @@ -1675,9 +1694,8 @@ def _build_output_paths(channel, program, start_time, end_time): # Ensure directory exists os.makedirs(os.path.dirname(final_path), exist_ok=True) - # Derive HLS working directory alongside the final MKV - base_no_ext = os.path.splitext(os.path.basename(final_path))[0] - hls_dir = os.path.join(os.path.dirname(final_path), f"{base_no_ext}_hls") + # HLS working directory: hidden and id-tagged alongside the final MKV. + hls_dir = os.path.join(os.path.dirname(final_path), f".dvr_{recording_id}_hls") return final_path, hls_dir, os.path.basename(final_path) @@ -1821,7 +1839,37 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): program.update(epg_match) cp["program"] = program - final_path, hls_dir, filename = _build_output_paths(channel, program, start_time, end_time) + # Resume into the existing working set if this task is a recovery run. + # `recover_recordings_on_startup` re-dispatches `run_recording` for an + # interrupted recording with `_hls_dir` and `file_path` already set in + # custom_properties. Re-running `_build_output_paths` would allocate a + # fresh `Foo_2.mkv` / `.dvr_{id}_hls` pair and orphan the original + # segments, producing two MKVs at the end. Reuse the original paths + # whenever the HLS directory is still on disk. + existing_hls = cp.get("_hls_dir") + existing_file = cp.get("file_path") + if ( + existing_hls + and existing_file + and os.path.isdir(existing_hls) + ): + final_path = existing_file + hls_dir = existing_hls + filename = os.path.basename(final_path) + try: + _seg_count = sum( + 1 for f in os.listdir(hls_dir) if f.endswith(".ts") + ) + except OSError: + _seg_count = 0 + logger.info( + f"run_recording {recording_id}: resuming into existing HLS dir " + f"{hls_dir} ({_seg_count} segment(s)), final={final_path}" + ) + else: + final_path, hls_dir, filename = _build_output_paths( + channel, program, start_time, end_time, recording_id + ) cp["file_name"] = filename cp["file_path"] = final_path cp["_hls_dir"] = hls_dir @@ -2213,6 +2261,37 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): except subprocess.TimeoutExpired: ffmpeg_proc.kill() + # If the loop broke because the Celery worker is shutting down (e.g. + # docker stop, container update) and the recording window is still open, + # leave the HLS working directory intact and mark the recording as + # interrupted. `recover_recordings_on_startup` will re-dispatch this task + # on the next boot, and the resume-path logic in the prep block will + # adopt the existing `_hls_dir` and `file_path` so FFmpeg appends to the + # same playlist and the eventual concat produces a single MKV. + if _DVR_SHUTTING_DOWN and time.time() < end_timestamp: + try: + _shutdown_rec = Recording.objects.filter(id=recording_id).first() + if _shutdown_rec: + _shutdown_cp = _shutdown_rec.custom_properties or {} + # Preserve _hls_dir / file_path / file_url exactly as they are + # so the HLS endpoint keeps serving until the worker dies and + # so recovery can find the working directory on next boot. + _shutdown_cp["status"] = "interrupted" + _shutdown_cp["interrupted_reason"] = "server_shutdown" + _shutdown_cp["ended_at"] = str(datetime.now()) + _shutdown_rec.custom_properties = _shutdown_cp + _shutdown_rec.save(update_fields=["custom_properties"]) + except Exception as _se: + logger.warning( + f"DVR recording {recording_id}: failed to flag interrupted on shutdown: {_se}" + ) + logger.info( + f"DVR recording {recording_id}: worker shutting down with " + f"{int(end_timestamp - time.time())}s of recording window remaining; " + f"leaving HLS dir intact for resume on next boot, skipping concat." + ) + return + if not _stream_confirmed and not interrupted: interrupted = True interrupted_reason = f"no_stream_data: {last_error or 'ffmpeg_failed'}" @@ -2355,17 +2434,101 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) - if ( + _ok = ( concat_result.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0 - ): - remux_success = True - logger.info( - f"DVR recording {recording_id}: HLS→MKV concat succeeded — " - f"{len(segments)} segments → {os.path.basename(final_path)} " - f"({os.path.getsize(final_path):,} bytes)" + ) + _fallback_used = False + # MP4-intermediate fallback: some MPEG-TS streams (parameter set + # changes mid-stream, weird PMT updates, audio discontinuities) + # fail to mux directly into Matroska but go through an MP4 + # container cleanly, which can then be remuxed losslessly to + # MKV. Try this before declaring the recording lost. + if not _ok: + logger.warning( + f"DVR recording {recording_id}: direct HLS\u2192MKV concat failed " + f"(rc={concat_result.returncode}); attempting MP4-intermediate " + f"fallback. stderr: {(concat_result.stderr or '')[:300]}" ) + try: + if os.path.exists(final_path) and os.path.getsize(final_path) == 0: + os.remove(final_path) + except OSError: + pass + _intermediate_mp4 = os.path.join( + hls_dir, f".dvr_{recording_id}_intermediate.mp4" + ) + try: + if os.path.exists(_intermediate_mp4): + os.remove(_intermediate_mp4) + except OSError: + pass + _mp4_concat = subprocess.run( + [ + "ffmpeg", "-y", + "-f", "concat", "-safe", "0", + "-i", concat_list_path, + "-c", "copy", + "-bsf:a", "aac_adtstoasc", + _intermediate_mp4, + ], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + if ( + _mp4_concat.returncode == 0 + and os.path.exists(_intermediate_mp4) + and os.path.getsize(_intermediate_mp4) > 0 + ): + _mp4_to_mkv = subprocess.run( + [ + "ffmpeg", "-y", + "-i", _intermediate_mp4, + "-c", "copy", + final_path, + ], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + if ( + _mp4_to_mkv.returncode == 0 + and os.path.exists(final_path) + and os.path.getsize(final_path) > 0 + ): + _ok = True + _fallback_used = True + else: + logger.error( + f"DVR recording {recording_id}: MP4\u2192MKV remux step " + f"failed (rc={_mp4_to_mkv.returncode}). stderr: " + f"{(_mp4_to_mkv.stderr or '')[:300]}" + ) + else: + logger.error( + f"DVR recording {recording_id}: HLS\u2192MP4 concat fallback " + f"failed (rc={_mp4_concat.returncode}). stderr: " + f"{(_mp4_concat.stderr or '')[:300]}" + ) + try: + if os.path.exists(_intermediate_mp4): + os.remove(_intermediate_mp4) + except OSError: + pass + + if _ok: + remux_success = True + if _fallback_used: + logger.info( + f"DVR recording {recording_id}: HLS\u2192MP4\u2192MKV fallback " + f"concat succeeded \u2014 {len(segments)} segments \u2192 " + f"{os.path.basename(final_path)} " + f"({os.path.getsize(final_path):,} bytes)" + ) + else: + logger.info( + f"DVR recording {recording_id}: HLS\u2192MKV concat succeeded \u2014 " + f"{len(segments)} segments \u2192 {os.path.basename(final_path)} " + f"({os.path.getsize(final_path):,} bytes)" + ) # Update DB so *new* client requests go to /file/ (the final MKV) # rather than the soon-to-be-removed /hls/ endpoint. Important: # we do NOT clear _hls_dir here, active viewers still in the @@ -2432,9 +2595,10 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): logger.warning(f"DVR recording {recording_id}: post-rmtree DB update failed: {_post_e}") else: logger.error( - f"DVR recording {recording_id}: HLS→MKV concat failed " - f"(rc={concat_result.returncode}). Keeping HLS segments for recovery. " - f"stderr: {(concat_result.stderr or '')[:500]}" + f"DVR recording {recording_id}: all HLS\u2192MKV concat attempts " + f"failed (direct rc={concat_result.returncode}, MP4 fallback also " + f"failed). Keeping HLS segments for recovery. stderr: " + f"{(concat_result.stderr or '')[:500]}" ) except Exception as _ce: logger.error(f"DVR recording {recording_id}: concat exception: {_ce}") diff --git a/frontend/src/components/cards/RecordingCard.jsx b/frontend/src/components/cards/RecordingCard.jsx index 80b5ba80..97481deb 100644 --- a/frontend/src/components/cards/RecordingCard.jsx +++ b/frontend/src/components/cards/RecordingCard.jsx @@ -285,7 +285,9 @@ const RecordingCard = ({ @@ -296,10 +298,7 @@ const RecordingCard = ({ e.stopPropagation(); handleWatchRecording(); }} - disabled={ - customProps.status === 'recording' || - !(customProps.file_url || customProps.output_file_url) - } + disabled={!(customProps.file_url || customProps.output_file_url)} > Watch diff --git a/frontend/src/components/forms/RecordingDetailsModal.jsx b/frontend/src/components/forms/RecordingDetailsModal.jsx index 39643d27..bc8b245f 100644 --- a/frontend/src/components/forms/RecordingDetailsModal.jsx +++ b/frontend/src/components/forms/RecordingDetailsModal.jsx @@ -122,7 +122,8 @@ const RecordingDetailsModal = ({ const canWatchRecording = (customProps.status === 'completed' || customProps.status === 'stopped' || - customProps.status === 'interrupted') && + customProps.status === 'interrupted' || + customProps.status === 'recording') && Boolean(fileUrl); const isSeriesGroup = Boolean( From aa693f7a6a778a7245a53c78e8b41e38da10a826 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 26 Apr 2026 11:48:02 -0500 Subject: [PATCH 05/55] Enhancement: Integrate HLS.js for improved handling of live and recorded streams --- frontend/src/components/FloatingVideo.jsx | 80 ++++++++++++++++++++++- 1 file changed, 77 insertions(+), 3 deletions(-) diff --git a/frontend/src/components/FloatingVideo.jsx b/frontend/src/components/FloatingVideo.jsx index 380f7153..752ab865 100644 --- a/frontend/src/components/FloatingVideo.jsx +++ b/frontend/src/components/FloatingVideo.jsx @@ -4,6 +4,7 @@ import Draggable from 'react-draggable'; import useVideoStore from '../store/useVideoStore'; import useAuthStore from '../store/auth'; import mpegts from 'mpegts.js'; +import Hls from 'hls.js'; import { CloseButton, Flex, Loader, Text, Box } from '@mantine/core'; import { applyConstraints, @@ -279,9 +280,75 @@ export default function FloatingVideo() { video.addEventListener('error', handleError); video.addEventListener('progress', handleProgress); - // Set the source - video.src = streamUrl; - video.load(); + // HLS handling: in-progress recordings expose .m3u8 playlists (and ended + // recordings whose MKV concat hasn't completed yet still serve via /hls/). + // Native