review: prevent helper backpressure and trim API reads

Author: sumitake

bin/youtube-autoencoder

@@ -17,6 +17,7 @@ import selectors
 import signal
 import subprocess
 import sys
+import tempfile
 import time
 from typing import Any
 from urllib.parse import urlsplit
@@ -39,6 +40,7 @@ URL_CREDENTIALS_RE = re.compile(r"(?P<scheme>[A-Za-z][A-Za-z0-9+.-]*://)[^/\s:@]
 RTMP_SECRET_PATH_RE = re.compile(r"(?i)\b(?P<prefix>rtmps?://[^\s/]+/)[^\s]+")
 YOUTUBE_SECRET_PATH_RE = re.compile(r"(?i)\b(?P<prefix>https?://[^\s/]*(?:youtube|google)[^\s/]*/)[^\s]+")
 PROGRESS_LINE_RE = re.compile(r"^[A-Za-z0-9_]+=.*$")
+RTSP_OPTION_CACHE: dict[tuple[str, str], bool] = {}
 SUPERVISOR_LOCK_FILE = pathlib.Path(
     os.environ.get("YTA_SUPERVISOR_LOCK_FILE", HOME / ".config/youtube-autoencoder/supervisor.lock")
 ).expanduser()
@@ -300,6 +302,9 @@ def stream_config() -> tuple[str, str]:
 
 
 def binary_supports_rtsp_option(binary: str, option: str) -> bool:
+    cache_key = (binary, option)
+    if cache_key in RTSP_OPTION_CACHE:
+        return RTSP_OPTION_CACHE[cache_key]
     try:
         result = subprocess.run(
             [binary, "-hide_banner", "-h", "demuxer=rtsp"],
@@ -312,7 +317,9 @@ def binary_supports_rtsp_option(binary: str, option: str) -> bool:
         log(f"could not inspect {pathlib.Path(binary).name} RTSP options: {exc}")
         return False
     output = "\n".join((result.stdout or "", result.stderr or ""))
-    return any(line.lstrip().startswith(f"-{option} ") for line in output.splitlines())
+    supported = any(line.lstrip().startswith(f"-{option} ") for line in output.splitlines())
+    RTSP_OPTION_CACHE[cache_key] = supported
+    return supported
 
 
 def ffmpeg_supports_rtsp_option(option: str) -> bool:
@@ -575,38 +582,46 @@ def api_command_while_streaming(
     timeout: int,
 ) -> dict[str, Any]:
     cmd = [env("YTA_YOUTUBE_API", "youtube-autoencoder-api"), *args]
-    try:
-        api_process = subprocess.Popen(
-            cmd,
-            text=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        )
-    except OSError as exc:
-        raise ApiCommandError(
-            returncode=-1,
-            payload={"message": f"could not start YouTube API helper: {exc}", "retry_class": "fatal"},
-        ) from exc
-    deadline = time.monotonic() + timeout
-    try:
-        while api_process.poll() is None:
-            remaining = max(0.0, deadline - time.monotonic())
-            drain_ffmpeg_output(runtime, timeout=min(0.5, remaining))
+    with (
+        tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as stdout_file,
+        tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as stderr_file,
+    ):
+        try:
+            api_process = subprocess.Popen(
+                cmd,
+                text=True,
+                stdout=stdout_file,
+                stderr=stderr_file,
+            )
+        except OSError as exc:
+            raise ApiCommandError(
+                returncode=-1,
+                payload={"message": f"could not start YouTube API helper: {exc}", "retry_class": "fatal"},
+            ) from exc
+        deadline = time.monotonic() + timeout
+        try:
+            while api_process.poll() is None:
+                remaining = max(0.0, deadline - time.monotonic())
+                drain_ffmpeg_output(runtime, timeout=min(0.5, remaining))
+                ensure_encoder_running(runtime)
+                if time.monotonic() >= deadline:
+                    raise ApiCommandError(
+                        returncode=-1,
+                        payload={
+                            "message": f"YouTube API command timed out after {timeout}s",
+                            "retry_class": "api",
+                        },
+                    )
+            drain_ffmpeg_output(runtime, timeout=0.0)
             ensure_encoder_running(runtime)
-            if time.monotonic() >= deadline:
-                raise ApiCommandError(
-                    returncode=-1,
-                    payload={
-                        "message": f"YouTube API command timed out after {timeout}s",
-                        "retry_class": "api",
-                    },
-                )
-        drain_ffmpeg_output(runtime, timeout=0.0)
-        ensure_encoder_running(runtime)
-        stdout, stderr = api_process.communicate()
-    except BaseException:
-        stop_process(api_process)
-        raise
+            api_process.wait()
+            stdout_file.seek(0)
+            stderr_file.seek(0)
+            stdout = stdout_file.read()
+            stderr = stderr_file.read()
+        except BaseException:
+            stop_process(api_process)
+            raise
     return parse_api_result(api_process.returncode, stdout, stderr)
 
 

bin/youtube-autoencoder-api

@@ -783,10 +783,7 @@ def stream_status(stream_id: str | None = None) -> dict[str, Any]:
     }
 
 
-def broadcast_status(broadcast_id: str) -> dict[str, Any]:
-    broadcast = broadcast_by_id(broadcast_id)
-    if broadcast is None:
-        raise ReconciliationError(f"liveBroadcast not found: {broadcast_id}")
+def broadcast_status_payload(broadcast: dict[str, Any]) -> dict[str, Any]:
     return {
         "broadcast_id": broadcast.get("id"),
         "title": (broadcast.get("snippet") or {}).get("title"),
@@ -797,6 +794,13 @@ def broadcast_status(broadcast_id: str) -> dict[str, Any]:
     }
 
 
+def broadcast_status(broadcast_id: str) -> dict[str, Any]:
+    broadcast = broadcast_by_id(broadcast_id)
+    if broadcast is None:
+        raise ReconciliationError(f"liveBroadcast not found: {broadcast_id}")
+    return broadcast_status_payload(broadcast)
+
+
 def set_broadcast_privacy(broadcast_id: str, privacy: str) -> dict[str, Any]:
     current = broadcast_by_id(broadcast_id)
     if current is None:
@@ -956,7 +960,7 @@ def set_privacy_command(args: argparse.Namespace) -> int:
             }
         )
         write_state(state)
-    print(json.dumps(broadcast_status(broadcast_id), indent=2, sort_keys=True))
+    print(json.dumps(broadcast_status_payload(result), indent=2, sort_keys=True))
     return 0
 
 

tests/test_api.py

@@ -500,6 +500,34 @@ def fake_api(method, path, params, body=None):
     ]
 
 
+def test_set_privacy_command_reuses_verified_readback(load_script, monkeypatch, tmp_path, capsys):
+    api = load_script("youtube-autoencoder-api", "yta_api_privacy_command")
+    configure_reconciliation(api, monkeypatch, tmp_path)
+    public = managed_broadcast(api, "broadcast-1", "live")
+    public["status"]["privacyStatus"] = "public"
+    api.write_state(
+        {
+            "schema_version": 2,
+            "instance_id": "encoder-1",
+            "stream_id": "stream-1",
+            "broadcast_id": "broadcast-1",
+        }
+    )
+    monkeypatch.setattr(api, "set_broadcast_privacy", lambda _broadcast_id, _privacy: public)
+    monkeypatch.setattr(
+        api,
+        "broadcast_status",
+        lambda _broadcast_id: pytest.fail("third broadcast read called"),
+    )
+
+    args = argparse.Namespace(broadcast_id="broadcast-1", privacy="public")
+
+    assert api.set_privacy_command(args) == 0
+    output = json.loads(capsys.readouterr().out)
+    assert output["broadcast_id"] == "broadcast-1"
+    assert output["privacy"] == "public"
+
+
 def test_retry_state_updates_preserve_broadcast_identity(load_script, monkeypatch, tmp_path):
     api = load_script("youtube-autoencoder-api", "yta_api_retry_state")
     configure_reconciliation(api, monkeypatch, tmp_path)

tests/test_supervisor.py

@@ -137,6 +137,26 @@ def test_ffmpeg_capability_probe_reads_rtsp_demuxer_options(load_script, monkeyp
     assert supervisor.ffmpeg_supports_rtsp_option("rw_timeout") is False
 
 
+def test_ffmpeg_capability_probe_is_cached_per_binary_and_option(load_script, monkeypatch):
+    supervisor = load_script("youtube-autoencoder", "yta_supervisor_capability_cache")
+    calls = []
+
+    def inspect(*_args, **_kwargs):
+        calls.append(True)
+        return subprocess.CompletedProcess(
+            args=["ffmpeg"],
+            returncode=0,
+            stdout="  -timeout <int64> set timeout\n",
+            stderr="",
+        )
+
+    monkeypatch.setattr(supervisor.subprocess, "run", inspect)
+
+    assert supervisor.ffmpeg_supports_rtsp_option("timeout") is True
+    assert supervisor.ffmpeg_supports_rtsp_option("timeout") is True
+    assert len(calls) == 1
+
+
 def test_ffmpeg_capability_probe_fails_closed_on_timeout(load_script, monkeypatch):
     supervisor = load_script("youtube-autoencoder", "yta_supervisor_capability_timeout")
 
@@ -256,7 +276,13 @@ def test_api_wait_parses_structured_error(load_script, monkeypatch):
         "retry_class": "quota",
     }
     api_process = FakeApiProcess(returncode=75, stderr=json.dumps(payload))
-    monkeypatch.setattr(supervisor.subprocess, "Popen", lambda *_args, **_kwargs: api_process)
+
+    def launch(*_args, **kwargs):
+        kwargs["stderr"].write(api_process.stderr_text)
+        kwargs["stderr"].flush()
+        return api_process
+
+    monkeypatch.setattr(supervisor.subprocess, "Popen", launch)
     runtime = supervisor.StreamRuntime(
         process=FakeFfmpegProcess(returncode=None),
         selector=FakeSelector(),
@@ -270,6 +296,34 @@ def test_api_wait_parses_structured_error(load_script, monkeypatch):
     assert raised.value.retry_after == 120
 
 
+def test_api_wait_spools_large_helper_output_without_pipe_backpressure(load_script, monkeypatch):
+    supervisor = load_script("youtube-autoencoder", "yta_supervisor_api_spool")
+    payload = {"stream_id": "stream-1", "padding": "x" * 100_000}
+    api_process = FakeApiProcess(returncode=0)
+
+    def launch(*_args, **kwargs):
+        assert kwargs["stdout"] != subprocess.PIPE
+        assert kwargs["stderr"] != subprocess.PIPE
+        kwargs["stdout"].write(json.dumps(payload))
+        kwargs["stdout"].flush()
+        return api_process
+
+    monkeypatch.setattr(supervisor.subprocess, "Popen", launch)
+    runtime = supervisor.StreamRuntime(
+        process=FakeFfmpegProcess(returncode=None),
+        selector=FakeSelector(),
+        watchdog=supervisor.ProgressWatchdog(
+            started_at=supervisor.time.monotonic(),
+            timeout=30.0,
+            last_progress_at=supervisor.time.monotonic(),
+        ),
+    )
+
+    result = supervisor.api_command_while_streaming(["stream-status"], runtime, timeout=5)
+
+    assert result == payload
+
+
 def test_api_wait_wraps_helper_launch_failure(load_script, monkeypatch):
     supervisor = load_script("youtube-autoencoder", "yta_supervisor_api_missing")
     monkeypatch.setattr(