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

nautilus_common/live/
clock.rs
