dash_runtime/timing/
clock.rs

1//! The UTC clock: an anchor pair mapping the caller's monotonic microseconds
2//! to UTC, with a trust-ordered replacement rule.
3//!
4//! `docs/dash-runtime/DESIGN.md` 3.2/3.3. Pure: the clock never reads a time
5//! source, it only maps instants it is given.
6
7use broadcast_common::impl_spec_display;
8use broadcast_common::time::MonoUtc;
9
10use super::utc_timing::UtcSample;
11
12/// Where an anchor came from, ordered by trust (lowest first).
13#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
14#[non_exhaustive]
15pub enum ClockSource {
16    /// The `Date` of the MPD response.
17    MpdDate,
18    /// The host's own wall clock.
19    Host,
20    /// A `UTCTiming` measurement.
21    UtcTiming,
22}
23
24impl ClockSource {
25    /// Stable snake_case label.
26    #[must_use]
27    pub fn name(&self) -> &'static str {
28        match self {
29            Self::MpdDate => "mpd_date",
30            Self::Host => "host",
31            Self::UtcTiming => "utc_timing",
32        }
33    }
34}
35impl_spec_display!(ClockSource);
36
37/// One anchor: at monotonic `mono_us` UTC was `utc_us`.
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39#[non_exhaustive]
40pub struct Anchor {
41    /// Monotonic microseconds.
42    pub mono_us: u64,
43    /// Unix microseconds.
44    pub utc_us: i64,
45    /// Where it came from.
46    pub source: ClockSource,
47    /// Half-width of the error bar.
48    pub uncertainty_us: u64,
49}
50
51impl Anchor {
52    /// Anchor from a measured sample.
53    #[must_use]
54    pub fn from_sample(sample: UtcSample, source: ClockSource) -> Self {
55        Self {
56            mono_us: sample.mono_us,
57            utc_us: sample.utc_us,
58            source,
59            uncertainty_us: sample.uncertainty_us,
60        }
61    }
62
63    /// UTC at `now_us`, saturating at the `i64` range.
64    #[must_use]
65    pub fn utc_at(&self, now_us: u64) -> i64 {
66        let m = MonoUtc::new(self.mono_us, i128::from(self.utc_us));
67        saturate(m.utc_at(now_us))
68    }
69
70    /// The monotonic instant at which UTC reaches `utc_us`, saturating at
71    /// `0..=u64::MAX`.
72    #[must_use]
73    pub fn mono_for_utc(&self, utc_us: i64) -> u64 {
74        let m = MonoUtc::new(self.mono_us, i128::from(self.utc_us));
75        let mono = m.mono_for_utc(i128::from(utc_us));
76        u64::try_from(mono.clamp(0, i128::from(u64::MAX))).unwrap_or(u64::MAX)
77    }
78}
79
80fn saturate(v: i128) -> i64 {
81    i64::try_from(v.clamp(i128::from(i64::MIN), i128::from(i64::MAX))).unwrap_or(i64::MAX)
82}
83
84/// Why an offered sample replaced the anchor.
85#[derive(Debug, Clone, Copy, PartialEq, Eq)]
86#[non_exhaustive]
87pub enum AdoptReason {
88    /// There was no anchor.
89    First,
90    /// The source is more trusted than the current one.
91    HigherTrust,
92    /// Same source, at least as precise.
93    Refined,
94    /// Same source, the old anchor was older than the resync period.
95    Stale,
96    /// Same source, the time disagrees with the prediction by more than both
97    /// error bars: a clock step.
98    Step,
99    /// A less trusted source, allowed because the better one failed.
100    Downgrade,
101}
102
103impl AdoptReason {
104    /// Stable snake_case label.
105    #[must_use]
106    pub fn name(&self) -> &'static str {
107        match self {
108            Self::First => "first",
109            Self::HigherTrust => "higher_trust",
110            Self::Refined => "refined",
111            Self::Stale => "stale",
112            Self::Step => "step",
113            Self::Downgrade => "downgrade",
114        }
115    }
116}
117impl_spec_display!(AdoptReason);
118
119/// Why an offered sample was not adopted.
120#[derive(Debug, Clone, Copy, PartialEq, Eq)]
121#[non_exhaustive]
122pub enum RejectReason {
123    /// A more trusted anchor exists.
124    LowerTrust,
125    /// Same source but less precise and not stale.
126    LessPrecise,
127}
128
129impl RejectReason {
130    /// Stable snake_case label.
131    #[must_use]
132    pub fn name(&self) -> &'static str {
133        match self {
134            Self::LowerTrust => "lower_trust",
135            Self::LessPrecise => "less_precise",
136        }
137    }
138}
139impl_spec_display!(RejectReason);
140
141/// The result of [`UtcClock::offer`].
142#[derive(Debug, Clone, Copy, PartialEq, Eq)]
143#[non_exhaustive]
144pub enum OfferOutcome {
145    /// The sample is now the anchor.
146    Adopted(AdoptReason),
147    /// The existing anchor was kept.
148    Rejected(RejectReason),
149}
150
151/// Parameters of the replacement rule.
152#[derive(Debug, Clone, Copy, PartialEq, Eq)]
153#[non_exhaustive]
154pub struct OfferPolicy {
155    /// An anchor older than this is replaced by any same-source sample.
156    pub resync_us: u64,
157    /// Let a less trusted source replace a more trusted one (the better one
158    /// has failed for the configured wait).
159    pub allow_downgrade: bool,
160}
161
162impl OfferPolicy {
163    /// A policy.
164    #[must_use]
165    pub fn new(resync_us: u64, allow_downgrade: bool) -> Self {
166        Self {
167            resync_us,
168            allow_downgrade,
169        }
170    }
171}
172
173/// The clock: at most one anchor.
174#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
175pub struct UtcClock {
176    anchor: Option<Anchor>,
177}
178
179impl UtcClock {
180    /// A clock with no anchor.
181    #[must_use]
182    pub fn new() -> Self {
183        Self::default()
184    }
185
186    /// The current anchor.
187    #[must_use]
188    pub fn anchor(&self) -> Option<&Anchor> {
189        self.anchor.as_ref()
190    }
191
192    /// UTC at `now_us`, or `None` before the first anchor.
193    #[must_use]
194    pub fn utc_at(&self, now_us: u64) -> Option<i64> {
195        self.anchor.map(|a| a.utc_at(now_us))
196    }
197
198    /// The monotonic instant at which UTC reaches `utc_us`.
199    #[must_use]
200    pub fn mono_for_utc(&self, utc_us: i64) -> Option<u64> {
201        self.anchor.map(|a| a.mono_for_utc(utc_us))
202    }
203
204    /// Offer a measurement; the anchor changes per the replacement rule.
205    ///
206    /// A more trusted source replaces at once; a less trusted one only when
207    /// `policy.allow_downgrade`. For the same source: a reading that disagrees
208    /// with the prediction by more than the two error bars together is a clock
209    /// step and is adopted immediately; otherwise it replaces the anchor when
210    /// it is at least as precise or the anchor is older than
211    /// `policy.resync_us`.
212    pub fn offer(
213        &mut self,
214        sample: UtcSample,
215        source: ClockSource,
216        now_us: u64,
217        policy: OfferPolicy,
218    ) -> OfferOutcome {
219        let new = Anchor::from_sample(sample, source);
220        let Some(old) = self.anchor else {
221            self.anchor = Some(new);
222            return OfferOutcome::Adopted(AdoptReason::First);
223        };
224        let reason = if new.source > old.source {
225            AdoptReason::HigherTrust
226        } else if new.source < old.source {
227            if policy.allow_downgrade {
228                AdoptReason::Downgrade
229            } else {
230                return OfferOutcome::Rejected(RejectReason::LowerTrust);
231            }
232        } else {
233            let predicted = i128::from(old.utc_at(new.mono_us));
234            let diff = (i128::from(new.utc_us) - predicted).unsigned_abs();
235            let bars = u128::from(old.uncertainty_us) + u128::from(new.uncertainty_us);
236            if diff > bars {
237                AdoptReason::Step
238            } else if now_us.saturating_sub(old.mono_us) >= policy.resync_us {
239                AdoptReason::Stale
240            } else if new.uncertainty_us <= old.uncertainty_us {
241                AdoptReason::Refined
242            } else {
243                return OfferOutcome::Rejected(RejectReason::LessPrecise);
244            }
245        };
246        self.anchor = Some(new);
247        OfferOutcome::Adopted(reason)
248    }
249}
dash_runtime/timing/clock.rs

dash_runtime/timing/
clock.rs
