Fix sync group member playing out of sync after concurrent group changes (#4189)

Author: marcelveldt

music_assistant/providers/sync_group/constants.py

@@ -15,6 +15,12 @@
 # from the user without disrupting the live sync session.
 IDLE_GRACE_SECONDS: Final[float] = 10.0
 
+# Maximum seconds to wait for the sync leader to confirm it has actually started
+# playing after a (re)form. The group's playback lock is held for the duration so
+# a concurrent (un)group command can't race a start that is still in flight at the
+# device — which would otherwise strand a player playing outside the group.
+PLAYBACK_START_TIMEOUT: Final[float] = 5.0
+
 CONF_ENTRY_SGP_NOTE = ConfigEntry(
     key="sgp_note",
     type=ConfigEntryType.ALERT,

music_assistant/providers/sync_group/player.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Any, cast
 
 from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
@@ -25,10 +26,13 @@
     CONF_ENTRY_SGP_NOTE,
     EXTRA_FEATURES_FROM_MEMBERS,
     IDLE_GRACE_SECONDS,
+    PLAYBACK_START_TIMEOUT,
     PROVIDERS_WITH_DYNAMIC_LEADER_SWITCH,
 )
 
 if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
     from music_assistant_models.player import PlayerSource
 
     from .provider import SyncGroupProvider
@@ -395,9 +399,13 @@ async def play(self) -> None:
         # formed (e.g. after _dissolve_and_reform left us powered with no leader).
         # _form_syncgroup is idempotent so calling it here is cheap when already formed.
         await self._form_syncgroup()
-        await self.mass.players.cmd_resume(
-            self.player_id, self._attr_active_source, self._attr_current_media
-        )
+        # Hold the group's playback lock until the leader actually reports playing
+        # so a concurrent (un)group command can't race the in-flight start — which
+        # would otherwise leave a player streaming outside the group.
+        async with self._await_leader_playback():
+            await self.mass.players.cmd_resume(
+                self.player_id, self._attr_active_source, self._attr_current_media
+            )
 
     async def poll(self) -> None:
         """Poll player for state updates."""
@@ -415,7 +423,10 @@ async def play_media(self, media: PlayerMedia) -> None:
         if sync_leader := self.sync_leader:
             # Use internal handler to target the sync leader directly,
             # bypassing group/sync redirect that would loop back to this player.
-            await self.mass.players._handle_play_media(sync_leader.player_id, media)
+            # Hold the group's playback lock until the leader confirms playback
+            # (see play()) so a concurrent (un)group command can't race the start.
+            async with self._await_leader_playback():
+                await self.mass.players._handle_play_media(sync_leader.player_id, media)
         else:
             raise RuntimeError("An empty group cannot play media, consider adding members first")
 
@@ -644,6 +655,28 @@ async def _form_syncgroup(self) -> None:
                     self.sync_leader, player_ids_to_add=members_to_sync
                 )
 
+    @asynccontextmanager
+    async def _await_leader_playback(self) -> AsyncIterator[None]:
+        """
+        Wait for the sync leader to confirm playback for the command run in the body.
+
+        Wrap the play/resume call that targets the leader in this context manager.
+        The group's playback lock (held by the caller) then stays acquired until the
+        leader actually reports playing, so a concurrent (un)group command cannot
+        race a start that has not yet taken effect at the device. A no-op when there
+        is no leader to wait on.
+        """
+        if (leader := self.sync_leader) is None:
+            yield
+            return
+        async with self.mass.players.wait_for_player_update(
+            leader.player_id,
+            attribute_name="playback_state",
+            attribute_value=PlaybackState.PLAYING,
+            timeout=PLAYBACK_START_TIMEOUT,
+        ):
+            yield
+
     async def _dissolve_syncgroup(self) -> None:
         """Dissolve the current syncgroup by ungrouping all members."""
         # a dissolve is happening now — any pending grace timer is no longer needed

tests/providers/test_sync_group.py

@@ -1166,3 +1166,101 @@ def _get_raw(_player_id: str, key: str, default: object = None) -> object:
         mass.config.get_raw_player_config_value = MagicMock(side_effect=_get_raw)
         sgp = _make_sync_group(mass)
         assert PlayerFeature.POWER in sgp.supported_features
+
+
+def _recording_wait(order: list[str]) -> MagicMock:
+    """
+    Build a wait_for_player_update replacement that records call/enter/exit order.
+
+    The returned mock records ``wait_for:<player_id>:<value>`` when invoked,
+    ``subscribe`` on context enter and ``await`` on context exit, so a test can
+    assert that the playback start was wrapped (subscribe before the command is
+    issued) rather than awaited after the fact.
+    """
+
+    class _RecordingWait:
+        async def __aenter__(self) -> None:
+            order.append("subscribe")
+
+        async def __aexit__(self, *_exc: object) -> bool:
+            order.append("await")
+            return False
+
+    def _wait(player_id: str, **kwargs: Any) -> _RecordingWait:
+        order.append(f"wait_for:{player_id}:{kwargs.get('attribute_value')}")
+        return _RecordingWait()
+
+    return MagicMock(side_effect=_wait)
+
+
+class TestLeaderPlaybackAwaited:
+    """A (re)form must not return until the leader confirms it has started playing."""
+
+    @pytest.mark.asyncio
+    async def test_play_waits_for_leader_before_returning(self) -> None:
+        """play() wraps the resume in a wait so the group lock isn't released early."""
+        mass = _make_mock_mass()
+        sgp = _make_sync_group(mass)
+        leader = _make_mock_player("leader", playback_state=PlaybackState.IDLE)
+        mass.players.get_player = _player_lookup({"leader": leader})
+        sgp.sync_leader = leader
+        sgp._attr_group_members = ["leader"]
+
+        order: list[str] = []
+        mass.players.wait_for_player_update = _recording_wait(order)
+        mass.players.cmd_resume = AsyncMock(side_effect=lambda *_a, **_k: order.append("resume"))
+
+        with patch.object(sgp, "_form_syncgroup", new=AsyncMock()):
+            await sgp.play()
+
+        # subscribe happens before the resume command, and the wait completes
+        # after — i.e. the start is wrapped, not fire-and-forget.
+        assert order == [
+            f"wait_for:leader:{PlaybackState.PLAYING}",
+            "subscribe",
+            "resume",
+            "await",
+        ]
+
+    @pytest.mark.asyncio
+    async def test_play_media_waits_for_leader_before_returning(self) -> None:
+        """play_media() wraps the leader start so a concurrent (un)group can't race it."""
+        mass = _make_mock_mass()
+        sgp = _make_sync_group(mass)
+        leader = _make_mock_player("leader", playback_state=PlaybackState.IDLE)
+        mass.players.get_player = _player_lookup({"leader": leader})
+        sgp.sync_leader = leader
+        sgp._attr_group_members = ["leader"]
+
+        order: list[str] = []
+        mass.players.wait_for_player_update = _recording_wait(order)
+        mass.players._handle_play_media = AsyncMock(
+            side_effect=lambda *_a, **_k: order.append("play_media")
+        )
+
+        media = MagicMock()
+        media.source_id = "syncgroup_test"
+        with patch.object(sgp, "_form_syncgroup", new=AsyncMock()):
+            await sgp.play_media(media)
+
+        assert order == [
+            f"wait_for:leader:{PlaybackState.PLAYING}",
+            "subscribe",
+            "play_media",
+            "await",
+        ]
+
+    @pytest.mark.asyncio
+    async def test_no_wait_when_group_has_no_leader(self) -> None:
+        """With no leader to wait on, play() resumes without arming a playback wait."""
+        mass = _make_mock_mass()
+        sgp = _make_sync_group(mass)
+        sgp.sync_leader = None
+
+        mass.players.cmd_resume = AsyncMock()
+
+        with patch.object(sgp, "_form_syncgroup", new=AsyncMock()):
+            await sgp.play()
+
+        mass.players.wait_for_player_update.assert_not_called()
+        mass.players.cmd_resume.assert_awaited_once()