nautilus_common/live/
clock.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Live clock implementation using Tokio for real-time operations.
17
18use std::{collections::BTreeMap, ops::Deref, sync::Arc};
19
20use nautilus_core::{
21    AtomicTime, DurationNanos, UnixNanos, correctness::check_predicate_true,
22    time::get_atomic_clock_realtime,
23};
24use ustr::Ustr;
25
26use super::timer::LiveTimer;
27use crate::{
28    clock::{
29        CallbackRegistry, Clock, replace_existing_timer, validate_and_prepare_schedule,
30        validate_and_prepare_time_alert, validate_and_prepare_timer,
31    },
32    runner::{TimeEventSender, purge_closed_time_event_callbacks, try_get_time_event_sender},
33    timer::{TimeEventCallback, TimerInterval, TimerSchedule, create_valid_interval},
34};
35
36/// A real-time clock which uses system time.
37///
38/// Timestamps are guaranteed to be unique and monotonically increasing.
39///
40/// # Threading
41///
42/// The clock holds thread-local runtime state and must remain on its originating thread.
43#[derive(Debug)]
44pub struct LiveClock {
45    time: &'static AtomicTime,
46    timers: BTreeMap<Ustr, LiveTimer>,
47    callbacks: CallbackRegistry,
48    sender: Option<Arc<dyn TimeEventSender>>,
49    sender_deferred: bool,
50}
51
52impl LiveClock {
53    /// Creates a new [`LiveClock`] instance.
54    #[must_use]
55    pub fn new(sender: Option<Arc<dyn TimeEventSender>>) -> Self {
56        Self {
57            time: get_atomic_clock_realtime(),
58            timers: BTreeMap::new(),
59            callbacks: CallbackRegistry::new(),
60            sender,
61            sender_deferred: false,
62        }
63    }
64
65    fn clear_expired_timers(&mut self) {
66        self.timers.retain(|_, timer| !timer.is_expired());
67        purge_closed_time_event_callbacks();
68    }
69
70    fn replace_existing_timer_if_needed(&mut self, name: &Ustr) {
71        replace_existing_timer(&mut self.timers, name);
72    }
73    fn register_timer(
74        &mut self,
75        name: Ustr,
76        interval: TimerInterval,
77        start_time_ns: UnixNanos,
78        stop_time_ns: Option<UnixNanos>,
79        callback: Option<TimeEventCallback>,
80        fire_immediately: bool,
81    ) -> anyhow::Result<()> {
82        check_predicate_true(
83            callback.is_some() | self.callbacks.has_any_callback(&name),
84            "No callbacks provided",
85        )?;
86
87        self.replace_existing_timer_if_needed(&name);
88        let callback = if let Some(callback) = callback {
89            self.callbacks.register_callback(name, callback.clone());
90            callback
91        } else {
92            self.callbacks
93                .get_callback(&name)
94                .expect("Callback should exist")
95        };
96
97        let sender = self.resolve_time_event_sender();
98
99        let mut timer = LiveTimer::new(
100            name,
101            interval,
102            start_time_ns,
103            stop_time_ns,
104            callback,
105            fire_immediately,
106            sender,
107        );
108        timer.start();
109
110        self.clear_expired_timers();
111        self.timers.insert(name, timer);
112
113        Ok(())
114    }
115}
116
117impl Default for LiveClock {
118    /// Creates a new default [`LiveClock`] instance.
119    ///
120    /// Uses `try_get_time_event_sender()` to allow creation before channels are initialized.
121    fn default() -> Self {
122        let mut clock = Self::new(try_get_time_event_sender());
123        clock.sender_deferred = clock.sender.is_none();
124        clock
125    }
126}
127
128impl Deref for LiveClock {
129    type Target = AtomicTime;
130
131    fn deref(&self) -> &Self::Target {
132        self.time
133    }
134}
135
136impl Clock for LiveClock {
137    fn timestamp_ns(&self) -> UnixNanos {
138        self.time.get_time_ns()
139    }
140
141    fn timestamp_us(&self) -> u64 {
142        self.time.get_time_us()
143    }
144
145    fn timestamp_ms(&self) -> u64 {
146        self.time.get_time_ms()
147    }
148
149    fn timestamp(&self) -> f64 {
150        self.time.get_time()
151    }
152
153    fn timer_names(&self) -> Vec<&str> {
154        self.timers
155            .iter()
156            .filter(|(_, timer)| !timer.is_expired())
157            .map(|(k, _)| k.as_str())
158            .collect()
159    }
160
161    fn timer_count(&self) -> usize {
162        self.timers
163            .iter()
164            .filter(|(_, timer)| !timer.is_expired())
165            .count()
166    }
167
168    fn timer_exists(&self, name: &Ustr) -> bool {
169        self.timers
170            .get(name)
171            .is_some_and(|timer| !timer.is_expired())
172    }
173
174    fn register_default_handler(&mut self, handler: TimeEventCallback) {
175        self.callbacks.register_default_handler(handler);
176    }
177
178    fn cancel_default_handler(&mut self) {
179        self.callbacks.cancel_default_handler();
180    }
181
182    fn cancel_callbacks(&mut self) {
183        self.callbacks.clear();
184    }
185
186    fn set_time_alert_ns(
187        &mut self,
188        name: &str,
189        alert_time_ns: UnixNanos,
190        callback: Option<TimeEventCallback>,
191        allow_past: Option<bool>,
192    ) -> anyhow::Result<()> {
193        let ts_now = self.get_time_ns();
194        let (name, alert_time_ns) =
195            validate_and_prepare_time_alert(name, alert_time_ns, allow_past, ts_now)?;
196
197        check_predicate_true(
198            callback.is_some() | self.callbacks.has_any_callback(&name),
199            "No callbacks provided",
200        )?;
201
202        self.replace_existing_timer_if_needed(&name);
203
204        let callback = if let Some(callback) = callback {
205            self.callbacks.register_callback(name, callback.clone());
206            callback
207        } else {
208            self.callbacks
209                .get_callback(&name)
210                .expect("Callback should exist")
211        };
212
213        // Safe to calculate interval now that we've ensured alert_time_ns >= ts_now
214        let interval_ns = create_valid_interval(alert_time_ns - ts_now);
215        let fire_immediately = alert_time_ns == ts_now;
216        let sender = self.resolve_time_event_sender();
217
218        let mut timer = LiveTimer::new(
219            name,
220            interval_ns,
221            ts_now,
222            Some(alert_time_ns),
223            callback,
224            fire_immediately,
225            sender,
226        );
227
228        timer.start();
229
230        self.clear_expired_timers();
231        self.timers.insert(name, timer);
232
233        Ok(())
234    }
235
236    fn set_timer_ns(
237        &mut self,
238        name: &str,
239        interval_ns: DurationNanos,
240        start_time_ns: Option<UnixNanos>,
241        stop_time_ns: Option<UnixNanos>,
242        callback: Option<TimeEventCallback>,
243        allow_past: Option<bool>,
244        fire_immediately: Option<bool>,
245    ) -> anyhow::Result<()> {
246        let (name, start_time_ns, stop_time_ns, _, fire_immediately) = validate_and_prepare_timer(
247            name,
248            interval_ns,
249            start_time_ns,
250            stop_time_ns,
251            allow_past,
252            fire_immediately,
253            self.get_time_ns(),
254        )?;
255        self.register_timer(
256            name,
257            TimerInterval::Fixed(create_valid_interval(interval_ns)),
258            start_time_ns,
259            stop_time_ns,
260            callback,
261            fire_immediately,
262        )
263    }
264
265    fn set_timer_schedule(
266        &mut self,
267        name: &str,
268        schedule: Arc<dyn TimerSchedule>,
269        stop_time_ns: Option<UnixNanos>,
270        callback: Option<TimeEventCallback>,
271        allow_past: Option<bool>,
272        fire_immediately: Option<bool>,
273    ) -> anyhow::Result<()> {
274        let start_time_ns = schedule.start_time_ns();
275        let interval = TimerInterval::Schedule(schedule);
276        let (name, start_time_ns, stop_time_ns, _, fire_immediately) =
277            validate_and_prepare_schedule(
278                name,
279                &interval,
280                Some(start_time_ns),
281                stop_time_ns,
282                allow_past,
283                fire_immediately,
284                self.get_time_ns(),
285            )?;
286        self.register_timer(
287            name,
288            interval,
289            start_time_ns,
290            stop_time_ns,
291            callback,
292            fire_immediately,
293        )
294    }
295
296    fn next_time_ns(&self, name: &str) -> Option<UnixNanos> {
297        self.timers
298            .get(&Ustr::from(name))
299            .filter(|timer| !timer.is_expired())
300            .map(LiveTimer::next_time_ns)
301    }
302
303    fn cancel_timer(&mut self, name: &str) {
304        let timer = self.timers.remove(&Ustr::from(name));
305        if let Some(mut timer) = timer {
306            timer.cancel();
307        }
308    }
309
310    fn cancel_timers(&mut self) {
311        for timer in &mut self.timers.values_mut() {
312            timer.cancel();
313        }
314
315        self.timers.clear();
316    }
317
318    fn reset(&mut self) {
319        self.cancel_timers();
320        self.callbacks.clear();
321    }
322}
323
324impl LiveClock {
325    fn resolve_time_event_sender(&mut self) -> Option<Arc<dyn TimeEventSender>> {
326        if self.sender.is_none() && self.sender_deferred {
327            self.sender = try_get_time_event_sender();
328        }
329
330        self.sender.clone()
331    }
332}
333
334#[cfg(test)]
335#[cfg(not(all(feature = "simulation", madsim)))]
336mod tests {
337    use std::{
338        sync::{
339            Arc,
340            atomic::{AtomicBool, Ordering},
341            mpsc,
342        },
343        time::Duration,
344    };
345
346    use nautilus_core::{DurationNanos, UnixNanos, time::get_atomic_clock_realtime};
347    use parking_lot::Mutex;
348    use rstest::rstest;
349    use ustr::Ustr;
350
351    use super::*;
352    use crate::{
353        calendar::CalendarSchedule,
354        clock::Clock,
355        runner::{TimeEventMessage, TimeEventSender, replace_time_event_sender},
356        testing::wait_until,
357        timer::{TimeEvent, TimeEventCallback},
358    };
359
360    #[derive(Debug)]
361    struct CollectingSender {
362        events: Arc<Mutex<Vec<(TimeEvent, UnixNanos)>>>,
363    }
364
365    impl CollectingSender {
366        fn new(events: Arc<Mutex<Vec<(TimeEvent, UnixNanos)>>>) -> Self {
367            Self { events }
368        }
369    }
370
371    impl TimeEventSender for CollectingSender {
372        fn send(&self, message: TimeEventMessage) {
373            let now_ns = get_atomic_clock_realtime().get_time_ns();
374            let event = message.event().clone();
375            message.dispatch();
376            self.events.lock().push((event, now_ns));
377        }
378    }
379
380    #[derive(Debug)]
381    struct PausingCollectingSender {
382        collector: CollectingSender,
383        paused_tx: mpsc::Sender<()>,
384        release_rx: Mutex<mpsc::Receiver<()>>,
385        pause_once: AtomicBool,
386    }
387
388    impl PausingCollectingSender {
389        fn new(
390            events: Arc<Mutex<Vec<(TimeEvent, UnixNanos)>>>,
391        ) -> (Arc<Self>, mpsc::Receiver<()>, mpsc::Sender<()>) {
392            let (paused_tx, paused_rx) = mpsc::channel();
393            let (release_tx, release_rx) = mpsc::channel();
394            let sender = Arc::new(Self {
395                collector: CollectingSender::new(events),
396                paused_tx,
397                release_rx: Mutex::new(release_rx),
398                pause_once: AtomicBool::new(true),
399            });
400            (sender, paused_rx, release_tx)
401        }
402    }
403
404    impl TimeEventSender for PausingCollectingSender {
405        fn send(&self, message: TimeEventMessage) {
406            self.collector.send(message);
407
408            if self.pause_once.swap(false, Ordering::SeqCst) {
409                self.paused_tx.send(()).expect("timer send should pause");
410                self.release_rx
411                    .lock()
412                    .recv()
413                    .expect("timer send should release");
414            }
415        }
416    }
417
418    fn wait_for_events(
419        events: &Arc<Mutex<Vec<(TimeEvent, UnixNanos)>>>,
420        target: usize,
421        timeout: Duration,
422    ) {
423        wait_until(|| events.lock().len() >= target, timeout);
424    }
425
426    #[rstest]
427    fn test_live_clock_timer_replacement_cancels_previous_task() {
428        let events = Arc::new(Mutex::new(Vec::new()));
429        let (sender, paused_rx, release_tx) = PausingCollectingSender::new(Arc::clone(&events));
430
431        let mut clock = LiveClock::new(Some(sender));
432        clock.register_default_handler(TimeEventCallback::from(|_| {}));
433
434        let fast_interval = DurationNanos::from_millis(10);
435        clock
436            .set_timer_ns("replace", fast_interval, None, None, None, None, None)
437            .unwrap();
438
439        paused_rx
440            .recv_timeout(Duration::from_secs(2))
441            .expect("fast timer send should pause");
442        events.lock().clear();
443
444        let slow_interval = DurationNanos::from_millis(30);
445        clock
446            .set_timer_ns("replace", slow_interval, None, None, None, None, None)
447            .unwrap();
448        release_tx.send(()).expect("fast timer send should release");
449
450        wait_for_events(&events, 3, Duration::from_secs(2));
451
452        let snapshot = events.lock().clone();
453        let diffs: Vec<DurationNanos> = snapshot
454            .array_windows()
455            .map(|[a, b]| b.0.ts_event - a.0.ts_event)
456            .collect();
457
458        assert!(!diffs.is_empty());
459        for diff in diffs {
460            assert_eq!(diff, slow_interval);
461        }
462
463        clock.cancel_timers();
464    }
465
466    #[rstest]
467    fn test_live_clock_time_alert_persists_callback() {
468        let events = Arc::new(Mutex::new(Vec::new()));
469        let sender = Arc::new(CollectingSender::new(Arc::clone(&events)));
470
471        let mut clock = LiveClock::new(Some(sender));
472        clock.register_default_handler(TimeEventCallback::from(|_| {}));
473
474        let now = clock.timestamp_ns();
475        let alert_time = now + DurationNanos::from_mins(1);
476
477        clock
478            .set_time_alert_ns("alert-callback", alert_time, None, None)
479            .unwrap();
480
481        assert!(
482            clock
483                .callbacks
484                .has_any_callback(&Ustr::from("alert-callback"))
485        );
486
487        clock.cancel_timers();
488    }
489
490    #[rstest]
491    fn test_default_live_clock_resolves_sender_after_initialization() {
492        std::thread::spawn(|| {
493            let events = Arc::new(Mutex::new(Vec::new()));
494            let sender = Arc::new(CollectingSender::new(Arc::clone(&events)));
495            let mut clock = LiveClock::default();
496            assert!(clock.sender.is_none());
497
498            replace_time_event_sender(sender);
499            let mut explicit_senderless = LiveClock::new(None);
500            assert!(explicit_senderless.resolve_time_event_sender().is_none());
501
502            let alert_time = clock.timestamp_ns();
503            clock
504                .set_time_alert_ns(
505                    "late-sender",
506                    alert_time,
507                    Some(TimeEventCallback::from(|_| {})),
508                    None,
509                )
510                .unwrap();
511            wait_for_events(&events, 1, Duration::from_secs(2));
512
513            assert!(clock.sender.is_some());
514            let events = events.lock();
515            assert_eq!(events.len(), 1);
516            assert_eq!(events[0].0.name, Ustr::from("late-sender"));
517        })
518        .join()
519        .expect("live clock sender test thread should join");
520    }
521
522    #[rstest]
523    fn test_live_clock_reset_stops_active_timers() {
524        let events = Arc::new(Mutex::new(Vec::new()));
525        let (sender, paused_rx, release_tx) = PausingCollectingSender::new(Arc::clone(&events));
526
527        let mut clock = LiveClock::new(Some(sender.clone()));
528
529        clock
530            .set_timer_ns(
531                "reset-test",
532                DurationNanos::from_millis(15),
533                None,
534                None,
535                Some(TimeEventCallback::from(|_| {})),
536                None,
537                None,
538            )
539            .unwrap();
540
541        paused_rx
542            .recv_timeout(Duration::from_secs(2))
543            .expect("timer send should pause");
544
545        assert_eq!(events.lock().len(), 1);
546        assert_eq!(clock.timer_count(), 1);
547
548        clock.reset();
549        release_tx.send(()).expect("timer send should release");
550
551        // Only the test and clock retain the sender after the canceled task exits
552        wait_until(|| Arc::strong_count(&sender) == 2, Duration::from_secs(2));
553
554        assert_eq!(events.lock().len(), 1);
555        assert_eq!(clock.timer_count(), 0);
556        assert!(clock.timer_names().is_empty());
557        assert!(!clock.callbacks.has_any_callback(&Ustr::from("reset-test")));
558    }
559
560    #[rstest]
561    fn test_live_clock_timer_exists_consistent_after_expiry() {
562        let events = Arc::new(Mutex::new(Vec::new()));
563        let (sender, paused_rx, release_tx) = PausingCollectingSender::new(Arc::clone(&events));
564
565        let mut clock = LiveClock::new(Some(sender));
566        clock.register_default_handler(TimeEventCallback::from(|_| {}));
567
568        let name = Ustr::from("expiring");
569        let interval_ns = DurationNanos::from_millis(10);
570        let start_time = clock.timestamp_ns();
571        let stop_time = start_time + DurationNanos::from_millis(30);
572
573        clock
574            .set_timer_ns(
575                name.as_str(),
576                interval_ns,
577                Some(start_time),
578                Some(stop_time),
579                None,
580                None,
581                None,
582            )
583            .unwrap();
584
585        paused_rx
586            .recv_timeout(Duration::from_secs(2))
587            .expect("timer send should pause");
588
589        assert!(clock.timer_exists(&name));
590        release_tx.send(()).expect("timer send should release");
591
592        // Wait for the timer task to run past its stop time and finish
593        wait_until(|| clock.timer_count() == 0, Duration::from_secs(2));
594
595        // An expired timer is purged only lazily on the next set/cancel call,
596        // so the entry still sits in the map; the introspection surfaces
597        // must nevertheless agree it is gone
598        assert!(clock.timers.contains_key(&name));
599        assert!(!clock.timer_exists(&name));
600        assert_eq!(clock.timer_count(), 0);
601        assert!(clock.timer_names().is_empty());
602        assert!(clock.next_time_ns(name.as_str()).is_none());
603    }
604
605    #[rstest]
606    fn test_live_clock_failed_set_time_alert_ns_preserves_existing_timer() {
607        let events = Arc::new(Mutex::new(Vec::new()));
608        let sender = Arc::new(CollectingSender::new(Arc::clone(&events)));
609
610        // No default handler registered
611        let mut clock = LiveClock::new(Some(sender));
612
613        let now = clock.timestamp_ns();
614        let alert_time = now + DurationNanos::from_mins(1);
615
616        clock
617            .set_time_alert_ns(
618                "alert",
619                alert_time,
620                Some(TimeEventCallback::from(|_| {})),
621                None,
622            )
623            .unwrap();
624        assert_eq!(clock.next_time_ns("alert"), Some(alert_time));
625
626        // Callbacks released (e.g. partial component teardown) while the alert still lives
627        clock.cancel_callbacks();
628
629        // Rescheduling without a callback fails the predicate check; the error
630        // return must not have destroyed the previously scheduled alert
631        let err = clock
632            .set_time_alert_ns("alert", alert_time + DurationNanos::new(1000), None, None)
633            .unwrap_err();
634        assert!(
635            err.to_string().contains("No callbacks provided"),
636            "unexpected error: {err}"
637        );
638        assert!(clock.timer_exists(&Ustr::from("alert")));
639        assert_eq!(clock.next_time_ns("alert"), Some(alert_time));
640
641        clock.cancel_timers();
642    }
643    #[rstest]
644    fn test_live_clock_registers_deadline_schedule_with_default_callback_and_inclusive_stop() {
645        let events = Arc::new(Mutex::new(Vec::new()));
646        let sender = Arc::new(CollectingSender::new(events.clone()));
647        let mut clock = LiveClock::new(Some(sender));
648        clock.register_default_handler(TimeEventCallback::from(|_| {}));
649        let origin = clock.timestamp_ns();
650        let first = origin + DurationNanos::from_millis(200);
651
652        let schedule = Arc::new(
653            CalendarSchedule::new(
654                &origin.to_datetime_utc().to_zoned(jiff::tz::TimeZone::UTC),
655                jiff::Span::new().milliseconds(200),
656            )
657            .unwrap(),
658        );
659        clock
660            .set_timer_schedule("deadline", schedule, Some(first), None, None, None)
661            .unwrap();
662        assert_eq!(clock.next_time_ns("deadline"), Some(first));
663        wait_until(|| clock.timer_count() == 0, Duration::from_secs(2));
664        let recorded = events.lock();
665        assert_eq!(recorded.len(), 1);
666        assert_eq!(recorded[0].0.name, Ustr::from("deadline"));
667        assert_eq!(recorded[0].0.ts_event, first);
668        assert!(recorded[0].0.ts_init >= first);
669        assert_eq!(clock.next_time_ns("deadline"), None);
670    }
671}
nautilus_common/live/clock.rs

nautilus_common/live/
clock.rs
