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

nautilus_infrastructure/redis/
mod.rs
