diff --git a/CHANGELOG.md b/CHANGELOG.md index a8bf3716..8f109106 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **DVR recordings no longer stop immediately when FFmpeg exits mid-stream.** If FFmpeg crashes, stalls, or loses the source before the scheduled end time, `run_recording` now restarts it in-process instead of going straight to HLS→MKV concat and marking the recording complete. Each restart reuses the same HLS working directory and continues segment numbering (`append_list`, `-start_number`) so the final MKV is a single continuous file. Retries are bounded by a per-outage time window matching live-proxy client tolerance (`STREAM_TIMEOUT` + `FAILOVER_GRACE_PERIOD`, default 80s); the window resets whenever new segments resume, so a long recording can survive multiple separate outages. Reconnect pacing between attempts follows the same `min(0.25 × attempt, 3s)` backoff used by the live-proxy `StreamManager`. Input demuxing also uses `-err_detect ignore_err` to tolerate minor TS corruption without aborting the process. If the outage window expires without recovery, the recording is saved as `interrupted` with `ffmpeg_outage_window_exhausted` rather than falsely reported as `completed`. (Closes #1170) +- **DVR HLS→MKV concat now tolerates timestamp splices and corrupt segments.** The finalize step (and startup recovery concat) previously used a bare `ffmpeg -f concat -c copy`, which could fail on truncated tail segments or PTS discontinuities from FFmpeg restarts mid-recording. Concat now uses a shared `_dvr_build_hls_concat_cmd` helper with `-fflags +genpts+igndts+discardcorrupt`, `-err_detect ignore_err`, and `-avoid_negative_ts make_zero`; the existing MP4-intermediate fallback path uses the same tolerant input flags. +- **DVR FFmpeg restart no longer skips HLS segment numbers.** On restart, `-start_number` was set to the existing segment count while `append_list` also incremented the internal sequence for each segment reloaded from `index.m3u8`, so the first post-restart segment could land at roughly double the expected index (e.g. `seg_00028.ts` after 14 segments). Restarts now pass `-start_number 0` when the playlist already lists segments and let `append_list` continue numbering; loose `.ts` files without a playlist still seed from the highest filename index + 1. - **Null TS keepalive packets during channel initialization caused Emby (and Jellyfin) to force a deinterlace transcode or fail playback entirely.** While waiting for a channel to become ready, the generator sent a null TS packet (PID `0x1FFF`) every 0.5 s to keep the HTTP connection alive. Emby probes the initial bytes of the response via ffprobe; receiving a null packet instead of real stream data produced incorrect codec metadata and triggered a forced transcode. The null-packet yield has been replaced with a plain `gevent.sleep(0.1)`. (Fixes #1280) - **VOD proxy stream URLs survive M3U import refresh cycles.** When `process_movie_batch` / `process_series_batch` create duplicate content records during a refresh, existing `M3UMovieRelation` / `M3UEpisodeRelation` rows are repointed at the new records, orphaning the old UUIDs. External players (Emby, Jellyfin, ChannelsDVR) that cached `.strm` URLs with the dead UUID would then 404 even though the same request carried a stable `stream_id` that maps to an active relation. The proxy now falls back to stream_id resolution when the UUID lookup misses, using strictest-match-first logic (account-scoped relation preferred, then highest-priority account). A `[STREAMID-FALLBACK]` WARNING log keeps the underlying import churn. — Thanks [@R3XCHRIS](https://github.com/R3XCHRIS) - **IP lookup no longer blocks the settings page load.** The `environment` endpoint previously made up to three sequential HTTP calls (ipify, ipapi.co, ip-api.com) with 5s timeouts each, blocking the page for up to 15s if any were unreachable. The lookups now run in a background thread on first request and results are cached in Redis for 1 hour, so the endpoint returns immediately on every call. — Thanks [@sethwv](https://github.com/sethwv) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index fb676b7d..aae62ef4 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -1736,6 +1736,190 @@ def _build_output_paths(channel, program, start_time, end_time, recording_id): return final_path, hls_dir, os.path.basename(final_path) +def _dvr_ffmpeg_retry_window_seconds(): + """Max continuous outage duration before giving up on FFmpeg restarts. + """ + try: + from apps.proxy.live_proxy.config_helper import ConfigHelper + return ConfigHelper.stream_timeout() + ConfigHelper.failover_grace_period() + except Exception: + return 80.0 + + +def _dvr_count_hls_segments(hls_dir): + """Return the number of HLS segment files in hls_dir.""" + if not hls_dir or not os.path.isdir(hls_dir): + return 0 + try: + return sum( + 1 for f in os.listdir(hls_dir) + if f.startswith("seg_") and f.endswith(".ts") + ) + except OSError: + return 0 + + +def _dvr_max_hls_segment_index(hls_dir): + """Return the highest ``seg_NNNNN.ts`` index in hls_dir, or -1 if none.""" + if not hls_dir or not os.path.isdir(hls_dir): + return -1 + max_idx = -1 + try: + for f in os.listdir(hls_dir): + if not (f.startswith("seg_") and f.endswith(".ts")): + continue + try: + max_idx = max(max_idx, int(f[4:9])) + except ValueError: + continue + except OSError: + return -1 + return max_idx + + +def _dvr_hls_playlist_has_segments(hls_m3u8): + """Return True if the m3u8 lists at least one segment URI.""" + if not hls_m3u8 or not os.path.isfile(hls_m3u8): + return False + try: + with open(hls_m3u8) as _f: + for _line in _f: + _line = _line.strip() + if _line and not _line.startswith('#'): + return True + except OSError: + pass + return False + + +def _dvr_hls_start_number(hls_dir, hls_m3u8): + """Return FFmpeg ``-start_number`` for an HLS recording (re)start. + + With ``append_list``, FFmpeg reloads existing segments from ``index.m3u8`` + and increments its internal sequence counter for each one. Passing our + segment count as ``-start_number`` double-counts (e.g. 14 on disk plus + 14 from ``-start_number`` → first new segment written as ``seg_00028.ts``). + When the playlist already lists segments, use 0 and let ``append_list`` + derive the continuation. When only loose ``.ts`` files remain, seed from + the highest filename index + 1. + """ + if _dvr_hls_playlist_has_segments(hls_m3u8): + return 0 + max_idx = _dvr_max_hls_segment_index(hls_dir) + return max_idx + 1 if max_idx >= 0 else 0 + + +def _dvr_ffmpeg_retry_backoff_seconds(retry_index): + """Backoff delay before FFmpeg restart (1-based retry index). + + Matches live-proxy StreamManager reconnect pacing so more attempts fit + inside the outage window than a long exponential series would allow. + """ + return min(0.25 * retry_index, 3.0) + + +def _dvr_build_ffmpeg_cmd(stream_url, recording_id, hls_m3u8, hls_seg_pattern, hls_start_number): + """Build the FFmpeg command for DVR HLS segment recording.""" + return [ + "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", + # Tolerate minor TS corruption without aborting the whole process. + "-err_detect", "ignore_err", + "-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, + ] + + +def _dvr_build_hls_concat_cmd(concat_list_path, output_path, extra_args=None): + """Build an error-tolerant FFmpeg concat command for HLS ``.ts`` segments. + + Tolerates truncated tail segments, timestamp discontinuities at FFmpeg + restart splices, and minor MPEG-TS corruption from IPTV sources. The + concat demuxer already re-bases timestamps between files; ``genpts``, + ``igndts``, and ``avoid_negative_ts`` keep the copied stream muxable. + """ + cmd = [ + "ffmpeg", "-y", + "-fflags", "+genpts+igndts+discardcorrupt", + "-err_detect", "ignore_err", + "-f", "concat", "-safe", "0", + "-i", concat_list_path, + "-c", "copy", + "-avoid_negative_ts", "make_zero", + ] + if extra_args: + cmd.extend(extra_args) + cmd.append(output_path) + return cmd + + +def _dvr_drain_ffmpeg_stderr(proc, rec_id, tail): + """Drain FFmpeg stderr in a background thread (see run_recording for rationale).""" + try: + buf = bytearray() + stream = proc.stderr + while True: + byte = stream.read(1) + if not byte: + break + 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]) + 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}") + 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}") + + +def _dvr_ensure_ffmpeg_exited(ffmpeg_proc, timeout=5): + """Wait for FFmpeg to exit, force-kill if necessary.""" + if ffmpeg_proc is None or ffmpeg_proc.poll() is not None: + return + try: + ffmpeg_proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + try: + ffmpeg_proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + pass + + def get_dvr_stream_base_url(): """Return the single correct base URL for DVR to reach the TS stream proxy. @@ -2046,294 +2230,372 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): except Exception: pass - if not interrupted: + end_timestamp = end_time.timestamp() + _stop_poll_interval = 2.0 + _first_segment_timeout = 15.0 + _stall_timeout = 60.0 # seconds without new segments → source stream gone + _active_lock_refresh_interval = 15.0 + _ffmpeg_stderr_tail = deque(maxlen=200) + _break_reason = None + _ffmpeg_retry_count = 0 + _ffmpeg_outage_started = None + _ffmpeg_retry_window = _dvr_ffmpeg_retry_window_seconds() + + if not interrupted and hls_dir: 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, + logger.info( + f"DVR recording {recording_id}: FFmpeg outage retry window " + f"is {_ffmpeg_retry_window:.0f}s (stream_timeout + failover_grace_period)" + ) + + while ( + not interrupted + and time.time() < end_timestamp + and not _DVR_SHUTTING_DOWN + and ( + _ffmpeg_outage_started is None + or time.time() - _ffmpeg_outage_started < _ffmpeg_retry_window ) - 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: - buf = bytearray() - stream = proc.stderr - while True: - byte = stream.read(1) - if not byte: - break - 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}") - - 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() - - end_timestamp = end_time.timestamp() - _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() - _last_active_lock_refresh = 0.0 - _active_lock_refresh_interval = 15.0 - - while ffmpeg_proc.poll() is None: - time.sleep(0.5) - now = time.time() - - # Refresh the per-recording active lock so a concurrent worker - # cannot acquire it and start a duplicate ffmpeg. - if ( - _active_lock_redis is not None - and now - _last_active_lock_refresh >= _active_lock_refresh_interval - ): - try: - _active_lock_redis.expire(_active_lock_key, _active_lock_ttl) - except Exception: - pass - _last_active_lock_refresh = now - - 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: + ): + if _ffmpeg_retry_count > 0: + _outage_elapsed = time.time() - _ffmpeg_outage_started + _backoff = min( + _dvr_ffmpeg_retry_backoff_seconds(_ffmpeg_retry_count), + max(0.0, _ffmpeg_retry_window - _outage_elapsed), + ) + if _backoff <= 0: logger.warning( - f"DVR recording {recording_id}: no HLS segments produced after " - f"{_first_segment_timeout}s from {base} — stream unavailable" + f"DVR recording {recording_id}: FFmpeg outage window " + f"({_ffmpeg_retry_window:.0f}s) exhausted after " + f"{_outage_elapsed:.0f}s (last reason={_break_reason})" ) - 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 - 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. - # Only stat the most recent segment (highest filename) to keep - # this O(1) per tick instead of O(N) over a long recording. + logger.info( + f"DVR recording {recording_id}: retrying FFmpeg in {_backoff:.0f}s " + f"(outage {_outage_elapsed:.0f}s / {_ffmpeg_retry_window:.0f}s, " + f"attempt {_ffmpeg_retry_count + 1}, last reason={_break_reason})" + ) + _backoff_deadline = time.time() + _backoff + while time.time() < _backoff_deadline: + if _DVR_SHUTTING_DOWN or time.time() >= end_timestamp: + break + if ( + _ffmpeg_outage_started is not None + and time.time() - _ffmpeg_outage_started >= _ffmpeg_retry_window + ): + break try: - if segs_now and hls_dir: - _newest = max(segs_now) - _newest_mtime = os.path.getmtime( - os.path.join(hls_dir, _newest) - ) - if _newest_mtime > _last_new_seg_time: - _last_new_seg_time = _newest_mtime + _pre_retry = Recording.objects.filter( + id=recording_id + ).only("custom_properties").first() + if _pre_retry is None: + interrupted = True + interrupted_reason = "recording_deleted" + break + if (_pre_retry.custom_properties or {}).get("status") == "stopped": + _break_reason = "stopped" + break except Exception: pass - if now - _last_new_seg_time > _stall_timeout: + time.sleep(0.5) + if interrupted or _break_reason == "stopped" or _DVR_SHUTTING_DOWN: + break + if time.time() >= end_timestamp: + break + if ( + _ffmpeg_outage_started is not None + and time.time() - _ffmpeg_outage_started >= _ffmpeg_retry_window + ): logger.warning( - f"DVR recording {recording_id}: no new HLS segments for " - f"{_stall_timeout:.0f}s — source stream stalled, stopping FFmpeg" + f"DVR recording {recording_id}: FFmpeg outage window " + f"({_ffmpeg_retry_window:.0f}s) exhausted during backoff " + f"(last reason={_break_reason})" ) - 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" + hls_start_number = _dvr_hls_start_number(hls_dir, hls_m3u8) + ffmpeg_cmd = _dvr_build_ffmpeg_cmd( + stream_url, recording_id, hls_m3u8, hls_seg_pattern, hls_start_number, + ) + + logger.info( + f"DVR recording {recording_id}: starting FFmpeg " + f"(attempt {_ffmpeg_retry_count + 1}, segment start={hls_start_number}, " + f"{_dvr_count_hls_segments(hls_dir)} existing segment(s))" + ) + logger.debug( + f"DVR recording {recording_id}: FFmpeg command: " + f"{' '.join(str(a) for a in ffmpeg_cmd)}" + ) + + ffmpeg_proc = None + _break_reason = None + _attempt_stream_confirmed = False + + try: + ffmpeg_proc = subprocess.Popen( + ffmpeg_cmd, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, ) - ffmpeg_proc.send_signal(signal.SIGINT) - try: - ffmpeg_proc.wait(timeout=10) - except subprocess.TimeoutExpired: - ffmpeg_proc.kill() + except Exception as _fe: + last_error = str(_fe) + logger.warning( + f"DVR recording {recording_id}: failed to launch FFmpeg: {_fe}" + ) + _break_reason = "launch_failed" + + if ffmpeg_proc is not None and ffmpeg_proc.stderr is not None: + _stderr_thread = threading.Thread( + target=_dvr_drain_ffmpeg_stderr, + args=(ffmpeg_proc, recording_id, _ffmpeg_stderr_tail), + daemon=True, + name=f"dvr-ffmpeg-stderr-{recording_id}-{_ffmpeg_retry_count}", + ) + _stderr_thread.start() + + _last_stop_poll = time.time() + _ffmpeg_start = time.time() + _last_seg_count = hls_start_number + _last_new_seg_time = time.time() + _last_active_lock_refresh = 0.0 + + if ffmpeg_proc is not None: + while ffmpeg_proc.poll() is None: + time.sleep(0.5) + now = time.time() + + if _DVR_SHUTTING_DOWN: + logger.info( + f"DVR recording {recording_id}: worker shutting down — stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + _break_reason = "server_shutdown" + break + + if ( + _active_lock_redis is not None + and now - _last_active_lock_refresh >= _active_lock_refresh_interval + ): + try: + _active_lock_redis.expire(_active_lock_key, _active_lock_ttl) + except Exception: + pass + _last_active_lock_refresh = now + + segs_now = [ + f for f in os.listdir(hls_dir) + if f.startswith("seg_") and f.endswith(".ts") + ] + + if not _attempt_stream_confirmed: + if len(segs_now) > _last_seg_count: + _attempt_stream_confirmed = True + _stream_confirmed = True + _ffmpeg_outage_started = None + _ffmpeg_retry_count = 0 + _last_seg_count = len(segs_now) + _last_new_seg_time = now + logger.info( + f"DVR recording {recording_id}: HLS segment written, " + f"stream confirmed" + ) + else: + try: + if segs_now: + _newest = max(segs_now) + _newest_mtime = os.path.getmtime( + os.path.join(hls_dir, _newest) + ) + if _newest_mtime >= _ffmpeg_start: + _attempt_stream_confirmed = True + _stream_confirmed = True + _ffmpeg_outage_started = None + _ffmpeg_retry_count = 0 + _last_new_seg_time = _newest_mtime + logger.info( + f"DVR recording {recording_id}: HLS segment " + f"activity detected, stream confirmed" + ) + except Exception: + pass + if ( + not _attempt_stream_confirmed + and now - _ffmpeg_start > _first_segment_timeout + ): + logger.warning( + f"DVR recording {recording_id}: no HLS segments produced " + f"after {_first_segment_timeout}s from {base} — stream unavailable" + ) + 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}" + _break_reason = "no_segments" + break + else: + if len(segs_now) > _last_seg_count: + _last_seg_count = len(segs_now) + _last_new_seg_time = now + else: + try: + if segs_now: + _newest = max(segs_now) + _newest_mtime = os.path.getmtime( + os.path.join(hls_dir, _newest) + ) + 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_reason = "stall" + break + + if now >= end_timestamp: + logger.info( + f"DVR recording {recording_id}: scheduled end time reached, " + f"stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + _break_reason = "end_time" + break + + 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_reason = "deleted" + 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_reason = "stopped" + break + 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 " + f"to {new_end}" + ) + end_timestamp = new_ts + except Exception: + pass + except Exception: + pass + + if _break_reason is None: + if time.time() >= end_timestamp: + _break_reason = "end_time" + elif ffmpeg_proc.poll() is not None: + _break_reason = "unexpected_exit" + _exit_code = ffmpeg_proc.poll() + if _stream_confirmed and _exit_code not in (0, None): + logger.warning( + f"DVR recording {recording_id}: FFmpeg exited unexpectedly " + f"(rc={_exit_code}) after stream was confirmed — " + f"source stream likely disconnected" + ) + elif not _stream_confirmed: + 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 " + f"(rc={_exit_code}) for {base} without producing " + f"segments.\nFFmpeg stderr tail:\n{_tail_text[-1000:]}" + ) + else: + logger.warning( + f"DVR recording {recording_id}: FFmpeg exited " + f"(rc={_exit_code}) for {base} without producing " + f"segments (no stderr output)" + ) + + _dvr_ensure_ffmpeg_exited(ffmpeg_proc) + + if _break_reason in ("end_time", "stopped", "deleted"): 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 _DVR_SHUTTING_DOWN or time.time() >= end_timestamp: + break - # 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: + if _ffmpeg_outage_started is None: + _ffmpeg_outage_started = time.time() + + _outage_elapsed = time.time() - _ffmpeg_outage_started + if _outage_elapsed >= _ffmpeg_retry_window: logger.warning( - f"DVR recording {recording_id}: FFmpeg exited unexpectedly (rc={_exit_code}) " - f"after stream was confirmed — source stream likely disconnected" + f"DVR recording {recording_id}: FFmpeg outage window " + f"({_ffmpeg_retry_window:.0f}s) exhausted " + f"(last reason={_break_reason})" ) + break - # 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 + _ffmpeg_retry_count += 1 + logger.info( + f"DVR recording {recording_id}: FFmpeg stopped early " + f"(reason={_break_reason}), scheduling retry " + f"({_outage_elapsed:.0f}s / {_ffmpeg_retry_window:.0f}s into outage window)" + ) - # 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() + _dvr_ensure_ffmpeg_exited(ffmpeg_proc) + + if ( + _stream_confirmed + and not interrupted + and _break_reason not in ("end_time", "stopped", "deleted", "server_shutdown") + and time.time() < end_timestamp + and not _DVR_SHUTTING_DOWN + ): + interrupted = True + interrupted_reason = ( + f"ffmpeg_outage_window_exhausted: {_break_reason or last_error or 'unknown'}" + ) # If the loop broke because the Celery worker is shutting down (e.g. # docker stop, container update) and the recording window is still open, @@ -2504,13 +2766,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): _cl.write(f"file '{_escaped}'\n") concat_result = subprocess.run( - [ - "ffmpeg", "-y", - "-f", "concat", "-safe", "0", - "-i", concat_list_path, - "-c", "copy", - final_path, - ], + _dvr_build_hls_concat_cmd(concat_list_path, final_path), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) _ok = ( @@ -2544,14 +2800,11 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): except OSError: pass _mp4_concat = subprocess.run( - [ - "ffmpeg", "-y", - "-f", "concat", "-safe", "0", - "-i", concat_list_path, - "-c", "copy", - "-bsf:a", "aac_adtstoasc", + _dvr_build_hls_concat_cmd( + concat_list_path, _intermediate_mp4, - ], + extra_args=["-bsf:a", "aac_adtstoasc"], + ), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) if ( @@ -2562,6 +2815,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): _mp4_to_mkv = subprocess.run( [ "ffmpeg", "-y", + "-err_detect", "ignore_err", "-i", _intermediate_mp4, "-c", "copy", final_path, @@ -2994,12 +3248,7 @@ def recover_recordings_on_startup(): _escaped = _s.replace("'", "'\\''") _cl.write(f"file '{_escaped}'\n") _res = subprocess.run( - [ - "ffmpeg", "-y", - "-f", "concat", "-safe", "0", - "-i", _concat_txt, - "-c", "copy", mkv_path, - ], + _dvr_build_hls_concat_cmd(_concat_txt, 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: diff --git a/apps/channels/tests/test_db_retry.py b/apps/channels/tests/test_db_retry.py index e65d929a..a13560fa 100644 --- a/apps/channels/tests/test_db_retry.py +++ b/apps/channels/tests/test_db_retry.py @@ -196,15 +196,17 @@ class InitialConnectionRetryTests(TestCase): base URL before falling back to the next candidate.""" def test_reconnect_max_constant_exists_in_run_recording(self): - """run_recording must define a max-reconnect limit to prevent - infinite retries on the same broken base URL.""" + """run_recording must use a time-bounded FFmpeg outage window to prevent + infinite restarts when the source stream is permanently down.""" import inspect - from apps.channels.tasks import run_recording + from apps.channels.tasks import run_recording, _dvr_ffmpeg_retry_window_seconds source = inspect.getsource(run_recording) - # The reconnection counter pattern must be present + self.assertGreater(_dvr_ffmpeg_retry_window_seconds(), 0) self.assertIn("reconnect", source.lower(), - "run_recording must contain reconnection logic") + "run_recording must contain input reconnection flags") + self.assertIn("_ffmpeg_outage_started", source) + self.assertIn("_ffmpeg_retry_window", source) # --------------------------------------------------------------------------- diff --git a/apps/channels/tests/test_recording_pipeline.py b/apps/channels/tests/test_recording_pipeline.py index b1362422..190687f4 100644 --- a/apps/channels/tests/test_recording_pipeline.py +++ b/apps/channels/tests/test_recording_pipeline.py @@ -365,46 +365,44 @@ class RecordingStatusLifecycleTests(TestCase): # ========================================================================= class ConcatFlagsTests(TestCase): - """Verify that the finalize phase uses error-tolerant ffmpeg flags - when concatenating pre-restart segments.""" + """Verify error-tolerant FFmpeg flags on the HLS segment concat command.""" - def test_concat_command_includes_error_tolerant_flags(self): - """Inspect the source code to confirm error-tolerant flags are present. - This is a static analysis test — no ffmpeg execution needed.""" + def test_hls_concat_cmd_includes_error_tolerant_flags(self): + from apps.channels.tasks import _dvr_build_hls_concat_cmd + + cmd = _dvr_build_hls_concat_cmd("/data/concat.txt", "/data/out.mkv") + self.assertIn("+genpts+igndts+discardcorrupt", cmd) + self.assertIn("-err_detect", cmd) + self.assertEqual(cmd[cmd.index("-err_detect") + 1], "ignore_err") + self.assertIn("-avoid_negative_ts", cmd) + self.assertEqual(cmd[cmd.index("-avoid_negative_ts") + 1], "make_zero") + self.assertIn("concat", cmd) + self.assertEqual(cmd[-1], "/data/out.mkv") + + def test_hls_concat_cmd_supports_mp4_fallback_extra_args(self): + from apps.channels.tasks import _dvr_build_hls_concat_cmd + + cmd = _dvr_build_hls_concat_cmd( + "/data/concat.txt", + "/data/intermediate.mp4", + extra_args=["-bsf:a", "aac_adtstoasc"], + ) + self.assertIn("aac_adtstoasc", cmd) + self.assertEqual(cmd[-1], "/data/intermediate.mp4") + + def test_run_recording_uses_hls_concat_helper(self): import inspect from apps.channels.tasks import run_recording + source = inspect.getsource(run_recording) + self.assertIn("_dvr_build_hls_concat_cmd", source) - # The concat subprocess.run call must include these flags - self.assertIn("+genpts+igndts+discardcorrupt", source, - "Concat must use +genpts+igndts+discardcorrupt fflags") - self.assertIn("ignore_err", source, - "Concat must use -err_detect ignore_err") - self.assertIn("-f", source) - self.assertIn("concat", source) - - def test_concat_goes_directly_to_mkv(self): - """Concat must produce MKV directly (not intermediate .ts) to - preserve timestamp boundaries and avoid playback freeze at splice.""" + def test_recover_recordings_uses_hls_concat_helper(self): import inspect - from apps.channels.tasks import run_recording - source = inspect.getsource(run_recording) + from apps.channels.tasks import recover_recordings_on_startup - # Must contain reset_timestamps for proper segment boundary handling - self.assertIn("reset_timestamps", source, - "Concat must use -reset_timestamps 1 for seamless seeking") - # Must write directly to final_path (MKV), not an intermediate .ts - self.assertIn("_concat_did_remux", source, - "Concat path must set flag to skip separate remux step") - - def test_segment_time_metadata_present(self): - """Verify concat uses -segment_time_metadata for boundary awareness.""" - import inspect - from apps.channels.tasks import run_recording - source = inspect.getsource(run_recording) - - self.assertIn("segment_time_metadata", source, - "Concat must use -segment_time_metadata 1 for segment boundary handling") + source = inspect.getsource(recover_recordings_on_startup) + self.assertIn("_dvr_build_hls_concat_cmd", source) # ========================================================================= @@ -486,6 +484,92 @@ class RecoverySkipListTests(TestCase): mock_run.apply_async.assert_not_called() +# ========================================================================= +# 7. FFmpeg in-process retry loop +# ========================================================================= + +class FfmpegRetryTests(TestCase): + """Verify FFmpeg restart logic for mid-recording crashes and stalls.""" + + def test_ffmpeg_retry_constants_and_helpers_exist(self): + from apps.channels import tasks as dvr_tasks + + self.assertGreater(dvr_tasks._dvr_ffmpeg_retry_window_seconds(), 0) + self.assertEqual(dvr_tasks._dvr_count_hls_segments(None), 0) + self.assertEqual(dvr_tasks._dvr_count_hls_segments("/nonexistent"), 0) + self.assertEqual(dvr_tasks._dvr_ffmpeg_retry_backoff_seconds(1), 0.25) + self.assertEqual(dvr_tasks._dvr_ffmpeg_retry_backoff_seconds(12), 3.0) + + @patch("apps.proxy.live_proxy.config_helper.ConfigHelper.stream_timeout", return_value=60) + @patch("apps.proxy.live_proxy.config_helper.ConfigHelper.failover_grace_period", return_value=20) + def test_retry_window_matches_live_proxy_timeouts(self, _grace, _stream): + from apps.channels.tasks import _dvr_ffmpeg_retry_window_seconds + + self.assertEqual(_dvr_ffmpeg_retry_window_seconds(), 80.0) + + def test_hls_start_number_zero_when_playlist_exists(self): + import tempfile + from apps.channels.tasks import _dvr_hls_start_number + + with tempfile.TemporaryDirectory() as tmp: + m3u8 = os.path.join(tmp, "index.m3u8") + open(os.path.join(tmp, "seg_00000.ts"), "wb").write(b"\x00") + open(os.path.join(tmp, "seg_00013.ts"), "wb").write(b"\x00") + with open(m3u8, "w") as f: + f.write("#EXTM3U\n#EXT-X-TARGETDURATION:4\n") + f.write("seg_00000.ts\nseg_00013.ts\n") + # append_list reloads playlist entries; start_number must stay 0. + self.assertEqual(_dvr_hls_start_number(tmp, m3u8), 0) + + def test_hls_start_number_from_max_index_without_playlist(self): + import tempfile + from apps.channels.tasks import _dvr_hls_start_number + + with tempfile.TemporaryDirectory() as tmp: + open(os.path.join(tmp, "seg_00000.ts"), "wb").write(b"\x00") + open(os.path.join(tmp, "seg_00013.ts"), "wb").write(b"\x00") + self.assertEqual(_dvr_hls_start_number(tmp, os.path.join(tmp, "index.m3u8")), 14) + + def test_hls_start_number_zero_on_fresh_dir(self): + import tempfile + from apps.channels.tasks import _dvr_hls_start_number + + with tempfile.TemporaryDirectory() as tmp: + self.assertEqual(_dvr_hls_start_number(tmp, os.path.join(tmp, "index.m3u8")), 0) + + def test_build_ffmpeg_cmd_continues_hls_numbering(self): + from apps.channels.tasks import _dvr_build_ffmpeg_cmd + + cmd = _dvr_build_ffmpeg_cmd( + "http://127.0.0.1:5656/proxy/ts/stream/uuid", + 71, + "/data/recordings/.dvr_71_hls/index.m3u8", + "/data/recordings/.dvr_71_hls/seg_%05d.ts", + 42, + ) + self.assertIn("-start_number", cmd) + self.assertEqual(cmd[cmd.index("-start_number") + 1], "42") + hls_flags = cmd[cmd.index("-hls_flags") + 1] + self.assertIn("append_list", hls_flags) + self.assertIn("omit_endlist", hls_flags) + self.assertIn("-err_detect", cmd) + self.assertEqual(cmd[cmd.index("-err_detect") + 1], "ignore_err") + + def test_run_recording_has_retry_loop(self): + import inspect + from apps.channels.tasks import run_recording + + source = inspect.getsource(run_recording) + self.assertIn("_ffmpeg_retry_count", source) + self.assertIn("_ffmpeg_outage_started", source) + self.assertIn("_ffmpeg_retry_window", source) + self.assertIn("_break_reason", source) + self.assertIn("ffmpeg_outage_window_exhausted", source) + self.assertIn("_dvr_build_ffmpeg_cmd", source) + self.assertIn("_dvr_hls_start_number", source) + self.assertIn("_ffmpeg_retry_count = 0", source) + + # ========================================================================= # 6. Frontend red-dot filter (guideUtils.mapRecordingsByProgramId) # =========================================================================