Track failed playback commands with bounded retry backoff

jukebox/domain/entities/__init__.py

@@ -3,13 +3,14 @@
 from .disc import Disc, DiscMetadata, DiscOption
 from .library import Library
 from .playback_action import PlaybackAction
-from .playback_session import PlaybackSession
+from .playback_session import PlaybackCommandRetry, PlaybackSession
 from .tag_event import TagEvent
 
 __all__ = [
     "CurrentTagAction",
     "CurrentTagStatus",
     "PlaybackAction",
+    "PlaybackCommandRetry",
     "PlaybackSession",
     "TagEvent",
     "Library",

jukebox/domain/entities/playback_session.py

@@ -1,4 +1,32 @@
-from pydantic import BaseModel
+from typing import Self
+
+from pydantic import BaseModel, model_validator
+
+from .playback_action import PlaybackAction
+
+
+class PlaybackCommandRetry(BaseModel):
+    """Tracks retry timing for a failed playback command."""
+
+    action: PlaybackAction
+    tag_id: str | None = None
+    first_failed_at: float
+    last_failed_at: float
+    attempt_count: int
+    next_retry_at: float | None
+    exhausted: bool = False
+
+    @model_validator(mode="after")
+    def validate_tag_id_matches_action(self) -> Self:
+        if self.action == PlaybackAction.PLAY:
+            if self.tag_id is None:
+                raise ValueError("tag_id is required for PLAY retry")
+        elif self.tag_id is not None:
+            raise ValueError("tag_id is only valid for PLAY retry")
+        return self
+
+    def matches(self, *, action: PlaybackAction, tag_id: str | None = None) -> bool:
+        return self.action == action and self.tag_id == tag_id
 
 
 class PlaybackSession(BaseModel):
@@ -15,3 +43,6 @@ class PlaybackSession(BaseModel):
 
     # Timestamp
     last_event_timestamp: float | None = None
+
+    # Playback command retry state
+    playback_command_retry: PlaybackCommandRetry | None = None

jukebox/domain/use_cases/handle_tag_event.py

@@ -1,14 +1,15 @@
 import logging
-from contextlib import contextmanager
+from collections.abc import Callable
 
-from jukebox.domain.entities import CurrentTagAction, PlaybackAction, PlaybackSession, TagEvent
+from jukebox.domain.entities import CurrentTagAction, PlaybackAction, PlaybackCommandRetry, PlaybackSession, TagEvent
 from jukebox.domain.errors import PlaybackError
 from jukebox.domain.ports import PlayerPort
 from jukebox.domain.repositories import CurrentTagRepository, LibraryRepository
 from jukebox.domain.use_cases.determine_action import DetermineAction
 from jukebox.domain.use_cases.determine_current_tag_action import DetermineCurrentTagAction
 
 LOGGER = logging.getLogger("jukebox")
+PLAYBACK_RETRY_DELAYS_SECONDS = (0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0)
 
 
 class HandleTagEvent:
@@ -21,12 +22,16 @@ def __init__(
         current_tag_repository: CurrentTagRepository,
         determine_action: DetermineAction,
         determine_current_tag_action: DetermineCurrentTagAction,
+        retry_delays_seconds: tuple[float, ...] = PLAYBACK_RETRY_DELAYS_SECONDS,
     ):
         self.player = player
         self.library = library
         self.current_tag_repository = current_tag_repository
         self.determine_action = determine_action
         self.determine_current_tag_action = determine_current_tag_action
+        if not retry_delays_seconds:
+            raise ValueError("retry_delays_seconds must not be empty")
+        self.retry_delays_seconds = retry_delays_seconds
 
     def execute(self, tag_event: TagEvent, session: PlaybackSession) -> PlaybackSession:
         self._apply_current_tag_action_best_effort(tag_event, session)
@@ -45,10 +50,20 @@ def execute(self, tag_event: TagEvent, session: PlaybackSession) -> PlaybackSess
             case PlaybackAction.CONTINUE:
                 # Reset when tag is present
                 session.playing_tag_removed_at = None
+                if (
+                    session.playback_command_retry is not None
+                    and session.playback_command_retry.action == PlaybackAction.PAUSE
+                ):
+                    session.playback_command_retry = None
 
             case PlaybackAction.RESUME:
-                with suppress_playback_error("Playback operation `RESUME` failed; stopping session update"):
-                    self.player.resume()
+                if self._run_playback_command(
+                    action=PlaybackAction.RESUME,
+                    timestamp=tag_event.timestamp,
+                    session=session,
+                    error_message="Playback operation `RESUME` failed; stopping session update",
+                    command=self.player.resume,
+                ):
                     session.paused_at = None
                     session.playing_tag_removed_at = None
 
@@ -58,14 +73,21 @@ def execute(self, tag_event: TagEvent, session: PlaybackSession) -> PlaybackSess
                 disc = self.library.get_disc(tag_event.tag_id) if tag_event.tag_id is not None else None
                 if disc is not None:
                     LOGGER.info("Found corresponding disc: %s", disc)
-                    with suppress_playback_error(
-                        f"Playback operation `PLAY` failed for tag_id='{tag_event.tag_id}'; stopping session update"
+                    if self._run_playback_command(
+                        action=PlaybackAction.PLAY,
+                        tag_id=tag_event.tag_id,
+                        timestamp=tag_event.timestamp,
+                        session=session,
+                        error_message=(
+                            f"Playback operation `PLAY` failed for tag_id='{tag_event.tag_id}'; stopping session update"
+                        ),
+                        command=lambda: self.player.play(disc.uri, disc.option.shuffle),
                     ):
-                        self.player.play(disc.uri, disc.option.shuffle)
                         session.playing_tag = tag_event.tag_id
                         session.paused_at = None
                         session.playing_tag_removed_at = None
                 else:
+                    session.playback_command_retry = None
                     LOGGER.warning("No disc found for UID: %s", tag_event.tag_id)
 
             case PlaybackAction.WAITING:
@@ -76,16 +98,26 @@ def execute(self, tag_event: TagEvent, session: PlaybackSession) -> PlaybackSess
                 LOGGER.debug("Grace period: %.3fs / %gs", grace_period_elapsed, self.determine_action.pause_delay)
 
             case PlaybackAction.PAUSE:
-                with suppress_playback_error("Playback operation `PAUSE` failed; continuing session update"):
-                    self.player.pause()
-                session.paused_at = tag_event.timestamp
+                if self._run_playback_command(
+                    action=PlaybackAction.PAUSE,
+                    timestamp=tag_event.timestamp,
+                    session=session,
+                    error_message="Playback operation `PAUSE` failed; stopping session update",
+                    command=self.player.pause,
+                ):
+                    session.paused_at = tag_event.timestamp
 
             case PlaybackAction.STOP:
-                with suppress_playback_error("Playback operation `STOP` failed; continuing session update"):
-                    self.player.stop()
-                session.playing_tag = None
-                session.paused_at = None
-                session.playing_tag_removed_at = None
+                if self._run_playback_command(
+                    action=PlaybackAction.STOP,
+                    timestamp=tag_event.timestamp,
+                    session=session,
+                    error_message="Playback operation `STOP` failed; stopping session update",
+                    command=self.player.stop,
+                ):
+                    session.playing_tag = None
+                    session.paused_at = None
+                    session.playing_tag_removed_at = None
 
             case PlaybackAction.IDLE:
                 pass
@@ -136,10 +168,86 @@ def _apply_current_tag_action(
             case CurrentTagAction.KEEP:
                 pass  # No state changed
 
+    def _run_playback_command(
+        self,
+        *,
+        action: PlaybackAction,
+        timestamp: float,
+        session: PlaybackSession,
+        error_message: str,
+        command: Callable[[], None],
+        tag_id: str | None = None,
+    ) -> bool:
+        retry = session.playback_command_retry
+        retry_matches = retry is not None and retry.matches(action=action, tag_id=tag_id)
+        if retry_matches and retry.exhausted:
+            LOGGER.debug(
+                "Skipping playback operation `%s`; retry exhausted after %d attempts",
+                action.value.upper(),
+                retry.attempt_count,
+            )
+            return False
 
-@contextmanager
-def suppress_playback_error(msg: str):
-    try:
-        yield
-    except PlaybackError:
-        LOGGER.warning(msg)
+        if retry_matches and retry.next_retry_at is not None and timestamp < retry.next_retry_at:
+            LOGGER.debug(
+                "Skipping playback operation `%s` until retry time %.3f",
+                action.value.upper(),
+                retry.next_retry_at,
+            )
+            return False
+
+        try:
+            command()
+        except PlaybackError:
+            retry = self._record_playback_command_failure(
+                action=action,
+                timestamp=timestamp,
+                session=session,
+                tag_id=tag_id,
+            )
+            if retry.exhausted:
+                LOGGER.warning("%s; retry exhausted after %d attempts", error_message, retry.attempt_count)
+            else:
+                next_retry_at = retry.next_retry_at
+                if next_retry_at is None:
+                    raise RuntimeError("retry is not exhausted but next_retry_at is missing")
+                LOGGER.warning("%s; retrying in %.3fs", error_message, next_retry_at - timestamp)
+            return False
+
+        session.playback_command_retry = None
+        return True
+
+    def _record_playback_command_failure(
+        self,
+        *,
+        action: PlaybackAction,
+        timestamp: float,
+        session: PlaybackSession,
+        tag_id: str | None = None,
+    ) -> PlaybackCommandRetry:
+        existing_retry = session.playback_command_retry
+        if existing_retry is None or not existing_retry.matches(action=action, tag_id=tag_id):
+            attempt_count = 1
+            first_failed_at = timestamp
+        else:
+            attempt_count = existing_retry.attempt_count + 1
+            first_failed_at = existing_retry.first_failed_at
+
+        retry_delay = self._retry_delay_for_attempt(attempt_count)
+        retry = PlaybackCommandRetry(
+            action=action,
+            tag_id=tag_id,
+            first_failed_at=first_failed_at,
+            last_failed_at=timestamp,
+            attempt_count=attempt_count,
+            next_retry_at=None if retry_delay is None else timestamp + retry_delay,
+            exhausted=retry_delay is None,
+        )
+        session.playback_command_retry = retry
+        return retry
+
+    def _retry_delay_for_attempt(self, attempt_count: int) -> float | None:
+        delay_index = attempt_count - 1
+        if delay_index >= len(self.retry_delays_seconds):
+            return None
+        return self.retry_delays_seconds[delay_index]

tests/jukebox/domain/entities/test_playback_session.py

@@ -0,0 +1,33 @@
+import pytest
+from pydantic import ValidationError
+
+from jukebox.domain.entities import PlaybackAction, PlaybackCommandRetry
+
+
+def make_retry(action: PlaybackAction, tag_id: str | None = None) -> PlaybackCommandRetry:
+    return PlaybackCommandRetry(
+        action=action,
+        tag_id=tag_id,
+        first_failed_at=100.0,
+        last_failed_at=100.0,
+        attempt_count=1,
+        next_retry_at=100.1,
+    )
+
+
+def test_playback_command_retry_requires_tag_id_for_play():
+    with pytest.raises(ValidationError, match="tag_id is required for PLAY retry"):
+        make_retry(PlaybackAction.PLAY)
+
+
+def test_playback_command_retry_rejects_tag_id_for_non_play():
+    with pytest.raises(ValidationError, match="tag_id is only valid for PLAY retry"):
+        make_retry(PlaybackAction.PAUSE, tag_id="test-tag")
+
+
+def test_playback_command_retry_matches_action_and_tag_id():
+    retry = make_retry(PlaybackAction.PLAY, tag_id="test-tag")
+
+    assert retry.matches(action=PlaybackAction.PLAY, tag_id="test-tag") is True
+    assert retry.matches(action=PlaybackAction.PLAY, tag_id="other-tag") is False
+    assert retry.matches(action=PlaybackAction.PAUSE, tag_id="test-tag") is False