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

nautilus_common/live/
clock.rs
