nautilus_live/node/
queue.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::fmt::Display;
17
18use nautilus_common::{
19    config::{ConfigErrorCollector, ConfigResult, check_valid_value},
20    messages::system::{QueueCondition, QueueState},
21    runner::SystemChannel,
22};
23use serde::{Deserialize, Serialize};
24
25use super::metrics::{RunnerMetricsDelta, RunnerMetricsSnapshot};
26
27/// Configuration for runner queue pressure monitoring.
28#[cfg_attr(
29    feature = "python",
30    pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
31)]
32#[cfg_attr(
33    feature = "python",
34    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
35)]
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
37#[serde(deny_unknown_fields)]
38pub struct QueueMonitorConfig {
39    /// Queue depth that triggers a backlogged state.
40    pub queue_depth_trigger: usize,
41    /// Queue depth at or below which a backlogged state clears.
42    pub queue_depth_clear: usize,
43    /// Mean dispatch time that triggers a slow state, in nanoseconds.
44    pub mean_dispatch_ns_trigger: u64,
45    /// Mean dispatch time at or below which a slow state clears, in nanoseconds.
46    pub mean_dispatch_ns_clear: u64,
47}
48
49impl QueueMonitorConfig {
50    pub(crate) fn validate(&self) -> ConfigResult<()> {
51        let mut collector = ConfigErrorCollector::new();
52
53        collector.collect(validate_hysteresis(
54            "LiveNodeConfig.queue_monitor.queue_depth",
55            self.queue_depth_trigger,
56            self.queue_depth_clear,
57        ));
58        collector.collect(validate_hysteresis(
59            "LiveNodeConfig.queue_monitor.mean_dispatch_ns",
60            self.mean_dispatch_ns_trigger,
61            self.mean_dispatch_ns_clear,
62        ));
63
64        collector.into_result()
65    }
66}
67
68fn validate_hysteresis<T>(field: impl Into<String>, trigger: T, clear: T) -> ConfigResult<()>
69where
70    T: Copy + Display + PartialOrd,
71{
72    check_valid_value(
73        field,
74        clear < trigger,
75        format!("clear threshold {clear} must be lower than trigger threshold {trigger}"),
76    )
77}
78
79pub(crate) const SYSTEM_CHANNELS: [SystemChannel; 5] = [
80    SystemChannel::TimeEvents,
81    SystemChannel::ExecEvents,
82    SystemChannel::ExecCommands,
83    SystemChannel::DataEvents,
84    SystemChannel::DataCommands,
85];
86
87pub(crate) const fn system_channel_index(channel: SystemChannel) -> usize {
88    match channel {
89        SystemChannel::TimeEvents => 0,
90        SystemChannel::ExecEvents => 1,
91        SystemChannel::ExecCommands => 2,
92        SystemChannel::DataEvents => 3,
93        SystemChannel::DataCommands => 4,
94    }
95}
96
97#[derive(Debug, Clone, Copy, PartialEq, Eq)]
98pub(crate) struct QueueStateTransition {
99    pub channel: SystemChannel,
100    pub condition: QueueCondition,
101    pub state: QueueState,
102    pub queue_depth: usize,
103    pub mean_dispatch_ns: u64,
104}
105
106#[derive(Debug)]
107pub(crate) struct QueueMonitor {
108    config: QueueMonitorConfig,
109    previous_snapshot: RunnerMetricsSnapshot,
110    states: [QueueChannelState; SYSTEM_CHANNELS.len()],
111}
112
113impl QueueMonitor {
114    pub(crate) fn new(
115        config: &QueueMonitorConfig,
116        previous_snapshot: RunnerMetricsSnapshot,
117    ) -> Self {
118        Self {
119            config: config.clone(),
120            previous_snapshot,
121            states: [QueueChannelState::default(); SYSTEM_CHANNELS.len()],
122        }
123    }
124
125    pub(crate) fn evaluate(
126        &mut self,
127        snapshot: RunnerMetricsSnapshot,
128    ) -> Vec<QueueStateTransition> {
129        let delta = RunnerMetricsDelta::from_snapshots(self.previous_snapshot, snapshot);
130        self.previous_snapshot = snapshot;
131        let mut transitions = Vec::new();
132
133        for channel in SYSTEM_CHANNELS {
134            let queue_depth = channel_queue_depth(snapshot, channel);
135            let mean_dispatch_ns = delta.channel_mean_dispatch_ns(channel);
136            let dispatched = channel_dispatched(delta, channel);
137            let state = &mut self.states[system_channel_index(channel)];
138
139            if let Some(queue_state) = condition_transition(
140                &mut state.backlogged,
141                &queue_depth,
142                &self.config.queue_depth_trigger,
143                &self.config.queue_depth_clear,
144            ) {
145                transitions.push(QueueStateTransition {
146                    channel,
147                    condition: QueueCondition::Backlogged,
148                    state: queue_state,
149                    queue_depth,
150                    mean_dispatch_ns,
151                });
152            }
153
154            // A window without dispatches has no mean sample, so retain the previous slow state.
155            if dispatched > 0
156                && let Some(queue_state) = condition_transition(
157                    &mut state.slow,
158                    &mean_dispatch_ns,
159                    &self.config.mean_dispatch_ns_trigger,
160                    &self.config.mean_dispatch_ns_clear,
161                )
162            {
163                transitions.push(QueueStateTransition {
164                    channel,
165                    condition: QueueCondition::Slow,
166                    state: queue_state,
167                    queue_depth,
168                    mean_dispatch_ns,
169                });
170            }
171        }
172
173        transitions
174    }
175}
176
177#[derive(Debug, Clone, Copy, Default)]
178struct QueueChannelState {
179    slow: bool,
180    backlogged: bool,
181}
182
183fn condition_transition<T>(
184    triggered: &mut bool,
185    value: &T,
186    trigger_threshold: &T,
187    clear_threshold: &T,
188) -> Option<QueueState>
189where
190    T: PartialOrd,
191{
192    if !*triggered && value >= trigger_threshold {
193        *triggered = true;
194        Some(QueueState::Triggered)
195    } else if *triggered && value <= clear_threshold {
196        *triggered = false;
197        Some(QueueState::Cleared)
198    } else {
199        None
200    }
201}
202
203const fn channel_queue_depth(snapshot: RunnerMetricsSnapshot, channel: SystemChannel) -> usize {
204    match channel {
205        SystemChannel::TimeEvents => snapshot.time_events.queue_depth,
206        SystemChannel::ExecEvents => snapshot.exec_events.queue_depth,
207        SystemChannel::ExecCommands => snapshot.exec_commands.queue_depth,
208        SystemChannel::DataEvents => snapshot.data_events.queue_depth,
209        SystemChannel::DataCommands => snapshot.data_commands.queue_depth,
210    }
211}
212
213const fn channel_dispatched(delta: RunnerMetricsDelta, channel: SystemChannel) -> u64 {
214    match channel {
215        SystemChannel::TimeEvents => delta.time_events,
216        SystemChannel::ExecEvents => delta.exec_events,
217        SystemChannel::ExecCommands => delta.exec_commands,
218        SystemChannel::DataEvents => delta.data_events,
219        SystemChannel::DataCommands => delta.data_commands,
220    }
221}
nautilus_live/node/queue.rs

nautilus_live/node/
queue.rs
