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

nautilus_common/
timer.rs
