Skip to main content

dash_runtime/client/
engine.rs

1//! [`DashClient`]: the sans-IO engine for static (VOD) and dynamic (live)
2//! presentations. The live half (clock ladder, refresh, join, availability)
3//! is in `live`.
4
5mod live;
6
7use std::collections::{BTreeMap, BTreeSet, VecDeque};
8use std::sync::Arc;
9
10use container_probe::segment;
11use transmux::{TrackId, TrackSpec};
12use url::Url;
13
14use super::action::{Action, RepId, ResourceId};
15use super::config::{DashConfig, PullLimits};
16use super::demux::{self, Staged};
17use super::error::{Error, Failure, FailureClass, FailureKind, FatalKind, Rejection, Result};
18use super::meta::ResponseMeta;
19use super::output::{AudioRendition, DropCause, ManifestSelection, Notice, Output, VariantInfo};
20use crate::address::base_index;
21use crate::address::{AddressError, PlanError, PlanOptions, SegmentPlan, plan_representation};
22use crate::error::Unsupported;
23use crate::mpd::descriptors::SCHEME_PERIOD_CONTINUITY;
24use crate::mpd::inherit::{SegmentInfo, effective_common, resolve_segment_info};
25use crate::mpd::model::{ByteRange, Descriptor, Mpd, PresentationType};
26use crate::mpd::urls::{BaseCandidate, append_query, representation_candidates, segment_url_query};
27use crate::selection::{RepRef, Selection, SetClass, select};
28use crate::timing::clock::UtcClock;
29use crate::timing::period::period_timings;
30
31/// A missing last segment shorter than this, in microseconds, is the tail a
32/// muxer declared past the media (one frame of the slowest common video
33/// rate), not a lost segment.
34const MISSING_TAIL_MAX_US: i128 = 50_000;
35/// Consecutive failures on one BaseURL alternate before the next is used
36/// (ETSI TS 103 285 10.8.2).
37const ROTATE_AFTER_FAILURES: u32 = 2;
38const STATUS_NOT_FOUND: u16 = 404;
39const STATUS_GONE: u16 = 410;
40const STATUS_TOO_EARLY: u16 = 425;
41/// Delivered resources remembered to answer a second delivery with
42/// `DuplicateResource`; older ones are forgotten so a long live session stays
43/// bounded.
44const MAX_REMEMBERED_DONE: usize = 4096;
45const START_WITH_SAP_1: u8 = 1;
46const START_WITH_SAP_2: u8 = 2;
47
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49enum Phase {
50    NeedMpd,
51    Running,
52    Ended,
53    Fatal,
54}
55
56enum Candidate {
57    Ready(i128),
58    Wait(i128),
59    Idle,
60}
61
62#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
63struct InitKey {
64    url: String,
65    range: Option<(u64, u64)>,
66}
67
68#[derive(Debug, Clone, Copy, PartialEq, Eq)]
69enum InitState {
70    Wanted,
71    InFlight,
72    Ready,
73}
74
75struct InitEntry {
76    state: InitState,
77    bytes: Option<Arc<Vec<u8>>>,
78    specs: Vec<TrackSpec>,
79    skipped_reported: BTreeSet<(Option<u32>, String)>,
80}
81
82#[derive(Debug, Clone, Copy, PartialEq, Eq)]
83enum ReqKind {
84    Init,
85    Index,
86    Segment,
87}
88
89struct ReqInfo {
90    rep: RepId,
91    reference: String,
92    range: Option<(u64, u64)>,
93    kind: ReqKind,
94    avail_from_us: Option<i128>,
95    avail_until_us: Option<i128>,
96}
97
98struct IndexReq {
99    range: ByteRange,
100}
101
102struct RepState {
103    id: RepId,
104    at: RepRef,
105    candidates: Vec<BaseCandidate>,
106    cand: usize,
107    cand_failures: u32,
108    query: Option<String>,
109    plan: Option<SegmentPlan>,
110    index: Option<IndexReq>,
111    init_key: InitKey,
112    next_issue: u64,
113    start_time_us: Option<i128>,
114    dead: bool,
115    set_id: Option<u64>,
116    set_pos: usize,
117    rep_key: String,
118    joined: bool,
119    last_time: Option<u64>,
120}
121
122enum SlotState {
123    Issued,
124    Held(Vec<u8>),
125    Staged(Staged),
126    Skipped { marker: bool },
127    Marker,
128}
129
130struct Slot {
131    id: Option<ResourceId>,
132    rep: usize,
133    plan_index: u64,
134    state: SlotState,
135}
136
137#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
138struct TrackIdentity {
139    kind: SetClass,
140    set_id: Option<u64>,
141    group: Option<u64>,
142    lang: Option<String>,
143    roles: Vec<(String, Option<String>)>,
144}
145
146struct Lane {
147    kind: SetClass,
148    set_id: Option<u64>,
149    identity: TrackIdentity,
150    reps: Vec<RepState>,
151    queue: VecDeque<Slot>,
152    base: u32,
153    count: u32,
154    announced_rep: Option<usize>,
155    announced_init: Option<InitKey>,
156    expired_run: u32,
157}
158
159impl Lane {
160    fn current(&self) -> Option<&RepState> {
161        self.reps.last().filter(|r| !r.dead)
162    }
163
164    fn exhausted(&self) -> bool {
165        match self.current() {
166            None => true,
167            Some(rep) => rep
168                .plan
169                .as_ref()
170                .is_some_and(|p| p.len().is_some_and(|n| rep.next_issue >= n)),
171        }
172    }
173}
174
175struct PeriodRun {
176    index: usize,
177    id: Option<String>,
178    start_us: i128,
179    lanes: Vec<Lane>,
180    selection: Selection,
181    notices: Vec<Output>,
182    started: bool,
183    issue_done: bool,
184    finished: bool,
185}
186
187struct PrevPeriod {
188    id: Option<String>,
189    lanes: Vec<(SetClass, Option<u64>, InitKey)>,
190}
191
192/// The sizes of a [`DashClient`]'s bookkeeping, see [`DashClient::state_size`].
193#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
194#[non_exhaustive]
195pub struct StateSize {
196    /// Period runs held (finished ones are dropped).
197    pub periods: usize,
198    /// Resources with a request on record.
199    pub requests: usize,
200    /// Requests awaiting an answer.
201    pub inflight: usize,
202    /// Failed resources awaiting the caller's retry.
203    pub parked: usize,
204    /// Resources with a consecutive-failure count.
205    pub failures: usize,
206    /// Delivered resources remembered for duplicate detection.
207    pub done: usize,
208    /// Initialization Segments held.
209    pub inits: usize,
210    /// Representations indexed.
211    pub reps: usize,
212    /// Fetches queued for the next poll.
213    pub queued: usize,
214    /// Outputs not yet drained.
215    pub outputs: usize,
216    /// Live-only bookkeeping entries (early waits, missing-segment flags).
217    pub live: usize,
218    /// Track identities holding an allocated `TrackId` range (current and previous Period only).
219    pub track_identities: usize,
220}
221
222/// The sans-IO DASH client for static and dynamic presentations.
223///
224/// Drive it like `hls_runtime::client::HlsClient`: call [`DashClient::poll`]
225/// for the next fetch, perform it, feed the result to the matching `on_*`
226/// call (or [`DashClient::on_failure`]), and drain
227/// [`DashClient::next_output`]. The core never opens a socket and never
228/// reads a clock; the `now_us` arguments are the caller's monotonic clock.
229/// A static presentation ignores it; a dynamic one maps it to UTC through an
230/// anchor (a `UTCTiming` sample the client takes through its own requests,
231/// [`DashClient::set_wall_clock`], or the MPD response `Date`) and uses that
232/// to join near the live edge, never request a segment before it exists,
233/// refresh the MPD and classify early, expired and missing segments. After
234/// each batch of calls ask [`DashClient::next_wake`] when to call
235/// [`DashClient::poll`] again if nothing else happens first.
236pub struct DashClient {
237    cfg: PullLimits,
238    dash: DashConfig,
239    mpd_url: String,
240    doc_url: Option<Url>,
241    mpd_query: Option<String>,
242    phase: Phase,
243    want_mpd: bool,
244    mpd_inflight: bool,
245    mpd: Option<Mpd>,
246    periods: Vec<PeriodRun>,
247    issue: usize,
248    deliver: usize,
249    next_rep: u32,
250    next_track_id: u32,
251    track_ids: BTreeMap<TrackIdentity, (u32, u32)>,
252    previous_identities: Vec<TrackIdentity>,
253    reqs: BTreeMap<ResourceId, ReqInfo>,
254    queue: VecDeque<ResourceId>,
255    inflight: BTreeSet<ResourceId>,
256    parked: BTreeSet<ResourceId>,
257    failures: BTreeMap<ResourceId, u32>,
258    done: BTreeSet<ResourceId>,
259    inits: BTreeMap<InitKey, InitEntry>,
260    rep_loc: BTreeMap<RepId, (usize, usize, usize)>,
261    outputs: VecDeque<Output>,
262    announced_tracks: Option<String>,
263    prev: Option<PrevPeriod>,
264    any_started: bool,
265    outstanding: usize,
266    notices_sent: Vec<Notice>,
267    clock: UtcClock,
268    mpd_failures: u32,
269    live: Option<live::Live>,
270    next_utc_id: u32,
271    last_was_discontinuity: bool,
272    remembered_done: VecDeque<ResourceId>,
273}
274
275fn without_query(url: &str) -> String {
276    match Url::parse(url) {
277        Ok(mut u) => {
278            u.set_query(None);
279            u.set_fragment(None);
280            u.to_string()
281        }
282        Err(_) => url.to_owned(),
283    }
284}
285
286fn same_origin(a: &Url, b: &Url) -> bool {
287    a.scheme() == b.scheme()
288        && a.host_str() == b.host_str()
289        && a.port_or_known_default() == b.port_or_known_default()
290}
291
292pub(crate) fn filter_candidates(
293    candidates: Vec<BaseCandidate>,
294    doc_url: Option<&Url>,
295    cross_origin_urls: bool,
296) -> Vec<BaseCandidate> {
297    if cross_origin_urls {
298        return candidates;
299    }
300    let Some(doc) = doc_url else {
301        return candidates;
302    };
303    candidates
304        .into_iter()
305        .filter(|c| {
306            c.resolve(Some(doc), "")
307                .and_then(|u| Url::parse(&u).ok())
308                .map(|p| same_origin(doc, &p))
309                .unwrap_or(false)
310        })
311        .collect()
312}
313
314fn address_name(e: &AddressError) -> &'static str {
315    match e {
316        AddressError::InvalidTemplate { .. } => "invalid_template",
317        AddressError::Unsupported(u) => u.name(),
318        AddressError::MissingMediaTemplate => "missing_media_template",
319        AddressError::MissingDuration => "missing_duration",
320        AddressError::TimelineNotIncreasing { .. } => "timeline_not_increasing",
321        AddressError::NumberNotIncreasing { .. } => "number_not_increasing",
322        AddressError::TimelineTooLong { .. } => "timeline_too_long",
323        AddressError::ListTimelineMismatch { .. } => "list_timeline_mismatch",
324        AddressError::Sidx(_) => "bad_sidx",
325        AddressError::Overflow => "overflow",
326        AddressError::IndexRequired => "index_required",
327        AddressError::EmptySegmentUrl => "empty_segment_url",
328        AddressError::ZeroSegmentDuration => "zero_segment_duration",
329    }
330}
331
332fn plan_error_name(e: &PlanError) -> &'static str {
333    match e {
334        PlanError::NoSuchRepresentation => "no_such_representation",
335        PlanError::Inherit(_) => "inherit",
336        PlanError::Address(a) => address_name(a),
337        PlanError::PeriodUnresolved => "period_unresolved",
338        PlanError::MissingAvailabilityStartTime => "missing_availability_start_time",
339    }
340}
341
342fn range_pair(r: Option<ByteRange>) -> std::result::Result<Option<(u64, u64)>, DropCause> {
343    r.map(|r| r.offset_len().map_err(|u| DropCause::Addressing(u.name())))
344        .transpose()
345}
346
347fn continuity_signals(set: &crate::mpd::model::AdaptationSet, previous: &str) -> bool {
348    let named = |d: &Descriptor| {
349        d.scheme_id_uri == SCHEME_PERIOD_CONTINUITY && d.value.as_deref() == Some(previous)
350    };
351    set.supplemental_properties.iter().any(named) || set.essential_properties.iter().any(named)
352}
353
354impl DashClient {
355    /// A client for the MPD at `mpd_url`. Performs no I/O.
356    #[must_use]
357    pub fn new(mpd_url: impl Into<String>, dash: DashConfig, limits: PullLimits) -> Self {
358        let mpd_url = mpd_url.into();
359        let doc_url = Url::parse(&mpd_url).ok();
360        let mpd_query = doc_url.as_ref().and_then(|u| u.query()).map(str::to_owned);
361        Self {
362            cfg: limits,
363            dash,
364            mpd_url,
365            doc_url,
366            mpd_query,
367            phase: Phase::NeedMpd,
368            want_mpd: true,
369            mpd_inflight: false,
370            mpd: None,
371            periods: Vec::new(),
372            issue: 0,
373            deliver: 0,
374            next_rep: 0,
375            next_track_id: 1,
376            track_ids: BTreeMap::new(),
377            previous_identities: Vec::new(),
378            reqs: BTreeMap::new(),
379            queue: VecDeque::new(),
380            inflight: BTreeSet::new(),
381            parked: BTreeSet::new(),
382            failures: BTreeMap::new(),
383            done: BTreeSet::new(),
384            inits: BTreeMap::new(),
385            rep_loc: BTreeMap::new(),
386            outputs: VecDeque::new(),
387            announced_tracks: None,
388            prev: None,
389            any_started: false,
390            outstanding: 0,
391            notices_sent: Vec::new(),
392            clock: UtcClock::new(),
393            mpd_failures: 0,
394            live: None,
395            next_utc_id: 0,
396            last_was_discontinuity: false,
397            remembered_done: VecDeque::new(),
398        }
399    }
400
401    /// The MPD URL given to [`DashClient::new`].
402    #[must_use]
403    pub fn mpd_url(&self) -> &str {
404        &self.mpd_url
405    }
406
407    /// Use `base` (the URL the MPD was finally served from, after redirects)
408    /// as the document URL every relative reference resolves against
409    /// (23009-1 A.11).
410    pub fn set_mpd_base(&mut self, base: impl Into<String>) {
411        self.doc_url = Url::parse(&base.into()).ok();
412    }
413
414    /// `Some(true)` for a static MPD, `Some(false)` for a dynamic one, `None`
415    /// before the MPD arrived.
416    #[must_use]
417    pub fn is_static(&self) -> Option<bool> {
418        self.mpd
419            .as_ref()
420            .map(|m| m.kind == PresentationType::Static)
421    }
422
423    /// Whether [`Output::EndOfStream`] was produced.
424    #[must_use]
425    pub fn presentation_ended(&self) -> bool {
426        self.phase == Phase::Ended
427    }
428
429    /// The `@maxSegmentDuration` of the MPD, whole seconds rounded up.
430    #[must_use]
431    pub fn max_segment_duration_secs(&self) -> Option<u64> {
432        let d = self.mpd.as_ref()?.max_segment_duration?;
433        Some(d.as_secs() + u64::from(d.subsec_nanos() > 0))
434    }
435
436    /// The sizes of the client's bookkeeping, for tests and monitoring: every
437    /// collection is bounded by the in-flight window and the Periods in play
438    /// (a live session of any length does not grow them).
439    #[must_use]
440    pub fn state_size(&self) -> StateSize {
441        StateSize {
442            periods: self.periods.len(),
443            requests: self.reqs.len(),
444            inflight: self.inflight.len(),
445            parked: self.parked.len(),
446            failures: self.failures.len(),
447            done: self.done.len(),
448            inits: self.inits.len(),
449            reps: self.rep_loc.len(),
450            queued: self.queue.len(),
451            outputs: self.outputs.len(),
452            live: self.live.as_ref().map_or(0, live::Live::bookkeeping),
453            track_identities: self.track_ids.len(),
454        }
455    }
456
457    /// Whether the client still waits for the answer to the fetch of `id`.
458    /// An MPD reset (or a rejoin) abandons the requests of the old timeline:
459    /// a caller may cancel those downloads; a late answer is rejected with
460    /// [`Error::UnrequestedResource`] either way.
461    #[must_use]
462    pub fn awaits(&self, id: ResourceId) -> bool {
463        self.inflight.contains(&id)
464    }
465
466    /// The next drained output, if any.
467    pub fn next_output(&mut self) -> Option<Output> {
468        self.outputs.pop_front()
469    }
470
471    fn stopped(&self) -> bool {
472        matches!(self.phase, Phase::Ended | Phase::Fatal)
473    }
474
475    fn ensure_live(&self) -> Result<()> {
476        if self.stopped() {
477            Err(Error::Stopped)
478        } else {
479            Ok(())
480        }
481    }
482
483    fn go_fatal(&mut self, kind: FatalKind) {
484        if self.phase == Phase::Fatal {
485            return;
486        }
487        self.phase = Phase::Fatal;
488        self.queue.clear();
489        self.live = None;
490        self.outputs.push_back(Output::Fatal(kind));
491    }
492
493    fn discontinuity(&mut self) {
494        if !self.last_was_discontinuity {
495            self.last_was_discontinuity = true;
496            self.outputs.push_back(Output::Discontinuity);
497        }
498    }
499
500    fn mark_done(&mut self, id: ResourceId) {
501        if self.done.insert(id) {
502            self.remembered_done.push_back(id);
503            while self.remembered_done.len() > MAX_REMEMBERED_DONE {
504                if let Some(old) = self.remembered_done.pop_front() {
505                    self.done.remove(&old);
506                }
507            }
508        }
509    }
510
511    fn notice_once(&mut self, n: Notice) {
512        if !self.notices_sent.contains(&n) {
513            self.notices_sent.push(n);
514            self.outputs.push_back(Output::Notice(n));
515        }
516    }
517
518    // MPD ------------------------------------------------------------------
519
520    /// Feed the MPD body of a response with no headers of interest (a static
521    /// presentation, or a caller with no `Date`, `ETag` or redirect to
522    /// report). See [`DashClient::on_mpd_with`].
523    ///
524    /// # Errors
525    /// As [`DashClient::on_mpd_with`].
526    pub fn on_mpd(&mut self, now_us: u64, bytes: &[u8]) -> Result<()> {
527        self.on_mpd_with(now_us, bytes, &ResponseMeta::at(now_us))
528    }
529
530    /// Feed the MPD body with what the caller measured and read: the first
531    /// MPD starts the presentation, a later one is a refresh of a dynamic
532    /// presentation.
533    ///
534    /// # Errors
535    /// [`Error::Mpd`] for a body that does not parse (request it again with
536    /// [`DashClient::retry_mpd`]; the MPD already held stays in force);
537    /// [`Error::MpdNotAwaited`] when no MPD is awaited; [`Error::Stopped`].
538    pub fn on_mpd_with(
539        &mut self,
540        now_us: u64,
541        bytes: &[u8],
542        meta: &ResponseMeta<'_>,
543    ) -> Result<()> {
544        self.ensure_live()?;
545        if !self.mpd_inflight {
546            return Err(Error::MpdNotAwaited);
547        }
548        self.mpd_inflight = false;
549        let mpd = match Mpd::parse_with(bytes, &self.cfg.limits) {
550            Ok(m) => m,
551            Err(e) => {
552                self.mpd_failed(now_us);
553                return Err(Error::Mpd(e));
554            }
555        };
556        if self.mpd.is_some() {
557            self.refresh_mpd(now_us, mpd, bytes, meta);
558            return Ok(());
559        }
560        if mpd.periods.is_empty() {
561            self.go_fatal(FatalKind::MpdInvalid {
562                reason: "no_period",
563            });
564            return Ok(());
565        }
566        let dynamic = mpd.kind == PresentationType::Dynamic;
567        if dynamic && mpd.availability_start_time_us.is_none() {
568            self.go_fatal(FatalKind::MpdInvalid {
569                reason: "missing_availability_start_time",
570            });
571            return Ok(());
572        }
573        let features = mpd.features;
574        self.mpd = Some(mpd);
575        self.phase = Phase::Running;
576        if features.event_stream || features.inband_event_stream {
577            self.notice_once(Notice::EventStreamIgnored);
578        }
579        if features.preselection {
580            self.notice_once(Notice::PreselectionIgnored);
581        }
582        if features.patch_location {
583            self.notice_once(Notice::PatchIgnored);
584        }
585        if dynamic && features.low_latency {
586            self.notice_once(Notice::LowLatencyIgnored);
587        }
588        if dynamic {
589            self.start_live(now_us, bytes, meta);
590        } else {
591            if let Some(u) = meta.final_url.and_then(|f| self.accept_final_url(f)) {
592                self.doc_url = Some(u);
593            }
594            self.activate(0);
595            self.issue = 0;
596        }
597        Ok(())
598    }
599
600    /// The MPD fetch failed; nothing is counted against the presentation, an
601    /// MPD is never given up on. The caller backs off and calls
602    /// [`DashClient::retry_mpd`]. The returned class carries the count of
603    /// consecutive MPD failures.
604    pub fn on_mpd_failure(&mut self, now_us: u64, failure: Failure) -> FailureClass {
605        self.mpd_inflight = false;
606        self.want_mpd = false;
607        let attempt = self.mpd_failed(now_us);
608        let kind = match failure {
609            Failure::Status(STATUS_NOT_FOUND | STATUS_GONE) => FailureKind::Missing,
610            Failure::Status(STATUS_TOO_EARLY) => FailureKind::TooEarly,
611            Failure::Status(_) => FailureKind::Server,
612            Failure::Transport | Failure::Timeout => FailureKind::Transport,
613            Failure::TooLarge => FailureKind::Invalid,
614        };
615        FailureClass::Retry { attempt, kind }
616    }
617
618    /// Ask for the MPD again after a rejected body or a failed fetch.
619    pub fn retry_mpd(&mut self, _now_us: u64) {
620        if self.mpd_inflight || self.stopped() {
621            return;
622        }
623        let waiting = self.phase == Phase::NeedMpd
624            || self.live.as_ref().is_some_and(|l| l.refresh.awaiting_retry);
625        if waiting {
626            self.want_mpd = true;
627        }
628    }
629
630    // Poll -----------------------------------------------------------------
631
632    /// The next fetch the caller must perform, if any. Call it until it
633    /// returns `None`, then wait for a response or for
634    /// [`DashClient::next_wake`], whichever is first.
635    pub fn poll(&mut self, now_us: u64) -> Option<Action> {
636        if self.stopped() {
637            return None;
638        }
639        self.tick(now_us);
640        if self.stopped() {
641            return None;
642        }
643        if let Some(a) = self.mpd_action(now_us) {
644            return Some(a);
645        }
646        if self.phase != Phase::Running {
647            return None;
648        }
649        if let Some(a) = self.utc_action(now_us) {
650            return Some(a);
651        }
652        loop {
653            if self.queue.is_empty() {
654                self.fill_queue(now_us);
655                if self.queue.is_empty() {
656                    self.drain();
657                }
658            }
659            let id = self.queue.pop_front()?;
660            if let Some(action) = self.emit(id) {
661                return Some(action);
662            }
663            if self.stopped() {
664                return None;
665            }
666        }
667    }
668
669    fn window_open(&self) -> bool {
670        self.outstanding < self.cfg.window()
671    }
672
673    fn fill_queue(&mut self, now_us: u64) {
674        self.reset_wake();
675        loop {
676            let Some(pi) = (self.issue < self.periods.len()).then_some(self.issue) else {
677                return;
678            };
679            if self.live_blocked() {
680                return;
681            }
682            let exhausted = self.periods[pi].lanes.iter().all(Lane::exhausted);
683            if self.periods[pi].issue_done || (exhausted && !self.period_open(pi, now_us)) {
684                self.periods[pi].issue_done = true;
685                let next = self.periods[pi].index + 1;
686                if self.can_activate(next) {
687                    self.activate(next);
688                    self.issue = self.periods.len() - 1;
689                    continue;
690                }
691                return;
692            }
693            if exhausted {
694                self.note_exhausted(now_us);
695                return;
696            }
697            if !self.window_open() {
698                return;
699            }
700            let utc = self.utc_now(now_us);
701            let mut pick: Option<(usize, i128)> = None;
702            for li in 0..self.periods[pi].lanes.len() {
703                match self.lane_candidate(pi, li, utc, now_us) {
704                    Candidate::Ready(start_us) => {
705                        if pick.is_none_or(|(_, best)| start_us < best) {
706                            pick = Some((li, start_us));
707                        }
708                    }
709                    Candidate::Wait(at_utc_us) => self.note_wake(at_utc_us),
710                    Candidate::Idle => {}
711                }
712            }
713            let Some((li, _)) = pick else {
714                return;
715            };
716            if !self.issue_segment(pi, li) {
717                return;
718            }
719        }
720    }
721
722    fn lane_candidate(
723        &mut self,
724        pi: usize,
725        li: usize,
726        utc: Option<i128>,
727        now_us: u64,
728    ) -> Candidate {
729        match utc {
730            Some(utc) if self.is_live_mpd() => self.live_lane_candidate(pi, li, utc, now_us),
731            _ => {
732                let lane = &self.periods[pi].lanes[li];
733                let Some(rep) = lane.current() else {
734                    return Candidate::Idle;
735                };
736                let Some(plan) = rep.plan.as_ref() else {
737                    return Candidate::Idle;
738                };
739                plan.get(rep.next_issue)
740                    .map_or(Candidate::Idle, |s| Candidate::Ready(s.start_us))
741            }
742        }
743    }
744
745    fn issue_segment(&mut self, pi: usize, li: usize) -> bool {
746        let lane = &mut self.periods[pi].lanes[li];
747        let ri = lane.reps.len() - 1;
748        let rep = &mut lane.reps[ri];
749        let Some(seg) = rep.plan.as_ref().and_then(|p| p.get(rep.next_issue)) else {
750            return false;
751        };
752        let range = match range_pair(seg.byte_range) {
753            Ok(r) => r,
754            Err(cause) => {
755                let reason = match cause {
756                    DropCause::Addressing(r) => r,
757                    _ => "unusable_url",
758                };
759                self.go_fatal(FatalKind::UnsupportedAddressing { reason });
760                return false;
761            }
762        };
763        let id = ResourceId::Segment {
764            rep: rep.id,
765            number: seg.number,
766        };
767        let plan_index = rep.next_issue;
768        rep.next_issue += 1;
769        rep.last_time = Some(seg.time_ticks);
770        let rep_id = rep.id;
771        lane.queue.push_back(Slot {
772            id: Some(id),
773            rep: ri,
774            plan_index,
775            state: SlotState::Issued,
776        });
777        self.outstanding += 1;
778        self.reqs.insert(
779            id,
780            ReqInfo {
781                rep: rep_id,
782                reference: seg.reference,
783                range,
784                kind: ReqKind::Segment,
785                avail_from_us: seg.available_from_us,
786                avail_until_us: seg.available_until_us,
787            },
788        );
789        self.queue.push_back(id);
790        true
791    }
792
793    fn rep(&self, id: RepId) -> Option<&RepState> {
794        let (p, l, r) = *self.rep_loc.get(&id)?;
795        self.periods.get(p)?.lanes.get(l)?.reps.get(r)
796    }
797
798    fn rep_mut(&mut self, id: RepId) -> Option<&mut RepState> {
799        let (p, l, r) = *self.rep_loc.get(&id)?;
800        self.periods.get_mut(p)?.lanes.get_mut(l)?.reps.get_mut(r)
801    }
802
803    fn resolve(
804        &self,
805        rep: &RepState,
806        reference: &str,
807        media: bool,
808    ) -> std::result::Result<String, DropCause> {
809        let cand = rep
810            .candidates
811            .get(rep.cand % rep.candidates.len().max(1))
812            .ok_or(DropCause::UnusableUrl)?;
813        let mut url = cand
814            .resolve(self.doc_url.as_ref(), reference)
815            .ok_or(DropCause::UnusableUrl)?;
816        if media && let Some(q) = &rep.query {
817            url = append_query(&url, q);
818        }
819        let parsed = Url::parse(&url).map_err(|_| DropCause::UnusableUrl)?;
820        if parsed.scheme() == "data" {
821            return Err(DropCause::Unsupported(Unsupported::DataUrl));
822        }
823        if !matches!(parsed.scheme(), "http" | "https") {
824            return Err(DropCause::UnusableUrl);
825        }
826        if !self.dash.cross_origin_urls
827            && let Some(doc) = &self.doc_url
828            && !same_origin(doc, &parsed)
829        {
830            return Err(DropCause::CrossOrigin);
831        }
832        Ok(url)
833    }
834
835    pub(super) fn accept_final_url(&self, final_url: &str) -> Option<Url> {
836        let parsed = Url::parse(final_url).ok()?;
837        if !matches!(parsed.scheme(), "http" | "https") {
838            return None;
839        }
840        let pinned_elsewhere = !self.dash.cross_origin_urls
841            && self
842                .doc_url
843                .as_ref()
844                .is_some_and(|d| !same_origin(d, &parsed));
845        (!pinned_elsewhere).then_some(parsed)
846    }
847
848    fn emit(&mut self, id: ResourceId) -> Option<Action> {
849        let (rep_id, reference, range, kind) = {
850            let info = self.reqs.get(&id)?;
851            (info.rep, info.reference.clone(), info.range, info.kind)
852        };
853        let url = {
854            let rep = self.rep(rep_id)?;
855            match self.resolve(rep, &reference, kind == ReqKind::Segment) {
856                Ok(u) => u,
857                Err(cause) => {
858                    let reason = match cause {
859                        DropCause::CrossOrigin => "cross_origin",
860                        DropCause::Unsupported(u) => u.name(),
861                        _ => "unusable_url",
862                    };
863                    self.go_fatal(FatalKind::UnsupportedAddressing { reason });
864                    return None;
865                }
866            }
867        };
868        self.parked.remove(&id);
869        self.inflight.insert(id);
870        match kind {
871            ReqKind::Init => {
872                let key = self.rep(rep_id)?.init_key.clone();
873                if let Some(e) = self.inits.get_mut(&key) {
874                    e.state = InitState::InFlight;
875                }
876                Some(Action::FetchInit {
877                    id,
878                    url,
879                    byte_range: range,
880                })
881            }
882            ReqKind::Index => Some(Action::FetchIndex {
883                id,
884                url,
885                byte_range: range,
886            }),
887            ReqKind::Segment => Some(Action::FetchSegment {
888                id,
889                url,
890                byte_range: range,
891            }),
892        }
893    }
894
895    // Period activation ----------------------------------------------------
896
897    fn activate(&mut self, index: usize) {
898        let Some(mpd) = self.mpd.as_ref() else { return };
899        let timing = period_timings(mpd).get(index).copied();
900        let start_us = timing.and_then(|t| t.start_us).unwrap_or(0);
901        let selection = select(mpd, index, &self.cfg.policy);
902        let period_id = mpd.periods.get(index).and_then(|p| p.id.clone());
903        let mut notices = Vec::new();
904        let mut encryption = false;
905        for d in &selection.dropped {
906            if d.reason == crate::selection::DropReason::UnsupportedEncryption {
907                encryption = true;
908            }
909            notices.push(Output::TrackDropped {
910                period: index,
911                set: d.set,
912                rep: d.rep,
913                cause: DropCause::Selection(d.reason),
914            });
915        }
916        if encryption {
917            notices.push(Output::Notice(Notice::UnsupportedEncryption));
918        }
919        if selection
920            .dropped
921            .iter()
922            .any(|d| d.reason == crate::selection::DropReason::UnresolvedXlink)
923        {
924            notices.push(Output::Notice(Notice::XlinkSkipped));
925        }
926        let video = selection
927            .chosen_variant
928            .and_then(|i| selection.variants.get(i).copied());
929        let audio = selection.chosen_audio;
930        let mut run = PeriodRun {
931            index,
932            id: period_id,
933            start_us,
934            lanes: Vec::new(),
935            selection,
936            notices,
937            started: false,
938            issue_done: false,
939            finished: false,
940        };
941        let pi = self.periods.len();
942        for (kind, at) in [(SetClass::Video, video), (SetClass::Audio, audio)] {
943            let Some(at) = at else { continue };
944            match self.build_rep(at) {
945                Ok(rep) => {
946                    let resolved = at.resolve(self.mpd.as_ref().expect("mpd present"));
947                    let set_id = resolved.and_then(|(_, s, _)| s.id);
948                    let identity = resolved.map_or_else(
949                        || TrackIdentity {
950                            kind,
951                            set_id: None,
952                            group: None,
953                            lang: None,
954                            roles: Vec::new(),
955                        },
956                        |(_, s, _)| TrackIdentity {
957                            kind,
958                            set_id: s.id,
959                            group: s.group,
960                            lang: s.lang.clone(),
961                            roles: s
962                                .roles
963                                .iter()
964                                .map(|d| (d.scheme_id_uri.clone(), d.value.clone()))
965                                .collect(),
966                        },
967                    );
968                    run.lanes.push(Lane {
969                        kind,
970                        set_id,
971                        identity,
972                        reps: vec![rep],
973                        queue: VecDeque::new(),
974                        base: 0,
975                        count: 0,
976                        announced_rep: None,
977                        announced_init: None,
978                        expired_run: 0,
979                    });
980                }
981                Err(cause) => run.notices.push(Output::TrackDropped {
982                    period: index,
983                    set: at.set,
984                    rep: Some(at.rep),
985                    cause,
986                }),
987            }
988        }
989        self.periods.push(run);
990        for li in 0..self.periods[pi].lanes.len() {
991            self.register_rep(pi, li, 0);
992        }
993    }
994
995    fn init_reference(
996        mpd: &Mpd,
997        at: RepRef,
998        planned: &std::result::Result<crate::address::PlannedRepresentation, PlanError>,
999    ) -> Option<(String, Option<ByteRange>)> {
1000        match planned {
1001            Ok(p) => p.plan.init().map(|r| (r.reference, r.byte_range)),
1002            Err(PlanError::Address(AddressError::IndexRequired)) => {
1003                let (period, set, rep) = at.resolve(mpd)?;
1004                match resolve_segment_info(period, set, rep).ok()? {
1005                    SegmentInfo::Base(b) => b
1006                        .initialization
1007                        .as_ref()
1008                        .map(|u| (u.source_url.clone().unwrap_or_default(), u.range)),
1009                    SegmentInfo::Template(_) | SegmentInfo::List(_) => None,
1010                }
1011            }
1012            Err(_) => None,
1013        }
1014    }
1015
1016    fn build_rep(&mut self, at: RepRef) -> std::result::Result<RepState, DropCause> {
1017        let mpd = self.mpd.as_ref().expect("mpd present");
1018        let (period, set, rep) = at
1019            .resolve(mpd)
1020            .ok_or(DropCause::Addressing("no_such_representation"))?;
1021        let candidates = filter_candidates(
1022            representation_candidates(mpd, period, set, rep),
1023            self.doc_url.as_ref(),
1024            self.dash.cross_origin_urls,
1025        );
1026        if candidates.is_empty() && !self.dash.cross_origin_urls {
1027            return Err(DropCause::CrossOrigin);
1028        }
1029        let query = segment_url_query(mpd, period, set, rep, self.mpd_query.as_deref())
1030            .map_err(|_| DropCause::UnusableUrl)?;
1031        let planned = plan_representation(mpd, at, &self.plan_options());
1032        let init = Self::init_reference(mpd, at, &planned).ok_or(DropCause::NoInitialization)?;
1033        let (plan, index) = match planned {
1034            Ok(p) => {
1035                if p.plan.len().is_none() && mpd.kind != PresentationType::Dynamic {
1036                    return Err(DropCause::UnboundedPlan);
1037                }
1038                (Some(p.plan), None)
1039            }
1040            Err(PlanError::Address(AddressError::IndexRequired)) => {
1041                let info = resolve_segment_info(period, set, rep)
1042                    .map_err(|_| DropCause::Addressing("inherit"))?;
1043                let range = match info {
1044                    SegmentInfo::Base(b) => b.index_range,
1045                    SegmentInfo::Template(_) | SegmentInfo::List(_) => None,
1046                };
1047                let range = range.ok_or(DropCause::Addressing("index_required"))?;
1048                (None, Some(IndexReq { range }))
1049            }
1050            Err(e) => return Err(DropCause::Addressing(plan_error_name(&e))),
1051        };
1052        let init_range = range_pair(init.1)?;
1053        let id = RepId(self.next_rep);
1054        self.next_rep = self.next_rep.saturating_add(1);
1055        let mut state = RepState {
1056            id,
1057            at,
1058            candidates,
1059            cand: 0,
1060            cand_failures: 0,
1061            query,
1062            plan,
1063            index,
1064            init_key: InitKey {
1065                url: String::new(),
1066                range: init_range,
1067            },
1068            next_issue: 0,
1069            start_time_us: None,
1070            dead: false,
1071            set_id: set.id,
1072            set_pos: at.set,
1073            rep_key: rep.id.clone(),
1074            joined: mpd.kind != PresentationType::Dynamic,
1075            last_time: None,
1076        };
1077        let init_url = self.resolve(&state, &init.0, false)?;
1078        state.init_key.url = without_query(&init_url);
1079        if let Some(ix) = &state.index {
1080            range_pair(Some(ix.range))?;
1081            self.resolve(&state, "", false)?;
1082        }
1083        if let Some(seg) = state.plan.as_ref().and_then(|p| p.get(0)) {
1084            range_pair(seg.byte_range)?;
1085            self.resolve(&state, &seg.reference, true)?;
1086        }
1087        Ok(state)
1088    }
1089
1090    fn register_rep(&mut self, pi: usize, li: usize, ri: usize) {
1091        let rep = &self.periods[pi].lanes[li].reps[ri];
1092        let rep_id = rep.id;
1093        let key = rep.init_key.clone();
1094        let has_index = rep.index.is_some();
1095        let index_range = rep.index.as_ref().map(|i| i.range);
1096        let init_reference = {
1097            let mpd = self.mpd.as_ref().expect("mpd present");
1098            let planned = rep.plan.as_ref().map(|p| p.init());
1099            match planned {
1100                Some(init) => init.map(|r| (r.reference, r.byte_range)),
1101                None => Self::init_reference(
1102                    mpd,
1103                    rep.at,
1104                    &Err(PlanError::Address(AddressError::IndexRequired)),
1105                ),
1106            }
1107        };
1108        self.rep_loc.insert(rep_id, (pi, li, ri));
1109        if let std::collections::btree_map::Entry::Vacant(v) = self.inits.entry(key) {
1110            v.insert(InitEntry {
1111                state: InitState::Wanted,
1112                bytes: None,
1113                specs: Vec::new(),
1114                skipped_reported: BTreeSet::new(),
1115            });
1116            let (reference, range) = init_reference.unwrap_or_default();
1117            let init_id = ResourceId::Init { rep: rep_id };
1118            self.reqs.insert(
1119                init_id,
1120                ReqInfo {
1121                    rep: rep_id,
1122                    reference,
1123                    range: range.and_then(|r| r.offset_len().ok()),
1124                    kind: ReqKind::Init,
1125                    avail_from_us: None,
1126                    avail_until_us: None,
1127                },
1128            );
1129            self.queue.push_back(init_id);
1130        }
1131        if has_index && let Some(r) = index_range {
1132            let index_id = ResourceId::Index { rep: rep_id };
1133            self.reqs.insert(
1134                index_id,
1135                ReqInfo {
1136                    rep: rep_id,
1137                    reference: String::new(),
1138                    range: r.offset_len().ok(),
1139                    kind: ReqKind::Index,
1140                    avail_from_us: None,
1141                    avail_until_us: None,
1142                },
1143            );
1144            self.queue.push_back(index_id);
1145        }
1146    }
1147
1148    // Deliveries -----------------------------------------------------------
1149
1150    fn requested(&self, id: ResourceId) -> Result<()> {
1151        if self.done.contains(&id) {
1152            return Err(Error::DuplicateResource { id });
1153        }
1154        if !self.inflight.contains(&id) {
1155            return Err(Error::UnrequestedResource { id });
1156        }
1157        Ok(())
1158    }
1159
1160    fn body_rejection(&self, id: ResourceId, bytes: &[u8]) -> Option<Rejection> {
1161        if bytes.len() > self.cfg.max_segment_bytes {
1162            return Some(Rejection::TooLarge);
1163        }
1164        let range = self.reqs.get(&id).and_then(|r| r.range)?;
1165        (u64::try_from(bytes.len()).ok() != Some(range.1)).then_some(Rejection::WrongLength)
1166    }
1167
1168    fn max_failures(&self, id: ResourceId) -> u32 {
1169        match id {
1170            ResourceId::Segment { .. } => self.cfg.max_segment_failures,
1171            ResourceId::Init { .. } | ResourceId::Index { .. } => self.cfg.max_init_failures,
1172        }
1173    }
1174
1175    fn count_failure(
1176        &mut self,
1177        now_us: u64,
1178        id: ResourceId,
1179        kind: FailureKind,
1180        status: Option<u16>,
1181    ) -> FailureClass {
1182        self.inflight.remove(&id);
1183        let mut rotated = false;
1184        let count = {
1185            let n = self.failures.entry(id).or_insert(0);
1186            *n = n.saturating_add(1);
1187            *n
1188        };
1189        if count >= self.max_failures(id) {
1190            if self.is_live_mpd() && matches!(id, ResourceId::Segment { .. }) {
1191                return self.give_up_live_segment(id, status);
1192            }
1193            let fatal = match id {
1194                ResourceId::Segment { .. } => FatalKind::SegmentFetchFailed { id, status },
1195                ResourceId::Init { .. } | ResourceId::Index { .. } => {
1196                    FatalKind::InitFetchFailed { id, status }
1197                }
1198            };
1199            self.go_fatal(fatal);
1200            return FailureClass::GiveUp;
1201        }
1202        let doc = self.doc_url.clone();
1203        let cross = self.dash.cross_origin_urls;
1204        if let Some(rep) = self.rep_mut(id.rep()) {
1205            rep.cand_failures += 1;
1206            if rep.cand_failures >= ROTATE_AFTER_FAILURES && rep.candidates.len() > 1 {
1207                let len = rep.candidates.len();
1208                let mut next = (rep.cand + 1) % len;
1209                let mut found = false;
1210                for _ in 0..len {
1211                    let ok = cross
1212                        || doc.as_ref().is_none_or(|d| {
1213                            rep.candidates[next]
1214                                .resolve(Some(d), "")
1215                                .and_then(|u| Url::parse(&u).ok())
1216                                .map(|p| same_origin(d, &p))
1217                                .unwrap_or(false)
1218                        });
1219                    if ok {
1220                        found = true;
1221                        break;
1222                    }
1223                    next = (next + 1) % len;
1224                }
1225                if found && next != rep.cand {
1226                    rep.cand = next;
1227                    rep.cand_failures = 0;
1228                    rotated = true;
1229                }
1230            }
1231        }
1232        if rotated {
1233            self.after_rotation(now_us);
1234        }
1235        self.parked.insert(id);
1236        FailureClass::Retry {
1237            attempt: count,
1238            kind,
1239        }
1240    }
1241
1242    fn reject(&mut self, now_us: u64, id: ResourceId, reason: Rejection) -> Error {
1243        let class = self.count_failure(now_us, id, FailureKind::Invalid, None);
1244        Error::Rejected { id, reason, class }
1245    }
1246
1247    fn clear_success(&mut self, id: ResourceId) {
1248        self.inflight.remove(&id);
1249        self.parked.remove(&id);
1250        self.failures.remove(&id);
1251        self.mark_done(id);
1252    }
1253
1254    /// Feed an Initialization Segment.
1255    ///
1256    /// # Errors
1257    /// [`Error::Rejected`] for an unusable body (counted; see
1258    /// [`FailureClass`]), also for a segment held for this init that the
1259    /// demuxer then refused; [`Error::UnrequestedResource`],
1260    /// [`Error::DuplicateResource`], [`Error::Stopped`].
1261    pub fn on_init(&mut self, now_us: u64, id: ResourceId, bytes: &[u8]) -> Result<()> {
1262        self.ensure_live()?;
1263        let ResourceId::Init { rep } = id else {
1264            return Err(Error::UnrequestedResource { id });
1265        };
1266        self.requested(id)?;
1267        if let Some(r) = self.body_rejection(id, bytes) {
1268            return Err(self.reject(now_us, id, r));
1269        }
1270        if !segment::progressive_movie(bytes) {
1271            return Err(self.reject(now_us, id, Rejection::InitUndecodable));
1272        }
1273        let info = match demux::init_info(bytes) {
1274            Ok(i) => i,
1275            Err(_) => return Err(self.reject(now_us, id, Rejection::DemuxFailed)),
1276        };
1277        let key = self
1278            .rep(rep)
1279            .ok_or(Error::UnrequestedResource { id })?
1280            .init_key
1281            .clone();
1282        if let Some(e) = self.inits.get_mut(&key) {
1283            e.state = InitState::Ready;
1284            e.bytes = Some(Arc::new(bytes.to_vec()));
1285            e.specs = info.specs;
1286            for s in info.skipped {
1287                let k = (s.track_id.map(TrackId::value), s.fourcc.clone());
1288                if e.skipped_reported.insert(k) {
1289                    self.outputs.push_back(Output::TrackSkipped {
1290                        track_id: None,
1291                        fourcc: s.fourcc,
1292                        reason: s.reason,
1293                    });
1294                }
1295            }
1296        }
1297        self.clear_success(id);
1298        self.reqs.remove(&id);
1299        let held = self.demux_held(now_us, &key);
1300        self.drain();
1301        held
1302    }
1303
1304    fn demux_held(&mut self, now_us: u64, key: &InitKey) -> Result<()> {
1305        let Some(init) = self.inits.get(key).and_then(|e| e.bytes.clone()) else {
1306            return Ok(());
1307        };
1308        let mut first_error = None;
1309        let from = self.deliver.min(self.periods.len());
1310        for pi in from..self.periods.len() {
1311            for li in 0..self.periods[pi].lanes.len() {
1312                let n = self.periods[pi].lanes[li].queue.len();
1313                for qi in 0..n {
1314                    let (id, matches) = {
1315                        let lane = &self.periods[pi].lanes[li];
1316                        let slot = &lane.queue[qi];
1317                        let m = matches!(slot.state, SlotState::Held(_))
1318                            && lane.reps.get(slot.rep).is_some_and(|r| &r.init_key == key);
1319                        (slot.id, m)
1320                    };
1321                    if !matches {
1322                        continue;
1323                    }
1324                    let slot = &mut self.periods[pi].lanes[li].queue[qi];
1325                    let SlotState::Held(bytes) =
1326                        std::mem::replace(&mut slot.state, SlotState::Issued)
1327                    else {
1328                        continue;
1329                    };
1330                    match demux::demux_segment(&init, &bytes) {
1331                        Ok(staged) => {
1332                            self.periods[pi].lanes[li].queue[qi].state = SlotState::Staged(staged);
1333                        }
1334                        Err(_) => {
1335                            if let Some(id) = id {
1336                                self.inflight.insert(id);
1337                                let e = self.reject(now_us, id, Rejection::DemuxFailed);
1338                                first_error.get_or_insert(e);
1339                            }
1340                        }
1341                    }
1342                }
1343            }
1344        }
1345        match first_error {
1346            Some(e) => Err(e),
1347            None => Ok(()),
1348        }
1349    }
1350
1351    /// Feed the `sidx` of a `SegmentBase` Representation.
1352    ///
1353    /// # Errors
1354    /// As [`DashClient::on_init`].
1355    pub fn on_index(&mut self, now_us: u64, id: ResourceId, bytes: &[u8]) -> Result<()> {
1356        self.ensure_live()?;
1357        let ResourceId::Index { rep } = id else {
1358            return Err(Error::UnrequestedResource { id });
1359        };
1360        self.requested(id)?;
1361        if let Some(r) = self.body_rejection(id, bytes) {
1362            return Err(self.reject(now_us, id, r));
1363        }
1364        let sidx = match base_index::parse_index(bytes) {
1365            Ok(s) => s,
1366            Err(_) => return Err(self.reject(now_us, id, Rejection::BadIndex)),
1367        };
1368        let (at, range) = {
1369            let r = self.rep(rep).ok_or(Error::UnrequestedResource { id })?;
1370            let range = r.index.as_ref().map(|i| i.range);
1371            (r.at, range)
1372        };
1373        let Some(range) = range else {
1374            return Err(Error::UnrequestedResource { id });
1375        };
1376        let anchor = match base_index::anchor_after(&range) {
1377            Ok(a) => a,
1378            Err(_) => return Err(self.reject(now_us, id, Rejection::BadIndex)),
1379        };
1380        let planned = {
1381            let mpd = self.mpd.as_ref().expect("mpd present");
1382            plan_representation(mpd, at, &self.plan_options().with_index(&sidx, anchor))
1383        };
1384        self.clear_success(id);
1385        self.reqs.remove(&id);
1386        match planned {
1387            Ok(p) => {
1388                if let Some(r) = self.rep_mut(rep) {
1389                    let start = r.start_time_us.take();
1390                    r.next_issue = start
1391                        .and_then(|t| p.plan.index_at_or_before(t))
1392                        .unwrap_or(0);
1393                    r.plan = Some(p.plan);
1394                }
1395            }
1396            Err(e) => {
1397                if let Some(r) = self.rep_mut(rep) {
1398                    r.dead = true;
1399                }
1400                if let Some((pi, _, _)) = self.rep_loc.get(&rep).copied() {
1401                    self.outputs.push_back(Output::TrackDropped {
1402                        period: self.periods[pi].index,
1403                        set: at.set,
1404                        rep: Some(at.rep),
1405                        cause: DropCause::Addressing(plan_error_name(&e)),
1406                    });
1407                }
1408            }
1409        }
1410        self.drain();
1411        Ok(())
1412    }
1413
1414    /// Feed a Media Segment.
1415    ///
1416    /// # Errors
1417    /// As [`DashClient::on_init`]; a progressive MP4 additionally ends the
1418    /// presentation.
1419    pub fn on_segment(&mut self, now_us: u64, id: ResourceId, bytes: &[u8]) -> Result<()> {
1420        self.ensure_live()?;
1421        let ResourceId::Segment { rep, .. } = id else {
1422            return Err(Error::UnrequestedResource { id });
1423        };
1424        self.requested(id)?;
1425        if let Some(r) = self.body_rejection(id, bytes) {
1426            return Err(self.reject(now_us, id, r));
1427        }
1428        if segment::progressive_movie(bytes) {
1429            self.inflight.remove(&id);
1430            self.go_fatal(FatalKind::NonFragmentedMp4 { id });
1431            return Err(Error::Rejected {
1432                id,
1433                reason: Rejection::NonFragmentedMp4,
1434                class: FailureClass::GiveUp,
1435            });
1436        }
1437        if !segment::media_fragment(bytes) {
1438            return Err(self.reject(now_us, id, Rejection::NotMedia));
1439        }
1440        let key = self
1441            .rep(rep)
1442            .ok_or(Error::UnrequestedResource { id })?
1443            .init_key
1444            .clone();
1445        let init = self.inits.get(&key).and_then(|e| e.bytes.clone());
1446        let state = match init {
1447            Some(init) => match demux::demux_segment(&init, bytes) {
1448                Ok(staged) => SlotState::Staged(staged),
1449                Err(_) => return Err(self.reject(now_us, id, Rejection::DemuxFailed)),
1450            },
1451            None => SlotState::Held(bytes.to_vec()),
1452        };
1453        let Some(slot) = self.slot_mut(id) else {
1454            return Err(Error::UnrequestedResource { id });
1455        };
1456        slot.state = state;
1457        self.inflight.remove(&id);
1458        self.parked.remove(&id);
1459        self.failures.remove(&id);
1460        self.mark_done(id);
1461        self.segment_delivered(id);
1462        self.drain();
1463        Ok(())
1464    }
1465
1466    fn slot_mut(&mut self, id: ResourceId) -> Option<&mut Slot> {
1467        let (p, l, _) = *self.rep_loc.get(&id.rep())?;
1468        self.periods
1469            .get_mut(p)?
1470            .lanes
1471            .get_mut(l)?
1472            .queue
1473            .iter_mut()
1474            .find(|s| s.id == Some(id))
1475    }
1476
1477    /// Report a failed fetch.
1478    ///
1479    /// # Errors
1480    /// [`Error::UnrequestedResource`] when `id` is not outstanding;
1481    /// [`Error::Stopped`].
1482    pub fn on_failure(
1483        &mut self,
1484        now_us: u64,
1485        id: ResourceId,
1486        failure: Failure,
1487    ) -> Result<FailureClass> {
1488        self.ensure_live()?;
1489        if !self.inflight.contains(&id) {
1490            return Err(Error::UnrequestedResource { id });
1491        }
1492        if self.is_live_mpd() && matches!(id, ResourceId::Segment { .. }) {
1493            return self.live_segment_failure(now_us, id, failure);
1494        }
1495        let status = match failure {
1496            Failure::Status(s) => Some(s),
1497            Failure::Transport | Failure::Timeout | Failure::TooLarge => None,
1498        };
1499        if matches!(status, Some(STATUS_NOT_FOUND | STATUS_GONE)) && self.is_missing_tail(id) {
1500            self.inflight.remove(&id);
1501            self.failures.remove(&id);
1502            if let Some(live) = self.live.as_mut() {
1503                live.early.remove(&id);
1504            }
1505            self.mark_done(id);
1506            if let Some(slot) = self.slot_mut(id) {
1507                slot.state = SlotState::Skipped { marker: false };
1508            }
1509            self.drain();
1510            return Ok(FailureClass::EndTail);
1511        }
1512        Ok(self.count_failure(now_us, id, live::failure_kind(failure), status))
1513    }
1514
1515    fn is_missing_tail(&self, id: ResourceId) -> bool {
1516        let ResourceId::Segment { rep, .. } = id else {
1517            return false;
1518        };
1519        let Some((pi, li, _)) = self.rep_loc.get(&rep).copied() else {
1520            return false;
1521        };
1522        let lane = &self.periods[pi].lanes[li];
1523        let Some(slot) = lane.queue.iter().find(|s| s.id == Some(id)) else {
1524            return false;
1525        };
1526        let Some(r) = lane.reps.get(slot.rep) else {
1527            return false;
1528        };
1529        let Some(plan) = r.plan.as_ref() else {
1530            return false;
1531        };
1532        let last = plan.len().and_then(|n| n.checked_sub(1));
1533        last == Some(slot.plan_index)
1534            && plan.len().is_some_and(|n| n > 1)
1535            && plan
1536                .get(slot.plan_index)
1537                .is_some_and(|s| s.duration_us < MISSING_TAIL_MAX_US)
1538    }
1539
1540    /// Request again a resource that [`DashClient::on_failure`] or a
1541    /// rejection classified [`FailureClass::Retry`], after the caller's
1542    /// backoff. `false` when `id` is not waiting for a retry.
1543    pub fn retry(&mut self, _now_us: u64, id: ResourceId) -> bool {
1544        if self.stopped() || !self.parked.remove(&id) {
1545            return false;
1546        }
1547        self.queue.push_front(id);
1548        true
1549    }
1550
1551    /// Give up on one segment: it counts as delivered and a
1552    /// [`Output::Discontinuity`] marks the gap. `false` when `id` is not a
1553    /// segment waiting for a retry.
1554    pub fn skip(&mut self, _now_us: u64, id: ResourceId) -> bool {
1555        if self.stopped() || !matches!(id, ResourceId::Segment { .. }) || !self.parked.remove(&id) {
1556            return false;
1557        }
1558        self.failures.remove(&id);
1559        if let Some(live) = self.live.as_mut() {
1560            live.early.remove(&id);
1561        }
1562        self.mark_done(id);
1563        if let Some(slot) = self.slot_mut(id) {
1564            slot.state = SlotState::Skipped { marker: true };
1565        }
1566        self.drain();
1567        true
1568    }
1569
1570    // Delivery -------------------------------------------------------------
1571
1572    fn drain(&mut self) {
1573        loop {
1574            if self.phase != Phase::Running {
1575                return;
1576            }
1577            let Some(pi) = (self.deliver < self.periods.len()).then_some(self.deliver) else {
1578                return;
1579            };
1580            if !self.periods[pi].started {
1581                if !self.ready_to_start(pi) {
1582                    return;
1583                }
1584                self.start_period(pi);
1585            }
1586            self.deliver_slots(pi);
1587            let p = &self.periods[pi];
1588            let complete = p.issue_done && p.lanes.iter().all(|l| l.queue.is_empty());
1589            if !complete {
1590                return;
1591            }
1592            self.periods[pi].finished = true;
1593            self.periods[pi].lanes.clear();
1594            let total = self.mpd.as_ref().map_or(0, |m| m.periods.len());
1595            let last_of_mpd = self.periods[pi].index + 1 >= total;
1596            self.deliver += 1;
1597            self.prune_finished();
1598            if last_of_mpd {
1599                if self.live_open() {
1600                    return;
1601                }
1602                if self.any_started {
1603                    self.outputs.push_back(Output::EndOfStream);
1604                    self.phase = Phase::Ended;
1605                    self.live = None;
1606                } else {
1607                    self.go_fatal(FatalKind::MpdInvalid {
1608                        reason: "no_playable_representation",
1609                    });
1610                }
1611                return;
1612            }
1613        }
1614    }
1615
1616    fn prune_finished(&mut self) {
1617        while self.deliver > 0 && self.issue > 0 && self.periods.first().is_some_and(|p| p.finished)
1618        {
1619            self.periods.remove(0);
1620            self.deliver -= 1;
1621            self.issue -= 1;
1622            self.rep_loc.retain(|_, loc| loc.0 != 0);
1623            for loc in self.rep_loc.values_mut() {
1624                loc.0 -= 1;
1625            }
1626        }
1627        let mut keep: BTreeSet<InitKey> = self
1628            .periods
1629            .iter()
1630            .flat_map(|p| p.lanes.iter())
1631            .flat_map(|l| l.reps.iter())
1632            .map(|r| r.init_key.clone())
1633            .collect();
1634        if let Some(prev) = &self.prev {
1635            keep.extend(prev.lanes.iter().map(|(_, _, k)| k.clone()));
1636        }
1637        self.inits.retain(|k, _| keep.contains(k));
1638    }
1639
1640    fn ready_to_start(&self, pi: usize) -> bool {
1641        !self.live_blocked()
1642            && self.periods[pi].lanes.iter().all(|lane| {
1643                lane.current().is_none_or(|rep| {
1644                    self.inits
1645                        .get(&rep.init_key)
1646                        .is_some_and(|e| e.state == InitState::Ready)
1647                })
1648            })
1649    }
1650
1651    fn start_period(&mut self, pi: usize) {
1652        self.periods[pi].started = true;
1653        let mut assignments: Vec<(usize, u32, u32)> = Vec::new();
1654        let mut used_ranges: Vec<(u32, u32)> = Vec::new();
1655        for li in 0..self.periods[pi].lanes.len() {
1656            let n = self.periods[pi].lanes[li]
1657                .current()
1658                .and_then(|r| self.inits.get(&r.init_key))
1659                .map_or(0, |e| u32::try_from(e.specs.len()).unwrap_or(u32::MAX));
1660            let identity = self.periods[pi].lanes[li].identity.clone();
1661            let base = allocate_track_base(
1662                &mut self.track_ids,
1663                &mut self.next_track_id,
1664                &mut used_ranges,
1665                identity,
1666                n,
1667            );
1668            assignments.push((li, base, n));
1669        }
1670        for (li, base, n) in assignments {
1671            let lane = &mut self.periods[pi].lanes[li];
1672            lane.base = base;
1673            lane.count = n;
1674        }
1675        let current: Vec<TrackIdentity> = self.periods[pi]
1676            .lanes
1677            .iter()
1678            .map(|lane| lane.identity.clone())
1679            .collect();
1680        prune_track_ids(
1681            &mut self.track_ids,
1682            current.iter().chain(self.previous_identities.iter()),
1683        );
1684        self.previous_identities = current;
1685        let notices = std::mem::take(&mut self.periods[pi].notices);
1686        self.outputs.extend(notices);
1687        let live: Vec<usize> = (0..self.periods[pi].lanes.len())
1688            .filter(|&li| self.periods[pi].lanes[li].current().is_some())
1689            .collect();
1690        if live.is_empty() {
1691            self.periods[pi].issue_done = true;
1692            return;
1693        }
1694        let first = !self.any_started;
1695        self.any_started = true;
1696        let continuous = !first && self.is_continuous(pi);
1697        let manifest = self.manifest(pi);
1698        let p = &self.periods[pi];
1699        self.outputs.push_back(Output::Manifest(manifest));
1700        self.outputs.push_back(Output::PeriodChanged {
1701            index: p.index,
1702            id: p.id.clone(),
1703            start_us: p.start_us,
1704            continuous,
1705        });
1706        if !first && !continuous {
1707            self.discontinuity();
1708        }
1709        let mut summary = Vec::new();
1710        for li in live {
1711            let lane = &mut self.periods[pi].lanes[li];
1712            let ri = lane.reps.len() - 1;
1713            let key = lane.reps[ri].init_key.clone();
1714            summary.push((lane.kind, lane.set_id, key.clone()));
1715            if !continuous && let Some(bytes) = self.inits.get(&key).and_then(|e| e.bytes.as_ref())
1716            {
1717                self.outputs.push_back(Output::Init(bytes.as_ref().clone()));
1718            }
1719            lane.announced_init = Some(key);
1720            lane.announced_rep = Some(ri);
1721        }
1722        self.announce_tracks(pi);
1723        self.prev = Some(PrevPeriod {
1724            id: self.periods[pi].id.clone(),
1725            lanes: summary,
1726        });
1727    }
1728
1729    fn is_continuous(&self, pi: usize) -> bool {
1730        let Some(prev) = &self.prev else { return false };
1731        let Some(prev_id) = prev.id.as_deref() else {
1732            return false;
1733        };
1734        let Some(mpd) = self.mpd.as_ref() else {
1735            return false;
1736        };
1737        let p = &self.periods[pi];
1738        let live: Vec<&Lane> = p.lanes.iter().filter(|l| l.current().is_some()).collect();
1739        if live.len() != prev.lanes.len() {
1740            return false;
1741        }
1742        live.iter()
1743            .zip(&prev.lanes)
1744            .all(|(lane, (kind, set_id, key))| {
1745                let Some(rep) = lane.current() else {
1746                    return false;
1747                };
1748                let Some((_, set, _)) = rep.at.resolve(mpd) else {
1749                    return false;
1750                };
1751                lane.kind == *kind
1752                    && lane.set_id.is_some()
1753                    && lane.set_id == *set_id
1754                    && &rep.init_key == key
1755                    && continuity_signals(set, prev_id)
1756            })
1757    }
1758
1759    fn announce_tracks(&mut self, pi: usize) {
1760        let mut specs: Vec<TrackSpec> = Vec::new();
1761        for lane in &self.periods[pi].lanes {
1762            let Some(rep) = lane.current() else { continue };
1763            let Some(entry) = self.inits.get(&rep.init_key) else {
1764                continue;
1765            };
1766            for (pos, spec) in entry.specs.iter().enumerate() {
1767                let pos = u32::try_from(pos).unwrap_or(u32::MAX);
1768                if pos >= lane.count {
1769                    break;
1770                }
1771                let mut s = spec.clone();
1772                s.track_id = TrackId::new(lane.base.saturating_add(pos));
1773                specs.push(s);
1774            }
1775        }
1776        let fingerprint = format!("{specs:?}");
1777        if self.announced_tracks.as_deref() != Some(fingerprint.as_str()) {
1778            self.announced_tracks = Some(fingerprint);
1779            self.outputs.push_back(Output::Tracks(specs));
1780        }
1781    }
1782
1783    fn deliver_slots(&mut self, pi: usize) {
1784        for li in 0..self.periods[pi].lanes.len() {
1785            while self.periods[pi].lanes[li]
1786                .queue
1787                .front()
1788                .is_some_and(|f| !matches!(f.state, SlotState::Issued | SlotState::Held(_)))
1789            {
1790                let Some(slot) = self.periods[pi].lanes[li].queue.pop_front() else {
1791                    break;
1792                };
1793                if !matches!(slot.state, SlotState::Marker) {
1794                    self.outstanding = self.outstanding.saturating_sub(1);
1795                }
1796                if let Some(id) = slot.id {
1797                    self.reqs.remove(&id);
1798                }
1799                match slot.state {
1800                    SlotState::Marker => self.discontinuity(),
1801                    SlotState::Skipped { marker } => {
1802                        if marker {
1803                            self.discontinuity();
1804                        }
1805                    }
1806                    SlotState::Staged(staged) => {
1807                        self.emit_staged(pi, li, slot.rep, slot.id, staged);
1808                    }
1809                    SlotState::Issued | SlotState::Held(_) => {}
1810                }
1811            }
1812        }
1813    }
1814
1815    fn emit_staged(
1816        &mut self,
1817        pi: usize,
1818        li: usize,
1819        rep_idx: usize,
1820        id: Option<ResourceId>,
1821        staged: Staged,
1822    ) {
1823        if self.periods[pi].lanes[li].announced_rep != Some(rep_idx) {
1824            let key = self.periods[pi].lanes[li].reps[rep_idx].init_key.clone();
1825            if self.periods[pi].lanes[li].announced_init.as_ref() != Some(&key) {
1826                if let Some(bytes) = self.inits.get(&key).and_then(|e| e.bytes.as_ref()) {
1827                    self.outputs.push_back(Output::Init(bytes.as_ref().clone()));
1828                }
1829                self.periods[pi].lanes[li].announced_init = Some(key);
1830            }
1831            self.periods[pi].lanes[li].announced_rep = Some(rep_idx);
1832            self.announce_tracks(pi);
1833        }
1834        let lane = &self.periods[pi].lanes[li];
1835        let key = &lane.reps[rep_idx].init_key;
1836        let locals: Vec<u32> = self
1837            .inits
1838            .get(key)
1839            .map(|e| e.specs.iter().map(|s| s.track_id.value()).collect())
1840            .unwrap_or_default();
1841        let (base, count) = (lane.base, lane.count);
1842        let map = |local: u32| -> Option<TrackId> {
1843            let pos = locals.iter().position(|l| *l == local)?;
1844            let pos = u32::try_from(pos).ok()?;
1845            (pos < count).then(|| TrackId::new(base.saturating_add(pos)))
1846        };
1847        if staged.damaged > 0
1848            && let Some(id) = id
1849        {
1850            self.outputs.push_back(Output::FragmentDamaged {
1851                id,
1852                skipped: staged.damaged,
1853            });
1854        }
1855        for s in staged.skipped {
1856            let k = (s.track_id.map(TrackId::value), s.fourcc.clone());
1857            let first = self
1858                .inits
1859                .get_mut(&self.periods[pi].lanes[li].reps[rep_idx].init_key.clone())
1860                .is_some_and(|e| e.skipped_reported.insert(k));
1861            if first {
1862                self.outputs.push_back(Output::TrackSkipped {
1863                    track_id: s.track_id.and_then(|t| map(t.value())),
1864                    fourcc: s.fourcc,
1865                    reason: s.reason,
1866                });
1867            }
1868        }
1869        for (local, samples) in staged.tracks {
1870            if samples.is_empty() {
1871                continue;
1872            }
1873            if let Some(track_id) = map(local) {
1874                self.last_was_discontinuity = false;
1875                self.outputs
1876                    .push_back(Output::Samples { track_id, samples });
1877            }
1878        }
1879    }
1880
1881    // Manifest and switching ----------------------------------------------
1882
1883    fn init_url_of(&self, at: RepRef) -> Option<String> {
1884        let mpd = self.mpd.as_ref()?;
1885        let planned = plan_representation(mpd, at, &PlanOptions::default());
1886        let (reference, _) = Self::init_reference(mpd, at, &planned)?;
1887        let (period, set, rep) = at.resolve(mpd)?;
1888        let cands = filter_candidates(
1889            representation_candidates(mpd, period, set, rep),
1890            self.doc_url.as_ref(),
1891            self.dash.cross_origin_urls,
1892        );
1893        cands.first()?.resolve(self.doc_url.as_ref(), &reference)
1894    }
1895
1896    fn manifest(&self, pi: usize) -> ManifestSelection {
1897        let mpd = self.mpd.as_ref().expect("mpd present");
1898        let p = &self.periods[pi];
1899        let period = &mpd.periods[p.index];
1900        let audio_group = p
1901            .selection
1902            .chosen_audio
1903            .and_then(|a| period.adaptation_sets.get(a.set))
1904            .and_then(|s| s.group.or(s.id))
1905            .map(|g| g.to_string());
1906        let variants = p
1907            .selection
1908            .variants
1909            .iter()
1910            .filter_map(|&at| {
1911                let (_, set, rep) = at.resolve(mpd)?;
1912                let common = effective_common(set, rep);
1913                Some(VariantInfo {
1914                    rep: at,
1915                    init_url: self.init_url_of(at),
1916                    bandwidth: rep.bandwidth,
1917                    resolution: common.width.zip(common.height),
1918                    codecs: common.codecs,
1919                    audio_group: audio_group.clone(),
1920                })
1921            })
1922            .collect();
1923        let audio: Vec<AudioRendition> = p
1924            .selection
1925            .audio_sets
1926            .iter()
1927            .filter_map(|a| {
1928                let set = period.adaptation_sets.get(a.set)?;
1929                let first = a.reps.first()?;
1930                let (_, _, first_rep) = first.resolve(mpd)?;
1931                let chosen = p
1932                    .selection
1933                    .chosen_audio
1934                    .filter(|c| c.set == a.set)
1935                    .unwrap_or(*first);
1936                Some(AudioRendition {
1937                    set: a.set,
1938                    group: set.group.or(set.id).map(|g| g.to_string()),
1939                    name: set
1940                        .labels
1941                        .first()
1942                        .cloned()
1943                        .or_else(|| set.lang.clone())
1944                        .unwrap_or_else(|| first_rep.id.clone()),
1945                    language: set.lang.clone(),
1946                    default: a.is_main,
1947                    autoselect: true,
1948                    init_url: self.init_url_of(chosen),
1949                    reps: a.reps.clone(),
1950                })
1951            })
1952            .collect();
1953        let chosen_audio = p
1954            .selection
1955            .chosen_audio
1956            .and_then(|c| p.selection.audio_sets.iter().position(|a| a.set == c.set));
1957        ManifestSelection {
1958            period: p.index,
1959            variants,
1960            chosen: p.selection.chosen_variant,
1961            audio,
1962            chosen_audio,
1963        }
1964    }
1965
1966    /// Switch the video lane to variant `index` of the latest
1967    /// [`Output::Manifest`]. Segments already requested keep their
1968    /// Representation; the new one continues from the segment containing the
1969    /// next unrequested media time, its init is fetched, and the track set is
1970    /// re-announced (DESIGN 5.6). A switch that is not segment-aligned
1971    /// (`@segmentAlignment` with `@startWithSAP` 1 or 2 on the new set) is
1972    /// marked with a [`Output::Discontinuity`].
1973    ///
1974    /// # Errors
1975    /// [`Error::NoSuchVariant`]; [`Error::NotSwitchable`] when the Period
1976    /// being played is not the one being fetched, there is no video lane, or
1977    /// nothing is left to fetch; [`Error::Stopped`].
1978    pub fn select_representation(&mut self, index: usize) -> Result<()> {
1979        self.ensure_live()?;
1980        let pi = self.switchable_period()?;
1981        let at = *self.periods[pi]
1982            .selection
1983            .variants
1984            .get(index)
1985            .ok_or(Error::NoSuchVariant)?;
1986        self.switch_lane(pi, SetClass::Video, at)?;
1987        self.periods[pi].selection.chosen_variant = Some(index);
1988        Ok(())
1989    }
1990
1991    /// Play audio set `index` of the latest [`Output::Manifest`]'s `audio`
1992    /// list, or stop audio with `None`. The best Representation of the set
1993    /// the codec policy allows is used; segments already requested keep their
1994    /// Representation, the new one continues from the segment containing the
1995    /// next unrequested media time and the track set is re-announced. The
1996    /// audio track ids keep their numbers. `None` stops requesting audio
1997    /// segments (those already requested are still delivered); a later
1998    /// `Some` starts audio again.
1999    ///
2000    /// # Errors
2001    /// [`Error::NoSuchAudio`] for an index outside the list;
2002    /// [`Error::NotSwitchable`] when the Period being played is not the one
2003    /// being fetched or it has no audio lane; [`Error::Stopped`].
2004    pub fn select_audio(&mut self, index: Option<usize>) -> Result<()> {
2005        self.ensure_live()?;
2006        let pi = self.switchable_period()?;
2007        let Some(index) = index else {
2008            let lane = self.periods[pi]
2009                .lanes
2010                .iter_mut()
2011                .find(|l| l.kind == SetClass::Audio)
2012                .ok_or(Error::NotSwitchable)?;
2013            if let Some(rep) = lane.reps.last_mut() {
2014                rep.dead = true;
2015            }
2016            self.periods[pi].selection.chosen_audio = None;
2017            return Ok(());
2018        };
2019        let at = self.best_audio_rep(pi, index)?;
2020        self.switch_lane(pi, SetClass::Audio, at)?;
2021        self.periods[pi].selection.chosen_audio = Some(at);
2022        Ok(())
2023    }
2024
2025    fn switchable_period(&self) -> Result<usize> {
2026        let pi = self.deliver;
2027        if pi != self.issue || self.periods.get(pi).is_none_or(|p| !p.started) {
2028            return Err(Error::NotSwitchable);
2029        }
2030        Ok(pi)
2031    }
2032
2033    fn best_audio_rep(&self, pi: usize, index: usize) -> Result<RepRef> {
2034        let mpd = self.mpd.as_ref().expect("mpd present");
2035        let set = self.periods[pi]
2036            .selection
2037            .audio_sets
2038            .get(index)
2039            .ok_or(Error::NoSuchAudio)?;
2040        let bandwidth = |r: &RepRef| r.resolve(mpd).map_or(0, |(_, _, rep)| rep.bandwidth);
2041        set.reps
2042            .iter()
2043            .max_by_key(|r| bandwidth(r))
2044            .copied()
2045            .ok_or(Error::NoSuchAudio)
2046    }
2047
2048    fn switch_lane(&mut self, pi: usize, kind: SetClass, at: RepRef) -> Result<()> {
2049        let revive = kind == SetClass::Audio;
2050        let li = self.periods[pi]
2051            .lanes
2052            .iter()
2053            .position(|l| l.kind == kind && (revive || l.current().is_some()))
2054            .ok_or(Error::NotSwitchable)?;
2055        let (current_at, was_dead, switch_us) = {
2056            let lane = &self.periods[pi].lanes[li];
2057            let cur = lane.reps.last().ok_or(Error::NotSwitchable)?;
2058            let plan = cur.plan.as_ref().ok_or(Error::NotSwitchable)?;
2059            let at_us = match plan.get(cur.next_issue) {
2060                Some(seg) => seg.start_us,
2061                None => plan
2062                    .len()
2063                    .and_then(|n| n.checked_sub(1))
2064                    .and_then(|last| plan.get(last))
2065                    .map(|seg| seg.start_us.saturating_add(seg.duration_us))
2066                    .ok_or(Error::NotSwitchable)?,
2067            };
2068            (cur.at, cur.dead, at_us)
2069        };
2070        if current_at == at && !was_dead {
2071            return Ok(());
2072        }
2073        let aligned = {
2074            let mpd = self.mpd.as_ref().expect("mpd present");
2075            at.resolve(mpd).is_some_and(|(_, set, _)| {
2076                set.segment_alignment
2077                    && matches!(
2078                        set.start_with_sap,
2079                        Some(START_WITH_SAP_1 | START_WITH_SAP_2)
2080                    )
2081            })
2082        };
2083        let mut rep = match self.build_rep(at) {
2084            Ok(r) => r,
2085            Err(cause) => {
2086                self.outputs.push_back(Output::TrackDropped {
2087                    period: self.periods[pi].index,
2088                    set: at.set,
2089                    rep: Some(at.rep),
2090                    cause,
2091                });
2092                return Err(Error::NotSwitchable);
2093            }
2094        };
2095        match rep.plan.as_ref() {
2096            Some(plan) => {
2097                rep.next_issue = plan.index_at_or_before(switch_us).unwrap_or(0);
2098                rep.joined = true;
2099                rep.last_time = rep
2100                    .next_issue
2101                    .checked_sub(1)
2102                    .and_then(|i| plan.get(i))
2103                    .map(|s| s.time_ticks);
2104            }
2105            None => rep.start_time_us = Some(switch_us),
2106        }
2107        if !aligned {
2108            self.periods[pi].lanes[li].queue.push_back(Slot {
2109                id: None,
2110                rep: 0,
2111                plan_index: 0,
2112                state: SlotState::Marker,
2113            });
2114        }
2115        let ri = self.periods[pi].lanes[li].reps.len();
2116        self.periods[pi].lanes[li].reps.push(rep);
2117        self.periods[pi].issue_done = false;
2118        self.register_rep(pi, li, ri);
2119        Ok(())
2120    }
2121}
2122
2123fn ranges_overlap(a_base: u32, a_len: u32, b_base: u32, b_len: u32) -> bool {
2124    let a_end = a_base.saturating_add(a_len);
2125    let b_end = b_base.saturating_add(b_len);
2126    a_base < b_end && b_base < a_end
2127}
2128
2129fn allocate_track_base(
2130    track_ids: &mut BTreeMap<TrackIdentity, (u32, u32)>,
2131    next_track_id: &mut u32,
2132    used: &mut Vec<(u32, u32)>,
2133    identity: TrackIdentity,
2134    n: u32,
2135) -> u32 {
2136    if n == 0 {
2137        return 0;
2138    }
2139    if let Some((existing, len)) = track_ids.get(&identity).copied()
2140        && len >= n
2141        && !used.iter().any(|&(b, l)| ranges_overlap(existing, n, b, l))
2142    {
2143        used.push((existing, n));
2144        return existing;
2145    }
2146    let base = *next_track_id;
2147    *next_track_id = next_track_id.saturating_add(n);
2148    track_ids.insert(identity, (base, n));
2149    used.push((base, n));
2150    base
2151}
2152
2153/// Drops identities outside `keep`. `next_track_id` is never lowered, so a
2154/// pruned identity that returns gets a fresh range and no id is announced for
2155/// two identities. `next_track_id` saturates at `u32::MAX`; at 20 tracks per
2156/// Period and one Period every 10 s that takes over 700 years.
2157fn prune_track_ids<'a>(
2158    track_ids: &mut BTreeMap<TrackIdentity, (u32, u32)>,
2159    keep: impl Iterator<Item = &'a TrackIdentity>,
2160) {
2161    let keep: BTreeSet<&TrackIdentity> = keep.collect();
2162    track_ids.retain(|identity, _| keep.contains(identity));
2163}
2164
2165#[cfg(test)]
2166mod track_id_allocation_tests {
2167    use super::*;
2168
2169    #[test]
2170    fn two_same_identity_lanes_get_unique_bases() {
2171        let mut track_ids: BTreeMap<TrackIdentity, (u32, u32)> = BTreeMap::new();
2172        let mut next = 1u32;
2173        let mut used = Vec::new();
2174        let id = TrackIdentity {
2175            kind: SetClass::Audio,
2176            set_id: None,
2177            group: None,
2178            lang: Some("en".to_string()),
2179            roles: Vec::new(),
2180        };
2181        let a = allocate_track_base(&mut track_ids, &mut next, &mut used, id.clone(), 1);
2182        let b = allocate_track_base(&mut track_ids, &mut next, &mut used, id, 1);
2183        assert_ne!(a, b, "duplicate base for two lanes with equal identity");
2184    }
2185
2186    #[test]
2187    fn growing_identity_does_not_overlap_neighbor() {
2188        let mut track_ids: BTreeMap<TrackIdentity, (u32, u32)> = BTreeMap::new();
2189        let mut next = 1u32;
2190        let mut used = Vec::new();
2191        let a = TrackIdentity {
2192            kind: SetClass::Video,
2193            set_id: None,
2194            group: None,
2195            lang: None,
2196            roles: Vec::new(),
2197        };
2198        let b = TrackIdentity {
2199            kind: SetClass::Audio,
2200            set_id: None,
2201            group: None,
2202            lang: Some("en".to_string()),
2203            roles: Vec::new(),
2204        };
2205        // Period 1: A has 1 track.
2206        let a1 = allocate_track_base(&mut track_ids, &mut next, &mut used, a.clone(), 1);
2207        // B allocated in between.
2208        let b1 = allocate_track_base(&mut track_ids, &mut next, &mut used, b.clone(), 1);
2209        // Period 2: A needs 2 tracks; used reset for the new Period.
2210        let mut used2 = Vec::new();
2211        let a2 = allocate_track_base(&mut track_ids, &mut next, &mut used2, a.clone(), 2);
2212        // All allocated ids unique.
2213        let mut ids = std::collections::BTreeSet::new();
2214        ids.insert(a1);
2215        ids.insert(b1);
2216        ids.insert(a2);
2217        ids.insert(a2 + 1);
2218        assert_eq!(ids.len(), 4, "range overlap or duplicate ids");
2219    }
2220
2221    fn video(set_id: u64) -> TrackIdentity {
2222        TrackIdentity {
2223            kind: SetClass::Video,
2224            set_id: Some(set_id),
2225            group: None,
2226            lang: None,
2227            roles: Vec::new(),
2228        }
2229    }
2230
2231    fn audio(set_id: u64) -> TrackIdentity {
2232        TrackIdentity {
2233            kind: SetClass::Audio,
2234            set_id: Some(set_id),
2235            group: None,
2236            lang: None,
2237            roles: Vec::new(),
2238        }
2239    }
2240
2241    #[test]
2242    fn renumbered_periods_keep_the_identity_map_bounded_and_ids_unique_per_kind() {
2243        let mut track_ids: BTreeMap<TrackIdentity, (u32, u32)> = BTreeMap::new();
2244        let mut next = 1u32;
2245        let mut owner: BTreeMap<u32, SetClass> = BTreeMap::new();
2246        let mut previous: Vec<TrackIdentity> = Vec::new();
2247        let mut peak = 0usize;
2248        for period in 0..200u64 {
2249            let current = vec![video(period), audio(period)];
2250            let mut used = Vec::new();
2251            for identity in &current {
2252                let base =
2253                    allocate_track_base(&mut track_ids, &mut next, &mut used, identity.clone(), 1);
2254                let kind = owner.entry(base).or_insert(identity.kind);
2255                assert_eq!(*kind, identity.kind, "id {base} announced for two kinds");
2256            }
2257            prune_track_ids(&mut track_ids, current.iter().chain(previous.iter()));
2258            peak = peak.max(track_ids.len());
2259            previous = current;
2260        }
2261        assert!(peak <= 4, "identity map grew to {peak}");
2262        assert_eq!(next, 401, "ids must stay monotonic, never recycled");
2263    }
2264}