mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 00:55:50 +00:00
Enhancement: Enhance DVR recording reliability and error tolerance
- Improve DVR recording stability by ensuring FFmpeg restarts on mid-stream outages, maintaining segment continuity and proper numbering. - Update HLS→MKV concatenation to handle timestamp splices and corrupt segments, utilizing error-tolerant FFmpeg flags. - Introduce helper functions for managing HLS segment counts and retry logic, ensuring robust handling of recording interruptions. - Update tests to validate new retry logic and error handling in recording processes. (Closes #1170)
This commit is contained in:
SergeantPanda
parent
2c6a636c73
commit
7d6bef5507
4 changed files with 662 additions and 324 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
# =========================================================================
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue