diff --git a/riva/client/__init__.py b/riva/client/__init__.py index 7656bd6..492a3af 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,10 @@ 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, +) +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 be9a362..29d7cd8 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]]: @@ -483,3 +490,154 @@ 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 8a4688d..942ed17 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, str]]] = [], + options: Optional[List[Tuple[str, Union[str, int]]]] = None, 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, str]]] = [], + options: Optional[List[Tuple[str, Union[str, int]]]] = None, 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, 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 new file mode 100644 index 0000000..b4c9227 --- /dev/null +++ b/riva/client/retry.py @@ -0,0 +1,76 @@ +# 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 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 logging +import random +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, +}) + +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 diff --git a/riva/client/tts.py b/riva/client/tts.py index c057956..2a1d79b 100644 --- a/riva/client/tts.py +++ b/riva/client/tts.py @@ -1,16 +1,23 @@ # SPDX-FileCopyrightText: Copyright (c) 2022 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: MIT -from typing import Dict, Generator, Optional, Union, Iterable +import logging +import time +from typing import Dict, Generator, Iterable, Iterator, List, Optional, Union +import grpc from grpc._channel import _MultiThreadedRendezvous import riva.client.proto.riva_tts_pb2 as rtts 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. @@ -217,4 +224,146 @@ 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()) + + +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), + ) + 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/scripts/asr/transcribe_file.py b/scripts/asr/transcribe_file.py index 160f9a5..ad4be50 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 31b5170..66a73a4 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 new file mode 100644 index 0000000..953442e --- /dev/null +++ b/tests/unit/test_retry.py @@ -0,0 +1,77 @@ +# SPDX-FileCopyrightText: Copyright (c) 2022 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: MIT + +import grpc +import pytest + +from riva.client.retry import ( + RETRYABLE_GRPC_CODES, + exponential_backoff, + is_retryable_grpc_error, +) +from riva.client.tts import ResilientStreamingTTS + + +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 TestResilientStreamingTTS: + def test_does_not_yield_partial_audio_from_a_failed_segment(self, monkeypatch): + class Service: + def __init__(self): + self.calls = 0 + + 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