Recover stalled live pipelines

README.md

@@ -157,15 +157,24 @@ Extensively optimized for minimal latency and CPU usage:
 - **HTTPS passthrough** - Auto-proxies HTTP streams when behind HTTPS
 
 Apple and web clients share the adaptive live transcoder for configured outputs
-above 720p: one provider ingest, a 720p startup rendition, and a separately warmed
+above 720p: one upstream ingest, a 720p startup rendition, and a separately warmed
 high-quality rendition. Both use the same backend buffer/throughput policy for
 upgrades, fallback, and recovery after sustained bandwidth improvement. The high
-encoder stays warm during bandwidth saver so recovery reuses the same provider
+encoder stays warm during bandwidth saver so recovery reuses the same upstream
 ingest. The web player switches Hls.js levels within one media
 element using matching playlist dates, and targets 12 seconds behind the live edge
 when enough media is available. Native browser HLS without Hls.js telemetry stays
 at the safe initial quality.
 
+A server-owned watchdog also checks process state and completed segment publication.
+An isolated high-quality failure keeps low playback and the upstream ingest running
+while only that encoder restarts. A stalled ingest or low output restarts the
+dependency chain only after the old upstream reader exits. Restarted outputs append
+to the retained playlists with an HLS discontinuity, preserving the available DVR
+window under the same session ID. Recovery attempts have cooldowns and a limit of
+three per five-minute window. HTTP inputs use a 30-second read timeout so an open
+but unresponsive connection becomes recoverable.
+
 Live bitrate targets/maximum settings are 4/6 Mbps for 720p, 6/8 Mbps for 1080p,
 10/14 Mbps for 1440p, and 16/20 Mbps for 4K, plus audio and transport overhead.
 The native gateway's standard live AI-upscale path shares the bitrate policy.

fast_start.py

@@ -18,13 +18,22 @@
 )
 
 
-def ingest_command(url: str, directory: str, user_agent: str | None) -> list[str]:
-    # Remux only. No probe subprocess and no encoder may open the provider URL.
+_PLAYLIST_ENTRIES = re.compile(r"#EXTINF:([\d.]+),[^\n]*\n(?:#[^\n]*\n)*([^#\n]+)")
+
+
+def ingest_command(
+    url: str,
+    directory: str,
+    user_agent: str | None,
+    *,
+    restarting: bool = False,
+) -> list[str]:
+    # Remux only. No probe subprocess and no encoder may open the upstream URL.
     cmd = ["ffmpeg", "-hide_banner", "-loglevel", "error"]
     cmd += http_reconnect_args(url, max_delay=5)
     if user_agent:
         cmd += ["-user_agent", user_agent]
-    return cmd + [
+    output = [
         "-probesize",
         "5000000",
         "-analyzeduration",
@@ -44,11 +53,16 @@ def ingest_command(url: str, directory: str, user_agent: str | None) -> list[str
         "-hls_list_size",
         "120",
         "-hls_flags",
-        "delete_segments+temp_file",
+        "delete_segments+temp_file" + ("+discont_start" if restarting else ""),
+    ]
+    if restarting:
+        output += ["-hls_start_number_source", "epoch_us"]
+    output += [
         "-hls_segment_filename",
         f"{directory}/input_%06d.ts",
         f"{directory}/input.m3u8",
     ]
+    return cmd + output
 
 
 def encoder_command(
@@ -61,6 +75,7 @@ def encoder_command(
     *,
     audio: MediaInfo | None = None,
     audio_passthrough: bool = False,
+    restarting: bool = False,
 ) -> list[str]:
     cmd = build_hls_ffmpeg_cmd(
         f"{directory}/input.m3u8",
@@ -78,14 +93,17 @@ def encoder_command(
         audio_passthrough=audio_passthrough,
     )
     index = cmd.index("-i")
-    cmd[index:index] = ["-live_start_index", "0"]
+    cmd[index:index] = ["-live_start_index", "-3" if restarting else "0"]
     cmd[cmd.index("-probesize") + 1] = "500000"
     cmd[cmd.index("-analyzeduration") + 1] = "500000"
     # Preserve the shared source timeline through both independent encoders.
     cmd[1:1] = ["-copyts"]
     prefix = "high" if high else "low"
     cmd[cmd.index("-hls_segment_filename") + 1] = f"{directory}/{prefix}_%06d.ts"
-    cmd[cmd.index("-hls_flags") + 1] = "delete_segments+temp_file"
+    flags = "delete_segments+temp_file"
+    if restarting:
+        flags += "+append_list+discont_start"
+    cmd[cmd.index("-hls_flags") + 1] = flags
     duration = "2"
     cmd[cmd.index("-hls_time") + 1] = duration
     cmd[cmd.index("-hls_list_size") + 1] = str(get_live_hls_list_size(float(duration)))
@@ -162,10 +180,29 @@ def dated_playlist(directory: str, content: str, origin_pts: float, origin_time:
     """Map both renditions to the same clock, independent of encoder warm-up."""
     lines = content.splitlines()
     result = []
+    reset_clock = False
+    duration = 0.0
     for line in lines:
+        if line.startswith("#EXT-X-PROGRAM-DATE-TIME:"):
+            continue
+        if line.startswith("#EXT-X-DISCONTINUITY"):
+            reset_clock = True
+        elif match := re.match(r"#EXTINF:([\d.]+)", line):
+            duration = float(match.group(1))
         if line and not line.startswith("#") and pathlib.Path(line).name == line:
-            pts = segment_pts(pathlib.Path(directory) / line)
+            segment = pathlib.Path(directory) / line
+            pts = segment_pts(segment)
             if pts is not None:
+                if reset_clock:
+                    try:
+                        segment_time = segment.stat().st_mtime - duration
+                        elapsed = (pts - origin_pts) % ((1 << 33) / 90000)
+                        if abs(origin_time + elapsed - segment_time) > 60:
+                            origin_time = segment_time
+                            origin_pts = pts
+                    except OSError:
+                        pass
+                    reset_clock = False
                 elapsed = (pts - origin_pts) % ((1 << 33) / 90000)
                 date = datetime.fromtimestamp(origin_time + elapsed, UTC)
                 result.append("#EXT-X-PROGRAM-DATE-TIME:" + date.isoformat(timespec="milliseconds"))
@@ -179,7 +216,7 @@ def aligned(directory: str) -> bool:
         ends = []
         for name in ("low.m3u8", "high.m3u8"):
             content = (pathlib.Path(directory) / name).read_text()
-            entries = re.findall(r"#EXTINF:([\d.]+),[^\n]*\n([^#\n]+)", content)
+            entries = _PLAYLIST_ENTRIES.findall(content)
             duration, filename = entries[-1]
             pts = segment_pts(pathlib.Path(directory) / filename)
             if pts is None:
@@ -190,13 +227,19 @@ def aligned(directory: str) -> bool:
         return False
 
 
-def ready_bitrate(directory: str, name: str, *, minimum_segments: int = 2) -> float:
+def ready_bitrate(
+    directory: str,
+    name: str,
+    *,
+    minimum_segments: int = 2,
+    max_age: float = 10,
+) -> float:
     """Require complete, fresh segments; use the largest recent segment bitrate."""
     path = pathlib.Path(directory) / name
     try:
-        if time.time() - path.stat().st_mtime > 10:
+        if time.time() - path.stat().st_mtime > max_age:
             return 0
-        entries = re.findall(r"#EXTINF:([\d.]+),[^\n]*\n([^#\n]+)", path.read_text())
+        entries = _PLAYLIST_ENTRIES.findall(path.read_text())
         if len(entries) < minimum_segments:
             return 0
         rates = []

fast_start_test.py

@@ -53,8 +53,10 @@ def test_timeline_and_readiness(tmp_path):
     assert "#EXT-X-PROGRAM-DATE-TIME:1970-01-01T00:00:00.000+00:00" in dated
     playlist(tmp_path, "high", start=0)
     assert not fast_start.aligned(str(tmp_path))
-    os.utime(tmp_path / "high.m3u8", (0, 0))
-    assert fast_start.ready_bitrate(str(tmp_path), "high.m3u8") == 0
+    os.utime(tmp_path / "high.m3u8", (80, 80))
+    with patch("fast_start.time.time", return_value=100):
+        assert fast_start.ready_bitrate(str(tmp_path), "high.m3u8") == 0
+        assert fast_start.ready_bitrate(str(tmp_path), "high.m3u8", max_age=30) == 7520
     (tmp_path / "low_2.ts").unlink()
     assert fast_start.ready_bitrate(str(tmp_path), "low.m3u8") == 0
 
@@ -63,6 +65,7 @@ def test_commands_only_ingest_opens_provider(tmp_path):
     url = "https://provider.example/live.ts"
     ingest = fast_start.ingest_command(url, str(tmp_path), "neTV")
     assert ingest[ingest.index("-i") + 1] == url
+    assert ingest[ingest.index("-rw_timeout") + 1] == "30000000"
     assert ingest[ingest.index("-reconnect_on_network_error") + 1] == "1"
     assert ingest[ingest.index("-analyzeduration") + 1] == "1000000"
     assert ingest[ingest.index("-probesize") + 1] == "5000000"
@@ -81,6 +84,65 @@ def test_commands_only_ingest_opens_provider(tmp_path):
         assert duration * int(cmd[cmd.index("-hls_list_size") + 1]) >= 30
 
 
+def test_restart_commands_use_unique_sequences_and_discontinuities(tmp_path):
+    ingest = fast_start.ingest_command(
+        "https://upstream.example/live.m3u8",
+        str(tmp_path),
+        None,
+        restarting=True,
+    )
+    assert ingest[ingest.index("-hls_start_number_source") + 1] == "epoch_us"
+    assert "discont_start" in ingest[ingest.index("-hls_flags") + 1]
+
+    encoder = fast_start.encoder_command(
+        str(tmp_path),
+        "software",
+        "1080p",
+        "high",
+        False,
+        True,
+        restarting=True,
+    )
+    assert encoder[encoder.index("-live_start_index") + 1] == "-3"
+    assert "-hls_start_number_source" not in encoder
+    flags = encoder[encoder.index("-hls_flags") + 1]
+    assert "append_list" in flags
+    assert "discont_start" in flags
+    assert encoder[encoder.index("-i") + 1] == f"{tmp_path}/input.m3u8"
+
+
+def test_dated_playlist_replaces_encoder_dates_after_discontinuity(tmp_path):
+    playlist(tmp_path, "high")
+    content = (
+        (tmp_path / "high.m3u8")
+        .read_text()
+        .replace(
+            "#EXTINF:2,\n",
+            "#EXTINF:2,\n#EXT-X-DISCONTINUITY\n#EXT-X-PROGRAM-DATE-TIME:2030-01-01T00:00:00Z\n",
+        )
+    )
+    dated = fast_start.dated_playlist(str(tmp_path), content, 10, 0)
+    assert dated.count("#EXT-X-PROGRAM-DATE-TIME:") == 3
+    assert "2030-" not in dated
+    assert fast_start.dated_playlist(str(tmp_path), dated, 10, 0) == dated
+    (tmp_path / "high.m3u8").write_text(content)
+    assert fast_start.ready_bitrate(str(tmp_path), "high.m3u8") == 7520
+
+
+def test_discontinuity_keeps_shared_clock_when_pts_remain_continuous(tmp_path):
+    playlist(tmp_path, "high")
+    content = (
+        (tmp_path / "high.m3u8")
+        .read_text()
+        .replace(
+            "#EXTINF:2,\nhigh_1.ts",
+            "#EXTINF:2,\n#EXT-X-DISCONTINUITY\nhigh_1.ts",
+        )
+    )
+    os.utime(tmp_path / "high_1.ts", (32, 32))
+    dated = fast_start.dated_playlist(str(tmp_path), content, 10, 0)
+    assert "#EXT-X-PROGRAM-DATE-TIME:1970-01-01T00:00:02.000+00:00" in dated
+
 
 @pytest.mark.parametrize("high", [False, True])
 def test_encoders_keep_surround_audio(tmp_path, high):
@@ -97,6 +159,7 @@ def test_encoders_keep_surround_audio(tmp_path, high):
     assert dolby[dolby.index("-c:a") + 1] == "copy"
     assert dolby[dolby.index("-c:v") + 1] != "copy"
 
+
 def test_ingest_can_warm_encoders_before_playback_is_ready(tmp_path):
     segment = tmp_path / "input_0.ts"
     segment.write_bytes(video_packet(10) * 10)
@@ -133,7 +196,6 @@ def test_upgrade_uses_bounded_bitrate(tmp_path, hardware: HwAccel, resolution, t
     assert not set(low) & {"-qp", "-qp_i", "-qp_p", "-global_quality", "-crf", "constqp"}
 
 
-
 def test_master_playlist_reserves_surround_audio_bandwidth():
     from ffmpeg_command import MediaInfo
 
@@ -148,6 +210,7 @@ def audio(codec, channels):
     content = fast_start.master_playlist("4k", include_high=False, audio_bitrate=640_000)
     assert "BANDWIDTH=7240000\nlow.m3u8" in content
 
+
 def test_master_playlist_only_exposes_local_renditions():
     content = fast_start.master_playlist("4k", include_high=True)
     assert content == (
@@ -272,6 +335,7 @@ def measure_duration(directory, name):
         session = ffmpeg_session.get_session(result["session_id"])
         assert session is not None
         assert session["playback_policy"].bandwidth_saver is bandwidth_saver
+        assert session["watchdog_stale_after"] == max(10, 2 * source_duration + 2)
         master = pathlib.Path(session["dir"]) / "master.m3u8"
         assert ("high.m3u8" in master.read_text()) is not fail_high
         reused = await ffmpeg_session.start_transcode("https://provider/live", fast_start=True)
@@ -305,7 +369,9 @@ async def test_disconnected_startup_stops_waiting_and_cleans_ingest(tmp_path):
     with (
         patch("ffmpeg_session.get_settings", return_value={"max_resolution": "4k"}),
         patch("ffmpeg_session.get_transcode_dir", return_value=tmp_path),
-        patch("ffmpeg_session.asyncio.create_subprocess_exec", new=AsyncMock(return_value=proc)) as launch,
+        patch(
+            "ffmpeg_session.asyncio.create_subprocess_exec", new=AsyncMock(return_value=proc)
+        ) as launch,
         patch("ffmpeg_session._spawn_background_task", side_effect=lambda coro: coro.close()),
     ):
         with pytest.raises(ffmpeg_session.HTTPException) as error:
@@ -494,6 +560,27 @@ def test_real_local_encoders_share_timestamps(tmp_path):
         assert fast_start.ready_bitrate(str(tmp_path), "low.m3u8") > 0
         assert fast_start.ready_bitrate(str(tmp_path), "high.m3u8") > 0
         assert fast_start.aligned(str(tmp_path))
+
+        low_playlist = tmp_path / "low.m3u8"
+        previous_segments = {
+            line for line in low_playlist.read_text().splitlines() if line.endswith(".ts")
+        }
+        subprocess.run(
+            fast_start.encoder_command(
+                str(tmp_path),
+                "software",
+                "720p",
+                "low",
+                False,
+                False,
+                restarting=True,
+            ),
+            check=True,
+        )
+        restarted = low_playlist.read_text()
+        restarted_segments = {line for line in restarted.splitlines() if line.endswith(".ts")}
+        assert previous_segments < restarted_segments
+        assert "#EXT-X-DISCONTINUITY" in restarted
     finally:
         for process in processes:
             if process.poll() is None:
@@ -506,11 +593,19 @@ def test_saver_does_not_recover_when_high_encoder_has_failed(tmp_path):
     playlist(tmp_path, "high")
     high = FakeProcess()
     high.returncode = 1
-    session = dict(dir=str(tmp_path), username="u", fast_start=True,
-        high_process=high, playback_policy=PlaybackPolicy(bandwidth_saver=True),
-        recovery_bitrate=7520)
+    session = dict(
+        dir=str(tmp_path),
+        username="u",
+        fast_start=True,
+        high_process=high,
+        playback_policy=PlaybackPolicy(bandwidth_saver=True),
+        recovery_bitrate=7520,
+    )
     good = PlaybackHealth(buffer_seconds=12, waiting=False, observed_bitrate=100000)
-    with patch.dict(ffmpeg_session._transcode_sessions, {"failed": session}), patch("ffmpeg_session.time.monotonic") as clock:
+    with (
+        patch.dict(ffmpeg_session._transcode_sessions, {"failed": session}),
+        patch("ffmpeg_session.time.monotonic") as clock,
+    ):
         for now in range(0, 120, 2):
             clock.return_value = now
             result = ffmpeg_session.report_playback_health("failed", "u", good)

ffmpeg_command.py

@@ -885,11 +885,18 @@ def probe_media(
 # ===========================================================================
 
 
-def http_reconnect_args(url: str, *, max_delay: int = 30) -> list[str]:
+def http_reconnect_args(
+    url: str,
+    *,
+    max_delay: int = 30,
+    read_timeout: int = 30,
+) -> list[str]:
     """HTTP reconnect options must not be passed to local or other protocol inputs."""
     if not url.startswith(("http://", "https://")):
         return []
     return [
+        "-rw_timeout",
+        str(read_timeout * 1_000_000),
         "-reconnect",
         "1",
         "-reconnect_streamed",

ffmpeg_command_test.py

@@ -608,10 +608,13 @@ def test_http_reconnect_options_shared_by_live_and_vod(url):
     for is_vod in (False, True):
         cmd = build_hls_ffmpeg_cmd(url, "software", "/tmp", is_vod=is_vod)
         expected = http_reconnect_args(url)
-        start = cmd.index("-reconnect")
-        assert cmd[start : start + len(expected)] == expected
-        assert start + len(expected) <= cmd.index("-i")
+        timeout_start = cmd.index("-rw_timeout")
+        assert cmd[timeout_start : timeout_start + len(expected)] == expected
+        assert timeout_start + len(expected) <= cmd.index("-i")
+        assert cmd[timeout_start + 1] == "30000000"
     assert http_reconnect_args(url, max_delay=5)[-1] == "5"
+    custom = http_reconnect_args(url, read_timeout=12)
+    assert custom[custom.index("-rw_timeout") + 1] == "12000000"
 
 
 @pytest.mark.parametrize("url", ["/tmp/input.m3u8", "file:/tmp/input.ts", "udp://test:1234"])