nautilus_common/cache/
config.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 nautilus_core::correctness::{CorrectnessResultExt, FAILED, check_positive_usize};
17use serde::{Deserialize, Deserializer, Serialize, de::Error};
18
19use crate::{
20    config::{ConfigError, ConfigErrorCollector, ConfigResult},
21    enums::SerializationEncoding,
22};
23
24/// Configuration for `Cache` instances.
25#[cfg_attr(
26    feature = "python",
27    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.common", from_py_object)
28)]
29#[cfg_attr(
30    feature = "python",
31    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")
32)]
33#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
34#[builder(finish_fn(name = build_inner, vis = ""))]
35#[serde(default, deny_unknown_fields)]
36pub struct CacheConfig {
37    /// The encoding for database operations, controls the type of serializer used.
38    #[builder(default = SerializationEncoding::Json)]
39    pub encoding: SerializationEncoding,
40    /// If timestamps should be persisted as ISO 8601 strings.
41    #[builder(default)]
42    pub timestamps_as_iso8601: bool,
43    /// The buffer interval (milliseconds) between pipelined/batched transactions.
44    pub buffer_interval_ms: Option<usize>,
45    /// The batch size for bulk read operations (e.g., MGET).
46    /// If set, bulk reads will be batched into chunks of this size.
47    pub bulk_read_batch_size: Option<usize>,
48    /// If a 'trader-' prefix is used for keys.
49    #[builder(default = true)]
50    pub use_trader_prefix: bool,
51    /// If the trader's instance ID is used for keys.
52    #[builder(default)]
53    pub use_instance_id: bool,
54    /// If the database should be flushed on start.
55    #[builder(default)]
56    pub flush_on_start: bool,
57    /// If instrument data should be dropped from the cache's memory on reset.
58    #[builder(default = true)]
59    pub drop_instruments_on_reset: bool,
60    /// The maximum length for internal tick deques.
61    #[builder(default = 10_000)]
62    #[serde(deserialize_with = "deserialize_positive_usize")]
63    pub tick_capacity: usize,
64    /// The maximum length for internal bar deques.
65    #[builder(default = 10_000)]
66    #[serde(deserialize_with = "deserialize_positive_usize")]
67    pub bar_capacity: usize,
68    /// If account events should be persisted to a backing database.
69    #[builder(default = true)]
70    pub persist_account_events: bool,
71    /// If market data should be persisted to disk.
72    #[builder(default)]
73    pub save_market_data: bool,
74}
75
76impl<S: cache_config_builder::IsComplete> CacheConfigBuilder<S> {
77    /// Validates and builds the [`CacheConfig`].
78    ///
79    /// # Errors
80    ///
81    /// Returns a [`ConfigError`] if any field fails validation
82    /// (see [`CacheConfig::validate`]).
83    pub fn build(self) -> ConfigResult<CacheConfig> {
84        let config = self.build_inner();
85        config.validate()?;
86        Ok(config)
87    }
88}
89
90impl Default for CacheConfig {
91    fn default() -> Self {
92        Self::builder()
93            .build()
94            .expect("default `CacheConfig` should be valid")
95    }
96}
97
98impl CacheConfig {
99    /// Creates a new [`CacheConfig`] instance.
100    ///
101    /// # Panics
102    ///
103    /// Panics if `tick_capacity` or `bar_capacity` is zero.
104    #[expect(clippy::too_many_arguments)]
105    #[must_use]
106    pub fn new(
107        encoding: SerializationEncoding,
108        timestamps_as_iso8601: bool,
109        buffer_interval_ms: Option<usize>,
110        bulk_read_batch_size: Option<usize>,
111        use_trader_prefix: bool,
112        use_instance_id: bool,
113        flush_on_start: bool,
114        drop_instruments_on_reset: bool,
115        tick_capacity: usize,
116        bar_capacity: usize,
117        persist_account_events: bool,
118        save_market_data: bool,
119    ) -> Self {
120        check_positive_usize(tick_capacity, stringify!(tick_capacity)).expect_display(FAILED);
121        check_positive_usize(bar_capacity, stringify!(bar_capacity)).expect_display(FAILED);
122
123        Self {
124            encoding,
125            timestamps_as_iso8601,
126            buffer_interval_ms,
127            bulk_read_batch_size,
128            use_trader_prefix,
129            use_instance_id,
130            flush_on_start,
131            drop_instruments_on_reset,
132            tick_capacity,
133            bar_capacity,
134            persist_account_events,
135            save_market_data,
136        }
137    }
138
139    /// Checks whether all cache settings are valid.
140    ///
141    /// # Errors
142    ///
143    /// Returns a [`ConfigError`] if a capacity setting is not positive.
144    pub fn validate(&self) -> ConfigResult<()> {
145        let mut errors = ConfigErrorCollector::new();
146
147        for (field, value) in [
148            ("tick_capacity", self.tick_capacity),
149            ("bar_capacity", self.bar_capacity),
150        ] {
151            errors.check(
152                value > 0,
153                ConfigError::range(field, format!("must be positive, was {value}")),
154            );
155        }
156
157        errors.into_result()
158    }
159}
160
161fn deserialize_positive_usize<'de, D>(deserializer: D) -> Result<usize, D::Error>
162where
163    D: Deserializer<'de>,
164{
165    let value = usize::deserialize(deserializer)?;
166    check_positive_usize(value, "capacity").map_err(D::Error::custom)?;
167    Ok(value)
168}
169
170#[cfg(test)]
171mod tests {
172    use rstest::rstest;
173
174    use super::*;
175
176    #[rstest]
177    fn test_default_uses_json_encoding() {
178        let config = CacheConfig::default();
179
180        assert_eq!(config.encoding, SerializationEncoding::Json);
181    }
182
183    #[rstest]
184    #[case(0, 1)]
185    #[case(1, 0)]
186    #[should_panic]
187    fn test_new_rejects_zero_capacities(#[case] tick_capacity: usize, #[case] bar_capacity: usize) {
188        let _ = CacheConfig::new(
189            SerializationEncoding::MsgPack,
190            false,
191            None,
192            None,
193            true,
194            false,
195            false,
196            true,
197            tick_capacity,
198            bar_capacity,
199            true,
200            false,
201        );
202    }
203
204    #[rstest]
205    fn test_builder_rejects_zero_tick_capacity() {
206        let result = CacheConfig::builder().tick_capacity(0).build();
207        assert!(
208            matches!(result, Err(ConfigError::Range { field, .. }) if field == "tick_capacity")
209        );
210    }
211
212    #[rstest]
213    fn test_builder_rejects_zero_bar_capacity() {
214        let result = CacheConfig::builder().bar_capacity(0).build();
215        assert!(matches!(result, Err(ConfigError::Range { field, .. }) if field == "bar_capacity"));
216    }
217
218    #[rstest]
219    #[case(0, 1, "tick_capacity")]
220    #[case(1, 0, "bar_capacity")]
221    fn test_validate_rejects_zero_capacities(
222        #[case] tick_capacity: usize,
223        #[case] bar_capacity: usize,
224        #[case] expected_field: &str,
225    ) {
226        let config = CacheConfig {
227            tick_capacity,
228            bar_capacity,
229            ..Default::default()
230        };
231
232        let err = config.validate().expect_err("zero capacity is invalid");
233
234        assert!(matches!(err, ConfigError::Range { field, .. } if field == expected_field));
235    }
236
237    #[rstest]
238    #[case(r#"{"tick_capacity":0}"#)]
239    #[case(r#"{"bar_capacity":0}"#)]
240    fn test_deserialize_rejects_zero_capacities(#[case] raw: &str) {
241        let err = serde_json::from_str::<CacheConfig>(raw)
242            .expect_err("zero capacity should fail deserialization");
243
244        assert!(
245            err.to_string()
246                .contains("invalid usize for 'capacity' not positive")
247        );
248    }
249
250    #[rstest]
251    fn test_deserialize_uses_positive_default_capacities() {
252        let config = serde_json::from_str::<CacheConfig>("{}").unwrap();
253
254        assert_eq!(config.tick_capacity, 10_000);
255        assert_eq!(config.bar_capacity, 10_000);
256    }
257}
nautilus_common/cache/config.rs

nautilus_common/cache/
config.rs
