Skip to main content

sedsnet/
timesync.rs

1//! Time synchronization primitives, packet helpers, and network-clock utilities.
2//!
3//! This module:
4//! - Defines the public time sync configuration and role-selection types.
5//! - Encodes and decodes time sync announce/request/response packets.
6//! - Tracks remote time sources and elects the current synchronization leader.
7//! - Maintains partial or complete network time assembled from one or more sources.
8//! - Provides a slewed clock that converges toward sampled network time without hard jumps.
9
10use alloc::collections::BTreeMap;
11use alloc::string::{String, ToString};
12use alloc::vec::Vec;
13use core::mem::size_of;
14
15use crate::router::{Router, encode_slice_le};
16use crate::{
17    DataEndpoint, DataType, TelemetryError, TelemetryResult, config::runtime_device_identifier,
18    message_meta, packet::Packet,
19};
20
21/// Number of `u64` words carried by a time sync announce payload.
22pub const TIMESYNC_ANNOUNCE_WORDS: usize = 2;
23/// Number of `u64` words carried by a time sync request payload.
24pub const TIMESYNC_REQUEST_WORDS: usize = 2;
25/// Number of `u64` words carried by a time sync response payload.
26pub const TIMESYNC_RESPONSE_WORDS: usize = 4;
27/// Synthetic source identifier used for a remotely learned full-network-time source.
28pub const INTERNAL_TIMESYNC_SOURCE_ID: &str = "__timesync_remote__";
29/// Synthetic source identifier used for a locally supplied complete date+time source.
30pub const LOCAL_TIMESYNC_FULL_SOURCE_ID: &str = "__timesync_local_full__";
31/// Synthetic source identifier used for a locally supplied date-only source.
32pub const LOCAL_TIMESYNC_DATE_SOURCE_ID: &str = "__timesync_local_date__";
33/// Synthetic source identifier used for a locally supplied time-of-day source.
34pub const LOCAL_TIMESYNC_TOD_SOURCE_ID: &str = "__timesync_local_tod__";
35/// Synthetic source identifier used for a locally supplied sub-second source.
36pub const LOCAL_TIMESYNC_SUBSEC_SOURCE_ID: &str = "__timesync_local_subsec__";
37
38/// Declares how a node participates in time synchronization.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum TimeSyncRole {
41    /// Never originates time unless promoted by consumer fallback rules.
42    Consumer,
43    /// Always advertises itself as a time source.
44    Source,
45    /// Advertises itself only when it has usable time and no better source is active.
46    Auto,
47}
48
49/// Configures time sync behavior for a router or tracker.
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51pub struct TimeSyncConfig {
52    /// Local participation mode used for leader election and serving behavior.
53    pub role: TimeSyncRole,
54    /// Lower values are preferred during source selection.
55    pub priority: u64,
56    /// Maximum age before an announced source is considered inactive.
57    pub source_timeout_ms: u64,
58    /// Interval between local announce packets while acting as leader.
59    pub announce_interval_ms: u64,
60    /// Interval between request packets while following a remote source.
61    pub request_interval_ms: u64,
62    /// Allows a consumer with usable local time to promote itself when no remote source exists.
63    pub consumer_promotion_enabled: bool,
64    /// Maximum slew rate applied by [`SlewedNetworkClock`] in parts per million.
65    pub max_slew_ppm: u32,
66}
67
68impl Default for TimeSyncConfig {
69    fn default() -> Self {
70        Self {
71            role: TimeSyncRole::Consumer,
72            priority: 100,
73            source_timeout_ms: 5_000,
74            announce_interval_ms: 1_000,
75            request_interval_ms: 1_000,
76            consumer_promotion_enabled: true,
77            max_slew_ppm: 50_000,
78        }
79    }
80}
81
82/// Describes the currently elected time sync leader.
83#[derive(Debug, Clone, PartialEq, Eq)]
84pub enum TimeSyncLeader {
85    /// The local node is the current leader at the given effective priority.
86    Local { priority: u64 },
87    /// A remote node is the current leader.
88    Remote(TimeSyncSource),
89}
90
91/// Snapshot of a remote time source learned from announce traffic.
92#[derive(Debug, Clone, PartialEq, Eq)]
93pub struct TimeSyncSource {
94    /// Sender identifier for the remote source.
95    pub sender: String,
96    /// Advertised source priority. Lower is better.
97    pub priority: u64,
98    /// Monotonic receive time, in milliseconds, of the last announce.
99    pub last_announce_ms: u64,
100    /// Network time value carried by the last announce, in Unix milliseconds.
101    pub last_time_ms: u64,
102}
103
104/// Reports whether tracker state changed after processing or pruning sources.
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
106pub enum TimeSyncUpdate {
107    /// The active source selection stayed the same.
108    NoChange,
109    /// The active source selection changed.
110    SourceChanged,
111}
112
113/// Decoded fields from a time sync request packet.
114#[derive(Debug, Clone, Copy, PartialEq, Eq)]
115pub struct TimeSyncRequestFields {
116    /// Request sequence number chosen by the requester.
117    pub seq: u64,
118    /// Requester transmit timestamp in monotonic milliseconds.
119    pub t1_ms: u64,
120}
121
122/// Decoded fields from a time sync response packet.
123#[derive(Debug, Clone, Copy, PartialEq, Eq)]
124pub struct TimeSyncResponseFields {
125    /// Sequence number copied from the corresponding request.
126    pub seq: u64,
127    /// Original requester transmit timestamp in monotonic milliseconds.
128    pub t1_ms: u64,
129    /// Responder receive timestamp in network milliseconds.
130    pub t2_ms: u64,
131    /// Responder transmit timestamp in network milliseconds.
132    pub t3_ms: u64,
133}
134
135/// Decoded fields from a time sync announce packet.
136#[derive(Debug, Clone, Copy, PartialEq, Eq)]
137pub struct TimeSyncAnnounceFields {
138    /// Advertised source priority. Lower is better.
139    pub priority: u64,
140    /// Advertised network time in Unix milliseconds.
141    pub time_ms: u64,
142}
143
144/// Result of a four-timestamp offset/delay computation.
145#[derive(Debug, Clone, Copy, PartialEq, Eq)]
146pub struct TimeSyncSample {
147    /// Estimated clock offset in milliseconds from requester to responder.
148    pub offset_ms: i64,
149    /// Estimated round-trip path delay in milliseconds.
150    pub delay_ms: u64,
151}
152
153/// Partial calendar/time-of-day reading assembled from one or more sources.
154#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
155pub struct PartialNetworkTime {
156    /// Calendar year, if known.
157    pub year: Option<i32>,
158    /// Calendar month in the range `1..=12`, if known.
159    pub month: Option<u8>,
160    /// Day of month in the range `1..=31`, if known.
161    pub day: Option<u8>,
162    /// Hour of day in the range `0..=23`, if known.
163    pub hour: Option<u8>,
164    /// Minute in the range `0..=59`, if known.
165    pub minute: Option<u8>,
166    /// Second in the range `0..=59`, if known.
167    pub second: Option<u8>,
168    /// Nanosecond within the current second, if known.
169    pub nanosecond: Option<u32>,
170}
171
172impl PartialNetworkTime {
173    /// Builds a complete partial-time value from Unix milliseconds when conversion succeeds.
174    pub fn from_unix_ms(unix_ms: u64) -> Self {
175        let unix_ns = (unix_ms as i128) * 1_000_000;
176        if let Some(full) = NetworkTime::from_unix_ns(unix_ns) {
177            Self::from(full)
178        } else {
179            Self::default()
180        }
181    }
182
183    /// Returns `true` when year, month, and day are all present.
184    pub fn is_complete_date(&self) -> bool {
185        self.year.is_some() && self.month.is_some() && self.day.is_some()
186    }
187
188    /// Returns `true` when hour, minute, and second are all present.
189    pub fn is_complete_time(&self) -> bool {
190        self.hour.is_some() && self.minute.is_some() && self.second.is_some()
191    }
192
193    /// Converts this value into a full [`NetworkTime`] when all required fields are present.
194    pub fn to_network_time(&self) -> Option<NetworkTime> {
195        Some(NetworkTime {
196            year: self.year?,
197            month: self.month?,
198            day: self.day?,
199            hour: self.hour?,
200            minute: self.minute?,
201            second: self.second?,
202            nanosecond: self.nanosecond.unwrap_or(0),
203        })
204    }
205}
206
207impl From<NetworkTime> for PartialNetworkTime {
208    fn from(value: NetworkTime) -> Self {
209        Self {
210            year: Some(value.year),
211            month: Some(value.month),
212            day: Some(value.day),
213            hour: Some(value.hour),
214            minute: Some(value.minute),
215            second: Some(value.second),
216            nanosecond: Some(value.nanosecond),
217        }
218    }
219}
220
221/// Fully specified calendar and time-of-day in UTC-like Unix time representation.
222#[derive(Debug, Clone, Copy, PartialEq, Eq)]
223pub struct NetworkTime {
224    /// Calendar year.
225    pub year: i32,
226    /// Calendar month in the range `1..=12`.
227    pub month: u8,
228    /// Day of month in the range `1..=31`.
229    pub day: u8,
230    /// Hour of day in the range `0..=23`.
231    pub hour: u8,
232    /// Minute in the range `0..=59`.
233    pub minute: u8,
234    /// Second in the range `0..=59`.
235    pub second: u8,
236    /// Nanosecond within the current second.
237    pub nanosecond: u32,
238}
239
240impl NetworkTime {
241    /// Converts a non-negative Unix nanosecond timestamp into calendar/time fields.
242    pub fn from_unix_ns(unix_ns: i128) -> Option<Self> {
243        if unix_ns < 0 {
244            return None;
245        }
246
247        let secs = u64::try_from(unix_ns / 1_000_000_000).ok()?;
248        let nanos = unix_ns.rem_euclid(1_000_000_000) as u32;
249        let days = secs / 86_400;
250        let sod = (secs % 86_400) as u32;
251        let (year, month, day) = civil_from_days(days)?;
252
253        Some(Self {
254            year,
255            month: month as u8,
256            day: day as u8,
257            hour: (sod / 3_600) as u8,
258            minute: ((sod % 3_600) / 60) as u8,
259            second: (sod % 60) as u8,
260            nanosecond: nanos,
261        })
262    }
263
264    /// Converts this calendar/time value into Unix nanoseconds if the fields are valid.
265    pub fn as_unix_ns(&self) -> Option<i128> {
266        if !(1..=12).contains(&self.month)
267            || !(1..=31).contains(&self.day)
268            || self.hour > 23
269            || self.minute > 59
270            || self.second > 59
271            || self.nanosecond >= 1_000_000_000
272        {
273            return None;
274        }
275
276        let days = days_from_civil(self.year, self.month as u32, self.day as u32);
277        let secs = (days as i128) * 86_400
278            + (self.hour as i128) * 3_600
279            + (self.minute as i128) * 60
280            + (self.second as i128);
281        Some(secs * 1_000_000_000 + self.nanosecond as i128)
282    }
283
284    /// Converts this calendar/time value into non-negative Unix milliseconds.
285    pub fn as_unix_ms(&self) -> Option<u64> {
286        let unix_ns = self.as_unix_ns()?;
287        if unix_ns < 0 {
288            return None;
289        }
290        u64::try_from(unix_ns / 1_000_000).ok()
291    }
292}
293
294/// Best-effort current network-time reading from the assembled clock state.
295#[derive(Debug, Clone, PartialEq, Eq)]
296pub struct NetworkTimeReading {
297    /// Partial or complete calendar/time-of-day reading.
298    pub time: PartialNetworkTime,
299    /// Absolute Unix milliseconds when the reading is complete enough to derive them.
300    pub unix_time_ms: Option<u64>,
301}
302
303/// Monotonic-anchor clock that slews toward target Unix time instead of stepping immediately.
304#[derive(Debug, Clone, Copy, PartialEq, Eq)]
305pub struct SlewedNetworkClock {
306    anchor_mono_ns: u64,
307    anchor_unix_ns: i128,
308    pending_adjust_ns: i128,
309    max_slew_ppm: u32,
310    initialized: bool,
311}
312
313impl Default for SlewedNetworkClock {
314    fn default() -> Self {
315        Self {
316            anchor_mono_ns: 0,
317            anchor_unix_ns: 0,
318            pending_adjust_ns: 0,
319            max_slew_ppm: 50_000,
320            initialized: false,
321        }
322    }
323}
324
325impl SlewedNetworkClock {
326    /// Creates a new slewed clock with the given slew-rate cap in parts per million.
327    pub fn new(max_slew_ppm: u32) -> Self {
328        Self {
329            max_slew_ppm: max_slew_ppm.min(999_999),
330            ..Default::default()
331        }
332    }
333
334    /// Returns `true` once the clock has received an initial anchor time.
335    pub fn is_initialized(&self) -> bool {
336        self.initialized
337    }
338
339    /// Reads the current estimated Unix time in nanoseconds from a monotonic timestamp.
340    pub fn read_unix_ns(&self, now_mono_ns: u64) -> Option<i128> {
341        if !self.initialized {
342            return None;
343        }
344
345        let elapsed_ns = now_mono_ns.saturating_sub(self.anchor_mono_ns) as i128;
346        let max_adjust_ns = elapsed_ns.saturating_mul(self.max_slew_ppm as i128) / 1_000_000;
347        let applied_adjust_ns = if self.pending_adjust_ns >= 0 {
348            self.pending_adjust_ns.min(max_adjust_ns)
349        } else {
350            -(-self.pending_adjust_ns).min(max_adjust_ns)
351        };
352
353        Some(
354            self.anchor_unix_ns
355                .saturating_add(elapsed_ns)
356                .saturating_add(applied_adjust_ns),
357        )
358    }
359
360    /// Reads the current estimated Unix time in milliseconds from a monotonic timestamp.
361    pub fn read_unix_ms(&self, now_mono_ns: u64) -> Option<u64> {
362        let unix_ns = self.read_unix_ns(now_mono_ns)?;
363        if unix_ns < 0 {
364            return None;
365        }
366        u64::try_from(unix_ns / 1_000_000).ok()
367    }
368
369    /// Steers the clock toward a target Unix millisecond reading at the configured slew rate.
370    pub fn steer_unix_ms(&mut self, now_mono_ns: u64, target_unix_ms: u64) {
371        let target_unix_ns = (target_unix_ms as i128) * 1_000_000;
372        if !self.initialized {
373            self.anchor_mono_ns = now_mono_ns;
374            self.anchor_unix_ns = target_unix_ns;
375            self.pending_adjust_ns = 0;
376            self.initialized = true;
377            return;
378        }
379
380        let current_unix_ns = self.read_unix_ns(now_mono_ns).unwrap_or(target_unix_ns);
381        self.anchor_mono_ns = now_mono_ns;
382        self.anchor_unix_ns = current_unix_ns;
383        // The target is an absolute observation at `now`, not an incremental
384        // correction. Repeated observations must replace the residual error.
385        self.pending_adjust_ns = target_unix_ns.saturating_sub(current_unix_ns);
386        self.initialized = true;
387    }
388}
389
390#[derive(Debug, Clone)]
391struct NetworkTimeSourceState {
392    priority: u64,
393    updated_mono_ms: u64,
394    anchor_mono_ns: u64,
395    ttl_ms: Option<u64>,
396    time: PartialNetworkTime,
397}
398
399/// Aggregates multiple partial or complete time sources into a current network-time reading.
400#[derive(Debug, Clone, Default)]
401pub struct NetworkClock {
402    sources: BTreeMap<String, NetworkTimeSourceState>,
403}
404
405impl NetworkClock {
406    /// Inserts or replaces a named source used when assembling current network time.
407    pub fn update_source(
408        &mut self,
409        source: &str,
410        priority: u64,
411        time: PartialNetworkTime,
412        updated_mono_ms: u64,
413        anchor_mono_ns: u64,
414        ttl_ms: Option<u64>,
415    ) {
416        self.sources.insert(
417            source.to_string(),
418            NetworkTimeSourceState {
419                priority,
420                updated_mono_ms,
421                anchor_mono_ns,
422                ttl_ms,
423                time,
424            },
425        );
426    }
427
428    /// Removes a previously registered source by identifier.
429    pub fn remove_source(&mut self, source: &str) {
430        self.sources.remove(source);
431    }
432
433    /// Drops sources whose optional TTL has expired relative to `now_mono_ms`.
434    pub fn prune_expired(&mut self, now_mono_ms: u64) {
435        self.sources.retain(|_, src| {
436            src.ttl_ms
437                .map(|ttl| now_mono_ms.saturating_sub(src.updated_mono_ms) <= ttl)
438                .unwrap_or(true)
439        });
440    }
441
442    /// Returns the best current reading by preferring complete sources, then merging partial ones.
443    pub fn current_time(&self, now_mono_ns: u64) -> Option<NetworkTimeReading> {
444        let sources: Vec<(&str, &NetworkTimeSourceState)> = self
445            .sources
446            .iter()
447            .map(|(name, src)| (name.as_str(), src))
448            .collect();
449        if sources.is_empty() {
450            return None;
451        }
452
453        let full = best_source(&sources, |src| src.time.to_network_time().is_some());
454        if let Some((_, src)) = full
455            && let Some(base) = src.time.to_network_time()
456        {
457            let elapsed_ns = now_mono_ns.saturating_sub(src.anchor_mono_ns);
458            if let Some(advanced) = advance_network_time(base, elapsed_ns) {
459                return Some(NetworkTimeReading {
460                    unix_time_ms: advanced.as_unix_ms(),
461                    time: PartialNetworkTime::from(advanced),
462                });
463            }
464        }
465
466        let year = best_source(&sources, |src| src.time.year.is_some()).map(|(_, src)| src);
467        let month = best_source(&sources, |src| src.time.month.is_some()).map(|(_, src)| src);
468        let day = best_source(&sources, |src| src.time.day.is_some()).map(|(_, src)| src);
469        let hour = best_source(&sources, |src| src.time.hour.is_some()).map(|(_, src)| src);
470        let minute = best_source(&sources, |src| src.time.minute.is_some()).map(|(_, src)| src);
471        let second = best_source(&sources, |src| src.time.second.is_some()).map(|(_, src)| src);
472        let subsec = best_source(&sources, |src| src.time.nanosecond.is_some()).map(|(_, src)| src);
473
474        let mut merged = PartialNetworkTime {
475            year: year.and_then(|src| src.time.year),
476            month: month.and_then(|src| src.time.month),
477            day: day.and_then(|src| src.time.day),
478            hour: hour.and_then(|src| src.time.hour),
479            minute: minute.and_then(|src| src.time.minute),
480            second: second.and_then(|src| src.time.second),
481            ..Default::default()
482        };
483        if let Some(src) = subsec {
484            merged.nanosecond = src.time.nanosecond;
485        }
486
487        if let Some(time_src) = second.or(minute).or(hour)
488            && merged.is_complete_date()
489            && merged.is_complete_time()
490        {
491            merged.nanosecond = merged.nanosecond.or(Some(0));
492            let base = merged.to_network_time()?;
493            let elapsed_ns = now_mono_ns.saturating_sub(time_src.anchor_mono_ns);
494            if let Some(advanced) = advance_network_time(base, elapsed_ns) {
495                return Some(NetworkTimeReading {
496                    unix_time_ms: advanced.as_unix_ms(),
497                    time: PartialNetworkTime::from(advanced),
498                });
499            }
500        }
501
502        Some(NetworkTimeReading {
503            unix_time_ms: None,
504            time: merged,
505        })
506    }
507}
508
509fn best_source<'a, F>(
510    sources: &'a [(&'a str, &'a NetworkTimeSourceState)],
511    predicate: F,
512) -> Option<(&'a str, &'a NetworkTimeSourceState)>
513where
514    F: Fn(&NetworkTimeSourceState) -> bool,
515{
516    let mut best: Option<(&str, &NetworkTimeSourceState)> = None;
517    for (name, src) in sources.iter().copied() {
518        if !predicate(src) {
519            continue;
520        }
521        best = match best {
522            None => Some((name, src)),
523            Some((best_name, best_src)) => {
524                if src.priority < best_src.priority
525                    || (src.priority == best_src.priority && name < best_name)
526                {
527                    Some((name, src))
528                } else {
529                    Some((best_name, best_src))
530                }
531            }
532        };
533    }
534    best
535}
536
537/// Advances a complete [`NetworkTime`] forward by an elapsed monotonic duration.
538pub(crate) fn advance_network_time(base: NetworkTime, elapsed_ns: u64) -> Option<NetworkTime> {
539    let unix_ns = base.as_unix_ns()?;
540    NetworkTime::from_unix_ns(unix_ns + elapsed_ns as i128)
541}
542
543fn days_from_civil(year: i32, month: u32, day: u32) -> i64 {
544    let mut era = year.div_euclid(400);
545    let mut yoe = year.rem_euclid(400) - (month <= 2) as i32;
546    if yoe < 0 {
547        era -= 1;
548        yoe += 400;
549    }
550    let mp = month as i32 + if month > 2 { -3 } else { 9 };
551    let doy = (153 * mp + 2) / 5 + day as i32 - 1;
552    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
553    era as i64 * 146_097 + doe as i64 - 719_468
554}
555
556fn civil_from_days(mut z: u64) -> Option<(i32, u32, u32)> {
557    z += 719_468;
558    let era = z / 146_097;
559    // A Gregorian era contains exactly 146097 days. Its remaining
560    // arithmetic fits in 32 bits, avoiding repeated software i64 divisions
561    // on Cortex-M while retaining the full representable year range.
562    let doe = (z - era * 146_097) as u32;
563    let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
564    let y = yoe as u64 + era * 400;
565    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
566    let mp = (5 * doy + 2) / 153;
567    let d = doy - (153 * mp + 2) / 5 + 1;
568    let m = if mp < 10 { mp + 3 } else { mp - 9 };
569    let y = i32::try_from(y + (m <= 2) as u64).ok()?;
570    Some((y, m, d))
571}
572
573/// Tracks remote time sources and performs leader/source selection decisions.
574#[derive(Debug, Clone)]
575pub struct TimeSyncTracker {
576    cfg: TimeSyncConfig,
577    local_id: String,
578    sources: BTreeMap<String, TimeSyncSource>,
579    current_source: Option<String>,
580}
581
582impl TimeSyncTracker {
583    /// Creates a tracker with the supplied time sync configuration.
584    pub fn new(cfg: TimeSyncConfig) -> Self {
585        Self {
586            cfg,
587            local_id: runtime_device_identifier(),
588            sources: BTreeMap::new(),
589            current_source: None,
590        }
591    }
592
593    /// Returns the tracker's active configuration.
594    pub fn config(&self) -> TimeSyncConfig {
595        self.cfg
596    }
597
598    /// Returns the currently selected remote source, if any.
599    pub fn current_source(&self) -> Option<&TimeSyncSource> {
600        self.current_source
601            .as_ref()
602            .and_then(|sender| self.sources.get(sender))
603    }
604
605    /// Removes stale sources and re-runs source selection.
606    pub fn refresh(&mut self, now_ms: u64) -> TimeSyncUpdate {
607        self.sources
608            .retain(|_, src| Self::source_is_active(src, now_ms, self.cfg.source_timeout_ms));
609        self.reselect_source(now_ms)
610    }
611
612    /// Returns the best currently active remote source according to priority and sender ID.
613    pub fn best_active_source(&self, now_ms: u64) -> Option<&TimeSyncSource> {
614        self.sources
615            .values()
616            .filter(|src| Self::source_is_active(src, now_ms, self.cfg.source_timeout_ms))
617            .min_by(|a, b| {
618                a.priority
619                    .cmp(&b.priority)
620                    .then_with(|| a.sender.cmp(&b.sender))
621            })
622    }
623
624    /// Returns the effective local candidate priority if the node may lead at `now_ms`.
625    pub fn local_candidate_priority(&self, now_ms: u64, has_usable_time: bool) -> Option<u64> {
626        match self.cfg.role {
627            TimeSyncRole::Consumer => {
628                if has_usable_time
629                    && self.cfg.consumer_promotion_enabled
630                    && self.best_active_source(now_ms).is_none()
631                {
632                    Some(self.cfg.priority)
633                } else {
634                    None
635                }
636            }
637            TimeSyncRole::Source => Some(self.cfg.priority),
638            TimeSyncRole::Auto => {
639                if has_usable_time && self.best_active_source(now_ms).is_none() {
640                    Some(self.cfg.priority)
641                } else {
642                    None
643                }
644            }
645        }
646    }
647
648    /// Returns the elected leader between the local node and active remote sources.
649    pub fn leader(&self, now_ms: u64, has_usable_time: bool) -> Option<TimeSyncLeader> {
650        let local_priority = self.local_candidate_priority(now_ms, has_usable_time);
651        let remote = self.best_active_source(now_ms).cloned();
652        match (local_priority, remote) {
653            (Some(priority), Some(remote)) => {
654                if priority < remote.priority
655                    || (priority == remote.priority
656                        && self.local_id.as_str() < remote.sender.as_str())
657                {
658                    Some(TimeSyncLeader::Local { priority })
659                } else {
660                    Some(TimeSyncLeader::Remote(remote))
661                }
662            }
663            (Some(priority), None) => Some(TimeSyncLeader::Local { priority }),
664            (None, Some(remote)) => Some(TimeSyncLeader::Remote(remote)),
665            (None, None) => None,
666        }
667    }
668
669    /// Returns the priority to advertise in local announce packets when leading.
670    pub fn local_announce_priority(&self, now_ms: u64, has_usable_time: bool) -> Option<u64> {
671        let Some(TimeSyncLeader::Local { priority }) = self.leader(now_ms, has_usable_time) else {
672            return None;
673        };
674        let tie_exists = self
675            .sources
676            .values()
677            .filter(|src| Self::source_is_active(src, now_ms, self.cfg.source_timeout_ms))
678            .any(|src| src.priority == priority);
679        Some(if tie_exists {
680            priority.saturating_sub(1)
681        } else {
682            priority
683        })
684    }
685
686    /// Returns `true` when the local node should currently emit announce packets.
687    pub fn should_announce(&self, now_ms: u64, has_usable_time: bool) -> bool {
688        self.local_announce_priority(now_ms, has_usable_time)
689            .is_some()
690    }
691
692    /// Returns `true` when the local node should currently answer incoming requests.
693    pub fn should_serve(&self, now_ms: u64, has_usable_time: bool) -> bool {
694        self.should_announce(now_ms, has_usable_time)
695    }
696
697    /// Updates tracker state from an incoming announce packet.
698    pub fn handle_announce(
699        &mut self,
700        pkt: &Packet,
701        recv_ms: u64,
702    ) -> TelemetryResult<TimeSyncUpdate> {
703        let ann = decode_timesync_announce(pkt)?;
704        if pkt.sender() == self.local_id.as_str() {
705            return Ok(TimeSyncUpdate::NoChange);
706        }
707
708        let incoming = TimeSyncSource {
709            sender: pkt.sender().to_string(),
710            priority: ann.priority,
711            last_announce_ms: recv_ms,
712            last_time_ms: ann.time_ms,
713        };
714        self.sources.insert(incoming.sender.clone(), incoming);
715        Ok(self.reselect_source(recv_ms))
716    }
717
718    /// Returns `true` when the current remote source is still within the active timeout window.
719    pub fn is_source_active(&self, now_ms: u64) -> bool {
720        self.current_source()
721            .map(|s| Self::source_is_active(s, now_ms, self.cfg.source_timeout_ms))
722            .unwrap_or(false)
723    }
724
725    fn source_is_active(src: &TimeSyncSource, now_ms: u64, timeout_ms: u64) -> bool {
726        now_ms.saturating_sub(src.last_announce_ms) <= timeout_ms
727    }
728
729    fn best_active_source_id(&self, now_ms: u64) -> Option<String> {
730        self.best_active_source(now_ms)
731            .map(|src| src.sender.clone())
732    }
733
734    fn reselect_source(&mut self, now_ms: u64) -> TimeSyncUpdate {
735        let prev = self.current_source.clone();
736        self.current_source = self.best_active_source_id(now_ms);
737        if self.current_source == prev {
738            TimeSyncUpdate::NoChange
739        } else {
740            TimeSyncUpdate::SourceChanged
741        }
742    }
743}
744
745/// Computes clock offset and round-trip delay from the standard four timestamps.
746pub fn compute_offset_delay(t1_ms: u64, t2_ms: u64, t3_ms: u64, t4_ms: u64) -> TimeSyncSample {
747    let t1 = t1_ms as i128;
748    let t2 = t2_ms as i128;
749    let t3 = t3_ms as i128;
750    let t4 = t4_ms as i128;
751
752    let offset = ((t2 - t1) + (t3 - t4)) / 2;
753    let delay = (t4 - t1) - (t3 - t2);
754    let delay_ms = if delay < 0 { 0 } else { delay as u64 };
755
756    TimeSyncSample {
757        offset_ms: offset as i64,
758        delay_ms,
759    }
760}
761
762/// Estimates current network time and one-way delay from a request/response exchange.
763pub fn compute_network_time_sample(
764    t1_monotonic_ms: u64,
765    t2_network_ms: u64,
766    t3_network_ms: u64,
767    t4_monotonic_ms: u64,
768) -> (u64, u64) {
769    let round_trip = t4_monotonic_ms.saturating_sub(t1_monotonic_ms);
770    let server_processing = t3_network_ms.saturating_sub(t2_network_ms);
771    let one_way_delay = round_trip.saturating_sub(server_processing) / 2;
772    let estimated_network_ms = t3_network_ms.saturating_add(one_way_delay);
773    (estimated_network_ms, one_way_delay)
774}
775
776/// Builds a time sync announce packet using the schema-configured sender metadata.
777pub fn build_timesync_announce(priority: u64, time_ms: u64) -> TelemetryResult<Packet> {
778    let meta = message_meta(DataType::TimeSyncAnnounce);
779    Packet::from_u64_slice(
780        DataType::TimeSyncAnnounce,
781        &[priority, time_ms],
782        meta.endpoints_ref(),
783        time_ms,
784    )
785}
786
787/// Builds a time sync announce packet with an explicit sender identifier.
788pub fn build_timesync_announce_with_sender(
789    sender: &'static str,
790    priority: u64,
791    time_ms: u64,
792) -> TelemetryResult<Packet> {
793    let payload = encode_slice_le(&[priority, time_ms]);
794    Packet::new(
795        DataType::TimeSyncAnnounce,
796        &[DataEndpoint::TimeSync],
797        sender,
798        time_ms,
799        payload,
800    )
801}
802
803/// Queues a time sync announce message through the router telemetry path.
804pub fn send_timesync_announce(router: &Router, priority: u64, time_ms: u64) -> TelemetryResult<()> {
805    router.log_ts(DataType::TimeSyncAnnounce, time_ms, &[priority, time_ms])
806}
807
808/// Builds a time sync request packet.
809pub fn build_timesync_request(seq: u64, t1_ms: u64) -> TelemetryResult<Packet> {
810    let meta = message_meta(DataType::TimeSyncRequest);
811    Packet::from_u64_slice(
812        DataType::TimeSyncRequest,
813        &[seq, t1_ms],
814        meta.endpoints_ref(),
815        t1_ms,
816    )
817}
818
819/// Queues a time sync request message through the router telemetry path.
820pub fn send_timesync_request(router: &Router, seq: u64, t1_ms: u64) -> TelemetryResult<()> {
821    router.log_ts(DataType::TimeSyncRequest, t1_ms, &[seq, t1_ms])
822}
823
824/// Builds a time sync response packet.
825pub fn build_timesync_response(
826    seq: u64,
827    t1_ms: u64,
828    t2_ms: u64,
829    t3_ms: u64,
830) -> TelemetryResult<Packet> {
831    let meta = message_meta(DataType::TimeSyncResponse);
832    Packet::from_u64_slice(
833        DataType::TimeSyncResponse,
834        &[seq, t1_ms, t2_ms, t3_ms],
835        meta.endpoints_ref(),
836        t3_ms,
837    )
838}
839
840/// Queues a time sync response message through the router telemetry path.
841pub fn send_timesync_response(
842    router: &Router,
843    seq: u64,
844    t1_ms: u64,
845    t2_ms: u64,
846    t3_ms: u64,
847) -> TelemetryResult<()> {
848    router.log_ts(
849        DataType::TimeSyncResponse,
850        t3_ms,
851        &[seq, t1_ms, t2_ms, t3_ms],
852    )
853}
854
855fn decode_u64_payload(
856    pkt: &Packet,
857    expected_ty: DataType,
858    expected_words: usize,
859) -> TelemetryResult<Vec<u64>> {
860    if pkt.data_type() != expected_ty {
861        return Err(TelemetryError::InvalidType);
862    }
863
864    let vals = pkt.data_as_u64()?;
865    if vals.len() != expected_words {
866        return Err(TelemetryError::SizeMismatch {
867            expected: expected_words * size_of::<u64>(),
868            got: vals.len() * size_of::<u64>(),
869        });
870    }
871
872    Ok(vals)
873}
874
875/// Decodes a time sync announce packet into strongly typed fields.
876pub fn decode_timesync_announce(pkt: &Packet) -> TelemetryResult<TimeSyncAnnounceFields> {
877    let vals = decode_u64_payload(pkt, DataType::TimeSyncAnnounce, TIMESYNC_ANNOUNCE_WORDS)?;
878    Ok(TimeSyncAnnounceFields {
879        priority: vals[0],
880        time_ms: vals[1],
881    })
882}
883
884/// Decodes a time sync request packet into strongly typed fields.
885pub fn decode_timesync_request(pkt: &Packet) -> TelemetryResult<TimeSyncRequestFields> {
886    let vals = decode_u64_payload(pkt, DataType::TimeSyncRequest, TIMESYNC_REQUEST_WORDS)?;
887    Ok(TimeSyncRequestFields {
888        seq: vals[0],
889        t1_ms: vals[1],
890    })
891}
892
893/// Decodes a time sync response packet into strongly typed fields.
894pub fn decode_timesync_response(pkt: &Packet) -> TelemetryResult<TimeSyncResponseFields> {
895    let vals = decode_u64_payload(pkt, DataType::TimeSyncResponse, TIMESYNC_RESPONSE_WORDS)?;
896    Ok(TimeSyncResponseFields {
897        seq: vals[0],
898        t1_ms: vals[1],
899        t2_ms: vals[2],
900        t3_ms: vals[3],
901    })
902}
903
904#[cfg(test)]
905mod calendar_tests {
906    use super::NetworkTime;
907
908    #[test]
909    fn calendar_roundtrips_era_and_year_boundaries() {
910        for year in [1970, 1999, 2000, 2024, 2100, 2400, 9999, i32::MAX] {
911            for (month, day) in [(1, 1), (2, 28), (3, 1), (12, 31)] {
912                let expected = NetworkTime {
913                    year,
914                    month,
915                    day,
916                    hour: 23,
917                    minute: 59,
918                    second: 59,
919                    nanosecond: 999_999_999,
920                };
921                assert_eq!(
922                    NetworkTime::from_unix_ns(expected.as_unix_ns().unwrap()),
923                    Some(expected)
924                );
925            }
926        }
927        let leap = NetworkTime {
928            year: 2024,
929            month: 2,
930            day: 29,
931            hour: 0,
932            minute: 0,
933            second: 0,
934            nanosecond: 123,
935        };
936        assert_eq!(
937            NetworkTime::from_unix_ns(leap.as_unix_ns().unwrap()),
938            Some(leap)
939        );
940        assert_eq!(NetworkTime::from_unix_ns(0).unwrap().year, 1970);
941    }
942
943    #[test]
944    fn out_of_range_calendar_timestamps_are_rejected_without_wraparound() {
945        assert_eq!(NetworkTime::from_unix_ns(-1), None);
946        assert_eq!(NetworkTime::from_unix_ns(i128::MAX), None);
947        let last = NetworkTime {
948            year: i32::MAX,
949            month: 12,
950            day: 31,
951            hour: 23,
952            minute: 59,
953            second: 59,
954            nanosecond: 999_999_999,
955        }
956        .as_unix_ns()
957        .unwrap();
958        assert_eq!(NetworkTime::from_unix_ns(last + 1), None);
959    }
960}
961
962#[cfg(test)]
963mod slew_absolute_target_tests {
964    use super::SlewedNetworkClock;
965
966    #[test]
967    fn remote_source_projection_survives_embedded_polling_and_source_updates() {
968        use super::{TimeSyncConfig, TimeSyncRole, build_timesync_announce_with_sender};
969        use crate::router::{Router, RouterConfig};
970        use alloc::sync::Arc;
971        use core::sync::atomic::{AtomicU64, Ordering};
972        const EPOCH: u64 = 1_767_225_600_000;
973        let now = Arc::new(AtomicU64::new(0));
974        let clock = now.clone();
975        let router = Router::new_with_clock(
976            RouterConfig::default().with_timesync(TimeSyncConfig {
977                role: TimeSyncRole::Consumer,
978                priority: 50,
979                source_timeout_ms: 10_000,
980                consumer_promotion_enabled: false,
981                ..Default::default()
982            }),
983            alloc::boxed::Box::new(move || clock.load(Ordering::SeqCst)),
984        );
985        for elapsed_ms in (0..=20_000).step_by(10) {
986            now.store(elapsed_ms, Ordering::SeqCst);
987            if elapsed_ms % 5_000 == 0 {
988                // Exercise a replacement leader as well as repeated updates.
989                let (source, priority) = if elapsed_ms < 10_000 {
990                    ("SOURCE_A", 2)
991                } else {
992                    ("SOURCE_B", 1)
993                };
994                router
995                    .rx(
996                        &build_timesync_announce_with_sender(source, priority, EPOCH + elapsed_ms)
997                            .unwrap(),
998                    )
999                    .unwrap();
1000            }
1001            for _ in 0..5 {
1002                router.poll_timesync().unwrap();
1003                assert_eq!(router.network_time_ms(), Some(EPOCH + elapsed_ms));
1004            }
1005        }
1006    }
1007
1008    #[test]
1009    fn repeated_absolute_target_does_not_accumulate_phase_error() {
1010        for target in [9_900, 10_100] {
1011            let mut once = SlewedNetworkClock::new(50_000);
1012            once.steer_unix_ms(0, 10_000);
1013            once.steer_unix_ms(0, target);
1014            let mut repeated = once;
1015            for _ in 0..100 {
1016                repeated.steer_unix_ms(0, target);
1017            }
1018            for elapsed_ms in [1, 100, 1_000, 2_000, 3_000, 10_000] {
1019                assert_eq!(
1020                    repeated.read_unix_ms(elapsed_ms * 1_000_000),
1021                    once.read_unix_ms(elapsed_ms * 1_000_000)
1022                );
1023            }
1024        }
1025    }
1026
1027    #[test]
1028    fn advancing_absolute_targets_converge_with_bounded_monotonic_slew() {
1029        for offset_ms in [-100i64, 100] {
1030            let mut clock = SlewedNetworkClock::new(50_000);
1031            clock.steer_unix_ms(0, 10_000);
1032            let mut previous = 10_000_000_000i128;
1033            for elapsed_ms in 1..=5_000u64 {
1034                let now_ns = elapsed_ms * 1_000_000;
1035                let target_ms = (10_000 + elapsed_ms as i64 + offset_ms) as u64;
1036                clock.steer_unix_ms(now_ns, target_ms);
1037                let current = clock.read_unix_ns(now_ns).unwrap();
1038                assert!((950_000..=1_050_000).contains(&(current - previous)));
1039                previous = current;
1040            }
1041            assert_eq!(
1042                clock.read_unix_ms(5_000_000_000),
1043                Some((15_000 + offset_ms) as u64)
1044            );
1045        }
1046    }
1047}