music-assistant-server

1"""Unified AirPlay/RAOP stream session logic for AirPlay devices."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from collections.abc import AsyncGenerator, Coroutine
8from contextlib import suppress
9from typing import TYPE_CHECKING, Any
10
11from music_assistant_models.enums import ContentType, PlaybackState
12from music_assistant_models.errors import MusicAssistantError, PlayerCommandFailed
13
14from music_assistant.constants import CONF_SYNC_ADJUST
15from music_assistant.controllers.streams.audio_processing import get_media_session_id
16from music_assistant.helpers.ffmpeg import FFMpeg
17
18from .constants import (
19    AIRPLAY_CLOCK_READY_LEAD_MS,
20    AIRPLAY_CLOCK_READY_TIMEOUT_MS,
21    AIRPLAY_COLD_GROUP_START_LEAD_MS,
22    AIRPLAY_GROUP_START_LEAD_MS,
23    AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS,
24    AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS,
25    AIRPLAY_LATE_JOIN_RING_MAX_BYTES,
26    AIRPLAY_LATE_JOIN_RING_MIN_SECONDS,
27    AIRPLAY_SPLICE_LEAD_MARGIN_MS,
28    AIRPLAY_START_LEAD_MS,
29    ClockReadiness,
30    StreamingProtocol,
31)
32from .helpers import get_final_output_format
33from .stream import AirPlayStream
34
35if TYPE_CHECKING:
36    from music_assistant_models.media_items import AudioFormat
37
38    from music_assistant.models.player import PlayerMedia
39
40    from .player import AirPlayPlayer
41    from .provider import AirPlayProvider
42
43# What each readiness outcome means for the join anchor: only a projection moves
44# it, every other outcome leaves the join floor carrying it alone, so the note
45# says which of the two happened and why. STALLED has no note because it never
46# reaches an anchor - a stalled joiner is refused the join before that.
47_CLOCK_READINESS_NOTES: dict[ClockReadiness, str] = {
48    ClockReadiness.PROJECTED: "usable in {out:.2f}s; anchoring just past that",
49    ClockReadiness.NOT_APPLICABLE: "runs on NTP timing, so there is none to wait for; "
50    "anchoring on the join floor",
51    ClockReadiness.UNREPORTED: "was not reported within {timeout:.1f}s (a slow device, or a "
52    "receiver that never answered); anchoring on the join floor",
53}
54
55
56class AirPlayStreamSession:
57    """Stream session (RAOP or AirPlay2) to one or more players."""
58
59    def __init__(
60        self,
61        airplay_provider: AirPlayProvider,
62        sync_clients: list[AirPlayPlayer],
63        pcm_format: AudioFormat,
64        media: PlayerMedia,
65    ) -> None:
66        """
67        Initialize AirPlayStreamSession.
68
69        :param airplay_provider: The AirPlay provider instance.
70        :param sync_clients: List of AirPlay players to stream to.
71        :param pcm_format: PCM format of the input stream.
72        :param media: Queue media that owns the stream session.
73        """
74        assert sync_clients
75        self.prov = airplay_provider
76        self.mass = airplay_provider.mass
77        self.pcm_format = pcm_format
78        self.media = media
79        self.sync_clients = sync_clients
80        self._audio_source_task: asyncio.Task[None] | None = None
81        self._player_ffmpeg: dict[str, FFMpeg] = {}
82        self._lock = asyncio.Lock()
83        self._ptp_lock = asyncio.Lock()
84        self.start_unix_ms: int = 0
85        self.start_time: float = 0.0
86        self.seconds_streamed: float = 0
87        # Timing source for the whole session, decided once in start() and applied
88        # identically to every native AirPlay 2 member (and any late joiner) so a
89        # sync group can never mix shared-PTP and NTP members.
90        self.use_shared_ptp: bool = False
91        self._shared_ptp_resolved = False
92        self._ptp_degraded_warning_logged = False
93        # Raw PCM ring buffer for late joiners. When a late joiner arrives we
94        # send this buffer to prime its pipeline so it starts playing quickly
95        # instead of waiting for the full pipeline to fill from scratch.
96        # It must cover the write-head lead - how far the feed runs ahead of the
97        # audible position - because that is exactly the span a joiner's anchor
98        # maps into. The lead is not a constant of the protocol: it is the sum
99        # of every buffer downstream of this counter (see
100        # AIRPLAY_LATE_JOIN_RING_MIN_SECONDS), so it is measured per session and
101        # the ring is grown to match.
102        self._pcm_buffer = bytearray()
103        # Largest write-head lead this session has shown, and the ring size
104        # derived from it.
105        self._peak_lead_seconds: float = 0.0
106        # Raw byte sizes of the PCM actually on the wire. At 24-bit the binary
107        # is fed s32le carriers (bit_depth stays 24 for the ALAC encode), so
108        # sizes MUST come from the content type: bit_depth-derived sizes are
109        # 6 bytes/frame while the wire frames are 8 — slicing or timing the
110        # feed on those boundaries cuts mid-sample (loud noise on a late
111        # joiner's prime) and inflates the position clock by 4/3.
112        bytes_per_sample = {
113            ContentType.PCM_S16LE: 2,
114            ContentType.PCM_S24LE: 3,
115            ContentType.PCM_S32LE: 4,
116            ContentType.PCM_F32LE: 4,
117        }.get(pcm_format.content_type, pcm_format.bit_depth // 8)
118        self._pcm_frame_size = bytes_per_sample * pcm_format.channels
119        self._pcm_byte_rate = self._pcm_frame_size * pcm_format.sample_rate
120        self._pcm_buffer_max = self._ring_bytes_for(0.0)
121        # Bytes still to skip off the head of the live feed for a late joiner
122        # whose anchor lands ahead of the current write head, keyed by player id.
123        self._client_skip_bytes: dict[str, int] = {}
124        # Cumulative bytes fed to the members (absolute stream coordinate of
125        # the write head). Source chunking makes no frame-alignment promise,
126        # so any slice or skip of the live feed must be aligned against THIS
127        # counter — aligning lengths in isolation can still land mid-sample,
128        # which a joiner renders as pure static.
129        self._pcm_total_fed: int = 0
130
131    @property
132    def effective_start_time(self) -> float:
133        """
134        Return the session anchor adjusted for the reference member's clock shift.
135
136        ``start_time`` is the wall-clock anchor set at the last group (re)start.
137        A member's cliairplay can re-anchor its playout LATER after recovering
138        from a PCM starvation, shifting the group's real timeline; that shift is
139        tracked per member. The first sync client is taken as the reference and
140        its accumulated shift is added, so a late joiner maps its first sample to
141        where the group actually plays. Divergent per-member shifts make any
142        single reference imperfect; the current first client is the best
143        available choice and the reference transfers to the new first client when
144        the old one is removed.
145        """
146        if not self.sync_clients:
147            return self.start_time
148        reference_stream = self.sync_clients[0].stream
149        if reference_stream is None:
150            return self.start_time
151        return self.start_time + reference_stream.cumulative_shift_seconds
152
153    async def start(self, audio_source: AsyncGenerator[bytes]) -> None:
154        """
155        Connect every member and anchor synchronized playback.
156
157        Spawns and connects each member's CLI, wires the per-seek ffmpeg into its
158        persistent stdin and starts feeding audio, then commands one shared
159        audible start instant. On any failure the whole session is stopped so the
160        caller can fall back to a cold restart.
161        """
162        ap2_members = sum(1 for p in self.sync_clients if p.protocol == StreamingProtocol.AIRPLAY2)
163        if ap2_members:
164            # Resolve the timing source before calculating the audible anchor so
165            # a bounded daemon-readiness wait cannot consume the setup lead.
166            self.use_shared_ptp = await self._resolve_shared_ptp(ap2_members)
167            self._shared_ptp_resolved = True
168        position_ms = int((self.media.elapsed_time or 0) * 1000)
169        try:
170            async with asyncio.TaskGroup() as task_group:
171                for player in self.sync_clients:
172                    task_group.create_task(
173                        self._member_start_step(
174                            player, "spawn its cli", self._start_client(player, self.use_shared_ptp)
175                        )
176                    )
177            await asyncio.gather(
178                *[
179                    self._member_start_step(
180                        player, "connect to its device", player.stream.wait_for_connection()
181                    )
182                    for player in self.sync_clients
183                    if player.stream
184                ]
185            )
186            # The binary buffers stdin into its ring from process start; feed
187            # audio first and wait for every member to confirm it flowing, then
188            # anchor with a short lead. Readiness is fully event-driven
189            # (connected + audio), so no guessed setup time is needed; the
190            # binary bursts the receiver pre-fill after START.
191            self._audio_source_task = asyncio.create_task(self._audio_streamer(audio_source))
192            await self._wait_members_audio_present()
193            # Members of a group have to agree on one instant, and a freshly
194            # connected receiver needs about a second before it even starts
195            # probing — too late for the binary to raise its own commit floor,
196            # and a first start is the one case it will not correct afterwards.
197            # So a group anchor waits for the receivers to say when they can
198            # play, rather than trusting the lead to have covered it.
199            ready_at_unix_ms = await self._wait_members_clock_ready()
200            await self._start_members(
201                position_ms, self._anchor_start_unix_ms(ready_at_unix_ms=ready_at_unix_ms)
202            )
203        except asyncio.CancelledError:
204            await self.stop()
205            raise
206        except Exception as err:
207            # playback failed to start, cleanup. This runs for every failure. A
208            # per-member one has already been named where it happened, so here
209            # the line says the whole session went down with it; for a failure
210            # no single member owns, it is the only line there is.
211            self.prov.logger.warning(
212                "AirPlay start failed for a session of %d member(s): %s",
213                len(self.sync_clients),
214                err,
215            )
216            await self.stop()
217            # A member can fail for a specific, user-actionable reason (a device
218            # that needs its password configured, for example). That error must
219            # reach the caller intact instead of being flattened into the generic
220            # message; the TaskGroup above nests it inside an exception group.
221            if (specific := _first_music_assistant_error(err)) is not None:
222                raise specific from err
223            raise PlayerCommandFailed("Playback failed to start") from err
224
225    def can_replace(self, sync_clients: list[AirPlayPlayer], pcm_format: AudioFormat) -> bool:
226        """
227        Return whether this live session can absorb a new play_media warm.
228
229        A warm replacement needs the same member set, the same session PCM
230        format and a connected stream on every member; anything else takes the
231        cold path.
232        """
233        if {p.player_id for p in sync_clients} != {p.player_id for p in self.sync_clients}:
234            return False
235        if (
236            pcm_format.sample_rate != self.pcm_format.sample_rate
237            or pcm_format.bit_depth != self.pcm_format.bit_depth
238        ):
239            return False
240        return all(
241            p.stream is not None and p.stream.running and p.stream.connected
242            for p in self.sync_clients
243        )
244
245    async def replace(self, audio_source: AsyncGenerator[bytes], media: PlayerMedia) -> bool:
246        """
247        Warm-replace the playing media with a new source (seek/next-track).
248
249        Stops feeding old audio and kills each member's ffmpeg (never the
250        persistent cli stdin), flushes every member's live stream in place, then
251        feeds a fresh ffmpeg into the same stdin and anchors all members at one
252        shared instant. A group flushes every member and awaits all acks before
253        the shared start. Returns False when anything fails so the caller can
254        fall back to the cold path.
255        """
256        position_ms = int((media.elapsed_time or 0) * 1000)
257        try:
258            # Stop feeding old audio and drop each member's buffered ffmpeg output
259            # before flushing: the binary drains stdin to EAGAIN on FLUSH, so no
260            # bytes may be written between the old ffmpeg dying and the flush ack.
261            # Killing ffmpeg never closes the cli stdin (MA holds the write end),
262            # so the binary keeps its stdin reader alive across the seek.
263            if self._audio_source_task and not self._audio_source_task.done():
264                self._audio_source_task.cancel()
265                with suppress(asyncio.CancelledError):
266                    await self._audio_source_task
267            for player in self.sync_clients:
268                if ffmpeg := self._player_ffmpeg.pop(player.player_id, None):
269                    await ffmpeg.kill()
270            flushed = await asyncio.gather(
271                *[self._flush_member(player) for player in self.sync_clients]
272            )
273            if not all(flushed):
274                raise PlayerCommandFailed("warm flush was not acknowledged")
275            for player in self.sync_clients:
276                await self._start_player_ffmpeg(player, media)
277            self.media = media
278            # The stream position counter and the late-join prime buffer both
279            # describe the OLD timeline; restart them before the new source pumps.
280            self.seconds_streamed = 0
281            self._pcm_total_fed = 0
282            self._pcm_buffer.clear()
283            # The shared START below re-establishes start_time, so the per-client
284            # late-join skip counters and every member's accumulated starvation
285            # shift describe a timeline that no longer exists: reset them.
286            self._client_skip_bytes.clear()
287            self._reset_member_shifts()
288            self._audio_source_task = asyncio.create_task(self._audio_streamer(audio_source))
289            # Anchor only after every member confirms the new audio flowing;
290            # the live connection and clock survive the flush, so a short
291            # re-anchor lead replaces the full setup lead.
292            await self._wait_members_audio_present()
293            await self._start_members(position_ms, self._anchor_start_unix_ms(warm=True))
294        except asyncio.CancelledError:
295            raise
296        except Exception as err:
297            self.prov.logger.warning(
298                "Warm replacement failed (%r); falling back to a cold restart", err
299            )
300            return False
301        return True
302
303    async def standby(self) -> bool:
304        """
305        Park the session: stall every member but keep the connections alive.
306
307        The next play_media (resume or seek) replaces the media warm over the
308        live connections — the same coordinated flush-refill as seek/next.
309        Returns False when any member lacks a running, connected stream or its
310        standby command cannot be delivered so the caller can fall back to a
311        full stop.
312        """
313        if not all(
314            p.stream is not None and p.stream.running and p.stream.connected
315            for p in self.sync_clients
316        ):
317            return False
318        if self._audio_source_task and not self._audio_source_task.done():
319            self._audio_source_task.cancel()
320            with suppress(asyncio.CancelledError):
321                await self._audio_source_task
322        for player in self.sync_clients:
323            if ffmpeg := self._player_ffmpeg.pop(player.player_id, None):
324                await ffmpeg.kill()
325            stream = player.stream
326            assert stream
327            try:
328                command_delivered = await stream.send_cli_command("ACTION=STANDBY")
329            except Exception as err:
330                self.prov.logger.warning(
331                    "Could not park AirPlay player %s: %s", player.player_id, err
332                )
333                return False
334            if not command_delivered:
335                self.prov.logger.warning(
336                    "Could not park AirPlay player %s: standby command was not delivered",
337                    player.player_id,
338                )
339                return False
340            player.set_state_from_stream(state=PlaybackState.PAUSED, stream=stream)
341        # a parked session has no live timeline; the resume re-anchors it
342        self.seconds_streamed = 0
343        self._pcm_total_fed = 0
344        self._pcm_buffer.clear()
345        self._client_skip_bytes.clear()
346        self._reset_member_shifts()
347        return True
348
349    async def stop(self) -> None:
350        """Stop playback and cleanup."""
351        if self._audio_source_task and not self._audio_source_task.done():
352            self._audio_source_task.cancel()
353            with suppress(asyncio.CancelledError):
354                await self._audio_source_task
355        await asyncio.gather(
356            *[self.remove_client(x, reason="session stop") for x in self.sync_clients],
357        )
358
359    async def remove_client(
360        self, airplay_player: AirPlayPlayer, reason: str = "client removed"
361    ) -> None:
362        """
363        Remove a sync client from the session.
364
365        :param airplay_player: The player to remove from the session.
366        :param reason: Short human-readable reason for the removal, used in teardown logs.
367        """
368        async with self._lock:
369            if airplay_player not in self.sync_clients:
370                return
371            self.sync_clients.remove(airplay_player)
372        await self._cleanup_after_removal(airplay_player, reason=reason)
373
374    async def stop_client(
375        self, airplay_player: AirPlayPlayer, reason: str = "stop_client called"
376    ) -> None:
377        """
378        Stop a client's stream and ffmpeg.
379
380        :param airplay_player: The player to stop.
381        :param reason: Short human-readable reason for the teardown, used in debug logs.
382        """
383        self.prov.logger.debug(
384            "AirPlay session teardown: session=%s client=%s reason=%s",
385            id(self),
386            airplay_player.player_id,
387            reason,
388        )
389        self._client_skip_bytes.pop(airplay_player.player_id, None)
390        ffmpeg = self._player_ffmpeg.pop(airplay_player.player_id, None)
391        # note that we use kill instead of graceful close here,
392        # because otherwise it can take a very long time for the process to exit.
393        if ffmpeg and not ffmpeg.closed:
394            await ffmpeg.kill()
395        if airplay_player.stream and airplay_player.stream.session == self:
396            airplay_player.stream.reset_reanchor_shift()
397            await airplay_player.stream.stop(force=True)
398
399    async def add_client(self, airplay_player: AirPlayPlayer) -> None:  # noqa: PLR0915
400        """
401        Add a sync client to the session as a late joiner.
402
403        The joiner cannot honour an anchor in the past — cliairplay makes the
404        first post-START stdin byte audible exactly at the instant it acks and
405        then freezes the anchor, with no catch-up. So the anchor is commanded
406        just past the instant the receiver reports its clock becomes usable (and
407        never inside the join floor), the binary acks the instant it can truly
408        honour, and the stream position due at that instant is derived from the
409        group's effective (shift-adjusted) timeline. Depending on where that
410        position falls relative to the ring buffer the joiner is either primed
411        from the ring tail (position at or behind the write head) or has the
412        leading bytes of the live feed skipped (position ahead of the write
413        head), so its first audible sample lands exactly where the group is
414        playing.
415
416        Timing-source readiness, the receiver's clock projection and the START
417        ack are all awaited without the session lock so the rest of the group
418        keeps being fed meanwhile; all buffer and anchor math stays under the
419        lock so ``seconds_streamed``, the ring buffer and the per-client skip
420        counter stay consistent with the live feed.
421        """
422        await self._resolve_late_joiner_ptp(airplay_player)
423        async with self._lock:
424            if not self._session_is_live():
425                return
426        try:
427            await self._start_client(airplay_player, self.use_shared_ptp)
428            stream = airplay_player.stream
429            assert stream
430            await stream.wait_for_connection()
431            # A receiver starts probing its clock at connect, so how long it
432            # still needs is measurable before any anchor is announced. Waiting
433            # for that projection here — outside the session lock, so the rest
434            # of the group keeps being fed — lets the join anchor on the
435            # device's own readiness instead of a fixed guess.
436            readiness, ready_at_unix_ms = await stream.wait_clock_ready(
437                timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
438            )
439        except asyncio.CancelledError:
440            await self.stop_client(airplay_player, reason="late joiner start cancelled")
441            raise
442        except Exception as err:
443            self.prov.logger.warning(
444                "Late joiner %s: failed to connect pipeline: %s",
445                airplay_player.player_id,
446                err,
447            )
448            await self.stop_client(airplay_player, reason="late joiner connection failed")
449            return
450
451        if readiness is ClockReadiness.STALLED:
452            # The receiver never answered our clock, so it would render silence
453            # for as long as it stayed in the group while every other signal
454            # said it was playing. Better to leave it out: the group keeps
455            # playing and the player stays idle, which is the visible truth.
456            # The binary's own report names the device and the ports to check.
457            self.prov.logger.warning(
458                "Late joiner %s: not adding it to the group - its receiver never answered "
459                "the server's PTP clock, so it would render silence",
460                airplay_player.player_id,
461            )
462            await self.stop_client(airplay_player, reason="receiver clock stalled")
463            return
464
465        pcm_sample_size = self._pcm_byte_rate
466        frame_size = self._pcm_frame_size
467
468        def map_to(anchor_at: float, *, committed: bool) -> tuple[float, float, bytes, int]:
469            """
470            Map an anchor instant onto the live feed (call under the lock).
471
472            Snapshots the ring and derives which stream position the group
473            plays at ``anchor_at``, returning the anchor, the due feed
474            position, the prime slice and the live skip.
475
476            :param anchor_at: Instant at which the joiner's first delivered
477                sample becomes audible.
478            :param committed: True once the binary has acked the instant, which
479                fixes it. A due position the ring can no longer reach is then
480                covered with silence rather than by moving the anchor, because
481                moving an instant the binary already owns would offset the
482                joiner from the group by exactly that much.
483            """
484            # Snapshot the ring, which ends at the write head: it covers the
485            # stream positions [seconds_streamed - len(ring)/rate, seconds_streamed].
486            effective_start_time = self.effective_start_time
487            buffered_pcm = bytes(self._pcm_buffer)
488            due = anchor_at - effective_start_time
489            skip = 0
490            prime_slice = b""
491            if due <= self.seconds_streamed:
492                # The due position is at or behind the write head: prime
493                # the joiner from the ring so the live feed then continues
494                # seamlessly.
495                keep_bytes = int((self.seconds_streamed - due) * pcm_sample_size)
496                # Frame-align the prime START in ABSOLUTE stream
497                # coordinates. The prime must end exactly at the write head
498                # (the live feed continues byte-contiguously from there),
499                # so only the start may move: shift it forward to the next
500                # frame boundary (dropping under one frame, ~23 us).
501                start_abs = self._pcm_total_fed - keep_bytes
502                realign = (frame_size - start_abs % frame_size) % frame_size
503                keep_bytes -= realign
504                if realign:
505                    self.prov.logger.debug(
506                        "Late joiner %s: prime start realigned +%d bytes "
507                        "(write head sits mid-frame)",
508                        airplay_player.player_id,
509                        realign,
510                    )
511                # The ring's own head is frame-aligned only by accident - it is
512                # trimmed on byte overflow - so align it too before measuring
513                # how much of the requested prime it can really serve.
514                ring_head_abs = self._pcm_total_fed - len(buffered_pcm)
515                ring_realign = (frame_size - ring_head_abs % frame_size) % frame_size
516                servable = buffered_pcm[ring_realign:]
517                if keep_bytes > len(servable):
518                    # The due position predates the ring: the write head ran
519                    # further ahead of the audible position than the ring was
520                    # sized for. Those samples are gone, so the join either
521                    # starts a little later (anchor still free to move) or opens
522                    # with that much silence (anchor already acked) - both keep
523                    # the real content on the instant the group plays it, which
524                    # dropping the missing head would not.
525                    missing = keep_bytes - len(servable)
526                    lead = self.seconds_streamed - (time.time() - effective_start_time)
527                    if committed:
528                        # One ring's worth is already far past any real
529                        # shortfall, so it bounds what an implausible ack (one
530                        # mapping back near the session start) can allocate.
531                        pad = min(missing, self._pcm_buffer_max)
532                        prime_slice = bytes(pad) + servable
533                        # A clamped pad cannot reach the due position, so the
534                        # joiner really does end up ahead of the group by the
535                        # remainder. Say which of the two happened.
536                        residual = (missing - pad) / pcm_sample_size
537                        self.prov.logger.warning(
538                            "Late joiner %s: the feed runs %.2fs ahead of the audible "
539                            "position, past the %.2fs of audio kept for a join; opening "
540                            "with %.2fs of silence to cover the missing head. %s Please "
541                            "report this with a debug log.",
542                            airplay_player.player_id,
543                            lead,
544                            len(servable) / pcm_sample_size,
545                            pad / pcm_sample_size,
546                            f"That still leaves it {residual:.2f}s ahead of the group."
547                            if residual
548                            else "The content itself still lands in sync; the joiner is "
549                            "only audible that much late.",
550                        )
551                    else:
552                        due += missing / pcm_sample_size
553                        anchor_at = effective_start_time + due
554                        prime_slice = servable
555                        self.prov.logger.debug(
556                            "Late joiner %s: due position predates the %.2fs ring "
557                            "(feed runs %.2fs ahead); anchoring %.2fs later, on the "
558                            "ring's oldest sample",
559                            airplay_player.player_id,
560                            len(servable) / pcm_sample_size,
561                            lead,
562                            missing / pcm_sample_size,
563                        )
564                elif keep_bytes > 0:
565                    prime_slice = servable[len(servable) - keep_bytes :]
566            else:
567                # The due position is ahead of the write head: skip that
568                # many bytes off the head of the live feed so the joiner's
569                # first delivered byte is the sample due at the anchor.
570                skip = int((due - self.seconds_streamed) * pcm_sample_size)
571                # The first delivered live byte must land on a frame
572                # boundary in ABSOLUTE stream coordinates (round the skip
573                # up, under one frame).
574                first_abs = self._pcm_total_fed + skip
575                skip += (frame_size - first_abs % frame_size) % frame_size
576            return anchor_at, due, prime_slice, skip
577
578        now = time.time()
579        self.prov.logger.debug(
580            "Late joiner %s: receiver clock %s",
581            airplay_player.player_id,
582            _CLOCK_READINESS_NOTES.get(readiness, "").format(
583                out=ready_at_unix_ms / 1000 - now,
584                timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000,
585            ),
586        )
587        async with self._lock:
588            if not self._session_is_live():
589                await self.stop_client(airplay_player, reason="session ended during late join")
590                return
591            # cliairplay makes the first post-START stdin byte audible exactly at
592            # the instant it acks, so the anchor is commanded first and the
593            # content mapped onto the ack afterwards. It sits just past the
594            # instant the receiver's clock becomes usable; the floor is only a
595            # lower bound, and carries the whole anchor when no projection came.
596            min_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
597            anchor_at = now + min_headroom
598            if ready_at_unix_ms:
599                anchor_at = max(
600                    anchor_at,
601                    (ready_at_unix_ms + AIRPLAY_CLOCK_READY_LEAD_MS) / 1000,
602                )
603            requested_at, fed_pos_due = map_to(anchor_at, committed=False)[:2]
604            start_unix_ms = int(requested_at * 1000)
605            sync_adjust = airplay_player.config.get_value(CONF_SYNC_ADJUST, 0)
606            adjust_ms = sync_adjust if isinstance(sync_adjust, int) else 0
607            position_ms = int(((self.media.elapsed_time or 0) + fed_pos_due) * 1000)
608
609        try:
610            # Anchor the joiner BEFORE feeding the prime: pre-START the binary
611            # only buffers stdin into its bounded ring and sends nothing, so a
612            # prime longer than that ring would wedge that write — and, through
613            # the session lock, stall the whole group's feed. Anchored, the
614            # binary drains the prime as it streams in. The START does not
615            # re-anchor the session timeline (the group keeps playing). Its ack
616            # is awaited WITHOUT the session lock: the binary can hold that ack
617            # until its receiver clock verification resolves, which would
618            # otherwise starve every other member's feed for that whole wait.
619            actual = await stream.start(start_unix_ms + adjust_ms, position_ms, join=True)
620        except asyncio.CancelledError:
621            await self.stop_client(airplay_player, reason="late joiner start cancelled")
622            raise
623        except Exception as err:
624            self.prov.logger.warning(
625                "Late joiner %s: failed to start/prime pipeline: %r",
626                airplay_player.player_id,
627                err,
628            )
629            await self.stop_client(airplay_player, reason="late joiner start/prime failed")
630            return
631
632        try:
633            async with self._lock:
634                if not self._session_is_live():
635                    await self.stop_client(airplay_player, reason="session ended during late join")
636                    return
637                if airplay_player.stream is not stream or not stream.running:
638                    await self.stop_client(
639                        airplay_player, reason="late joiner stopped during start"
640                    )
641                    return
642                # Map the content onto the instant the binary acked — its
643                # verified truth. The ring is re-snapshotted here, so the mapping
644                # covers whatever the group was fed while the ack was outstanding.
645                acked_at = (actual - adjust_ms) / 1000
646                start_at, fed_pos_due, prime, skip_bytes = map_to(acked_at, committed=True)
647                self._client_skip_bytes[airplay_player.player_id] = skip_bytes
648                # An instant that moved carries the content mapped onto it
649                # further into the stream than the position sent with the
650                # command, so progress is reported against the sample that
651                # actually lands on the anchor.
652                acked_position_ms = int(((self.media.elapsed_time or 0) + fed_pos_due) * 1000)
653                if acked_position_ms != position_ms:
654                    stream.rebase_position(acked_position_ms)
655
656                self.prov.logger.debug(
657                    "Late joiner %s: priming %.2fs, skipping %.2fs, stream_pos=%.2fs, "
658                    "fed_pos_due=%.2fs, start_at is %.2fs from now, acked %+d ms off the "
659                    "commanded instant (min_headroom=%.2fs, effective_shift=%.2fs, "
660                    "write_head_lead=%.2fs, peak=%.2fs, ring=%.2fs)",
661                    airplay_player.player_id,
662                    len(prime) / pcm_sample_size,
663                    skip_bytes / pcm_sample_size,
664                    self.seconds_streamed,
665                    fed_pos_due,
666                    start_at - time.time(),
667                    int(acked_at * 1000) - start_unix_ms,
668                    min_headroom,
669                    self.effective_start_time - self.start_time,
670                    self.seconds_streamed - (time.time() - self.effective_start_time),
671                    self._peak_lead_seconds,
672                    self._pcm_buffer_max / pcm_sample_size,
673                )
674
675                if prime:
676                    try:
677                        await self._write_chunk_to_player(airplay_player, prime)
678                    except Exception as err:
679                        self.prov.logger.warning(
680                            "Late joiner %s: failed to start/prime pipeline: %r",
681                            airplay_player.player_id,
682                            err,
683                        )
684                        await self.stop_client(
685                            airplay_player, reason="late joiner start/prime failed"
686                        )
687                        return
688
689                if airplay_player not in self.sync_clients:
690                    self.sync_clients.append(airplay_player)
691                if (
692                    airplay_player.protocol == StreamingProtocol.AIRPLAY2
693                    and not self.use_shared_ptp
694                ):
695                    ap2_members = sum(
696                        1
697                        for player in self.sync_clients
698                        if player.protocol == StreamingProtocol.AIRPLAY2
699                    )
700                    self._warn_degraded_shared_ptp(ap2_members)
701        except asyncio.CancelledError:
702            # the joiner is anchored and audible from here on, so a cancellation
703            # before it is part of the session must still tear it down
704            await self.stop_client(airplay_player, reason="late joiner start cancelled")
705            raise
706
707        self.prov.logger.debug(
708            "Late joiner %s: started after %.2fs",
709            airplay_player.player_id,
710            time.time() - now,
711        )
712
713    def _ring_bytes_for(self, lead_seconds: float) -> int:
714        """
715        Return the late-join ring size that covers a given write-head lead.
716
717        :param lead_seconds: Lead the ring has to span, before margin.
718        """
719        seconds = max(
720            lead_seconds + AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS,
721            AIRPLAY_LATE_JOIN_RING_MIN_SECONDS,
722        )
723        size = min(int(seconds * self._pcm_byte_rate), AIRPLAY_LATE_JOIN_RING_MAX_BYTES)
724        # Keep it a whole number of frames so it can bound a silence pad without
725        # knocking the prime off its frame boundaries.
726        return size - size % self._pcm_frame_size
727
728    def _observe_write_head_lead(self) -> None:
729        """
730        Track how far the feed runs ahead of the audible position (call under the lock).
731
732        A joiner's anchor maps into exactly this span, so the ring is grown to
733        the largest lead the session has shown. It only ever grows: shrinking it
734        mid-session would discard history a joiner still needs, and the lead is
735        at its largest right after the anchor is placed - the whole downstream
736        pipeline is full by then - so the ring is sized long before any late
737        join can arrive.
738        """
739        if self.start_time <= 0:
740            return  # not anchored yet, so there is no audible position to measure against
741        now = time.time()
742        if now < self.effective_start_time:
743            # Still inside the start lead: everything fed is ahead of an anchor
744            # that has not arrived, which would read as a lead seconds larger
745            # than the pipeline really holds.
746            return
747        lead = self.seconds_streamed - (now - self.effective_start_time)
748        if lead <= self._peak_lead_seconds:
749            return
750        self._peak_lead_seconds = lead
751        self._pcm_buffer_max = self._ring_bytes_for(lead)
752
753    def _session_is_live(self) -> bool:
754        """Return whether the session still plays a joinable timeline (call under the lock)."""
755        if not self.sync_clients:
756            return False
757        reference = self.sync_clients[0]
758        reference_stream = reference.stream
759        if reference_stream is None or not reference_stream.running:
760            return False
761        # A parked (standby) session keeps every member's stream running while
762        # its timeline is gone - the anchor is stale and nothing is being fed -
763        # so only a member that is actually playing can absorb a joiner.
764        return reference.playback_state == PlaybackState.PLAYING
765
766    async def _resolve_shared_ptp(self, ap2_members: int | None = None) -> bool:
767        """
768        Decide, once per session, whether members attach to the shared PTP daemon.
769
770        The decision is session-wide and gated on the daemon actually being ready
771        (bound to 319/320 with its control channel open), not merely spawned, so a
772        group start cannot race the daemon and end up mixing PTP and NTP members.
773
774        :param ap2_members: Number of native AirPlay 2 members that will use the
775            decision. Defaults to the current session members.
776        :return: True if every native AirPlay 2 member should use the shared PTP
777            clock; False to degrade the whole session consistently (no member
778            attaches to the daemon).
779        """
780        if ap2_members is None:
781            ap2_members = sum(
782                1 for p in self.sync_clients if p.protocol == StreamingProtocol.AIRPLAY2
783            )
784        if not ap2_members:
785            return False
786        if await self.prov.wait_ptp_daemon_ready():
787            return True
788        # Daemon not ready: keep the group coherent by attaching no one to the
789        # shared clock. A lone native AP2 player can still self-bind its own PTP
790        # with no partner to drift against; only a real multi-room group loses
791        # tight sync, so warn just for that case.
792        self._warn_degraded_shared_ptp(ap2_members)
793        return False
794
795    async def _resolve_late_joiner_ptp(self, airplay_player: AirPlayPlayer) -> None:
796        """Resolve the session timing source before its first late AirPlay 2 join."""
797        if airplay_player.protocol != StreamingProtocol.AIRPLAY2:
798            return
799        async with self._ptp_lock:
800            if self._shared_ptp_resolved:
801                return
802            async with self._lock:
803                ap2_members = sum(
804                    1
805                    for player in self.sync_clients
806                    if player.protocol == StreamingProtocol.AIRPLAY2
807                ) + (airplay_player not in self.sync_clients)
808            self.use_shared_ptp = await self._resolve_shared_ptp(ap2_members)
809            self._shared_ptp_resolved = True
810
811    def _warn_degraded_shared_ptp(self, ap2_members: int) -> None:
812        """Warn once when multiple AirPlay 2 members cannot share the PTP clock."""
813        if ap2_members <= 1 or self._ptp_degraded_warning_logged:
814            return
815        self._ptp_degraded_warning_logged = True
816        self.prov.logger.warning(
817            "Shared PTP clock daemon not ready - native AirPlay 2 multi-room sync "
818            "for this group of %d players is degraded and members may drift. The "
819            "server likely cannot bind the privileged PTP ports (UDP 319/320); "
820            "running without root or CAP_NET_BIND_SERVICE is the common cause.",
821            ap2_members,
822        )
823
824    async def _cleanup_after_removal(
825        self, airplay_player: AirPlayPlayer, reason: str = "client removed"
826    ) -> None:
827        """
828        Clean up processes and state after a client has been removed from sync_clients.
829
830        :param airplay_player: The player whose processes should be stopped.
831        :param reason: Short human-readable reason, forwarded to stop_client for logging.
832        """
833        stream = airplay_player.stream
834        if stream is not None and stream.session != self:
835            stream = None
836        await self.stop_client(airplay_player, reason=reason)
837        # Only set IDLE if the player's stream still belongs to this session,
838        # otherwise a re-add to a new session may have already set a new state.
839        if stream is not None:
840            airplay_player.set_state_from_stream(PlaybackState.IDLE, stream=stream)
841        # Re-check sync_clients under the lock to avoid racing with add_client.
842        async with self._lock:
843            should_stop = not self.sync_clients
844        if should_stop:
845            await self.stop()
846
847    async def _audio_streamer(self, audio_source: AsyncGenerator[bytes]) -> None:
848        """Stream audio to all players."""
849        stream_error: BaseException | None = None
850        try:
851            async for chunk in audio_source:
852                if not self.sync_clients:
853                    break
854
855                has_running_clients = await self._write_chunk_to_all_players(chunk)
856                if not has_running_clients:
857                    self.prov.logger.debug("No running clients remaining, stopping audio streamer")
858                    break
859        except asyncio.CancelledError:
860            self.prov.logger.debug("Audio streamer cancelled after %.1fs", self.seconds_streamed)
861            raise
862        except Exception as err:
863            stream_error = err
864            self.prov.logger.error(
865                "Audio source error after %.1fs of streaming: %s",
866                self.seconds_streamed,
867                err,
868                exc_info=err,
869            )
870        finally:
871            if stream_error:
872                self.prov.logger.warning(
873                    "Stream ended prematurely due to error - notifying players"
874                )
875        async with self._lock:
876            await asyncio.gather(
877                *[
878                    self._write_eof_to_player(x)
879                    for x in self.sync_clients
880                    if x.stream and x.stream.running
881                ],
882                return_exceptions=True,
883            )
884
885    async def _write_chunk_to_all_players(self, chunk: bytes) -> bool:
886        """
887        Write a chunk to all connected players.
888
889        :return: True if there are still running clients, False otherwise.
890        """
891        async with self._lock:
892            sync_clients = [x for x in self.sync_clients if x.stream and x.stream.running]
893            if not sync_clients:
894                return False
895
896            # Update seconds_streamed and ring buffer under the lock so
897            # add_client always reads consistent values.
898            self.seconds_streamed += len(chunk) / self._pcm_byte_rate
899            self._pcm_total_fed += len(chunk)
900            self._observe_write_head_lead()
901            self._pcm_buffer.extend(chunk)
902            overflow = len(self._pcm_buffer) - self._pcm_buffer_max
903            if overflow > 0:
904                del self._pcm_buffer[:overflow]
905
906            # Write chunk to all players
907            write_tasks = [self._write_chunk_to_player(x, chunk) for x in sync_clients if x.stream]
908            results = await asyncio.gather(*write_tasks, return_exceptions=True)
909
910            # Check for write errors or timeouts
911            players_to_remove: list[tuple[AirPlayPlayer, str]] = []
912            for i, result in enumerate(results):
913                if i >= len(sync_clients):
914                    continue
915                player = sync_clients[i]
916
917                if isinstance(result, TimeoutError):
918                    self.prov.logger.warning(
919                        "Removing player %s from session: stopped reading data (write timeout)",
920                        player.player_id,
921                    )
922                    players_to_remove.append((player, "audio write timeout"))
923                elif isinstance(result, Exception):
924                    self.prov.logger.warning(
925                        "Removing player %s from session due to write error: %s",
926                        player.player_id,
927                        result,
928                    )
929                    players_to_remove.append((player, f"audio write error: {result}"))
930
931            # Remove failed players from sync_clients immediately under the lock
932            # so they are excluded from future write cycles. Only defer process
933            # cleanup (_cleanup_after_removal) — this prevents fire-and-forget
934            # remove_client calls from racing with a subsequent add_client when
935            # a player is being moved between groups.
936            for player, removal_reason in players_to_remove:
937                if player in self.sync_clients:
938                    self.sync_clients.remove(player)
939                self.mass.create_task(self._cleanup_after_removal(player, reason=removal_reason))
940
941            remaining_clients = len(sync_clients) - len(players_to_remove)
942            return remaining_clients > 0
943
944    async def _write_chunk_to_player(self, airplay_player: AirPlayPlayer, chunk: bytes) -> None:
945        """Write audio chunk to a player's ffmpeg process."""
946        player_id = airplay_player.player_id
947        # Drain any pending late-join skip first: a joiner anchored ahead of the
948        # write head must drop that many leading bytes of the live feed so its
949        # first delivered byte is the sample due at its anchor.
950        if skip := self._client_skip_bytes.get(player_id, 0):
951            if skip >= len(chunk):
952                self._client_skip_bytes[player_id] = skip - len(chunk)
953                return
954            chunk = chunk[skip:]
955            self._client_skip_bytes[player_id] = 0
956        if ffmpeg := self._player_ffmpeg.get(player_id):
957            if ffmpeg.closed:
958                return
959            await asyncio.wait_for(ffmpeg.write(chunk), timeout=35.0)
960
961    async def _write_eof_to_player(self, airplay_player: AirPlayPlayer) -> None:
962        """Write EOF to a specific player."""
963        if ffmpeg := self._player_ffmpeg.pop(airplay_player.player_id, None):
964            await ffmpeg.write_eof()
965            await ffmpeg.wait_with_timeout(30)
966            if airplay_player.stream:
967                await airplay_player.stream.write_audio_eof()
968
969    async def _member_start_step(
970        self, airplay_player: AirPlayPlayer, step: str, awaitable: Coroutine[Any, Any, None]
971    ) -> None:
972        """
973        Run one per-member step of a group start, naming the member if it fails.
974
975        A group start fans its members out over a task group and a gather, both
976        of which collapse into a single exception at the caller - so with five
977        speakers connecting, nothing in the log says which one failed. That,
978        with whatever reason its binary reported, is the whole diagnostic.
979
980        :param airplay_player: The member the step belongs to.
981        :param step: What the member was doing, for the failure message.
982        :param awaitable: The step to run.
983        """
984        try:
985            await awaitable
986        except asyncio.CancelledError:
987            raise
988        except Exception as err:
989            self.prov.logger.warning(
990                "AirPlay group start: %s failed to %s: %s",
991                airplay_player.display_name,
992                step,
993                err,
994            )
995            raise
996
997    async def _start_client(self, airplay_player: AirPlayPlayer, use_shared_ptp: bool) -> None:
998        """
999        Connect a CLI process and start its ffmpeg for a single client.
1000
1001        :param airplay_player: The player to start streaming to.
1002        :param use_shared_ptp: The session-wide shared-PTP decision applied to
1003            this member so the whole group shares one timing source.
1004        """
1005        # joining a session supersedes any pending automatic group re-join
1006        airplay_player.cancel_group_rejoin()
1007        # sync volume from parent player if needed
1008        airplay_player.sync_volume_level()
1009        if airplay_player.stream and airplay_player.stream.running:
1010            await airplay_player.stream.stop()
1011        stream_pcm_format = airplay_player.get_stream_pcm_format(self.pcm_format)
1012        airplay_player.stream = AirPlayStream(airplay_player, pcm_format=stream_pcm_format)
1013        airplay_player.stream.session = self
1014        await airplay_player.stream.connect(use_shared_ptp)
1015        await self._start_player_ffmpeg(airplay_player, self.media)
1016
1017    def _anchor_start_unix_ms(self, *, warm: bool = False, ready_at_unix_ms: int = 0) -> int:
1018        """
1019        Return the shared audible-start instant for a readiness-confirmed start.
1020
1021        :param warm: True for a warm re-start over live connections (seek/next/
1022            resume-from-park). Members on the splice timeline report a
1023            minimum warm lead — their queued audio plays out before the new
1024            content can begin — and the shared anchor must sit beyond the
1025            largest member value so every member splices at the same instant.
1026        :param ready_at_unix_ms: Latest instant at which a member's receiver
1027            clock becomes usable, as the binaries reported it. The anchor never
1028            lands before it. 0 when no member reported a projection, leaving the
1029            lead below as the whole anchor.
1030        """
1031        if len(self.sync_clients) == 1:
1032            lead_ms = AIRPLAY_START_LEAD_MS
1033        elif warm:
1034            lead_ms = AIRPLAY_GROUP_START_LEAD_MS
1035        else:
1036            # Cold group start: cover the members' receiver-side clock
1037            # acquisition (see AIRPLAY_COLD_GROUP_START_LEAD_MS).
1038            lead_ms = AIRPLAY_COLD_GROUP_START_LEAD_MS
1039        anchor = int(time.time() * 1000) + lead_ms
1040        if ready_at_unix_ms:
1041            anchor = max(anchor, ready_at_unix_ms + AIRPLAY_CLOCK_READY_LEAD_MS)
1042        if not warm:
1043            return anchor
1044        # Splice-timeline members honor the commanded instant only when that
1045        # instant (plus their sync_adjust) lands beyond their queued audio.
1046        # A NEGATIVE sync_adjust moves a member's commanded instant earlier,
1047        # eating into the lead, so it must be added to that member's
1048        # requirement — otherwise the first round can never succeed for that
1049        # member and every group start pays a corrective round.
1050        member_requirement = 0
1051        for player in self.sync_clients:
1052            stream = player.stream
1053            if stream is None or stream.warm_lead_ms <= 0:
1054                continue
1055            sync_adjust = player.config.get_value(CONF_SYNC_ADJUST, 0)
1056            adjust_ms = sync_adjust if isinstance(sync_adjust, int) else 0
1057            member_requirement = max(member_requirement, stream.warm_lead_ms - min(0, adjust_ms))
1058        if member_requirement > 0:
1059            anchor = max(
1060                anchor,
1061                int(time.time() * 1000) + member_requirement + AIRPLAY_SPLICE_LEAD_MARGIN_MS,
1062            )
1063        for player in self.sync_clients:
1064            stream = player.stream
1065            if stream is None or stream.flushed_head_unix_ms <= 0:
1066                continue
1067            sync_adjust = player.config.get_value(CONF_SYNC_ADJUST, 0)
1068            adjust_ms = sync_adjust if isinstance(sync_adjust, int) else 0
1069            # The member's commanded instant is anchor + adjust; it must clear
1070            # the member's frozen head with margin for the command round-trip.
1071            anchor = max(
1072                anchor,
1073                stream.flushed_head_unix_ms - adjust_ms + AIRPLAY_SPLICE_LEAD_MARGIN_MS,
1074            )
1075        return anchor
1076
1077    async def _wait_members_audio_present(self) -> None:
1078        """Wait until every member's binary reports the new audio flowing."""
1079        members = [(p, p.stream) for p in self.sync_clients if p.stream]
1080        results = await asyncio.gather(*[stream.wait_audio_present() for _, stream in members])
1081        if all(results):
1082            return
1083        # Name the members that never reported audio: they are what has to be
1084        # looked at, and the group start below is abandoned for all of them.
1085        silent = [
1086            player.display_name
1087            for (player, _), present in zip(members, results, strict=True)
1088            if not present
1089        ]
1090        raise PlayerCommandFailed(f"audio feed was not confirmed by {', '.join(silent)}")
1091
1092    async def _wait_members_clock_ready(self) -> int:
1093        """
1094        Return the latest receiver-clock readiness any member reported.
1095
1096        Members that report nothing contribute no instant to the maximum; the
1097        caller anchors those on its lead alone.
1098
1099        :return: Unix epoch ms of the latest projection any member reported, or 0
1100            when there is nothing to wait for — a receiver on NTP timing, or one
1101            that never answered. The caller then anchors on its lead alone.
1102        """
1103        # A solo start waits for the projection too: a receiver that has not
1104        # seated its clock renders silence at an anchor it cannot honor, and
1105        # enough of them need that time (WiiM, Edifier) that no start may assume
1106        # otherwise. It costs a warm clock nothing - the binary reports it ready
1107        # with a past instant right after connect - and a cold one anchors just
1108        # past its own projection instead of being corrected there by the binary
1109        # afterwards: the same instant, planned rather than repaired, with the
1110        # readiness lead's slack on top.
1111        results = await asyncio.gather(
1112            *[
1113                p.stream.wait_clock_ready(timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000)
1114                for p in self.sync_clients
1115                if p.stream
1116            ]
1117        )
1118        ready_at_unix_ms = max(
1119            (at for readiness, at in results if readiness is ClockReadiness.PROJECTED), default=0
1120        )
1121        # A group start does not drop a member that stalled - the rest of the
1122        # group would still be started, and the member is already warned about
1123        # by name, with the ports to check, where the binary reported it. Say
1124        # which outcomes were seen so the anchor decision is readable.
1125        unprojected = [
1126            readiness for readiness, _ in results if readiness is not ClockReadiness.PROJECTED
1127        ]
1128        if unprojected:
1129            self.prov.logger.debug(
1130                "AirPlay start: %d of %d member(s) reported no receiver clock projection (%s); "
1131                "anchoring those on the start lead alone",
1132                len(unprojected),
1133                len(results),
1134                ", ".join(sorted({readiness.value for readiness in unprojected})),
1135            )
1136        return ready_at_unix_ms
1137
1138    async def _start_members(self, position_ms: int, start_unix_ms: int) -> None:
1139        """
1140        Anchor every member's playback at one shared audible instant.
1141
1142        The binaries verify the instant: each ack carries the TRUE scheduled
1143        instant (an infeasible one is corrected forward, never silently
1144        misplaced). When any member was corrected, every member is re-STARTed
1145        at the largest reported instant so the group converges on one shared
1146        instant; the recorded session anchor is always the verified truth.
1147        A solo member is never re-STARTed: its corrected instant is simply
1148        adopted as the anchor, since there is no partner to converge with.
1149
1150        :param position_ms: Media position mapped to the first sample of the anchor.
1151        :param start_unix_ms: Shared audible-start instant in unix epoch ms.
1152        """
1153        target_ms = start_unix_ms
1154        corrected_ms = start_unix_ms
1155        for _ in range(4):
1156            member_tasks: list[tuple[int, asyncio.Task[int]]] = []
1157            async with asyncio.TaskGroup() as task_group:
1158                for player in self.sync_clients:
1159                    stream = player.stream
1160                    if stream is None:
1161                        continue
1162                    sync_adjust = player.config.get_value(CONF_SYNC_ADJUST, 0)
1163                    adjust_ms = sync_adjust if isinstance(sync_adjust, int) else 0
1164                    member_tasks.append(
1165                        (
1166                            adjust_ms,
1167                            task_group.create_task(
1168                                stream.start(target_ms + adjust_ms, position_ms)
1169                            ),
1170                        )
1171                    )
1172            corrected_ms = target_ms
1173            for adjust_ms, task in member_tasks:
1174                corrected_ms = max(corrected_ms, task.result() - adjust_ms)
1175            if corrected_ms <= target_ms + 2:
1176                break
1177            if len(member_tasks) == 1:
1178                # A lone member has no partner to converge with and its binary
1179                # already scheduled the corrected instant exactly, so adopt that
1180                # as the anchor. Re-STARTing it only does damage: the command
1181                # re-bases reported position on the raw position_ms (start()
1182                # writes that base unconditionally), throwing away the
1183                # correction the anchor already folded into it, and if the
1184                # second ack times out the session records an instant the
1185                # binary never played on.
1186                target_ms = corrected_ms
1187                break
1188            self.prov.logger.warning(
1189                "AirPlay group start corrected: a member could not honor %d, "
1190                "re-anchoring all members at %d (+%d ms)",
1191                target_ms,
1192                corrected_ms,
1193                corrected_ms - target_ms,
1194            )
1195            # The corrected instant already carries the binary's own
1196            # command-latency slack; the extra margin covers fanning the
1197            # retry out to every member so the next round lands.
1198            target_ms = corrected_ms + AIRPLAY_SPLICE_LEAD_MARGIN_MS
1199        else:
1200            # No round landed, so the members sit on the instants they last
1201            # reported, not on the retry that was about to be commanded. The
1202            # anchor has to record where they actually are, or every later
1203            # joiner maps against a timeline the group never played.
1204            target_ms = corrected_ms
1205            self.prov.logger.error(
1206                "AirPlay group start did not converge after 4 rounds "
1207                "(members last reported %d) - they may be audibly out of sync; "
1208                "please report this with a debug log",
1209                target_ms,
1210            )
1211        self.start_unix_ms = target_ms
1212        self.start_time = target_ms / 1000
1213
1214    async def _flush_member(self, player: AirPlayPlayer) -> bool:
1215        """Flush one member's live stream in place and report the binary's ack."""
1216        stream = player.stream
1217        if stream is None:
1218            return False
1219        return await stream.flush()
1220
1221    def _reset_member_shifts(self) -> None:
1222        """Zero every member's accumulated starvation shift for a fresh anchor."""
1223        for player in self.sync_clients:
1224            if player.stream is not None:
1225                player.stream.reset_reanchor_shift()
1226
1227    async def _start_player_ffmpeg(self, player: AirPlayPlayer, media: PlayerMedia) -> None:
1228        """
1229        Start the per-seek ffmpeg feeding a member's persistent cli stdin.
1230
1231        Retires any ffmpeg still tracked for the player and wires a fresh one to
1232        the same cli stdin fd. Killing ffmpeg never closes cli stdin (MA holds
1233        the write end via its process transport), so the binary's stdin reader
1234        survives a warm seek.
1235
1236        :param player: The member whose ffmpeg is (re)started.
1237        :param media: Media whose queue/session identify the output plan.
1238        """
1239        if ffmpeg := self._player_ffmpeg.pop(player.player_id, None):
1240            await ffmpeg.close()
1241        stream = player.stream
1242        assert stream
1243        handoff_format = stream.pcm_format
1244        output_plan = self.mass.streams.audio.get_player_output_plan(
1245            player.player_id,
1246            input_format=self.pcm_format,
1247            output_format=get_final_output_format(handoff_format),
1248            handoff_format=handoff_format,
1249            queue_id=media.source_id,
1250            session_id=get_media_session_id(media),
1251        )
1252        cli_proc = stream._cli_proc
1253        assert cli_proc
1254        assert cli_proc.proc
1255        assert cli_proc.proc.stdin
1256        stdin_transport = cli_proc.proc.stdin.transport
1257        audio_output: str | int = stdin_transport.get_extra_info("pipe").fileno()
1258        ffmpeg = FFMpeg(
1259            audio_input="-",
1260            input_format=self.pcm_format,
1261            output_format=handoff_format,
1262            filter_params=output_plan.filter_params,
1263            audio_output=audio_output,
1264        )
1265        await ffmpeg.start()
1266        self._player_ffmpeg[player.player_id] = ffmpeg
1267
1268
1269def _first_music_assistant_error(err: BaseException) -> MusicAssistantError | None:
1270    """
1271    Return the first MusicAssistantError inside an error or (nested) exception group.
1272
1273    :param err: The error to inspect.
1274    """
1275    if isinstance(err, MusicAssistantError):
1276        return err
1277    if isinstance(err, BaseExceptionGroup):
1278        for nested in err.exceptions:
1279            if (found := _first_music_assistant_error(nested)) is not None:
1280                return found
1281    return None
1282