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

nautilus_common/
timer.rs
