ll-hls-runtime 0.1.1

Sans-IO Low-Latency HLS (RFC 8216bis) client + server engines in one crate (blocking reload, part prefetch, rolling-window origin), with an optional tokio+reqwest IO adapter.
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
719
720
721
722
723
724
725
726
727
728
729
730
731
732
//! Per-stream, protocol-neutral in-RAM rolling window of the segmenter's
//! init/segments/parts, with a runtime-agnostic change notification
//! ([`event_listener::Event`]) so a blocking-reload wait works under any
//! async runtime — not `tokio::sync::watch`.
//!
//! [`MediaStore`] holds bytes + timing only — no playlist/manifest syntax.
//! Rendering a manifest (LL-HLS `#EXT-M3U8`) is [`super::media_playlist_m3u8`]'s
//! concern, layered on top.

use std::collections::VecDeque;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::SystemTime;

use event_listener::{Event, EventListener};
use transmux::ll_hls::{PartInfo, SegmentInfo};
use transmux::pipeline::TrackSpec;

/// Minimum live-parts cap, regardless of the timing-derived bound — keeps a
/// usable window even for pathologically small `target_duration_secs`/
/// `part_target_ms` configs.
const MIN_MAX_LIVE_PARTS: usize = 8;

/// Safety margin (in parts) added on top of one nominal segment's worth of
/// parts, absorbing jitter in part sizes/timing without needing an exact fit.
const MAX_LIVE_PARTS_SAFETY_MARGIN: usize = 4;

/// Bound on `Inner::live_parts` derived from segment timing: roughly one
/// nominal segment's worth of parts (`target_duration_secs / part_target`),
/// plus a small safety margin, floored at [`MIN_MAX_LIVE_PARTS`].
///
/// Without a cap, a segment that never closes (GOP much longer than
/// `target_duration_secs`, or a source that stops sending keyframes) would
/// grow `live_parts` unboundedly — `add_segment`'s clear-on-close only trims
/// it when a segment *does* close. This bound keeps RAM use flat regardless;
/// an LL-HLS playlist need only advertise the most recent parts of the open
/// segment (RFC 8216bis has no requirement to retain every part ever
/// produced for an in-progress segment).
pub(crate) fn compute_max_live_parts(target_duration_secs: f64, part_target_ms: u32) -> usize {
    let part_target_secs = f64::from(part_target_ms) / 1000.0;
    let nominal_parts = if part_target_secs > 0.0 {
        (target_duration_secs / part_target_secs).ceil() as usize
    } else {
        0
    };
    (nominal_parts + MAX_LIVE_PARTS_SAFETY_MARGIN).max(MIN_MAX_LIVE_PARTS)
}

struct Inner {
    init: Option<Vec<u8>>,
    segments: VecDeque<SegmentInfo>,
    live_parts: Vec<PartInfo>,
    /// Parts of *just-closed* segments, kept briefly (bounded, oldest-evicted)
    /// after `add_segment` moves them out of `live_parts`. They are no longer
    /// rendered in the playlist (the segment is advertised as a whole `seg-…`),
    /// but stay **fetchable** so an in-flight LL-HLS preload-hint request for a
    /// segment's *final* part still resolves: the segmenter emits that final
    /// part and closes the segment in the same pipeline step, so without this
    /// the part is evicted microseconds after it appears — before the blocked
    /// part request can wake — and every segment boundary 404s its hinted part.
    recent_parts: VecDeque<PartInfo>,
    window_segments: usize,
    /// Current ingest health, set by the feeding pipeline's supervisor (e.g.
    /// `multimux::origin::supervisor`) and read by adapters/metrics.
    health: HealthState,
    /// The largest `SegmentInfo.duration` ever seen by `add_segment`, over
    /// the whole lifetime of this store (never reset when the window slides
    /// or a segment is evicted). RFC 8216bis §4.4.3.1 requires
    /// `#EXT-X-TARGETDURATION` to be at least the rounded duration of every
    /// Media Segment ever advertised — since a real segment can exceed the
    /// *configured* target duration (the segmenter cuts on the next keyframe
    /// after the target, not exactly at it), an all-time max is the only way
    /// to guarantee the MUST holds for every segment this store has ever
    /// produced, including ones already evicted from the window.
    max_segment_duration: f64,
    /// The track specs the feeding pipeline built its segmenter from (issue
    /// #663 P4) — set once via [`MediaStore::set_track_specs`], before the
    /// first sample lands. Protocol-neutral (any `Output` can read it), but
    /// only [`crate`]'s DASH sibling in `multimux::output::dash` actually
    /// needs it today: a DASH `Representation` must advertise a real RFC 6381
    /// `codecs` string, which the LL-HLS playlist never needs.
    track_specs: Vec<TrackSpec>,
}

/// Ingest health of the pipeline feeding a [`MediaStore`], set by the
/// caller's supervisor loop (e.g. `multimux::origin::supervisor::supervise`)
/// as it connects/reconnects the source.
///
/// `Failed` is reserved for an unrecoverable connect error class; a
/// well-behaved supervisor never gives up on a route (sources like cameras
/// come back), so in practice it cycles `Connecting` -> `Live` <->
/// `Reconnecting`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum HealthState {
    /// Never yet connected; the supervisor's first connect attempt is in
    /// flight.
    Connecting,
    /// Connected and actively receiving media.
    Live,
    /// Lost the source (connect failure, pipeline error, or source EOF) and
    /// the supervisor is retrying with backoff.
    Reconnecting,
    /// Unrecoverable — the supervisor has given up on this route.
    Failed,
}

impl HealthState {
    /// The spec/field-enum label (workspace #204 convention): a stable,
    /// lowercase token per state, suitable for logs/metrics.
    pub fn name(&self) -> &'static str {
        match self {
            HealthState::Connecting => "connecting",
            HealthState::Live => "live",
            HealthState::Reconnecting => "reconnecting",
            HealthState::Failed => "failed",
        }
    }
}

impl std::fmt::Display for HealthState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.name())
    }
}

/// One closed segment's identity/timing — a protocol-neutral snapshot entry
/// returned by [`MediaStore::window_segments`] (issue #663 P4).
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct SegmentWindowEntry {
    /// This segment's sequence number — matches the `seg-{track}-{seq}.m4s`
    /// filename [`MediaStore::resolve_resource`](super::MediaStore::resolve_resource)
    /// serves.
    pub segment_seq: u32,
    /// This segment's actual duration, in seconds.
    pub duration_secs: f64,
}

/// In-RAM rolling window for one served stream: bytes + timing, shared by
/// every adapter serving that stream.
pub struct MediaStore {
    inner: Mutex<Inner>,
    target_duration_secs: f64,
    part_target_ms: u32,
    max_live_parts: usize,
    /// Monotonic version bumped by every mutation (`add_*`/`set_init`/
    /// `set_health` that actually changes health) — never reset, only ever
    /// grows (wrapping on overflow, which at any realistic mutation rate
    /// never happens within a process lifetime).
    progress_version: AtomicU64,
    /// Runtime-agnostic wakeup: `listen()` registers a listener *before*
    /// returning (so a `notify` racing the caller's next check is never
    /// missed — the standard `event-listener` idiom), and every mutation
    /// calls `notify(usize::MAX)` to wake every parked waiter.
    progress_event: Event,
    /// Wall-clock time this store was constructed — used as the live
    /// presentation's `availabilityStartTime` anchor (issue #663 P4; see
    /// [`Self::created_at`]). Not bumped/mutated after construction, so it
    /// needs no lock.
    created_at: SystemTime,
}

impl MediaStore {
    /// New empty store; `window_segments` = full segments retained.
    pub fn new(target_duration_secs: f64, part_target_ms: u32, window_segments: usize) -> Self {
        MediaStore {
            inner: Mutex::new(Inner {
                init: None,
                segments: VecDeque::new(),
                live_parts: Vec::new(),
                recent_parts: VecDeque::new(),
                window_segments,
                health: HealthState::Connecting,
                max_segment_duration: 0.0,
                track_specs: Vec::new(),
            }),
            target_duration_secs,
            part_target_ms,
            max_live_parts: compute_max_live_parts(target_duration_secs, part_target_ms),
            progress_version: AtomicU64::new(0),
            progress_event: Event::new(),
            created_at: SystemTime::now(),
        }
    }

    fn bump(&self) {
        self.progress_version.fetch_add(1, Ordering::SeqCst);
        self.progress_event.notify(usize::MAX);
    }

    /// Store the fMP4 init segment.
    pub fn set_init(&self, bytes: Vec<u8>) {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .init = Some(bytes);
        self.bump();
    }

    /// Append a completed part to the in-progress segment.
    ///
    /// Caps `live_parts` at `max_live_parts` worth of entries, dropping the
    /// *oldest* live part(s) first if the cap is exceeded — this bounds RAM
    /// use even if the current segment never closes (see
    /// `compute_max_live_parts`).
    pub fn add_part(&self, part: PartInfo) {
        let mut g = self
            .inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        g.live_parts.push(part);
        while g.live_parts.len() > self.max_live_parts {
            g.live_parts.remove(0);
        }
        drop(g);
        self.bump();
    }

    /// Count of currently-retained live parts (test accessor for the
    /// `live_parts` cap).
    #[cfg(test)]
    pub(crate) fn live_part_count(&self) -> usize {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .live_parts
            .len()
    }

    /// Close a full segment into the window (evicting the oldest). Its
    /// in-progress parts move out of `live_parts` into a bounded `recent_parts`
    /// buffer — still fetchable (so an in-flight preload-hint request for the
    /// segment's final part resolves) but no longer rendered as open parts.
    /// `recent_parts` is capped like `live_parts`, oldest-first.
    pub fn add_segment(&self, seg: SegmentInfo) {
        let mut g = self
            .inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let seq = seg.segment_seq;
        g.max_segment_duration = g.max_segment_duration.max(seg.duration);
        let (closed, still_live): (Vec<PartInfo>, Vec<PartInfo>) =
            core::mem::take(&mut g.live_parts)
                .into_iter()
                .partition(|p| p.segment_seq <= seq);
        g.live_parts = still_live;
        for p in closed {
            g.recent_parts.push_back(p);
        }
        while g.recent_parts.len() > self.max_live_parts {
            g.recent_parts.pop_front();
        }
        g.segments.push_back(seg);
        while g.segments.len() > g.window_segments {
            g.segments.pop_front();
        }
        drop(g);
        self.bump();
    }

    /// The fMP4 init segment bytes, if present — the one accessor kept
    /// public beyond `add_*`/`set_*`/`health`/`listen`, since callers (e.g.
    /// `multimux`'s pipeline/supervisor tests) commonly need to assert media
    /// has actually landed without going through `resolve_resource`.
    pub fn init_bytes(&self) -> Option<Vec<u8>> {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .init
            .clone()
    }

    /// A full segment's bytes by sequence number.
    pub(crate) fn segment_bytes(&self, seq: u32) -> Option<Vec<u8>> {
        let g = self
            .inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        g.segments
            .iter()
            .find(|s| s.segment_seq == seq)
            .map(|s| s.bytes.clone())
    }

    /// A part's bytes by (segment seq, part index). Checks the in-progress
    /// segment's `live_parts` first, then the just-closed `recent_parts` — the
    /// latter so an LL-HLS client's in-flight preload-hint request for a
    /// segment's final part still resolves after `add_segment` closed it. Parts
    /// older than the `recent_parts` bound are no longer individually
    /// addressable (only the whole segment is).
    pub(crate) fn part_bytes(&self, seq: u32, part_index: u32) -> Option<Vec<u8>> {
        let g = self
            .inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let matches = |p: &&PartInfo| p.segment_seq == seq && p.part_index == part_index;
        g.live_parts
            .iter()
            .find(matches)
            .or_else(|| g.recent_parts.iter().find(matches))
            .map(|p| p.bytes.clone())
    }

    /// `(in-progress segment seq, count of live parts available for it)` —
    /// used to resolve blocking `_HLS_msn`/`_HLS_part` requests.
    ///
    /// The second value is a **count**, not the last part's index: the
    /// blocking-reload resolver treats "part `P` ready" as `count > P` (0
    /// means no parts of the in-progress segment are available yet).
    ///
    /// `pub` (not `pub(crate)`) since issue #663 P4.2: `multimux::output::ll_dash`
    /// needs the in-progress segment's identity to address its live parts
    /// (`part-{track}-{seq}.{idx}.m4s`) from a `SegmentTemplate` — the same
    /// cross-crate exception already made for
    /// [`Self::target_duration_secs`]/[`Self::part_target_ms`]/
    /// [`Self::track_specs`].
    pub fn latest_progress(&self) -> (u32, u32) {
        let g = self
            .inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let last_closed_seg = g.segments.back().map(|s| s.segment_seq).unwrap_or(0);
        let in_progress_seg = g
            .live_parts
            .last()
            .map(|p| p.segment_seq)
            .unwrap_or(last_closed_seg);
        let part_count = g
            .live_parts
            .iter()
            .filter(|p| p.segment_seq == in_progress_seg)
            .count() as u32;
        (in_progress_seg, part_count)
    }

    /// The current monotonic progress version — bumped by every mutation.
    /// Mostly useful for a caller wanting to detect "did anything change"
    /// without registering a listener (e.g. a cheap pre-check).
    pub fn progress_version(&self) -> u64 {
        self.progress_version.load(Ordering::SeqCst)
    }

    /// Register for the next change notification. **Register before
    /// re-checking the condition you're waiting on** (see this module's
    /// `super`-level doc for the wait-loop shape) — `event-listener`
    /// guarantees any `notify` call that happens after `listen()` returns
    /// will wake this listener, so there is no missed-wakeup race as long as
    /// the re-check happens after `listen()`, not before.
    ///
    /// The returned [`EventListener`] is a plain `Future<Output = ()>` — any
    /// async runtime (or none, via its blocking `.wait()`) can drive it; this
    /// is what keeps `server` runtime-agnostic (unlike a
    /// `tokio::sync::watch::Receiver`, which only ever paired with tokio).
    pub fn listen(&self) -> EventListener {
        self.progress_event.listen()
    }

    /// Set the route's ingest health. Bumps the progress notification
    /// **only when the state actually changes**, so a caller blocked on
    /// [`Self::listen`] (e.g. an LL-HLS blocking playlist reload) wakes on a
    /// health transition too, not just new media.
    pub fn set_health(&self, state: HealthState) {
        let mut g = self
            .inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if g.health != state {
            g.health = state;
            drop(g);
            self.bump();
        }
    }

    /// The current ingest health (default [`HealthState::Connecting`] until
    /// the supervisor sets it).
    pub fn health(&self) -> HealthState {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .health
    }

    /// The full-segment target duration, in seconds, this store was built
    /// with — timing configuration a manifest renderer needs (e.g. LL-HLS's
    /// `#EXT-X-TARGETDURATION`, or a DASH `Output`'s `minimumUpdatePeriod`/
    /// `timeShiftBufferDepth`). `pub` (not `pub(crate)`) since issue #663 P4:
    /// `multimux::output::dash` needs it too, not just this crate's own
    /// `engine`.
    pub fn target_duration_secs(&self) -> f64 {
        self.target_duration_secs
    }

    /// The part target duration, in milliseconds, this store was built with.
    /// `pub` for the same cross-`Output` reason as
    /// [`Self::target_duration_secs`].
    pub fn part_target_ms(&self) -> u32 {
        self.part_target_ms
    }

    /// Wall-clock time this store was constructed (issue #663 P4) — used as
    /// a live DASH presentation's `availabilityStartTime` anchor. An
    /// approximation (the *route*'s start time, not the first segment's
    /// exact cut time — the first segment typically closes
    /// `target_duration_secs` or so later), acceptable for a manifest
    /// attribute that only needs to establish a consistent, monotonic
    /// timeline, not wall-clock precision.
    pub fn created_at(&self) -> SystemTime {
        self.created_at
    }

    /// Store the track specs the feeding pipeline built its segmenter from —
    /// called once, before the first sample is pushed. See
    /// `Inner::track_specs` for why this exists (DASH's `codecs` string
    /// needs real codec identity; LL-HLS never reads this).
    pub fn set_track_specs(&self, specs: Vec<TrackSpec>) {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .track_specs = specs;
    }

    /// The track specs set by [`Self::set_track_specs`], empty if never
    /// called (e.g. in a test that only exercises playlist rendering).
    pub fn track_specs(&self) -> Vec<TrackSpec> {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .track_specs
            .clone()
    }

    /// Snapshot of the closed segments currently retained in the rolling
    /// window, oldest first — enough for a manifest renderer to enumerate
    /// fetchable segments (issue #663 P4: DASH's `SegmentTemplate`/
    /// `$Number$` addressing) without depending on the LL-HLS-specific
    /// playlist rendering in the playlist renderer.
    pub fn window_segments(&self) -> Vec<SegmentWindowEntry> {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .segments
            .iter()
            .map(|s| SegmentWindowEntry {
                segment_seq: s.segment_seq,
                duration_secs: s.duration,
            })
            .collect()
    }

    /// The largest `SegmentInfo.duration` ever seen by [`Self::add_segment`],
    /// `0.0` if no segment has closed yet. The playlist renderer combines
    /// this with [`Self::target_duration_secs`] to compute a
    /// spec-conformant `#EXT-X-TARGETDURATION` (RFC 8216bis §4.4.3.1).
    pub(crate) fn max_segment_duration(&self) -> f64 {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .max_segment_duration
    }

    /// The sequence number of the most-recently-closed segment, `0` if none
    /// has closed yet. Unlike [`Self::latest_progress`]'s first element
    /// (which reflects the *in-progress* segment once any of its parts have
    /// landed), this is specifically the last **closed** segment — used to
    /// implement RFC 8216bis §6.2.5.2's bare-`_HLS_msn` blocking-reload
    /// semantics, which must wait for segment `msn` to be a fully-present
    /// Media Segment, not merely an in-progress one with live parts.
    pub(crate) fn last_closed_segment_seq(&self) -> u32 {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .segments
            .back()
            .map(|s| s.segment_seq)
            .unwrap_or(0)
    }

    /// Run `f` against a consistent snapshot of the closed `segments` and the
    /// in-progress segment's `live_parts`, taken under a single lock
    /// acquisition — used by the playlist renderer, which needs both
    /// collections together.
    pub(crate) fn with_segments_and_parts<R>(
        &self,
        f: impl FnOnce(&VecDeque<SegmentInfo>, &[PartInfo]) -> R,
    ) -> R {
        let g = self
            .inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        f(&g.segments, &g.live_parts)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use event_listener::Listener;

    fn seg(seq: u32, parts: u32) -> SegmentInfo {
        SegmentInfo {
            bytes: vec![seq as u8; 8],
            duration: 4.0,
            segment_seq: seq,
            part_count: parts,
        }
    }
    fn part(seq: u32, idx: u32) -> PartInfo {
        PartInfo {
            bytes: vec![idx as u8; 4],
            duration: 0.5,
            independent: idx == 0,
            segment_seq: seq,
            part_index: idx,
        }
    }

    #[test]
    fn window_evicts_oldest_and_serves_bytes() {
        let s = MediaStore::new(4.0, 500, 2);
        s.set_init(vec![0xAA; 10]);
        s.add_segment(seg(1, 8));
        s.add_segment(seg(2, 8));
        s.add_segment(seg(3, 8)); // evicts seq 1
        assert!(s.segment_bytes(1).is_none(), "seq 1 evicted");
        assert!(s.segment_bytes(2).is_some());
        assert!(s.segment_bytes(3).is_some());
        assert_eq!(s.init_bytes().unwrap(), vec![0xAA; 10]);
    }

    #[test]
    fn recent_parts_bounded_across_many_closes() {
        // Closing many segments must not grow recent_parts unboundedly.
        let s = MediaStore::new(4.0, 500, 4);
        s.set_init(vec![0; 4]);
        let cap = compute_max_live_parts(4.0, 500);
        for seq in 1..=20u32 {
            for idx in 0..4u32 {
                s.add_part(part(seq, idx));
            }
            s.add_segment(seg(seq, 4));
        }
        // Only the most recent ~cap parts remain individually fetchable; a very
        // old one has been evicted from recent_parts.
        assert!(s.part_bytes(1, 0).is_none(), "old closed part evicted");
        assert!(
            s.part_bytes(20, 3).is_some(),
            "most-recent closed part retained (within the {cap}-part bound)"
        );
    }

    #[test]
    fn progress_version_bumps_on_new_data() {
        let s = MediaStore::new(4.0, 500, 4);
        let before = s.progress_version();
        s.add_part(part(1, 0));
        assert_ne!(s.progress_version(), before, "progress version changed");
    }

    #[test]
    fn listen_wakes_on_new_data() {
        // Proves `listen()` actually observes a `notify()` triggered by a
        // mutation — not just that the version counter moves (that's
        // `progress_version_bumps_on_new_data`). Registers first (the
        // documented no-missed-wakeup ordering), then mutates, then blocks on
        // the listener with `EventListener::wait` (bounded so a broken wakeup
        // fails the test instead of hanging it).
        let s = MediaStore::new(4.0, 500, 4);
        let listener = s.listen();
        s.add_part(part(1, 0));
        assert!(
            listener
                .wait_deadline(std::time::Instant::now() + std::time::Duration::from_secs(2))
                .is_some(),
            "listener must wake within 2s of add_part"
        );
    }

    #[test]
    fn health_defaults_to_connecting() {
        let s = MediaStore::new(4.0, 500, 4);
        assert_eq!(s.health(), HealthState::Connecting);
    }

    #[test]
    fn set_health_updates_and_bumps_progress_only_on_change() {
        let s = MediaStore::new(4.0, 500, 4);
        let before = s.progress_version();

        // No-op: setting the same state again must not bump.
        s.set_health(HealthState::Connecting);
        assert_eq!(
            s.progress_version(),
            before,
            "unchanged state does not bump progress"
        );

        s.set_health(HealthState::Live);
        assert_eq!(s.health(), HealthState::Live);
        assert_ne!(
            s.progress_version(),
            before,
            "state change bumps progress so blocked readers wake"
        );

        let mid = s.progress_version();
        s.set_health(HealthState::Reconnecting);
        assert_eq!(s.health(), HealthState::Reconnecting);
        assert_ne!(s.progress_version(), mid);
    }

    #[test]
    fn max_segment_duration_tracks_lifetime_max_not_just_current_window() {
        let s = MediaStore::new(4.0, 500, 2);
        s.set_init(vec![0; 4]);
        assert_eq!(s.max_segment_duration(), 0.0, "nothing closed yet");

        let mut over = seg(1, 8);
        over.duration = 4.0;
        s.add_segment(over);
        assert_eq!(s.max_segment_duration(), 4.0);

        // A real segment that overshoots the configured target (the
        // segmenter cuts on the next keyframe after the target, so this is
        // routine, not pathological).
        let mut over = seg(2, 8);
        over.duration = 7.5;
        s.add_segment(over);
        assert_eq!(s.max_segment_duration(), 7.5);

        // Window slides (window_segments=2 evicts seq 1 and seq 2 eventually)
        // but the lifetime max must NOT reset/shrink back down.
        let mut small = seg(3, 8);
        small.duration = 3.0;
        s.add_segment(small); // evicts seq 1 from the window
        let mut small2 = seg(4, 8);
        small2.duration = 3.0;
        s.add_segment(small2); // evicts seq 2 from the window
        assert!(
            s.segment_bytes(2).is_none(),
            "seq 2 (the 7.5s segment) evicted from the window"
        );
        assert_eq!(
            s.max_segment_duration(),
            7.5,
            "lifetime max must survive window eviction"
        );
    }

    #[test]
    fn last_closed_segment_seq_tracks_the_newest_close() {
        let s = MediaStore::new(4.0, 500, 4);
        assert_eq!(s.last_closed_segment_seq(), 0, "nothing closed yet");
        s.add_segment(seg(1, 8));
        assert_eq!(s.last_closed_segment_seq(), 1);
        s.add_segment(seg(2, 8));
        assert_eq!(s.last_closed_segment_seq(), 2);
    }

    #[test]
    fn health_state_name_and_display_agree() {
        for (state, label) in [
            (HealthState::Connecting, "connecting"),
            (HealthState::Live, "live"),
            (HealthState::Reconnecting, "reconnecting"),
            (HealthState::Failed, "failed"),
        ] {
            assert_eq!(state.name(), label);
            assert_eq!(state.to_string(), label);
        }
    }

    // --- issue #663 P4: DASH-facing accessors ---

    #[test]
    fn track_specs_round_trip_and_default_empty() {
        use transmux::pipeline::CodecConfig;

        let s = MediaStore::new(4.0, 500, 4);
        assert!(
            s.track_specs().is_empty(),
            "no specs set yet -> empty, not a panic/placeholder"
        );

        let spec = TrackSpec::new(
            1,
            90_000,
            CodecConfig::Vp8 {
                width: 0,
                height: 0,
            },
        );
        s.set_track_specs(vec![spec.clone()]);
        let got = s.track_specs();
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].track_id, spec.track_id);
        assert_eq!(got[0].timescale, spec.timescale);
    }

    #[test]
    fn window_segments_reflects_closed_segments_oldest_first_and_evicts() {
        let s = MediaStore::new(4.0, 500, 2);
        assert!(s.window_segments().is_empty(), "nothing closed yet");

        s.add_segment(seg(1, 4));
        s.add_segment(seg(2, 4));
        let window = s.window_segments();
        assert_eq!(
            window.iter().map(|e| e.segment_seq).collect::<Vec<_>>(),
            vec![1, 2],
            "oldest first"
        );
        assert_eq!(window[0].duration_secs, 4.0);

        s.add_segment(seg(3, 4)); // evicts seq 1 (window_segments == 2)
        assert_eq!(
            s.window_segments()
                .iter()
                .map(|e| e.segment_seq)
                .collect::<Vec<_>>(),
            vec![2, 3],
            "eviction reflected in the snapshot"
        );
    }

    #[test]
    fn created_at_is_set_at_construction() {
        let before = SystemTime::now();
        let s = MediaStore::new(4.0, 500, 4);
        let after = SystemTime::now();
        assert!(s.created_at() >= before && s.created_at() <= after);
    }
}