playout_runtime/transition.rs
1//! Transition planning: given a [`Schedule`] and
2//! the current channel-timeline position, what transition happens next, at
3//! what instant, and what timeline-continuity actions it implies.
4//!
5//! Issue [#748](https://github.com/fishloa/rust-broadcast/issues/748)'s
6//! framing: **the hard part is the transition, not the schedule.** Joining
7//! two sources means timeline continuity across the join — a PTS-rebase
8//! offset so the incoming source's own clock lands continuously on the
9//! shared channel timeline (the parameter `transmux`'s IR transform would
10//! consume to actually rewrite sample timestamps — this crate computes the
11//! number, it does not touch sample bytes), plus a discontinuity flag when
12//! the codec configuration changes across the join (the signal
13//! `broadcast_hls::mark_init_discontinuities` would act on downstream). "A
14//! schedule that plays the right thing at the wrong timestamp is worse than
15//! no scheduler" — so [`TransitionPlan`] is the primary thing this crate's
16//! tests hold to timeline-correctness, not the schedule bookkeeping.
17
18use crate::schedule::{Schedule, ScheduleEntry};
19
20/// The computed plan for joining `from` into `to` at [`Self::at_pts`].
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub struct TransitionPlan {
23 /// The channel-timeline instant the join happens at — `to`'s
24 /// `planned_start`.
25 pub at_pts: u64,
26 /// Add this to one of `to`'s own PTS values to place it on the shared
27 /// channel timeline (see [`Self::rebase`]). `i128` so it can hold the
28 /// full possible difference of two `u64` instants without clamping —
29 /// this crate never silently narrows a rebase computation. May be
30 /// negative when `to`'s own clock numbering starts after the channel
31 /// instant it is joining at (an unusual but not-impossible schedule).
32 pub pts_rebase_offset: i128,
33 /// Whether `from` and `to` carry different
34 /// [`CodecConfigId`](crate::schedule::CodecConfigId)s — a discontinuity
35 /// that must be signalled downstream (init-segment change / new codec
36 /// config), never something this crate transcodes or conforms away.
37 pub discontinuity: bool,
38}
39
40impl TransitionPlan {
41 /// Plan the transition from `from` into `to`. `to.planned_start` is the
42 /// channel-timeline instant the join happens at; the rebase offset is
43 /// derived so that `to.source_start_pts`, once rebased, lands exactly on
44 /// that instant.
45 #[must_use]
46 pub fn plan(from: &ScheduleEntry, to: &ScheduleEntry) -> Self {
47 let at_pts = to.planned_start;
48 let pts_rebase_offset = i128::from(at_pts) - i128::from(to.source_start_pts);
49 TransitionPlan {
50 at_pts,
51 pts_rebase_offset,
52 discontinuity: from.codec_config != to.codec_config,
53 }
54 }
55
56 /// Map one of `to`'s own PTS values onto the shared channel timeline
57 /// (`source_pts + pts_rebase_offset`).
58 ///
59 /// Returns `None` rather than wrapping if the rebased instant would fall
60 /// outside `u64`'s range (e.g. a source PTS before the source's own
61 /// declared start, rebasing to before the join instant) — this crate
62 /// never silently produces a timeline instant that doesn't correspond to
63 /// the input.
64 #[must_use]
65 pub fn rebase(&self, source_pts: u64) -> Option<u64> {
66 let rebased = i128::from(source_pts) + self.pts_rebase_offset;
67 u64::try_from(rebased).ok()
68 }
69}
70
71/// A planned transition together with the two entries either side of it.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub struct PlannedTransition<'a> {
74 /// The entry currently playing.
75 pub from: &'a ScheduleEntry,
76 /// The entry the channel joins into.
77 pub to: &'a ScheduleEntry,
78 /// The computed plan for that join.
79 pub plan: TransitionPlan,
80}
81
82/// Find and plan the next transition due at or after `now_pts`.
83///
84/// Returns `None` if `now_pts` precedes every entry, or if the entry active
85/// at `now_pts` is the schedule's last entry (nothing scheduled to follow it
86/// yet).
87#[must_use]
88pub fn next_transition(schedule: &Schedule, now_pts: u64) -> Option<PlannedTransition<'_>> {
89 let idx = schedule.active_index_at(now_pts)?;
90 let entries = schedule.entries();
91 let to = entries.get(idx + 1)?;
92 let from = &entries[idx];
93 Some(PlannedTransition {
94 from,
95 to,
96 plan: TransitionPlan::plan(from, to),
97 })
98}
99
100#[cfg(test)]
101mod tests {
102 use super::*;
103 use crate::schedule::{CodecConfigId, EntryKind, ScheduleEntry};
104
105 fn entry(
106 id: &str,
107 kind: EntryKind,
108 planned_start: u64,
109 source_start_pts: u64,
110 codec_config: u64,
111 ) -> ScheduleEntry {
112 ScheduleEntry {
113 id: id.into(),
114 kind,
115 planned_start,
116 source_start_pts,
117 codec_config: CodecConfigId(codec_config),
118 }
119 }
120
121 /// The primary property under test: continuity across a join. Once
122 /// rebased, the incoming source's own PTS sequence must land exactly on
123 /// the channel timeline with no gap and no overlap — `rebase(start) ==
124 /// at_pts`, and every later source PTS advances the channel timeline by
125 /// exactly the same amount it advanced on the source's own clock.
126 #[test]
127 fn rebase_lands_the_incoming_source_continuously_on_the_channel_timeline() {
128 let programme = entry("prog-1", EntryKind::Programme, 0, 0, 1);
129 // The ad's own asset timeline starts at PTS 5_000 (e.g. it is itself
130 // a clip cut from a longer source), and is scheduled to join the
131 // channel at channel-timeline instant 1_000_000.
132 let ad = entry("ad-1", EntryKind::Ad, 1_000_000, 5_000, 1);
133
134 let plan = TransitionPlan::plan(&programme, &ad);
135 assert_eq!(plan.at_pts, 1_000_000);
136
137 // The ad's very first sample lands exactly at the join instant.
138 assert_eq!(plan.rebase(5_000), Some(1_000_000));
139 // One second later (90 kHz) on the ad's own clock is one second
140 // later on the channel timeline too — no drift introduced by the
141 // rebase itself.
142 assert_eq!(plan.rebase(5_000 + 90_000), Some(1_090_000));
143 // Ten seconds later.
144 assert_eq!(plan.rebase(5_000 + 900_000), Some(1_900_000));
145 }
146
147 #[test]
148 fn rebase_offset_can_be_negative_when_source_starts_after_the_join_instant() {
149 // A degenerate but legal schedule: the source's own clock numbering
150 // starts past the channel instant it joins at.
151 let from = entry("prog-1", EntryKind::Programme, 0, 0, 1);
152 let to = entry("ad-1", EntryKind::Ad, 100, 500, 1);
153 let plan = TransitionPlan::plan(&from, &to);
154 assert_eq!(plan.pts_rebase_offset, -400);
155 assert_eq!(plan.rebase(500), Some(100));
156 // A source PTS below its own declared start rebases to a negative
157 // channel instant — reported as `None`, never wrapped.
158 assert_eq!(plan.rebase(300), None);
159 }
160
161 #[test]
162 fn rebase_returns_none_rather_than_wrapping_out_of_range() {
163 let from = entry("prog-1", EntryKind::Programme, 0, 0, 1);
164 let to = entry("ad-1", EntryKind::Ad, 0, u64::MAX, 1);
165 let plan = TransitionPlan::plan(&from, &to);
166
167 // The source's own declared start rebases exactly onto the join
168 // instant, even though the raw offset is a huge negative i128.
169 assert_eq!(plan.rebase(u64::MAX), Some(0));
170 // Anything before the source's own start would rebase to a negative
171 // channel instant — reported honestly as `None`, not wrapped into a
172 // huge bogus `u64`.
173 assert_eq!(plan.rebase(0), None);
174 }
175
176 #[test]
177 fn discontinuity_flags_a_codec_config_change_across_the_join() {
178 let from = entry("prog-1", EntryKind::Programme, 0, 0, 1);
179 let same = entry("prog-2", EntryKind::Programme, 1_000, 0, 1);
180 let different = entry("ad-1", EntryKind::Ad, 1_000, 0, 2);
181
182 assert!(!TransitionPlan::plan(&from, &same).discontinuity);
183 assert!(TransitionPlan::plan(&from, &different).discontinuity);
184 }
185
186 #[test]
187 fn next_transition_walks_the_schedule() {
188 let mut sched = Schedule::default();
189 sched
190 .push(entry("prog-1", EntryKind::Programme, 0, 0, 1))
191 .unwrap();
192 sched
193 .push(entry("ad-1", EntryKind::Ad, 1_000, 0, 2))
194 .unwrap();
195 sched
196 .push(entry("prog-2", EntryKind::Programme, 2_000, 0, 1))
197 .unwrap();
198
199 let t = next_transition(&sched, 500).unwrap();
200 assert_eq!(t.from.id, "prog-1");
201 assert_eq!(t.to.id, "ad-1");
202 assert!(t.plan.discontinuity);
203
204 let t = next_transition(&sched, 1_500).unwrap();
205 assert_eq!(t.from.id, "ad-1");
206 assert_eq!(t.to.id, "prog-2");
207
208 // No entry follows the last one yet.
209 assert!(next_transition(&sched, 2_500).is_none());
210 // pts 0 IS prog-1's start, so the next transition is still found.
211 assert!(next_transition(&sched, 0).is_some());
212 }
213}