music-assistant-server

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