transmux 0.29.1

Any-to-any media container muxing hub: demux TS, fMP4/CMAF, MPEG-PS, WebM, FLV, or RTMP into one neutral IR and mux to CMAF/fMP4, progressive MP4, TS, DASH, low-latency DASH, HLS, low-latency HLS, Smooth Streaming, or RTMP. CENC/CBCS encrypt+decrypt, SSAI splice, RTP/RTCP, and an fMP4/CMAF conformance validator; parses codec config headers only, samples stay opaque. no_std + alloc, except the XML manifest paths (DASH/Smooth/PlayReady), which need the std feature.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
//! Low-latency DASH — chunked CMAF packaging + LL-DASH MPD signalling.
//!
//! Whole-segment CMAF ([`crate::segmenter::Segmenter`], [`crate::media::CmafMux`])
//! delivers one `moof`+`mdat` per segment: a client cannot start playing a
//! segment until the whole thing has been produced, so end-to-end latency is at
//! least one segment duration. **Low-latency DASH** (DASH-IF Low-latency Live
//! Interoperability, "LL-DASH IOP") cuts that by splitting each CMAF segment
//! into several **CMAF chunks** — each a self-contained `moof`+`mdat` covering a
//! sub-run of the segment's samples — delivered (over HTTP chunked transfer) as
//! they are produced. The client can begin decoding the first chunk while the
//! rest of the segment is still being generated.
//!
//! # Chunked-CMAF structure (ISO/IEC 14496-12:2015)
//!
//! A CMAF chunk is exactly the fragment structure the batch builder
//! [`crate::build_media_segment`] already emits, but scoped to K samples:
//!
//! - the **first chunk** of a segment carries the segment-type box
//!   `styp` (§8.16.4) followed by `moof` (§8.8.4) + `mdat` (§8.1.1);
//! - **subsequent chunks** are a bare `moof` + `mdat`.
//!
//! Each chunk's `moof` has its own `mfhd.sequence_number` (§8.8.5) — contiguous
//! and increasing across all chunks of all segments — and each track fragment's
//! `tfdt.baseMediaDecodeTime` (§8.8.12) is the decode time of that chunk's first
//! sample. Concatenating a segment's chunks yields the same coded samples, in
//! the same decode order, with the same per-track decode timeline, as the
//! whole-segment [`crate::build_media_segment`] output — chunking splits, it
//! never loses, duplicates, or reorders a sample. Segment boundaries stay
//! keyframe-aligned (the anchor track's first chunk of every segment starts on a
//! sync sample), exactly as [`crate::segmenter::Segmenter`] guarantees.
//!
//! # LL-DASH MPD signalling (ISO/IEC 23009-1 + DASH-IF LL IOP)
//!
//! [`LlDashPackager`] renders an MPD that advertises the chunked availability so
//! a player can request a segment before it is complete:
//!
//! - **`SegmentTemplate@availabilityTimeComplete="false"`** (ISO/IEC 23009-1
//!   §5.3.9.5.3) — a segment becomes available (its first chunk) before the whole
//!   segment is produced.
//! - **`SegmentTemplate@availabilityTimeOffset`** (§5.3.9.5.3) — how many seconds
//!   *before* the nominal segment-complete time the segment first becomes
//!   available. Per DASH-IF LL IOP this is `segment_duration − chunk_duration`
//!   (the first chunk is ready one chunk-duration into the segment).
//! - **`MPD/ServiceDescription/Latency@target`** (§5.13.2) — the target
//!   end-to-end latency in milliseconds, with an optional `<PlaybackRate>` giving
//!   the min/max catch-up rate.
//! - **`MPD/UTCTiming`** (§5.8.4.11) — how a client obtains wall-clock UTC. A
//!   `dynamic` MPD is timed against `availabilityStartTime`, which is only usable
//!   if the client's clock agrees with the server's; DASH-IF LL IOP §4.1 makes
//!   the element effectively required for a low-latency presentation. The
//!   `@schemeIdUri` (one of the six ISO-registered `urn:mpeg:dash:utc:*:2014`
//!   values) and `@value` are supplied by the caller.
//! - `MPD@type="dynamic"` with `@availabilityStartTime` (§5.3.1.2) — LL-DASH is a
//!   live profile.
//!
//! `<ProducerReferenceTime>` (§5.12) is **optional** and is *not* emitted by this
//! packager: it requires a wall-clock capture timestamp per segment that the
//! samples-in IR does not carry. A caller with that timing can add it out of
//! band; its absence does not affect the availability signalling above.

use crate::TrackId;
use alloc::collections::VecDeque;
use alloc::vec;
use alloc::vec::Vec;

use broadcast_common::{Demand, Stage, Timestamp};

use crate::error::{Error, Result};
use crate::media::Media;
use crate::movie_fragment::{
    MediaRunSample, MovieFragmentBox, MovieFragmentHeaderBox, TrackFragmentBaseMediaDecodeTimeBox,
    TrackFragmentBox, media_fragment_run,
};
use crate::pipeline::{FragmentTrackData, Sample, TrackSpec, build_init_segment};
use crate::segmenter::{MAX_PENDING_SAMPLES_PER_TRACK, MediaClock, choose_anchor};
use crate::segments::{MediaDataBox, SegmentTypeBox};
use broadcast_common::{Package, Serialize};

/// `styp` major brand for a low-latency CMAF chunk (§8.16.4). Matches the brand
/// `build_media_segment` writes so chunk concat == whole segment.
const STYP_MAJOR_BRAND: [u8; 4] = *b"msdh";
/// `styp` compatible brands.
const STYP_COMPATIBLE_BRANDS: [[u8; 4]; 2] = [*b"msdh", *b"msix"];

// ===========================================================================
// LlSegmenter — chunked-CMAF output packager
// ===========================================================================

/// One finished CMAF chunk: its bytes plus the metadata a caller needs to place
/// it (which segment it belongs to, whether it opens that segment).
#[derive(Debug, Clone)]
pub struct Chunk {
    /// The chunk bytes: `styp`+`moof`+`mdat` for the first chunk of a segment,
    /// or a bare `moof`+`mdat` for a continuation chunk.
    pub data: Vec<u8>,
    /// 1-based index of the segment this chunk belongs to.
    pub segment_number: u64,
    /// `true` if this is the first chunk of its segment (carries `styp`, and its
    /// anchor-track first sample is a keyframe).
    pub is_segment_start: bool,
    /// `mfhd.sequence_number` of this chunk's `moof`.
    pub sequence_number: u32,
}

/// Per-track accumulation state for the segment currently being built.
struct TrackState {
    spec: TrackSpec,
    /// Samples buffered for the current (not-yet-cut) segment, in decode order.
    pending: Vec<Sample>,
    /// Decode time of the first *pending* sample (media-timescale ticks) — the
    /// `tfdt.baseMediaDecodeTime` of this track's next chunk.
    base_decode: u64,
    /// Advances `base_decode` past each sample as it is drained into a chunk,
    /// under the same duration-then-dts-delta rule as the anchor accumulator
    /// (see [`MediaClock`]). A separate clock instance from the anchor's: this
    /// one walks *this* track's samples once at drain time. A plain `duration`
    /// sum would pin `base_decode` at 0 forever on a `duration: Some(0)`
    /// stream, collapsing every chunk's `tfdt` onto the same decode time.
    flush_clock: MediaClock,
}

/// A stateful **chunked** CMAF segmenter for low-latency DASH.
///
/// Same segmentation state machine as [`crate::segmenter::Segmenter`] — segments
/// are cut on the anchor track's sync samples once the target duration is reached
/// — but each segment is emitted as an ordered list of [`Chunk`]s: one
/// `moof`+`mdat` per `chunk_samples` anchor-track samples (e.g. `1` for per-frame
/// LL). Concatenating a segment's chunks reproduces the whole-segment
/// [`crate::build_media_segment`] coded-sample stream (see the module docs).
///
/// ```
/// use transmux::{CodecConfig, LlSegmenter, Sample, TrackSpec};
/// # use transmux::{AVCConfigurationBox, AVCDecoderConfigurationRecord, AvcPps, AvcSps};
/// # fn spec() -> TrackSpec {
/// #     let record = AVCDecoderConfigurationRecord {
/// #         configuration_version: 1,
/// #         profile_indication: 66,
/// #         profile_compatibility: 0,
/// #         level_indication: 30,
/// #         length_size_minus_one: 3,
/// #         sps: vec![AvcSps(vec![0x67, 0x42, 0xc0, 0x1e, 0xd9, 0x00, 0x80, 0x1e, 0x24])],
/// #         pps: vec![AvcPps(vec![0x68, 0xce, 0x3c, 0x80])],
/// #         chroma_format: None,
/// #         bit_depth_luma_minus8: None,
/// #         bit_depth_chroma_minus8: None,
/// #         sps_ext: vec![],
/// #     };
/// #     TrackSpec::new(transmux::TrackId::new(1), 90_000, CodecConfig::Avc {
/// #         config: AVCConfigurationBox::new(record),
/// #         width: 16,
/// #         height: 16,
/// #     })
/// # }
/// # fn au(sync: bool) -> Sample {
/// #     use std::sync::atomic::{AtomicI64, Ordering};
/// #     static NEXT_DTS: AtomicI64 = AtomicI64::new(0);
/// #     let dts = NEXT_DTS.fetch_add(1000, Ordering::Relaxed);
/// #     Sample::new(vec![0u8; 4], Some(dts), Some(dts), Some(1000), sync)
/// # }
/// // 2 s target segments, one video frame per chunk (per-frame LL).
/// let mut seg = LlSegmenter::new(vec![spec()], 1000, 2.0, 1).unwrap();
/// let init = seg.init_segment().unwrap();       // ftyp + moov
/// assert_eq!(&init[4..8], b"ftyp");
/// seg.push(transmux::TrackId::new(1), au(true)).unwrap();               // keyframe
/// seg.push(transmux::TrackId::new(1), au(false)).unwrap();
/// let chunks = seg.take_ready();                // write chunk.data over HTTP
/// assert_eq!(chunks.len(), 2);                  // one chunk per pushed frame
/// assert!(chunks[0].is_segment_start);
/// seg.flush().unwrap();                         // trailing chunks (none pending here)
/// ```
pub struct LlSegmenter {
    tracks: Vec<TrackState>,
    movie_timescale: u32,
    /// Index into `tracks` of the segmentation anchor (keyframe cut boundary).
    anchor: usize,
    /// Target segment duration in the *anchor track's* media timescale.
    target_ticks: u64,
    /// Number of anchor-track samples per chunk (>= 1).
    chunk_samples: usize,
    /// Buffered duration of the anchor's `pending` samples (media-timescale ticks).
    anchor_pending_dur: u64,
    /// Anchor-progress clock: advances `anchor_pending_dur` from each anchor
    /// sample's `duration`, or from its `dts` delta when `duration` is absent
    /// or zero (see [`MediaClock`]).
    anchor_clock: MediaClock,
    /// `mfhd.sequence_number` of the next chunk, 1-based and contiguous.
    next_seq: u32,
    /// 1-based number of the segment currently being built.
    current_segment: u64,
    /// `true` until the first chunk of the current segment has been emitted.
    segment_open: bool,
    /// Chunks finished but not yet taken by the caller. The single source of
    /// truth for both the inherent [`take_ready`](Self::take_ready) drain and
    /// [`Stage::poll`] — whichever API the caller uses pops from this same
    /// queue, so a chunk is delivered exactly once no matter which (or both)
    /// APIs drive this segmenter.
    ready: VecDeque<Chunk>,
}

impl LlSegmenter {
    /// Create a chunked segmenter for `tracks`, cutting segments roughly every
    /// `target_duration_secs` on the anchor track's keyframes, and subdividing
    /// each segment into chunks of `chunk_samples` anchor-track samples.
    ///
    /// The anchor is chosen by the shared
    /// [`crate::segmenter::choose_anchor`] — the first **video**
    /// track (any codec, not just AVC), falling back to the first
    /// anchor-capable track — exactly as [`crate::segmenter::Segmenter`].
    /// `movie_timescale` matches [`build_init_segment`].
    ///
    /// # Errors
    /// [`Error::InvalidInput`] if `tracks` is empty, has duplicate `track_id`s,
    /// `target_duration_secs` is not positive and finite, `chunk_samples == 0`,
    /// or no track is anchor-capable (every track is section-carried).
    pub fn new(
        tracks: Vec<TrackSpec>,
        movie_timescale: u32,
        target_duration_secs: f64,
        chunk_samples: usize,
    ) -> Result<Self> {
        if tracks.is_empty() {
            return Err(Error::InvalidInput("ll segmenter needs at least one track"));
        }
        if !(target_duration_secs.is_finite() && target_duration_secs > 0.0) {
            return Err(Error::InvalidInput(
                "target_duration_secs must be positive and finite",
            ));
        }
        if chunk_samples == 0 {
            return Err(Error::InvalidInput("chunk_samples must be >= 1"));
        }
        for (i, a) in tracks.iter().enumerate() {
            if tracks[i + 1..].iter().any(|b| b.track_id == a.track_id) {
                return Err(Error::InvalidInput("duplicate track_id"));
            }
        }

        // MUX = strict but filterable (media plane step-2 fix wave 1,
        // B2-B4): a track `build_init_segment` cannot place into an ISOBMFF
        // `trak` (opaque `CodecConfig::Data` or `CodecConfig::Subtitle`) is
        // no longer silently omitted here — it surfaces the same named
        // error every other mux entry point does, the first time
        // `init_segment`/a chunk build actually needs the `trak` (this
        // constructor does not itself need to build one). The caller must
        // pre-filter first (e.g. with
        // `tracks.retain(|t| t.config.is_muxable_in_bmff())`) if it wants
        // to drop such tracks rather than fail.

        // Anchor = first **video** track (any `CodecConfig::is_video` codec),
        // else the first anchor-capable track — the shared rule
        // `Segmenter`/`ts_hls` use. This used to match only
        // `CodecConfig::Avc`, so an HEVC+AAC media with audio first anchored on
        // the *audio* track and its segments did not begin on an IRAP.
        let anchor = choose_anchor(tracks.iter().map(|t| &t.config))?;

        let anchor_timescale = tracks[anchor].timescale as f64;
        let target_ticks = ((target_duration_secs * anchor_timescale) as u64).max(1);

        let tracks = tracks
            .into_iter()
            .map(|spec| TrackState {
                spec,
                pending: Vec::new(),
                base_decode: 0,
                flush_clock: MediaClock::new(),
            })
            .collect();

        Ok(Self {
            tracks,
            movie_timescale,
            anchor,
            target_ticks,
            chunk_samples,
            anchor_pending_dur: 0,
            anchor_clock: MediaClock::new(),
            next_seq: 1,
            current_segment: 1,
            segment_open: false,
            ready: VecDeque::new(),
        })
    }

    /// The initialization segment (`ftyp` + fragmented-init `moov`). Stable for
    /// the life of the segmenter; write it once before any chunk.
    pub fn init_segment(&self) -> Result<Vec<u8>> {
        let specs: Vec<TrackSpec> = self.tracks.iter().map(|t| t.spec.clone()).collect();
        build_init_segment(&specs, self.movie_timescale)
    }

    /// Push one coded sample for `track_id`, in decode order.
    ///
    /// When the anchor track reaches a sync sample past the target duration, the
    /// current segment is finalized (all its remaining buffered samples flushed
    /// as a final chunk) *before* this sample is buffered, so the new keyframe
    /// opens the next segment's first chunk on a random-access point.
    ///
    /// Anchor progress comes from [`MediaClock`]: each anchor sample's own
    /// `duration` when that is a real, non-zero span, else the `dts` delta from
    /// the previous anchor sample — `duration` alone stalls on the `Some(0)`
    /// durations live input legitimately produces.
    ///
    /// # Errors
    /// [`Error::InvalidInput`] if `track_id` matches no track, a chunk build
    /// fails, or this track already holds [`MAX_PENDING_SAMPLES_PER_TRACK`]
    /// un-cut samples (no anchor sync sample to cut on — call
    /// [`flush`](Self::flush) to close a trailing partial segment).
    pub fn push(&mut self, track_id: TrackId, sample: Sample) -> Result<()> {
        let idx = self
            .tracks
            .iter()
            .position(|t| t.spec.track_id == track_id)
            .ok_or(Error::InvalidInput("push: unknown track_id"))?;

        // Segment boundary: anchor keyframe past target → flush the segment.
        if idx == self.anchor
            && sample.flags.is_sync
            && self.anchor_pending_dur >= self.target_ticks
            && !self.tracks[self.anchor].pending.is_empty()
        {
            self.finish_segment()?;
        }

        if self.tracks[idx].pending.len() >= MAX_PENDING_SAMPLES_PER_TRACK {
            return Err(crate::segmenter::no_sync_sample_error());
        }

        if idx == self.anchor {
            self.anchor_pending_dur += self.anchor_clock.tick(&sample);
        }
        self.tracks[idx].pending.push(sample);

        // Emit a chunk once the anchor has buffered a full chunk's worth of
        // samples (but hold the segment's last chunk until the boundary/flush so
        // audio that trails the last video frame stays in the same segment).
        if idx == self.anchor
            && self.tracks[self.anchor].pending.len() >= self.chunk_samples
            && self.anchor_pending_dur < self.target_ticks
        {
            self.emit_chunk(false)?;
        }
        Ok(())
    }

    /// Finalize the trailing segment at end-of-stream. A no-op if nothing is
    /// buffered. Any remaining chunks are appended to the ready queue.
    ///
    /// # Errors
    /// Propagates a chunk-build failure.
    pub fn flush(&mut self) -> Result<()> {
        if self.tracks.iter().any(|t| !t.pending.is_empty()) {
            self.finish_segment()?;
        }
        Ok(())
    }

    /// Remove and return every chunk finished since the last call, in order.
    pub fn take_ready(&mut self) -> Vec<Chunk> {
        self.ready.drain(..).collect()
    }

    /// Flush the whole current segment: emit all remaining buffered anchor
    /// samples (and every non-anchor sample) as chunks, then open the next
    /// segment. The final chunk of a segment carries any trailing samples.
    fn finish_segment(&mut self) -> Result<()> {
        // Emit whole chunks of anchor samples first, then a final chunk for the
        // remainder (which also carries all non-anchor buffered samples).
        while self.tracks[self.anchor].pending.len() > self.chunk_samples {
            self.emit_chunk(false)?;
        }
        // Final chunk of the segment (anchor remainder + all other tracks).
        self.emit_chunk(true)?;

        self.current_segment += 1;
        self.segment_open = false;
        self.anchor_pending_dur = 0;
        Ok(())
    }

    /// Emit one chunk. If `final_chunk`, drain every buffered sample of every
    /// track; otherwise drain `chunk_samples` anchor samples and no other track's
    /// samples (non-anchor samples ride the segment's final chunk, matching the
    /// whole-segment layout's single per-track run).
    fn emit_chunk(&mut self, final_chunk: bool) -> Result<()> {
        let is_start = !self.segment_open;
        let anchor = self.anchor;

        // The decode span this chunk covers, in the anchor's timescale: the
        // first `chunk_samples` of the anchor's pending samples. Every track then
        // drains its own samples that fall inside that span, so audio rides each
        // chunk instead of only the segment's final one (audit r05-W27).
        let anchor_scale = self.tracks[anchor].spec.timescale.max(1) as u64;
        let anchor_take = if final_chunk {
            self.tracks[anchor].pending.len()
        } else {
            self.chunk_samples.min(self.tracks[anchor].pending.len())
        };
        let chunk_span = if final_chunk {
            u64::MAX
        } else {
            let t = &self.tracks[anchor];
            let mut clock =
                MediaClock::resumed_at(Some(i64::try_from(t.base_decode).unwrap_or(i64::MAX)));
            // Seed with the anchor's own first pending sample's real dts so the
            // delta fallback measures from the right baseline.
            let mut span: u64 = 0;
            for s in &t.pending[..anchor_take] {
                span += clock.tick(s);
            }
            span
        };

        // Decide, per track, how many leading pending samples this chunk drains.
        let take_counts: Vec<usize> = self
            .tracks
            .iter()
            .enumerate()
            .map(|(i, t)| {
                if final_chunk {
                    return t.pending.len();
                }
                if i == anchor {
                    return anchor_take;
                }
                let track_scale = t.spec.timescale.max(1) as u64;
                let track_span = chunk_span.saturating_mul(track_scale) / anchor_scale;
                take_span(t, track_span)
            })
            .collect();

        // Nothing to emit (e.g. flush with only-empty tracks) → skip.
        if take_counts.iter().all(|&n| n == 0) {
            return Ok(());
        }

        let seq = self.next_seq;
        let chunk_bytes = {
            let frags: Vec<FragmentTrackData<'_>> = self
                .tracks
                .iter()
                .zip(&take_counts)
                .filter(|&(_, &n)| n > 0)
                .map(|(t, &n)| FragmentTrackData {
                    track_id: t.spec.track_id,
                    base_media_decode_time: t.base_decode,
                    samples: &t.pending[..n],
                })
                .collect();
            build_chunk(seq, &frags, is_start)?
        };

        // Advance decode times and drop the drained samples. The drained
        // ranges are contiguous and in decode order across calls, so one
        // per-track `flush_clock` walks each sample exactly once and its
        // dts-delta fallback stays correct across chunk boundaries.
        for (t, &n) in self.tracks.iter_mut().zip(&take_counts) {
            let clock = &mut t.flush_clock;
            let dur: u64 = t.pending[..n].iter().map(|s| clock.tick(s)).sum();
            t.base_decode += dur;
            t.pending.drain(..n);
        }

        self.next_seq += 1;
        self.segment_open = true;
        self.ready.push_back(Chunk {
            data: chunk_bytes,
            segment_number: self.current_segment,
            is_segment_start: is_start,
            sequence_number: seq,
        });
        Ok(())
    }
}

/// How many of `t`'s pending samples fit inside a decode span of `target_ticks`
/// (in this track's timescale).
///
/// A lower-rate track (audio) is drained by *time*, not by the anchor's sample
/// count: a chunk that takes 3 video AUs also takes every audio frame inside
/// those 3 AUs, so audio accompanies the video in each chunk (audit r05-W27)
/// instead of waiting for the segment's final chunk. Always takes at least one
/// sample when one is available, so an over-target first sample cannot stall.
fn take_span(t: &TrackState, target_ticks: u64) -> usize {
    if t.pending.is_empty() {
        return 0;
    }
    let mut clock = MediaClock::resumed_at(Some(i64::try_from(t.base_decode).unwrap_or(i64::MAX)));
    let mut span: u64 = 0;
    let mut end = 0usize;
    for s in &t.pending {
        let next = clock.tick(s);
        if end > 0 && span + next > target_ticks {
            break;
        }
        span += next;
        end += 1;
    }
    end.max(1).min(t.pending.len())
}

/// [`Stage`] adoption (media plane step 2e-2): `In = (u32, Sample)`, same
/// reasoning as [`crate::segmenter::Segmenter`]'s impl. `Out = Chunk` —
/// [`take_ready`](Self::take_ready)'s own item type, not unified with any
/// other segmenter's `Out`.
///
/// Every inherent method — [`push`](Self::push), [`take_ready`](Self::take_ready),
/// [`flush`](Self::flush) — keeps working unchanged; this impl is an
/// additional, uniform way to drive the same engine. [`Stage::poll`] and the
/// inherent [`take_ready`](Self::take_ready) drain both read from the *same*
/// `ready` queue (there is no separate staging copy), so a chunk is
/// delivered exactly once no matter which API — inherent, `Stage`, or a mix
/// of both on the same instance — the caller uses to retrieve it.
impl Stage for LlSegmenter {
    type In<'a> = (TrackId, Sample);
    type Out = Chunk;
    type Error = Error;

    fn feed(&mut self, (track_id, sample): Self::In<'_>, _now: Timestamp) -> Result<()> {
        self.push(track_id, sample)
    }

    fn poll(&mut self) -> Option<Self::Out> {
        self.ready.pop_front()
    }

    fn finish(&mut self) -> Result<()> {
        self.flush()
    }

    fn next_deadline(&self) -> Option<Timestamp> {
        // Chunks are only cut in reaction to `push`/`flush` — no
        // rate-scheduled or timeout work.
        None
    }

    fn on_deadline(&mut self, _now: Timestamp) {}

    /// `saturated` once any track holds [`MAX_PENDING_SAMPLES_PER_TRACK`]
    /// un-cut samples — the same bound (and reasoning) as
    /// [`Segmenter`](crate::segmenter::Segmenter)'s: past it
    /// [`feed`](Stage::feed) errors rather than buffering a stream that never
    /// produces the sync sample a segment must open on.
    fn demand(&self) -> Demand {
        if self
            .tracks
            .iter()
            .any(|t| t.pending.len() >= MAX_PENDING_SAMPLES_PER_TRACK)
        {
            Demand::saturated()
        } else {
            Demand::default()
        }
    }
}

impl Package for LlSegmenter {
    type Media = Media;
    /// The ordered chunk list for the whole media (drained via the state machine
    /// in one shot). Streaming callers use [`LlSegmenter::push`] /
    /// [`LlSegmenter::take_ready`] instead.
    type Output = Vec<Chunk>;
    type Error = Error;

    /// Package a complete [`Media`] into its full ordered chunk list.
    ///
    /// Samples are pushed **interleaved by decode time** across tracks so each
    /// segment carries the audio temporally aligned to its video (a per-track
    /// batch push would dump all audio into segment 1). At equal decode times the
    /// non-anchor tracks are pushed *before* the anchor, so a segment-boundary
    /// keyframe cut sees that segment's audio already buffered. Then flush.
    ///
    /// # Errors
    /// [`Error::InvalidInput`] on an empty track list, or a chunk-build failure.
    fn package(&mut self, media: &Media) -> Result<Vec<Chunk>> {
        if media.tracks.is_empty() {
            return Err(Error::InvalidInput("cannot package a Media with no tracks"));
        }
        let anchor_id = self.tracks[self.anchor].spec.track_id;

        // A cursor per input track: next-sample index + its decode time in
        // *seconds* (cross-track comparable despite differing timescales),
        // accumulated from sample durations.
        struct Cursor<'a> {
            track_id: TrackId,
            timescale: u64,
            samples: &'a [Sample],
            idx: usize,
            dts_ticks: u64,
            /// Same duration-then-dts-delta rule the push path uses, so the
            /// merge order does not collapse for samples carrying
            /// `duration: Some(0)`.
            clock: MediaClock,
            is_anchor: bool,
        }
        let mut cursors: Vec<Cursor<'_>> = media
            .tracks
            .iter()
            .map(|t| Cursor {
                track_id: t.spec.track_id,
                timescale: (t.spec.timescale as u64).max(1),
                samples: &t.samples,
                idx: 0,
                dts_ticks: 0,
                clock: MediaClock::new(),
                is_anchor: t.spec.track_id == anchor_id,
            })
            .collect();

        // Merge: repeatedly push the track whose next sample has the smallest
        // decode time (seconds = dts_ticks / timescale, compared cross-multiplied
        // to stay integer). Anchor loses ties so audio at the same instant is
        // buffered first.
        loop {
            let mut best: Option<usize> = None;
            for (i, c) in cursors.iter().enumerate() {
                if c.idx >= c.samples.len() {
                    continue;
                }
                best = Some(match best {
                    None => i,
                    Some(b) => {
                        // c.dts (sec) < best.dts (sec)  ⇔  c.ticks*best.ts < best.ticks*c.ts
                        let lhs = c.dts_ticks as u128 * cursors[b].timescale as u128;
                        let rhs = cursors[b].dts_ticks as u128 * c.timescale as u128;
                        if lhs < rhs || (lhs == rhs && !c.is_anchor && cursors[b].is_anchor) {
                            i
                        } else {
                            b
                        }
                    }
                });
            }
            let Some(i) = best else { break };
            let (track_id, sample) = {
                let c = &mut cursors[i];
                let s = c.samples[c.idx].clone();
                c.dts_ticks += c.clock.tick(&s);
                c.idx += 1;
                (c.track_id, s)
            };
            self.push(track_id, sample)?;
        }

        self.flush()?;
        Ok(self.take_ready())
    }
}

/// Build one CMAF **chunk** (`[styp] moof mdat`) over a sub-run of samples,
/// mirroring [`crate::build_media_segment`] exactly (same `trun` flags, same
/// `default-base-is-moof` offset resolution) so a segment's chunks concatenate to
/// the whole-segment coded-sample stream. `with_styp` gates the leading `styp`
/// (present on the first chunk of a segment only).
///
/// Shared with the LL-HLS **partial-segment** builder
/// ([`crate::ll_hls`]): an LL-HLS part is exactly this fragment structure scoped
/// to a sub-duration, so both layers emit byte-compatible `moof`+`mdat`.
pub(crate) fn build_chunk(
    sequence_number: u32,
    tracks: &[FragmentTrackData<'_>],
    with_styp: bool,
) -> Result<Vec<u8>> {
    let styp = with_styp.then(|| SegmentTypeBox {
        major_brand: STYP_MAJOR_BRAND,
        minor_version: 0,
        compatible_brands: STYP_COMPATIBLE_BRANDS.to_vec(),
    });

    let mut traf_boxes = Vec::with_capacity(tracks.len());
    for ft in tracks {
        let (tfhd, trun) = media_fragment_run(
            ft.track_id,
            ft.samples.iter().map(|s| MediaRunSample {
                duration: s.duration.unwrap_or(0),
                size: s.data.len(),
                is_sync: s.flags.is_sync,
                composition_offset: s.composition_offset(),
            }),
        )?;
        let tfdt = TrackFragmentBaseMediaDecodeTimeBox::new_v1(ft.base_media_decode_time);
        traf_boxes.push(TrackFragmentBox::new(tfhd, Some(tfdt), vec![trun]));
    }

    let mut moof = MovieFragmentBox::new(MovieFragmentHeaderBox::new(sequence_number), traf_boxes);

    // default-base-is-moof: data_offset measured from the moof start; the mdat
    // payload begins at moof_size + 8 (the mdat header).
    let moof_size = moof.serialized_len();
    let mut cursor = moof_size + 8;
    let mut mdat_data = Vec::new();
    for (i, ft) in tracks.iter().enumerate() {
        moof.traf[i].trun[0].data_offset = Some(cursor as i32);
        for s in ft.samples {
            mdat_data.extend_from_slice(&s.data);
            cursor += s.data.len();
        }
    }
    let mdat = MediaDataBox { data: mdat_data };

    let styp_len = styp.as_ref().map_or(0, |s| s.serialized_len());
    let total = styp_len + moof.serialized_len() + mdat.serialized_len();
    let mut out = vec![0u8; total];
    let mut c = 0usize;
    if let Some(styp) = &styp {
        c += styp.serialize_into(&mut out[c..])?;
    }
    c += moof.serialize_into(&mut out[c..])?;
    c += mdat.serialize_into(&mut out[c..])?;
    out.truncate(c);
    Ok(out)
}

// The LL-DASH MPD packager renders and rewrites XML (quick-xml), so it is
// `std`-only; the chunked-CMAF segmenter above stays `no_std` + `alloc`.
#[cfg(feature = "std")]
mod mpd;
#[cfg(feature = "std")]
pub use mpd::*;