From 80fee8e8a566ef23c62df9b10b735571250678dc Mon Sep 17 00:00:00 2001 From: volschin Date: Sat, 1 Aug 2026 04:23:21 +0000 Subject: [PATCH] fix(tts-server): bound admission, not generation, by the 5s deadline Root cause of the live 503 that dropped a response turn. api.py computed deadline = now + DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS at request entry and then enforced it over the whole request: the polling loop raised SynthesisAdmissionTimeout while the worker already held the model lock and was generating, and a successfully completed result was discarded if it landed after the deadline. The name said admission; the semantics were 'total request must finish in 5 s'. Measured idle generation is 1.5-3.5 s, so a busy shared GPU crossed it routinely. CloneRuntime.synthesize now takes an 'admitted' event and sets it once the model lock is held and the post-acquire deadline check has passed - never earlier. _run_stable_synthesis enforces the deadline only while that event is unset, so queue and lock-wait time stay bounded (a caller may already be gone) while in-flight generation runs to completion, bounded by client disconnect and the client's own HTTP timeout. Also makes the budget configurable via SYNTHESIS_ADMISSION_TIMEOUT_SECONDS (malformed or non-positive falls back to the default rather than refusing to start) and records the investigation in TODO.md. --- TODO.md | 52 ++++++++++++-- dgx/tts/api.py | 33 +++++++-- dgx/tts/clone_runtime.py | 6 ++ dgx/tts/server.py | 5 +- tests/test_tts_api.py | 119 +++++++++++++++++++++++++++++++- tests/test_tts_clone_runtime.py | 57 +++++++++++++++ 6 files changed, 260 insertions(+), 12 deletions(-) diff --git a/TODO.md b/TODO.md index b97c298..0078aef 100644 --- a/TODO.md +++ b/TODO.md @@ -28,11 +28,53 @@ lost silently. case) and logs each retry at WARNING so recovered outages stay visible. A 4xx or the request-body 500 is not retried. The backoff sleeps are cancellation points, so barge-in still cuts a retrying turn. -- [ ] **Root-cause the server-side 503.** Determine why the DGX - `/v1/audio/speech` route rejects a request under conversational load — - exclusive model lock contention, worker restart, or Traefik-level shedding. - The agent-side retry above treats the symptom; this is the cause. Decide - from the finding whether the retry needs a backoff longer than one turn. +- [x] **Root-caused the server-side 503 (2026-08-01).** Not lock contention, + not a worker restart, not Traefik shedding — all three are ruled out by the + evidence below. The request path applies the *admission* budget to the + *whole* request, so a synthesis that generates too slowly is killed + mid-flight and reported as an admission timeout. + + Mechanism, in `dgx/tts/api.py::_run_stable_synthesis`: `deadline` is set at + request entry (`monotonic() + DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS`, + **5.0 s**, `api.py:24`) and then enforced over the entire request. Line 150 + raises `SynthesisAdmissionTimeout` once the deadline passes even though the + worker thread already holds the model lock and is actively generating, and + lines 159–161 discard a *successfully completed* result if it landed after + the deadline. `api.py:99` maps that to `503 "synthesis admission timed out"`. + The name says admission; the semantics are "total request must finish in + 5 s". `create_app` is called without the argument (`server.py:55`), so the + 5 s is effectively hardcoded — there is no env knob to widen it. + + Evidence (DGX `gx10-d624`, all timestamps UTC): + - `qwen3-tts` access log: `18:28:54 200`, `18:29:08 503`, `18:29:11 200`. + Requests are serialized ~14 s apart, so the agent had nothing else in + flight; no `499` was logged between them, so the model lock was free and + admission was instant. The 5 s therefore elapsed inside generation. + - The 503 was logged by uvicorn (the app), and `proxy-traefik-1` logged zero + 503s in the hour — so it is not proxy-level shedding. + - No ERROR, traceback, or restart in the container log; the container has + been `Up (healthy)` throughout — so it is not a worker restart. + - Measured idle latency inside the container (bypassing Traefik): + `"Ja."` 0.3–0.5 s, the sentence that failed 1.5–1.7 s, a long sentence + 2.9–3.5 s. A normal long sentence already consumes ~70 % of the budget + when the GPU is otherwise idle. + - Contention at the moment of failure: the request started ~18:29:03.8 + (5 s before the logged 503) while the shared GPU was busy — vLLM + `companion-llm` reported `Running: 1` with generation throughput at + 18:29:05, and `qwen3-asr` had transcribed at 18:29:01.7. A ~3.3x slowdown + on a 1.5 s sentence is enough to cross 5 s. + +- [ ] **Fix the TTS admission budget (server side, `dgx/tts/`).** Separate + admission from execution: have `clone_runtime.synthesize` set an `admitted` + event once `_acquire_synthesis` returns, enforce the deadline in + `_run_stable_synthesis` only while `not admitted.is_set()`, and delete the + post-completion deadline check that throws away good audio. After admission + the request should be bounded by client disconnect (already detected) and + the client's own 30 s HTTP timeout. Make the timeout env-configurable while + there. Needs a GPU image rebuild + redeploy on the DGX. + The agent-side retry (above) already masks this: the live 503 was followed + by a 200 three seconds later, which is exactly what the retry now does + automatically. Backoff does **not** need to be longer than one turn. - [x] **Caller ID format confirmed live.** Lifecycle logging in `agent/answer_policy.py` is verified for both an internal FRITZ!Box extension (`**613`, 18:27 UTC) and an external mobile call (`015100000001`, 18:33 UTC). diff --git a/dgx/tts/api.py b/dgx/tts/api.py index a45d07b..c5d98c1 100644 --- a/dgx/tts/api.py +++ b/dgx/tts/api.py @@ -4,7 +4,7 @@ import asyncio import io -from collections.abc import AsyncIterator, Iterator +from collections.abc import AsyncIterator, Iterator, Mapping from contextlib import suppress from dataclasses import dataclass from threading import Event, Lock @@ -21,7 +21,25 @@ from dgx.tts.profiles import ProfileError MAX_SPEECH_INPUT_CHARACTERS = 2_000 +# Budget for *admission* only — executor queue time plus the wait for the +# exclusive model lock. Generation itself is not bounded by it; see +# _run_stable_synthesis. DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS = 5.0 +SYNTHESIS_ADMISSION_TIMEOUT_ENV = "SYNTHESIS_ADMISSION_TIMEOUT_SECONDS" + + +def admission_timeout_from_env(environ: Mapping[str, str]) -> float: + """Read the admission budget from the environment, falling back to default. + + Deliberately forgiving: a malformed or non-positive value falls back rather + than refusing to start, because a TTS outage takes every call down with it. + """ + + try: + value = float(environ.get(SYNTHESIS_ADMISSION_TIMEOUT_ENV, "")) + except ValueError: + return DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS + return value if value > 0 else DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS @dataclass(frozen=True) @@ -133,6 +151,12 @@ async def _run_stable_synthesis( ) -> tuple[list[Any], int]: deadline = monotonic() + admission_timeout_seconds cancelled = Event() + # Set by the runtime once it holds the model lock. The deadline bounds + # admission only — time spent queued in the executor or waiting for the + # lock, where the caller may already be gone. Enforcing it past admission + # killed in-flight generation and reported it as a 503, losing a whole + # response turn whenever the shared GPU made a sentence run long. + admitted = Event() operation = asyncio.create_task( asyncio.to_thread( runtime.synthesize, @@ -142,21 +166,22 @@ async def _run_stable_synthesis( cancel_event=cancelled, lock_timeout=admission_timeout_seconds, admission_deadline=deadline, + admitted=admitted, ) ) try: while True: remaining = deadline - monotonic() - if remaining <= 0: + if remaining <= 0 and not admitted.is_set(): await _cancel_stable_operation(operation, cancelled) raise SynthesisAdmissionTimeout() done, _pending = await asyncio.wait( {operation}, - timeout=min(0.025, remaining), + timeout=0.025 if admitted.is_set() else min(0.025, remaining), ) if operation in done: result = operation.result() - if monotonic() >= deadline: + if monotonic() >= deadline and not admitted.is_set(): await _cancel_stable_operation(operation, cancelled) raise SynthesisAdmissionTimeout() return result diff --git a/dgx/tts/clone_runtime.py b/dgx/tts/clone_runtime.py index 245f96e..01ad9a7 100644 --- a/dgx/tts/clone_runtime.py +++ b/dgx/tts/clone_runtime.py @@ -58,12 +58,18 @@ def synthesize( cancel_event: Event | None = None, lock_timeout: float | None = None, admission_deadline: float | None = None, + admitted: Event | None = None, ) -> tuple[list[Any], int]: profile = resolve_profile(voice, self._profiles, self._default_profile_id) deadline = self._acquire_synthesis(cancel_event, lock_timeout, admission_deadline) try: if deadline is not None and monotonic() >= deadline: raise SynthesisAdmissionTimeout() + # Admission is complete: the model lock is held and generation is + # about to start. The caller stops enforcing the admission deadline + # from here, so this must never be signalled earlier. + if admitted is not None: + admitted.set() try: return self._model.generate_voice_clone( text=text, diff --git a/dgx/tts/server.py b/dgx/tts/server.py index e0f3d81..3613de0 100644 --- a/dgx/tts/server.py +++ b/dgx/tts/server.py @@ -7,7 +7,7 @@ import torch -from dgx.tts.api import HealthMetadata, create_app +from dgx.tts.api import HealthMetadata, admission_timeout_from_env, create_app from dgx.tts.clone_runtime import CloneRuntime from dgx.tts.profiles import load_profiles from dgx.tts.runtime import require_gb10_cuda @@ -54,7 +54,8 @@ def _build_app(): ) return create_app( runtime, - HealthMetadata( + synthesis_admission_timeout_seconds=admission_timeout_from_env(os.environ), + health=HealthMetadata( model_revision=MODEL_REVISION, default_profile=DEFAULT_PROFILE, profiles_loaded=tuple(sorted(profiles)), diff --git a/tests/test_tts_api.py b/tests/test_tts_api.py index 19b8351..17e69ea 100644 --- a/tests/test_tts_api.py +++ b/tests/test_tts_api.py @@ -18,12 +18,14 @@ import pytest from dgx.tts.api import ( + DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS, MAX_SPEECH_INPUT_CHARACTERS, HealthMetadata, SpeechRequest, _cancel_stable_operation, _ClientDisconnected, _run_stable_synthesis, + admission_timeout_from_env, create_app, encode_pcm_stream, ) @@ -511,7 +513,122 @@ def synthesize(self, *args, **kwargs): executor.shutdown(wait=True) -async def test_stable_absolute_deadline_rejects_late_success() -> None: +def test_admission_timeout_from_env_defaults_when_unset_or_invalid() -> None: + assert admission_timeout_from_env({}) == DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS + for bad in ("", "abc", "0", "-1"): + assert ( + admission_timeout_from_env({"SYNTHESIS_ADMISSION_TIMEOUT_SECONDS": bad}) + == DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS + ) + + +def test_admission_timeout_from_env_reads_positive_override() -> None: + assert admission_timeout_from_env({"SYNTHESIS_ADMISSION_TIMEOUT_SECONDS": "12.5"}) == 12.5 + + +async def test_admitted_synthesis_outlives_the_admission_deadline() -> None: + """Generation slower than the admission budget must still be delivered. + + The live 503 came from here: the deadline bounds *admission*, but it was + enforced over the whole request, so a sentence whose generation ran long + (a busy shared GPU) was killed mid-flight and reported as an admission + timeout. Once the worker holds the model lock, only client disconnect and + the client's own HTTP timeout may end the request. + """ + + class AdmittedSlowRuntime(_StableRuntime): + def __init__(self) -> None: + super().__init__() + self.started = threading.Event() + self.release = threading.Event() + self.finished = threading.Event() + self.cancel_event: threading.Event | None = None + self.completed_after_deadline: bool | None = None + + def synthesize( + self, + text: str, + _voice: str | None, + _language: str | None, + *, + cancel_event: threading.Event, + **_admission: Any, + ) -> tuple[list[np.ndarray], int]: + deadline = _admission["admission_deadline"] + # The real CloneRuntime signals this once it holds the model lock. + _admission["admitted"].set() + self.calls.append(text) + self.cancel_event = cancel_event + self.started.set() + try: + self.release.wait(timeout=2) + self.completed_after_deadline = time.monotonic() >= deadline + return [np.array([0.25], dtype=np.float32)], 24_000 + finally: + self.finished.set() + + runtime = AdmittedSlowRuntime() + operation = asyncio.create_task( + _run_stable_synthesis( + runtime, + SpeechRequest(input="slow-but-admitted"), + _DisconnectRequest(), + 0.10, + ) + ) + timer: threading.Timer | None = None + try: + await _wait_for_thread_event(runtime.started) + timer = threading.Timer(0.12, runtime.release.set) + timer.start() + + audios, sample_rate = await operation + + assert runtime.completed_after_deadline is True + assert sample_rate == 24_000 + assert np.asarray(audios[0]).size == 1 + assert runtime.cancel_event is not None + assert not runtime.cancel_event.is_set() + finally: + runtime.release.set() + if timer is not None: + timer.cancel() + if not operation.done(): + operation.cancel() + with suppress(asyncio.CancelledError, Exception): + await operation + await _wait_for_thread_event(runtime.finished) + + +async def test_slow_generation_returns_audio_not_503() -> None: + """End-to-end counterpart: the route must not turn slow audio into a 503.""" + + class SlowAdmittedRuntime(FakeRuntime): + def synthesize( + self, + text: str, + voice: str | None, + language: str | None, + **admission, + ): + admission["admitted"].set() + time.sleep(0.15) # generation outlasts the 0.05 s admission budget + return [np.array([0.0, 0.5, -0.5], dtype=np.float32)], 24_000 + + app = create_app(SlowAdmittedRuntime(), _health(), synthesis_admission_timeout_seconds=0.05) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post("/v1/audio/speech", json={"input": "langsam"}) + + assert response.status_code == 200 + assert response.headers["content-type"] == "audio/wav" + + +async def test_stable_absolute_deadline_rejects_never_admitted_late_success() -> None: + # A worker that never signals admission is still bounded by the deadline: + # it may be stuck in the executor queue with the caller long gone, so a + # result that arrives late is discarded and the model is cancelled. class LateSuccessRuntime(_StableRuntime): def __init__(self) -> None: super().__init__() diff --git a/tests/test_tts_clone_runtime.py b/tests/test_tts_clone_runtime.py index 132f4aa..2424153 100644 --- a/tests/test_tts_clone_runtime.py +++ b/tests/test_tts_clone_runtime.py @@ -286,6 +286,63 @@ def generate_voice_clone(self, **kwargs): assert model.generated_texts == ["active", "following"] +def test_synthesize_signals_admission_before_generation(tmp_path: Path) -> None: + # The request path stops enforcing the admission deadline once this event + # is set, so it must be set only after the model lock is held and never + # before generation starts. + profile = _profile(tmp_path) + admitted = threading.Event() + admitted_at_generation: bool | None = None + + class SignalCheckingModel(ExactCloneModel): + def generate_voice_clone(self, **kwargs): + nonlocal admitted_at_generation + admitted_at_generation = admitted.is_set() + return self.warm_audio, self.sample_rate + + model = SignalCheckingModel(profile) + runtime = CloneRuntime(model, {profile.profile_id: profile}, profile.profile_id) + + runtime.synthesize("hallo", profile.profile_id, "de", admitted=admitted) + + assert admitted_at_generation is True + assert admitted.is_set() + + +def test_admission_timeout_leaves_admission_unsignalled(tmp_path: Path) -> None: + profile = _profile(tmp_path) + admitted = threading.Event() + active_started = threading.Event() + release_active = threading.Event() + + class BlockingSynthesisModel(ExactCloneModel): + def generate_voice_clone(self, **kwargs): + if kwargs["text"] == "active": + active_started.set() + release_active.wait(timeout=2) + return self.warm_audio, self.sample_rate + + model = BlockingSynthesisModel(profile) + runtime = CloneRuntime(model, {profile.profile_id: profile}, profile.profile_id) + + with ThreadPoolExecutor(max_workers=2) as executor: + active = executor.submit(runtime.synthesize, "active", profile.profile_id, "de") + assert active_started.wait(timeout=1) + timed_out = executor.submit( + runtime.synthesize, + "timed-out", + profile.profile_id, + "de", + lock_timeout=0.05, + admitted=admitted, + ) + with pytest.raises(SynthesisAdmissionTimeout): + timed_out.result(timeout=1) + assert not admitted.is_set() + release_active.set() + active.result(timeout=1) + + def test_clone_runtime_holds_lock_for_complete_stream_lifetime(tmp_path: Path) -> None: profile = _profile(tmp_path) first_chunk = threading.Event()