Fix: Prevent Timeout on long running streams

Only use TimedAsyncGenerator in one location of the chain and with a high timeout, just to detect it, not to recover.

diff --git a/music_assistant/helpers/audio.py b/music_assistant/helpers/audio.py
index 5ce3b6b4f012a012f49dfef07339b8b7687490f6..ef3774b08e738875daf413e1f6d325e782053b3c 100644 (file)
--- a/music_assistant/helpers/audio.py
+++ b/music_assistant/helpers/audio.py
@@ -47,7 +47,7 @@ from .dsp import filter_to_ffmpeg_params
 from .ffmpeg import FFMpeg, get_ffmpeg_stream
 from .playlists import IsHLSPlaylist, PlaylistItem, fetch_playlist, parse_m3u
 from .process import AsyncProcess, communicate
-from .util import TimedAsyncGenerator, create_tempfile, detect_charset
+from .util import create_tempfile, detect_charset
 
 if TYPE_CHECKING:
     from music_assistant_models.config_entries import CoreConfig, PlayerConfig
@@ -384,9 +384,7 @@ async def get_media_stream(
             pcm_format.content_type.value,
             ffmpeg_proc.proc.pid,
         )
-        async for chunk in TimedAsyncGenerator(
-            ffmpeg_proc.iter_chunked(pcm_format.pcm_sample_size), timeout=30
-        ):
+        async for chunk in ffmpeg_proc.iter_chunked(pcm_format.pcm_sample_size):
             # for non-tracks we just yield all chunks directly
             if streamdetails.media_type != MediaType.TRACK:
                 yield chunk
@@ -467,7 +465,7 @@ async def get_media_stream(
             log_tail = ""
         logger.debug(
             "stream %s (with code %s) for %s - seconds streamed: %s %s",
-            "finished" if finished else "aborted",
+            "cancelled" if cancelled else "finished" if finished else "aborted",
             ffmpeg_proc.returncode,
             streamdetails.uri,
             seconds_streamed,

diff --git a/music_assistant/helpers/ffmpeg.py b/music_assistant/helpers/ffmpeg.py
index d187271efdea620bba9843f9f1104ec274f2a958..b3913008756dd17170a79260fd5d07020cfa3b9d 100644 (file)
--- a/music_assistant/helpers/ffmpeg.py
+++ b/music_assistant/helpers/ffmpeg.py
@@ -134,7 +134,12 @@ class FFMpeg(AsyncProcess):
         try:
             start = time.time()
             self.logger.log(VERBOSE_LOG_LEVEL, "Start reading audio data from source...")
-            async for chunk in TimedAsyncGenerator(self.audio_input, timeout=30):
+            # use TimedAsyncGenerator to catch we're stuck waiting on data forever
+            # don't set this timeout too low because in some cases it can indeed take a while
+            # for data to arrive (e.g. when there is X amount of seconds in the buffer)
+            # so this timeout is just to catch if the source is stuck and rpeort it and not
+            # to recover from it.
+            async for chunk in TimedAsyncGenerator(self.audio_input, timeout=300):
                 await self.write(chunk)
             self.logger.log(
                 VERBOSE_LOG_LEVEL, "Audio data source exhausted in %.2fs", time.time() - start

diff --git a/music_assistant/providers/spotify/__init__.py b/music_assistant/providers/spotify/__init__.py
index a59f85eded21b0b9a19c0ede70f05f4dfe8bea08..fe4ef2c1f9d1d907550e70a1fd2ecbaf6f0581e2 100644 (file)
--- a/music_assistant/providers/spotify/__init__.py
+++ b/music_assistant/providers/spotify/__init__.py
@@ -48,7 +48,7 @@ from music_assistant.helpers.auth import AuthenticationHelper
 from music_assistant.helpers.json import json_loads
 from music_assistant.helpers.process import AsyncProcess, check_output
 from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
-from music_assistant.helpers.util import TimedAsyncGenerator, lock, parse_title_and_version
+from music_assistant.helpers.util import lock, parse_title_and_version
 from music_assistant.models.music_provider import MusicProvider
 
 from .helpers import get_librespot_binary
@@ -608,7 +608,7 @@ class SpotifyProvider(MusicProvider):
             if stderr:
                 log_reader = asyncio.create_task(_read_stderr())
 
-            async for chunk in TimedAsyncGenerator(librespot_proc.iter_any(chunk_size), 20):
+            async for chunk in librespot_proc.iter_any(chunk_size):
                 yield chunk
                 bytes_received += len(chunk)
             if stderr: