nautilus_live/node/
metrics.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
16use std::{
17    sync::atomic::{AtomicU64, AtomicUsize, Ordering},
18    time::Duration,
19};
20
21use nautilus_common::{
22    live::dispatch::DispatchMessage,
23    messages::{DataEvent, ExecutionEvent, data::DataCommand},
24    runner::{SystemChannel, TimeEventMessage, TradingCommandMessage},
25};
26
27/// Primitive metrics for one `LiveNode::run` dispatch channel after startup.
28#[non_exhaustive]
29#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
30pub struct RunnerChannelMetricsSnapshot {
31    /// Number of messages dispatched from this channel.
32    pub dispatched: u64,
33    /// Cumulative nanoseconds spent dispatching from this channel.
34    pub dispatch_busy_ns: u64,
35    /// Receiver backlog sampled on the runner loop thread.
36    pub queue_depth: usize,
37    /// Runner-loop elapsed nanoseconds at this channel's last dispatch.
38    pub last_dispatch_at_ns: u64,
39}
40
41/// Primitive metrics for `LiveNode::run` dispatch and loop work after startup.
42///
43/// Rates, mean dispatch time, backlog pressure, and utilization are derived by callers from
44/// successive snapshots. Values reset each time `LiveNode::run` enters steady state.
45/// Residual channel dispatch during shutdown grace is included, but the final post-loop
46/// drain is not. Snapshots are lock-free and may not be a consistent cross-field view.
47#[non_exhaustive]
48#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
49pub struct RunnerMetricsSnapshot {
50    /// Time event channel metrics.
51    pub time_events: RunnerChannelMetricsSnapshot,
52    /// Execution event channel metrics.
53    pub exec_events: RunnerChannelMetricsSnapshot,
54    /// Execution command channel metrics.
55    pub exec_commands: RunnerChannelMetricsSnapshot,
56    /// Data event channel metrics.
57    pub data_events: RunnerChannelMetricsSnapshot,
58    /// Data command channel metrics.
59    pub data_commands: RunnerChannelMetricsSnapshot,
60    /// Cumulative nanoseconds spent in the five dispatch branches.
61    pub dispatch_busy_ns: u64,
62    /// Cumulative nanoseconds spent in maintenance and reconciliation report processing.
63    pub maintenance_busy_ns: u64,
64    /// Cumulative nanoseconds spent handling external message bus ingress.
65    pub external_msgbus_busy_ns: u64,
66    /// Monotonic nanoseconds since the steady-state runner loop started.
67    pub elapsed_ns: u64,
68}
69
70/// Derived deltas between two `LiveNode::run` runner metrics snapshots.
71///
72/// Values are saturating differences between two [`RunnerMetricsSnapshot`] samples. Queue depths
73/// and last-dispatch timestamps remain snapshot-only point-in-time values.
74#[non_exhaustive]
75#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
76pub struct RunnerMetricsDelta {
77    /// Number of time events dispatched during the sample window.
78    pub time_events: u64,
79    /// Number of execution events dispatched during the sample window.
80    pub exec_events: u64,
81    /// Number of execution commands dispatched during the sample window.
82    pub exec_commands: u64,
83    /// Number of data events dispatched during the sample window.
84    pub data_events: u64,
85    /// Number of data commands dispatched during the sample window.
86    pub data_commands: u64,
87    /// Nanoseconds spent dispatching time events during the sample window.
88    pub time_events_busy_ns: u64,
89    /// Nanoseconds spent dispatching execution events during the sample window.
90    pub exec_events_busy_ns: u64,
91    /// Nanoseconds spent dispatching execution commands during the sample window.
92    pub exec_commands_busy_ns: u64,
93    /// Nanoseconds spent dispatching data events during the sample window.
94    pub data_events_busy_ns: u64,
95    /// Nanoseconds spent dispatching data commands during the sample window.
96    pub data_commands_busy_ns: u64,
97    /// Nanoseconds spent in the five dispatch branches during the sample window.
98    pub dispatch_busy_ns: u64,
99    /// Nanoseconds spent in maintenance and reconciliation processing during the sample window.
100    pub maintenance_busy_ns: u64,
101    /// Nanoseconds spent handling external message bus ingress during the sample window.
102    pub external_msgbus_busy_ns: u64,
103    /// Monotonic nanoseconds elapsed during the sample window.
104    pub elapsed_ns: u64,
105}
106
107impl RunnerMetricsDelta {
108    /// Returns the saturating delta between two runner metrics snapshots.
109    #[must_use]
110    pub fn from_snapshots(before: RunnerMetricsSnapshot, after: RunnerMetricsSnapshot) -> Self {
111        let (time_events, time_events_busy_ns) =
112            channel_dispatch_delta(before.time_events, after.time_events);
113        let (exec_events, exec_events_busy_ns) =
114            channel_dispatch_delta(before.exec_events, after.exec_events);
115        let (exec_commands, exec_commands_busy_ns) =
116            channel_dispatch_delta(before.exec_commands, after.exec_commands);
117        let (data_events, data_events_busy_ns) =
118            channel_dispatch_delta(before.data_events, after.data_events);
119        let (data_commands, data_commands_busy_ns) =
120            channel_dispatch_delta(before.data_commands, after.data_commands);
121
122        Self {
123            time_events,
124            exec_events,
125            exec_commands,
126            data_events,
127            data_commands,
128            time_events_busy_ns,
129            exec_events_busy_ns,
130            exec_commands_busy_ns,
131            data_events_busy_ns,
132            data_commands_busy_ns,
133            dispatch_busy_ns: after
134                .dispatch_busy_ns
135                .saturating_sub(before.dispatch_busy_ns),
136            maintenance_busy_ns: after
137                .maintenance_busy_ns
138                .saturating_sub(before.maintenance_busy_ns),
139            external_msgbus_busy_ns: after
140                .external_msgbus_busy_ns
141                .saturating_sub(before.external_msgbus_busy_ns),
142            elapsed_ns: after.elapsed_ns.saturating_sub(before.elapsed_ns),
143        }
144    }
145
146    /// Returns the total messages dispatched across all runner channels.
147    #[must_use]
148    pub const fn total_dispatched(&self) -> u64 {
149        self.time_events
150            .saturating_add(self.exec_events)
151            .saturating_add(self.exec_commands)
152            .saturating_add(self.data_events)
153            .saturating_add(self.data_commands)
154    }
155
156    /// Returns dispatch busy time divided by elapsed time for the sample window.
157    ///
158    /// Returns `0.0` when the elapsed window is zero.
159    #[must_use]
160    #[expect(
161        clippy::cast_precision_loss,
162        reason = "sample-window utilization is an approximate ratio"
163    )]
164    pub fn dispatch_utilization(&self) -> f64 {
165        if self.elapsed_ns == 0 {
166            0.0
167        } else {
168            self.dispatch_busy_ns as f64 / self.elapsed_ns as f64
169        }
170    }
171
172    /// Returns total timed runner-loop work divided by elapsed time for the sample window.
173    ///
174    /// Total work includes dispatch, maintenance, reconciliation, and external message bus ingress
175    /// handling. Returns `0.0` when the elapsed window is zero.
176    #[must_use]
177    #[expect(
178        clippy::cast_precision_loss,
179        reason = "sample-window utilization is an approximate ratio"
180    )]
181    pub fn loop_utilization(&self) -> f64 {
182        if self.elapsed_ns == 0 {
183            0.0
184        } else {
185            self.total_busy_ns() as f64 / self.elapsed_ns as f64
186        }
187    }
188
189    /// Returns mean dispatch time in nanoseconds for the sample window.
190    ///
191    /// Returns zero when no dispatches were recorded.
192    #[must_use]
193    pub fn mean_dispatch_ns(&self) -> u64 {
194        self.dispatch_busy_ns
195            .checked_div(self.total_dispatched())
196            .unwrap_or(0)
197    }
198
199    /// Returns mean dispatch time in nanoseconds for `channel` during the sample window.
200    ///
201    /// Returns zero when the channel dispatched no messages.
202    #[must_use]
203    pub fn channel_mean_dispatch_ns(&self, channel: SystemChannel) -> u64 {
204        let (dispatched, dispatch_busy_ns) = match channel {
205            SystemChannel::TimeEvents => (self.time_events, self.time_events_busy_ns),
206            SystemChannel::ExecEvents => (self.exec_events, self.exec_events_busy_ns),
207            SystemChannel::ExecCommands => (self.exec_commands, self.exec_commands_busy_ns),
208            SystemChannel::DataEvents => (self.data_events, self.data_events_busy_ns),
209            SystemChannel::DataCommands => (self.data_commands, self.data_commands_busy_ns),
210        };
211
212        dispatch_busy_ns.checked_div(dispatched).unwrap_or(0)
213    }
214
215    /// Returns the total nanoseconds spent in timed runner-loop work.
216    #[must_use]
217    pub const fn total_busy_ns(&self) -> u64 {
218        self.dispatch_busy_ns
219            .saturating_add(self.maintenance_busy_ns)
220            .saturating_add(self.external_msgbus_busy_ns)
221    }
222}
223
224fn channel_dispatch_delta(
225    before: RunnerChannelMetricsSnapshot,
226    after: RunnerChannelMetricsSnapshot,
227) -> (u64, u64) {
228    (
229        after.dispatched.saturating_sub(before.dispatched),
230        after
231            .dispatch_busy_ns
232            .saturating_sub(before.dispatch_busy_ns),
233    )
234}
235
236#[derive(Debug, Default)]
237pub(crate) struct RunnerMetrics {
238    time_events: RunnerChannelMetrics,
239    exec_events: RunnerChannelMetrics,
240    exec_commands: RunnerChannelMetrics,
241    data_events: RunnerChannelMetrics,
242    data_commands: RunnerChannelMetrics,
243    maintenance_busy_ns: AtomicU64,
244    external_msgbus_busy_ns: AtomicU64,
245    elapsed_ns: AtomicU64,
246}
247
248impl RunnerMetrics {
249    pub(crate) fn reset(&self) {
250        self.time_events.reset();
251        self.exec_events.reset();
252        self.exec_commands.reset();
253        self.data_events.reset();
254        self.data_commands.reset();
255        self.maintenance_busy_ns.store(0, Ordering::Relaxed);
256        self.external_msgbus_busy_ns.store(0, Ordering::Relaxed);
257        self.elapsed_ns.store(0, Ordering::Relaxed);
258    }
259
260    pub(crate) fn snapshot(&self) -> RunnerMetricsSnapshot {
261        let time_events = self.time_events.snapshot();
262        let exec_events = self.exec_events.snapshot();
263        let exec_commands = self.exec_commands.snapshot();
264        let data_events = self.data_events.snapshot();
265        let data_commands = self.data_commands.snapshot();
266
267        RunnerMetricsSnapshot {
268            time_events,
269            exec_events,
270            exec_commands,
271            data_events,
272            data_commands,
273            dispatch_busy_ns: time_events
274                .dispatch_busy_ns
275                .saturating_add(exec_events.dispatch_busy_ns)
276                .saturating_add(exec_commands.dispatch_busy_ns)
277                .saturating_add(data_events.dispatch_busy_ns)
278                .saturating_add(data_commands.dispatch_busy_ns),
279            maintenance_busy_ns: self.maintenance_busy_ns.load(Ordering::Relaxed),
280            external_msgbus_busy_ns: self.external_msgbus_busy_ns.load(Ordering::Relaxed),
281            elapsed_ns: self.elapsed_ns.load(Ordering::Relaxed),
282        }
283    }
284
285    pub(crate) fn record_dispatch(
286        &self,
287        channel: SystemChannel,
288        dispatch_elapsed: Duration,
289        elapsed_since_start: Duration,
290    ) {
291        let elapsed_ns = duration_ns(elapsed_since_start);
292        self.channel(channel)
293            .record_dispatch(duration_ns(dispatch_elapsed), elapsed_ns);
294        self.elapsed_ns.store(elapsed_ns, Ordering::Relaxed);
295    }
296
297    pub(crate) fn record_maintenance(&self, work_elapsed: Duration, elapsed_since_start: Duration) {
298        self.record_loop_work(&self.maintenance_busy_ns, work_elapsed, elapsed_since_start);
299    }
300
301    pub(crate) fn record_external_msgbus(
302        &self,
303        work_elapsed: Duration,
304        elapsed_since_start: Duration,
305    ) {
306        self.record_loop_work(
307            &self.external_msgbus_busy_ns,
308            work_elapsed,
309            elapsed_since_start,
310        );
311    }
312
313    pub(crate) fn publish_queue_depths(
314        &self,
315        depths: RunnerChannelQueueDepths,
316        elapsed_since_start: Duration,
317    ) {
318        self.time_events.set_queue_depth(depths.time_events);
319        self.exec_events.set_queue_depth(depths.exec_events);
320        self.exec_commands.set_queue_depth(depths.exec_commands);
321        self.data_events.set_queue_depth(depths.data_events);
322        self.data_commands.set_queue_depth(depths.data_commands);
323        self.elapsed_ns
324            .store(duration_ns(elapsed_since_start), Ordering::Relaxed);
325    }
326
327    fn channel(&self, channel: SystemChannel) -> &RunnerChannelMetrics {
328        match channel {
329            SystemChannel::TimeEvents => &self.time_events,
330            SystemChannel::ExecEvents => &self.exec_events,
331            SystemChannel::ExecCommands => &self.exec_commands,
332            SystemChannel::DataEvents => &self.data_events,
333            SystemChannel::DataCommands => &self.data_commands,
334        }
335    }
336
337    fn record_loop_work(
338        &self,
339        busy_ns: &AtomicU64,
340        work_elapsed: Duration,
341        elapsed_since_start: Duration,
342    ) {
343        saturating_fetch_add(busy_ns, duration_ns(work_elapsed));
344        self.elapsed_ns
345            .store(duration_ns(elapsed_since_start), Ordering::Relaxed);
346    }
347}
348
349#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
350pub(crate) struct RunnerChannelQueueDepths {
351    time_events: usize,
352    exec_events: usize,
353    exec_commands: usize,
354    data_events: usize,
355    data_commands: usize,
356}
357
358impl RunnerChannelQueueDepths {
359    pub(crate) fn from_receivers(
360        time_events: &tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<TimeEventMessage>>,
361        exec_events: &tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<ExecutionEvent>>,
362        exec_commands: &tokio::sync::mpsc::UnboundedReceiver<
363            DispatchMessage<TradingCommandMessage>,
364        >,
365        data_events: &tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<DataEvent>>,
366        data_commands: &tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<DataCommand>>,
367    ) -> Self {
368        Self {
369            time_events: time_events.len(),
370            exec_events: exec_events.len(),
371            exec_commands: exec_commands.len(),
372            data_events: data_events.len(),
373            data_commands: data_commands.len(),
374        }
375    }
376}
377
378#[derive(Debug, Default)]
379struct RunnerChannelMetrics {
380    dispatched: AtomicU64,
381    dispatch_busy_ns: AtomicU64,
382    queue_depth: AtomicUsize,
383    last_dispatch_at_ns: AtomicU64,
384}
385
386impl RunnerChannelMetrics {
387    fn reset(&self) {
388        self.dispatched.store(0, Ordering::Relaxed);
389        self.dispatch_busy_ns.store(0, Ordering::Relaxed);
390        self.queue_depth.store(0, Ordering::Relaxed);
391        self.last_dispatch_at_ns.store(0, Ordering::Relaxed);
392    }
393
394    fn snapshot(&self) -> RunnerChannelMetricsSnapshot {
395        RunnerChannelMetricsSnapshot {
396            dispatched: self.dispatched.load(Ordering::Relaxed),
397            dispatch_busy_ns: self.dispatch_busy_ns.load(Ordering::Relaxed),
398            queue_depth: self.queue_depth.load(Ordering::Relaxed),
399            last_dispatch_at_ns: self.last_dispatch_at_ns.load(Ordering::Relaxed),
400        }
401    }
402
403    fn record_dispatch(&self, dispatch_busy_ns: u64, last_dispatch_at_ns: u64) {
404        self.dispatched.fetch_add(1, Ordering::Relaxed);
405        saturating_fetch_add(&self.dispatch_busy_ns, dispatch_busy_ns);
406        self.last_dispatch_at_ns
407            .store(last_dispatch_at_ns, Ordering::Relaxed);
408    }
409
410    fn set_queue_depth(&self, queue_depth: usize) {
411        self.queue_depth.store(queue_depth, Ordering::Relaxed);
412    }
413}
414
415fn duration_ns(duration: Duration) -> u64 {
416    u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)
417}
418
419fn saturating_fetch_add(atomic: &AtomicU64, value: u64) {
420    atomic
421        .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
422            Some(current.saturating_add(value))
423        })
424        .expect("try_update closure returns Some");
425}
426
427#[cfg(test)]
428mod tests {
429    use std::time::Duration;
430
431    use nautilus_common::{
432        live::dispatch::DispatchMessage,
433        messages::{
434            data::{SubscribeCommand, subscribe::SubscribeInstruments},
435            execution::{QueryAccount, TradingCommand},
436            system::{QueueCondition, QueueState},
437        },
438        msgbus::MessagingSwitchboard,
439        timer::{TimeEvent, TimeEventCallback},
440    };
441    use nautilus_core::{UUID4, UnixNanos};
442    use nautilus_model::{
443        enums::AccountType,
444        events::account::state::AccountState,
445        identifiers::{AccountId, TraderId, Venue},
446        instruments::{InstrumentAny, stubs::crypto_perpetual_ethusdt},
447    };
448    use rstest::rstest;
449    use ustr::Ustr;
450
451    use super::{
452        super::queue::{QueueMonitor, QueueMonitorConfig, QueueStateTransition},
453        *,
454    };
455
456    #[rstest]
457    fn test_runner_metrics_default_snapshot_is_zero() {
458        let metrics = RunnerMetrics::default();
459
460        assert_eq!(metrics.snapshot(), RunnerMetricsSnapshot::default());
461    }
462
463    #[rstest]
464    fn test_runner_metrics_delta_saturates_when_after_is_lower_than_before() {
465        let before = runner_snapshot([10, 9, 8, 7, 6], [20, 20, 20, 20, 20], 90, 80, 70);
466        let after = runner_snapshot([5, 4, 3, 2, 1], [10, 10, 10, 10, 10], 40, 30, 20);
467
468        let delta = RunnerMetricsDelta::from_snapshots(before, after);
469
470        assert_eq!(delta, RunnerMetricsDelta::default());
471    }
472
473    #[rstest]
474    fn test_runner_metrics_delta_zero_elapsed_window_returns_zero_utilization() {
475        let delta = RunnerMetricsDelta::from_snapshots(
476            RunnerMetricsSnapshot::default(),
477            runner_snapshot([1, 0, 0, 0, 0], [10, 0, 0, 0, 0], 20, 30, 0),
478        );
479
480        assert!(delta.dispatch_utilization().abs() < f64::EPSILON);
481        assert!(delta.loop_utilization().abs() < f64::EPSILON);
482    }
483
484    #[rstest]
485    fn test_runner_metrics_delta_zero_dispatched_returns_zero_mean_dispatch_time() {
486        let delta = RunnerMetricsDelta::from_snapshots(
487            RunnerMetricsSnapshot::default(),
488            runner_snapshot([0, 0, 0, 0, 0], [100, 0, 0, 0, 0], 0, 0, 100),
489        );
490
491        assert_eq!(delta.mean_dispatch_ns(), 0);
492        assert_eq!(delta.channel_mean_dispatch_ns(SystemChannel::TimeEvents), 0);
493    }
494
495    #[rstest]
496    fn test_runner_metrics_delta_total_dispatched_sums_all_channels() {
497        let delta = RunnerMetricsDelta::from_snapshots(
498            RunnerMetricsSnapshot::default(),
499            runner_snapshot([1, 2, 3, 4, 5], [0, 0, 0, 0, 0], 0, 0, 100),
500        );
501
502        assert_eq!(delta.total_dispatched(), 15);
503    }
504
505    #[rstest]
506    fn test_runner_metrics_delta_derived_metrics_use_sample_window_values() {
507        let before = runner_snapshot([1, 2, 0, 0, 0], [40, 20, 30, 10, 0], 10, 5, 200);
508        let after = runner_snapshot([4, 3, 2, 1, 0], [70, 40, 40, 10, 0], 30, 15, 300);
509
510        let delta = RunnerMetricsDelta::from_snapshots(before, after);
511
512        assert_eq!(delta.total_dispatched(), 7);
513        assert_eq!(delta.dispatch_busy_ns, 60);
514        assert_eq!(delta.total_busy_ns(), 90);
515        assert_eq!(delta.mean_dispatch_ns(), 8);
516        assert!((delta.dispatch_utilization() - 0.6).abs() < f64::EPSILON);
517        assert!((delta.loop_utilization() - 0.9).abs() < f64::EPSILON);
518    }
519
520    #[rstest]
521    #[case(SystemChannel::TimeEvents, 10)]
522    #[case(SystemChannel::ExecEvents, 20)]
523    #[case(SystemChannel::ExecCommands, 5)]
524    #[case(SystemChannel::DataEvents, 4)]
525    #[case(SystemChannel::DataCommands, 0)]
526    fn test_runner_metrics_delta_channel_mean_dispatch_ns_divides_selected_channel(
527        #[case] channel: SystemChannel,
528        #[case] expected_mean_ns: u64,
529    ) {
530        let before = runner_snapshot([1, 2, 3, 4, 5], [10, 20, 30, 40, 50], 0, 0, 100);
531        let after = runner_snapshot([4, 3, 5, 9, 5], [40, 40, 40, 60, 90], 0, 0, 200);
532
533        let delta = RunnerMetricsDelta::from_snapshots(before, after);
534
535        assert_eq!(delta.channel_mean_dispatch_ns(channel), expected_mean_ns);
536    }
537
538    #[rstest]
539    fn test_runner_metrics_delta_channel_busy_ns_sums_to_dispatch_busy_ns() {
540        let before = runner_snapshot([1, 2, 3, 4, 5], [10, 20, 30, 40, 50], 0, 0, 100);
541        let after = runner_snapshot([4, 3, 5, 9, 5], [40, 40, 40, 60, 90], 0, 0, 200);
542
543        let delta = RunnerMetricsDelta::from_snapshots(before, after);
544
545        assert_eq!(delta.time_events_busy_ns, 30);
546        assert_eq!(delta.exec_events_busy_ns, 20);
547        assert_eq!(delta.exec_commands_busy_ns, 10);
548        assert_eq!(delta.data_events_busy_ns, 20);
549        assert_eq!(delta.data_commands_busy_ns, 40);
550        assert_eq!(delta.dispatch_busy_ns, 120);
551    }
552
553    #[rstest]
554    fn test_queue_monitor_uses_successive_snapshot_delta_and_crossing_values() {
555        let previous = with_queue_depths(
556            runner_snapshot([10, 0, 0, 0, 0], [1_000, 0, 0, 0, 0], 0, 0, 100),
557            [999, 0, 0, 0, 0],
558        );
559        let mut monitor = QueueMonitor::new(&queue_monitor_config(), previous);
560        let snapshot = with_queue_depths(
561            runner_snapshot([12, 0, 0, 0, 0], [1_300, 0, 0, 0, 0], 0, 0, 200),
562            [10, 0, 0, 0, 0],
563        );
564
565        let transitions = monitor.evaluate(snapshot);
566
567        assert_eq!(
568            transitions,
569            vec![
570                QueueStateTransition {
571                    channel: SystemChannel::TimeEvents,
572                    condition: QueueCondition::Backlogged,
573                    state: QueueState::Triggered,
574                    queue_depth: 10,
575                    mean_dispatch_ns: 150,
576                },
577                QueueStateTransition {
578                    channel: SystemChannel::TimeEvents,
579                    condition: QueueCondition::Slow,
580                    state: QueueState::Triggered,
581                    queue_depth: 10,
582                    mean_dispatch_ns: 150,
583                },
584            ]
585        );
586    }
587
588    #[rstest]
589    fn test_queue_monitor_hysteresis_does_not_flap_between_thresholds() {
590        let mut monitor =
591            QueueMonitor::new(&queue_monitor_config(), RunnerMetricsSnapshot::default());
592        let triggered = with_queue_depths(
593            runner_snapshot([1, 0, 0, 0, 0], [100, 0, 0, 0, 0], 0, 0, 100),
594            [10, 0, 0, 0, 0],
595        );
596        let between = with_queue_depths(
597            runner_snapshot([2, 0, 0, 0, 0], [175, 0, 0, 0, 0], 0, 0, 200),
598            [7, 0, 0, 0, 0],
599        );
600        let cleared = with_queue_depths(
601            runner_snapshot([3, 0, 0, 0, 0], [225, 0, 0, 0, 0], 0, 0, 300),
602            [5, 0, 0, 0, 0],
603        );
604
605        assert_eq!(monitor.evaluate(triggered).len(), 2);
606        assert!(monitor.evaluate(between).is_empty());
607        assert_eq!(
608            monitor.evaluate(cleared),
609            vec![
610                QueueStateTransition {
611                    channel: SystemChannel::TimeEvents,
612                    condition: QueueCondition::Backlogged,
613                    state: QueueState::Cleared,
614                    queue_depth: 5,
615                    mean_dispatch_ns: 50,
616                },
617                QueueStateTransition {
618                    channel: SystemChannel::TimeEvents,
619                    condition: QueueCondition::Slow,
620                    state: QueueState::Cleared,
621                    queue_depth: 5,
622                    mean_dispatch_ns: 50,
623                },
624            ]
625        );
626    }
627
628    #[rstest]
629    fn test_queue_monitor_holds_slow_state_without_dispatch_sample() {
630        let mut monitor =
631            QueueMonitor::new(&queue_monitor_config(), RunnerMetricsSnapshot::default());
632        let triggered = runner_snapshot([1, 0, 0, 0, 0], [100, 0, 0, 0, 0], 0, 0, 100);
633        let idle = runner_snapshot([1, 0, 0, 0, 0], [100, 0, 0, 0, 0], 0, 0, 200);
634        let cleared = runner_snapshot([2, 0, 0, 0, 0], [150, 0, 0, 0, 0], 0, 0, 300);
635
636        assert_eq!(
637            monitor.evaluate(triggered),
638            vec![QueueStateTransition {
639                channel: SystemChannel::TimeEvents,
640                condition: QueueCondition::Slow,
641                state: QueueState::Triggered,
642                queue_depth: 0,
643                mean_dispatch_ns: 100,
644            }]
645        );
646        assert!(monitor.evaluate(idle).is_empty());
647        assert_eq!(
648            monitor.evaluate(cleared),
649            vec![QueueStateTransition {
650                channel: SystemChannel::TimeEvents,
651                condition: QueueCondition::Slow,
652                state: QueueState::Cleared,
653                queue_depth: 0,
654                mean_dispatch_ns: 50,
655            }]
656        );
657    }
658
659    #[rstest]
660    fn test_queue_monitor_conditions_trigger_and_clear_independently() {
661        let mut monitor =
662            QueueMonitor::new(&queue_monitor_config(), RunnerMetricsSnapshot::default());
663        let triggered = with_queue_depths(
664            runner_snapshot([1, 0, 0, 0, 0], [100, 0, 0, 0, 0], 0, 0, 100),
665            [10, 0, 0, 0, 0],
666        );
667        let slow_cleared = with_queue_depths(
668            runner_snapshot([2, 0, 0, 0, 0], [150, 0, 0, 0, 0], 0, 0, 200),
669            [7, 0, 0, 0, 0],
670        );
671        let backlog_cleared = with_queue_depths(
672            runner_snapshot([3, 0, 0, 0, 0], [225, 0, 0, 0, 0], 0, 0, 300),
673            [5, 0, 0, 0, 0],
674        );
675
676        assert_eq!(monitor.evaluate(triggered).len(), 2);
677        assert_eq!(
678            monitor.evaluate(slow_cleared),
679            vec![QueueStateTransition {
680                channel: SystemChannel::TimeEvents,
681                condition: QueueCondition::Slow,
682                state: QueueState::Cleared,
683                queue_depth: 7,
684                mean_dispatch_ns: 50,
685            }]
686        );
687        assert_eq!(
688            monitor.evaluate(backlog_cleared),
689            vec![QueueStateTransition {
690                channel: SystemChannel::TimeEvents,
691                condition: QueueCondition::Backlogged,
692                state: QueueState::Cleared,
693                queue_depth: 5,
694                mean_dispatch_ns: 75,
695            }]
696        );
697    }
698
699    #[rstest]
700    fn test_queue_monitor_keeps_channel_state_isolated() {
701        let mut monitor =
702            QueueMonitor::new(&queue_monitor_config(), RunnerMetricsSnapshot::default());
703        let first = with_queue_depths(
704            runner_snapshot([0, 0, 0, 1, 0], [0, 0, 0, 100, 0], 0, 0, 100),
705            [0, 0, 0, 10, 0],
706        );
707        let second = with_queue_depths(
708            runner_snapshot([0, 1, 0, 2, 0], [0, 100, 0, 175, 0], 0, 0, 200),
709            [0, 10, 0, 7, 0],
710        );
711
712        assert_eq!(
713            monitor
714                .evaluate(first)
715                .iter()
716                .map(|transition| transition.channel)
717                .collect::<Vec<_>>(),
718            vec![SystemChannel::DataEvents, SystemChannel::DataEvents]
719        );
720        assert_eq!(
721            monitor.evaluate(second),
722            vec![
723                QueueStateTransition {
724                    channel: SystemChannel::ExecEvents,
725                    condition: QueueCondition::Backlogged,
726                    state: QueueState::Triggered,
727                    queue_depth: 10,
728                    mean_dispatch_ns: 100,
729                },
730                QueueStateTransition {
731                    channel: SystemChannel::ExecEvents,
732                    condition: QueueCondition::Slow,
733                    state: QueueState::Triggered,
734                    queue_depth: 10,
735                    mean_dispatch_ns: 100,
736                },
737            ]
738        );
739    }
740
741    #[rstest]
742    fn test_runner_metrics_snapshot_reflects_dispatch_updates() {
743        let metrics = RunnerMetrics::default();
744
745        metrics.record_dispatch(
746            SystemChannel::ExecCommands,
747            Duration::from_nanos(10),
748            Duration::from_nanos(50),
749        );
750        metrics.record_dispatch(
751            SystemChannel::DataEvents,
752            Duration::from_nanos(7),
753            Duration::from_nanos(90),
754        );
755
756        let snapshot = metrics.snapshot();
757
758        assert_eq!(snapshot.exec_commands.dispatched, 1);
759        assert_eq!(snapshot.exec_commands.dispatch_busy_ns, 10);
760        assert_eq!(snapshot.exec_commands.last_dispatch_at_ns, 50);
761        assert_eq!(snapshot.data_events.dispatched, 1);
762        assert_eq!(snapshot.data_events.dispatch_busy_ns, 7);
763        assert_eq!(snapshot.data_events.last_dispatch_at_ns, 90);
764        assert_eq!(snapshot.dispatch_busy_ns, 17);
765        assert_eq!(snapshot.maintenance_busy_ns, 0);
766        assert_eq!(snapshot.external_msgbus_busy_ns, 0);
767        assert_eq!(snapshot.elapsed_ns, 90);
768    }
769
770    #[rstest]
771    #[case(SystemChannel::TimeEvents, [1, 0, 0, 0, 0], [10, 0, 0, 0, 0], [50, 0, 0, 0, 0])]
772    #[case(SystemChannel::ExecEvents, [0, 1, 0, 0, 0], [0, 10, 0, 0, 0], [0, 50, 0, 0, 0])]
773    #[case(SystemChannel::ExecCommands, [0, 0, 1, 0, 0], [0, 0, 10, 0, 0], [0, 0, 50, 0, 0])]
774    #[case(SystemChannel::DataEvents, [0, 0, 0, 1, 0], [0, 0, 0, 10, 0], [0, 0, 0, 50, 0])]
775    #[case(SystemChannel::DataCommands, [0, 0, 0, 0, 1], [0, 0, 0, 0, 10], [0, 0, 0, 0, 50])]
776    fn test_runner_metrics_record_dispatch_updates_selected_channel(
777        #[case] channel: SystemChannel,
778        #[case] expected_dispatched: [u64; 5],
779        #[case] expected_dispatch_busy_ns: [u64; 5],
780        #[case] expected_last_dispatch: [u64; 5],
781    ) {
782        let metrics = RunnerMetrics::default();
783
784        metrics.record_dispatch(channel, Duration::from_nanos(10), Duration::from_nanos(50));
785        let snapshot = metrics.snapshot();
786
787        assert_eq!(snapshot_dispatch_counts(snapshot), expected_dispatched);
788        assert_eq!(
789            snapshot_dispatch_busy_ns(snapshot),
790            expected_dispatch_busy_ns
791        );
792        assert_eq!(
793            snapshot_last_dispatch_at_ns(snapshot),
794            expected_last_dispatch
795        );
796        assert_eq!(snapshot.dispatch_busy_ns, 10);
797        assert_eq!(snapshot.elapsed_ns, 50);
798    }
799
800    #[rstest]
801    fn test_runner_metrics_snapshot_reflects_loop_work_updates() {
802        let metrics = RunnerMetrics::default();
803
804        metrics.record_maintenance(Duration::from_nanos(10), Duration::from_nanos(50));
805        metrics.record_external_msgbus(Duration::from_nanos(7), Duration::from_nanos(90));
806
807        let snapshot = metrics.snapshot();
808
809        assert_eq!(snapshot.dispatch_busy_ns, 0);
810        assert_eq!(snapshot.maintenance_busy_ns, 10);
811        assert_eq!(snapshot.external_msgbus_busy_ns, 7);
812        assert_eq!(snapshot.elapsed_ns, 90);
813    }
814
815    #[rstest]
816    fn test_runner_metrics_reset_clears_populated_snapshot() {
817        let metrics = RunnerMetrics::default();
818
819        metrics.record_dispatch(
820            SystemChannel::TimeEvents,
821            Duration::from_nanos(10),
822            Duration::from_nanos(30),
823        );
824        metrics.record_maintenance(Duration::from_nanos(5), Duration::from_nanos(40));
825        metrics.record_external_msgbus(Duration::from_nanos(7), Duration::from_nanos(45));
826        metrics.publish_queue_depths(
827            RunnerChannelQueueDepths {
828                time_events: 1,
829                exec_events: 2,
830                exec_commands: 3,
831                data_events: 4,
832                data_commands: 5,
833            },
834            Duration::from_nanos(50),
835        );
836
837        metrics.reset();
838
839        assert_eq!(metrics.snapshot(), RunnerMetricsSnapshot::default());
840    }
841
842    #[rstest]
843    fn test_runner_metrics_queue_depths_use_receiver_lengths() {
844        let (time_tx, time_rx) =
845            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
846        let (exec_evt_tx, exec_evt_rx) =
847            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
848        let (exec_cmd_tx, exec_cmd_rx) =
849            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
850        let (data_evt_tx, data_evt_rx) =
851            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
852        let (data_cmd_tx, data_cmd_rx) =
853            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
854        let metrics = RunnerMetrics::default();
855
856        time_tx.send(stub_time_event_handler().into()).unwrap();
857
858        for _ in 0..2 {
859            exec_evt_tx.send((stub_exec_event()).into()).unwrap();
860        }
861
862        for _ in 0..3 {
863            exec_cmd_tx
864                .send(
865                    TradingCommandMessage::new(
866                        MessagingSwitchboard::exec_engine_execute(),
867                        stub_trading_command(),
868                    )
869                    .into(),
870                )
871                .unwrap();
872        }
873
874        for _ in 0..4 {
875            data_evt_tx.send((stub_data_event()).into()).unwrap();
876        }
877
878        for _ in 0..5 {
879            data_cmd_tx.send(stub_data_command().into()).unwrap();
880        }
881
882        metrics.publish_queue_depths(
883            RunnerChannelQueueDepths::from_receivers(
884                &time_rx,
885                &exec_evt_rx,
886                &exec_cmd_rx,
887                &data_evt_rx,
888                &data_cmd_rx,
889            ),
890            Duration::from_nanos(25),
891        );
892        let snapshot = metrics.snapshot();
893
894        assert_eq!(snapshot.time_events.queue_depth, 1);
895        assert_eq!(snapshot.exec_events.queue_depth, 2);
896        assert_eq!(snapshot.exec_commands.queue_depth, 3);
897        assert_eq!(snapshot.data_events.queue_depth, 4);
898        assert_eq!(snapshot.data_commands.queue_depth, 5);
899        assert_eq!(snapshot.elapsed_ns, 25);
900    }
901
902    fn runner_snapshot(
903        dispatched: [u64; 5],
904        dispatch_busy_ns: [u64; 5],
905        maintenance_busy_ns: u64,
906        external_msgbus_busy_ns: u64,
907        elapsed_ns: u64,
908    ) -> RunnerMetricsSnapshot {
909        let total_dispatch_busy_ns = dispatch_busy_ns.into_iter().fold(0, u64::saturating_add);
910        let [
911            time_events,
912            exec_events,
913            exec_commands,
914            data_events,
915            data_commands,
916        ] = dispatched;
917        let [
918            time_events_busy_ns,
919            exec_events_busy_ns,
920            exec_commands_busy_ns,
921            data_events_busy_ns,
922            data_commands_busy_ns,
923        ] = dispatch_busy_ns;
924
925        RunnerMetricsSnapshot {
926            time_events: channel_snapshot(time_events, time_events_busy_ns),
927            exec_events: channel_snapshot(exec_events, exec_events_busy_ns),
928            exec_commands: channel_snapshot(exec_commands, exec_commands_busy_ns),
929            data_events: channel_snapshot(data_events, data_events_busy_ns),
930            data_commands: channel_snapshot(data_commands, data_commands_busy_ns),
931            dispatch_busy_ns: total_dispatch_busy_ns,
932            maintenance_busy_ns,
933            external_msgbus_busy_ns,
934            elapsed_ns,
935        }
936    }
937
938    fn with_queue_depths(
939        mut snapshot: RunnerMetricsSnapshot,
940        depths: [usize; 5],
941    ) -> RunnerMetricsSnapshot {
942        let [
943            time_events,
944            exec_events,
945            exec_commands,
946            data_events,
947            data_commands,
948        ] = depths;
949        snapshot.time_events.queue_depth = time_events;
950        snapshot.exec_events.queue_depth = exec_events;
951        snapshot.exec_commands.queue_depth = exec_commands;
952        snapshot.data_events.queue_depth = data_events;
953        snapshot.data_commands.queue_depth = data_commands;
954        snapshot
955    }
956
957    fn queue_monitor_config() -> QueueMonitorConfig {
958        QueueMonitorConfig {
959            queue_depth_trigger: 10,
960            queue_depth_clear: 5,
961            mean_dispatch_ns_trigger: 100,
962            mean_dispatch_ns_clear: 50,
963        }
964    }
965
966    fn channel_snapshot(dispatched: u64, dispatch_busy_ns: u64) -> RunnerChannelMetricsSnapshot {
967        RunnerChannelMetricsSnapshot {
968            dispatched,
969            dispatch_busy_ns,
970            ..Default::default()
971        }
972    }
973
974    fn snapshot_dispatch_counts(snapshot: RunnerMetricsSnapshot) -> [u64; 5] {
975        [
976            snapshot.time_events.dispatched,
977            snapshot.exec_events.dispatched,
978            snapshot.exec_commands.dispatched,
979            snapshot.data_events.dispatched,
980            snapshot.data_commands.dispatched,
981        ]
982    }
983
984    fn snapshot_dispatch_busy_ns(snapshot: RunnerMetricsSnapshot) -> [u64; 5] {
985        [
986            snapshot.time_events.dispatch_busy_ns,
987            snapshot.exec_events.dispatch_busy_ns,
988            snapshot.exec_commands.dispatch_busy_ns,
989            snapshot.data_events.dispatch_busy_ns,
990            snapshot.data_commands.dispatch_busy_ns,
991        ]
992    }
993
994    fn snapshot_last_dispatch_at_ns(snapshot: RunnerMetricsSnapshot) -> [u64; 5] {
995        [
996            snapshot.time_events.last_dispatch_at_ns,
997            snapshot.exec_events.last_dispatch_at_ns,
998            snapshot.exec_commands.last_dispatch_at_ns,
999            snapshot.data_events.last_dispatch_at_ns,
1000            snapshot.data_commands.last_dispatch_at_ns,
1001        ]
1002    }
1003
1004    fn stub_time_event_handler() -> TimeEventMessage {
1005        TimeEventMessage::new(
1006            TimeEvent::new(
1007                Ustr::from("test-timer"),
1008                UUID4::new(),
1009                UnixNanos::default(),
1010                UnixNanos::default(),
1011            ),
1012            TimeEventCallback::from(|_| {}),
1013        )
1014    }
1015
1016    fn stub_exec_event() -> ExecutionEvent {
1017        ExecutionEvent::Account(AccountState::new(
1018            AccountId::from("TEST-001"),
1019            AccountType::Cash,
1020            vec![],
1021            vec![],
1022            true,
1023            UUID4::new(),
1024            UnixNanos::default(),
1025            UnixNanos::default(),
1026            None,
1027        ))
1028    }
1029
1030    fn stub_trading_command() -> TradingCommand {
1031        TradingCommand::QueryAccount(QueryAccount::new(
1032            TraderId::from("TESTER-001"),
1033            None,
1034            AccountId::from("TEST-001"),
1035            UUID4::new(),
1036            UnixNanos::default(),
1037            None,
1038            None,
1039        ))
1040    }
1041
1042    fn stub_data_event() -> DataEvent {
1043        DataEvent::Instrument(InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt()))
1044    }
1045
1046    fn stub_data_command() -> DataCommand {
1047        DataCommand::Subscribe(SubscribeCommand::Instruments(SubscribeInstruments::new(
1048            None,
1049            Venue::from("TEST"),
1050            UUID4::new(),
1051            UnixNanos::default(),
1052            None,
1053            None,
1054        )))
1055    }
1056}
nautilus_live/node/metrics.rs

nautilus_live/node/
metrics.rs
