nautilus_infrastructure/redis/
mod.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//! Provides Redis-backed cache database and message bus backing implementations.
17
18pub mod cache;
19pub mod msgbus;
20pub mod queries;
21
22use std::{fmt::Write as _, time::Duration};
23
24use nautilus_common::{logging::log_task_awaiting, msgbus::MessageBusConfig};
25use nautilus_core::{UUID4, string::semver::SemVer};
26use nautilus_model::identifiers::TraderId;
27use redis::RedisError;
28
29const REDIS_MIN_VERSION: &str = "6.2.0";
30const REDIS_DELIMITER: char = ':';
31const REDIS_INDEX_PATTERN: &str = ":index:";
32const REDIS_XTRIM: &str = "XTRIM";
33const REDIS_MINID: &str = "MINID";
34const REDIS_FLUSHDB: &str = "FLUSHDB";
35
36/// Extracts the index key from a full Redis key.
37///
38/// Handles keys with `instance_id` prefix by finding the `:index:` pattern.
39/// For example, `trader-id:uuid:index:order_position` -> `index:order_position`.
40pub(crate) fn get_index_key(key: &str) -> anyhow::Result<&str> {
41    if let Some(pos) = key.find(REDIS_INDEX_PATTERN) {
42        return Ok(&key[pos + 1..]);
43    }
44
45    if key.starts_with("index:") {
46        return Ok(key);
47    }
48
49    anyhow::bail!("Invalid index key format: {key}")
50}
51
52async fn await_handle(handle: Option<tokio::task::JoinHandle<()>>, task_name: &str) {
53    if let Some(handle) = handle {
54        log_task_awaiting(task_name);
55
56        let timeout = Duration::from_secs(2);
57        match tokio::time::timeout(timeout, handle).await {
58            Ok(result) => {
59                if let Err(e) = result {
60                    log::error!("Error awaiting task '{task_name}': {e:?}");
61                }
62            }
63            Err(_) => {
64                log::warn!("Timeout {timeout:?} awaiting task '{task_name}'");
65            }
66        }
67    }
68}
69
70/// Redis connection settings shared by Redis-owned config structs.
71pub trait RedisConnectionConfig {
72    fn host(&self) -> Option<&str>;
73    fn port(&self) -> Option<u16>;
74    fn username(&self) -> Option<&str>;
75    fn password(&self) -> Option<&str>;
76    fn ssl(&self) -> bool;
77    fn connection_timeout(&self) -> u16;
78    fn response_timeout(&self) -> u16;
79    fn number_of_retries(&self) -> usize;
80    fn exponent_base(&self) -> u64;
81    fn max_delay(&self) -> u64;
82    fn factor(&self) -> u64;
83}
84
85/// Parses a Redis connection URL from the given Redis config, returning the
86/// full URL and a redacted version with the password obfuscated.
87///
88/// Authentication matrix handled:
89/// ┌───────────┬───────────┬────────────────────────────┐
90/// │ Username  │ Password  │ Resulting user-info part   │
91/// ├───────────┼───────────┼────────────────────────────┤
92/// │ non-empty │ non-empty │ user:pass@                 │
93/// │ empty     │ non-empty │ :pass@                     │
94/// │ empty     │ empty     │ (omitted)                  │
95/// └───────────┴───────────┴────────────────────────────┘
96///
97/// # Panics
98///
99/// Panics if a username is provided without a corresponding password.
100#[must_use]
101pub fn get_redis_url(config: &impl RedisConnectionConfig) -> (String, String) {
102    let host = config.host().unwrap_or("127.0.0.1");
103    let port = config.port().unwrap_or(6379);
104    let username = config.username().unwrap_or_default();
105    let password = config.password().unwrap_or_default();
106    let ssl = config.ssl();
107
108    // Redact the password for logging/metrics: keep the first & last two chars.
109    let redact_pw = |pw: &str| {
110        if pw.len() > 4 {
111            format!("{}...{}", &pw[..2], &pw[pw.len() - 2..])
112        } else {
113            pw.to_owned()
114        }
115    };
116
117    // Build the `userinfo@` portion for both the real and redacted URLs.
118    let (auth, auth_redacted) = match (username.is_empty(), password.is_empty()) {
119        // user:pass@
120        (false, false) => (
121            format!("{username}:{password}@"),
122            format!("{username}:{}@", redact_pw(password)),
123        ),
124        // :pass@
125        (true, false) => (
126            format!(":{password}@"),
127            format!(":{}@", redact_pw(password)),
128        ),
129        // username but no password ⇒  configuration error
130        (false, true) => panic!(
131            "Redis config error: username supplied without password. \
132            Either supply a password or omit the username."
133        ),
134        // no credentials
135        (true, true) => (String::new(), String::new()),
136    };
137
138    let scheme = if ssl { "rediss" } else { "redis" };
139
140    let url = format!("{scheme}://{auth}{host}:{port}");
141    let redacted_url = format!("{scheme}://{auth_redacted}{host}:{port}");
142
143    (url, redacted_url)
144}
145
146/// Creates a new Redis connection manager based on the provided database `config` and connection name.
147///
148/// # Errors
149///
150/// Returns an error if:
151/// - Constructing the Redis client fails.
152/// - Establishing or configuring the connection manager fails.
153///
154/// In case of reconnection issues, the connection will retry reconnection
155/// `number_of_retries` times, with an exponentially increasing delay, calculated as
156/// `factor * (exponent_base ^ current-try)`, bounded by `max_delay`.
157///
158/// The new connection will time out operations after `response_timeout` has passed.
159/// Each connection attempt to the server will time out after `connection_timeout`.
160pub async fn create_redis_connection(
161    con_name: &str,
162    config: &impl RedisConnectionConfig,
163) -> anyhow::Result<redis::aio::ConnectionManager> {
164    log::debug!("Creating {con_name} redis connection");
165    let (redis_url, redacted_url) = get_redis_url(config);
166    log::debug!("Connecting to {redacted_url}");
167
168    let connection_timeout = Duration::from_secs(u64::from(config.connection_timeout()));
169    let response_timeout = Duration::from_secs(u64::from(config.response_timeout()));
170    let number_of_retries = config.number_of_retries();
171    #[expect(
172        clippy::cast_precision_loss,
173        reason = "redis connection manager API accepts exponent base as f32"
174    )]
175    let exponent_base = config.exponent_base() as f32;
176
177    // Use factor as min_delay base for backoff: factor * (exponent_base ^ tries)
178    let min_delay = Duration::from_millis(config.factor());
179    let max_delay = Duration::from_secs(config.max_delay());
180
181    let client = redis::Client::open(redis_url)?;
182
183    let connection_manager_config = redis::aio::ConnectionManagerConfig::new()
184        .set_exponent_base(exponent_base)
185        .set_number_of_retries(number_of_retries)
186        .set_response_timeout(Some(response_timeout))
187        .set_connection_timeout(Some(connection_timeout))
188        .set_min_delay(min_delay)
189        .set_max_delay(max_delay);
190
191    let mut con = client
192        .get_connection_manager_with_config(connection_manager_config)
193        .await?;
194
195    let version = get_redis_version(&mut con).await?;
196    let min_version = SemVer::parse(REDIS_MIN_VERSION)?;
197    let con_msg = format!("Connected to redis v{version}");
198
199    if version >= min_version {
200        log::info!("{con_msg}");
201    } else {
202        log::error!("{con_msg}, but minimum supported version is {REDIS_MIN_VERSION}");
203    }
204
205    Ok(con)
206}
207
208/// Flushes the entire Redis database for the specified connection.
209///
210/// # Errors
211///
212/// Returns an error if the FLUSHDB command fails.
213pub async fn flush_redis(
214    con: &mut redis::aio::ConnectionManager,
215) -> anyhow::Result<(), RedisError> {
216    redis::cmd(REDIS_FLUSHDB).exec_async(con).await
217}
218
219/// Parse the stream key from the given identifiers and config.
220#[must_use]
221pub fn get_stream_key(
222    trader_id: TraderId,
223    instance_id: UUID4,
224    config: &MessageBusConfig,
225) -> String {
226    let mut stream_key = String::new();
227
228    if config.use_trader_prefix {
229        stream_key.push_str("trader-");
230    }
231
232    if config.use_trader_id {
233        stream_key.push_str(trader_id.as_str());
234        stream_key.push(REDIS_DELIMITER);
235    }
236
237    if config.use_instance_id {
238        write!(stream_key, "{instance_id}").expect("writing to String cannot fail");
239        stream_key.push(REDIS_DELIMITER);
240    }
241
242    stream_key.push_str(&config.streams_prefix);
243    stream_key
244}
245
246async fn get_redis_version(conn: &mut redis::aio::ConnectionManager) -> anyhow::Result<SemVer> {
247    let info: String = redis::cmd("INFO").query_async(conn).await?;
248    let Some(version_str) = info.lines().find_map(|line| {
249        if line.starts_with("redis_version:") {
250            line.split(':').nth(1).map(|s| s.trim().to_string())
251        } else {
252            None
253        }
254    }) else {
255        anyhow::bail!("Redis version not available");
256    };
257
258    SemVer::parse(&version_str)
259}
260
261#[cfg(test)]
262mod tests {
263    use rstest::rstest;
264    use serde_json::json;
265
266    use super::*;
267    use crate::redis::cache::RedisCacheConfig;
268
269    #[rstest]
270    fn test_get_redis_url_default_values() {
271        let config: RedisCacheConfig = serde_json::from_value(json!({})).unwrap();
272        let (url, redacted_url) = get_redis_url(&config);
273        assert_eq!(url, "redis://127.0.0.1:6379");
274        assert_eq!(redacted_url, "redis://127.0.0.1:6379");
275    }
276
277    #[rstest]
278    fn test_get_redis_url_password_only() {
279        // Username omitted, but password present
280        let config_json = json!({
281            "host": "example.com",
282            "port": 6380,
283            "password": "secretpw",   // >4 chars ⇒ will be redacted
284        });
285        let config: RedisCacheConfig = serde_json::from_value(config_json).unwrap();
286        let (url, redacted_url) = get_redis_url(&config);
287        assert_eq!(url, "redis://:secretpw@example.com:6380");
288        assert_eq!(redacted_url, "redis://:se...pw@example.com:6380");
289    }
290
291    #[rstest]
292    fn test_get_redis_url_full_config_with_ssl() {
293        let config_json = json!({
294            "host": "example.com",
295            "port": 6380,
296            "username": "user",
297            "password": "pass",
298            "ssl": true,
299        });
300        let config: RedisCacheConfig = serde_json::from_value(config_json).unwrap();
301        let (url, redacted_url) = get_redis_url(&config);
302        assert_eq!(url, "rediss://user:pass@example.com:6380");
303        assert_eq!(redacted_url, "rediss://user:pass@example.com:6380");
304    }
305
306    #[rstest]
307    fn test_get_redis_url_full_config_without_ssl() {
308        let config_json = json!({
309            "host": "example.com",
310            "port": 6380,
311            "username": "username",
312            "password": "password",
313            "ssl": false,
314        });
315        let config: RedisCacheConfig = serde_json::from_value(config_json).unwrap();
316        let (url, redacted_url) = get_redis_url(&config);
317        assert_eq!(url, "redis://username:password@example.com:6380");
318        assert_eq!(redacted_url, "redis://username:pa...rd@example.com:6380");
319    }
320
321    #[rstest]
322    fn test_get_redis_url_missing_username_and_password() {
323        let config_json = json!({
324            "host": "example.com",
325            "port": 6380,
326            "ssl": false,
327        });
328        let config: RedisCacheConfig = serde_json::from_value(config_json).unwrap();
329        let (url, redacted_url) = get_redis_url(&config);
330        assert_eq!(url, "redis://example.com:6380");
331        assert_eq!(redacted_url, "redis://example.com:6380");
332    }
333
334    #[rstest]
335    fn test_get_redis_url_ssl_default_false() {
336        let config_json = json!({
337            "host": "example.com",
338            "port": 6380,
339            "username": "username",
340            "password": "password",
341            // "ssl" is intentionally omitted to test default behavior
342        });
343        let config: RedisCacheConfig = serde_json::from_value(config_json).unwrap();
344        let (url, redacted_url) = get_redis_url(&config);
345        assert_eq!(url, "redis://username:password@example.com:6380");
346        assert_eq!(redacted_url, "redis://username:pa...rd@example.com:6380");
347    }
348
349    #[rstest]
350    fn test_get_stream_key_with_trader_prefix_and_instance_id() {
351        let trader_id = TraderId::from("tester-123");
352        let instance_id = UUID4::new();
353        let config = MessageBusConfig {
354            use_instance_id: true,
355            ..Default::default()
356        };
357
358        let key = get_stream_key(trader_id, instance_id, &config);
359        assert_eq!(key, format!("trader-tester-123:{instance_id}:stream"));
360    }
361
362    #[rstest]
363    fn test_get_stream_key_without_trader_prefix_or_instance_id() {
364        let trader_id = TraderId::from("tester-123");
365        let instance_id = UUID4::new();
366        let config = MessageBusConfig {
367            use_trader_prefix: false,
368            use_trader_id: false,
369            ..Default::default()
370        };
371
372        let key = get_stream_key(trader_id, instance_id, &config);
373        assert_eq!(key, "stream".to_string());
374    }
375
376    #[rstest]
377    fn test_get_index_key_without_prefix() {
378        let key = "index:order_position";
379        assert_eq!(get_index_key(key).unwrap(), "index:order_position");
380    }
381
382    #[rstest]
383    fn test_get_index_key_with_trader_prefix() {
384        let key = "trader-tester-123:index:order_position";
385        assert_eq!(get_index_key(key).unwrap(), "index:order_position");
386    }
387
388    #[rstest]
389    fn test_get_index_key_with_instance_id() {
390        let key = "trader-tester-123:abc-uuid-123:index:order_position";
391        assert_eq!(get_index_key(key).unwrap(), "index:order_position");
392    }
393
394    #[rstest]
395    fn test_get_index_key_invalid() {
396        let key = "no_index_pattern";
397        assert!(get_index_key(key).is_err());
398    }
399}
nautilus_infrastructure/redis/mod.rs

nautilus_infrastructure/redis/
mod.rs
