diff --git a/drb-edge-node/Dockerfile b/drb-edge-node/Dockerfile index af71952..532dd72 100644 --- a/drb-edge-node/Dockerfile +++ b/drb-edge-node/Dockerfile @@ -5,6 +5,7 @@ RUN apt-get update && apt-get install -y \ libopus0 \ libopus-dev \ libpulse0 \ + pulseaudio-utils \ && rm -rf /var/lib/apt/lists/* WORKDIR /app diff --git a/drb-edge-node/app/internal/call_recorder.py b/drb-edge-node/app/internal/call_recorder.py index 1f8c18c..3e01a65 100644 --- a/drb-edge-node/app/internal/call_recorder.py +++ b/drb-edge-node/app/internal/call_recorder.py @@ -96,6 +96,19 @@ TAIL_WAIT_POLL_SECONDS = 0.05 RESTART_BACKOFF_MIN = 1.0 RESTART_BACKOFF_MAX = 15.0 +# Substrings looked for in FFmpeg's stderr to classify why capture exited. +# "No such process" is what FFmpeg's pulse input prints when the daemon is up +# but the named source does not exist — a real PULSE_SOURCE misconfiguration, +# not a startup race. This is exactly the failure mode that hid unnoticed +# behind a generic "restarting" log in the April outage: silently retrying +# forever against a wrong source name looks identical to a normal startup +# wait unless it is logged differently. +_SOURCE_MISSING_MARKERS = ("no such process", "no such device") +# "Connection refused"/"Connection failure" is what the pulse client library +# prints when nothing is listening on the socket at all — expected while the +# op25 container's daemon is still coming up. +_NO_DAEMON_MARKERS = ("connection refused", "connection failure") + @dataclass class _ActiveRecording: @@ -154,6 +167,10 @@ class CallRecorder: # Active recording state (None when idle) self._active: Optional[_ActiveRecording] = None + # Last few lines of the most recent FFmpeg stderr, used to classify + # why a capture process exited (see _classify_capture_exit). + self._last_stderr_lines: deque[str] = deque(maxlen=10) + # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ @@ -206,7 +223,7 @@ class CallRecorder: continue await self._run_capture() - logger.warning("PulseAudio capture process exited — restarting.") + self._log_capture_exit() except asyncio.CancelledError: await self._terminate_proc() raise @@ -220,6 +237,7 @@ class CallRecorder: async def _run_capture(self) -> None: cmd = self._ffmpeg_command() logger.info(f"Starting capture: ffmpeg -f pulse -i {settings.pulse_source}") + self._last_stderr_lines.clear() proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, @@ -253,12 +271,48 @@ class CallRecorder: line = await proc.stderr.readline() if not line: return - logger.warning(f"ffmpeg(pulse): {line.decode(errors='replace').strip()}") + text = line.decode(errors="replace").strip() + self._last_stderr_lines.append(text) + logger.warning(f"ffmpeg(pulse): {text}") except asyncio.CancelledError: return except Exception: return + def _log_capture_exit(self) -> None: + """ + Log why the just-finished capture process exited, distinguishing the + two failure modes that matter operationally instead of one generic + "restarting" line for both: + + - no daemon / connection refused: infrastructure isn't up yet. This + is expected during startup/op25 restarts, so it stays at INFO — + the retry loop above already handles it. + - daemon up but the named source is missing: almost always a real + PULSE_SOURCE misconfiguration. This gets a loud, distinct ERROR + naming the configured source, because silently retrying forever + against a wrong source name is exactly how this hid in the past. + """ + text = " ".join(self._last_stderr_lines).lower() + + if any(marker in text for marker in _SOURCE_MISSING_MARKERS): + logger.error( + f"PulseAudio capture exited: source '{settings.pulse_source}' does not exist on the " + "daemon (FFmpeg reported 'No such process'). This looks like a real PULSE_SOURCE " + "misconfiguration or a missing drb_sink — retrying will not fix it by itself. " + "Restarting anyway." + ) + return + + if any(marker in text for marker in _NO_DAEMON_MARKERS): + logger.info( + "PulseAudio capture exited: daemon not accepting connections — " + "infrastructure still coming up, restarting." + ) + return + + logger.warning("PulseAudio capture process exited — restarting.") + async def _terminate_proc(self) -> None: proc, self._proc = self._proc, None if proc is None or proc.returncode is not None: diff --git a/drb-edge-node/app/internal/pulse.py b/drb-edge-node/app/internal/pulse.py index 37b3d3d..be00d6b 100644 --- a/drb-edge-node/app/internal/pulse.py +++ b/drb-edge-node/app/internal/pulse.py @@ -5,11 +5,24 @@ The PulseAudio daemon lives in the `op25` container and exposes its native socket on the shared `pulse_socket` docker volume (mounted at /run/pulse in both containers, with PULSE_SERVER=unix:/run/pulse/native). -`op25-container/docker-entrypoint.sh` waits up to ~10 s for that socket before -starting its own app, but the edge-node historically had *no* equivalent wait: -FFmpeg would be launched with `-f pulse` before the socket existed, fail -instantly, and the audio path would stay dead for the lifetime of the process. -This module is the missing wait. +`op25-container/docker-entrypoint.sh` waits (bounded) for that daemon to +actually answer before starting its own app, and the edge-node needs the same +guarantee before launching FFmpeg: FFmpeg with `-f pulse` fails instantly if +nothing is listening, and used to stay dead for the lifetime of the process. +This module is the wait. + +HISTORY / WHY THIS CHECKS LIVENESS, NOT FILE EXISTENCE: the `pulse_socket` +named volume survives container recreation, but the daemon process that +created the socket does not. Observed on live hardware: a stale +`/run/pulse/native` socket file and `/run/pulse/pid` from a killed daemon +were still in the volume after `docker compose up -d --build` recreated the +containers. PulseAudio refused to start ("Daemon already running") because of +the stale pid file, so nothing was actually listening on the socket — but the +socket *file* still existed. An earlier version of this module (and of the +op25 entrypoint) only checked `stat.S_ISSOCK` on the path, so it reported +"ready" against a dead daemon, FFmpeg launched anyway, and immediately failed +with "No such process" in a tight restart loop. Readiness here means "a +PulseAudio connection actually succeeds," never "a file exists at this path." NOTE on the source name: the op25 entrypoint starts pulseaudio with `-n`, which skips /etc/pulse/system.pa entirely and loads modules from the command line @@ -19,7 +32,8 @@ monitor explicitly via settings.pulse_source (default "drb_sink.monitor"). """ import asyncio import os -import stat +import shutil +import subprocess from typing import Optional from app.config import settings @@ -28,6 +42,14 @@ from app.internal.logger import logger DEFAULT_SOCKET_PATH = "/run/pulse/native" POLL_INTERVAL = 0.5 +# Bounded timeout for a single `pactl info` liveness probe. Kept short: this +# runs synchronously on the calling thread (see is_ready()), and callers of +# is_ready() include a sync code path inside the Discord voice bot, so a slow +# probe would stall its event loop. wait_until_ready() runs probes off-thread +# via asyncio.to_thread and can afford this bound comfortably within its own +# much larger PULSE_WAIT_TIMEOUT. +PROBE_TIMEOUT_SECONDS = 1.5 + def socket_path() -> str: """Resolve the PulseAudio socket path from PULSE_SERVER (`unix:/path` form).""" @@ -39,38 +61,81 @@ def socket_path() -> str: return DEFAULT_SOCKET_PATH -def is_ready() -> bool: - """True when the PulseAudio native socket exists and really is a socket.""" - try: - return stat.S_ISSOCK(os.stat(socket_path()).st_mode) - except OSError: +def _probe_env(path: str) -> dict: + env = dict(os.environ) + env["PULSE_SERVER"] = f"unix:{path}" + return env + + +def _daemon_responds() -> bool: + """ + True only when a PulseAudio daemon actually answers on the configured + socket. Shells out to `pactl info` (from `pulseaudio-utils`, installed + alongside `libpulse0` in the edge-node image) rather than re-implementing + the native protocol handshake in Python — this container has no other use + for talking to PulseAudio directly, so a subprocess call is the smallest + correct implementation. + + Deliberately does NOT check `os.path.exists`/`stat.S_ISSOCK` first: a + stale socket file from a killed daemon passes that check and always did, + which is the exact defect this function replaces. + """ + path = socket_path() + pactl = shutil.which("pactl") + if pactl is None: + logger.error("pactl not found in PATH — cannot verify PulseAudio liveness.") return False + try: + result = subprocess.run( + [pactl, "info"], + env=_probe_env(path), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=PROBE_TIMEOUT_SECONDS, + ) + return result.returncode == 0 + except (subprocess.TimeoutExpired, OSError): + return False + + +def is_ready() -> bool: + """ + True when a PulseAudio daemon is alive and answering right now. + + Synchronous and bounded by PROBE_TIMEOUT_SECONDS — used from a sync call + site (discord_radio._play_stream). Prefer `wait_until_ready()` from async + code so the probe doesn't block the event loop. + """ + return _daemon_responds() async def wait_until_ready(timeout: Optional[float] = None) -> bool: """ - Block until the PulseAudio socket appears, or `timeout` seconds elapse. + Block until a PulseAudio daemon actually answers, or `timeout` seconds + elapse. - Bounded on purpose — never hang the caller forever. Returns True if the - socket is present, False on timeout (caller decides whether to retry). + Bounded on purpose — never hang the caller forever. Returns True once a + live connection succeeds, False on timeout (caller decides whether to + retry). Each probe runs via asyncio.to_thread so the subprocess call never + blocks the event loop. """ limit = settings.pulse_wait_timeout if timeout is None else timeout path = socket_path() - if is_ready(): + if await asyncio.to_thread(_daemon_responds): return True - logger.info(f"Waiting up to {limit:.0f}s for PulseAudio socket at {path}…") + logger.info(f"Waiting up to {limit:.0f}s for a live PulseAudio daemon at {path}…") waited = 0.0 while waited < limit: await asyncio.sleep(POLL_INTERVAL) waited += POLL_INTERVAL - if is_ready(): - logger.info(f"PulseAudio socket ready after {waited:.1f}s.") + if await asyncio.to_thread(_daemon_responds): + logger.info(f"PulseAudio daemon live after {waited:.1f}s.") return True logger.error( - f"PulseAudio socket {path} not present after {limit:.0f}s — " - "is the op25 container running? Audio capture will retry." + f"PulseAudio daemon at {path} not responding after {limit:.0f}s — " + "is the op25 container's daemon actually up? Audio capture will retry." ) return False diff --git a/drb-edge-node/tests/test_call_recorder.py b/drb-edge-node/tests/test_call_recorder.py index a307fdb..0727932 100644 --- a/drb-edge-node/tests/test_call_recorder.py +++ b/drb-edge-node/tests/test_call_recorder.py @@ -421,4 +421,64 @@ def test_memory_ceiling_covers_the_longest_allowed_call(): """The cap must bound RAM without ever being able to truncate a legal call.""" bytes_per_second = 16_000 // 8 assert MAX_RECORDING_BYTES >= MAX_RECORDING_SECONDS * bytes_per_second + + +# --------------------------------------------------------------------------- +# Capture-exit classification — the two failure modes must be told apart +# instead of both logging the same generic "restarting" line. This is what +# let a wrong PULSE_SOURCE hide behind normal-looking startup retries before. +# --------------------------------------------------------------------------- + +def _log_levels(caplog, logger_name="drb-edge-node"): + return [r.levelname for r in caplog.records if r.name == logger_name] + + +def test_capture_exit_logs_error_when_source_missing(recorder, caplog): + """FFmpeg's pulse input prints 'No such process' when the daemon is up + but the configured source name does not exist — a real misconfiguration, + not a startup race, so this must stand out as an error naming the source.""" + recorder._last_stderr_lines.append( + "[pulse @ 0x...] pa_stream_connect_record failed: No such process" + ) + with caplog.at_level("INFO", logger="drb-edge-node"): + recorder._log_capture_exit() + + assert "ERROR" in _log_levels(caplog) + error_messages = [r.message for r in caplog.records if r.levelname == "ERROR"] + assert any(settings.pulse_source in m for m in error_messages) + + +def test_capture_exit_logs_info_when_no_daemon(recorder, caplog): + """Connection refused means nothing is listening yet — expected during + startup, so it must NOT be logged at the same severity as a real + misconfiguration.""" + recorder._last_stderr_lines.append( + "[pulse @ 0x...] pa_context_connect() failed: Connection refused" + ) + with caplog.at_level("INFO", logger="drb-edge-node"): + recorder._log_capture_exit() + + levels = _log_levels(caplog) + assert "ERROR" not in levels + assert "INFO" in levels + + +def test_capture_exit_falls_back_to_generic_warning(recorder, caplog): + """An FFmpeg failure that matches neither known marker keeps the original + generic behavior rather than guessing.""" + recorder._last_stderr_lines.append("[pulse @ 0x...] some other unexpected failure") + with caplog.at_level("INFO", logger="drb-edge-node"): + recorder._log_capture_exit() + + assert _log_levels(caplog) == ["WARNING"] + + +def test_capture_exit_with_no_stderr_captured_is_generic_warning(recorder, caplog): + """No stderr at all (e.g. FFmpeg killed before printing anything) must not + crash the classifier and must fall back to the generic message.""" + assert list(recorder._last_stderr_lines) == [] + with caplog.at_level("INFO", logger="drb-edge-node"): + recorder._log_capture_exit() + + assert _log_levels(caplog) == ["WARNING"] assert MAX_RECORDING_BYTES <= 8 * 1024 * 1024, "must stay small enough for a Pi" diff --git a/drb-edge-node/tests/test_pulse.py b/drb-edge-node/tests/test_pulse.py new file mode 100644 index 0000000..aef07fe --- /dev/null +++ b/drb-edge-node/tests/test_pulse.py @@ -0,0 +1,135 @@ +""" +Unit tests for PulseAudio readiness helpers (app.internal.pulse). + +The whole point of this module is that readiness means "a PulseAudio +connection actually succeeds," never "a socket file exists at this path" — +that was the exact bug reproduced on live hardware: a killed daemon left its +pid file and native socket behind in the shared `pulse_socket` volume, the +old file-existence check reported "ready", and FFmpeg launched against a +dead daemon. + +No real PulseAudio daemon or `pactl` binary is required for these tests: +`_daemon_responds` (the one function that actually shells out) is monkeypatched +everywhere except the dedicated subprocess-layer tests, which fake out +`shutil.which`/`subprocess.run` directly so the "stale file, dead daemon" case +is proven at the layer that matters. +""" +import subprocess +from unittest.mock import Mock + +from app.internal import pulse + + +# --------------------------------------------------------------------------- +# socket_path() +# --------------------------------------------------------------------------- + +def test_socket_path_defaults_when_pulse_server_unset(monkeypatch): + monkeypatch.delenv("PULSE_SERVER", raising=False) + assert pulse.socket_path() == pulse.DEFAULT_SOCKET_PATH + + +def test_socket_path_parses_unix_prefixed_pulse_server(monkeypatch): + monkeypatch.setenv("PULSE_SERVER", "unix:/tmp/somewhere/native") + assert pulse.socket_path() == "/tmp/somewhere/native" + + +def test_socket_path_falls_back_on_malformed_pulse_server(monkeypatch): + monkeypatch.setenv("PULSE_SERVER", "not-a-unix-uri") + assert pulse.socket_path() == pulse.DEFAULT_SOCKET_PATH + + +# --------------------------------------------------------------------------- +# is_ready() / wait_until_ready() against a monkeypatched probe +# --------------------------------------------------------------------------- + +def test_is_ready_true_when_daemon_responds(monkeypatch): + monkeypatch.setattr(pulse, "_daemon_responds", lambda: True) + assert pulse.is_ready() is True + + +def test_is_ready_false_when_daemon_does_not_respond(monkeypatch): + monkeypatch.setattr(pulse, "_daemon_responds", lambda: False) + assert pulse.is_ready() is False + + +async def test_wait_until_ready_short_circuits_when_already_live(monkeypatch): + calls = Mock(return_value=True) + monkeypatch.setattr(pulse, "_daemon_responds", calls) + assert await pulse.wait_until_ready(timeout=5) is True + assert calls.call_count == 1 + + +async def test_wait_until_ready_polls_until_daemon_comes_up(monkeypatch): + monkeypatch.setattr(pulse, "POLL_INTERVAL", 0.01) + responses = iter([False, False, True]) + monkeypatch.setattr(pulse, "_daemon_responds", lambda: next(responses)) + assert await pulse.wait_until_ready(timeout=5) is True + + +async def test_wait_until_ready_times_out_when_daemon_never_responds(monkeypatch): + monkeypatch.setattr(pulse, "POLL_INTERVAL", 0.01) + monkeypatch.setattr(pulse, "_daemon_responds", lambda: False) + assert await pulse.wait_until_ready(timeout=0.05) is False + + +async def test_wait_until_ready_uses_settings_default_when_timeout_omitted(monkeypatch): + monkeypatch.setattr(pulse.settings, "pulse_wait_timeout", 0.05) + monkeypatch.setattr(pulse, "POLL_INTERVAL", 0.01) + monkeypatch.setattr(pulse, "_daemon_responds", lambda: False) + assert await pulse.wait_until_ready() is False + + +# --------------------------------------------------------------------------- +# _daemon_responds() at the subprocess layer — proves a stale FILE is not +# enough, which is the actual regression this module fixes. +# --------------------------------------------------------------------------- + +def test_daemon_responds_false_when_pactl_missing(monkeypatch): + monkeypatch.setattr(pulse.shutil, "which", lambda name: None) + assert pulse._daemon_responds() is False + + +def test_daemon_responds_false_on_probe_timeout(monkeypatch, tmp_path): + monkeypatch.setattr(pulse.shutil, "which", lambda name: "/usr/bin/pactl") + + def fake_run(*args, **kwargs): + raise subprocess.TimeoutExpired(cmd="pactl", timeout=pulse.PROBE_TIMEOUT_SECONDS) + + monkeypatch.setattr(pulse.subprocess, "run", fake_run) + assert pulse._daemon_responds() is False + + +def test_daemon_responds_false_when_stale_socket_file_exists_but_daemon_dead(monkeypatch, tmp_path): + """ + The regression, reproduced at the layer that matters: a plain FILE sits + at the socket path (exactly what a killed daemon leaves behind), but + `pactl info` against it fails (nonzero exit — connection refused). This + must NOT be treated as ready. + """ + stale_socket = tmp_path / "native" + stale_socket.write_bytes(b"") # a stale file, not a live socket + monkeypatch.setenv("PULSE_SERVER", f"unix:{stale_socket}") + + monkeypatch.setattr(pulse.shutil, "which", lambda name: "/usr/bin/pactl") + monkeypatch.setattr( + pulse.subprocess, "run", + lambda *a, **k: subprocess.CompletedProcess(args=a, returncode=1), + ) + + assert stale_socket.exists() # sanity: the old file-existence check would pass + assert pulse._daemon_responds() is False + + +def test_daemon_responds_true_when_pactl_succeeds(monkeypatch, tmp_path): + live_socket = tmp_path / "native" + live_socket.write_bytes(b"") + monkeypatch.setenv("PULSE_SERVER", f"unix:{live_socket}") + + monkeypatch.setattr(pulse.shutil, "which", lambda name: "/usr/bin/pactl") + monkeypatch.setattr( + pulse.subprocess, "run", + lambda *a, **k: subprocess.CompletedProcess(args=a, returncode=0), + ) + + assert pulse._daemon_responds() is True diff --git a/op25-container/docker-entrypoint.sh b/op25-container/docker-entrypoint.sh index 9e1a6d1..786f703 100644 --- a/op25-container/docker-entrypoint.sh +++ b/op25-container/docker-entrypoint.sh @@ -1,32 +1,65 @@ -#!/bin/bash - -# --- Start PulseAudio Daemon --- -# -n: skip default config (load modules inline — avoids system.pa parsing issues) -# --system: run as system-wide daemon -# --log-target=stderr: makes errors visible in Docker logs -# &: background so this script continues; output still captured by Docker -echo "Starting PulseAudio daemon..." -mkdir -p /run/pulse -chmod 777 /run/pulse -pulseaudio --exit-idle-time=-1 -n --system \ - --load="module-native-protocol-unix socket=/run/pulse/native auth-anonymous=1" \ - --load="module-null-sink sink_name=drb_sink sink_properties=device.description=DRB-Sink" \ - --log-target=stderr & - -# Wait for the socket to actually exist before continuing -echo "Waiting for PulseAudio socket..." -for i in $(seq 1 20); do - if [ -S /run/pulse/native ]; then - echo "PulseAudio socket ready." - break - fi - sleep 0.5 -done -if [ ! -S /run/pulse/native ]; then - echo "WARNING: PulseAudio socket not found after 10s — edge-node audio will fail." -fi -ls -la /run/pulse/ - -# --- Execute the main command (uvicorn) --- -echo "Starting FastAPI application..." -exec "$@" \ No newline at end of file +#!/bin/bash + +PULSE_SOCKET=/run/pulse/native +PULSE_PIDFILE=/run/pulse/pid + +mkdir -p /run/pulse +chmod 777 /run/pulse + +# Returns 0 (true) only when a PulseAudio daemon actually answers on +# $PULSE_SOCKET. A socket/pid FILE existing proves nothing by itself — that +# is exactly the bug this script works around (see stale-state check below). +pulse_daemon_alive() { + PULSE_SERVER="unix:${PULSE_SOCKET}" timeout 2 pactl info >/dev/null 2>&1 +} + +# --- Clear stale PulseAudio state left behind by a killed daemon --- +# The `pulse_socket` named volume survives container recreation, but the +# PulseAudio process that owned it does not. If the previous container was +# recreated (not gracefully stopped), its pid file and native socket are +# still sitting in the volume; pulseaudio's pid.c sees the pid file and +# refuses to start ("Daemon already running") even though nothing is +# listening. Only remove these when nothing actually answers on the socket — +# never delete a socket a live daemon is using. +if [ -S "$PULSE_SOCKET" ] || [ -f "$PULSE_PIDFILE" ]; then + if pulse_daemon_alive; then + echo "PulseAudio daemon already alive and responding at ${PULSE_SOCKET} — leaving state as-is." + else + echo "STALE STATE: found ${PULSE_PIDFILE} / ${PULSE_SOCKET} from a previous container, but no daemon answers — clearing before start." + rm -f "$PULSE_SOCKET" "$PULSE_PIDFILE" + fi +fi + +# --- Start PulseAudio Daemon --- +# -n: skip default config (load modules inline — avoids system.pa parsing issues) +# --system: run as system-wide daemon +# --log-target=stderr: makes errors visible in Docker logs +# &: background so this script continues; output still captured by Docker +echo "Starting PulseAudio daemon..." +pulseaudio --exit-idle-time=-1 -n --system \ + --load="module-native-protocol-unix socket=${PULSE_SOCKET} auth-anonymous=1" \ + --load="module-null-sink sink_name=drb_sink sink_properties=device.description=DRB-Sink" \ + --log-target=stderr & + +# Wait for the daemon to actually answer — NOT just for the socket file to +# exist. A stale socket file from a killed daemon exists but nothing is +# listening on it; a file-existence check reports "ready" against a dead +# daemon, which is exactly how this class of bug slipped through before. +echo "Waiting for PulseAudio to become live..." +PULSE_LIVE=0 +for i in $(seq 1 20); do + if pulse_daemon_alive; then + echo "PulseAudio daemon is live (pactl info succeeded)." + PULSE_LIVE=1 + break + fi + sleep 0.5 +done +if [ "$PULSE_LIVE" -ne 1 ]; then + echo "WARNING: PulseAudio daemon not responding after 10s — edge-node audio will fail until it recovers." +fi +ls -la /run/pulse/ + +# --- Execute the main command (uvicorn) --- +echo "Starting FastAPI application..." +exec "$@"