From f4f0d3631545f492eb30e972fb280d77efa73abd Mon Sep 17 00:00:00 2001 From: Prakruthi B Gowda Date: Sun, 30 Aug 2026 19:03:19 +0530 Subject: [PATCH 1/3] feat: add automatic recovery for streaming ASR/TTS Add resilient streaming wrappers that automatically reconnect on transient gRPC failures (UNAVAILABLE, DEADLINE_EXCEEDED, INTERNAL, etc.). - riva/client/retry.py: shared retry utilities with exponential backoff - riva/client/asr.py: ResilientStreamingASR with audio lookback buffer and final-transcript deduplication - riva/client/tts.py: ResilientStreamingTTS with segment-level retry - riva/client/auth.py: default gRPC keepalive for faster dead-connection detection - scripts/asr/transcribe_file.py: --auto-recover, --max-retries, --lookback-seconds - scripts/tts/talk.py: --auto-recover, --max-retries - tests/unit/test_retry.py: unit tests for retry logic --- riva/client/__init__.py | 9 +- riva/client/asr.py | 183 +++++++++++++++++++++++++++++++++++++++ riva/client/auth.py | 13 +++ riva/client/retry.py | 150 ++++++++++++++++++++++++++++++++ riva/client/tts.py | 155 ++++++++++++++++++++++++++++++++- tests/unit/test_retry.py | 116 +++++++++++++++++++++++++ 6 files changed, 624 insertions(+), 2 deletions(-) create mode 100644 riva/client/retry.py create mode 100644 tests/unit/test_retry.py diff --git a/riva/client/__init__.py b/riva/client/__init__.py index 7656bd69..6feab24a 100644 --- a/riva/client/__init__.py +++ b/riva/client/__init__.py @@ -4,6 +4,7 @@ from riva.client.asr import ( AudioChunkFileIterator, ASRService, + ResilientStreamingASR, add_audio_file_specs_to_config, add_word_boosting_to_config, add_speaker_diarization_to_config, @@ -39,5 +40,11 @@ from riva.client.proto.riva_audio_pb2 import AudioEncoding from riva.client.proto.riva_nlp_pb2 import AnalyzeIntentOptions from riva.client.proto.riva_nmt_pb2 import StreamingTranslateSpeechToSpeechConfig, TranslationConfig, SynthesizeSpeechConfig, StreamingTranslateSpeechToTextConfig -from riva.client.tts import SpeechSynthesisService +from riva.client.retry import ( + RETRYABLE_GRPC_CODES, + exponential_backoff, + is_retryable_grpc_error, + retry_streaming, +) +from riva.client.tts import SpeechSynthesisService, ResilientStreamingTTS from riva.client.nmt import NeuralMachineTranslationClient diff --git a/riva/client/asr.py b/riva/client/asr.py index be9a362e..a8abe70a 100644 --- a/riva/client/asr.py +++ b/riva/client/asr.py @@ -483,3 +483,186 @@ def offline_recognize( request = rasr.RecognizeRequest(config=config, audio=audio_bytes) func = self.stub.Recognize.future if future else self.stub.Recognize return func(request, metadata=self.auth.get_auth_metadata()) + + + +import hashlib +import logging +from collections import deque +from typing import Set + +import grpc + +from riva.client.retry import is_retryable_grpc_error, exponential_backoff + +LOGGER = logging.getLogger(__name__) + + +class ResilientStreamingASR: + """A resilient wrapper around :class:`ASRService` for streaming recognition. + + This class buffers recent audio and automatically reconnects on transient + gRPC failures, replaying buffered audio so that recognition can continue + with minimal data loss. Final transcripts are deduplicated across + reconnections. + + Example: + >>> auth = Auth(uri="localhost:50051") + >>> asr = ASRService(auth) + >>> config = StreamingRecognitionConfig( + ... config=RecognitionConfig(enable_automatic_punctuation=True), + ... interim_results=True, + ... ) + >>> resilient_asr = ResilientStreamingASR(asr, config) + >>> for response in resilient_asr.stream(audio_chunks): + ... print(response) + + Args: + asr_service: The underlying :class:`ASRService` instance. + streaming_config: Configuration for streaming recognition. + max_retries: Maximum number of reconnection attempts per failure. + lookback_seconds: Duration of audio to replay after reconnecting. + A larger value improves recovery at the cost of higher latency. + sample_rate_hz: Sample rate of the audio stream (used to size the + lookback buffer). + base_delay: Initial backoff delay in seconds. + max_delay: Maximum backoff delay in seconds. + """ + + def __init__( + self, + asr_service: ASRService, + streaming_config: rasr.StreamingRecognitionConfig, + max_retries: int = 3, + lookback_seconds: float = 2.0, + sample_rate_hz: int = 16000, + base_delay: float = 1.0, + max_delay: float = 60.0, + ) -> None: + self.asr_service = asr_service + self.streaming_config = streaming_config + self.max_retries = max_retries + self.base_delay = base_delay + self.max_delay = max_delay + + # Size the lookback buffer in chunks. We assume a typical chunk + # duration of ~0.1 s (e.g. 1600 frames at 16 kHz). The exact + # chunk size does not matter for correctness; we simply keep a + # bounded number of recent chunks. + self._lookback_chunk_count = max(1, int(lookback_seconds * 10)) + self._audio_buffer: deque = deque(maxlen=self._lookback_chunk_count) + self._finalized_hashes: Set[str] = set() + self._retry_count = 0 + + def _buffered_request_generator( + self, + audio_source: Iterable[bytes], + ) -> Generator[rasr.StreamingRecognizeRequest, None, None]: + """Yield the config message, buffered audio, then new audio. + + Each chunk from *audio_source* is appended to the lookback buffer + before being yielded so that it is available for the next reconnect. + """ + yield rasr.StreamingRecognizeRequest(streaming_config=self.streaming_config) + + # Replay buffered audio from previous (partial) stream + for chunk in self._audio_buffer: + yield rasr.StreamingRecognizeRequest(audio_content=chunk) + + for chunk in audio_source: + self._audio_buffer.append(chunk) + yield rasr.StreamingRecognizeRequest(audio_content=chunk) + + def _is_duplicate(self, response: rasr.StreamingRecognizeResponse) -> bool: + """Return ``True`` if every final transcript in *response* was already emitted.""" + if not response.results: + return False + all_final = True + for result in response.results: + if not result.is_final: + all_final = False + continue + if not result.alternatives: + continue + transcript = result.alternatives[0].transcript + h = hashlib.sha256(transcript.encode("utf-8")).hexdigest() + if h not in self._finalized_hashes: + return False + # Only consider it a duplicate if *all* results are final and known. + return all_final and len(response.results) > 0 + + def _record_final(self, response: rasr.StreamingRecognizeResponse) -> None: + """Store hashes of any new final transcripts.""" + for result in response.results: + if result.is_final and result.alternatives: + transcript = result.alternatives[0].transcript + h = hashlib.sha256(transcript.encode("utf-8")).hexdigest() + self._finalized_hashes.add(h) + + def stream( + self, + audio_source: Iterable[bytes], + ) -> Generator[rasr.StreamingRecognizeResponse, None, None]: + """Stream audio for recognition with automatic recovery. + + Args: + audio_source: An iterable of raw audio chunks. + + Yields: + :obj:`StreamingRecognizeResponse` objects. On a successful + reconnect, duplicate final transcripts are suppressed. + + Raises: + :obj:`grpc.RpcError`: If a non-retryable error occurs or the + maximum number of retries is exceeded. + """ + audio_iterator = iter(audio_source) + attempt = 0 + last_exception: Optional[grpc.RpcError] = None + + while True: + try: + generator = self._buffered_request_generator(audio_iterator) + for response in self.asr_service.stub.StreamingRecognize( + generator, metadata=self.asr_service.auth.get_auth_metadata() + ): + if self._is_duplicate(response): + LOGGER.debug("Suppressing duplicate final transcript after reconnect.") + continue + self._record_final(response) + yield response + # Stream completed normally. + if self._retry_count > 0: + LOGGER.info("Streaming ASR recovered after %d retry(s).", self._retry_count) + return + + except grpc.RpcError as exc: + last_exception = exc + if not is_retryable_grpc_error(exc): + LOGGER.warning( + "Non-retryable gRPC error in streaming ASR: %s – %s", + exc.code() if hasattr(exc, "code") else "UNKNOWN", + exc.details() if hasattr(exc, "details") else str(exc), + ) + raise + if attempt >= self.max_retries: + LOGGER.error( + "Streaming ASR failed permanently after %d retries. Last error: %s", + self.max_retries, + exc.details() if hasattr(exc, "details") else str(exc), + ) + raise + delay = exponential_backoff(attempt, self.base_delay, self.max_delay) + LOGGER.info( + "Streaming ASR connection lost (%s). Reconnecting in %.2f s " + "(attempt %d/%d).", + exc.code(), + delay, + attempt + 1, + self.max_retries, + ) + self._retry_count += 1 + attempt += 1 + time.sleep(delay) + # Loop continues: _buffered_request_generator will replay + # self._audio_buffer and then consume audio_iterator. diff --git a/riva/client/auth.py b/riva/client/auth.py index 8a4688d4..a20177bd 100644 --- a/riva/client/auth.py +++ b/riva/client/auth.py @@ -17,6 +17,19 @@ def create_channel( options: Optional[List[Tuple[str, str]]] = [], use_aio: Optional[bool] = False, ) -> grpc.Channel: + """Create a gRPC channel with sensible defaults for resilient streaming. + + Default keepalive settings are injected so that dead connections are + detected quickly, enabling faster recovery on transient failures. + """ + default_options = [ + ("grpc.keepalive_time_ms", "10000"), + ("grpc.keepalive_timeout_ms", "5000"), + ("grpc.keepalive_permit_without_calls", "1"), + ("grpc.http2.max_pings_without_data", "0"), + ] + options = default_options + (options or []) + def metadata_callback(context, callback): callback(metadata, None) diff --git a/riva/client/retry.py b/riva/client/retry.py new file mode 100644 index 00000000..495bcf7b --- /dev/null +++ b/riva/client/retry.py @@ -0,0 +1,150 @@ +# SPDX-FileCopyrightText: Copyright (c) 2022 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: MIT + +"""Shared retry utilities for resilient streaming ASR and TTS clients. + +This module provides constants, helpers, and decorators for handling transient +gRPC failures with exponential backoff. It is designed to be used by both +:mod:`riva.client.asr` and :mod:`riva.client.tts`. +""" + +import functools +import logging +import random +import time +from typing import Any, Callable, Optional, Tuple, TypeVar + +import grpc + +LOGGER = logging.getLogger(__name__) + +# gRPC status codes that are generally considered transient and safe to retry. +RETRYABLE_GRPC_CODES = frozenset({ + grpc.StatusCode.UNAVAILABLE, + grpc.StatusCode.DEADLINE_EXCEEDED, + grpc.StatusCode.INTERNAL, + grpc.StatusCode.RESOURCE_EXHAUSTED, + grpc.StatusCode.ABORTED, +}) + +# gRPC status codes that should NEVER be retried (client-side errors). +NON_RETRYABLE_GRPC_CODES = frozenset({ + grpc.StatusCode.INVALID_ARGUMENT, + grpc.StatusCode.PERMISSION_DENIED, + grpc.StatusCode.UNAUTHENTICATED, + grpc.StatusCode.NOT_FOUND, + grpc.StatusCode.ALREADY_EXISTS, + grpc.StatusCode.FAILED_PRECONDITION, + grpc.StatusCode.OUT_OF_RANGE, + grpc.StatusCode.UNIMPLEMENTED, +}) + +F = TypeVar("F", bound=Callable[..., Any]) + + +def is_retryable_grpc_error(exc: grpc.RpcError) -> bool: + """Return ``True`` if *exc* is a transient gRPC error that is safe to retry. + + Args: + exc: The exception raised by a gRPC call. + + Returns: + ``True`` if the error code is in :data:`RETRYABLE_GRPC_CODES`. + """ + code = exc.code() if hasattr(exc, "code") else None + if code is None: + return False + return code in RETRYABLE_GRPC_CODES + + +def exponential_backoff( + attempt: int, + base_delay: float = 1.0, + max_delay: float = 60.0, + jitter: bool = True, +) -> float: + """Compute a sleep duration for the *attempt*-th retry. + + Uses capped exponential backoff with optional full jitter to avoid + thundering-herd behaviour when many clients reconnect simultaneously. + + Args: + attempt: Zero-based retry attempt number. + base_delay: Initial delay in seconds. + max_delay: Upper bound for the delay in seconds. + jitter: If ``True``, multiply the delay by a random factor in ``[0, 1)``. + + Returns: + The number of seconds to sleep before the next attempt. + """ + delay = min(base_delay * (2 ** attempt), max_delay) + if jitter: + delay = delay * random.random() + return delay + + +def retry_streaming( + max_retries: int = 3, + base_delay: float = 1.0, + max_delay: float = 60.0, + on_retry: Optional[Callable[[grpc.RpcError, int, float], None]] = None, +) -> Callable[[F], F]: + """Decorator that retries a streaming gRPC call on transient failures. + + The decorated function must be a generator (i.e. use ``yield``). When a + :class:`grpc.RpcError` with a retryable code is raised, the generator is + closed and the function is re-invoked up to *max_retries* times. + + Args: + max_retries: Maximum number of retry attempts after the initial failure. + base_delay: Initial backoff delay in seconds. + max_delay: Maximum backoff delay in seconds. + on_retry: Optional callback ``fn(exc, attempt, delay)`` invoked before + each retry sleep. + + Returns: + A decorator that wraps generator functions with retry logic. + """ + def decorator(func: F) -> F: + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + last_exception: Optional[grpc.RpcError] = None + for attempt in range(max_retries + 1): + try: + yield from func(*args, **kwargs) + return + except grpc.RpcError as exc: + last_exception = exc + if not is_retryable_grpc_error(exc): + LOGGER.warning( + "Non-retryable gRPC error %s: %s", + exc.code() if hasattr(exc, "code") else "UNKNOWN", + exc.details() if hasattr(exc, "details") else str(exc), + ) + raise + if attempt >= max_retries: + LOGGER.error( + "Max retries (%d) exceeded for %s. Last error: %s", + max_retries, + func.__name__, + exc.details() if hasattr(exc, "details") else str(exc), + ) + raise + delay = exponential_backoff(attempt, base_delay, max_delay) + LOGGER.info( + "Retryable gRPC error %s on attempt %d/%d for %s. " + "Sleeping %.2f s before retry.", + exc.code(), + attempt + 1, + max_retries + 1, + func.__name__, + delay, + ) + if on_retry is not None: + on_retry(exc, attempt + 1, delay) + time.sleep(delay) + # Should never reach here, but satisfy type checker. + if last_exception is not None: + raise last_exception + return wrapper # type: ignore[return-value] + return decorator diff --git a/riva/client/tts.py b/riva/client/tts.py index c0579563..d6239de6 100644 --- a/riva/client/tts.py +++ b/riva/client/tts.py @@ -1,8 +1,10 @@ # SPDX-FileCopyrightText: Copyright (c) 2022 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: MIT +import time from typing import Dict, Generator, Optional, Union, Iterable +import grpc from grpc._channel import _MultiThreadedRendezvous import riva.client.proto.riva_tts_pb2 as rtts @@ -217,4 +219,155 @@ def request_generator(text): else: raise ValueError(f"Invalid text type: {type(text)}") - return self.stub.SynthesizeOnline(request_generator(text), metadata=self.auth.get_auth_metadata()) \ No newline at end of file + return self.stub.SynthesizeOnline(request_generator(text), metadata=self.auth.get_auth_metadata()) + + +import logging +from typing import Iterator + +import grpc + +from riva.client.retry import is_retryable_grpc_error, exponential_backoff + +LOGGER = logging.getLogger(__name__) + + +class ResilientStreamingTTS: + """A resilient wrapper around :class:`SpeechSynthesisService` for streaming TTS. + + This class retries individual text segments on transient gRPC failures, + yielding audio chunks as they arrive. It is designed for long-running + streaming synthesis where network blips should not terminate the session. + + Example: + >>> auth = Auth(uri="localhost:50051") + >>> tts = SpeechSynthesisService(auth) + >>> resilient_tts = ResilientStreamingTTS(tts) + >>> for audio_chunk in resilient_tts.synthesize_stream( + ... ["Hello world", "This is a test."], + ... voice_name="English-US-Female-1", + ... ): + ... play_audio(audio_chunk) + + Args: + tts_service: The underlying :class:`SpeechSynthesisService` instance. + max_retries: Maximum number of retry attempts per text segment. + base_delay: Initial backoff delay in seconds. + max_delay: Maximum backoff delay in seconds. + """ + + def __init__( + self, + tts_service: SpeechSynthesisService, + max_retries: int = 3, + base_delay: float = 1.0, + max_delay: float = 60.0, + ) -> None: + self.tts_service = tts_service + self.max_retries = max_retries + self.base_delay = base_delay + self.max_delay = max_delay + self._retry_count = 0 + + def synthesize_stream( + self, + text_segments: Union[str, list[str], Iterable[str]], + voice_name: Optional[str] = None, + language_code: str = 'en-US', + encoding: AudioEncoding = AudioEncoding.LINEAR_PCM, + sample_rate_hz: int = 22050, + zero_shot_audio_prompt_file: Optional[str] = None, + audio_prompt_encoding: AudioEncoding = AudioEncoding.ENCODING_UNSPECIFIED, + zero_shot_quality: int = 20, + custom_dictionary: Optional[dict] = None, + custom_configuration: Optional[Dict[str, str]] = None, + enable_word_time_offsets: Optional[bool] = None, + ) -> Generator[rtts.SynthesizeSpeechResponse, None, None]: + """Synthesize speech from text segments with automatic recovery. + + Each text segment is sent independently. If the gRPC stream fails + while synthesizing a segment, that segment is retried up to + *max_retries* times before the error is propagated. + + Args: + text_segments: Input text. A single string, a list, or any iterable + of strings. Each element is treated as one retryable unit. + voice_name: See :meth:`SpeechSynthesisService.synthesize_online`. + language_code: See :meth:`SpeechSynthesisService.synthesize_online`. + encoding: See :meth:`SpeechSynthesisService.synthesize_online`. + sample_rate_hz: See :meth:`SpeechSynthesisService.synthesize_online`. + zero_shot_audio_prompt_file: See :meth:`SpeechSynthesisService.synthesize_online`. + audio_prompt_encoding: See :meth:`SpeechSynthesisService.synthesize_online`. + zero_shot_quality: See :meth:`SpeechSynthesisService.synthesize_online`. + custom_dictionary: See :meth:`SpeechSynthesisService.synthesize_online`. + custom_configuration: See :meth:`SpeechSynthesisService.synthesize_online`. + enable_word_time_offsets: See :meth:`SpeechSynthesisService.synthesize_online`. + + Yields: + :obj:`SynthesizeSpeechResponse` objects containing audio chunks. + + Raises: + :obj:`grpc.RpcError`: If a non-retryable error occurs or the + maximum number of retries is exceeded for a segment. + """ + # Normalise input to an iterator of strings. + if isinstance(text_segments, str): + segment_iter: Iterator[str] = iter([text_segments]) + else: + segment_iter = iter(text_segments) + + for segment in segment_iter: + attempt = 0 + last_exception: Optional[grpc.RpcError] = None + + while True: + try: + responses = self.tts_service.synthesize_online( + text=segment, + voice_name=voice_name, + language_code=language_code, + encoding=encoding, + sample_rate_hz=sample_rate_hz, + zero_shot_audio_prompt_file=zero_shot_audio_prompt_file, + audio_prompt_encoding=audio_prompt_encoding, + zero_shot_quality=zero_shot_quality, + custom_dictionary=custom_dictionary, + custom_configuration=custom_configuration, + enable_word_time_offsets=enable_word_time_offsets, + ) + for resp in responses: + yield resp + break # Segment completed successfully. + + except grpc.RpcError as exc: + last_exception = exc + if not is_retryable_grpc_error(exc): + LOGGER.warning( + "Non-retryable gRPC error in streaming TTS: %s – %s", + exc.code() if hasattr(exc, "code") else "UNKNOWN", + exc.details() if hasattr(exc, "details") else str(exc), + ) + raise + if attempt >= self.max_retries: + LOGGER.error( + "Streaming TTS failed permanently after %d retries for segment %r. " + "Last error: %s", + self.max_retries, + segment, + exc.details() if hasattr(exc, "details") else str(exc), + ) + raise + delay = exponential_backoff(attempt, self.base_delay, self.max_delay) + LOGGER.info( + "Streaming TTS connection lost (%s) on segment %r. " + "Retrying in %.2f s (attempt %d/%d).", + exc.code(), + segment, + delay, + attempt + 1, + self.max_retries, + ) + self._retry_count += 1 + attempt += 1 + time.sleep(delay) + # Loop continues: retry the same segment. diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py new file mode 100644 index 00000000..bcf192de --- /dev/null +++ b/tests/unit/test_retry.py @@ -0,0 +1,116 @@ +# SPDX-FileCopyrightText: Copyright (c) 2022 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: MIT + +import time +from unittest.mock import Mock + +import grpc +import pytest + +from riva.client.retry import ( + RETRYABLE_GRPC_CODES, + exponential_backoff, + is_retryable_grpc_error, + retry_streaming, +) + + +class FakeRpcError(grpc.RpcError): + def __init__(self, code): + self._code = code + + def code(self): + return self._code + + def details(self): + return "fake error" + + +class TestIsRetryableGrpcError: + def test_retryable_codes(self): + for code in RETRYABLE_GRPC_CODES: + exc = FakeRpcError(code) + assert is_retryable_grpc_error(exc) is True + + def test_non_retryable_code(self): + exc = FakeRpcError(grpc.StatusCode.INVALID_ARGUMENT) + assert is_retryable_grpc_error(exc) is False + + def test_no_code_method(self): + exc = Exception("plain exception") + assert is_retryable_grpc_error(exc) is False + + +class TestExponentialBackoff: + def test_no_jitter_growth(self): + assert exponential_backoff(0, base_delay=1.0, jitter=False) == 1.0 + assert exponential_backoff(1, base_delay=1.0, jitter=False) == 2.0 + assert exponential_backoff(2, base_delay=1.0, jitter=False) == 4.0 + + def test_max_delay_cap(self): + assert exponential_backoff(10, base_delay=1.0, max_delay=8.0, jitter=False) == 8.0 + + def test_jitter_reduces_delay(self): + for _ in range(20): + d = exponential_backoff(2, base_delay=1.0, jitter=True) + assert 0.0 <= d < 4.0 + + +class TestRetryStreamingDecorator: + def test_success_no_retry(self): + @retry_streaming(max_retries=2) + def gen(): + yield 1 + yield 2 + + assert list(gen()) == [1, 2] + + def test_retries_then_succeeds(self): + call_count = 0 + + @retry_streaming(max_retries=2, base_delay=0.01) + def gen(): + nonlocal call_count + call_count += 1 + if call_count < 2: + raise FakeRpcError(grpc.StatusCode.UNAVAILABLE) + yield "ok" + + assert list(gen()) == ["ok"] + assert call_count == 2 + + def test_exhausts_retries(self): + @retry_streaming(max_retries=1, base_delay=0.01) + def gen(): + raise FakeRpcError(grpc.StatusCode.UNAVAILABLE) + yield # make it a generator + + with pytest.raises(grpc.RpcError): + list(gen()) + + def test_non_retryable_raises_immediately(self): + @retry_streaming(max_retries=3) + def gen(): + raise FakeRpcError(grpc.StatusCode.INVALID_ARGUMENT) + yield + + with pytest.raises(grpc.RpcError): + list(gen()) + + def test_on_retry_callback(self): + callback_log = [] + + def on_retry(exc, attempt, delay): + callback_log.append((exc.code(), attempt, delay)) + + @retry_streaming(max_retries=1, base_delay=0.01, on_retry=on_retry) + def gen(): + raise FakeRpcError(grpc.StatusCode.UNAVAILABLE) + yield + + with pytest.raises(grpc.RpcError): + list(gen()) + + assert len(callback_log) == 1 + assert callback_log[0][0] == grpc.StatusCode.UNAVAILABLE + assert callback_log[0][1] == 1 From dbe22a36691cc6286283fde7cdc4cc3509b99d99 Mon Sep 17 00:00:00 2001 From: Prakruthi B Gowda Date: Sun, 30 Aug 2026 19:47:25 +0530 Subject: [PATCH 2/3] fix: make streaming recovery bounded and safe --- riva/client/__init__.py | 1 - riva/client/asr.py | 93 +++++++++++++--------------------- riva/client/auth.py | 21 ++------ riva/client/retry.py | 78 +--------------------------- riva/client/tts.py | 28 +++++----- scripts/asr/transcribe_file.py | 30 +++++++++-- scripts/tts/talk.py | 16 +++++- tests/unit/test_retry.py | 77 +++++++--------------------- 8 files changed, 111 insertions(+), 233 deletions(-) diff --git a/riva/client/__init__.py b/riva/client/__init__.py index 6feab24a..492a3af8 100644 --- a/riva/client/__init__.py +++ b/riva/client/__init__.py @@ -44,7 +44,6 @@ RETRYABLE_GRPC_CODES, exponential_backoff, is_retryable_grpc_error, - retry_streaming, ) from riva.client.tts import SpeechSynthesisService, ResilientStreamingTTS from riva.client.nmt import NeuralMachineTranslationClient diff --git a/riva/client/asr.py b/riva/client/asr.py index a8abe70a..29d7cd8f 100644 --- a/riva/client/asr.py +++ b/riva/client/asr.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: MIT import io +import logging import os import sys import time @@ -11,8 +12,10 @@ import wave from itertools import groupby from pathlib import Path -from typing import Callable, Dict, Generator, Iterable, List, Optional, TextIO, Union +from collections import deque +from typing import Callable, Deque, Dict, Generator, Iterable, List, Optional, TextIO, Union +import grpc from google.protobuf.json_format import MessageToJson from grpc._channel import _MultiThreadedRendezvous @@ -20,6 +23,10 @@ import riva.client.proto.riva_asr_pb2 as rasr import riva.client.proto.riva_asr_pb2_grpc as rasr_srv from riva.client.auth import Auth +from riva.client.retry import exponential_backoff, is_retryable_grpc_error + + +LOGGER = logging.getLogger(__name__) def get_wav_file_parameters(input_file: Union[str, os.PathLike]) -> Dict[str, Union[int, float]]: @@ -486,25 +493,14 @@ def offline_recognize( -import hashlib -import logging -from collections import deque -from typing import Set - -import grpc - -from riva.client.retry import is_retryable_grpc_error, exponential_backoff - -LOGGER = logging.getLogger(__name__) - - class ResilientStreamingASR: """A resilient wrapper around :class:`ASRService` for streaming recognition. This class buffers recent audio and automatically reconnects on transient gRPC failures, replaying buffered audio so that recognition can continue - with minimal data loss. Final transcripts are deduplicated across - reconnections. + with a bounded audio lookback. Recovery is best effort: audio older than + the configured lookback window may not be replayed, and applications must + tolerate repeated transcripts after a reconnect. Example: >>> auth = Auth(uri="localhost:50051") @@ -523,8 +519,9 @@ class ResilientStreamingASR: max_retries: Maximum number of reconnection attempts per failure. lookback_seconds: Duration of audio to replay after reconnecting. A larger value improves recovery at the cost of higher latency. - sample_rate_hz: Sample rate of the audio stream (used to size the - lookback buffer). + sample_rate_hz: Sample rate of linear PCM audio. + audio_channel_count: Number of PCM audio channels. + sample_width_bytes: Bytes per PCM sample. base_delay: Initial backoff delay in seconds. max_delay: Maximum backoff delay in seconds. """ @@ -536,6 +533,8 @@ def __init__( max_retries: int = 3, lookback_seconds: float = 2.0, sample_rate_hz: int = 16000, + audio_channel_count: int = 1, + sample_width_bytes: int = 2, base_delay: float = 1.0, max_delay: float = 60.0, ) -> None: @@ -545,13 +544,15 @@ def __init__( self.base_delay = base_delay self.max_delay = max_delay - # Size the lookback buffer in chunks. We assume a typical chunk - # duration of ~0.1 s (e.g. 1600 frames at 16 kHz). The exact - # chunk size does not matter for correctness; we simply keep a - # bounded number of recent chunks. - self._lookback_chunk_count = max(1, int(lookback_seconds * 10)) - self._audio_buffer: deque = deque(maxlen=self._lookback_chunk_count) - self._finalized_hashes: Set[str] = set() + if lookback_seconds <= 0: + raise ValueError("lookback_seconds must be greater than zero") + if min(sample_rate_hz, audio_channel_count, sample_width_bytes) <= 0: + raise ValueError("PCM format values must be greater than zero") + self._lookback_max_bytes = int( + lookback_seconds * sample_rate_hz * audio_channel_count * sample_width_bytes + ) + self._audio_buffer: Deque[bytes] = deque() + self._buffered_bytes = 0 self._retry_count = 0 def _buffered_request_generator( @@ -570,34 +571,14 @@ def _buffered_request_generator( yield rasr.StreamingRecognizeRequest(audio_content=chunk) for chunk in audio_source: - self._audio_buffer.append(chunk) + self._append_audio(chunk) yield rasr.StreamingRecognizeRequest(audio_content=chunk) - def _is_duplicate(self, response: rasr.StreamingRecognizeResponse) -> bool: - """Return ``True`` if every final transcript in *response* was already emitted.""" - if not response.results: - return False - all_final = True - for result in response.results: - if not result.is_final: - all_final = False - continue - if not result.alternatives: - continue - transcript = result.alternatives[0].transcript - h = hashlib.sha256(transcript.encode("utf-8")).hexdigest() - if h not in self._finalized_hashes: - return False - # Only consider it a duplicate if *all* results are final and known. - return all_final and len(response.results) > 0 - - def _record_final(self, response: rasr.StreamingRecognizeResponse) -> None: - """Store hashes of any new final transcripts.""" - for result in response.results: - if result.is_final and result.alternatives: - transcript = result.alternatives[0].transcript - h = hashlib.sha256(transcript.encode("utf-8")).hexdigest() - self._finalized_hashes.add(h) + def _append_audio(self, chunk: bytes) -> None: + self._audio_buffer.append(chunk) + self._buffered_bytes += len(chunk) + while self._audio_buffer and self._buffered_bytes > self._lookback_max_bytes: + self._buffered_bytes -= len(self._audio_buffer.popleft()) def stream( self, @@ -609,8 +590,9 @@ def stream( audio_source: An iterable of raw audio chunks. Yields: - :obj:`StreamingRecognizeResponse` objects. On a successful - reconnect, duplicate final transcripts are suppressed. + :obj:`StreamingRecognizeResponse` objects. A reconnect replays + the configured audio lookback, so callers should deduplicate + transcripts if their application requires exactly-once output. Raises: :obj:`grpc.RpcError`: If a non-retryable error occurs or the @@ -618,18 +600,12 @@ def stream( """ audio_iterator = iter(audio_source) attempt = 0 - last_exception: Optional[grpc.RpcError] = None - while True: try: generator = self._buffered_request_generator(audio_iterator) for response in self.asr_service.stub.StreamingRecognize( generator, metadata=self.asr_service.auth.get_auth_metadata() ): - if self._is_duplicate(response): - LOGGER.debug("Suppressing duplicate final transcript after reconnect.") - continue - self._record_final(response) yield response # Stream completed normally. if self._retry_count > 0: @@ -637,7 +613,6 @@ def stream( return except grpc.RpcError as exc: - last_exception = exc if not is_retryable_grpc_error(exc): LOGGER.warning( "Non-retryable gRPC error in streaming ASR: %s – %s", diff --git a/riva/client/auth.py b/riva/client/auth.py index a20177bd..942ed17f 100644 --- a/riva/client/auth.py +++ b/riva/client/auth.py @@ -14,22 +14,9 @@ def create_channel( use_ssl: bool = False, uri: str = "localhost:50051", metadata: Optional[List[Tuple[str, str]]] = None, - options: Optional[List[Tuple[str, str]]] = [], + options: Optional[List[Tuple[str, Union[str, int]]]] = None, use_aio: Optional[bool] = False, ) -> grpc.Channel: - """Create a gRPC channel with sensible defaults for resilient streaming. - - Default keepalive settings are injected so that dead connections are - detected quickly, enabling faster recovery on transient failures. - """ - default_options = [ - ("grpc.keepalive_time_ms", "10000"), - ("grpc.keepalive_timeout_ms", "5000"), - ("grpc.keepalive_permit_without_calls", "1"), - ("grpc.http2.max_pings_without_data", "0"), - ] - options = default_options + (options or []) - def metadata_callback(context, callback): callback(metadata, None) @@ -74,7 +61,7 @@ def __init__( metadata_args: List[List[str]] = None, ssl_client_cert: Optional[Union[str, os.PathLike]] = None, ssl_client_key: Optional[Union[str, os.PathLike]] = None, - options: Optional[List[Tuple[str, str]]] = [], + options: Optional[List[Tuple[str, Union[str, int]]]] = None, use_aio: bool = False, ) -> None: """ @@ -95,8 +82,8 @@ def __init__( Used for mutual TLS authentication. Defaults to None. ssl_client_key (Optional[Union[str, os.PathLike]], optional): Path to the SSL client private key file. Used for mutual TLS authentication. Defaults to None. - options (Optional[List[Tuple[str, str]]], optional): Additional gRPC channel options. - Each tuple should contain (option_name, option_value). Defaults to []. + options (Optional[List[Tuple[str, Union[str, int]]]], optional): Additional gRPC channel options. + Each tuple should contain an option name and value. use_aio (bool, optional): Whether to use asyncio for the channel. Defaults to False. Raises: diff --git a/riva/client/retry.py b/riva/client/retry.py index 495bcf7b..b4c9227e 100644 --- a/riva/client/retry.py +++ b/riva/client/retry.py @@ -3,17 +3,13 @@ """Shared retry utilities for resilient streaming ASR and TTS clients. -This module provides constants, helpers, and decorators for handling transient -gRPC failures with exponential backoff. It is designed to be used by both +This module provides constants and helpers for handling transient gRPC +failures with exponential backoff. It is designed to be used by both :mod:`riva.client.asr` and :mod:`riva.client.tts`. """ -import functools import logging import random -import time -from typing import Any, Callable, Optional, Tuple, TypeVar - import grpc LOGGER = logging.getLogger(__name__) @@ -39,9 +35,6 @@ grpc.StatusCode.UNIMPLEMENTED, }) -F = TypeVar("F", bound=Callable[..., Any]) - - def is_retryable_grpc_error(exc: grpc.RpcError) -> bool: """Return ``True`` if *exc* is a transient gRPC error that is safe to retry. @@ -81,70 +74,3 @@ def exponential_backoff( if jitter: delay = delay * random.random() return delay - - -def retry_streaming( - max_retries: int = 3, - base_delay: float = 1.0, - max_delay: float = 60.0, - on_retry: Optional[Callable[[grpc.RpcError, int, float], None]] = None, -) -> Callable[[F], F]: - """Decorator that retries a streaming gRPC call on transient failures. - - The decorated function must be a generator (i.e. use ``yield``). When a - :class:`grpc.RpcError` with a retryable code is raised, the generator is - closed and the function is re-invoked up to *max_retries* times. - - Args: - max_retries: Maximum number of retry attempts after the initial failure. - base_delay: Initial backoff delay in seconds. - max_delay: Maximum backoff delay in seconds. - on_retry: Optional callback ``fn(exc, attempt, delay)`` invoked before - each retry sleep. - - Returns: - A decorator that wraps generator functions with retry logic. - """ - def decorator(func: F) -> F: - @functools.wraps(func) - def wrapper(*args: Any, **kwargs: Any) -> Any: - last_exception: Optional[grpc.RpcError] = None - for attempt in range(max_retries + 1): - try: - yield from func(*args, **kwargs) - return - except grpc.RpcError as exc: - last_exception = exc - if not is_retryable_grpc_error(exc): - LOGGER.warning( - "Non-retryable gRPC error %s: %s", - exc.code() if hasattr(exc, "code") else "UNKNOWN", - exc.details() if hasattr(exc, "details") else str(exc), - ) - raise - if attempt >= max_retries: - LOGGER.error( - "Max retries (%d) exceeded for %s. Last error: %s", - max_retries, - func.__name__, - exc.details() if hasattr(exc, "details") else str(exc), - ) - raise - delay = exponential_backoff(attempt, base_delay, max_delay) - LOGGER.info( - "Retryable gRPC error %s on attempt %d/%d for %s. " - "Sleeping %.2f s before retry.", - exc.code(), - attempt + 1, - max_retries + 1, - func.__name__, - delay, - ) - if on_retry is not None: - on_retry(exc, attempt + 1, delay) - time.sleep(delay) - # Should never reach here, but satisfy type checker. - if last_exception is not None: - raise last_exception - return wrapper # type: ignore[return-value] - return decorator diff --git a/riva/client/tts.py b/riva/client/tts.py index d6239de6..2a1d79bf 100644 --- a/riva/client/tts.py +++ b/riva/client/tts.py @@ -1,8 +1,9 @@ # SPDX-FileCopyrightText: Copyright (c) 2022 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: MIT +import logging import time -from typing import Dict, Generator, Optional, Union, Iterable +from typing import Dict, Generator, Iterable, Iterator, List, Optional, Union import grpc from grpc._channel import _MultiThreadedRendezvous @@ -11,8 +12,12 @@ import riva.client.proto.riva_tts_pb2_grpc as rtts_srv from riva.client import Auth from riva.client.proto.riva_audio_pb2 import AudioEncoding +from riva.client.retry import exponential_backoff, is_retryable_grpc_error import wave + +LOGGER = logging.getLogger(__name__) + def parse_custom_configuration(custom_configuration: str) -> Dict[str, str]: """Parse a comma-separated ``key:value`` string into a dictionary. @@ -222,22 +227,13 @@ def request_generator(text): return self.stub.SynthesizeOnline(request_generator(text), metadata=self.auth.get_auth_metadata()) -import logging -from typing import Iterator - -import grpc - -from riva.client.retry import is_retryable_grpc_error, exponential_backoff - -LOGGER = logging.getLogger(__name__) - - class ResilientStreamingTTS: """A resilient wrapper around :class:`SpeechSynthesisService` for streaming TTS. - This class retries individual text segments on transient gRPC failures, - yielding audio chunks as they arrive. It is designed for long-running - streaming synthesis where network blips should not terminate the session. + This class retries individual text segments on transient gRPC failures. + Responses for a segment are buffered until that segment completes, so a + retry cannot duplicate audio that was already delivered to the caller. + Segment size is therefore the latency/recovery trade-off. Example: >>> auth = Auth(uri="localhost:50051") @@ -335,8 +331,8 @@ def synthesize_stream( custom_configuration=custom_configuration, enable_word_time_offsets=enable_word_time_offsets, ) - for resp in responses: - yield resp + completed_segment: List[rtts.SynthesizeSpeechResponse] = list(responses) + yield from completed_segment break # Segment completed successfully. except grpc.RpcError as exc: diff --git a/scripts/asr/transcribe_file.py b/scripts/asr/transcribe_file.py index 160f9a5f..ad4be508 100644 --- a/scripts/asr/transcribe_file.py +++ b/scripts/asr/transcribe_file.py @@ -60,6 +60,16 @@ def parse_args() -> argparse.Namespace: help="Option to simulate realtime transcription. Audio fragments are sent to a server at a pace that mimics " "normal speech.", ) + parser.add_argument( + "--auto-recover", + action="store_true", + help="Retry retryable streaming gRPC failures using a bounded audio lookback.", + ) + parser.add_argument("--max-retries", type=int, default=3, help="Maximum reconnect attempts.") + parser.add_argument( + "--lookback-seconds", type=float, default=2.0, + help="PCM audio duration to replay after a reconnect.", + ) parser.add_argument( "--print-confidence", action="store_true", @@ -158,11 +168,23 @@ def main() -> int: with riva.client.AudioChunkFileIterator( args.input_file, args.file_streaming_chunk, delay_callback, ) as audio_chunk_iterator: + if args.auto_recover: + wav_parameters = riva.client.get_wav_file_parameters(args.input_file) or {} + responses = riva.client.ResilientStreamingASR( + asr_service, + config, + max_retries=args.max_retries, + lookback_seconds=args.lookback_seconds, + sample_rate_hz=wav_parameters.get("framerate", 16000), + audio_channel_count=wav_parameters.get("nchannels", 1), + sample_width_bytes=wav_parameters.get("sampwidth", 2), + ).stream(audio_chunk_iterator) + else: + responses = asr_service.streaming_response_generator( + audio_chunks=audio_chunk_iterator, streaming_config=config + ) riva.client.print_streaming( - responses=asr_service.streaming_response_generator( - audio_chunks=audio_chunk_iterator, - streaming_config=config, - ), + responses=responses, show_intermediate=args.show_intermediate, additional_info="time" if (args.word_time_offsets or args.speaker_diarization) else ("confidence" if args.print_confidence else "no"), word_time_offsets=args.word_time_offsets or args.speaker_diarization, diff --git a/scripts/tts/talk.py b/scripts/tts/talk.py index 31b51705..66a73a48 100644 --- a/scripts/tts/talk.py +++ b/scripts/tts/talk.py @@ -85,6 +85,12 @@ def parse_args() -> argparse.Namespace: "as it gets ready. If `--stream` is not set, then a synthesized audio is returned in 1 response only when " "all text is processed.", ) + parser.add_argument( + "--auto-recover", + action="store_true", + help="Retry a failed streaming synthesis segment before writing its audio.", + ) + parser.add_argument("--max-retries", type=int, default=3, help="Maximum retry attempts per text segment.") parser.add_argument( "--zero_shot_transcript", type=str, @@ -206,8 +212,8 @@ def main() -> int: print("Generating audio for request...") start = time.time() if args.stream: - responses = service.synthesize_online( - text_list, args.voice, args.language_code, sample_rate_hz=args.sample_rate_hz, + synthesize_kwargs = dict( + voice_name=args.voice, language_code=args.language_code, sample_rate_hz=args.sample_rate_hz, encoding=(AudioEncoding.OGGOPUS if args.encoding == "OGGOPUS" else AudioEncoding.LINEAR_PCM), zero_shot_audio_prompt_file=args.zero_shot_audio_prompt_file, zero_shot_quality=(20 if args.zero_shot_quality is None else args.zero_shot_quality), @@ -215,6 +221,12 @@ def main() -> int: enable_word_time_offsets=args.word_time_offsets, **custom_configuration_kwargs, ) + if args.auto_recover: + responses = riva.client.ResilientStreamingTTS( + service, max_retries=args.max_retries + ).synthesize_stream(text_list, **synthesize_kwargs) + else: + responses = service.synthesize_online(text_list, **synthesize_kwargs) first = True for resp in responses: stop = time.time() diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index bcf192de..953442e4 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -1,9 +1,6 @@ # SPDX-FileCopyrightText: Copyright (c) 2022 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: MIT -import time -from unittest.mock import Mock - import grpc import pytest @@ -11,8 +8,8 @@ RETRYABLE_GRPC_CODES, exponential_backoff, is_retryable_grpc_error, - retry_streaming, ) +from riva.client.tts import ResilientStreamingTTS class FakeRpcError(grpc.RpcError): @@ -56,61 +53,25 @@ def test_jitter_reduces_delay(self): assert 0.0 <= d < 4.0 -class TestRetryStreamingDecorator: - def test_success_no_retry(self): - @retry_streaming(max_retries=2) - def gen(): - yield 1 - yield 2 - - assert list(gen()) == [1, 2] - - def test_retries_then_succeeds(self): - call_count = 0 - - @retry_streaming(max_retries=2, base_delay=0.01) - def gen(): - nonlocal call_count - call_count += 1 - if call_count < 2: - raise FakeRpcError(grpc.StatusCode.UNAVAILABLE) - yield "ok" - - assert list(gen()) == ["ok"] - assert call_count == 2 - - def test_exhausts_retries(self): - @retry_streaming(max_retries=1, base_delay=0.01) - def gen(): - raise FakeRpcError(grpc.StatusCode.UNAVAILABLE) - yield # make it a generator - - with pytest.raises(grpc.RpcError): - list(gen()) - - def test_non_retryable_raises_immediately(self): - @retry_streaming(max_retries=3) - def gen(): - raise FakeRpcError(grpc.StatusCode.INVALID_ARGUMENT) - yield - - with pytest.raises(grpc.RpcError): - list(gen()) - - def test_on_retry_callback(self): - callback_log = [] +class TestResilientStreamingTTS: + def test_does_not_yield_partial_audio_from_a_failed_segment(self, monkeypatch): + class Service: + def __init__(self): + self.calls = 0 - def on_retry(exc, attempt, delay): - callback_log.append((exc.code(), attempt, delay)) + def synthesize_online(self, **_kwargs): + self.calls += 1 + if self.calls == 1: + def failed_stream(): + yield "partial-audio" + raise FakeRpcError(grpc.StatusCode.UNAVAILABLE) + return failed_stream() - @retry_streaming(max_retries=1, base_delay=0.01, on_retry=on_retry) - def gen(): - raise FakeRpcError(grpc.StatusCode.UNAVAILABLE) - yield + return iter(["complete-audio"]) - with pytest.raises(grpc.RpcError): - list(gen()) + monkeypatch.setattr("riva.client.tts.time.sleep", lambda _delay: None) + service = Service() + client = ResilientStreamingTTS(service, max_retries=1, base_delay=0) - assert len(callback_log) == 1 - assert callback_log[0][0] == grpc.StatusCode.UNAVAILABLE - assert callback_log[0][1] == 1 + assert list(client.synthesize_stream("hello")) == ["complete-audio"] + assert service.calls == 2 From c73599aa69435625240b51e538f9fdc1b5fec017 Mon Sep 17 00:00:00 2001 From: Prakruthi B Gowda Date: Mon, 7 Sep 2026 13:37:39 +0530 Subject: [PATCH 3/3] refactor: configure streaming recovery through custom configuration --- riva/client/__init__.py | 8 +- riva/client/asr.py | 210 +++++++++------------------------ riva/client/auth.py | 8 +- riva/client/retry.py | 34 +++++- riva/client/tts.py | 189 +++++++---------------------- scripts/asr/transcribe_file.py | 30 +---- scripts/tts/talk.py | 16 +-- tests/unit/test_retry.py | 38 +++--- 8 files changed, 155 insertions(+), 378 deletions(-) diff --git a/riva/client/__init__.py b/riva/client/__init__.py index 492a3af8..7656bd69 100644 --- a/riva/client/__init__.py +++ b/riva/client/__init__.py @@ -4,7 +4,6 @@ from riva.client.asr import ( AudioChunkFileIterator, ASRService, - ResilientStreamingASR, add_audio_file_specs_to_config, add_word_boosting_to_config, add_speaker_diarization_to_config, @@ -40,10 +39,5 @@ from riva.client.proto.riva_audio_pb2 import AudioEncoding from riva.client.proto.riva_nlp_pb2 import AnalyzeIntentOptions from riva.client.proto.riva_nmt_pb2 import StreamingTranslateSpeechToSpeechConfig, TranslationConfig, SynthesizeSpeechConfig, StreamingTranslateSpeechToTextConfig -from riva.client.retry import ( - RETRYABLE_GRPC_CODES, - exponential_backoff, - is_retryable_grpc_error, -) -from riva.client.tts import SpeechSynthesisService, ResilientStreamingTTS +from riva.client.tts import SpeechSynthesisService from riva.client.nmt import NeuralMachineTranslationClient diff --git a/riva/client/asr.py b/riva/client/asr.py index 29d7cd8f..82eb43c6 100644 --- a/riva/client/asr.py +++ b/riva/client/asr.py @@ -23,7 +23,7 @@ import riva.client.proto.riva_asr_pb2 as rasr import riva.client.proto.riva_asr_pb2_grpc as rasr_srv from riva.client.auth import Auth -from riva.client.retry import exponential_backoff, is_retryable_grpc_error +from riva.client.retry import exponential_backoff, is_retryable_grpc_error, split_recovery_configuration LOGGER = logging.getLogger(__name__) @@ -439,6 +439,11 @@ def streaming_response_generator( config = RecognitionConfig(enable_automatic_punctuation=True) streaming_config = StreamingRecognitionConfig(config, interim_results=True) + Set ``client_auto_recover:true`` in ``config.custom_configuration`` to enable + retry. ``client_max_retries`` and ``client_lookback_seconds`` control the + retry count and bounded PCM lookback buffer. These client-only keys are not + sent to Riva. + Yields: :obj:`riva.client.proto.riva_asr_pb2.StreamingRecognizeResponse`: responses for audio chunks in :param:`audio_chunks`. You may find description of response fields in declaration of @@ -446,9 +451,55 @@ def streaming_response_generator( message `here `_. """ - generator = streaming_request_generator(audio_chunks, streaming_config) - for response in self.stub.StreamingRecognize(generator, metadata=self.auth.get_auth_metadata()): - yield response + server_configuration, auto_recover, max_retries, lookback_seconds = split_recovery_configuration( + streaming_config.config.custom_configuration + ) + server_streaming_config = rasr.StreamingRecognitionConfig() + server_streaming_config.CopyFrom(streaming_config) + server_streaming_config.config.custom_configuration.clear() + server_streaming_config.config.custom_configuration.update(server_configuration) + + if not auto_recover: + generator = streaming_request_generator(audio_chunks, server_streaming_config) + yield from self.stub.StreamingRecognize(generator, metadata=self.auth.get_auth_metadata()) + return + + sample_rate_hz = server_streaming_config.config.sample_rate_hertz or 16000 + channel_count = server_streaming_config.config.audio_channel_count or 1 + max_buffered_bytes = int(lookback_seconds * sample_rate_hz * channel_count * 2) + buffered_audio: Deque[bytes] = deque() + buffered_bytes = 0 + audio_iterator = iter(audio_chunks) + attempt = 0 + + while True: + def request_generator() -> Generator[rasr.StreamingRecognizeRequest, None, None]: + nonlocal buffered_bytes + yield rasr.StreamingRecognizeRequest(streaming_config=server_streaming_config) + for chunk in buffered_audio: + yield rasr.StreamingRecognizeRequest(audio_content=chunk) + for chunk in audio_iterator: + buffered_audio.append(chunk) + buffered_bytes += len(chunk) + while buffered_audio and buffered_bytes > max_buffered_bytes: + buffered_bytes -= len(buffered_audio.popleft()) + yield rasr.StreamingRecognizeRequest(audio_content=chunk) + + try: + yield from self.stub.StreamingRecognize( + request_generator(), metadata=self.auth.get_auth_metadata() + ) + return + except grpc.RpcError as exc: + if not is_retryable_grpc_error(exc) or attempt >= max_retries: + raise + delay = exponential_backoff(attempt) + LOGGER.info( + "Streaming ASR connection lost (%s); retrying in %.2f seconds (attempt %d/%d).", + exc.code(), delay, attempt + 1, max_retries, + ) + attempt += 1 + time.sleep(delay) def offline_recognize( self, audio_bytes: bytes, config: rasr.RecognitionConfig, future: bool = False @@ -490,154 +541,3 @@ def offline_recognize( request = rasr.RecognizeRequest(config=config, audio=audio_bytes) func = self.stub.Recognize.future if future else self.stub.Recognize return func(request, metadata=self.auth.get_auth_metadata()) - - - -class ResilientStreamingASR: - """A resilient wrapper around :class:`ASRService` for streaming recognition. - - This class buffers recent audio and automatically reconnects on transient - gRPC failures, replaying buffered audio so that recognition can continue - with a bounded audio lookback. Recovery is best effort: audio older than - the configured lookback window may not be replayed, and applications must - tolerate repeated transcripts after a reconnect. - - Example: - >>> auth = Auth(uri="localhost:50051") - >>> asr = ASRService(auth) - >>> config = StreamingRecognitionConfig( - ... config=RecognitionConfig(enable_automatic_punctuation=True), - ... interim_results=True, - ... ) - >>> resilient_asr = ResilientStreamingASR(asr, config) - >>> for response in resilient_asr.stream(audio_chunks): - ... print(response) - - Args: - asr_service: The underlying :class:`ASRService` instance. - streaming_config: Configuration for streaming recognition. - max_retries: Maximum number of reconnection attempts per failure. - lookback_seconds: Duration of audio to replay after reconnecting. - A larger value improves recovery at the cost of higher latency. - sample_rate_hz: Sample rate of linear PCM audio. - audio_channel_count: Number of PCM audio channels. - sample_width_bytes: Bytes per PCM sample. - base_delay: Initial backoff delay in seconds. - max_delay: Maximum backoff delay in seconds. - """ - - def __init__( - self, - asr_service: ASRService, - streaming_config: rasr.StreamingRecognitionConfig, - max_retries: int = 3, - lookback_seconds: float = 2.0, - sample_rate_hz: int = 16000, - audio_channel_count: int = 1, - sample_width_bytes: int = 2, - base_delay: float = 1.0, - max_delay: float = 60.0, - ) -> None: - self.asr_service = asr_service - self.streaming_config = streaming_config - self.max_retries = max_retries - self.base_delay = base_delay - self.max_delay = max_delay - - if lookback_seconds <= 0: - raise ValueError("lookback_seconds must be greater than zero") - if min(sample_rate_hz, audio_channel_count, sample_width_bytes) <= 0: - raise ValueError("PCM format values must be greater than zero") - self._lookback_max_bytes = int( - lookback_seconds * sample_rate_hz * audio_channel_count * sample_width_bytes - ) - self._audio_buffer: Deque[bytes] = deque() - self._buffered_bytes = 0 - self._retry_count = 0 - - def _buffered_request_generator( - self, - audio_source: Iterable[bytes], - ) -> Generator[rasr.StreamingRecognizeRequest, None, None]: - """Yield the config message, buffered audio, then new audio. - - Each chunk from *audio_source* is appended to the lookback buffer - before being yielded so that it is available for the next reconnect. - """ - yield rasr.StreamingRecognizeRequest(streaming_config=self.streaming_config) - - # Replay buffered audio from previous (partial) stream - for chunk in self._audio_buffer: - yield rasr.StreamingRecognizeRequest(audio_content=chunk) - - for chunk in audio_source: - self._append_audio(chunk) - yield rasr.StreamingRecognizeRequest(audio_content=chunk) - - def _append_audio(self, chunk: bytes) -> None: - self._audio_buffer.append(chunk) - self._buffered_bytes += len(chunk) - while self._audio_buffer and self._buffered_bytes > self._lookback_max_bytes: - self._buffered_bytes -= len(self._audio_buffer.popleft()) - - def stream( - self, - audio_source: Iterable[bytes], - ) -> Generator[rasr.StreamingRecognizeResponse, None, None]: - """Stream audio for recognition with automatic recovery. - - Args: - audio_source: An iterable of raw audio chunks. - - Yields: - :obj:`StreamingRecognizeResponse` objects. A reconnect replays - the configured audio lookback, so callers should deduplicate - transcripts if their application requires exactly-once output. - - Raises: - :obj:`grpc.RpcError`: If a non-retryable error occurs or the - maximum number of retries is exceeded. - """ - audio_iterator = iter(audio_source) - attempt = 0 - while True: - try: - generator = self._buffered_request_generator(audio_iterator) - for response in self.asr_service.stub.StreamingRecognize( - generator, metadata=self.asr_service.auth.get_auth_metadata() - ): - yield response - # Stream completed normally. - if self._retry_count > 0: - LOGGER.info("Streaming ASR recovered after %d retry(s).", self._retry_count) - return - - except grpc.RpcError as exc: - if not is_retryable_grpc_error(exc): - LOGGER.warning( - "Non-retryable gRPC error in streaming ASR: %s – %s", - exc.code() if hasattr(exc, "code") else "UNKNOWN", - exc.details() if hasattr(exc, "details") else str(exc), - ) - raise - if attempt >= self.max_retries: - LOGGER.error( - "Streaming ASR failed permanently after %d retries. Last error: %s", - self.max_retries, - exc.details() if hasattr(exc, "details") else str(exc), - ) - raise - delay = exponential_backoff(attempt, self.base_delay, self.max_delay) - LOGGER.info( - "Streaming ASR connection lost (%s). Reconnecting in %.2f s " - "(attempt %d/%d).", - exc.code(), - delay, - attempt + 1, - self.max_retries, - ) - self._retry_count += 1 - attempt += 1 - time.sleep(delay) - # Loop continues: _buffered_request_generator will replay - # self._audio_buffer and then consume audio_iterator. diff --git a/riva/client/auth.py b/riva/client/auth.py index 942ed17f..8a4688d4 100644 --- a/riva/client/auth.py +++ b/riva/client/auth.py @@ -14,7 +14,7 @@ def create_channel( use_ssl: bool = False, uri: str = "localhost:50051", metadata: Optional[List[Tuple[str, str]]] = None, - options: Optional[List[Tuple[str, Union[str, int]]]] = None, + options: Optional[List[Tuple[str, str]]] = [], use_aio: Optional[bool] = False, ) -> grpc.Channel: def metadata_callback(context, callback): @@ -61,7 +61,7 @@ def __init__( metadata_args: List[List[str]] = None, ssl_client_cert: Optional[Union[str, os.PathLike]] = None, ssl_client_key: Optional[Union[str, os.PathLike]] = None, - options: Optional[List[Tuple[str, Union[str, int]]]] = None, + options: Optional[List[Tuple[str, str]]] = [], use_aio: bool = False, ) -> None: """ @@ -82,8 +82,8 @@ def __init__( Used for mutual TLS authentication. Defaults to None. ssl_client_key (Optional[Union[str, os.PathLike]], optional): Path to the SSL client private key file. Used for mutual TLS authentication. Defaults to None. - options (Optional[List[Tuple[str, Union[str, int]]]], optional): Additional gRPC channel options. - Each tuple should contain an option name and value. + options (Optional[List[Tuple[str, str]]], optional): Additional gRPC channel options. + Each tuple should contain (option_name, option_value). Defaults to []. use_aio (bool, optional): Whether to use asyncio for the channel. Defaults to False. Raises: diff --git a/riva/client/retry.py b/riva/client/retry.py index b4c9227e..15f1f4a1 100644 --- a/riva/client/retry.py +++ b/riva/client/retry.py @@ -8,11 +8,19 @@ :mod:`riva.client.asr` and :mod:`riva.client.tts`. """ -import logging import random +from typing import Dict, Mapping, Tuple + import grpc -LOGGER = logging.getLogger(__name__) +CLIENT_AUTO_RECOVER = "client_auto_recover" +CLIENT_MAX_RETRIES = "client_max_retries" +CLIENT_LOOKBACK_SECONDS = "client_lookback_seconds" +CLIENT_RECOVERY_KEYS = frozenset({ + CLIENT_AUTO_RECOVER, + CLIENT_MAX_RETRIES, + CLIENT_LOOKBACK_SECONDS, +}) # gRPC status codes that are generally considered transient and safe to retry. RETRYABLE_GRPC_CODES = frozenset({ @@ -74,3 +82,25 @@ def exponential_backoff( if jitter: delay = delay * random.random() return delay + + +def split_recovery_configuration( + custom_configuration: Mapping[str, str], +) -> Tuple[Dict[str, str], bool, int, float]: + """Separate client-only streaming recovery options from server options. + + The reserved ``client_*`` keys control retry behaviour in the Python + client and are intentionally not forwarded to Riva. All other key/value + pairs are returned unchanged for the server. + """ + server_configuration = { + key: value for key, value in custom_configuration.items() if key not in CLIENT_RECOVERY_KEYS + } + enabled = str(custom_configuration.get(CLIENT_AUTO_RECOVER, "false")).lower() == "true" + max_retries = int(custom_configuration.get(CLIENT_MAX_RETRIES, "3")) + lookback_seconds = float(custom_configuration.get(CLIENT_LOOKBACK_SECONDS, "2.0")) + if max_retries < 0: + raise ValueError(f"{CLIENT_MAX_RETRIES} must be non-negative") + if lookback_seconds <= 0: + raise ValueError(f"{CLIENT_LOOKBACK_SECONDS} must be greater than zero") + return server_configuration, enabled, max_retries, lookback_seconds diff --git a/riva/client/tts.py b/riva/client/tts.py index 2a1d79bf..f43c85f9 100644 --- a/riva/client/tts.py +++ b/riva/client/tts.py @@ -3,7 +3,7 @@ import logging import time -from typing import Dict, Generator, Iterable, Iterator, List, Optional, Union +from typing import Dict, Generator, Iterable, List, Optional, Union import grpc from grpc._channel import _MultiThreadedRendezvous @@ -12,7 +12,7 @@ import riva.client.proto.riva_tts_pb2_grpc as rtts_srv from riva.client import Auth from riva.client.proto.riva_audio_pb2 import AudioEncoding -from riva.client.retry import exponential_backoff, is_retryable_grpc_error +from riva.client.retry import exponential_backoff, is_retryable_grpc_error, split_recovery_configuration import wave @@ -131,8 +131,11 @@ def synthesize( if zero_shot_transcript is not None: req.zero_shot_data.transcript = zero_shot_transcript - if custom_configuration: - for key, value in custom_configuration.items(): + server_configuration, _, _, _ = split_recovery_configuration( + custom_configuration or {} + ) + if server_configuration: + for key, value in server_configuration.items(): req.custom_configuration[key] = str(value) add_custom_dictionary_to_config(req, custom_dictionary) @@ -174,7 +177,10 @@ def synthesize_online( zero_shot_quality: (:obj:`int`): Required quality of output audio, ranges between 1-40. custom_dictionary (:obj:`dict`, `optional`): Dictionary with key-value pair containing grapheme and corresponding phoneme custom_configuration (:obj:`Dict[str, str]`, `optional`): Free-form key/value parameters forwarded - to the synthesizer (e.g. ``{"exaggeration_factor": "1.5"}``). Model-specific. + to the synthesizer (e.g. ``{"exaggeration_factor": "1.5"}``). Set + ``client_auto_recover`` to ``"true"`` to enable client-side retry; use + ``client_max_retries`` to set the retry count. These reserved client keys are + not forwarded to the synthesizer. enable_word_time_offsets (:obj:`bool`, `optional`): If :obj:`True`, request per-word start/end timestamps, returned in ``response.meta.words`` (supported by models that produce word alignment, e.g. Magpie TTS). @@ -203,8 +209,11 @@ def synthesize_online( req.zero_shot_data.encoding = audio_prompt_encoding req.zero_shot_data.quality = zero_shot_quality - if custom_configuration: - for key, value in custom_configuration.items(): + server_configuration, auto_recover, max_retries, _ = split_recovery_configuration( + custom_configuration or {} + ) + if server_configuration: + for key, value in server_configuration.items(): req.custom_configuration[key] = str(value) add_custom_dictionary_to_config(req, custom_dictionary) @@ -224,146 +233,30 @@ def request_generator(text): else: raise ValueError(f"Invalid text type: {type(text)}") - return self.stub.SynthesizeOnline(request_generator(text), metadata=self.auth.get_auth_metadata()) - - -class ResilientStreamingTTS: - """A resilient wrapper around :class:`SpeechSynthesisService` for streaming TTS. - - This class retries individual text segments on transient gRPC failures. - Responses for a segment are buffered until that segment completes, so a - retry cannot duplicate audio that was already delivered to the caller. - Segment size is therefore the latency/recovery trade-off. - - Example: - >>> auth = Auth(uri="localhost:50051") - >>> tts = SpeechSynthesisService(auth) - >>> resilient_tts = ResilientStreamingTTS(tts) - >>> for audio_chunk in resilient_tts.synthesize_stream( - ... ["Hello world", "This is a test."], - ... voice_name="English-US-Female-1", - ... ): - ... play_audio(audio_chunk) - - Args: - tts_service: The underlying :class:`SpeechSynthesisService` instance. - max_retries: Maximum number of retry attempts per text segment. - base_delay: Initial backoff delay in seconds. - max_delay: Maximum backoff delay in seconds. - """ - - def __init__( - self, - tts_service: SpeechSynthesisService, - max_retries: int = 3, - base_delay: float = 1.0, - max_delay: float = 60.0, - ) -> None: - self.tts_service = tts_service - self.max_retries = max_retries - self.base_delay = base_delay - self.max_delay = max_delay - self._retry_count = 0 - - def synthesize_stream( - self, - text_segments: Union[str, list[str], Iterable[str]], - voice_name: Optional[str] = None, - language_code: str = 'en-US', - encoding: AudioEncoding = AudioEncoding.LINEAR_PCM, - sample_rate_hz: int = 22050, - zero_shot_audio_prompt_file: Optional[str] = None, - audio_prompt_encoding: AudioEncoding = AudioEncoding.ENCODING_UNSPECIFIED, - zero_shot_quality: int = 20, - custom_dictionary: Optional[dict] = None, - custom_configuration: Optional[Dict[str, str]] = None, - enable_word_time_offsets: Optional[bool] = None, - ) -> Generator[rtts.SynthesizeSpeechResponse, None, None]: - """Synthesize speech from text segments with automatic recovery. - - Each text segment is sent independently. If the gRPC stream fails - while synthesizing a segment, that segment is retried up to - *max_retries* times before the error is propagated. - - Args: - text_segments: Input text. A single string, a list, or any iterable - of strings. Each element is treated as one retryable unit. - voice_name: See :meth:`SpeechSynthesisService.synthesize_online`. - language_code: See :meth:`SpeechSynthesisService.synthesize_online`. - encoding: See :meth:`SpeechSynthesisService.synthesize_online`. - sample_rate_hz: See :meth:`SpeechSynthesisService.synthesize_online`. - zero_shot_audio_prompt_file: See :meth:`SpeechSynthesisService.synthesize_online`. - audio_prompt_encoding: See :meth:`SpeechSynthesisService.synthesize_online`. - zero_shot_quality: See :meth:`SpeechSynthesisService.synthesize_online`. - custom_dictionary: See :meth:`SpeechSynthesisService.synthesize_online`. - custom_configuration: See :meth:`SpeechSynthesisService.synthesize_online`. - enable_word_time_offsets: See :meth:`SpeechSynthesisService.synthesize_online`. - - Yields: - :obj:`SynthesizeSpeechResponse` objects containing audio chunks. - - Raises: - :obj:`grpc.RpcError`: If a non-retryable error occurs or the - maximum number of retries is exceeded for a segment. - """ - # Normalise input to an iterator of strings. - if isinstance(text_segments, str): - segment_iter: Iterator[str] = iter([text_segments]) - else: - segment_iter = iter(text_segments) - - for segment in segment_iter: - attempt = 0 - last_exception: Optional[grpc.RpcError] = None - - while True: - try: - responses = self.tts_service.synthesize_online( - text=segment, - voice_name=voice_name, - language_code=language_code, - encoding=encoding, - sample_rate_hz=sample_rate_hz, - zero_shot_audio_prompt_file=zero_shot_audio_prompt_file, - audio_prompt_encoding=audio_prompt_encoding, - zero_shot_quality=zero_shot_quality, - custom_dictionary=custom_dictionary, - custom_configuration=custom_configuration, - enable_word_time_offsets=enable_word_time_offsets, - ) - completed_segment: List[rtts.SynthesizeSpeechResponse] = list(responses) - yield from completed_segment - break # Segment completed successfully. - - except grpc.RpcError as exc: - last_exception = exc - if not is_retryable_grpc_error(exc): - LOGGER.warning( - "Non-retryable gRPC error in streaming TTS: %s – %s", - exc.code() if hasattr(exc, "code") else "UNKNOWN", - exc.details() if hasattr(exc, "details") else str(exc), + if not auto_recover: + return self.stub.SynthesizeOnline(request_generator(text), metadata=self.auth.get_auth_metadata()) + + def recovery_generator() -> Generator[rtts.SynthesizeSpeechResponse, None, None]: + segments = [text] if isinstance(text, str) else text + for segment in segments: + attempt = 0 + while True: + try: + responses = self.stub.SynthesizeOnline( + request_generator(segment), metadata=self.auth.get_auth_metadata() ) - raise - if attempt >= self.max_retries: - LOGGER.error( - "Streaming TTS failed permanently after %d retries for segment %r. " - "Last error: %s", - self.max_retries, - segment, - exc.details() if hasattr(exc, "details") else str(exc), + # Do not expose a partial segment: retrying it must not duplicate audio. + yield from list(responses) + break + except grpc.RpcError as exc: + if not is_retryable_grpc_error(exc) or attempt >= max_retries: + raise + delay = exponential_backoff(attempt) + LOGGER.info( + "Streaming TTS connection lost (%s); retrying in %.2f seconds (attempt %d/%d).", + exc.code(), delay, attempt + 1, max_retries, ) - raise - delay = exponential_backoff(attempt, self.base_delay, self.max_delay) - LOGGER.info( - "Streaming TTS connection lost (%s) on segment %r. " - "Retrying in %.2f s (attempt %d/%d).", - exc.code(), - segment, - delay, - attempt + 1, - self.max_retries, - ) - self._retry_count += 1 - attempt += 1 - time.sleep(delay) - # Loop continues: retry the same segment. + attempt += 1 + time.sleep(delay) + + return recovery_generator() diff --git a/scripts/asr/transcribe_file.py b/scripts/asr/transcribe_file.py index ad4be508..160f9a5f 100644 --- a/scripts/asr/transcribe_file.py +++ b/scripts/asr/transcribe_file.py @@ -60,16 +60,6 @@ def parse_args() -> argparse.Namespace: help="Option to simulate realtime transcription. Audio fragments are sent to a server at a pace that mimics " "normal speech.", ) - parser.add_argument( - "--auto-recover", - action="store_true", - help="Retry retryable streaming gRPC failures using a bounded audio lookback.", - ) - parser.add_argument("--max-retries", type=int, default=3, help="Maximum reconnect attempts.") - parser.add_argument( - "--lookback-seconds", type=float, default=2.0, - help="PCM audio duration to replay after a reconnect.", - ) parser.add_argument( "--print-confidence", action="store_true", @@ -168,23 +158,11 @@ def main() -> int: with riva.client.AudioChunkFileIterator( args.input_file, args.file_streaming_chunk, delay_callback, ) as audio_chunk_iterator: - if args.auto_recover: - wav_parameters = riva.client.get_wav_file_parameters(args.input_file) or {} - responses = riva.client.ResilientStreamingASR( - asr_service, - config, - max_retries=args.max_retries, - lookback_seconds=args.lookback_seconds, - sample_rate_hz=wav_parameters.get("framerate", 16000), - audio_channel_count=wav_parameters.get("nchannels", 1), - sample_width_bytes=wav_parameters.get("sampwidth", 2), - ).stream(audio_chunk_iterator) - else: - responses = asr_service.streaming_response_generator( - audio_chunks=audio_chunk_iterator, streaming_config=config - ) riva.client.print_streaming( - responses=responses, + responses=asr_service.streaming_response_generator( + audio_chunks=audio_chunk_iterator, + streaming_config=config, + ), show_intermediate=args.show_intermediate, additional_info="time" if (args.word_time_offsets or args.speaker_diarization) else ("confidence" if args.print_confidence else "no"), word_time_offsets=args.word_time_offsets or args.speaker_diarization, diff --git a/scripts/tts/talk.py b/scripts/tts/talk.py index 66a73a48..31b51705 100644 --- a/scripts/tts/talk.py +++ b/scripts/tts/talk.py @@ -85,12 +85,6 @@ def parse_args() -> argparse.Namespace: "as it gets ready. If `--stream` is not set, then a synthesized audio is returned in 1 response only when " "all text is processed.", ) - parser.add_argument( - "--auto-recover", - action="store_true", - help="Retry a failed streaming synthesis segment before writing its audio.", - ) - parser.add_argument("--max-retries", type=int, default=3, help="Maximum retry attempts per text segment.") parser.add_argument( "--zero_shot_transcript", type=str, @@ -212,8 +206,8 @@ def main() -> int: print("Generating audio for request...") start = time.time() if args.stream: - synthesize_kwargs = dict( - voice_name=args.voice, language_code=args.language_code, sample_rate_hz=args.sample_rate_hz, + responses = service.synthesize_online( + text_list, args.voice, args.language_code, sample_rate_hz=args.sample_rate_hz, encoding=(AudioEncoding.OGGOPUS if args.encoding == "OGGOPUS" else AudioEncoding.LINEAR_PCM), zero_shot_audio_prompt_file=args.zero_shot_audio_prompt_file, zero_shot_quality=(20 if args.zero_shot_quality is None else args.zero_shot_quality), @@ -221,12 +215,6 @@ def main() -> int: enable_word_time_offsets=args.word_time_offsets, **custom_configuration_kwargs, ) - if args.auto_recover: - responses = riva.client.ResilientStreamingTTS( - service, max_retries=args.max_retries - ).synthesize_stream(text_list, **synthesize_kwargs) - else: - responses = service.synthesize_online(text_list, **synthesize_kwargs) first = True for resp in responses: stop = time.time() diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 953442e4..9a1a4c07 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -5,11 +5,14 @@ import pytest from riva.client.retry import ( + CLIENT_AUTO_RECOVER, + CLIENT_LOOKBACK_SECONDS, + CLIENT_MAX_RETRIES, RETRYABLE_GRPC_CODES, exponential_backoff, is_retryable_grpc_error, + split_recovery_configuration, ) -from riva.client.tts import ResilientStreamingTTS class FakeRpcError(grpc.RpcError): @@ -53,25 +56,16 @@ def test_jitter_reduces_delay(self): assert 0.0 <= d < 4.0 -class TestResilientStreamingTTS: - def test_does_not_yield_partial_audio_from_a_failed_segment(self, monkeypatch): - class Service: - def __init__(self): - self.calls = 0 +class TestRecoveryConfiguration: + def test_client_options_are_not_forwarded_to_riva(self): + server_configuration, enabled, max_retries, lookback_seconds = split_recovery_configuration({ + "exaggeration_factor": "1.5", + CLIENT_AUTO_RECOVER: "true", + CLIENT_MAX_RETRIES: "4", + CLIENT_LOOKBACK_SECONDS: "3.5", + }) - def synthesize_online(self, **_kwargs): - self.calls += 1 - if self.calls == 1: - def failed_stream(): - yield "partial-audio" - raise FakeRpcError(grpc.StatusCode.UNAVAILABLE) - return failed_stream() - - return iter(["complete-audio"]) - - monkeypatch.setattr("riva.client.tts.time.sleep", lambda _delay: None) - service = Service() - client = ResilientStreamingTTS(service, max_retries=1, base_delay=0) - - assert list(client.synthesize_stream("hello")) == ["complete-audio"] - assert service.calls == 2 + assert server_configuration == {"exaggeration_factor": "1.5"} + assert enabled is True + assert max_retries == 4 + assert lookback_seconds == 3.5