Compare commits
3 Commits
b0a8ed2a5a
...
f1de157d69
| Author | SHA1 | Date | |
|---|---|---|---|
| f1de157d69 | |||
| ceb2836371 | |||
| 7c4a3f2f20 |
+7
-3
@@ -38,9 +38,13 @@ CALL_IDLE_TIMEOUT=3
|
||||
|
||||
# Seconds of audio kept after the last transmission ends. This is the only
|
||||
# headroom protecting the final word of a transmission — usually the disposition
|
||||
# or the address. Measured at 0.5s it left ~0.3s of real margin and one recording
|
||||
# ended mid-word, hence 1.0.
|
||||
CALL_TAIL_PAD_SECONDS=1.0
|
||||
# or the address. Raised 1.0 -> 3.0 after field measurement showed the
|
||||
# grant-to-speech offset is ~0.84-1.62s (typically ~1.5s): at 1.0s pad, short
|
||||
# calls had their recording window close before the voice even started,
|
||||
# clipping speech mid-word. Safe to be generous — trim_silence already strips
|
||||
# the extra back off long calls before upload, so only short transmissions
|
||||
# actually benefit from the larger window.
|
||||
CALL_TAIL_PAD_SECONDS=3.0
|
||||
|
||||
# Strip leading/trailing dead air before upload. ~63% of an untrimmed recording
|
||||
# is silence, which costs Whisper spend and makes it hallucinate text that was
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -47,10 +47,29 @@ class Settings(BaseSettings):
|
||||
# Audio kept after the observed end of the last transmission. The srcaddr
|
||||
# 1→0 edge can be up to one poll (0.5 s) late and the encoder adds its own
|
||||
# latency, so this is the only headroom protecting the last word of a
|
||||
# transmission — which is usually the disposition or the address. Field
|
||||
# measurement at 0.5 s left only 0.29–0.37 s of real trailing margin and one
|
||||
# recording ended mid-word, hence 1.0 s.
|
||||
call_tail_pad_seconds: float = 1.0
|
||||
# transmission — which is usually the disposition or the address.
|
||||
#
|
||||
# Raised 1.0 -> 3.0 after field measurement showed the recording WINDOW
|
||||
# (anchored to OP25 control-channel timestamps) closing well before the
|
||||
# actual voice audio arrives: grant->speech offset measured 0.84-1.62s
|
||||
# across 7 calls (~1.5s typical). At the old 1.0s pad, a short
|
||||
# transmission (e.g. a 0.97s control-channel call) had its window close
|
||||
# at T+1.97 while voice didn't start until ~T+1.5 — leaving ~0.4s of
|
||||
# captured speech, clipped mid-word. Confirmed by a 0.57s output file
|
||||
# whose final 0.10s measured -12.2dB, louder than its own -18.2dB
|
||||
# average (i.e. clipped speech, not trailing silence), and by two short
|
||||
# calls that produced no "Trimmed" log line at all because there was no
|
||||
# trailing silence left to trim.
|
||||
#
|
||||
# Safe to be generous here: trim_silence already strips trailing silence
|
||||
# back to trim_silence_guard_seconds before upload, so a larger pad costs
|
||||
# long calls nothing (the extra is trimmed away) while giving short
|
||||
# transmissions enough window to actually capture the voice. Over-capture
|
||||
# is free; under-capture loses words permanently. Do not tune this back
|
||||
# down without new field data showing the grant->speech offset has
|
||||
# shrunk — see DEFERRED.md for the call_idle_timeout coupling this value
|
||||
# now sits at.
|
||||
call_tail_pad_seconds: float = 3.0
|
||||
|
||||
# Strip leading/trailing dead air before upload. ~63% of a typical recording
|
||||
# is silence (the grant→speech delay plus the tail pad), which inflates
|
||||
|
||||
@@ -89,13 +89,30 @@ MAX_RECORDING_BYTES = MAX_RECORDING_SECONDS * _MP3_BYTES_PER_SECOND * 4
|
||||
# immediately therefore cuts the tail short — which costs the last word of the
|
||||
# transmission, usually the disposition or the address. Bounded so a dead capture
|
||||
# can never hang the upload path.
|
||||
TAIL_WAIT_TIMEOUT_SECONDS = 2.0
|
||||
#
|
||||
# Must exceed settings.call_tail_pad_seconds (default 3.0), otherwise a
|
||||
# tgid_change close — which pads past a timestamp that is still ~now — gives up
|
||||
# before the padded audio has been captured and warns on every talkgroup switch.
|
||||
TAIL_WAIT_TIMEOUT_SECONDS = 4.0
|
||||
TAIL_WAIT_POLL_SECONDS = 0.05
|
||||
|
||||
# Backoff bounds for restarting a dead capture process.
|
||||
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 +171,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 +227,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 +241,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 +275,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:
|
||||
|
||||
@@ -58,8 +58,24 @@ def _tail_pad() -> float:
|
||||
edge (up to one poll late) plus encoder latency never clips the tail.
|
||||
|
||||
Read live from settings (env CALL_TAIL_PAD_SECONDS) rather than frozen into a
|
||||
module constant, so it is tunable per node. See the setting for why the
|
||||
default moved 0.5 → 1.0.
|
||||
module constant, so it is tunable per node. See the setting in config.py for
|
||||
why the default moved 1.0 → 3.0 (short calls' recording window was closing
|
||||
before the ~1.5s grant→speech offset let voice audio even start).
|
||||
|
||||
All three close paths use this pad: idle_timeout, tgid_change, and
|
||||
tgid_change_unlogged. An earlier version of this docstring claimed the
|
||||
latter two close at "an exact, already-known boundary" (the new grant's
|
||||
timestamp, or the same poll tick) and so intentionally added no pad — THAT
|
||||
REASONING WAS WRONG and produced real truncated recordings. The boundary is
|
||||
exact only in CONTROL-CHANNEL time; the buffered AUDIO lags control-channel
|
||||
timestamps by ~1.5s (measured: 0.84-1.62s of lead trimmed across 7 field
|
||||
calls), so slicing the outgoing call at the new grant's exact timestamp cut
|
||||
roughly the last 1.5s of its real speech — calls ending mid-word with ~0s
|
||||
trailing silence. Do not reintroduce a zero-pad close for tgid_change or
|
||||
tgid_change_unlogged; if the outgoing and incoming recordings end up
|
||||
overlapping in the underlying audio because of this pad, that is correct —
|
||||
the audio genuinely contains both. See _handle_call_log and
|
||||
_handle_channels for how each path sources the timestamp this gets added to.
|
||||
"""
|
||||
return settings.call_tail_pad_seconds
|
||||
|
||||
@@ -189,9 +205,16 @@ class MetadataWatcher:
|
||||
return
|
||||
|
||||
# SPLIT: different talkgroup. The new grant's OP25 timestamp is the most
|
||||
# precise end available for the outgoing segment — the new call's audio
|
||||
# starts exactly there, so no tail pad.
|
||||
await self._close_segment(started_at, reason="tgid_change")
|
||||
# precise CONTROL-CHANNEL end for the outgoing segment, but the buffered
|
||||
# audio lags control by ~1.5s, so slicing exactly there cut the outgoing
|
||||
# call's last words. Pad past it and let the recorder's bounded tail wait
|
||||
# block until that audio has actually been captured.
|
||||
#
|
||||
# The incoming call's pre-roll comes from the ring buffer, so the delay
|
||||
# costs it nothing, and the two slices overlapping in the underlying
|
||||
# audio is correct — the stream genuinely contains one call's tail and
|
||||
# then the next call's start.
|
||||
await self._close_segment(started_at + _tail_pad(), reason="tgid_change")
|
||||
await self._open_segment(entry, tgid, started_at, now)
|
||||
|
||||
async def _handle_channels(self, channels: List[Dict[str, Any]], now: float) -> None:
|
||||
@@ -236,7 +259,7 @@ class MetadataWatcher:
|
||||
f"tgid {foreign_active_tgid} active without a call_log entry — "
|
||||
f"closing segment for tgid {self._current_tgid} (call_log event likely dropped)."
|
||||
)
|
||||
await self._close_segment(now, reason="tgid_change_unlogged")
|
||||
await self._close_segment(now + _tail_pad(), reason="tgid_change_unlogged")
|
||||
return
|
||||
|
||||
if (now - self._last_activity) >= settings.call_idle_timeout:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -233,15 +233,17 @@ async def test_srcaddr_edge_then_idle_timeout_ends_call(watcher, clock):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tail_pad_is_configurable_and_defaults_to_one_second(watcher, clock, monkeypatch):
|
||||
async def test_tail_pad_is_configurable_and_defaults_to_three_seconds(watcher, clock, monkeypatch):
|
||||
"""
|
||||
0.5s left only ~0.3s of real trailing margin in field measurement and one
|
||||
recording ended mid-word, so the default moved to 1.0 — and it has to be a
|
||||
setting, not a magic number, so it can be tuned per node.
|
||||
Field measurement showed the grant->speech offset runs ~0.84-1.62s, so a
|
||||
1.0s pad let short calls' windows close before voice audio even started
|
||||
(clipping mid-word). The default moved to 3.0 — and it has to be a
|
||||
setting, not a magic number, so it can be tuned per node without a code
|
||||
change (and so tests can prove it isn't hardcoded anywhere downstream).
|
||||
"""
|
||||
assert settings.call_tail_pad_seconds == 1.0
|
||||
assert settings.call_tail_pad_seconds == 3.0
|
||||
|
||||
monkeypatch.setattr(settings, "call_tail_pad_seconds", 2.5)
|
||||
monkeypatch.setattr(settings, "call_tail_pad_seconds", 5.0)
|
||||
|
||||
await tick(watcher, update(
|
||||
call_log=[grant(1234, clock.now)],
|
||||
@@ -251,11 +253,51 @@ async def test_tail_pad_is_configurable_and_defaults_to_one_second(watcher, cloc
|
||||
edge_time = clock.now
|
||||
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
||||
|
||||
clock.advance(settings.call_idle_timeout + 1.0)
|
||||
# Advance well past both the idle timeout AND the monkeypatched 5.0s pad so
|
||||
# the "now" cap in _handle_channels never masks the pad value under test.
|
||||
clock.advance(settings.call_idle_timeout + 6.0)
|
||||
await tick(watcher, update(channels=[channel()]))
|
||||
|
||||
payload = watcher.on_call_end.call_args[0][0]
|
||||
assert payload["ended_at_epoch"] == pytest.approx(edge_time + 2.5)
|
||||
assert payload["ended_at_epoch"] == pytest.approx(edge_time + 5.0)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_short_call_window_now_covers_delayed_voice_arrival(watcher, clock):
|
||||
"""
|
||||
Regression test for the truncation bug: a ~0.97s control-channel call
|
||||
(grant to srcaddr-drop) previously closed its window at
|
||||
last_tx_end + 1.0s pad, i.e. ~1.97s after the grant — but field
|
||||
measurement shows voice audio doesn't start until ~1.5s after the grant
|
||||
(0.84-1.62s measured), so the old window left as little as ~0.4s of
|
||||
captured speech and clipped it mid-word.
|
||||
|
||||
With the 3.0s default pad, the same short call's window must extend well
|
||||
past the ~1.5s point where voice actually starts.
|
||||
"""
|
||||
call_start = clock.now
|
||||
await tick(watcher, update(
|
||||
call_log=[grant(1234, call_start)],
|
||||
channels=[channel(tgid=1234, srcaddr=555)],
|
||||
))
|
||||
|
||||
# The control-channel call itself is short — under 1 second.
|
||||
clock.advance(0.97)
|
||||
edge_time = clock.now
|
||||
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
||||
|
||||
clock.advance(settings.call_idle_timeout + 0.5)
|
||||
await tick(watcher, update(channels=[channel()]))
|
||||
|
||||
payload = watcher.on_call_end.call_args[0][0]
|
||||
assert payload["end_reason"] == "idle_timeout"
|
||||
|
||||
voice_arrival = call_start + 1.5 # measured grant->speech offset, typical case
|
||||
assert payload["ended_at_epoch"] == pytest.approx(edge_time + settings.call_tail_pad_seconds)
|
||||
assert payload["ended_at_epoch"] > voice_arrival, (
|
||||
"recording window must extend past the point voice audio actually arrives, "
|
||||
"not just past the control-channel call_log timestamps"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@@ -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
|
||||
@@ -1,29 +1,62 @@
|
||||
#!/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..."
|
||||
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-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 socket to actually exist before continuing
|
||||
echo "Waiting for PulseAudio socket..."
|
||||
# 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 [ -S /run/pulse/native ]; then
|
||||
echo "PulseAudio socket ready."
|
||||
if pulse_daemon_alive; then
|
||||
echo "PulseAudio daemon is live (pactl info succeeded)."
|
||||
PULSE_LIVE=1
|
||||
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."
|
||||
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/
|
||||
|
||||
|
||||
Reference in New Issue
Block a user