1mod 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
31const MISSING_TAIL_MAX_US: i128 = 50_000;
35const ROTATE_AFTER_FAILURES: u32 = 2;
38const STATUS_NOT_FOUND: u16 = 404;
39const STATUS_GONE: u16 = 410;
40const STATUS_TOO_EARLY: u16 = 425;
41const 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
194#[non_exhaustive]
195pub struct StateSize {
196 pub periods: usize,
198 pub requests: usize,
200 pub inflight: usize,
202 pub parked: usize,
204 pub failures: usize,
206 pub done: usize,
208 pub inits: usize,
210 pub reps: usize,
212 pub queued: usize,
214 pub outputs: usize,
216 pub live: usize,
218 pub track_identities: usize,
220}
221
222pub 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 #[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 #[must_use]
403 pub fn mpd_url(&self) -> &str {
404 &self.mpd_url
405 }
406
407 pub fn set_mpd_base(&mut self, base: impl Into<String>) {
411 self.doc_url = Url::parse(&base.into()).ok();
412 }
413
414 #[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 #[must_use]
425 pub fn presentation_ended(&self) -> bool {
426 self.phase == Phase::Ended
427 }
428
429 #[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 #[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 #[must_use]
462 pub fn awaits(&self, id: ResourceId) -> bool {
463 self.inflight.contains(&id)
464 }
465
466 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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
2153fn 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 let a1 = allocate_track_base(&mut track_ids, &mut next, &mut used, a.clone(), 1);
2207 let b1 = allocate_track_base(&mut track_ids, &mut next, &mut used, b.clone(), 1);
2209 let mut used2 = Vec::new();
2211 let a2 = allocate_track_base(&mut track_ids, &mut next, &mut used2, a.clone(), 2);
2212 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 ¤t {
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}