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