From c7fd50084903d87fb8f3400f8858e76bc38ae49c Mon Sep 17 00:00:00 2001 From: nullraptor Date: Sat, 3 Oct 2026 20:56:16 +0000 Subject: [PATCH] asr: bad audio fails the request instead of unloading the model uploading something whisper or qwen3-asr can't decode raised inside the worker, which marks it an inference failure and unloads the model. one garbage upload took the model down until someone reloaded it. decode errors now raise AudioDecodeError (a ValueError, so the route returns 400) and the worker fails just that request via packet.request_error. the model stays loaded. --- src/engine/audio.py | 5 +++ .../openvino/qwen3_asr/qwen3_asr_utils.py | 11 +++++ src/engine/ov_genai/whisper.py | 6 ++- src/server/worker_registry.py | 13 ++++++ tests/unit/test_audio_bad_input_unit.py | 16 ++++++++ tests/unit/test_ov_genai_whisper_unit.py | 7 ++++ tests/unit/test_worker_registry_unit.py | 41 +++++++++++++++++++ 7 files changed, 98 insertions(+), 1 deletion(-) create mode 100644 src/engine/audio.py create mode 100644 tests/unit/test_audio_bad_input_unit.py diff --git a/src/engine/audio.py b/src/engine/audio.py new file mode 100644 index 00000000..cf650800 --- /dev/null +++ b/src/engine/audio.py @@ -0,0 +1,5 @@ +"""Audio helpers shared by the speech engines.""" + + +class AudioDecodeError(ValueError): + """The request carried audio that cannot be decoded: a client error, not a model failure.""" diff --git a/src/engine/openvino/qwen3_asr/qwen3_asr_utils.py b/src/engine/openvino/qwen3_asr/qwen3_asr_utils.py index c60c3e1d..6f89f559 100644 --- a/src/engine/openvino/qwen3_asr/qwen3_asr_utils.py +++ b/src/engine/openvino/qwen3_asr/qwen3_asr_utils.py @@ -24,6 +24,8 @@ import numpy as np import soundfile as sf +from src.engine.audio import AudioDecodeError + AudioLike = Union[ str, # wav path / URL / base64 Tuple[np.ndarray, int], # (waveform, sr) @@ -198,6 +200,15 @@ def decode_base64_bytes(b64: str) -> bytes: def load_audio_any(x: str) -> Tuple[np.ndarray, int]: + try: + return _load_audio_any(x) + except AudioDecodeError: + raise + except Exception as e: + raise AudioDecodeError(f"Unreadable audio: {e}") from e + + +def _load_audio_any(x: str) -> Tuple[np.ndarray, int]: if is_url(x): with urllib.request.urlopen(x) as resp: audio_bytes = resp.read() diff --git a/src/engine/ov_genai/whisper.py b/src/engine/ov_genai/whisper.py index db5814c5..b12af9ff 100644 --- a/src/engine/ov_genai/whisper.py +++ b/src/engine/ov_genai/whisper.py @@ -9,6 +9,7 @@ import numpy as np from openvino_genai import WhisperPipeline +from src.engine.audio import AudioDecodeError from src.server.model_registry import ModelRegistry from src.server.schemas.registration import ModelLoadConfig from src.server.schemas.modeling.contract_whisper import OVGenAI_WhisperGenConfig @@ -31,7 +32,10 @@ def prepare_audio(self, gen_config: OVGenAI_WhisperGenConfig) -> list[float]: audio_buffer = io.BytesIO(audio_bytes) - audio, sr = librosa.load(audio_buffer, sr=16000, mono=True) + try: + audio, sr = librosa.load(audio_buffer, sr=16000, mono=True) + except Exception as e: + raise AudioDecodeError(f"Unreadable audio: {e}") from e return audio.astype(np.float32).tolist() diff --git a/src/server/worker_registry.py b/src/server/worker_registry.py index 2400dd11..df77099b 100644 --- a/src/server/worker_registry.py +++ b/src/server/worker_registry.py @@ -12,6 +12,7 @@ from src.engine.ov_genai.llm import OVGenAI_LLM from src.engine.ov_genai.vlm import OVGenAI_VLM from src.engine.ov_genai.whisper import OVGenAI_Whisper +from src.engine.audio import AudioDecodeError from src.engine.openvino.kokoro import OV_Kokoro from src.engine.openvino.qwen3_asr.qwen3_asr import OVQwen3ASR from src.engine.openvino.qwen3_tts.qwen3_tts import OVQwen3TTS @@ -70,6 +71,8 @@ class WorkerPacket: response: Optional[str] = None metrics: Optional[Dict[str, Any]] = None segments: Optional[List[Dict[str, Any]]] = None + # A bad request (e.g. undecodable audio): fail this request, keep the model loaded. + request_error: Optional[BaseException] = None # Orchestration plumbing result_future: Optional[asyncio.Future] = None stream_queue: Optional[asyncio.Queue] = None @@ -96,6 +99,10 @@ def _commit_completed_packet( registry: ModelRegistry, ) -> bool: """Complete the request future. Return True if the worker should exit.""" + if completed.request_error is not None: + if packet.result_future is not None and not packet.result_future.done(): + packet.result_future.set_exception(completed.request_error) + return False if completed.error is not None: logger.error( f"[{model_name}] Inference failed, triggering model unload..." @@ -214,6 +221,9 @@ async def infer_whisper(packet: WorkerPacket, whisper_model: OVGenAI_Whisper) -> packet.response = final_text packet.metrics = metrics + except AudioDecodeError as e: + logger.warning(f"Whisper request rejected: {e}") + packet.request_error = e except Exception as e: logger.error("Whisper inference failed!", exc_info=True) _mark_inference_error(packet, e) @@ -228,6 +238,9 @@ async def infer_qwen3_asr(packet: WorkerPacket, asr_model: OVQwen3ASR) -> Worker try: assert isinstance(packet.gen_config, OV_Qwen3ASRGenConfig), "Expected OV_Qwen3ASRGenConfig for Qwen3 ASR inference" packet.response, packet.metrics, packet.segments = await asr_model.transcribe(packet.gen_config) + except AudioDecodeError as e: + logger.warning(f"Qwen3 ASR request rejected: {e}") + packet.request_error = e except Exception as e: logger.error("Qwen3 ASR inference failed!", exc_info=True) _mark_inference_error(packet, e) diff --git a/tests/unit/test_audio_bad_input_unit.py b/tests/unit/test_audio_bad_input_unit.py new file mode 100644 index 00000000..8b823e80 --- /dev/null +++ b/tests/unit/test_audio_bad_input_unit.py @@ -0,0 +1,16 @@ +import base64 + +import pytest # type: ignore[import] + +from src.engine.audio import AudioDecodeError +from src.engine.openvino.qwen3_asr.qwen3_asr_utils import load_audio_any + + +def test_qwen3_load_audio_rejects_garbage_base64() -> None: + with pytest.raises(AudioDecodeError, match="Unreadable audio"): + load_audio_any(base64.b64encode(b"not audio" * 100).decode()) + + +def test_decode_error_is_a_value_error() -> None: + # routes map ValueError to HTTP 400 + assert issubclass(AudioDecodeError, ValueError) diff --git a/tests/unit/test_ov_genai_whisper_unit.py b/tests/unit/test_ov_genai_whisper_unit.py index ff2d2f41..155caede 100644 --- a/tests/unit/test_ov_genai_whisper_unit.py +++ b/tests/unit/test_ov_genai_whisper_unit.py @@ -152,3 +152,10 @@ def test_unload_model_resets_state(monkeypatch: pytest.MonkeyPatch, load_config: registry.register_unload.assert_called_once_with("model-name") gc_mock.assert_called_once() + +def test_prepare_audio_rejects_garbage(load_config: ModelLoadConfig) -> None: + from src.engine.audio import AudioDecodeError + + whisper = OVGenAI_Whisper(load_config) + with pytest.raises(AudioDecodeError, match="Unreadable audio"): + whisper.prepare_audio(OVGenAI_WhisperGenConfig(audio_base64=base64.b64encode(b"not audio" * 100).decode())) diff --git a/tests/unit/test_worker_registry_unit.py b/tests/unit/test_worker_registry_unit.py index 415f14ae..50533551 100644 --- a/tests/unit/test_worker_registry_unit.py +++ b/tests/unit/test_worker_registry_unit.py @@ -1,4 +1,5 @@ import asyncio +from unittest.mock import AsyncMock, MagicMock import pytest # type: ignore[import] @@ -402,3 +403,43 @@ def test_missing_model_queue(worker_registry: worker_module.WorkerRegistry) -> N with pytest.raises(ValueError): worker_registry._get_model_queue("missing") + + +def test_undecodable_audio_fails_whisper_and_qwen3_requests_not_models() -> None: + from src.engine.audio import AudioDecodeError + + class BadWhisper: + async def transcribe(self, cfg): + raise AudioDecodeError("Unreadable audio: nope") + yield # pragma: no cover + + class BadQwen3: + async def transcribe(self, cfg): + raise AudioDecodeError("Unreadable audio: nope") + + w = worker_module.WorkerPacket(request_id="w", id_model="whisper", gen_config=OVGenAI_WhisperGenConfig(audio_base64="AAA")) + done = asyncio.run(worker_module.InferWorker.infer_whisper(w, BadWhisper())) # type: ignore[arg-type] + assert done.error is None and isinstance(done.request_error, AudioDecodeError) + + q = worker_module.WorkerPacket(request_id="q", id_model="qwen3", gen_config=OV_Qwen3ASRGenConfig(audio_base64="AAA")) + done = asyncio.run(worker_module.InferWorker.infer_qwen3_asr(q, BadQwen3())) # type: ignore[arg-type] + assert done.error is None and isinstance(done.request_error, AudioDecodeError) + + +def test_commit_request_error_fails_future_and_keeps_worker() -> None: + async def _run(): + registry = MagicMock() + registry.register_unload = AsyncMock() + fut = asyncio.get_running_loop().create_future() + packet = worker_module.WorkerPacket(request_id="r", id_model="m", gen_config=OVGenAI_WhisperGenConfig(audio_base64="AAA"), + result_future=fut) + packet.request_error = ValueError("bad audio") + exit_worker = worker_module._commit_completed_packet(packet, packet, "m", registry) + await asyncio.sleep(0) + return exit_worker, fut, registry + + exit_worker, fut, registry = asyncio.run(_run()) + assert exit_worker is False + with pytest.raises(ValueError, match="bad audio"): + fut.result() + registry.register_unload.assert_not_called()