nautilus_common/
timer.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//! Real-time and test timers for use with `Clock` implementations.
17//!
18//! Defines [`TimeEvent`] values, callback and handler types, heap scheduling order, and the
19//! deterministic [`TestTimer`] iterator. The event and callback primitives are shared by test and
20//! live clock implementations.
21
22use std::{
23    cmp::Ordering,
24    fmt::{Debug, Display},
25    num::NonZeroU64,
26    rc::Rc,
27    sync::Arc,
28};
29
30use nautilus_core::{
31    UUID4, UnixNanos,
32    correctness::{FAILED, check_valid_string_utf8},
33};
34#[cfg(feature = "python")]
35use pyo3::{Py, PyAny, Python};
36use ustr::Ustr;
37
38/// Returns a positive nanosecond interval, coercing zero to one nanosecond.
39#[must_use]
40pub fn create_valid_interval(interval_ns: u64) -> NonZeroU64 {
41    NonZeroU64::new(interval_ns).unwrap_or(NonZeroU64::MIN)
42}
43
44#[repr(C)]
45#[derive(Clone, Debug, PartialEq, Eq)]
46#[cfg_attr(
47    feature = "python",
48    pyo3::pyclass(module = "nautilus_trader.common", from_py_object)
49)]
50#[cfg_attr(
51    feature = "python",
52    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")
53)]
54/// Represents a named timer event.
55///
56/// `ts_event` records the scheduled event time, while `ts_init` records
57/// when the event instance was initialized.
58pub struct TimeEvent {
59    /// The timer event name.
60    pub name: Ustr,
61    /// The unique identifier for the event.
62    pub event_id: UUID4,
63    /// UNIX timestamp (nanoseconds) when the event is scheduled to occur.
64    pub ts_event: UnixNanos,
65    /// UNIX timestamp (nanoseconds) when the instance was initialized.
66    pub ts_init: UnixNanos,
67}
68
69impl TimeEvent {
70    /// Creates a time event with the supplied identity and timestamps.
71    #[must_use]
72    pub const fn new(name: Ustr, event_id: UUID4, ts_event: UnixNanos, ts_init: UnixNanos) -> Self {
73        Self {
74            name,
75            event_id,
76            ts_event,
77            ts_init,
78        }
79    }
80}
81
82impl Display for TimeEvent {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        write!(
85            f,
86            "{}(name={}, event_id={}, ts_event={}, ts_init={})",
87            stringify!(TimeEvent),
88            self.name,
89            self.event_id,
90            self.ts_event,
91            self.ts_init
92        )
93    }
94}
95
96/// Orders a [`TimeEvent`] for earliest-first scheduling in a
97/// [`BinaryHeap`](std::collections::BinaryHeap).
98///
99/// The reversed ordering makes the heap pop events in ascending order by `ts_event`, then `name`,
100/// `ts_init`, and `event_id`.
101#[repr(transparent)] // Guarantees zero-cost abstraction with identical memory layout
102#[derive(Clone, Debug, PartialEq, Eq)]
103pub struct ScheduledTimeEvent(
104    /// The time event to schedule.
105    pub TimeEvent,
106);
107
108impl ScheduledTimeEvent {
109    /// Creates a scheduled wrapper for `event`.
110    #[must_use]
111    pub const fn new(event: TimeEvent) -> Self {
112        Self(event)
113    }
114
115    /// Returns the wrapped time event.
116    #[must_use]
117    pub fn into_inner(self) -> TimeEvent {
118        self.0
119    }
120}
121
122impl PartialOrd for ScheduledTimeEvent {
123    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
124        Some(self.cmp(other))
125    }
126}
127
128impl Ord for ScheduledTimeEvent {
129    fn cmp(&self, other: &Self) -> Ordering {
130        // Reverse order for max heap: earlier timestamps have higher priority
131        cmp_time_events(&other.0, &self.0)
132    }
133}
134
135#[cfg(feature = "python")]
136/// Wraps a Python callable that handles time events.
137pub struct PythonTimeEventCallback {
138    callback: Py<PyAny>,
139}
140
141#[cfg(feature = "python")]
142impl PythonTimeEventCallback {
143    /// Wraps a Python callable as a time event callback.
144    #[must_use]
145    pub const fn new(callback: Py<PyAny>) -> Self {
146        Self { callback }
147    }
148
149    /// Invokes the Python callback for `event`.
150    ///
151    /// Logs and suppresses any exception raised by the callback.
152    pub fn call(&self, event: TimeEvent) {
153        Python::attach(|py| {
154            if let Err(e) = self.callback.call1(py, (event,)) {
155                log::error!("Python time event callback raised exception: {e}");
156            }
157        });
158    }
159}
160
161#[cfg(feature = "python")]
162impl Debug for PythonTimeEventCallback {
163    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
164        f.debug_struct(stringify!(PythonTimeEventCallback))
165            .finish_non_exhaustive()
166    }
167}
168
169#[derive(Clone)]
170/// Represents a callback invoked for time events.
171///
172/// # Variants
173///
174/// - `Python`: For Python callbacks (requires `python` feature).
175/// - `Rust`: Thread-safe callbacks using `Arc`. Use when the closure is `Send + Sync`.
176/// - `RustLocal`: Single-threaded callbacks using `Rc`. Use when capturing `Rc<RefCell<...>>`.
177///
178/// # Choosing Between `Rust` and `RustLocal`
179///
180/// Use `Rust` (thread-safe) when:
181/// - The callback does not capture `Rc<RefCell<...>>` or other non-`Send` types.
182/// - The closure is `Send + Sync` (most simple closures qualify).
183///
184/// Use `RustLocal` when:
185/// - The callback captures `Rc<RefCell<...>>` for shared mutable state.
186/// - Thread safety constraints prevent using `Arc`.
187///
188/// `RustLocal` works with `TestClock` and with `LiveClock` when its event channel
189/// is drained on the callback's originating thread.
190///
191/// # Automatic Conversion
192///
193/// - Closures that are `Fn + Send + Sync + 'static` automatically convert to `Rust`.
194/// - `Rc<dyn Fn(TimeEvent)>` converts to `RustLocal`.
195/// - `Arc<dyn Fn(TimeEvent) + Send + Sync>` converts to `Rust`.
196pub enum TimeEventCallback {
197    /// Python callable for use from Python via PyO3.
198    #[cfg(feature = "python")]
199    Python(Arc<PythonTimeEventCallback>),
200    /// Thread-safe Rust callback using `Arc` (`Send + Sync`).
201    Rust(Arc<dyn Fn(TimeEvent) + Send + Sync>),
202    /// Local Rust callback using `Rc` (not `Send`/`Sync`).
203    RustLocal(Rc<dyn Fn(TimeEvent)>),
204}
205
206impl Debug for TimeEventCallback {
207    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
208        match self {
209            #[cfg(feature = "python")]
210            Self::Python(_) => f.write_str("Python callback"),
211            Self::Rust(_) => f.write_str("Rust callback (thread-safe)"),
212            Self::RustLocal(_) => f.write_str("Rust callback (local)"),
213        }
214    }
215}
216
217impl TimeEventCallback {
218    /// Returns `true` if this is a local (non-thread-safe) Rust callback.
219    ///
220    /// Local callbacks use `Rc` internally and require creation, cloning, dropping,
221    /// and invocation to stay on the originating thread.
222    #[must_use]
223    pub const fn is_local(&self) -> bool {
224        matches!(self, Self::RustLocal(_))
225    }
226
227    /// Invokes the callback for the given `TimeEvent`.
228    ///
229    /// For Python callbacks, exceptions are logged as errors rather than panicking.
230    ///
231    /// # Panics
232    ///
233    /// Panics from Rust callbacks propagate to the caller.
234    pub fn call(&self, event: TimeEvent) {
235        match self {
236            #[cfg(feature = "python")]
237            Self::Python(callback) => callback.call(event),
238            Self::Rust(callback) => callback(event),
239            Self::RustLocal(callback) => callback(event),
240        }
241    }
242}
243
244impl<F> From<F> for TimeEventCallback
245where
246    F: Fn(TimeEvent) + Send + Sync + 'static,
247{
248    fn from(value: F) -> Self {
249        Self::Rust(Arc::new(value))
250    }
251}
252
253impl From<Arc<dyn Fn(TimeEvent) + Send + Sync>> for TimeEventCallback {
254    fn from(value: Arc<dyn Fn(TimeEvent) + Send + Sync>) -> Self {
255        Self::Rust(value)
256    }
257}
258
259impl From<Rc<dyn Fn(TimeEvent)>> for TimeEventCallback {
260    fn from(value: Rc<dyn Fn(TimeEvent)>) -> Self {
261        Self::RustLocal(value)
262    }
263}
264
265#[cfg(feature = "python")]
266impl From<Py<PyAny>> for TimeEventCallback {
267    fn from(value: Py<PyAny>) -> Self {
268        Self::from_python_time_event(value)
269    }
270}
271
272#[cfg(feature = "python")]
273impl TimeEventCallback {
274    /// Creates a Python callback that receives a PyO3 `TimeEvent`.
275    #[must_use]
276    pub fn from_python_time_event(callback: Py<PyAny>) -> Self {
277        Self::Python(Arc::new(PythonTimeEventCallback::new(callback)))
278    }
279}
280
281#[repr(C)]
282#[derive(Clone, Debug)]
283/// Pairs a [`TimeEvent`] with its callback for ordered dispatch.
284///
285/// Natural ordering is ascending by `ts_event`, then `name`, `ts_init`, and `event_id`.
286pub struct TimeEventHandler {
287    /// The time event.
288    pub event: TimeEvent,
289    /// The callable handler for the event.
290    pub callback: TimeEventCallback,
291}
292
293impl TimeEventHandler {
294    /// Creates a handler for `event` and `callback`.
295    #[must_use]
296    pub const fn new(event: TimeEvent, callback: TimeEventCallback) -> Self {
297        Self { event, callback }
298    }
299
300    /// Dispatches the event to the installed message-bus tap, then invokes its callback.
301    ///
302    /// # Panics
303    ///
304    /// Panics from the message-bus tap or a Rust callback propagate to the caller.
305    pub fn run(self) {
306        let Self { event, callback } = self;
307        crate::msgbus::dispatch_tap_time_event(&event);
308        callback.call(event);
309    }
310}
311
312impl PartialOrd for TimeEventHandler {
313    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
314        Some(self.cmp(other))
315    }
316}
317
318impl PartialEq for TimeEventHandler {
319    fn eq(&self, other: &Self) -> bool {
320        self.cmp(other).is_eq()
321    }
322}
323
324impl Eq for TimeEventHandler {}
325
326impl Ord for TimeEventHandler {
327    fn cmp(&self, other: &Self) -> Ordering {
328        cmp_time_events(&self.event, &other.event)
329    }
330}
331
332fn cmp_time_events(left: &TimeEvent, right: &TimeEvent) -> Ordering {
333    left.ts_event
334        .cmp(&right.ts_event)
335        .then_with(|| left.name.cmp(&right.name))
336        .then_with(|| left.ts_init.cmp(&right.ts_init))
337        .then_with(|| left.event_id.as_str().cmp(right.event_id.as_str()))
338}
339
340pub(crate) trait Timer {
341    fn is_expired(&self) -> bool;
342    fn cancel(&mut self);
343}
344
345/// A deterministic interval timer for use with a [`TestClock`](crate::clock::TestClock).
346///
347/// The timer generates scheduled events through an optional inclusive stop time as its iterator is
348/// consumed.
349#[derive(Clone, Debug)]
350pub struct TestTimer {
351    /// The name of the timer.
352    pub name: Ustr,
353    /// The interval between timer events in nanoseconds.
354    pub interval_ns: NonZeroU64,
355    /// The start time of the timer in UNIX nanoseconds.
356    pub start_time_ns: UnixNanos,
357    /// The optional inclusive stop time of the timer in UNIX nanoseconds.
358    pub stop_time_ns: Option<UnixNanos>,
359    /// Whether the first event fires at the start time instead of after one interval.
360    pub fire_immediately: bool,
361    next_time_ns: UnixNanos,
362    is_expired: bool,
363}
364
365impl TestTimer {
366    /// Creates a test timer with the supplied schedule.
367    ///
368    /// # Panics
369    ///
370    /// Panics if:
371    /// - `name` is not a valid string.
372    /// - `fire_immediately` is `false` and `start_time_ns + interval_ns` exceeds the
373    ///   [`UnixNanos`] range.
374    #[must_use]
375    pub fn new(
376        name: Ustr,
377        interval_ns: NonZeroU64,
378        start_time_ns: UnixNanos,
379        stop_time_ns: Option<UnixNanos>,
380        fire_immediately: bool,
381    ) -> Self {
382        check_valid_string_utf8(name, stringify!(name)).expect(FAILED);
383
384        let next_time_ns = if fire_immediately {
385            start_time_ns
386        } else {
387            start_time_ns + interval_ns.get()
388        };
389
390        Self {
391            name,
392            interval_ns,
393            start_time_ns,
394            stop_time_ns,
395            fire_immediately,
396            next_time_ns,
397            is_expired: false,
398        }
399    }
400
401    /// Returns the next time in UNIX nanoseconds when the timer will fire.
402    #[must_use]
403    pub const fn next_time_ns(&self) -> UnixNanos {
404        self.next_time_ns
405    }
406
407    /// Returns whether the timer is expired.
408    #[must_use]
409    pub const fn is_expired(&self) -> bool {
410        self.is_expired
411    }
412
413    /// Returns a lazy iterator over events scheduled at or before `to_time_ns`.
414    ///
415    /// Consuming the iterator advances the timer. Events at `to_time_ns` and at the configured stop
416    /// time are included.
417    pub fn advance(&mut self, to_time_ns: UnixNanos) -> impl Iterator<Item = TimeEvent> + '_ {
418        // Calculate how many events should fire up to and including to_time_ns
419        let advances = if self.next_time_ns <= to_time_ns {
420            ((to_time_ns.as_u64() - self.next_time_ns.as_u64()) / self.interval_ns.get())
421                .saturating_add(1)
422        } else {
423            0
424        };
425        self.take(advances as usize).map(|(event, _)| event)
426    }
427
428    /// Cancels the timer so it produces no further events.
429    pub const fn cancel(&mut self) {
430        self.is_expired = true;
431    }
432}
433
434impl Timer for TestTimer {
435    fn is_expired(&self) -> bool {
436        Self::is_expired(self)
437    }
438
439    fn cancel(&mut self) {
440        Self::cancel(self);
441    }
442}
443
444impl Iterator for TestTimer {
445    type Item = (TimeEvent, UnixNanos);
446
447    fn next(&mut self) -> Option<Self::Item> {
448        if self.is_expired {
449            return None;
450        }
451
452        // Check if current event would exceed stop time before creating the event
453        if let Some(stop_time_ns) = self.stop_time_ns
454            && self.next_time_ns > stop_time_ns
455        {
456            self.is_expired = true;
457            return None;
458        }
459
460        let event_time_ns = self.next_time_ns;
461
462        let item = (
463            TimeEvent {
464                name: self.name,
465                event_id: UUID4::new(),
466                ts_event: event_time_ns,
467                ts_init: event_time_ns,
468            },
469            event_time_ns,
470        );
471
472        if let Some(following_time_ns) = event_time_ns.checked_add(self.interval_ns.get()) {
473            self.next_time_ns = following_time_ns;
474        } else {
475            self.is_expired = true;
476        }
477
478        if self.stop_time_ns == Some(event_time_ns) {
479            self.is_expired = true;
480        }
481
482        Some(item)
483    }
484}
485
486#[cfg(test)]
487mod tests {
488    use std::{cell::RefCell, collections::BinaryHeap, num::NonZeroU64, rc::Rc};
489
490    use nautilus_core::{UUID4, UnixNanos};
491    #[cfg(feature = "python")]
492    use pyo3::{
493        Bound, PyResult, Python,
494        types::{
495            PyAnyMethods, PyCFunction, PyDict, PyList, PyListMethods, PyTuple, PyTupleMethods,
496            PyTypeMethods,
497        },
498    };
499    use rstest::*;
500    use ustr::Ustr;
501
502    use super::{
503        ScheduledTimeEvent, TestTimer, TimeEvent, TimeEventCallback, TimeEventHandler,
504        create_valid_interval,
505    };
506    use crate::msgbus::{
507        BusTap, Endpoint, MStr, MessagingSwitchboard, Topic, clear_bus_tap, set_bus_tap,
508    };
509
510    #[rstest]
511    #[case(0, 1)]
512    #[case(1, 1)]
513    #[case(25, 25)]
514    fn test_create_valid_interval(#[case] interval_ns: u64, #[case] expected: u64) {
515        assert_eq!(create_valid_interval(interval_ns).get(), expected);
516    }
517
518    #[rstest]
519    fn test_test_timer_advance_within_next_time_ns() {
520        let mut timer = TestTimer::new(
521            Ustr::from("TEST_TIMER"),
522            NonZeroU64::new(5).unwrap(),
523            UnixNanos::default(),
524            None,
525            false,
526        );
527        let _: Vec<TimeEvent> = timer.advance(UnixNanos::from(1)).collect();
528        let _: Vec<TimeEvent> = timer.advance(UnixNanos::from(2)).collect();
529        let _: Vec<TimeEvent> = timer.advance(UnixNanos::from(3)).collect();
530        assert_eq!(timer.advance(UnixNanos::from(4)).count(), 0);
531        assert_eq!(timer.next_time_ns, 5);
532        assert!(!timer.is_expired);
533    }
534
535    #[rstest]
536    fn test_test_timer_advance_up_to_next_time_ns() {
537        let mut timer = TestTimer::new(
538            Ustr::from("TEST_TIMER"),
539            NonZeroU64::new(1).unwrap(),
540            UnixNanos::default(),
541            None,
542            false,
543        );
544        assert_eq!(timer.advance(UnixNanos::from(1)).count(), 1);
545        assert!(!timer.is_expired);
546    }
547
548    #[rstest]
549    fn test_test_timer_advance_up_to_next_time_ns_with_stop_time() {
550        let mut timer = TestTimer::new(
551            Ustr::from("TEST_TIMER"),
552            NonZeroU64::new(1).unwrap(),
553            UnixNanos::default(),
554            Some(UnixNanos::from(2)),
555            false,
556        );
557        assert_eq!(timer.advance(UnixNanos::from(2)).count(), 2);
558        assert!(timer.is_expired);
559    }
560
561    #[rstest]
562    fn test_test_timer_advance_beyond_next_time_ns() {
563        let mut timer = TestTimer::new(
564            Ustr::from("TEST_TIMER"),
565            NonZeroU64::new(1).unwrap(),
566            UnixNanos::default(),
567            Some(UnixNanos::from(5)),
568            false,
569        );
570        assert_eq!(timer.advance(UnixNanos::from(5)).count(), 5);
571        assert!(timer.is_expired);
572    }
573
574    #[rstest]
575    fn test_test_timer_advance_beyond_stop_time() {
576        let mut timer = TestTimer::new(
577            Ustr::from("TEST_TIMER"),
578            NonZeroU64::new(1).unwrap(),
579            UnixNanos::default(),
580            Some(UnixNanos::from(5)),
581            false,
582        );
583        assert_eq!(timer.advance(UnixNanos::from(10)).count(), 5);
584        assert!(timer.is_expired);
585    }
586
587    #[rstest]
588    fn test_test_timer_advance_exact_boundary() {
589        let mut timer = TestTimer::new(
590            Ustr::from("TEST_TIMER"),
591            NonZeroU64::new(5).unwrap(),
592            UnixNanos::from(0),
593            None,
594            false,
595        );
596        assert_eq!(
597            timer.advance(UnixNanos::from(5)).count(),
598            1,
599            "Expected one event at the 5 ns boundary"
600        );
601        assert_eq!(
602            timer.advance(UnixNanos::from(10)).count(),
603            1,
604            "Expected one event at the 10 ns boundary"
605        );
606    }
607
608    #[rstest]
609    fn test_test_timer_fire_immediately_true() {
610        let mut timer = TestTimer::new(
611            Ustr::from("TEST_TIMER"),
612            NonZeroU64::new(5).unwrap(),
613            UnixNanos::from(10),
614            None,
615            true, // fire_immediately = true
616        );
617
618        // With fire_immediately=true, next_time_ns should be start_time_ns
619        assert_eq!(timer.next_time_ns(), UnixNanos::from(10));
620
621        // Advance to start time should produce an event
622        let events: Vec<TimeEvent> = timer.advance(UnixNanos::from(10)).collect();
623        assert_eq!(events.len(), 1);
624        assert_eq!(events[0].ts_event, UnixNanos::from(10));
625
626        // Next event should be at start_time + interval
627        assert_eq!(timer.next_time_ns(), UnixNanos::from(15));
628    }
629
630    #[rstest]
631    fn test_test_timer_fire_immediately_false() {
632        let mut timer = TestTimer::new(
633            Ustr::from("TEST_TIMER"),
634            NonZeroU64::new(5).unwrap(),
635            UnixNanos::from(10),
636            None,
637            false, // fire_immediately = false
638        );
639
640        // With fire_immediately=false, next_time_ns should be start_time_ns + interval
641        assert_eq!(timer.next_time_ns(), UnixNanos::from(15));
642
643        // Advance to start time should produce no events
644        assert_eq!(timer.advance(UnixNanos::from(10)).count(), 0);
645
646        // Advance to first interval should produce an event
647        let events: Vec<TimeEvent> = timer.advance(UnixNanos::from(15)).collect();
648        assert_eq!(events.len(), 1);
649        assert_eq!(events[0].ts_event, UnixNanos::from(15));
650    }
651
652    #[rstest]
653    fn test_time_event_handler_ordering_uses_tie_breakers() {
654        let callback = TimeEventCallback::from(|_: TimeEvent| {});
655
656        let later_name = TimeEventHandler::new(
657            TimeEvent::new(
658                Ustr::from("TIME_BAR_ESM4-2-MINUTE-ASK-INTERNAL"),
659                UUID4::from("00000000-0000-4000-8000-000000000003"),
660                100.into(),
661                100.into(),
662            ),
663            callback.clone(),
664        );
665        let earlier_name = TimeEventHandler::new(
666            TimeEvent::new(
667                Ustr::from("SPREAD_QUOTE_ESM4"),
668                UUID4::from("00000000-0000-4000-8000-000000000002"),
669                100.into(),
670                100.into(),
671            ),
672            callback.clone(),
673        );
674        let later_init = TimeEventHandler::new(
675            TimeEvent::new(
676                Ustr::from("SPREAD_QUOTE_ESM4"),
677                UUID4::from("00000000-0000-4000-8000-000000000004"),
678                100.into(),
679                101.into(),
680            ),
681            callback.clone(),
682        );
683        let later_id = TimeEventHandler::new(
684            TimeEvent::new(
685                Ustr::from("SPREAD_QUOTE_ESM4"),
686                UUID4::from("00000000-0000-4000-8000-000000000005"),
687                100.into(),
688                100.into(),
689            ),
690            callback,
691        );
692
693        assert!(earlier_name < later_name);
694        assert!(earlier_name < later_init);
695        assert!(earlier_name < later_id);
696        assert_ne!(earlier_name, later_id);
697    }
698
699    #[rstest]
700    fn test_scheduled_time_event_ordering_laws() {
701        let base = ScheduledTimeEvent::new(TimeEvent::new(
702            Ustr::from("ALPHA"),
703            UUID4::from("00000000-0000-4000-8000-000000000001"),
704            100.into(),
705            10.into(),
706        ));
707        let variants = [
708            base.clone(),
709            ScheduledTimeEvent::new(TimeEvent::new(
710                Ustr::from("BETA"),
711                base.0.event_id,
712                base.0.ts_event,
713                base.0.ts_init,
714            )),
715            ScheduledTimeEvent::new(TimeEvent::new(
716                base.0.name,
717                UUID4::from("00000000-0000-4000-8000-000000000002"),
718                base.0.ts_event,
719                base.0.ts_init,
720            )),
721            ScheduledTimeEvent::new(TimeEvent::new(
722                base.0.name,
723                base.0.event_id,
724                101.into(),
725                base.0.ts_init,
726            )),
727            ScheduledTimeEvent::new(TimeEvent::new(
728                base.0.name,
729                base.0.event_id,
730                base.0.ts_event,
731                11.into(),
732            )),
733        ];
734
735        for a in &variants {
736            for b in &variants {
737                assert_eq!(a == b, a.cmp(b).is_eq());
738                assert_eq!(a.partial_cmp(b), Some(a.cmp(b)));
739                assert_eq!(a.cmp(b), b.cmp(a).reverse());
740            }
741        }
742    }
743
744    #[rstest]
745    fn test_scheduled_time_event_heap_ordering() {
746        let expected = [
747            TimeEvent::new(
748                Ustr::from("ALPHA"),
749                UUID4::from("00000000-0000-4000-8000-000000000001"),
750                100.into(),
751                10.into(),
752            ),
753            TimeEvent::new(
754                Ustr::from("ALPHA"),
755                UUID4::from("00000000-0000-4000-8000-000000000002"),
756                100.into(),
757                10.into(),
758            ),
759            TimeEvent::new(
760                Ustr::from("ALPHA"),
761                UUID4::from("00000000-0000-4000-8000-000000000003"),
762                100.into(),
763                11.into(),
764            ),
765            TimeEvent::new(
766                Ustr::from("BETA"),
767                UUID4::from("00000000-0000-4000-8000-000000000004"),
768                100.into(),
769                10.into(),
770            ),
771            TimeEvent::new(
772                Ustr::from("ALPHA"),
773                UUID4::from("00000000-0000-4000-8000-000000000005"),
774                101.into(),
775                10.into(),
776            ),
777        ];
778        let insertion_order = [4, 1, 3, 0, 2];
779        let mut heap = BinaryHeap::new();
780
781        for index in insertion_order {
782            heap.push(ScheduledTimeEvent::new(expected[index].clone()));
783        }
784
785        let popped = std::iter::from_fn(|| heap.pop().map(ScheduledTimeEvent::into_inner))
786            .collect::<Vec<_>>();
787        assert_eq!(popped, expected);
788    }
789
790    #[cfg(feature = "python")]
791    #[rstest]
792    fn test_python_callback_passes_time_event() {
793        Python::initialize();
794
795        Python::attach(|py| {
796            let seen = PyList::empty(py);
797            let seen_obj = seen.clone().unbind().into_any();
798            let callback = new_sync_py_callback(
799                py,
800                move |args: &Bound<'_, PyTuple>,
801                      _kwargs: Option<&Bound<'_, PyDict>>|
802                      -> PyResult<()> {
803                    let arg = args.get_item(0)?;
804                    let type_name = arg.get_type().name()?.to_string();
805                    Python::attach(|py| seen_obj.call_method1(py, "append", (type_name,)))?;
806                    Ok(())
807                },
808            )
809            .expect("callback should create")
810            .into_any()
811            .unbind();
812
813            let event = TimeEvent::new(
814                Ustr::from("PY_CALLBACK_MODE"),
815                UUID4::from("00000000-0000-4000-8000-000000000007"),
816                UnixNanos::from(100),
817                UnixNanos::from(99),
818            );
819
820            TimeEventCallback::from_python_time_event(callback).call(event);
821
822            assert_eq!(seen.len(), 1);
823            assert_eq!(
824                seen.get_item(0).unwrap().extract::<String>().unwrap(),
825                "TimeEvent"
826            );
827        });
828    }
829
830    #[cfg(feature = "python")]
831    fn new_sync_py_callback<F>(py: Python<'_>, closure: F) -> PyResult<Bound<'_, PyCFunction>>
832    where
833        F: Fn(&Bound<'_, PyTuple>, Option<&Bound<'_, PyDict>>) -> PyResult<()>
834            + Send
835            + Sync
836            + 'static,
837    {
838        PyCFunction::new_closure(py, None, None, closure)
839    }
840
841    #[derive(Default)]
842    struct RecordingTimeEventTap {
843        time_events: RefCell<Vec<(String, TimeEvent)>>,
844    }
845
846    impl RecordingTimeEventTap {
847        fn time_events(&self) -> Vec<(String, TimeEvent)> {
848            self.time_events.borrow().clone()
849        }
850    }
851
852    impl BusTap for RecordingTimeEventTap {
853        fn on_publish(&self, topic: MStr<Topic>, message: &dyn std::any::Any) {
854            if let Some(event) = message.downcast_ref::<TimeEvent>() {
855                self.time_events
856                    .borrow_mut()
857                    .push((topic.to_string(), event.clone()));
858            }
859        }
860
861        fn on_send(&self, _endpoint: MStr<Endpoint>, _message: &dyn std::any::Any) {}
862    }
863
864    #[rstest]
865    fn test_time_event_handler_run_dispatches_tap_before_callback() {
866        let event = TimeEvent::new(
867            Ustr::from("strategy.heartbeat"),
868            UUID4::from("00000000-0000-4000-8000-000000000006"),
869            UnixNanos::from(100),
870            UnixNanos::from(99),
871        );
872        let tap = Rc::new(RecordingTimeEventTap::default());
873        let callback_seen: Rc<RefCell<Vec<TimeEvent>>> = Rc::new(RefCell::new(Vec::new()));
874        let expected_topic = MessagingSwitchboard::time_event_topic().to_string();
875        let callback_expected = event.clone();
876        let callback_expected_topic = expected_topic.clone();
877        let callback_tap = Rc::clone(&tap);
878        let callback_seen_ref = Rc::clone(&callback_seen);
879        let callback: Rc<dyn Fn(TimeEvent)> = Rc::new(move |callback_event| {
880            assert_eq!(
881                callback_tap.time_events(),
882                vec![(callback_expected_topic.clone(), callback_expected.clone())],
883            );
884            callback_seen_ref.borrow_mut().push(callback_event);
885        });
886
887        set_bus_tap(tap.clone());
888        TimeEventHandler::new(event.clone(), TimeEventCallback::from(callback)).run();
889        clear_bus_tap();
890
891        assert_eq!(tap.time_events(), vec![(expected_topic, event.clone())]);
892        assert_eq!(*callback_seen.borrow(), vec![event]);
893    }
894
895    use proptest::{prelude::*, test_runner::TestCaseResult};
896
897    #[derive(Clone, Debug)]
898    enum TimerOperation {
899        AdvanceTime(u64),
900        Cancel,
901    }
902
903    fn timer_operation_strategy() -> impl Strategy<Value = TimerOperation> {
904        prop_oneof![
905            8 => (0u64..=1000).prop_map(TimerOperation::AdvanceTime),
906            2 => Just(TimerOperation::Cancel),
907        ]
908    }
909
910    fn timer_config_strategy() -> impl Strategy<Value = (u64, u64, Option<u64>, bool)> {
911        (
912            1u64..=1000,
913            timer_start_time_strategy(),
914            prop::option::of(0u64..=20_000),
915            prop::bool::ANY,
916        )
917            .prop_map(
918                |(interval_ns, start_time_ns, stop_after_ns, fire_immediately)| {
919                    (
920                        interval_ns,
921                        start_time_ns,
922                        stop_after_ns.map(|offset| start_time_ns + offset),
923                        fire_immediately,
924                    )
925                },
926            )
927    }
928
929    fn timer_start_time_strategy() -> impl Strategy<Value = u64> {
930        prop_oneof![
931            6 => 0u64..=u64::MAX - TIMER_TIME_HEADROOM,
932            2 => 0u64..=1_000_000,
933            1 => Just(1_700_000_000_000_000_000),
934            1 => Just(u64::MAX - TIMER_TIME_HEADROOM),
935        ]
936    }
937
938    fn timer_test_strategy()
939    -> impl Strategy<Value = (Vec<TimerOperation>, (u64, u64, Option<u64>, bool))> {
940        (
941            prop::collection::vec(timer_operation_strategy(), 5..=75),
942            timer_config_strategy(),
943        )
944    }
945
946    fn test_timer_with_operations(
947        operations: Vec<TimerOperation>,
948        (interval_ns, start_time_ns, stop_time_ns, fire_immediately): (u64, u64, Option<u64>, bool),
949    ) -> TestCaseResult {
950        let mut timer = TestTimer::new(
951            Ustr::from("PROP_TEST_TIMER"),
952            NonZeroU64::new(interval_ns).unwrap(),
953            UnixNanos::from(start_time_ns),
954            stop_time_ns.map(UnixNanos::from),
955            fire_immediately,
956        );
957
958        let mut current_time = start_time_ns;
959        let mut expected_next = if fire_immediately {
960            start_time_ns
961        } else {
962            start_time_ns + interval_ns
963        };
964        let mut expected_expired = false;
965
966        for operation in operations {
967            match operation {
968                TimerOperation::AdvanceTime(delta) => {
969                    let to_time = current_time + delta;
970                    let actual: Vec<(Ustr, u64, u64)> = timer
971                        .advance(UnixNanos::from(to_time))
972                        .map(|event| time_event_state(&event))
973                        .collect();
974                    let expected = expected_event_states(
975                        expected_event_times(
976                            to_time,
977                            interval_ns,
978                            stop_time_ns,
979                            &mut expected_next,
980                            &mut expected_expired,
981                        ),
982                        Ustr::from("PROP_TEST_TIMER"),
983                    );
984                    current_time = to_time;
985
986                    prop_assert_eq!(actual, expected);
987                }
988                TimerOperation::Cancel => {
989                    timer.cancel();
990                    expected_expired = true;
991                }
992            }
993
994            prop_assert_eq!(timer.is_expired(), expected_expired);
995            prop_assert_eq!(timer.next_time_ns().as_u64(), expected_next);
996        }
997
998        if !expected_expired && let Some(stop_time_ns) = stop_time_ns {
999            let to_time = stop_time_ns.saturating_add(interval_ns);
1000            let actual: Vec<(Ustr, u64, u64)> = timer
1001                .advance(UnixNanos::from(to_time))
1002                .map(|event| time_event_state(&event))
1003                .collect();
1004            let expected = expected_event_states(
1005                expected_event_times(
1006                    to_time,
1007                    interval_ns,
1008                    Some(stop_time_ns),
1009                    &mut expected_next,
1010                    &mut expected_expired,
1011                ),
1012                Ustr::from("PROP_TEST_TIMER"),
1013            );
1014            prop_assert_eq!(actual, expected);
1015            prop_assert!(expected_expired);
1016            prop_assert!(timer.is_expired());
1017            prop_assert_eq!(timer.next_time_ns().as_u64(), expected_next);
1018        }
1019
1020        Ok(())
1021    }
1022
1023    fn expected_event_times(
1024        to_time: u64,
1025        interval_ns: u64,
1026        stop_time_ns: Option<u64>,
1027        next_time: &mut u64,
1028        is_expired: &mut bool,
1029    ) -> Vec<u64> {
1030        let mut events = Vec::new();
1031
1032        while !*is_expired && *next_time <= to_time {
1033            if let Some(stop_time_ns) = stop_time_ns
1034                && *next_time > stop_time_ns
1035            {
1036                *is_expired = true;
1037                break;
1038            }
1039
1040            let event_time = *next_time;
1041            events.push(event_time);
1042            let Some(following_time) = event_time.checked_add(interval_ns) else {
1043                *is_expired = true;
1044                break;
1045            };
1046            *next_time = following_time;
1047
1048            if Some(event_time) == stop_time_ns {
1049                *is_expired = true;
1050                break;
1051            }
1052        }
1053
1054        events
1055    }
1056
1057    proptest! {
1058        #[rstest]
1059        fn prop_timer_advance_operations((operations, config) in timer_test_strategy()) {
1060            test_timer_with_operations(operations, config)?;
1061        }
1062
1063        #[rstest]
1064        fn prop_timer_advance_batching_is_consistent(
1065            interval_ns in 1u64..=1000,
1066            start_time_ns in timer_start_time_strategy(),
1067            fire_immediately in prop::bool::ANY,
1068            advance_count in 1u64..=20,
1069        ) {
1070            let mut timer = TestTimer::new(
1071                Ustr::from("CONSISTENCY_TEST"),
1072                NonZeroU64::new(interval_ns).unwrap(),
1073                UnixNanos::from(start_time_ns),
1074                None, // No stop time for this test
1075                fire_immediately,
1076            );
1077
1078            let first_event_time = if fire_immediately { start_time_ns } else { start_time_ns + interval_ns };
1079            let final_event_time = first_event_time + interval_ns * (advance_count - 1);
1080            let expected = expected_event_states(
1081                (0..advance_count)
1082                    .map(|index| first_event_time + interval_ns * index)
1083                    .collect(),
1084                Ustr::from("CONSISTENCY_TEST"),
1085            );
1086
1087            let mut batched_timer = timer.clone();
1088            let batched: Vec<(Ustr, u64, u64)> = batched_timer
1089                .advance(UnixNanos::from(final_event_time))
1090                .map(|event| time_event_state(&event))
1091                .collect();
1092
1093            let mut stepped = Vec::new();
1094
1095            for event_time in
1096                (0..advance_count).map(|index| first_event_time + interval_ns * index)
1097            {
1098                stepped.extend(
1099                    timer
1100                        .advance(UnixNanos::from(event_time))
1101                        .map(|event| time_event_state(&event)),
1102                );
1103            }
1104
1105            prop_assert_eq!(&batched, &expected);
1106            prop_assert_eq!(&stepped, &expected);
1107            prop_assert_eq!(timer.next_time_ns(), batched_timer.next_time_ns());
1108            prop_assert_eq!(timer.is_expired(), batched_timer.is_expired());
1109        }
1110
1111        #[rstest]
1112        fn prop_timer_terminal_time_does_not_require_following_time(
1113            (interval_ns, event_headroom) in terminal_time_strategy(),
1114            fire_immediately in prop::bool::ANY,
1115            bounded in prop::bool::ANY,
1116        ) {
1117            let event_time_ns = u64::MAX - event_headroom;
1118            let start_time_ns = if fire_immediately {
1119                event_time_ns
1120            } else {
1121                event_time_ns - interval_ns
1122            };
1123            let mut timer = TestTimer::new(
1124                Ustr::from("TERMINAL_STOP_TEST"),
1125                NonZeroU64::new(interval_ns).unwrap(),
1126                UnixNanos::from(start_time_ns),
1127                bounded.then_some(UnixNanos::max()),
1128                fire_immediately,
1129            );
1130
1131            let events: Vec<(Ustr, u64, u64)> = timer
1132                .advance(UnixNanos::max())
1133                .map(|event| time_event_state(&event))
1134                .collect();
1135
1136            prop_assert_eq!(
1137                events,
1138                vec![(Ustr::from("TERMINAL_STOP_TEST"), event_time_ns, event_time_ns)]
1139            );
1140            prop_assert!(timer.is_expired());
1141            prop_assert_eq!(timer.next_time_ns(), UnixNanos::from(event_time_ns));
1142        }
1143    }
1144
1145    const TIMER_TIME_HEADROOM: u64 = 100_000;
1146
1147    fn time_event_state(event: &TimeEvent) -> (Ustr, u64, u64) {
1148        (event.name, event.ts_event.as_u64(), event.ts_init.as_u64())
1149    }
1150
1151    fn expected_event_states(times: Vec<u64>, name: Ustr) -> Vec<(Ustr, u64, u64)> {
1152        times.into_iter().map(|time| (name, time, time)).collect()
1153    }
1154
1155    fn terminal_time_strategy() -> impl Strategy<Value = (u64, u64)> {
1156        (1u64..=1000).prop_flat_map(|interval_ns| (Just(interval_ns), 0u64..interval_ns))
1157    }
1158}
nautilus_common/timer.rs

nautilus_common/
timer.rs
