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