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