Skip to main content

smooth_runtime/server/
engine.rs

1//! [`SmoothOrigin`]: the live Manifest and fragment resolution over a `Trunk`.
2
3use std::collections::HashMap;
4use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
5
6use broadcast_common::{Timestamp, Unpackage};
7use broadcast_smooth::{C, QualityLevel, SmoothManifest, StreamIndex, StreamType};
8use bytes::Bytes;
9use media_plane::Trunk;
10use media_plane::egress::{AwaitPolicy, CachePolicy, EgressResponse, ServedEgress};
11use media_plane::ids::SegmentSeq;
12use transmux::{
13    Fmp4Demux, Media, SmoothCodec, SmoothPackager, SmoothStreamType, Track, TrackSpec,
14    resolve_codec, to_smooth_ticks,
15};
16
17/// The prefix of this origin's fragment-URL template and of every fragment
18/// request path: the `QualityLevels(N)` segment of MS-SSTR §2.2.3.
19pub const QUALITY_LEVELS_PREFIX: &str = "QualityLevels(";
20
21/// The segment of a fragment URL carrying its start time: `Fragments(N={start time})`
22/// (MS-SSTR §2.2.3).
23pub const FRAGMENTS_SEGMENT: &str = "Fragments(";
24
25/// `StreamIndex@Subtype` for a text stream: `SUBT`, "Subtitles that are used for
26/// foreign-language audio" (MS-SSTR §2.2.2.3). The IR does not say whether a
27/// track is a caption service for the deaf (`CAPT`), so the subtitle category is
28/// used.
29pub const SUBTYPE_SUBT: &str = "SUBT";
30
31/// `QualityLevel@AudioTag` for raw AAC (MS-SSTR §2.2.2.5).
32const AUDIO_TAG_AAC: u32 = 255;
33/// `QualityLevel@BitsPerSample` advertised for AAC audio.
34const AAC_BITS_PER_SAMPLE: u16 = 16;
35/// The quality-level count every stream this origin emits advertises: one
36/// `QualityLevel` per track, whose fragments are served at that one rate.
37const STREAM_QUALITY_LEVELS: u32 = 1;
38/// The 1-based ordinal of the first stream of a `Type`; its `Name` is the bare
39/// `Type`.
40const FIRST_STREAM_ORDINAL: usize = 1;
41/// `SmoothStreamingMedia@LookAheadFragmentCount`: this origin publishes a
42/// fragment only once its segment has closed, so it never runs ahead of the
43/// manifest it advertises.
44const LOOK_AHEAD_FRAGMENT_COUNT: u32 = 0;
45/// Target fragment duration above which a live segment is never cut: set far
46/// above any real segment so one `Trunk` segment maps to exactly one Smooth
47/// fragment. Whole seconds because [`SmoothPackager::target_duration_secs`] is
48/// an integer.
49const SINGLE_SEGMENT_FRAGMENT_TARGET_SECS: u32 = 3600;
50/// Bits per second floor of the advertised `Bitrate` for an AAC track.
51const MIN_AUDIO_BITRATE: u64 = 64_000;
52/// Bits per second ceiling of the advertised `Bitrate` for an AAC track (320
53/// kbps, stereo AAC).
54const MAX_AUDIO_BITRATE: u64 = 320_000;
55/// Bits per second floor of the advertised `Bitrate` for a video track.
56const MIN_VIDEO_BITRATE: u64 = 500_000;
57/// Bits per second ceiling of the advertised `Bitrate` for a video track.
58const MAX_VIDEO_BITRATE: u64 = 20_000_000;
59/// Bits per coded pixel per second in the video `Bitrate` estimate.
60const VIDEO_BITS_PER_PIXEL: u64 = 4;
61/// Width assumed for the video `Bitrate` estimate when the codec carries none.
62const DEFAULT_VIDEO_WIDTH: u64 = 1280;
63/// Height assumed for the video `Bitrate` estimate when the codec carries none.
64const DEFAULT_VIDEO_HEIGHT: u64 = 720;
65/// Sampling rate assumed for the audio `Bitrate` estimate when the codec carries
66/// none.
67const DEFAULT_SAMPLING_RATE: u64 = 48_000;
68/// Channel count assumed for the audio `Bitrate` estimate when the codec carries
69/// none.
70const DEFAULT_CHANNELS: u64 = 2;
71/// Bits per sample assumed for the audio `Bitrate` estimate.
72const ESTIMATE_BITS_PER_SAMPLE: u64 = 16;
73
74/// What identifies one advertised stream: its media kind and the 1-based
75/// ordinal of that kind, `(kind, ordinal)`, exactly the pair the manifest's
76/// `StreamIndex@Name`/`@Url` and the request grammar name. Two tracks of one
77/// kind are therefore distinct.
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79pub struct StreamKey {
80    /// The `StreamIndex@Type` kind.
81    pub kind: SmoothStreamType,
82    /// 1-based position among the streams of `kind`.
83    pub ordinal: u32,
84}
85
86impl StreamKey {
87    /// The `StreamIndex@Name` this key is advertised under: the bare `Type` for
88    /// the first stream of a kind, `TYPE_N` for the rest (MS-SSTR §2.2.2.3).
89    pub fn name(&self) -> String {
90        let kind = self.kind.name();
91        match self.ordinal as usize {
92            FIRST_STREAM_ORDINAL => kind.to_string(),
93            ordinal => format!("{kind}_{ordinal}"),
94        }
95    }
96
97    /// The `StreamIndex@Url` fragment-URL template whose `Fragments(N=...)`
98    /// token carries this key's [`Self::name`], so each stream of a kind is
99    /// addressed distinctly.
100    pub fn url_template(&self) -> String {
101        format!(
102            "{QUALITY_LEVELS_PREFIX}{{bitrate}})/{FRAGMENTS_SEGMENT}{}={{start time}})",
103            self.name()
104        )
105    }
106}
107
108/// One fragment a manifest advertises: its stream, absolute start time and
109/// duration in [`SMOOTH_TIMESCALE`](transmux::SMOOTH_TIMESCALE) ticks.
110#[derive(Debug, Clone, Copy, PartialEq, Eq)]
111#[non_exhaustive]
112pub struct FragmentLocation {
113    /// The stream the fragment belongs to.
114    pub stream: StreamKey,
115    /// `c@t` — the fragment's absolute start time on its stream's timeline.
116    pub start: u64,
117    /// `c@d` — the fragment's duration.
118    pub duration: u64,
119}
120
121/// What a caller asks a [`SmoothOrigin`] for: the client Manifest, or one
122/// fragment of it.
123#[derive(Debug, Clone, Copy, PartialEq, Eq)]
124#[non_exhaustive]
125pub enum SmoothRequest {
126    /// `GET .../Manifest` — the live client Manifest.
127    Manifest,
128    /// `GET ...  /QualityLevels(BITRATE)/Fragments(KIND=START)`.
129    Fragment {
130        /// Which advertised stream (by name) the request names.
131        stream: StreamKey,
132        /// The `QualityLevel@Bitrate` the request's `QualityLevels(N)` carries.
133        bitrate: u64,
134        /// The requested `c@t`, in [`SMOOTH_TIMESCALE`](transmux::SMOOTH_TIMESCALE)
135        /// ticks.
136        start: u64,
137    },
138}
139
140/// The body of a [`SmoothOrigin`] answer.
141#[derive(Debug, Clone, PartialEq, Eq)]
142#[non_exhaustive]
143pub enum SmoothBody {
144    /// The client Manifest document.
145    Manifest(String),
146    /// One fragment's `styp`+`moof`(+`tfxd`)+`mdat` bytes.
147    Fragment(Bytes),
148}
149
150/// Why a [`SmoothOrigin`] cannot describe the trunk's media.
151#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
152#[non_exhaustive]
153pub enum SmoothOriginError {
154    /// A track's codec has no MS-SSTR representation (HEVC, Opus, a DVB
155    /// subtitle, …).
156    #[error("track {track} has no Smooth Streaming codec representation")]
157    UnrepresentableCodec {
158        /// The track's id.
159        track: u32,
160    },
161    /// A track's codec is representable in principle but unusable in fact (an
162    /// AAC track whose `esds` carries no DecoderSpecificInfo, an AVC track
163    /// whose SPS does not decode) — the packager's own error, rendered.
164    #[error("track {track} cannot be described: {reason}")]
165    Codec {
166        /// The track's id.
167        track: u32,
168        /// The codec resolver's error, rendered.
169        reason: String,
170    },
171    /// The manifest model's writer refused the assembled document.
172    #[error("manifest could not be rendered: {reason}")]
173    Manifest {
174        /// The writer's error, rendered.
175        reason: String,
176    },
177}
178
179/// One advertised stream's chunk timeline: per segment, the fragment's absolute
180/// start and duration on the track's own timeline.
181#[derive(Debug, Clone)]
182struct ChunkTiming {
183    sequence: SegmentSeq,
184    start: u64,
185    duration: u64,
186}
187
188/// One resolved track: its manifest codec parameters and chunk timeline.
189#[derive(Debug, Clone)]
190struct StreamPlan {
191    key: StreamKey,
192    codec: SmoothCodec,
193    interleaved_ordinal: u32,
194    chunks: Vec<ChunkTiming>,
195}
196
197/// An assembled manifest plus the query a [`SmoothRequest::Fragment`] needs.
198#[derive(Debug, Clone)]
199struct Plan {
200    manifest: String,
201    streams: Vec<StreamPlan>,
202}
203
204impl Plan {
205    /// The manifest model `broadcast-smooth` serializes.
206    fn manifest(&self) -> SmoothManifest {
207        let duration = self
208            .streams
209            .iter()
210            .filter_map(|stream| stream.chunks.last())
211            .map(|chunk| chunk.start.saturating_add(chunk.duration))
212            .max()
213            .unwrap_or(0);
214        let streams = self.streams.iter().map(stream_index).collect();
215        SmoothManifest::new(streams)
216            .with_duration(duration)
217            .live(duration, LOOK_AHEAD_FRAGMENT_COUNT)
218    }
219
220    /// The chunk of `stream` starting at `start`, or `None` when the window
221    /// holds no chunk of that stream starting there (a fragment older than the
222    /// window's first chunk, or later than its last one).
223    fn chunk(&self, stream: StreamKey, start: u64) -> Option<(StreamPlan, ChunkTiming)> {
224        let plan = self.streams.iter().find(|plan| plan.key == stream)?;
225        if !window_holds(plan, start) {
226            return None;
227        }
228        let chunk = plan.chunks.iter().find(|chunk| chunk.start == start)?;
229        Some((plan.clone(), chunk.clone()))
230    }
231}
232
233/// `true` when `start` falls inside the window `plan` describes: at or after
234/// its first chunk and before the end of its last one, each bound computed
235/// with checked arithmetic so no comparison can wrap.
236///
237/// A start outside that span is a fragment the window does not hold — an
238/// older form of "this byte range is gone", the same answer an evicted HLS
239/// segment gets — never a fragment assembled from a neighbouring chunk.
240fn window_holds(plan: &StreamPlan, start: u64) -> bool {
241    let (Some(first), Some(last)) = (plan.chunks.first(), plan.chunks.last()) else {
242        return false;
243    };
244    let Some(end) = last.start.checked_add(last.duration) else {
245        return start >= first.start;
246    };
247    start >= first.start && start < end
248}
249
250/// The `StreamIndex` element for one resolved track: its `Name`/`Url` from the
251/// stream key (so two streams of one kind stay distinct), the one `QualityLevel`
252/// its fragments are served at, and one `c` per chunk.
253fn stream_index(stream: &StreamPlan) -> StreamIndex {
254    let codec = &stream.codec;
255    let mut level = QualityLevel::new(
256        codec.fourcc,
257        advertised_bitrate(codec),
258        codec.codec_private_data.clone(),
259    );
260    if let (Some(width), Some(height)) = (codec.width, codec.height) {
261        level = level.with_size(width, height);
262    }
263    if codec.stream_type == SmoothStreamType::Audio
264        && let (Some(rate), Some(channels)) = (codec.sampling_rate, codec.channels)
265    {
266        level = level.with_audio(rate, channels, AAC_BITS_PER_SAMPLE, 0, AUDIO_TAG_AAC);
267    }
268    level = level.without_packet_size();
269
270    let chunks = stream
271        .chunks
272        .iter()
273        .enumerate()
274        .map(|(i, chunk)| {
275            let ordinal = u32::try_from(i).unwrap_or(u32::MAX);
276            C::new(chunk.duration).numbered(ordinal).at(chunk.start)
277        })
278        .collect();
279    let count = u32::try_from(stream.chunks.len()).unwrap_or(u32::MAX);
280    let mut index = StreamIndex::new(
281        manifest_stream_type(stream.codec.stream_type),
282        stream.key.url_template(),
283        vec![level],
284    )
285    .with_counts(count, STREAM_QUALITY_LEVELS)
286    .with_chunks(chunks);
287    if stream.codec.stream_type == SmoothStreamType::Text {
288        index = index.with_subtype(SUBTYPE_SUBT);
289    }
290    if stream.key.ordinal as usize > FIRST_STREAM_ORDINAL {
291        index = index.with_name(stream.key.name());
292    }
293    index
294}
295
296/// The `StreamIndex@Type` token for a stream kind.
297fn manifest_stream_type(kind: SmoothStreamType) -> StreamType {
298    match kind {
299        SmoothStreamType::Video => StreamType::Video,
300        SmoothStreamType::Audio => StreamType::Audio,
301        _ => StreamType::Text,
302    }
303}
304
305/// The `QualityLevel@Bitrate` advertised for `codec`: a documented per-codec
306/// estimate, because the live window carries no per-track byte count at
307/// manifest-render time (Smooth's `Bitrate` is advisory ABR metadata that the
308/// fragment URLs carry back).
309fn advertised_bitrate(codec: &SmoothCodec) -> u64 {
310    match codec.stream_type {
311        SmoothStreamType::Audio => {
312            let rate = u64::from(codec.sampling_rate.unwrap_or(DEFAULT_SAMPLING_RATE as u32));
313            let channels = u64::from(codec.channels.unwrap_or(DEFAULT_CHANNELS as u16));
314            rate.saturating_mul(channels)
315                .saturating_mul(ESTIMATE_BITS_PER_SAMPLE)
316                .clamp(MIN_AUDIO_BITRATE, MAX_AUDIO_BITRATE)
317        }
318        SmoothStreamType::Video => {
319            let width = u64::from(codec.width.unwrap_or(DEFAULT_VIDEO_WIDTH as u32));
320            let height = u64::from(codec.height.unwrap_or(DEFAULT_VIDEO_HEIGHT as u32));
321            width
322                .saturating_mul(height)
323                .saturating_mul(VIDEO_BITS_PER_PIXEL)
324                .clamp(MIN_VIDEO_BITRATE, MAX_VIDEO_BITRATE)
325        }
326        _ => MIN_AUDIO_BITRATE,
327    }
328}
329
330/// The segments the origin should describe, given the `Trunk` it reads.
331///
332/// A `Trunk` holds the media; which *published* segments a program presents
333/// is the host's own bookkeeping — the same source `multimux` renders its
334/// DASH window from — so the origin asks for it rather than inferring it. A
335/// presence check with no store of its own is the plain adapter.
336pub trait SegmentProvider: Send + Sync {
337    /// The program's published (sequence number, bytes) pairs, oldest first.
338    /// The bytes are shared handles: a provider hands out the segment it
339    /// already holds, never a copy.
340    fn segments(&self) -> Vec<(SegmentSeq, Bytes)>;
341}
342
343impl<F> SegmentProvider for F
344where
345    F: Fn() -> Vec<(SegmentSeq, Bytes)> + Send + Sync,
346{
347    fn segments(&self) -> Vec<(SegmentSeq, Bytes)> {
348        self()
349    }
350}
351
352/// The tracks the origin should describe, given the `Trunk` it reads.
353///
354/// Like [`SegmentProvider`], which track set a program presents is the host's
355/// own bookkeeping (the one its DASH window and HLS origin already use), so
356/// the origin asks for it. A `Trunk`'s own announcement is the plain adapter
357/// for a host that publishes there.
358pub trait TrackProvider: Send + Sync {
359    /// The program's current track specs, in manifest order.
360    fn tracks(&self) -> Vec<TrackSpec>;
361}
362
363impl<F> TrackProvider for F
364where
365    F: Fn() -> Vec<TrackSpec> + Send + Sync,
366{
367    fn tracks(&self) -> Vec<TrackSpec> {
368        self()
369    }
370}
371
372/// The live Smooth origin over one `Trunk`.
373pub struct SmoothOrigin {
374    trunk: Arc<Trunk>,
375    /// The init segment the trunk's segments were muxed with — the one thing
376    /// no `Trunk` ring holds, and what a fragment request is re-demuxed
377    /// against.
378    init: Mutex<InitSlot>,
379    /// The program's published segments.
380    provider: Arc<dyn SegmentProvider>,
381    /// The program's current track set, as the host records it.
382    tracks: Arc<dyn TrackProvider>,
383    /// The plan built for the window it was last asked about; a request whose
384    /// [`PlanKey`] is unchanged reuses it.
385    plan_cache: Mutex<Option<CachedPlan>>,
386    /// Held for the whole of a plan build, so concurrent requests that all
387    /// miss build it once and the rest find it cached.
388    build_gate: Mutex<()>,
389    /// Packaged fragments of the cached plan's window, each built once.
390    fragments: Mutex<HashMap<FragmentKey, FragmentSlot>>,
391    #[cfg(test)]
392    plan_builds: std::sync::atomic::AtomicUsize,
393    #[cfg(test)]
394    fragment_builds: std::sync::atomic::AtomicUsize,
395}
396
397/// The stored init segment and how many distinct ones have been stored.
398#[derive(Default)]
399struct InitSlot {
400    bytes: Option<Bytes>,
401    generation: u64,
402}
403
404/// What a [`Plan`] depends on: the init segment generation, the track set and
405/// the window's segment sequence numbers. Equal keys yield equal plans.
406#[derive(Debug, Clone, PartialEq, Eq)]
407struct PlanKey {
408    init_generation: u64,
409    tracks: Vec<TrackSpec>,
410    window: Vec<SegmentSeq>,
411}
412
413struct CachedPlan {
414    key: PlanKey,
415    plan: Result<Arc<Plan>, SmoothOriginError>,
416}
417
418/// One packaged fragment: the stream, its `c@t` and the segment it came from.
419#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
420struct FragmentKey {
421    kind: &'static str,
422    ordinal: u32,
423    start: u64,
424    sequence: u32,
425}
426
427impl FragmentKey {
428    fn new(stream: StreamKey, chunk: &ChunkTiming) -> Self {
429        Self {
430            kind: stream.kind.name(),
431            ordinal: stream.ordinal,
432            start: chunk.start,
433            sequence: chunk.sequence.value(),
434        }
435    }
436
437    fn is_in(&self, plan: &Plan) -> bool {
438        plan.streams.iter().any(|stream| {
439            stream.key.kind.name() == self.kind
440                && stream.key.ordinal == self.ordinal
441                && stream.chunks.iter().any(|chunk| {
442                    chunk.start == self.start && chunk.sequence.value() == self.sequence
443                })
444        })
445    }
446}
447
448/// A fragment packaged at most once; concurrent requests for it wait on the
449/// first.
450type FragmentSlot = Arc<OnceLock<Result<Bytes, SmoothOriginError>>>;
451
452/// Everything one request reads: the plan for the window it saw, that window's
453/// segments and the init segment they were planned against.
454struct Snapshot {
455    plan: Arc<Plan>,
456    init: Bytes,
457    segments: Vec<(SegmentSeq, Bytes)>,
458}
459
460/// What a [`SmoothOrigin`] is built from: every part is required, so
461/// [`SmoothOriginConfig::new`] takes all three.
462#[non_exhaustive]
463pub struct SmoothOriginConfig {
464    /// The trunk whose segments the origin describes.
465    pub trunk: Arc<Trunk>,
466    /// Which published segments the origin presents.
467    pub segments: Arc<dyn SegmentProvider>,
468    /// The program's current track set.
469    pub tracks: Arc<dyn TrackProvider>,
470}
471
472impl SmoothOriginConfig {
473    /// The origin's trunk, the segments it presents and the program's track set.
474    pub fn new(
475        trunk: Arc<Trunk>,
476        segments: Arc<dyn SegmentProvider>,
477        tracks: Arc<dyn TrackProvider>,
478    ) -> Self {
479        Self {
480            trunk,
481            segments,
482            tracks,
483        }
484    }
485}
486
487impl SmoothOrigin {
488    /// An origin over `config.trunk`; it describes whatever segments the
489    /// trunk currently holds, once an init segment is stored and at least one
490    /// track is announced.
491    pub fn new(config: SmoothOriginConfig) -> Self {
492        Self {
493            trunk: config.trunk,
494            init: Mutex::new(InitSlot::default()),
495            provider: config.segments,
496            tracks: config.tracks,
497            plan_cache: Mutex::new(None),
498            build_gate: Mutex::new(()),
499            fragments: Mutex::new(HashMap::new()),
500            #[cfg(test)]
501            plan_builds: std::sync::atomic::AtomicUsize::new(0),
502            #[cfg(test)]
503            fragment_builds: std::sync::atomic::AtomicUsize::new(0),
504        }
505    }
506
507    /// The `Trunk` this origin reads.
508    pub fn trunk(&self) -> &Arc<Trunk> {
509        &self.trunk
510    }
511
512    /// The streams this origin currently advertises, in manifest order — the
513    /// vocabulary [`SmoothRequest::parse`] needs to read a request path.
514    /// Empty while the trunk has nothing to describe.
515    pub fn stream_keys(&self) -> Vec<StreamKey> {
516        match self.snapshot() {
517            Ok(snapshot) => snapshot
518                .plan
519                .streams
520                .iter()
521                .map(|stream| stream.key)
522                .collect(),
523            Err(_) => Vec::new(),
524        }
525    }
526
527    /// Store the fMP4 init segment the trunk's segments are muxed with — the
528    /// one thing no `Trunk` ring holds, and what a fragment request is
529    /// re-demuxed against. Storing the segment already held changes nothing,
530    /// so a host may call this on every request without invalidating the
531    /// cached plan.
532    pub fn set_init(&self, bytes: impl Into<Bytes>) {
533        let bytes = bytes.into();
534        let mut slot = locked(&self.init);
535        if slot.bytes.as_ref() == Some(&bytes) {
536            return;
537        }
538        slot.bytes = Some(bytes);
539        slot.generation = slot.generation.saturating_add(1);
540    }
541
542    /// The init segment currently stored, if any.
543    pub fn init(&self) -> Option<Bytes> {
544        locked(&self.init).bytes.clone()
545    }
546
547    /// The fragments currently advertised, in manifest order, or `None` while
548    /// the trunk has nothing to advertise yet.
549    pub fn fragments(&self) -> Result<Vec<FragmentLocation>, SmoothOriginError> {
550        let snapshot = match self.snapshot() {
551            Ok(snapshot) => snapshot,
552            Err(Unavailable::Waiting { .. }) => return Ok(Vec::new()),
553            Err(Unavailable::Unusable(error)) => return Err(error),
554        };
555        Ok(snapshot
556            .plan
557            .streams
558            .iter()
559            .flat_map(|stream| {
560                stream.chunks.iter().map(|chunk| FragmentLocation {
561                    stream: stream.key,
562                    start: chunk.start,
563                    duration: chunk.duration,
564                })
565            })
566            .collect())
567    }
568
569    /// Render the live client Manifest, or why it cannot be rendered yet
570    /// (`EgressResponse::Await`) or at all (`EgressResponse::BadRequest`).
571    pub fn manifest(&self, now: Timestamp, await_policy: AwaitPolicy) -> EgressResponse<String> {
572        match self.snapshot() {
573            Ok(snapshot) => EgressResponse::Ready {
574                body: snapshot.plan.manifest.clone(),
575                cache: CachePolicy::NoCache,
576            },
577            Err(Unavailable::Waiting {
578                retry_no_earlier_than,
579            }) => EgressResponse::pending(await_policy, now, retry_no_earlier_than),
580            Err(Unavailable::Unusable(_)) => EgressResponse::BadRequest {
581                reason: UNUSABLE_MANIFEST,
582            },
583        }
584    }
585}
586
587fn locked<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
588    m.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
589}
590
591/// Why no plan could be built: the trunk has nothing to describe yet, or what
592/// it has cannot be described.
593enum Unavailable {
594    Waiting { retry_no_earlier_than: Timestamp },
595    Unusable(SmoothOriginError),
596}
597
598impl From<SmoothOriginError> for Unavailable {
599    fn from(error: SmoothOriginError) -> Self {
600        Unavailable::Unusable(error)
601    }
602}
603
604impl SmoothOrigin {
605    /// The plan for the window the host presents now: the cached one when
606    /// nothing it depends on has changed, otherwise a rebuilt one (built once
607    /// however many requests miss together).
608    fn snapshot(&self) -> Result<Snapshot, Unavailable> {
609        let waiting = || Unavailable::Waiting {
610            retry_no_earlier_than: Timestamp::from_nanos(0),
611        };
612        let (init, init_generation) = self.init_slot().ok_or_else(waiting)?;
613        let specs = self.tracks.tracks();
614        if specs.is_empty() {
615            return Err(waiting());
616        }
617        let segments = self.provider.segments();
618        if segments.is_empty() {
619            return Err(waiting());
620        }
621        let key = plan_key(init_generation, &specs, &segments);
622        let plan = match self.cached_plan(&key) {
623            Some(cached) => cached,
624            None => self.build(key, &specs, &init, &segments),
625        }?;
626        Ok(Snapshot {
627            plan,
628            init,
629            segments,
630        })
631    }
632
633    fn init_slot(&self) -> Option<(Bytes, u64)> {
634        let slot = locked(&self.init);
635        slot.bytes.clone().map(|bytes| (bytes, slot.generation))
636    }
637
638    fn cached_plan(&self, key: &PlanKey) -> Option<Result<Arc<Plan>, SmoothOriginError>> {
639        locked(&self.plan_cache)
640            .as_ref()
641            .filter(|cached| cached.key == *key)
642            .map(|cached| cached.plan.clone())
643    }
644
645    fn build(
646        &self,
647        key: PlanKey,
648        specs: &[TrackSpec],
649        init: &[u8],
650        segments: &[(SegmentSeq, Bytes)],
651    ) -> Result<Arc<Plan>, SmoothOriginError> {
652        let _gate = locked(&self.build_gate);
653        if let Some(cached) = self.cached_plan(&key) {
654            return cached;
655        }
656        #[cfg(test)]
657        self.plan_builds
658            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
659        let built = build_plan(specs, init, segments).map(Arc::new);
660        self.retain_fragments_in(built.as_deref().ok());
661        *locked(&self.plan_cache) = Some(CachedPlan {
662            key,
663            plan: built.clone(),
664        });
665        built
666    }
667
668    /// Drop every packaged fragment the new plan's window no longer holds, so
669    /// the fragment cache never outgrows the window.
670    fn retain_fragments_in(&self, plan: Option<&Plan>) {
671        locked(&self.fragments).retain(|key, _| plan.is_some_and(|plan| key.is_in(plan)));
672    }
673}
674
675fn plan_key(
676    init_generation: u64,
677    specs: &[TrackSpec],
678    segments: &[(SegmentSeq, Bytes)],
679) -> PlanKey {
680    PlanKey {
681        init_generation,
682        tracks: specs.to_vec(),
683        window: segments.iter().map(|(sequence, _)| *sequence).collect(),
684    }
685}
686
687/// Resolve `specs` and the window's `segments` into the live plan.
688fn build_plan(
689    specs: &[TrackSpec],
690    init: &[u8],
691    segments: &[(SegmentSeq, Bytes)],
692) -> Result<Plan, SmoothOriginError> {
693    let mut ordinals: Vec<(SmoothStreamType, u32)> = Vec::new();
694    let mut streams: Vec<StreamPlan> = Vec::new();
695    for spec in specs {
696        let codec = resolve_codec(&spec.config).map_err(|error| SmoothOriginError::Codec {
697            track: spec.track_id.value(),
698            reason: error.to_string(),
699        })?;
700        let key = StreamKey {
701            kind: codec.stream_type,
702            ordinal: match ordinals
703                .iter_mut()
704                .find(|(kind, _)| *kind == codec.stream_type)
705            {
706                Some((_, count)) => {
707                    *count = count.saturating_add(1);
708                    *count
709                }
710                None => {
711                    let first = FIRST_STREAM_ORDINAL as u32;
712                    ordinals.push((codec.stream_type, first));
713                    first
714                }
715            },
716        };
717        let chunks = chunk_timeline(init, segments, spec, key)?;
718        if chunks.is_empty() {
719            continue;
720        }
721        streams.push(StreamPlan {
722            key,
723            codec,
724            interleaved_ordinal: spec.track_id.value().saturating_sub(FIRST_TRACK_ID) + 1,
725            chunks,
726        });
727    }
728    if streams.is_empty() {
729        return Err(SmoothOriginError::Codec {
730            track: 0,
731            reason: "the window carries none of the trunk's tracks".to_string(),
732        });
733    }
734    let mut plan = Plan {
735        manifest: String::new(),
736        streams,
737    };
738    plan.manifest = plan.manifest().to_xml();
739    Ok(plan)
740}
741
742/// Track ids are 1-based, so the first interleaved track is ordinal 1.
743const FIRST_TRACK_ID: u32 = 1;
744
745/// One track's chunk timeline over the window, in the track's own timeline:
746/// each chunk starts where the previous one ended, and the first starts at its
747/// segment's own absolute first-sample time, so the timeline is stable as the
748/// window slides.
749fn chunk_timeline(
750    init: &[u8],
751    segments: &[(SegmentSeq, Bytes)],
752    spec: &TrackSpec,
753    key: StreamKey,
754) -> Result<Vec<ChunkTiming>, SmoothOriginError> {
755    let packager = SmoothPackager {
756        target_duration_secs: SINGLE_SEGMENT_FRAGMENT_TARGET_SECS,
757    };
758    let mut chunks: Vec<ChunkTiming> = Vec::with_capacity(segments.len());
759    for (sequence, bytes) in segments {
760        let media = demux(init, bytes).map_err(|reason| SmoothOriginError::Codec {
761            track: spec.track_id.value(),
762            reason,
763        })?;
764        let Some(track) = find_track(&media, key) else {
765            continue;
766        };
767        let Some(first) = track.samples.first() else {
768            continue;
769        };
770        // A sample without a `dts` starts at 0; a negative one (only a
771        // rebased B-frame lead-in) has no earlier position to report.
772        let timescale = track.spec.timescale.max(1);
773        let decode_time = first.dts.unwrap_or(0).max(0);
774        let absolute = u64::try_from(decode_time).unwrap_or(u64::MAX);
775        let start = chunks
776            .last()
777            .map(|chunk: &ChunkTiming| chunk.start.saturating_add(chunk.duration))
778            .unwrap_or_else(|| to_smooth_ticks(absolute, timescale));
779        let sequence_number = sequence.value();
780        // The chunk's duration is exactly the fragment's own `trun` sum, so
781        // `Σtrun == c@d` holds by construction; a track the packager refuses
782        // contributes no chunk rather than an empty entry with `d="0"`.
783        let Ok(fragment) = packager.package_track_fragment(track, sequence_number, start) else {
784            continue;
785        };
786        let duration = fragment.duration;
787        chunks.push(ChunkTiming {
788            sequence: *sequence,
789            start,
790            duration,
791        });
792    }
793    Ok(chunks)
794}
795
796/// Demux one init + media segment, rendering the demuxer's error.
797fn demux(init: &[u8], segment: &[u8]) -> Result<Media, String> {
798    let mut combined = Vec::with_capacity(init.len() + segment.len());
799    combined.extend_from_slice(init);
800    combined.extend_from_slice(segment);
801    Fmp4Demux::new()
802        .unpackage(combined.as_slice())
803        .map_err(|error| error.to_string())
804}
805
806/// The demuxed track a [`StreamKey`] names: the `ordinal`-th track of its kind
807/// in the segment's own interleaved track order.
808fn find_track(media: &Media, key: StreamKey) -> Option<&Track> {
809    media
810        .tracks
811        .iter()
812        .filter(|track| stream_kind(&track.spec.config) == Some(key.kind))
813        .nth(usize::try_from(key.ordinal.checked_sub(1)?).ok()?)
814}
815
816/// The stream kind a track's codec maps to, or `None` for a codec MS-SSTR (as
817/// implemented here) cannot carry.
818fn stream_kind(config: &transmux::CodecConfig) -> Option<SmoothStreamType> {
819    match config {
820        transmux::CodecConfig::Avc { .. } => Some(SmoothStreamType::Video),
821        transmux::CodecConfig::Aac { .. } => Some(SmoothStreamType::Audio),
822        transmux::CodecConfig::Subtitle {
823            sample_entry: Some(_),
824            ..
825        } => Some(SmoothStreamType::Text),
826        _ => None,
827    }
828}
829
830impl ServedEgress for SmoothOrigin {
831    type Request = SmoothRequest;
832    type Body = SmoothBody;
833
834    fn resolve(
835        &self,
836        request: SmoothRequest,
837        now: Timestamp,
838        await_policy: AwaitPolicy,
839    ) -> EgressResponse<SmoothBody> {
840        self.answer(request, now, await_policy)
841    }
842
843    fn is_warm(&self, request: &SmoothRequest) -> bool {
844        let Some((_, init_generation)) = self.init_slot() else {
845            return true;
846        };
847        let specs = self.tracks.tracks();
848        let segments = self.provider.segments();
849        if specs.is_empty() || segments.is_empty() {
850            return true;
851        }
852        let Some(cached) = self.cached_plan(&plan_key(init_generation, &specs, &segments)) else {
853            return false;
854        };
855        let (Ok(plan), SmoothRequest::Fragment { stream, start, .. }) = (cached, request) else {
856            return true;
857        };
858        let Some((_, chunk)) = plan.chunk(*stream, *start) else {
859            return true;
860        };
861        locked(&self.fragments)
862            .get(&FragmentKey::new(*stream, &chunk))
863            .is_some_and(|slot| slot.get().is_some())
864    }
865
866    fn prepare(&self, request: SmoothRequest) {
867        drop(self.answer(
868            request,
869            Timestamp::from_nanos(0),
870            AwaitPolicy::new(Timestamp::from_nanos(0)),
871        ));
872    }
873}
874
875impl SmoothOrigin {
876    fn answer(
877        &self,
878        request: SmoothRequest,
879        now: Timestamp,
880        await_policy: AwaitPolicy,
881    ) -> EgressResponse<SmoothBody> {
882        let snapshot = match self.snapshot() {
883            Ok(snapshot) => snapshot,
884            Err(Unavailable::Waiting {
885                retry_no_earlier_than,
886            }) => return EgressResponse::pending(await_policy, now, retry_no_earlier_than),
887            Err(Unavailable::Unusable(_)) => {
888                return EgressResponse::BadRequest {
889                    reason: UNUSABLE_MANIFEST,
890                };
891            }
892        };
893        match request {
894            SmoothRequest::Manifest => EgressResponse::Ready {
895                body: SmoothBody::Manifest(snapshot.plan.manifest.clone()),
896                cache: CachePolicy::NoCache,
897            },
898            SmoothRequest::Fragment {
899                stream,
900                bitrate,
901                start,
902            } => match snapshot.plan.chunk(stream, start) {
903                Some((stream, chunk)) if bitrate == advertised_bitrate(&stream.codec) => {
904                    match self.fragment(&stream, chunk, &snapshot) {
905                        Ok(Some(bytes)) => EgressResponse::Ready {
906                            body: SmoothBody::Fragment(bytes),
907                            cache: CachePolicy::Immutable,
908                        },
909                        Ok(None) => EgressResponse::NotFound,
910                        Err(_) => EgressResponse::BadRequest {
911                            reason: UNUSABLE_FRAGMENT,
912                        },
913                    }
914                }
915                _ => EgressResponse::NotFound,
916            },
917        }
918    }
919}
920
921impl SmoothOrigin {
922    /// Build the fragment for `chunk` of `stream` from the window segment it
923    /// came from: demux, take the track, and re-mux one fragment whose `tfxd`
924    /// is the manifest's `c@t`.
925    ///
926    /// `Ok(None)` is a fragment the window no longer holds: its segment left
927    /// between describing the window and reading it back, which is the same
928    /// "gone" answer as a start time outside it, not a broken request.
929    fn fragment(
930        &self,
931        stream: &StreamPlan,
932        chunk: ChunkTiming,
933        snapshot: &Snapshot,
934    ) -> Result<Option<Bytes>, SmoothOriginError> {
935        let Some((_, bytes)) = snapshot
936            .segments
937            .iter()
938            .find(|(sequence, _)| *sequence == chunk.sequence)
939        else {
940            return Ok(None);
941        };
942        let key = FragmentKey::new(stream.key, &chunk);
943        let slot = Arc::clone(locked(&self.fragments).entry(key).or_default());
944        slot.get_or_init(|| {
945            #[cfg(test)]
946            self.fragment_builds
947                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
948            package_fragment(stream, &chunk, &snapshot.init, bytes)
949        })
950        .clone()
951        .map(Some)
952    }
953}
954
955/// Demux `bytes` and re-mux the `chunk` of `stream` as one fragment whose
956/// `tfxd` is the manifest's `c@t`.
957fn package_fragment(
958    stream: &StreamPlan,
959    chunk: &ChunkTiming,
960    init: &[u8],
961    bytes: &[u8],
962) -> Result<Bytes, SmoothOriginError> {
963    let media = demux(init, bytes).map_err(|reason| SmoothOriginError::Codec {
964        track: stream.interleaved_ordinal,
965        reason,
966    })?;
967    let track = find_track(&media, stream.key).ok_or(SmoothOriginError::Codec {
968        track: stream.interleaved_ordinal,
969        reason: "the segment does not carry the requested stream".to_string(),
970    })?;
971    let fragment = SmoothPackager {
972        target_duration_secs: SINGLE_SEGMENT_FRAGMENT_TARGET_SECS,
973    }
974    .package_track_fragment(track, chunk.sequence.value(), chunk.start)
975    .map_err(|error| SmoothOriginError::Codec {
976        track: stream.interleaved_ordinal,
977        reason: error.to_string(),
978    })?;
979    Ok(Bytes::from(fragment.data))
980}
981
982impl SmoothRequest {
983    /// Parse a request from a fragment path or its file name
984    /// (`QualityLevels(BITRATE)/Fragments(KIND=START)`, MS-SSTR §2.2.3), or a
985    /// Manifest file name. `kinds` supplies the advertised stream ordinals the
986    /// `KIND` token may name (`video`, `audio`, `text`, and `TYPE_N`). `None`
987    /// for an unrecognised path.
988    pub fn parse(path: &str, kinds: &[StreamKey]) -> Option<SmoothRequest> {
989        let manifest_name = path.rsplit('/').find(|segment| !segment.is_empty())?;
990        if manifest_name == MANIFEST_FILE_NAME {
991            return Some(SmoothRequest::Manifest);
992        }
993        // The URL template has two path segments: `QualityLevels(N)` and
994        // `Fragments(KIND=START)` (relative to the Manifest's own directory).
995        let mut trailing = path
996            .rsplit('/')
997            .filter(|segment| !segment.is_empty())
998            .take(2);
999        let fragment = trailing.next()?;
1000        let quality = trailing.next()?.strip_prefix(QUALITY_LEVELS_PREFIX)?;
1001        let bitrate = quality.strip_suffix(')')?.parse().ok()?;
1002        let (kind, start) = fragment.strip_prefix(FRAGMENTS_SEGMENT)?.split_once('=')?;
1003        let start = start.strip_suffix(')')?.parse().ok()?;
1004        let stream = kinds
1005            .iter()
1006            .copied()
1007            .find(|key| key.name() == kind)
1008            .or_else(|| kinds.iter().copied().find(|key| key.kind.name() == kind))?;
1009        Some(SmoothRequest::Fragment {
1010            stream,
1011            bitrate,
1012            start,
1013        })
1014    }
1015}
1016
1017/// The bad-request reason a manifest answers when a track it must describe has
1018/// no usable Smooth codec description.
1019const UNUSABLE_MANIFEST: &str =
1020    "a track has no usable Smooth Streaming codec description; the Manifest cannot be rendered";
1021
1022/// The bad-request reason a fragment answers when its segment cannot be
1023/// re-muxed into the requested stream's fragment.
1024const UNUSABLE_FRAGMENT: &str = "the requested fragment cannot be built from the live window";
1025
1026/// The file name of the client Manifest (MS-SSTR §2.2.1).
1027pub const MANIFEST_FILE_NAME: &str = "Manifest";
1028
1029#[cfg(test)]
1030mod tests {
1031    use std::num::NonZeroUsize;
1032    use std::time::Duration;
1033
1034    use broadcast_smooth::StreamType;
1035    use media_plane::TrunkConfig;
1036    use media_plane::trunk::SegmentEntry;
1037    use transmux::SegmentMeta;
1038    use transmux::avc_config_from_sprop;
1039    use transmux::ll_hls::LlHlsSegmenter;
1040    use transmux::pipeline::Sample;
1041    use transmux::smooth_fragment::fragment_tfxd;
1042    use transmux::{
1043        AVCConfigurationBox, AudioSpecificConfig, CodecConfig, DecoderConfigDescriptor,
1044        DecoderSpecificInfo, ESDescriptor, EsdsBox, SLConfigDescriptor, TrackId, TrackSpec,
1045    };
1046
1047    use super::*;
1048
1049    /// The SPS/PPS pair of a real H.264 stream, as an `sprop-parameter-sets`
1050    /// value: its SPS decodes to 640x480.
1051    const SPROP: &str = "Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==";
1052    /// The SPS's coded width (the `sprop-parameter-sets` stream is 640x360).
1053    const SPROP_WIDTH: u32 = 640;
1054    /// The SPS's coded height.
1055    const SPROP_HEIGHT: u32 = 480;
1056    /// Video timescale: the H.264 clock.
1057    const VIDEO_TIMESCALE: u32 = 90_000;
1058    /// A 30 fps frame duration in `VIDEO_TIMESCALE` ticks.
1059    const FRAME_DUR: u32 = VIDEO_TIMESCALE / 30;
1060    /// Frames per segment: two seconds of video.
1061    const FRAMES_PER_SEGMENT: u32 = 60;
1062    /// Repeat requests the cache tests make over one window.
1063    const REQUESTS: usize = 8;
1064    /// Segments published per test route.
1065    const SEGMENTS: u32 = 2;
1066    /// The AAC media timescale, in ticks per second.
1067    const AUDIO_TIMESCALE: u32 = 48_000;
1068    /// AAC-LC frame duration, in `AUDIO_TIMESCALE` ticks (1024 samples).
1069    const AAC_FRAME_DUR: u32 = 1024;
1070    /// Sample rate the AAC config reports.
1071    const SAMPLE_RATE: u32 = 48_000;
1072    /// Channel count the AAC config reports.
1073    const CHANNELS: u16 = 2;
1074    /// An AAC-LC `AudioSpecificConfig` for 48 kHz stereo (`0x11 0x90`).
1075    const ASC_48K_STEREO: [u8; 2] = [0x11, 0x90];
1076    /// `SLConfigDescriptor@predefined` 2: the MP4-storage form (ISO/IEC
1077    /// 14496-14 §3.1.2).
1078    const SL_CONFIG_PREDEFINED_MP4: u8 = 2;
1079    /// The `stpp` namespace a TTML track carries.
1080    const TTML_NAMESPACE: &str = "http://www.w3.org/ns/ttml";
1081    /// The `sprop-parameter-sets` the golden Manifest is rendered from: a
1082    /// Baseline-profile SPS that decodes to the 1280x720 the golden records,
1083    /// plus its PPS.
1084    ///
1085    /// The golden's own `CodecPrivateData` records a truncated 4-byte SPS
1086    /// (`00000001674D401F`), a placeholder an earlier hand-built config_
1087    /// carried rather than a decodable record, so it cannot be the source of
1088    /// the geometry. The same geometry cannot be reached by editing it either:
1089    /// the size fields are `pic_width_in_mbs_minus1 = 79` and
1090    /// `pic_height_in_map_units_minus1 = 44`, which need more than the 17 and
1091    /// 16 bits a 4-byte SPS leaves. The values below are the ones this pair
1092    /// actually produces.
1093    const GOLDEN_SPROP: &str = "Z0LAH/QCgC3AgA==,aO48gA==";
1094    /// Segment duration published to the trunk.
1095    const SEGMENT_DURATION: Duration = Duration::from_secs(2);
1096    /// [`SEGMENT_DURATION`] in nanoseconds, for the trunk's timeline position.
1097    const SEGMENT_NANOS: u64 = 2_000_000_000;
1098
1099    /// An `avcC` box from the real SPS/PPS pair above.
1100    fn avc_config() -> AVCConfigurationBox {
1101        avc_config_from_sprop(SPROP).expect("valid sprop")
1102    }
1103
1104    /// A video track (id `1`) whose `CodecConfig` reports **no** geometry — the
1105    /// shape RTP/WHEP ingest builds before the SPS is known.
1106    fn video_spec() -> TrackSpec {
1107        TrackSpec::new(
1108            TrackId::new(1),
1109            VIDEO_TIMESCALE,
1110            CodecConfig::Avc {
1111                config: avc_config(),
1112                width: 0,
1113                height: 0,
1114            },
1115        )
1116    }
1117
1118    /// An `esds` carrying `dsi` as its DecoderSpecificInfo, or none when `dsi`
1119    /// is `None`.
1120    fn esds(dsi: Option<Vec<u8>>) -> EsdsBox {
1121        let decoder = DecoderConfigDescriptor::new(
1122            0x40,
1123            0x15,
1124            false,
1125            0,
1126            0,
1127            0,
1128            dsi.map(DecoderSpecificInfo::new),
1129        );
1130        EsdsBox::new(ESDescriptor::new(
1131            1,
1132            0,
1133            Some(decoder),
1134            Some(SLConfigDescriptor::new(vec![SL_CONFIG_PREDEFINED_MP4])),
1135        ))
1136    }
1137
1138    /// An AAC track with the given id.
1139    fn audio_spec(track_id: u32) -> TrackSpec {
1140        TrackSpec::new(
1141            TrackId::new(track_id),
1142            AUDIO_TIMESCALE,
1143            CodecConfig::Aac {
1144                esds: esds(Some(ASC_48K_STEREO.to_vec())),
1145                sample_rate: SAMPLE_RATE,
1146                channel_count: CHANNELS,
1147                sample_size: AAC_BITS_PER_SAMPLE,
1148            },
1149        )
1150    }
1151
1152    /// A TTML (`stpp`) subtitle track with the given id.
1153    fn subtitle_spec(track_id: u32) -> TrackSpec {
1154        TrackSpec::new(
1155            TrackId::new(track_id),
1156            1000,
1157            CodecConfig::Subtitle {
1158                format: transmux::ir::SubtitleFormat::Ttml,
1159                sample_entry: Some(transmux::pipeline::SubtitleSampleEntry::Ttml(
1160                    transmux::subtitle_entries::XmlSubtitleSampleEntry::new(TTML_NAMESPACE),
1161                )),
1162            },
1163        )
1164    }
1165
1166    fn trunk() -> Arc<Trunk> {
1167        Trunk::new(
1168            TrunkConfig::default()
1169                .with_timed_capacity(NonZeroUsize::new(8).expect("non-zero"))
1170                .with_sparse_capacity(NonZeroUsize::new(8).expect("non-zero"))
1171                .with_segment_capacity(NonZeroUsize::new(8).expect("non-zero"))
1172                .with_event_capacity(NonZeroUsize::new(8).expect("non-zero"))
1173                .with_part_capacity(NonZeroUsize::new(8).expect("non-zero")),
1174        )
1175    }
1176
1177    /// A trunk whose segment log holds every test segment.
1178    fn big_trunk() -> Arc<Trunk> {
1179        Trunk::new(
1180            TrunkConfig::default()
1181                .with_timed_capacity(NonZeroUsize::new(64).expect("non-zero"))
1182                .with_sparse_capacity(NonZeroUsize::new(64).expect("non-zero"))
1183                .with_segment_capacity(NonZeroUsize::new(64).expect("non-zero"))
1184                .with_event_capacity(NonZeroUsize::new(64).expect("non-zero"))
1185                .with_part_capacity(NonZeroUsize::new(64).expect("non-zero")),
1186        )
1187    }
1188
1189    /// Everything a published route holds: the trunk and its init segment.
1190    struct Route {
1191        trunk: Arc<Trunk>,
1192        init: Vec<u8>,
1193        published: Vec<(SegmentSeq, Bytes)>,
1194        specs: Vec<TrackSpec>,
1195    }
1196
1197    impl Route {
1198        /// The track specs this route published.
1199        fn specs(&self) -> Vec<TrackSpec> {
1200            self.specs.clone()
1201        }
1202
1203        /// The provider handing back this route's published segments.
1204        fn published(&self) -> impl Fn() -> Vec<(SegmentSeq, Bytes)> + Send + Sync + 'static {
1205            let published = self.published.clone();
1206            move || published.clone()
1207        }
1208    }
1209
1210    /// Publish `specs` through a real LL-HLS segmenter: `SEGMENTS` two-second
1211    /// segments of video and/or audio, and one subtitle sample per video frame.
1212    fn publish(specs: Vec<TrackSpec>) -> Route {
1213        let trunk = big_trunk();
1214        let mut segmenter =
1215            LlHlsSegmenter::with_part_target(specs.clone(), transmux::VIDEO_CLOCK_RATE, 2.0, 500)
1216                .expect("segmenter builds");
1217        let truncated = trunk.writer().expect("fresh trunk");
1218        truncated.set_tracks(specs.clone());
1219        let writer = trunk.segment_writer().expect("fresh trunk");
1220        let init = segmenter.init_segment().expect("init segment");
1221        let mut published = 0u32;
1222        let mut window: Vec<(SegmentSeq, Bytes)> = Vec::new();
1223        for index in 0..SEGMENTS * FRAMES_PER_SEGMENT {
1224            let dts = i64::from(index) * i64::from(FRAME_DUR);
1225            for spec in &specs {
1226                let sample = match spec.config {
1227                    CodecConfig::Avc { .. } => Sample::new(
1228                        vec![0xAA_u8.wrapping_add((index % 251) as u8); 64],
1229                        Some(dts),
1230                        Some(dts),
1231                        Some(FRAME_DUR),
1232                        index % FRAMES_PER_SEGMENT == 0,
1233                    ),
1234                    CodecConfig::Aac { .. } => {
1235                        let dts = i64::from(index) * i64::from(AAC_FRAME_DUR);
1236                        Sample::new(
1237                            vec![0x5A; 32],
1238                            Some(dts),
1239                            Some(dts),
1240                            Some(AAC_FRAME_DUR),
1241                            true,
1242                        )
1243                    }
1244                    _ => Sample::new(vec![0x54; 8], Some(dts), Some(dts), Some(FRAME_DUR), true),
1245                };
1246                segmenter.push(spec.track_id, sample).expect("push");
1247            }
1248            for segment in segmenter.take_ready_segments() {
1249                publish_segment(&writer, segment, &mut published, &mut window);
1250            }
1251        }
1252        segmenter.flush().expect("flush");
1253        for segment in segmenter.take_ready_segments() {
1254            publish_segment(&writer, segment, &mut published, &mut window);
1255        }
1256        Route {
1257            trunk,
1258            init,
1259            published: window,
1260            specs,
1261        }
1262    }
1263
1264    /// Publish one closed segment to the trunk, numbered in publication order.
1265    fn publish_segment(
1266        writer: &media_plane::trunk::SegmentWriter,
1267        segment: transmux::ll_hls::SegmentInfo,
1268        published: &mut u32,
1269        window: &mut Vec<(SegmentSeq, Bytes)>,
1270    ) {
1271        let bytes = segment.bytes.clone();
1272        writer
1273            .publish_segment(SegmentEntry::new(
1274                segment.bytes,
1275                SegmentSeq::new(*published + 1),
1276                SEGMENT_DURATION,
1277                Timestamp::from_nanos(u64::from(*published) * SEGMENT_NANOS),
1278                SegmentMeta {
1279                    discontinuous: false,
1280                },
1281            ))
1282            .expect("publish segment");
1283        window.push((SegmentSeq::new(*published + 1), Bytes::from(bytes)));
1284        *published += 1;
1285    }
1286
1287    /// The manifest a fresh origin renders for `route`.
1288    fn manifest(route: &Route) -> SmoothManifest {
1289        let origin = origin(route);
1290        let response = origin.manifest(
1291            Timestamp::from_nanos(0),
1292            AwaitPolicy::new(Timestamp::from_nanos(u64::MAX)),
1293        );
1294        match response {
1295            EgressResponse::Ready { body, .. } => {
1296                SmoothManifest::parse(&body).expect("manifest parses")
1297            }
1298            other => panic!("expected a rendered manifest, got {other:?}"),
1299        }
1300    }
1301
1302    /// An origin over `route` with its init segment stored.
1303    fn origin(route: &Route) -> SmoothOrigin {
1304        let origin = SmoothOrigin::new(SmoothOriginConfig::new(
1305            Arc::clone(&route.trunk),
1306            Arc::new(route.published()),
1307            Arc::new({
1308                let specs = route.specs();
1309                move || specs.clone()
1310            }),
1311        ));
1312        origin.set_init(route.init.clone());
1313        origin
1314    }
1315
1316    /// The advertised streams' keys, in manifest order.
1317    fn keys(route: &Route) -> Vec<StreamKey> {
1318        let origin = origin(route);
1319        origin
1320            .fragments()
1321            .expect("fragments")
1322            .iter()
1323            .fold(Vec::new(), |mut acc, fragment| {
1324                if !acc.contains(&fragment.stream) {
1325                    acc.push(fragment.stream);
1326                }
1327                acc
1328            })
1329    }
1330
1331    /// An empty trunk answers `Await` until the caller's policy expires, then
1332    /// `NotFound` — never a fabricated manifest.
1333    #[test]
1334    fn an_empty_origin_waits_then_gives_up() {
1335        let origin = SmoothOrigin::new(SmoothOriginConfig::new(
1336            trunk(),
1337            Arc::new(Vec::new),
1338            Arc::new(Vec::new),
1339        ));
1340        let policy = AwaitPolicy::new(Timestamp::from_nanos(10));
1341        assert!(matches!(
1342            origin.manifest(Timestamp::from_nanos(0), policy),
1343            EgressResponse::Await { .. }
1344        ));
1345        assert!(matches!(
1346            origin.manifest(Timestamp::from_nanos(10), policy),
1347            EgressResponse::NotFound
1348        ));
1349    }
1350
1351    /// Byte-for-byte golden: the rendered live Manifest equals the file in
1352    /// `fixtures/golden/smooth/`, which `multimux` recorded from the live
1353    /// output `origin/main` hand-wrote before this engine existed (see that
1354    /// directory's `PROVENANCE.md`). The window is that test's own: a
1355    /// H.264 1280x720 + AAC-LC 48 kHz stereo track pair over three and five
1356    /// chunks.
1357    ///
1358    /// The golden is compared as a `(name, value)` map of bytes per element
1359    /// (a multi-line `Prop` per line):
1360    ///
1361    /// 1. `(element, Some(attribute name), value)` for every attribute, in
1362    ///    document order;
1363    /// 2. `(element, None, text)` for every leaf element carrying text
1364    ///    (`Prop`), in document order;
1365    /// 3. `("", None, line)` for every non-blank line outside a `Prop`.
1366    ///
1367    /// The first two sequences must be **identical** to the golden's: not a
1368    /// single attribute value or name may change. The third compares the
1369    /// document's structure and every literal line (the XML declaration, each
1370    /// indented open/empty/close tag) but normalises a `Prop` to a single line
1371    /// and treats an escaped character (`&`, `<`, `>`, `"`, `\'`) as equal to
1372    /// the character it stands for — `broadcast-smooth`'s writer chooses
1373    /// those spellings, and the round-trip test
1374    /// ([`the_rendered_manifest_round_trips`]) pins them instead.
1375    #[test]
1376    fn render_manifest_matches_golden() {
1377        fn chunks(count: u64, duration: u64) -> Vec<ChunkTiming> {
1378            (0..count)
1379                .map(|i| ChunkTiming {
1380                    sequence: SegmentSeq::new(u32::try_from(i + 1).expect("sequence fits")),
1381                    start: i * duration,
1382                    duration,
1383                })
1384                .collect()
1385        }
1386        // `SmoothCodec` is `#[non_exhaustive]`: build the two tracks the
1387        // golden records through the real resolver, which is also what keeps
1388        // the test honest about where each manifest value comes from.
1389        let video = resolve_codec(&CodecConfig::Avc {
1390            config: avc_config_from_sprop(GOLDEN_SPROP).expect("valid sprop"),
1391            width: 1280,
1392            height: 720,
1393        })
1394        .expect("AVC resolves");
1395        let audio = resolve_codec(&CodecConfig::Aac {
1396            esds: esds(Some(ASC_48K_STEREO.to_vec())),
1397            sample_rate: SAMPLE_RATE,
1398            channel_count: CHANNELS,
1399            sample_size: AAC_BITS_PER_SAMPLE,
1400        })
1401        .expect("AAC resolves");
1402        let layouts = [
1403            (
1404                "video-audio",
1405                Plan {
1406                    manifest: String::new(),
1407                    streams: vec![
1408                        StreamPlan {
1409                            key: StreamKey {
1410                                kind: SmoothStreamType::Video,
1411                                ordinal: 1,
1412                            },
1413                            codec: video,
1414                            interleaved_ordinal: 1,
1415                            chunks: chunks(3, 20_000_000),
1416                        },
1417                        StreamPlan {
1418                            key: StreamKey {
1419                                kind: SmoothStreamType::Audio,
1420                                ordinal: 1,
1421                            },
1422                            codec: audio,
1423                            interleaved_ordinal: 2,
1424                            chunks: chunks(5, 21_333_333),
1425                        },
1426                    ],
1427                },
1428            ),
1429            (
1430                "empty",
1431                Plan {
1432                    manifest: String::new(),
1433                    streams: Vec::new(),
1434                },
1435            ),
1436        ];
1437        for (name, layout) in layouts {
1438            let actual = layout.manifest().to_xml();
1439            let file = format!("smooth-{name}.xml");
1440            let path = test_fixtures::path(&format!("golden/smooth/{file}"));
1441            let expected = std::fs::read_to_string(&path).expect("read golden");
1442            assert_manifest_matches(&actual, &expected, &file);
1443        }
1444    }
1445
1446    /// One parsed manifest: its elements and attributes in document order.
1447    struct ManifestDocument {
1448        elements: Vec<(String, Option<String>, String)>,
1449    }
1450
1451    /// Parse `xml` into [`ManifestDocument`], requiring it to be well formed.
1452    fn parse_manifest(xml: &str) -> ManifestDocument {
1453        use quick_xml::XmlVersion;
1454        use quick_xml::events::Event;
1455        use quick_xml::reader::Reader;
1456
1457        let mut reader = Reader::from_str(xml);
1458        let mut elements: Vec<(String, Option<String>, String)> = Vec::new();
1459        loop {
1460            match reader.read_event().expect("well-formed manifest") {
1461                Event::Start(e) | Event::Empty(e) => {
1462                    let name = e.name().into_inner().to_string();
1463                    for attribute in e.attributes() {
1464                        let attribute = attribute.expect("attribute");
1465                        elements.push((
1466                            name.clone(),
1467                            Some(attribute.key.into_inner().to_string()),
1468                            attribute
1469                                .normalized_value(XmlVersion::Implicit1_0)
1470                                .expect("attribute value")
1471                                .into_owned(),
1472                        ));
1473                    }
1474                }
1475                Event::Text(e) => {
1476                    let text = e.into_inner();
1477                    if !text.trim().is_empty() {
1478                        elements.push(("Prop".to_string(), None, text.trim().to_string()));
1479                    }
1480                }
1481                Event::Eof => break,
1482                _ => {}
1483            }
1484        }
1485        ManifestDocument { elements }
1486    }
1487
1488    /// A value the golden records and the value this origin produces for the
1489    /// same attribute, with the reason the difference is intended. Every
1490    /// `QualityLevel` attribute is scoped to its stream.
1491    struct GoldenDifference {
1492        element: &'static str,
1493        attribute: &'static str,
1494        stream: Option<&'static str>,
1495        /// The golden's value, or `None` when the golden does not record this
1496        /// attribute for that stream at all.
1497        golden: Option<&'static str>,
1498        /// This origin's value, or `None` when it does not emit it.
1499        ours: Option<&'static str>,
1500        reason: &'static str,
1501    }
1502
1503    /// Every intended difference between this origin's live Manifest and the
1504    /// golden `origin/main` recorded. Nothing else may differ, and nothing the
1505    /// golden records may be absent.
1506    const GOLDEN_DIFFERENCES: &[GoldenDifference] = &[
1507        GoldenDifference {
1508            element: "SmoothStreamingMedia",
1509            attribute: "IsLive",
1510            stream: None,
1511            golden: Some("true"),
1512            ours: Some("TRUE"),
1513            reason: "broadcast-smooth's writer spells IsLive=\"TRUE\", the value its own parser accepts",
1514        },
1515        GoldenDifference {
1516            element: "StreamIndex",
1517            attribute: "Subtype",
1518            stream: Some("video"),
1519            golden: Some(""),
1520            ours: None,
1521            reason: "the old writer always emitted Subtype=\"\"; broadcast-smooth emits it only where MS-SSTR requires it (a text stream)",
1522        },
1523        GoldenDifference {
1524            element: "StreamIndex",
1525            attribute: "Subtype",
1526            stream: Some("audio"),
1527            golden: Some(""),
1528            ours: None,
1529            reason: "as for video: MS-SSTR §2.2.2.3 makes Subtype a text-stream attribute",
1530        },
1531        GoldenDifference {
1532            element: "QualityLevel",
1533            attribute: "Bitrate",
1534            stream: Some("video"),
1535            golden: Some("2500000"),
1536            ours: Some("3686400"),
1537            reason: "a live window has no per-track byte count, so the origin advertises a resolution-scaled estimate (width x height x 4 bits/pixel), clamped 0.5-20 Mbps; the old output estimated from the same geometry with different constants",
1538        },
1539        GoldenDifference {
1540            element: "QualityLevel",
1541            attribute: "Bitrate",
1542            stream: Some("audio"),
1543            golden: Some("128000"),
1544            ours: Some("320000"),
1545            reason: "same estimate: sampling_rate x channels x 16 bits (48 kHz x 2 x 16 = 1.536 Mbps) clamped to 320 kbps, the band's ceiling",
1546        },
1547        GoldenDifference {
1548            element: "QualityLevel",
1549            attribute: "CodecPrivateData",
1550            stream: Some("video"),
1551            golden: Some("00000001674D401F0000000168EE3C80"),
1552            ours: Some("000000016742C01FF402802DC0800000000168EE3C80"),
1553            reason: "the golden's SPS is a truncated 4-byte placeholder that decodes to no geometry; the origin's Annex-B bytes and its 1280x720 geometry both come from the real SPS at GOLDEN_SPROP",
1554        },
1555    ];
1556
1557    /// Assert `actual` matches `expected` — see [`render_manifest_matches_golden`]
1558    /// for exactly what is and is not compared.
1559    fn assert_manifest_matches(actual: &str, expected: &str, file: &str) {
1560        let got = parse_manifest(actual);
1561        let want = parse_manifest(expected);
1562        let intended = |element: &str, attribute: &str, stream: Option<&str>| {
1563            GOLDEN_DIFFERENCES
1564                .iter()
1565                .find(|d| d.element == element && d.attribute == attribute && d.stream == stream)
1566        };
1567        /// One attribute occurrence, with the stream-kind segment it sits in
1568        /// (a `c@n` repeats per chunk, `Index` per stream, so occurrences are
1569        /// matched per segment rather than by value).
1570        type Slot<'a> = (&'a str, &'a str, Option<&'a str>);
1571
1572        fn slots<'a>(
1573            doc: &'a ManifestDocument,
1574            skip: &dyn Fn(&str, &str, Option<&str>) -> bool,
1575        ) -> Vec<Slot<'a>> {
1576            let mut kind = None;
1577            doc.elements
1578                .iter()
1579                .filter_map(|(element, attribute, value)| {
1580                    let attribute = attribute.as_deref()?;
1581                    let mut stream = kind.filter(|k| *k != "other");
1582                    if element == "StreamIndex" && attribute == "Type" {
1583                        kind = match value.as_str() {
1584                            "video" => Some("video"),
1585                            "audio" => Some("audio"),
1586                            _ => Some("other"),
1587                        };
1588                        stream = None;
1589                    }
1590                    if skip(element, attribute, stream) {
1591                        return None;
1592                    }
1593                    Some((element.as_str(), attribute, stream))
1594                })
1595                .collect()
1596        }
1597        fn dropped_values(doc: &ManifestDocument) -> Vec<&str> {
1598            let mut kind: Option<&'static str> = None;
1599            let mut out: Vec<&str> = Vec::new();
1600            for (element, attribute, value) in &doc.elements {
1601                let Some(attribute) = attribute.as_deref() else {
1602                    continue;
1603                };
1604                let stream = kind.filter(|k| *k != "other");
1605                if element == "StreamIndex" && attribute == "Type" {
1606                    kind = match value.as_str() {
1607                        "video" => Some("video"),
1608                        "audio" => Some("audio"),
1609                        _ => Some("other"),
1610                    };
1611                    out.push(value.as_str());
1612                    continue;
1613                }
1614                let dropped = element == "StreamIndex"
1615                    && GOLDEN_DIFFERENCES.iter().any(|d| {
1616                        d.element == element.as_str()
1617                            && d.attribute == attribute
1618                            && d.stream == stream
1619                            && d.ours.is_none()
1620                    });
1621                if !dropped {
1622                    out.push(value.as_str());
1623                }
1624            }
1625            out
1626        }
1627        fn values(doc: &ManifestDocument) -> Vec<&str> {
1628            doc.elements
1629                .iter()
1630                .filter_map(|(_, attribute, value)| attribute.as_deref().map(|_| value.as_str()))
1631                .collect()
1632        }
1633
1634        let dropped = |element: &str, attribute: &str, stream: Option<&str>| {
1635            GOLDEN_DIFFERENCES.iter().any(|d| {
1636                d.element == element
1637                    && d.attribute == attribute
1638                    && d.stream == stream
1639                    && d.ours.is_none()
1640            })
1641        };
1642        let ours_slots = slots(&got, &|_, _, _| false);
1643        let wanted_slots = slots(&want, &dropped);
1644        assert_eq!(
1645            ours_slots, wanted_slots,
1646            "{file}: the sequence of (element, attribute, stream) differs"
1647        );
1648
1649        let wanted_values = dropped_values(&want);
1650        let ours_values = values(&got);
1651        assert_eq!(
1652            ours_values.len(),
1653            wanted_values.len(),
1654            "{file}: value count"
1655        );
1656
1657        let mut differences: Vec<String> = Vec::new();
1658        for ((slot, got_value), want_value) in
1659            ours_slots.iter().zip(&ours_values).zip(&wanted_values)
1660        {
1661            let (element, attribute, stream) = *slot;
1662            match intended(element, attribute, stream) {
1663                Some(difference) => {
1664                    assert_eq!(
1665                        difference.golden,
1666                        Some(*want_value),
1667                        "{file}: the recorded golden value of {element}@{attribute} is stale"
1668                    );
1669                    if Some(*got_value) != difference.ours {
1670                        differences.push(format!(
1671                            "{element}@{attribute} ({stream:?}): golden {want_value:?}, ours {got_value:?}, intended {:?} ({})",
1672                            difference.ours, difference.reason
1673                        ));
1674                    }
1675                }
1676                None => {
1677                    if got_value != want_value {
1678                        differences.push(format!(
1679                            "{element}@{attribute}: golden {want_value:?}, ours {got_value:?}"
1680                        ));
1681                    }
1682                }
1683            }
1684        }
1685        assert!(
1686            differences.is_empty(),
1687            "{file}: {differences:#?}\nintended differences:\n{}",
1688            GOLDEN_DIFFERENCES
1689                .iter()
1690                .map(|d| format!(
1691                    "  {}.{} ({:?}): golden {:?} -> ours {:?} — {}",
1692                    d.element, d.attribute, d.stream, d.golden, d.ours, d.reason
1693                ))
1694                .collect::<Vec<_>>()
1695                .join("\n")
1696        );
1697
1698        // Every attribute the live Manifest emits is one the golden records:
1699        // nothing is invented.
1700        for (element, attribute, _) in &got.elements {
1701            let Some(attribute) = attribute else {
1702                continue;
1703            };
1704            assert!(
1705                want.elements
1706                    .iter()
1707                    .any(|(w, a, _)| w == element && a.as_deref() == Some(attribute.as_str())),
1708                "{file}: ours emits ({element}, {attribute}), which the golden does not record"
1709            );
1710        }
1711
1712        // Text elements, verbatim.
1713        let props = |doc: &ManifestDocument| -> Vec<(String, Option<String>, String)> {
1714            doc.elements
1715                .iter()
1716                .filter(|(name, attribute, _)| name == "Prop" && attribute.is_none())
1717                .cloned()
1718                .collect()
1719        };
1720        assert_eq!(props(&got), props(&want), "{file}: text elements");
1721
1722        // Every literal line, in order. A line that differs may differ only
1723        // in its attributes: its element name must be the same.
1724        let lines = |xml: &str| -> Vec<String> {
1725            xml.lines()
1726                .map(str::trim)
1727                .filter(|line| !line.is_empty())
1728                .map(|line| match line.strip_prefix("<Prop>") {
1729                    Some(_) => "<Prop>".to_string(),
1730                    None => line.to_string(),
1731                })
1732                .collect()
1733        };
1734        let ours = lines(actual);
1735        let theirs = lines(expected);
1736        assert_eq!(ours.len(), theirs.len(), "{file}: line count");
1737        for (number, (ours, theirs)) in ours.iter().zip(&theirs).enumerate() {
1738            if ours == theirs {
1739                continue;
1740            }
1741            let element = |line: &str| line.split_whitespace().next().unwrap_or("").to_string();
1742            assert_eq!(
1743                element(ours),
1744                element(theirs),
1745                "{file}: line {number} differs beyond its attributes"
1746            );
1747        }
1748    }
1749
1750    /// The manifest is the live model's own document: one `StreamIndex` per
1751    /// track, each with its own chunk timeline, and a duration covering the
1752    /// window.
1753    #[test]
1754    fn the_manifest_describes_every_track() {
1755        let route = publish(vec![video_spec(), audio_spec(2)]);
1756        let manifest = manifest(&route);
1757        assert!(manifest.is_live);
1758        assert_eq!(manifest.timescale, transmux::SMOOTH_TIMESCALE);
1759        let kinds: Vec<StreamType> = manifest.streams.iter().map(|s| s.stream_type).collect();
1760        assert_eq!(kinds, vec![StreamType::Video, StreamType::Audio]);
1761        for stream in &manifest.streams {
1762            let chunks = stream.enumerate_chunks().expect("chunks");
1763            assert_eq!(chunks.len(), SEGMENTS as usize);
1764            assert!(chunks.iter().all(|(_, d)| *d > 0));
1765        }
1766        let duration = manifest.duration.expect("duration");
1767        assert!(duration > 0, "{duration}");
1768        assert_eq!(manifest.dvr_window_length, Some(duration));
1769    }
1770
1771    /// Two AAC tracks are two streams: distinct `Name` and `Url`, the second
1772    /// named `audio_2` (MS-SSTR §2.2.2.3).
1773    #[test]
1774    fn two_tracks_of_one_kind_get_distinct_names_and_urls() {
1775        let route = publish(vec![audio_spec(1), audio_spec(2)]);
1776        let manifest = manifest(&route);
1777        assert_eq!(
1778            manifest.streams.len(),
1779            2,
1780            "both audio tracks are advertised"
1781        );
1782        let first = &manifest.streams[0];
1783        let second = &manifest.streams[1];
1784        assert_eq!(first.name, None, "the first stream of a kind is unnamed");
1785        assert_eq!(second.name.as_deref(), Some("audio_2"));
1786        assert_ne!(first.url, second.url);
1787        assert!(second.url.contains("Fragments(audio_2={start time})"));
1788        for stream in &manifest.streams {
1789            assert_eq!(stream.quality_levels, Some(STREAM_QUALITY_LEVELS));
1790        }
1791    }
1792
1793    /// A fragment request for the second audio stream resolves to that
1794    /// stream's own fragment, whose `tfxd` is the `c@t` the manifest gives it.
1795    #[test]
1796    fn the_second_stream_serves_its_own_fragments() {
1797        let route = publish(vec![audio_spec(1), audio_spec(2)]);
1798        let origin = origin(&route);
1799        let manifest = manifest(&route);
1800        let second = &manifest.streams[1];
1801        let url = second.resolve_fragment_url(second.qualities[0].bitrate, 0);
1802        let path = format!("/cam/{url}");
1803        let request = SmoothRequest::parse(&path, &keys(&route))
1804            .unwrap_or_else(|| panic!("request parses: {path} from {:?}", keys(&route)));
1805        let response = origin.resolve(
1806            request,
1807            Timestamp::from_nanos(0),
1808            AwaitPolicy::new(Timestamp::from_nanos(u64::MAX)),
1809        );
1810        let EgressResponse::Ready {
1811            body: SmoothBody::Fragment(bytes),
1812            ..
1813        } = response
1814        else {
1815            panic!("expected fragment bytes from {path}, got {response:?}");
1816        };
1817        let tfxd = fragment_tfxd(&bytes).expect("fragment carries a tfxd");
1818        assert_eq!(tfxd.fragment_absolute_time, 0);
1819        assert!(tfxd.fragment_duration > 0);
1820    }
1821
1822    /// An AAC track whose `esds` carries no DecoderSpecificInfo has no
1823    /// `CodecPrivateData` a client could initialise from, so the manifest is
1824    /// refused rather than advertising an empty value.
1825    #[test]
1826    fn aac_without_decoder_specific_info_is_refused() {
1827        let mut spec = audio_spec(2);
1828        spec.config = CodecConfig::Aac {
1829            esds: esds(None),
1830            sample_rate: SAMPLE_RATE,
1831            channel_count: CHANNELS,
1832            sample_size: AAC_BITS_PER_SAMPLE,
1833        };
1834        let route = publish(vec![video_spec(), spec]);
1835        let origin = origin(&route);
1836        let response = origin.manifest(
1837            Timestamp::from_nanos(0),
1838            AwaitPolicy::new(Timestamp::from_nanos(u64::MAX)),
1839        );
1840        assert!(
1841            matches!(response, EgressResponse::BadRequest { .. }),
1842            "expected a refusal, got {response:?}"
1843        );
1844    }
1845
1846    /// Video geometry comes from the SPS, not from a `CodecConfig` that
1847    /// reports none (RTSP/WHEP ingest builds it with width and height `0`).
1848    #[test]
1849    fn video_geometry_comes_from_the_sps() {
1850        let route = publish(vec![video_spec()]);
1851        let manifest = manifest(&route);
1852        let level = &manifest.streams[0].qualities[0];
1853        assert_eq!(level.width, Some(SPROP_WIDTH));
1854        assert_eq!(level.height, Some(SPROP_HEIGHT));
1855        assert_eq!(level.four_cc, transmux::FOURCC_H264);
1856        assert!(!level.codec_private_data.is_empty());
1857    }
1858
1859    /// AAC's `SamplingRate` is the AudioSpecificConfig's own effective rate.
1860    #[test]
1861    fn aac_advertises_its_audio_specific_config() {
1862        let route = publish(vec![audio_spec(1)]);
1863        let manifest = manifest(&route);
1864        let level = &manifest.streams[0].qualities[0];
1865        assert_eq!(level.sampling_rate, Some(SAMPLE_RATE));
1866        assert_eq!(level.channels, Some(CHANNELS));
1867        assert_eq!(level.audio_tag, Some(AUDIO_TAG_AAC));
1868        let asc = <AudioSpecificConfig as broadcast_common::Parse>::parse(&ASC_48K_STEREO)
1869            .expect("asc parses");
1870        assert_eq!(asc.effective_sampling_frequency(), Some(SAMPLE_RATE));
1871        assert_eq!(level.codec_private_data, ASC_48K_STEREO.to_vec());
1872    }
1873
1874    /// A subtitle track is advertised as a `text` stream with the required
1875    /// `Subtype="SUBT"` and its sample-entry FourCC.
1876    #[test]
1877    fn a_subtitle_track_is_advertised() {
1878        let route = publish(vec![video_spec(), subtitle_spec(2)]);
1879        let manifest = manifest(&route);
1880        let text = manifest
1881            .streams
1882            .iter()
1883            .find(|s| s.stream_type == StreamType::Text)
1884            .expect("the subtitle track is advertised");
1885        assert_eq!(text.subtype.as_deref(), Some(SUBTYPE_SUBT));
1886        assert_eq!(text.qualities[0].four_cc, transmux::smooth::FOURCC_STPP);
1887        assert!(!text.enumerate_chunks().expect("chunks").is_empty());
1888    }
1889
1890    /// The manifest round-trips through the parser byte-for-byte, so what a
1891    /// client reads is what `broadcast-smooth` wrote.
1892    #[test]
1893    fn the_rendered_manifest_round_trips() {
1894        let route = publish(vec![video_spec(), audio_spec(2)]);
1895        let origin = origin(&route);
1896        let EgressResponse::Ready { body, .. } = origin.manifest(
1897            Timestamp::from_nanos(0),
1898            AwaitPolicy::new(Timestamp::from_nanos(u64::MAX)),
1899        ) else {
1900            panic!("manifest renders");
1901        };
1902        let parsed = SmoothManifest::parse(&body).expect("manifest parses");
1903        assert_eq!(parsed.to_xml(), body);
1904    }
1905
1906    /// A fragment path names its stream by the manifest's own `Name`
1907    /// (`TYPE` for the first, `TYPE_N` for the rest); an unknown one, a
1908    /// malformed one and a manifest name all parse as expected.
1909    #[test]
1910    fn requests_parse_from_their_paths() {
1911        let kinds = [
1912            StreamKey {
1913                kind: SmoothStreamType::Video,
1914                ordinal: 1,
1915            },
1916            StreamKey {
1917                kind: SmoothStreamType::Audio,
1918                ordinal: 2,
1919            },
1920        ];
1921        assert_eq!(
1922            SmoothRequest::parse("/cam/Manifest", &kinds),
1923            Some(SmoothRequest::Manifest)
1924        );
1925        assert_eq!(
1926            SmoothRequest::parse(
1927                "/cam/QualityLevels(128000)/Fragments(audio_2=21333333)",
1928                &kinds
1929            ),
1930            Some(SmoothRequest::Fragment {
1931                stream: kinds[1],
1932                bitrate: 128_000,
1933                start: 21_333_333,
1934            })
1935        );
1936        assert_eq!(
1937            SmoothRequest::parse("QualityLevels(2500000)/Fragments(video=0)", &kinds),
1938            Some(SmoothRequest::Fragment {
1939                stream: kinds[0],
1940                bitrate: 2_500_000,
1941                start: 0,
1942            })
1943        );
1944        assert_eq!(SmoothRequest::parse("/cam/nope", &kinds), None);
1945        assert_eq!(
1946            SmoothRequest::parse("/cam/QualityLevels(1)/Fragments(text=0)", &kinds),
1947            None,
1948            "a kind the manifest does not advertise is not a request"
1949        );
1950    }
1951
1952    /// A fragment start outside the window the manifest describes is
1953    /// `NotFound`, never a fragment assembled from a neighbouring chunk and
1954    /// never an arithmetic panic: one tick before the first chunk and one tick
1955    /// after the end of the last are both outside, and so is the largest
1956    /// representable start.
1957    #[test]
1958    fn a_fragment_outside_the_window_is_not_found() {
1959        let route = publish(vec![audio_spec(1)]);
1960        let origin = origin(&route);
1961        let manifest = manifest(&route);
1962        let stream = &manifest.streams[0];
1963        let bitrate = stream.qualities[0].bitrate;
1964        let chunks = stream.enumerate_chunks().expect("chunks");
1965        let first = chunks.first().expect("a first chunk").0;
1966        let (last_start, last_duration) = chunks.last().copied().expect("a last chunk");
1967        let end = last_start.checked_add(last_duration).expect("the end fits");
1968        for start in [first.checked_sub(1), Some(end), Some(u64::MAX)]
1969            .into_iter()
1970            .flatten()
1971        {
1972            let request = SmoothRequest::Fragment {
1973                stream: keys(&route)[0],
1974                bitrate,
1975                start,
1976            };
1977            let response = origin.resolve(
1978                request,
1979                Timestamp::from_nanos(0),
1980                AwaitPolicy::new(Timestamp::from_nanos(u64::MAX)),
1981            );
1982            assert_eq!(
1983                response,
1984                EgressResponse::NotFound,
1985                "start {start} is outside the window"
1986            );
1987        }
1988        for (start, _) in &chunks {
1989            let request = SmoothRequest::Fragment {
1990                stream: keys(&route)[0],
1991                bitrate,
1992                start: *start,
1993            };
1994            assert!(
1995                matches!(
1996                    origin.resolve(
1997                        request,
1998                        Timestamp::from_nanos(0),
1999                        AwaitPolicy::new(Timestamp::from_nanos(u64::MAX))
2000                    ),
2001                    EgressResponse::Ready { .. }
2002                ),
2003                "the window holds a chunk starting at {start}"
2004            );
2005        }
2006    }
2007
2008    /// A window with no chunk at all holds no start, whatever the value, and
2009    /// the bounds comparison never wraps for a chunk whose end is not
2010    /// representable.
2011    #[test]
2012    fn the_window_bounds_are_checked() {
2013        let codec = resolve_codec(&CodecConfig::Aac {
2014            esds: esds(Some(ASC_48K_STEREO.to_vec())),
2015            sample_rate: SAMPLE_RATE,
2016            channel_count: CHANNELS,
2017            sample_size: AAC_BITS_PER_SAMPLE,
2018        })
2019        .expect("AAC resolves");
2020        let key = StreamKey {
2021            kind: SmoothStreamType::Audio,
2022            ordinal: 1,
2023        };
2024        let empty = StreamPlan {
2025            key,
2026            codec: codec.clone(),
2027            interleaved_ordinal: 1,
2028            chunks: Vec::new(),
2029        };
2030        assert!(!window_holds(&empty, 0));
2031        let huge = StreamPlan {
2032            key,
2033            codec,
2034            interleaved_ordinal: 1,
2035            chunks: vec![ChunkTiming {
2036                sequence: SegmentSeq::new(1),
2037                start: u64::MAX,
2038                duration: 1,
2039            }],
2040        };
2041        assert!(window_holds(&huge, u64::MAX));
2042        assert!(!window_holds(&huge, u64::MAX - 1));
2043    }
2044
2045    /// A start inside the window's span but not on an advertised `c@t` (a
2046    /// tick between two chunks) is a fragment the window does not hold, so it
2047    /// is `NotFound` — never a fragment assembled from a neighbouring chunk.
2048    #[test]
2049    fn a_start_between_chunks_is_not_found() {
2050        let route = publish(vec![audio_spec(1)]);
2051        let origin = origin(&route);
2052        let manifest = manifest(&route);
2053        let stream = &manifest.streams[0];
2054        let bitrate = stream.qualities[0].bitrate;
2055        let chunks = stream.enumerate_chunks().expect("chunks");
2056        let (first, _) = chunks.first().copied().expect("a first chunk");
2057        let (second, duration) = chunks.get(1).copied().expect("a second chunk");
2058        let between = [
2059            first.checked_add(1).expect("a tick after the first chunk"),
2060            second
2061                .checked_add(duration.checked_div(2).unwrap_or(0))
2062                .expect("mid-way through the last chunk"),
2063        ];
2064        for start in between {
2065            let request = SmoothRequest::Fragment {
2066                stream: keys(&route)[0],
2067                bitrate,
2068                start,
2069            };
2070            assert_eq!(
2071                origin.resolve(
2072                    request,
2073                    Timestamp::from_nanos(0),
2074                    AwaitPolicy::new(Timestamp::from_nanos(u64::MAX))
2075                ),
2076                EgressResponse::NotFound,
2077                "no chunk starts at {start}"
2078            );
2079        }
2080    }
2081
2082    /// A stream key's name and URL template agree: the template's `Fragments`
2083    /// token carries exactly the name the `c` timeline is advertised under.
2084    #[test]
2085    fn a_stream_keys_url_carries_its_name() {
2086        let first = StreamKey {
2087            kind: SmoothStreamType::Video,
2088            ordinal: 1,
2089        };
2090        let second = StreamKey {
2091            kind: SmoothStreamType::Video,
2092            ordinal: 2,
2093        };
2094        assert_eq!(first.name(), "video");
2095        assert_eq!(second.name(), "video_2");
2096        assert!(
2097            second
2098                .url_template()
2099                .contains("Fragments(video_2={start time})")
2100        );
2101        assert_eq!(
2102            first.url_template(),
2103            "QualityLevels({bitrate})/Fragments(video={start time})"
2104        );
2105    }
2106
2107    /// The origin is shared across threads by its host, so the cache it holds
2108    /// must not take that away.
2109    #[test]
2110    fn the_origin_is_send_and_sync() {
2111        fn assert_send_sync<T: Send + Sync>() {}
2112        assert_send_sync::<SmoothOrigin>();
2113    }
2114
2115    type Window = Vec<(SegmentSeq, Bytes)>;
2116
2117    /// An origin whose window the test moves by publishing segments, and the
2118    /// handle it moves it with.
2119    fn sliding(route: &Route, initial: usize) -> (SmoothOrigin, Arc<Mutex<Window>>) {
2120        let window = Arc::new(Mutex::new(route.published[..initial].to_vec()));
2121        let shared = Arc::clone(&window);
2122        let specs = route.specs();
2123        let origin = SmoothOrigin::new(SmoothOriginConfig::new(
2124            Arc::clone(&route.trunk),
2125            Arc::new(move || locked(&shared).clone()),
2126            Arc::new(move || specs.clone()),
2127        ));
2128        origin.set_init(route.init.clone());
2129        (origin, window)
2130    }
2131
2132    fn render(origin: &SmoothOrigin) -> EgressResponse<String> {
2133        origin.manifest(
2134            Timestamp::from_nanos(0),
2135            AwaitPolicy::new(Timestamp::from_nanos(u64::MAX)),
2136        )
2137    }
2138
2139    fn builds(origin: &SmoothOrigin) -> usize {
2140        origin.plan_builds.load(std::sync::atomic::Ordering::SeqCst)
2141    }
2142
2143    /// N requests over one window build the plan once; publishing a segment
2144    /// changes the window and the next request builds it again, once.
2145    #[test]
2146    fn the_plan_is_built_once_per_window() {
2147        let route = publish(vec![video_spec()]);
2148        let (origin, window) = sliding(&route, 1);
2149        let keys = origin.stream_keys();
2150        assert_eq!(keys.len(), 1);
2151        let first = render(&origin);
2152        for _ in 0..REQUESTS {
2153            assert_eq!(render(&origin), first);
2154            origin.fragments().expect("fragments");
2155            origin.stream_keys();
2156        }
2157        assert_eq!(builds(&origin), 1);
2158
2159        locked(&window).push(route.published[1].clone());
2160        let second = render(&origin);
2161        assert_ne!(second, first);
2162        for _ in 0..REQUESTS {
2163            assert_eq!(render(&origin), second);
2164        }
2165        assert_eq!(builds(&origin), 2);
2166    }
2167
2168    /// A new init segment invalidates the plan; storing the one already held
2169    /// does not.
2170    #[test]
2171    fn only_a_changed_init_segment_rebuilds_the_plan() {
2172        let route = publish(vec![video_spec()]);
2173        let (origin, _window) = sliding(&route, 1);
2174        render(&origin);
2175        origin.set_init(route.init.clone());
2176        render(&origin);
2177        assert_eq!(builds(&origin), 1);
2178        let mut changed = route.init.clone();
2179        changed.push(0);
2180        origin.set_init(changed);
2181        render(&origin);
2182        assert_eq!(builds(&origin), 2);
2183    }
2184
2185    /// A fragment requested repeatedly is packaged once; one that left the
2186    /// window is dropped from the cache when the plan is rebuilt.
2187    #[test]
2188    fn a_fragment_is_packaged_once_and_evicted_with_its_segment() {
2189        let route = publish(vec![video_spec()]);
2190        let (origin, window) = sliding(&route, 2);
2191        let stream = origin.stream_keys()[0];
2192        let chunks = origin.fragments().expect("fragments");
2193        let bitrate =
2194            advertised_bitrate(&origin.snapshot().ok().expect("a snapshot").plan.streams[0].codec);
2195        let fetch = |start: u64| {
2196            origin.resolve(
2197                SmoothRequest::Fragment {
2198                    stream,
2199                    bitrate,
2200                    start,
2201                },
2202                Timestamp::from_nanos(0),
2203                AwaitPolicy::new(Timestamp::from_nanos(u64::MAX)),
2204            )
2205        };
2206        let first = fetch(chunks[0].start);
2207        for _ in 0..REQUESTS {
2208            assert_eq!(fetch(chunks[0].start), first);
2209        }
2210        assert_eq!(
2211            origin
2212                .fragment_builds
2213                .load(std::sync::atomic::Ordering::SeqCst),
2214            1
2215        );
2216        assert!(matches!(first, EgressResponse::Ready { .. }));
2217        fetch(chunks[1].start);
2218        assert_eq!(locked(&origin.fragments).len(), 2);
2219
2220        let kept = locked(&window).split_off(1);
2221        *locked(&window) = kept;
2222        render(&origin);
2223        assert!(locked(&origin.fragments).len() <= 1);
2224    }
2225
2226    /// A track-set change with no new segment yields a new Manifest.
2227    #[test]
2228    fn a_track_change_without_a_new_segment_rebuilds_the_plan() {
2229        let route = publish(vec![video_spec(), audio_spec(2)]);
2230        let current = Arc::new(Mutex::new(vec![route.specs()[0].clone()]));
2231        let shared = Arc::clone(&current);
2232        let origin = SmoothOrigin::new(SmoothOriginConfig::new(
2233            Arc::clone(&route.trunk),
2234            Arc::new(route.published()),
2235            Arc::new(move || locked(&shared).clone()),
2236        ));
2237        origin.set_init(route.init.clone());
2238        let video_only = render(&origin);
2239        assert_eq!(render(&origin), video_only);
2240        assert_eq!(builds(&origin), 1);
2241
2242        *locked(&current) = route.specs();
2243        let both = render(&origin);
2244        assert_ne!(both, video_only);
2245        assert_eq!(builds(&origin), 2);
2246    }
2247}