nautilus_infrastructure/redis/
msgbus.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//! Redis-backed message bus backing for the system.
17//!
18//! # Architecture
19//!
20//! Runs background tasks on `get_runtime()` for publishing, stream reading,
21//! and heartbeats. Messages are sent via an unbounded `tokio::sync::mpsc`
22//! channel to the publish task, which buffers and writes them to Redis
23//! streams. Each background task owns its own Redis connection created on
24//! the Nautilus runtime.
25//!
26//! Handles are stored as `Option<JoinHandle>` for idempotent shutdown via
27//! `close_async()`. The synchronous `close()` uses `block_in_place` to
28//! bridge into the async shutdown path and must be called from outside any
29//! `current_thread` Tokio runtime.
30
31use std::{
32    collections::{HashMap, VecDeque},
33    fmt::Debug,
34    sync::{
35        Arc,
36        atomic::{AtomicBool, Ordering},
37    },
38    time::Duration,
39};
40
41use bytes::Bytes;
42use futures::stream::Stream;
43use nautilus_common::{
44    enums::SerializationEncoding,
45    live::get_runtime,
46    logging::{log_task_error, log_task_started, log_task_stopped},
47    msgbus::{
48        BusMessage, BusPayloadType, MessageBusBacking, MessageBusBackingFactory, MessageBusConfig,
49        switchboard::CLOSE_TOPIC,
50    },
51};
52use nautilus_core::{
53    UUID4,
54    time::{duration_since_unix_epoch, get_atomic_clock_realtime},
55};
56use nautilus_cryptography::providers::install_cryptographic_provider;
57use nautilus_model::identifiers::TraderId;
58use redis::{AsyncCommands, RetryMethod, aio::ConnectionManager, streams};
59use serde::{Deserialize, Serialize};
60use streams::StreamReadOptions;
61use ustr::Ustr;
62
63use super::{REDIS_MINID, REDIS_XTRIM, await_handle};
64use crate::redis::{RedisConnectionConfig, create_redis_connection, get_stream_key};
65
66const MSGBUS_PUBLISH: &str = "msgbus-publish";
67const MSGBUS_STREAM: &str = "msgbus-stream";
68const MSGBUS_HEARTBEAT: &str = "msgbus-heartbeat";
69const HEARTBEAT_TOPIC: &str = "health:heartbeat";
70const TRIM_BUFFER_SECS: u64 = 60;
71
72type RedisStreamBulk = Vec<HashMap<String, Vec<HashMap<String, redis::Value>>>>;
73
74/// Configuration for a Redis-backed message bus backing.
75///
76/// Redis 6.2 or higher is required for correct operation.
77#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
78#[serde(default, deny_unknown_fields)]
79#[cfg_attr(
80    feature = "python",
81    pyo3::pyclass(
82        module = "nautilus_trader.core.nautilus_pyo3.infrastructure",
83        from_py_object
84    )
85)]
86#[cfg_attr(
87    feature = "python",
88    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.infrastructure")
89)]
90pub struct RedisMessageBusConfig {
91    /// The Redis host address. If `None`, `127.0.0.1` is used.
92    pub host: Option<String>,
93    /// The Redis port. If `None`, `6379` is used.
94    pub port: Option<u16>,
95    /// The Redis account username.
96    pub username: Option<String>,
97    /// The Redis account password.
98    pub password: Option<String>,
99    /// If Redis should use an SSL-enabled connection.
100    pub ssl: bool,
101    /// The timeout (in seconds) to wait for a new connection.
102    pub connection_timeout: u16,
103    /// The timeout (in seconds) to wait for a response.
104    pub response_timeout: u16,
105    /// The number of retry attempts with exponential backoff for connection attempts.
106    pub number_of_retries: usize,
107    /// The base value for exponential backoff calculation.
108    pub exponent_base: u64,
109    /// The maximum delay between retry attempts (in seconds).
110    pub max_delay: u64,
111    /// The multiplication factor for retry delay calculation.
112    pub factor: u64,
113}
114
115impl Debug for RedisMessageBusConfig {
116    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
117        let redacted = self.password.as_ref().map(|_| "***");
118        f.debug_struct(stringify!(RedisMessageBusConfig))
119            .field("host", &self.host)
120            .field("port", &self.port)
121            .field("username", &self.username)
122            .field("password", &redacted)
123            .field("ssl", &self.ssl)
124            .field("connection_timeout", &self.connection_timeout)
125            .field("response_timeout", &self.response_timeout)
126            .field("number_of_retries", &self.number_of_retries)
127            .field("exponent_base", &self.exponent_base)
128            .field("max_delay", &self.max_delay)
129            .field("factor", &self.factor)
130            .finish()
131    }
132}
133
134impl Default for RedisMessageBusConfig {
135    fn default() -> Self {
136        Self {
137            host: None,
138            port: None,
139            username: None,
140            password: None,
141            ssl: false,
142            connection_timeout: 20,
143            response_timeout: 20,
144            number_of_retries: 100,
145            exponent_base: 2,
146            max_delay: 1000,
147            factor: 2,
148        }
149    }
150}
151
152impl RedisConnectionConfig for RedisMessageBusConfig {
153    fn host(&self) -> Option<&str> {
154        self.host.as_deref()
155    }
156
157    fn port(&self) -> Option<u16> {
158        self.port
159    }
160
161    fn username(&self) -> Option<&str> {
162        self.username.as_deref()
163    }
164
165    fn password(&self) -> Option<&str> {
166        self.password.as_deref()
167    }
168
169    fn ssl(&self) -> bool {
170        self.ssl
171    }
172
173    fn connection_timeout(&self) -> u16 {
174        self.connection_timeout
175    }
176
177    fn response_timeout(&self) -> u16 {
178        self.response_timeout
179    }
180
181    fn number_of_retries(&self) -> usize {
182        self.number_of_retries
183    }
184
185    fn exponent_base(&self) -> u64 {
186        self.exponent_base
187    }
188
189    fn max_delay(&self) -> u64 {
190        self.max_delay
191    }
192
193    fn factor(&self) -> u64 {
194        self.factor
195    }
196}
197
198/// Factory for constructing Redis message bus backings.
199#[derive(Debug, Clone)]
200pub struct RedisMessageBusFactory {
201    config: RedisMessageBusConfig,
202}
203
204impl RedisMessageBusFactory {
205    /// Creates a new [`RedisMessageBusFactory`] from the given Redis configuration.
206    #[must_use]
207    pub const fn new(config: RedisMessageBusConfig) -> Self {
208        Self { config }
209    }
210}
211
212impl MessageBusBackingFactory for RedisMessageBusFactory {
213    fn create(
214        &self,
215        trader_id: TraderId,
216        instance_id: UUID4,
217        config: MessageBusConfig,
218    ) -> anyhow::Result<Box<dyn MessageBusBacking>> {
219        Ok(Box::new(RedisMessageBusBacking::new(
220            trader_id,
221            instance_id,
222            config,
223            self.config.clone(),
224        )?))
225    }
226}
227
228#[cfg_attr(
229    feature = "python",
230    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.infrastructure")
231)]
232pub struct RedisMessageBusBacking {
233    /// The trader ID for this message bus backing.
234    pub trader_id: TraderId,
235    /// The instance ID for this message bus backing.
236    pub instance_id: UUID4,
237    pub_tx: tokio::sync::mpsc::UnboundedSender<BusMessage>,
238    pub_handle: Option<tokio::task::JoinHandle<()>>,
239    stream_rx: Option<tokio::sync::mpsc::Receiver<BusMessage>>,
240    stream_handle: Option<tokio::task::JoinHandle<()>>,
241    stream_signal: Arc<AtomicBool>,
242    heartbeat_handle: Option<tokio::task::JoinHandle<()>>,
243    heartbeat_signal: Arc<AtomicBool>,
244}
245
246impl Debug for RedisMessageBusBacking {
247    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
248        f.debug_struct(stringify!(RedisMessageBusBacking))
249            .field("trader_id", &self.trader_id)
250            .field("instance_id", &self.instance_id)
251            .finish_non_exhaustive()
252    }
253}
254
255impl RedisMessageBusBacking {
256    /// Creates a new [`RedisMessageBusBacking`] instance for the given `trader_id`, `instance_id`, and `config`.
257    ///
258    /// # Errors
259    ///
260    /// Returns an error if the heartbeat interval is configured as zero seconds.
261    pub fn new(
262        trader_id: TraderId,
263        instance_id: UUID4,
264        config: MessageBusConfig,
265        backing: RedisMessageBusConfig,
266    ) -> anyhow::Result<Self> {
267        install_cryptographic_provider();
268
269        if config.heartbeat_interval_secs == Some(0) {
270            anyhow::bail!("heartbeat_interval_secs must be greater than 0");
271        }
272
273        let external_streams = config.external_streams.clone().unwrap_or_default();
274        let heartbeat_interval_secs = config.heartbeat_interval_secs;
275        let publish = backing.clone();
276
277        let (pub_tx, pub_rx) = tokio::sync::mpsc::unbounded_channel::<BusMessage>();
278
279        // Create publish task (start the runtime here for now)
280        let pub_handle = Some(get_runtime().spawn(async move {
281            if let Err(e) = publish_messages(pub_rx, trader_id, instance_id, config, publish).await
282            {
283                log_task_error(MSGBUS_PUBLISH, &e);
284            }
285        }));
286
287        // Conditionally create stream task and channel if external streams configured
288        let stream_signal = Arc::new(AtomicBool::new(false));
289        let (stream_rx, stream_handle) = if external_streams.is_empty() {
290            (None, None)
291        } else {
292            let stream_signal_clone = stream_signal.clone();
293            let (stream_tx, stream_rx) = tokio::sync::mpsc::channel::<BusMessage>(100_000);
294            (
295                Some(stream_rx),
296                Some(get_runtime().spawn(async move {
297                    if let Err(e) = stream_messages(
298                        stream_tx,
299                        backing.clone(),
300                        external_streams,
301                        stream_signal_clone,
302                    )
303                    .await
304                    {
305                        log_task_error(MSGBUS_STREAM, &e);
306                    }
307                })),
308            )
309        };
310
311        // Create heartbeat task
312        let heartbeat_signal = Arc::new(AtomicBool::new(false));
313        let heartbeat_handle = if let Some(heartbeat_interval_secs) = heartbeat_interval_secs {
314            let signal = heartbeat_signal.clone();
315            let pub_tx_clone = pub_tx.clone();
316
317            Some(get_runtime().spawn(async move {
318                run_heartbeat(heartbeat_interval_secs, signal, pub_tx_clone).await;
319            }))
320        } else {
321            None
322        };
323
324        Ok(Self {
325            trader_id,
326            instance_id,
327            pub_tx,
328            pub_handle,
329            stream_rx,
330            stream_handle,
331            stream_signal,
332            heartbeat_handle,
333            heartbeat_signal,
334        })
335    }
336}
337
338impl MessageBusBacking for RedisMessageBusBacking {
339    /// Returns whether the message bus backing publishing channel is closed.
340    fn is_closed(&self) -> bool {
341        self.pub_tx.is_closed()
342    }
343
344    /// Queues a serialized bus message for external publication.
345    fn publish(&self, message: BusMessage) {
346        if let Err(e) = self.pub_tx.send(message) {
347            log::error!("Failed to send message: {e}");
348        }
349    }
350
351    fn take_receiver(&mut self) -> anyhow::Result<tokio::sync::mpsc::Receiver<BusMessage>> {
352        self.get_stream_receiver()
353    }
354
355    /// Closes the message bus backing.
356    fn close(&mut self) {
357        log::debug!("Closing");
358
359        self.stream_signal.store(true, Ordering::Relaxed);
360        self.heartbeat_signal.store(true, Ordering::Relaxed);
361
362        if !self.pub_tx.is_closed() {
363            let msg = BusMessage::new_close();
364
365            if let Err(e) = self.pub_tx.send(msg) {
366                log::warn!("Failed to send close message: {e:?}");
367            }
368        }
369
370        // Keep close sync for now to avoid async trait method
371        tokio::task::block_in_place(|| {
372            get_runtime().block_on(async {
373                self.close_async().await;
374            });
375        });
376
377        log::debug!("Closed");
378    }
379}
380
381impl RedisMessageBusBacking {
382    /// Retrieves the Redis stream receiver for this message bus instance.
383    ///
384    /// # Errors
385    ///
386    /// Returns an error if the stream receiver has already been taken.
387    pub fn get_stream_receiver(
388        &mut self,
389    ) -> anyhow::Result<tokio::sync::mpsc::Receiver<BusMessage>> {
390        self.stream_rx
391            .take()
392            .ok_or_else(|| anyhow::anyhow!("Stream receiver already taken"))
393    }
394
395    /// Streams messages arriving on the stream receiver channel.
396    pub fn stream(
397        mut stream_rx: tokio::sync::mpsc::Receiver<BusMessage>,
398    ) -> impl Stream<Item = BusMessage> + 'static {
399        async_stream::stream! {
400            while let Some(msg) = stream_rx.recv().await {
401                yield msg;
402            }
403        }
404    }
405
406    pub async fn close_async(&mut self) {
407        await_handle(self.pub_handle.take(), MSGBUS_PUBLISH).await;
408        await_handle(self.stream_handle.take(), MSGBUS_STREAM).await;
409        await_handle(self.heartbeat_handle.take(), MSGBUS_HEARTBEAT).await;
410    }
411}
412
413/// Publishes messages received on `rx` to Redis streams for the given `trader_id` and `instance_id`, using `config`.
414///
415/// # Errors
416///
417/// Returns an error if:
418/// - The backing configuration is missing in `config`.
419/// - Establishing the Redis connection fails.
420/// - Any Redis command fails during publishing.
421pub async fn publish_messages(
422    mut rx: tokio::sync::mpsc::UnboundedReceiver<BusMessage>,
423    trader_id: TraderId,
424    instance_id: UUID4,
425    config: MessageBusConfig,
426    backing: RedisMessageBusConfig,
427) -> anyhow::Result<()> {
428    log_task_started(MSGBUS_PUBLISH);
429
430    let mut con = create_redis_connection(MSGBUS_PUBLISH, &backing).await?;
431    let stream_key = get_stream_key(trader_id, instance_id, &config);
432
433    // Auto-trimming
434    let autotrim_duration = config
435        .autotrim_mins
436        .filter(|&mins| mins > 0)
437        .map(|mins| Duration::from_secs(u64::from(mins) * 60));
438    let mut last_trim_index: HashMap<String, usize> = HashMap::new();
439
440    // Buffering
441    let mut buffer: VecDeque<BusMessage> = VecDeque::new();
442    let buffer_interval = Duration::from_millis(u64::from(config.buffer_interval_ms.unwrap_or(0)));
443
444    // A sleep used to trigger periodic flushing of the buffer.
445    // When `buffer_interval` is zero we skip using the timer and flush immediately
446    // after every message.
447    let flush_timer = tokio::time::sleep(buffer_interval);
448    tokio::pin!(flush_timer);
449
450    loop {
451        tokio::select! {
452            maybe_msg = rx.recv() => {
453                if let Some(msg) = maybe_msg {
454                    if msg.topic == CLOSE_TOPIC {
455                        log::debug!("Received close message");
456                        // Ensure we exit the loop after flushing any remaining messages.
457                        if !buffer.is_empty() {
458                            drain_buffer(
459                                &mut con,
460                                &stream_key,
461                                config.stream_per_topic,
462                                autotrim_duration,
463                                &mut last_trim_index,
464                                &mut buffer,
465                            ).await?;
466                        }
467                        break;
468                    }
469
470                    buffer.push_back(msg);
471
472                    if buffer_interval.is_zero() {
473                        // Immediate flush mode
474                        drain_buffer(
475                            &mut con,
476                            &stream_key,
477                            config.stream_per_topic,
478                            autotrim_duration,
479                            &mut last_trim_index,
480                            &mut buffer,
481                        ).await?;
482                    }
483                } else {
484                    log::debug!("Channel hung up");
485                    break;
486                }
487            }
488            // Only poll the timer when the interval is non-zero. This avoids
489            // unnecessarily waking the task when immediate flushing is enabled.
490            () = &mut flush_timer, if !buffer_interval.is_zero() => {
491                if !buffer.is_empty() {
492                    drain_buffer(
493                        &mut con,
494                        &stream_key,
495                        config.stream_per_topic,
496                        autotrim_duration,
497                        &mut last_trim_index,
498                        &mut buffer,
499                    ).await?;
500                }
501
502                // Schedule the next tick
503                flush_timer.as_mut().reset(tokio::time::Instant::now() + buffer_interval);
504            }
505        }
506    }
507
508    // Drain any remaining messages
509    if !buffer.is_empty() {
510        drain_buffer(
511            &mut con,
512            &stream_key,
513            config.stream_per_topic,
514            autotrim_duration,
515            &mut last_trim_index,
516            &mut buffer,
517        )
518        .await?;
519    }
520
521    log_task_stopped(MSGBUS_PUBLISH);
522    Ok(())
523}
524
525async fn drain_buffer(
526    conn: &mut redis::aio::ConnectionManager,
527    stream_key: &str,
528    stream_per_topic: bool,
529    autotrim_duration: Option<Duration>,
530    last_trim_index: &mut HashMap<String, usize>,
531    buffer: &mut VecDeque<BusMessage>,
532) -> anyhow::Result<()> {
533    let mut pipe = redis::pipe();
534    pipe.atomic();
535
536    for msg in buffer.drain(..) {
537        let encoding = msg.encoding.to_string();
538        let items: Vec<(&str, &[u8])> = vec![
539            ("topic", msg.topic.as_ref()),
540            ("type", msg.payload_type.as_str().as_bytes()),
541            ("payload", msg.payload.as_ref()),
542            ("encoding", encoding.as_bytes()),
543        ];
544        let stream_key = if stream_per_topic {
545            format!("{stream_key}:{}", msg.topic)
546        } else {
547            stream_key.to_string()
548        };
549        pipe.xadd(&stream_key, "*", &items);
550
551        if autotrim_duration.is_none() {
552            continue; // Nothing else to do
553        }
554
555        // Autotrim stream
556        let last_trim_ms = last_trim_index.entry(stream_key.clone()).or_insert(0); // Remove clone
557        let unix_duration_now = duration_since_unix_epoch();
558        let trim_buffer = Duration::from_secs(TRIM_BUFFER_SECS);
559
560        // Improve efficiency of this by batching
561        if *last_trim_ms < unix_duration_now.saturating_sub(trim_buffer).as_millis() as usize {
562            let min_timestamp_ms = unix_duration_now
563                .saturating_sub(autotrim_duration.unwrap())
564                .as_millis() as usize;
565            let result: Result<(), redis::RedisError> = redis::cmd(REDIS_XTRIM)
566                .arg(stream_key.clone())
567                .arg(REDIS_MINID)
568                .arg(min_timestamp_ms)
569                .query_async(conn)
570                .await;
571
572            if let Err(e) = result {
573                log::error!("Error trimming stream '{stream_key}': {e}");
574            } else {
575                last_trim_index.insert(stream_key.clone(), unix_duration_now.as_millis() as usize);
576            }
577        }
578    }
579
580    pipe.query_async(conn).await.map_err(anyhow::Error::from)
581}
582
583/// Streams messages from Redis streams and sends them over the provided `tx` channel.
584///
585/// # Errors
586///
587/// Returns an error if:
588/// - Establishing the Redis connection fails before the terminate signal is received.
589/// - A Redis read operation returns a non-retryable error.
590pub async fn stream_messages(
591    tx: tokio::sync::mpsc::Sender<BusMessage>,
592    config: RedisMessageBusConfig,
593    stream_keys: Vec<String>,
594    stream_signal: Arc<AtomicBool>,
595) -> anyhow::Result<()> {
596    log_task_started(MSGBUS_STREAM);
597
598    let Some(mut con) = connect_stream_connection(&config, &stream_signal).await? else {
599        log_task_stopped(MSGBUS_STREAM);
600        return Ok(());
601    };
602
603    let mut read_error_count = 0;
604
605    let stream_keys = &stream_keys
606        .iter()
607        .map(String::as_str)
608        .collect::<Vec<&str>>();
609
610    log::debug!("Listening to streams: [{}]", stream_keys.join(", "));
611
612    // Start streaming from current timestamp
613    let clock = get_atomic_clock_realtime();
614    let timestamp_ms = clock.get_time_ms();
615    let initial_id = timestamp_ms.to_string();
616
617    let mut last_ids: HashMap<String, String> = stream_keys
618        .iter()
619        .map(|&key| (key.to_string(), initial_id.clone()))
620        .collect();
621
622    let opts = StreamReadOptions::default().block(100);
623
624    'outer: loop {
625        if stream_signal.load(Ordering::Relaxed) {
626            log::debug!("Received streaming terminate signal");
627            break;
628        }
629
630        let ids: Vec<String> = stream_keys
631            .iter()
632            .map(|&key| last_ids[key].clone())
633            .collect();
634        let id_refs: Vec<&str> = ids.iter().map(String::as_str).collect();
635
636        let result: Result<RedisStreamBulk, _> =
637            con.xread_options(&[&stream_keys], &[&id_refs], &opts).await;
638
639        match result {
640            Ok(stream_bulk) => {
641                read_error_count = 0;
642
643                if stream_bulk.is_empty() {
644                    // Timeout occurred: no messages received
645                    continue;
646                }
647
648                for entry in &stream_bulk {
649                    for (stream_key, stream_msgs) in entry {
650                        for stream_msg in stream_msgs {
651                            for (id, array) in stream_msg {
652                                last_ids.insert(stream_key.clone(), id.clone());
653
654                                match decode_bus_message(array) {
655                                    Ok(msg) => {
656                                        if let Err(e) = tx.send(msg).await {
657                                            log::debug!("Channel closed: {e:?}");
658                                            break 'outer; // End streaming
659                                        }
660                                    }
661                                    Err(e) => {
662                                        log::error!("{e:?}");
663                                    }
664                                }
665                            }
666                        }
667                    }
668                }
669            }
670            Err(e) => {
671                if !is_retryable_stream_error(&e) {
672                    anyhow::bail!("Error reading from stream: {e:?}");
673                }
674
675                log::error!("Error reading from stream: {e:?}");
676
677                let Some(reconnected) =
678                    reconnect_stream_connection(&config, &stream_signal, &mut read_error_count)
679                        .await?
680                else {
681                    break;
682                };
683                con = reconnected;
684            }
685        }
686    }
687
688    log_task_stopped(MSGBUS_STREAM);
689    Ok(())
690}
691
692async fn connect_stream_connection(
693    config: &RedisMessageBusConfig,
694    stream_signal: &Arc<AtomicBool>,
695) -> anyhow::Result<Option<ConnectionManager>> {
696    let connect = create_redis_connection(MSGBUS_STREAM, config);
697    let terminate = wait_for_stream_signal(stream_signal);
698
699    tokio::pin!(connect);
700    tokio::pin!(terminate);
701
702    tokio::select! {
703        result = &mut connect => result.map(Some),
704        () = &mut terminate => Ok(None),
705    }
706}
707
708async fn reconnect_stream_connection(
709    config: &RedisMessageBusConfig,
710    stream_signal: &Arc<AtomicBool>,
711    read_error_count: &mut usize,
712) -> anyhow::Result<Option<ConnectionManager>> {
713    loop {
714        let retry_delay = stream_retry_delay(config, *read_error_count);
715        *read_error_count = (*read_error_count).saturating_add(1);
716
717        if !wait_for_retry_delay(retry_delay, stream_signal).await {
718            return Ok(None);
719        }
720
721        match connect_stream_connection(config, stream_signal).await {
722            Ok(Some(con)) => return Ok(Some(con)),
723            Ok(None) => return Ok(None),
724            Err(e) => {
725                log::error!("Error reconnecting to stream: {e:?}");
726            }
727        }
728    }
729}
730
731fn stream_retry_delay(config: &RedisMessageBusConfig, attempt: usize) -> Duration {
732    let exponent = u32::try_from(attempt.min(32)).unwrap_or(32);
733    let delay_ms = config
734        .factor
735        .saturating_mul(config.exponent_base.saturating_pow(exponent));
736    let max_delay = Duration::from_secs(config.max_delay);
737
738    Duration::from_millis(delay_ms)
739        .min(max_delay)
740        .max(Duration::from_millis(1))
741}
742
743fn is_retryable_stream_error(error: &redis::RedisError) -> bool {
744    matches!(
745        error.retry_method(),
746        RetryMethod::Reconnect
747            | RetryMethod::ReconnectFromInitialConnections
748            | RetryMethod::RetryImmediately
749            | RetryMethod::WaitAndRetry
750    )
751}
752
753async fn wait_for_retry_delay(retry_delay: Duration, stream_signal: &Arc<AtomicBool>) -> bool {
754    let retry_timer = tokio::time::sleep(retry_delay);
755    let terminate = wait_for_stream_signal(stream_signal);
756
757    tokio::pin!(retry_timer);
758    tokio::pin!(terminate);
759
760    tokio::select! {
761        () = &mut retry_timer => true,
762        () = &mut terminate => false,
763    }
764}
765
766async fn wait_for_stream_signal(stream_signal: &Arc<AtomicBool>) {
767    let check_timer = tokio::time::interval(Duration::from_millis(100));
768
769    tokio::pin!(check_timer);
770
771    while !stream_signal.load(Ordering::Relaxed) {
772        check_timer.tick().await;
773    }
774}
775
776// Redis fields are unordered, and older streams may omit type or encoding headers
777fn decode_bus_message(stream_msg: &redis::Value) -> anyhow::Result<BusMessage> {
778    let redis::Value::Array(fields) = stream_msg else {
779        anyhow::bail!("Invalid stream message format: {stream_msg:?}");
780    };
781
782    if fields.len() < 4 || fields.len() % 2 != 0 {
783        anyhow::bail!("Invalid stream message format: {stream_msg:?}");
784    }
785
786    let mut topic: Option<String> = None;
787    let mut payload_type = BusPayloadType::Custom(Ustr::default());
788    let mut encoding = SerializationEncoding::default();
789    let mut payload: Option<Bytes> = None;
790
791    for pair in fields.chunks_exact(2) {
792        let redis::Value::BulkString(key) = &pair[0] else {
793            anyhow::bail!("Invalid stream field key: {stream_msg:?}");
794        };
795
796        match key.as_slice() {
797            b"topic" => {
798                let redis::Value::BulkString(bytes) = &pair[1] else {
799                    anyhow::bail!("Invalid topic format: {stream_msg:?}");
800                };
801                topic = Some(
802                    String::from_utf8(bytes.clone())
803                        .map_err(|e| anyhow::anyhow!("Error parsing topic: {e}"))?,
804                );
805            }
806            b"type" => {
807                let redis::Value::BulkString(bytes) = &pair[1] else {
808                    anyhow::bail!("Invalid type format: {stream_msg:?}");
809                };
810                let type_name = std::str::from_utf8(bytes)
811                    .map_err(|e| anyhow::anyhow!("Error parsing type: {e}"))?;
812                payload_type = BusPayloadType::from_name(type_name);
813            }
814            b"encoding" => {
815                let redis::Value::BulkString(bytes) = &pair[1] else {
816                    anyhow::bail!("Invalid encoding format: {stream_msg:?}");
817                };
818                let value = std::str::from_utf8(bytes)
819                    .map_err(|e| anyhow::anyhow!("Error parsing encoding: {e}"))?;
820                encoding = value
821                    .parse()
822                    .map_err(|e| anyhow::anyhow!("Error parsing encoding: {e}"))?;
823            }
824            b"payload" => {
825                let redis::Value::BulkString(bytes) = &pair[1] else {
826                    anyhow::bail!("Invalid payload format: {stream_msg:?}");
827                };
828                payload = Some(Bytes::copy_from_slice(bytes));
829            }
830            _ => {}
831        }
832    }
833
834    let Some(topic) = topic else {
835        anyhow::bail!("Stream message missing topic: {stream_msg:?}");
836    };
837    let Some(payload) = payload else {
838        anyhow::bail!("Stream message missing payload: {stream_msg:?}");
839    };
840
841    Ok(BusMessage::with_str_topic(
842        topic,
843        payload_type,
844        payload,
845        encoding,
846    ))
847}
848
849async fn run_heartbeat(
850    heartbeat_interval_secs: u16,
851    signal: Arc<AtomicBool>,
852    pub_tx: tokio::sync::mpsc::UnboundedSender<BusMessage>,
853) {
854    log_task_started("heartbeat");
855    log::debug!("Heartbeat at {heartbeat_interval_secs} second intervals");
856
857    let heartbeat_interval = Duration::from_secs(u64::from(heartbeat_interval_secs));
858    let heartbeat_timer = tokio::time::interval(heartbeat_interval);
859
860    let check_interval = Duration::from_millis(100);
861    let check_timer = tokio::time::interval(check_interval);
862
863    tokio::pin!(heartbeat_timer);
864    tokio::pin!(check_timer);
865
866    loop {
867        if signal.load(Ordering::Relaxed) {
868            log::debug!("Received heartbeat terminate signal");
869            break;
870        }
871
872        tokio::select! {
873            _ = heartbeat_timer.tick() => {
874                let heartbeat = create_heartbeat_msg();
875                if let Err(e) = pub_tx.send(heartbeat) {
876                    // We expect an error if the channel is closed during shutdown
877                    log::debug!("Error sending heartbeat: {e}");
878                }
879            },
880            _ = check_timer.tick() => {}
881        }
882    }
883
884    log_task_stopped("heartbeat");
885}
886
887fn create_heartbeat_msg() -> BusMessage {
888    let payload = Bytes::from(chrono::Utc::now().to_rfc3339().into_bytes());
889    BusMessage::with_str_topic(
890        HEARTBEAT_TOPIC,
891        BusPayloadType::Custom(Ustr::default()),
892        payload,
893        SerializationEncoding::default(),
894    )
895}
896
897#[cfg(test)]
898mod tests {
899    use nautilus_common::{msgbus::external_io_from_backing, testing::wait_until_async};
900    use redis::Value;
901    use rstest::*;
902    use serde_json::json;
903
904    use super::*;
905
906    #[rstest]
907    fn test_default_redis_message_bus_config() {
908        let config = RedisMessageBusConfig::default();
909
910        assert_eq!(config.host, None);
911        assert_eq!(config.port, None);
912        assert_eq!(config.username, None);
913        assert_eq!(config.password, None);
914        assert!(!config.ssl);
915        assert_eq!(config.connection_timeout, 20);
916        assert_eq!(config.response_timeout, 20);
917        assert_eq!(config.number_of_retries, 100);
918        assert_eq!(config.exponent_base, 2);
919        assert_eq!(config.max_delay, 1000);
920        assert_eq!(config.factor, 2);
921    }
922
923    #[rstest]
924    fn test_deserialize_redis_message_bus_config() {
925        let config_json = json!({
926            "host": "localhost",
927            "port": 6379,
928            "username": "user",
929            "password": "pass",
930            "ssl": true,
931            "connection_timeout": 30,
932            "response_timeout": 10,
933            "number_of_retries": 3,
934            "exponent_base": 2,
935            "max_delay": 10,
936            "factor": 2
937        });
938
939        let config: RedisMessageBusConfig = serde_json::from_value(config_json).unwrap();
940
941        assert_eq!(config.host, Some("localhost".to_string()));
942        assert_eq!(config.port, Some(6379));
943        assert_eq!(config.username, Some("user".to_string()));
944        assert_eq!(config.password, Some("pass".to_string()));
945        assert!(config.ssl);
946        assert_eq!(config.connection_timeout, 30);
947        assert_eq!(config.response_timeout, 10);
948        assert_eq!(config.number_of_retries, 3);
949        assert_eq!(config.exponent_base, 2);
950        assert_eq!(config.max_delay, 10);
951        assert_eq!(config.factor, 2);
952    }
953
954    #[rstest]
955    fn test_deserialize_redis_message_bus_config_rejects_type_selector() {
956        let config_json = json!({
957            "type": "redis",
958        });
959
960        let error = serde_json::from_value::<RedisMessageBusConfig>(config_json).unwrap_err();
961
962        assert!(error.to_string().contains("unknown field `type`"));
963    }
964
965    #[rstest]
966    fn test_decode_bus_message_valid() {
967        let stream_msg = Value::Array(vec![
968            Value::BulkString(b"topic".to_vec()),
969            Value::BulkString(b"topic1".to_vec()),
970            Value::BulkString(b"type".to_vec()),
971            Value::BulkString(b"QuoteTick".to_vec()),
972            Value::BulkString(b"payload".to_vec()),
973            Value::BulkString(b"data1".to_vec()),
974            Value::BulkString(b"encoding".to_vec()),
975            Value::BulkString(b"msgpack".to_vec()),
976        ]);
977
978        let result = decode_bus_message(&stream_msg);
979        assert!(result.is_ok());
980        let msg = result.unwrap();
981        assert_eq!(msg.topic, "topic1");
982        assert_eq!(msg.payload_type, BusPayloadType::QuoteTick);
983        assert_eq!(msg.encoding, SerializationEncoding::MsgPack);
984        assert_eq!(msg.payload, Bytes::from("data1"));
985    }
986
987    #[rstest]
988    fn test_decode_bus_message_defaults_legacy_headers() {
989        let stream_msg = Value::Array(vec![
990            Value::BulkString(b"topic".to_vec()),
991            Value::BulkString(b"topic1".to_vec()),
992            Value::BulkString(b"payload".to_vec()),
993            Value::BulkString(b"data1".to_vec()),
994        ]);
995
996        let result = decode_bus_message(&stream_msg);
997        assert!(result.is_ok());
998        let msg = result.unwrap();
999        assert_eq!(msg.topic, "topic1");
1000        assert_eq!(msg.payload_type, BusPayloadType::Custom(Ustr::default()));
1001        assert_eq!(msg.encoding, SerializationEncoding::Json);
1002        assert_eq!(msg.payload, Bytes::from("data1"));
1003    }
1004
1005    #[rstest]
1006    fn test_decode_bus_message_unknown_type_is_custom() {
1007        let stream_msg = Value::Array(vec![
1008            Value::BulkString(b"topic".to_vec()),
1009            Value::BulkString(b"topic1".to_vec()),
1010            Value::BulkString(b"type".to_vec()),
1011            Value::BulkString(b"UnknownPayload".to_vec()),
1012            Value::BulkString(b"payload".to_vec()),
1013            Value::BulkString(b"data1".to_vec()),
1014        ]);
1015
1016        let result = decode_bus_message(&stream_msg);
1017        assert!(result.is_ok());
1018        let msg = result.unwrap();
1019        assert_eq!(
1020            msg.payload_type,
1021            BusPayloadType::Custom(Ustr::from("UnknownPayload"))
1022        );
1023        assert_eq!(msg.encoding, SerializationEncoding::Json);
1024    }
1025
1026    #[rstest]
1027    fn test_decode_bus_message_accepts_unordered_metadata_fields() {
1028        let stream_msg = Value::Array(vec![
1029            Value::BulkString(b"payload".to_vec()),
1030            Value::BulkString(b"data1".to_vec()),
1031            Value::BulkString(b"encoding".to_vec()),
1032            Value::BulkString(b"msgpack".to_vec()),
1033            Value::BulkString(b"type".to_vec()),
1034            Value::BulkString(b"TradeTick".to_vec()),
1035            Value::BulkString(b"topic".to_vec()),
1036            Value::BulkString(b"topic1".to_vec()),
1037        ]);
1038
1039        let msg = decode_bus_message(&stream_msg).unwrap();
1040
1041        assert_eq!(msg.topic, "topic1");
1042        assert_eq!(msg.payload_type, BusPayloadType::TradeTick);
1043        assert_eq!(msg.encoding, SerializationEncoding::MsgPack);
1044        assert_eq!(msg.payload, Bytes::from("data1"));
1045    }
1046
1047    #[rstest]
1048    fn test_decode_bus_message_rejects_invalid_encoding_header() {
1049        let stream_msg = Value::Array(vec![
1050            Value::BulkString(b"topic".to_vec()),
1051            Value::BulkString(b"topic1".to_vec()),
1052            Value::BulkString(b"encoding".to_vec()),
1053            Value::BulkString(b"invalid".to_vec()),
1054            Value::BulkString(b"payload".to_vec()),
1055            Value::BulkString(b"data1".to_vec()),
1056        ]);
1057
1058        let error = decode_bus_message(&stream_msg).unwrap_err();
1059
1060        assert!(
1061            error.to_string().contains("Error parsing encoding"),
1062            "{error:?}"
1063        );
1064    }
1065
1066    #[rstest]
1067    fn test_decode_bus_message_missing_fields() {
1068        let stream_msg = Value::Array(vec![
1069            Value::BulkString(b"0".to_vec()),
1070            Value::BulkString(b"topic1".to_vec()),
1071        ]);
1072
1073        let result = decode_bus_message(&stream_msg);
1074        assert!(result.is_err());
1075        assert_eq!(
1076            format!("{}", result.unwrap_err()),
1077            "Invalid stream message format: array([bulk-string('\"0\"'), bulk-string('\"topic1\"')])"
1078        );
1079    }
1080
1081    #[rstest]
1082    fn test_decode_bus_message_invalid_topic_format() {
1083        let stream_msg = Value::Array(vec![
1084            Value::BulkString(b"topic".to_vec()),
1085            Value::Int(42),
1086            Value::BulkString(b"payload".to_vec()),
1087            Value::BulkString(b"data1".to_vec()),
1088        ]);
1089
1090        let result = decode_bus_message(&stream_msg);
1091        assert!(result.is_err());
1092        assert_eq!(
1093            format!("{}", result.unwrap_err()),
1094            "Invalid topic format: array([bulk-string('\"topic\"'), int(42), bulk-string('\"payload\"'), bulk-string('\"data1\"')])"
1095        );
1096    }
1097
1098    #[rstest]
1099    fn test_decode_bus_message_invalid_type_format() {
1100        let stream_msg = Value::Array(vec![
1101            Value::BulkString(b"topic".to_vec()),
1102            Value::BulkString(b"topic1".to_vec()),
1103            Value::BulkString(b"type".to_vec()),
1104            Value::Int(42),
1105            Value::BulkString(b"payload".to_vec()),
1106            Value::BulkString(b"data1".to_vec()),
1107        ]);
1108
1109        let result = decode_bus_message(&stream_msg);
1110        assert!(result.is_err());
1111        assert_eq!(
1112            format!("{}", result.unwrap_err()),
1113            "Invalid type format: array([bulk-string('\"topic\"'), bulk-string('\"topic1\"'), bulk-string('\"type\"'), int(42), bulk-string('\"payload\"'), bulk-string('\"data1\"')])"
1114        );
1115    }
1116
1117    #[rstest]
1118    fn test_decode_bus_message_invalid_encoding_format() {
1119        let stream_msg = Value::Array(vec![
1120            Value::BulkString(b"topic".to_vec()),
1121            Value::BulkString(b"topic1".to_vec()),
1122            Value::BulkString(b"encoding".to_vec()),
1123            Value::Int(42),
1124            Value::BulkString(b"payload".to_vec()),
1125            Value::BulkString(b"data1".to_vec()),
1126        ]);
1127
1128        let result = decode_bus_message(&stream_msg);
1129        assert!(result.is_err());
1130        assert_eq!(
1131            format!("{}", result.unwrap_err()),
1132            "Invalid encoding format: array([bulk-string('\"topic\"'), bulk-string('\"topic1\"'), bulk-string('\"encoding\"'), int(42), bulk-string('\"payload\"'), bulk-string('\"data1\"')])"
1133        );
1134    }
1135
1136    #[rstest]
1137    fn test_decode_bus_message_invalid_payload_format() {
1138        let stream_msg = Value::Array(vec![
1139            Value::BulkString(b"topic".to_vec()),
1140            Value::BulkString(b"topic1".to_vec()),
1141            Value::BulkString(b"payload".to_vec()),
1142            Value::Int(42),
1143        ]);
1144
1145        let result = decode_bus_message(&stream_msg);
1146        assert!(result.is_err());
1147        assert_eq!(
1148            format!("{}", result.unwrap_err()),
1149            "Invalid payload format: array([bulk-string('\"topic\"'), bulk-string('\"topic1\"'), bulk-string('\"payload\"'), int(42)])"
1150        );
1151    }
1152
1153    #[rstest]
1154    fn test_decode_bus_message_invalid_stream_msg_format() {
1155        let stream_msg = Value::BulkString(b"not an array".to_vec());
1156
1157        let result = decode_bus_message(&stream_msg);
1158        assert!(result.is_err());
1159        assert_eq!(
1160            format!("{}", result.unwrap_err()),
1161            "Invalid stream message format: bulk-string('\"not an array\"')"
1162        );
1163    }
1164
1165    #[rstest]
1166    fn test_new_rejects_zero_heartbeat_interval() {
1167        let trader_id = TraderId::from("tester-001");
1168        let instance_id = UUID4::new();
1169        let config = MessageBusConfig {
1170            heartbeat_interval_secs: Some(0),
1171            ..Default::default()
1172        };
1173
1174        let result = RedisMessageBusBacking::new(
1175            trader_id,
1176            instance_id,
1177            config,
1178            RedisMessageBusConfig::default(),
1179        );
1180
1181        assert!(result.is_err());
1182        assert_eq!(
1183            result.unwrap_err().to_string(),
1184            "heartbeat_interval_secs must be greater than 0"
1185        );
1186    }
1187
1188    #[rstest]
1189    fn test_stream_retry_delay_uses_config_bounds() {
1190        let config = RedisMessageBusConfig {
1191            factor: 10,
1192            exponent_base: 2,
1193            max_delay: 1,
1194            ..Default::default()
1195        };
1196
1197        assert_eq!(stream_retry_delay(&config, 0), Duration::from_millis(10));
1198        assert_eq!(stream_retry_delay(&config, 1), Duration::from_millis(20));
1199        assert_eq!(stream_retry_delay(&config, 10), Duration::from_secs(1));
1200    }
1201
1202    #[rstest]
1203    fn test_stream_error_retry_classification() {
1204        let dropped =
1205            redis::RedisError::from(std::io::Error::from(std::io::ErrorKind::ConnectionReset));
1206        let client: redis::RedisError = (redis::ErrorKind::Client, "client error").into();
1207
1208        assert!(is_retryable_stream_error(&dropped));
1209        assert!(!is_retryable_stream_error(&client));
1210    }
1211
1212    #[tokio::test]
1213    async fn test_wait_for_retry_delay_returns_false_when_signaled() {
1214        let stream_signal = Arc::new(AtomicBool::new(true));
1215        let signal = stream_signal.clone();
1216        let fut = async move { wait_for_retry_delay(Duration::from_secs(30), &signal).await };
1217
1218        let handle = tokio::spawn(fut);
1219
1220        wait_until_async(|| async { handle.is_finished() }, Duration::from_secs(1)).await;
1221
1222        assert!(!handle.await.unwrap());
1223    }
1224
1225    #[rstest]
1226    fn test_external_io_from_backing_takes_stream_receiver() {
1227        let (stream_tx, stream_rx) = tokio::sync::mpsc::channel::<BusMessage>(1);
1228        let backing = backing_with_stream_receiver(stream_rx);
1229        let message = BusMessage::with_str_topic(
1230            "events/data",
1231            BusPayloadType::QuoteTick,
1232            Bytes::from_static(b"payload"),
1233            SerializationEncoding::Json,
1234        );
1235
1236        let (_egress, mut ingress) = external_io_from_backing(Box::new(backing));
1237        stream_tx.try_send(message.clone()).unwrap();
1238        let mut receiver = ingress.take_receiver().unwrap();
1239        let received = receiver.try_recv().unwrap();
1240
1241        assert_eq!(received.topic, message.topic);
1242        assert_eq!(received.payload, message.payload);
1243        assert!(ingress.take_receiver().is_err());
1244    }
1245
1246    fn backing_with_stream_receiver(
1247        stream_rx: tokio::sync::mpsc::Receiver<BusMessage>,
1248    ) -> RedisMessageBusBacking {
1249        let (pub_tx, _pub_rx) = tokio::sync::mpsc::unbounded_channel::<BusMessage>();
1250        RedisMessageBusBacking {
1251            trader_id: TraderId::from("tester-001"),
1252            instance_id: UUID4::new(),
1253            pub_tx,
1254            pub_handle: None,
1255            stream_rx: Some(stream_rx),
1256            stream_handle: None,
1257            stream_signal: Arc::new(AtomicBool::new(false)),
1258            heartbeat_handle: None,
1259            heartbeat_signal: Arc::new(AtomicBool::new(false)),
1260        }
1261    }
1262}
1263
1264#[cfg(target_os = "linux")] // Run Redis tests on Linux platforms only
1265#[cfg(test)]
1266mod serial_tests {
1267    use std::{sync::mpsc, thread};
1268
1269    use nautilus_common::{
1270        enums::Environment,
1271        msgbus::{self, TypedHandler},
1272        testing::wait_until_async,
1273    };
1274    use nautilus_live::{
1275        builder::LiveNodeBuilder,
1276        config::{LiveExecEngineConfig, LiveNodeConfig},
1277    };
1278    use nautilus_model::data::{QuoteTick, TradeTick};
1279    use redis::aio::ConnectionManager;
1280    use rstest::*;
1281
1282    use super::*;
1283
1284    #[fixture]
1285    async fn redis_connection() -> ConnectionManager {
1286        let config = RedisMessageBusConfig::default();
1287        create_redis_connection(MSGBUS_STREAM, &config)
1288            .await
1289            .unwrap()
1290    }
1291
1292    #[rstest]
1293    #[tokio::test(flavor = "multi_thread")]
1294    async fn test_stream_messages_terminate_signal(#[future] redis_connection: ConnectionManager) {
1295        let _con = redis_connection.await;
1296        let (tx, mut rx) = tokio::sync::mpsc::channel::<BusMessage>(100);
1297
1298        let trader_id = TraderId::from("tester-001");
1299        let instance_id = UUID4::new();
1300        let config = MessageBusConfig {
1301            use_instance_id: true,
1302            ..Default::default()
1303        };
1304
1305        let stream_key = get_stream_key(trader_id, instance_id, &config);
1306        let external_streams = vec![stream_key.clone()];
1307        let stream_signal = Arc::new(AtomicBool::new(false));
1308        let stream_signal_clone = stream_signal.clone();
1309
1310        // Start the message streaming task
1311        let handle = tokio::spawn(async move {
1312            stream_messages(
1313                tx,
1314                RedisMessageBusConfig::default(),
1315                external_streams,
1316                stream_signal_clone,
1317            )
1318            .await
1319            .unwrap();
1320        });
1321
1322        stream_signal.store(true, Ordering::Relaxed);
1323        let _ = rx.recv().await; // Wait for the tx to close
1324
1325        // Shutdown and cleanup
1326        rx.close();
1327        handle.await.unwrap();
1328    }
1329
1330    #[rstest]
1331    #[tokio::test(flavor = "multi_thread")]
1332    async fn test_stream_messages_when_receiver_closed(
1333        #[future] redis_connection: ConnectionManager,
1334    ) {
1335        let mut con = redis_connection.await;
1336        let (tx, mut rx) = tokio::sync::mpsc::channel::<BusMessage>(100);
1337
1338        let trader_id = TraderId::from("tester-001");
1339        let instance_id = UUID4::new();
1340        let config = MessageBusConfig {
1341            use_instance_id: true,
1342            ..Default::default()
1343        };
1344
1345        let stream_key = get_stream_key(trader_id, instance_id, &config);
1346        let external_streams = vec![stream_key.clone()];
1347        let stream_signal = Arc::new(AtomicBool::new(false));
1348        let stream_signal_clone = stream_signal.clone();
1349
1350        // Use a message ID in the future, as streaming begins
1351        // around the timestamp the task is spawned.
1352        let clock = get_atomic_clock_realtime();
1353        let future_id = (clock.get_time_ms() + 1_000_000).to_string();
1354
1355        // Publish test message
1356        let _: () = con
1357            .xadd(
1358                stream_key,
1359                future_id,
1360                &[("topic", "topic1"), ("payload", "data1")],
1361            )
1362            .await
1363            .unwrap();
1364
1365        // Immediately close channel
1366        rx.close();
1367
1368        // Start the message streaming task
1369        let handle = tokio::spawn(async move {
1370            stream_messages(
1371                tx,
1372                RedisMessageBusConfig::default(),
1373                external_streams,
1374                stream_signal_clone,
1375            )
1376            .await
1377            .unwrap();
1378        });
1379
1380        // Shutdown and cleanup
1381        handle.await.unwrap();
1382    }
1383
1384    #[rstest]
1385    #[tokio::test(flavor = "multi_thread")]
1386    async fn test_stream_messages(#[future] redis_connection: ConnectionManager) {
1387        let mut con = redis_connection.await;
1388        let (tx, mut rx) = tokio::sync::mpsc::channel::<BusMessage>(100);
1389
1390        let trader_id = TraderId::from("tester-001");
1391        let instance_id = UUID4::new();
1392        let config = MessageBusConfig {
1393            use_instance_id: true,
1394            ..Default::default()
1395        };
1396
1397        let stream_key = get_stream_key(trader_id, instance_id, &config);
1398        let external_streams = vec![stream_key.clone()];
1399        let stream_signal = Arc::new(AtomicBool::new(false));
1400        let stream_signal_clone = stream_signal.clone();
1401
1402        // Use a message ID in the future, as streaming begins
1403        // around the timestamp the task is spawned.
1404        let clock = get_atomic_clock_realtime();
1405        let future_id = (clock.get_time_ms() + 1_000_000).to_string();
1406
1407        // Publish test message
1408        let _: () = con
1409            .xadd(
1410                stream_key,
1411                future_id,
1412                &[("topic", "topic1"), ("payload", "data1")],
1413            )
1414            .await
1415            .unwrap();
1416
1417        // Start the message streaming task
1418        let handle = tokio::spawn(async move {
1419            stream_messages(
1420                tx,
1421                RedisMessageBusConfig::default(),
1422                external_streams,
1423                stream_signal_clone,
1424            )
1425            .await
1426            .unwrap();
1427        });
1428
1429        // Receive and verify the message
1430        let msg = receive_bus_message(&mut rx, Duration::from_secs(2)).await;
1431        assert_eq!(msg.topic, "topic1");
1432        assert_eq!(msg.payload, Bytes::from("data1"));
1433
1434        // Shutdown and cleanup
1435        rx.close();
1436        stream_signal.store(true, Ordering::Relaxed);
1437        handle.await.unwrap();
1438    }
1439
1440    #[rstest]
1441    #[tokio::test(flavor = "multi_thread")]
1442    async fn test_stream_messages_skips_malformed_entry(
1443        #[future] redis_connection: ConnectionManager,
1444    ) {
1445        let mut con = redis_connection.await;
1446        let (tx, mut rx) = tokio::sync::mpsc::channel::<BusMessage>(100);
1447
1448        let suffix = UUID4::new();
1449        let stream_key = format!("test:stream:malformed:{suffix}");
1450        let external_streams = vec![stream_key.clone()];
1451        let stream_signal = Arc::new(AtomicBool::new(false));
1452        let stream_signal_clone = stream_signal.clone();
1453
1454        let clock = get_atomic_clock_realtime();
1455        let base_id = clock.get_time_ms() + 1_000_000;
1456
1457        let _: () = con
1458            .xadd(
1459                &stream_key,
1460                format!("{}", base_id + 1),
1461                &[("topic", "missing-payload")],
1462            )
1463            .await
1464            .unwrap();
1465        let _: () = con
1466            .xadd(
1467                &stream_key,
1468                format!("{}", base_id + 2),
1469                &[("topic", "valid"), ("payload", "data")],
1470            )
1471            .await
1472            .unwrap();
1473
1474        let handle = tokio::spawn(async move {
1475            stream_messages(
1476                tx,
1477                RedisMessageBusConfig::default(),
1478                external_streams,
1479                stream_signal_clone,
1480            )
1481            .await
1482            .unwrap();
1483        });
1484
1485        let msg = receive_bus_message(&mut rx, Duration::from_secs(2)).await;
1486
1487        rx.close();
1488        stream_signal.store(true, Ordering::Relaxed);
1489        handle.await.unwrap();
1490
1491        assert_eq!(msg.topic, "valid");
1492        assert_eq!(msg.payload, Bytes::from("data"));
1493    }
1494
1495    #[rstest]
1496    #[tokio::test(flavor = "multi_thread")]
1497    async fn test_stream_messages_returns_unrecoverable_read_error(
1498        #[future] redis_connection: ConnectionManager,
1499    ) {
1500        let mut con = redis_connection.await;
1501        let (tx, _rx) = tokio::sync::mpsc::channel::<BusMessage>(100);
1502
1503        let suffix = UUID4::new();
1504        let stream_key = format!("test:stream:wrong-type:{suffix}");
1505        let external_streams = vec![stream_key.clone()];
1506        let stream_signal = Arc::new(AtomicBool::new(false));
1507
1508        let _: () = con.set(&stream_key, "not-a-stream").await.unwrap();
1509
1510        let result = stream_messages(
1511            tx,
1512            RedisMessageBusConfig::default(),
1513            external_streams,
1514            stream_signal,
1515        )
1516        .await;
1517
1518        let _: () = con.del(&stream_key).await.unwrap();
1519
1520        assert!(result.is_err());
1521        assert!(
1522            result
1523                .unwrap_err()
1524                .to_string()
1525                .contains("Error reading from stream")
1526        );
1527    }
1528
1529    #[rstest]
1530    #[tokio::test(flavor = "multi_thread")]
1531    async fn test_stream_connection_returns_none_when_signaled() {
1532        let config = RedisMessageBusConfig {
1533            port: Some(1),
1534            connection_timeout: 20,
1535            ..Default::default()
1536        };
1537        let stream_signal = Arc::new(AtomicBool::new(true));
1538        let signal = stream_signal.clone();
1539        let handle = tokio::spawn(async move { connect_stream_connection(&config, &signal).await });
1540
1541        wait_until_async(|| async { handle.is_finished() }, Duration::from_secs(1)).await;
1542
1543        assert!(handle.await.unwrap().unwrap().is_none());
1544    }
1545
1546    #[rstest]
1547    #[tokio::test(flavor = "multi_thread")]
1548    async fn test_publish_messages(#[future] redis_connection: ConnectionManager) {
1549        let mut con = redis_connection.await;
1550        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<BusMessage>();
1551
1552        let trader_id = TraderId::from("tester-001");
1553        let instance_id = UUID4::new();
1554        let config = MessageBusConfig {
1555            use_instance_id: true,
1556            stream_per_topic: false,
1557            ..Default::default()
1558        };
1559        let stream_key = get_stream_key(trader_id, instance_id, &config);
1560
1561        // Start the publish_messages task
1562        let handle = tokio::spawn(async move {
1563            publish_messages(
1564                rx,
1565                trader_id,
1566                instance_id,
1567                config,
1568                RedisMessageBusConfig::default(),
1569            )
1570            .await
1571            .unwrap();
1572        });
1573
1574        // Send a test message
1575        let msg = BusMessage::with_str_topic(
1576            "test_topic",
1577            BusPayloadType::QuoteTick,
1578            Bytes::from("test_payload"),
1579            SerializationEncoding::Json,
1580        );
1581        tx.send(msg).unwrap();
1582
1583        // Wait until the message is published to Redis
1584        wait_until_async(
1585            || {
1586                let mut con = con.clone();
1587                let stream_key = stream_key.clone();
1588                async move {
1589                    let messages: RedisStreamBulk =
1590                        con.xread(&[&stream_key], &["0"]).await.unwrap();
1591                    !messages.is_empty()
1592                }
1593            },
1594            Duration::from_secs(3),
1595        )
1596        .await;
1597
1598        // Verify the message was published to Redis
1599        let messages: RedisStreamBulk = con.xread(&[&stream_key], &["0"]).await.unwrap();
1600        assert_eq!(messages.len(), 1);
1601        let stream_msgs = messages[0].get(&stream_key).unwrap();
1602        let stream_msg_array = &stream_msgs[0].values().next().unwrap();
1603        let decoded_message = decode_bus_message(stream_msg_array).unwrap();
1604        assert_eq!(decoded_message.topic, "test_topic");
1605        assert_eq!(decoded_message.payload_type, BusPayloadType::QuoteTick);
1606        assert_eq!(decoded_message.encoding, SerializationEncoding::Json);
1607        assert_eq!(decoded_message.payload, Bytes::from("test_payload"));
1608
1609        // Stop publishing task
1610        let msg = BusMessage::new_close();
1611        tx.send(msg).unwrap();
1612
1613        // Shutdown and cleanup
1614        handle.await.unwrap();
1615    }
1616
1617    #[rstest]
1618    #[tokio::test(flavor = "multi_thread")]
1619    async fn test_two_live_nodes_publish_and_ingest_external_redis_stream(
1620        #[future] redis_connection: ConnectionManager,
1621    ) {
1622        let _con = redis_connection.await;
1623        let redis_config = RedisMessageBusConfig::default();
1624        let trader_a = TraderId::from("NODEA-001");
1625        let instance_a = UUID4::new();
1626        let node_a_msgbus = MessageBusConfig {
1627            use_instance_id: true,
1628            stream_per_topic: false,
1629            ..Default::default()
1630        };
1631        let stream_key = get_stream_key(trader_a, instance_a, &node_a_msgbus);
1632        let node_b_msgbus = MessageBusConfig {
1633            external_streams: Some(vec![stream_key]),
1634            stream_per_topic: false,
1635            ..Default::default()
1636        };
1637        let quote = QuoteTick::default();
1638        let trade = TradeTick::default();
1639        let (ready_tx, ready_rx) = mpsc::channel::<()>();
1640        let (quote_tx, quote_rx) = mpsc::channel::<QuoteTick>();
1641        let (trade_tx, trade_rx) = mpsc::channel::<TradeTick>();
1642
1643        let node_b = thread::spawn({
1644            let redis_config = redis_config.clone();
1645            move || -> anyhow::Result<()> {
1646                let runtime = tokio::runtime::Builder::new_multi_thread()
1647                    .worker_threads(2)
1648                    .enable_all()
1649                    .build()?;
1650
1651                runtime.block_on(async move {
1652                    let config = LiveNodeConfig {
1653                        environment: Environment::Sandbox,
1654                        trader_id: TraderId::from("NODEB-001"),
1655                        msgbus: Some(node_b_msgbus),
1656                        exec_engine: LiveExecEngineConfig {
1657                            reconciliation: false,
1658                            ..Default::default()
1659                        },
1660                        delay_post_stop: Duration::ZERO,
1661                        timeout_connection: Duration::from_millis(500),
1662                        timeout_disconnection: Duration::from_millis(500),
1663                        ..Default::default()
1664                    };
1665                    let mut node = LiveNodeBuilder::from_config(config)?
1666                        .with_external_msgbus_factory(Box::new(RedisMessageBusFactory::new(
1667                            redis_config,
1668                        )))
1669                        .build()?;
1670                    let handle = node.handle();
1671                    let quote_handler = TypedHandler::from({
1672                        let quote_tx = quote_tx.clone();
1673                        let handle = handle.clone();
1674                        move |quote: &QuoteTick| {
1675                            let _ = quote_tx.send(*quote);
1676                            handle.stop();
1677                        }
1678                    });
1679                    let trade_handler = TypedHandler::from(move |trade: &TradeTick| {
1680                        let _ = trade_tx.send(*trade);
1681                    });
1682
1683                    msgbus::subscribe_quotes("data.quotes.*".into(), quote_handler, None);
1684                    msgbus::subscribe_trades("data.trades.*".into(), trade_handler, None);
1685                    msgbus::get_message_bus()
1686                        .borrow_mut()
1687                        .add_streaming_type(BusPayloadType::QuoteTick);
1688                    let result = tokio::time::timeout(Duration::from_secs(10), async {
1689                        let run = node.run();
1690                        tokio::pin!(run);
1691
1692                        let announce_ready = async {
1693                            for _ in 0..100 {
1694                                if handle.is_running() {
1695                                    ready_tx.send(())?;
1696                                    return Ok(());
1697                                }
1698                                tokio::time::sleep(Duration::from_millis(10)).await;
1699                            }
1700
1701                            anyhow::bail!("node B did not reach running state")
1702                        };
1703
1704                        tokio::select! {
1705                            result = &mut run => result,
1706                            ready = announce_ready => {
1707                                ready?;
1708                                run.await
1709                            }
1710                        }
1711                    })
1712                    .await;
1713                    msgbus::get_message_bus().borrow_mut().dispose();
1714
1715                    match result {
1716                        Ok(Ok(())) => Ok(()),
1717                        Ok(Err(e)) => Err(e),
1718                        Err(e) => anyhow::bail!("node B timed out: {e}"),
1719                    }
1720                })
1721            }
1722        });
1723
1724        ready_rx
1725            .recv_timeout(Duration::from_secs(5))
1726            .expect("node B should start Redis ingress");
1727
1728        let node_a = thread::spawn(move || -> anyhow::Result<()> {
1729            let runtime = tokio::runtime::Builder::new_multi_thread()
1730                .worker_threads(2)
1731                .enable_all()
1732                .build()?;
1733
1734            runtime.block_on(async move {
1735                let config = LiveNodeConfig {
1736                    environment: Environment::Sandbox,
1737                    trader_id: trader_a,
1738                    instance_id: Some(instance_a),
1739                    msgbus: Some(node_a_msgbus),
1740                    exec_engine: LiveExecEngineConfig {
1741                        reconciliation: false,
1742                        ..Default::default()
1743                    },
1744                    delay_post_stop: Duration::ZERO,
1745                    timeout_connection: Duration::from_millis(500),
1746                    timeout_disconnection: Duration::from_millis(500),
1747                    ..Default::default()
1748                };
1749                let _node = LiveNodeBuilder::from_config(config)?
1750                    .with_external_msgbus_factory(Box::new(RedisMessageBusFactory::new(
1751                        redis_config,
1752                    )))
1753                    .build()?;
1754
1755                msgbus::publish_trade("data.trades.TEST".into(), &trade);
1756                msgbus::publish_quote("data.quotes.TEST".into(), &quote);
1757                msgbus::get_message_bus().borrow_mut().dispose();
1758
1759                Ok(())
1760            })
1761        });
1762
1763        node_a
1764            .join()
1765            .expect("node A thread should not panic")
1766            .expect("node A should publish externally");
1767        let received_quote = quote_rx
1768            .recv_timeout(Duration::from_secs(10))
1769            .expect("node B should republish the registered quote type");
1770        node_b
1771            .join()
1772            .expect("node B thread should not panic")
1773            .expect("node B should ingest and stop cleanly");
1774
1775        assert_eq!(received_quote, quote);
1776        assert!(
1777            trade_rx.try_recv().is_err(),
1778            "unregistered trade type should not republish internally"
1779        );
1780    }
1781
1782    #[rstest]
1783    #[tokio::test(flavor = "multi_thread")]
1784    async fn test_stream_messages_multiple_streams(#[future] redis_connection: ConnectionManager) {
1785        let mut con = redis_connection.await;
1786        let (tx, mut rx) = tokio::sync::mpsc::channel::<BusMessage>(100);
1787
1788        // Setup multiple stream keys
1789        let suffix = UUID4::new();
1790        let stream_key1 = format!("test:stream:{suffix}:1");
1791        let stream_key2 = format!("test:stream:{suffix}:2");
1792        let external_streams = vec![stream_key1.clone(), stream_key2.clone()];
1793        let stream_signal = Arc::new(AtomicBool::new(false));
1794        let stream_signal_clone = stream_signal.clone();
1795
1796        let clock = get_atomic_clock_realtime();
1797        let base_id = clock.get_time_ms() + 1_000_000;
1798
1799        // Start streaming task
1800        let handle = tokio::spawn(async move {
1801            stream_messages(
1802                tx,
1803                RedisMessageBusConfig::default(),
1804                external_streams,
1805                stream_signal_clone,
1806            )
1807            .await
1808            .unwrap();
1809        });
1810
1811        // Publish to stream 1 at higher ID
1812        let _: () = con
1813            .xadd(
1814                &stream_key1,
1815                format!("{}", base_id + 100),
1816                &[("topic", "stream1-first"), ("payload", "data")],
1817            )
1818            .await
1819            .unwrap();
1820
1821        let msg = receive_bus_message(&mut rx, Duration::from_secs(2)).await;
1822        assert_eq!(msg.topic, "stream1-first");
1823
1824        // Publish to stream 2 at lower ID (tests independent cursor tracking)
1825        let _: () = con
1826            .xadd(
1827                &stream_key2,
1828                format!("{}", base_id + 50),
1829                &[("topic", "stream2-second"), ("payload", "data")],
1830            )
1831            .await
1832            .unwrap();
1833
1834        let msg = receive_bus_message(&mut rx, Duration::from_secs(2)).await;
1835        assert_eq!(msg.topic, "stream2-second");
1836
1837        // Shutdown and cleanup
1838        rx.close();
1839        stream_signal.store(true, Ordering::Relaxed);
1840        handle.await.unwrap();
1841    }
1842
1843    #[rstest]
1844    #[tokio::test(flavor = "multi_thread")]
1845    async fn test_stream_messages_interleaved_at_different_rates(
1846        #[future] redis_connection: ConnectionManager,
1847    ) {
1848        let mut con = redis_connection.await;
1849        let (tx, mut rx) = tokio::sync::mpsc::channel::<BusMessage>(100);
1850
1851        // Setup multiple stream keys
1852        let suffix = UUID4::new();
1853        let stream_key1 = format!("test:stream:interleaved:{suffix}:1");
1854        let stream_key2 = format!("test:stream:interleaved:{suffix}:2");
1855        let stream_key3 = format!("test:stream:interleaved:{suffix}:3");
1856        let external_streams = vec![
1857            stream_key1.clone(),
1858            stream_key2.clone(),
1859            stream_key3.clone(),
1860        ];
1861        let stream_signal = Arc::new(AtomicBool::new(false));
1862        let stream_signal_clone = stream_signal.clone();
1863
1864        let clock = get_atomic_clock_realtime();
1865        let base_id = clock.get_time_ms() + 1_000_000;
1866
1867        let handle = tokio::spawn(async move {
1868            stream_messages(
1869                tx,
1870                RedisMessageBusConfig::default(),
1871                external_streams,
1872                stream_signal_clone,
1873            )
1874            .await
1875            .unwrap();
1876        });
1877
1878        // Stream 1 advances with high ID
1879        let _: () = con
1880            .xadd(
1881                &stream_key1,
1882                format!("{}", base_id + 100),
1883                &[("topic", "s1m1"), ("payload", "data")],
1884            )
1885            .await
1886            .unwrap();
1887        let msg = receive_bus_message(&mut rx, Duration::from_secs(2)).await;
1888        assert_eq!(msg.topic, "s1m1");
1889
1890        // Stream 2 gets message at lower ID - would be skipped with global cursor
1891        let _: () = con
1892            .xadd(
1893                &stream_key2,
1894                format!("{}", base_id + 50),
1895                &[("topic", "s2m1"), ("payload", "data")],
1896            )
1897            .await
1898            .unwrap();
1899        let msg = receive_bus_message(&mut rx, Duration::from_secs(2)).await;
1900        assert_eq!(msg.topic, "s2m1");
1901
1902        // Stream 3 gets message at even lower ID
1903        let _: () = con
1904            .xadd(
1905                &stream_key3,
1906                format!("{}", base_id + 25),
1907                &[("topic", "s3m1"), ("payload", "data")],
1908            )
1909            .await
1910            .unwrap();
1911        let msg = receive_bus_message(&mut rx, Duration::from_secs(2)).await;
1912        assert_eq!(msg.topic, "s3m1");
1913
1914        // Shutdown and cleanup
1915        rx.close();
1916        stream_signal.store(true, Ordering::Relaxed);
1917        handle.await.unwrap();
1918    }
1919
1920    #[rstest]
1921    #[tokio::test(flavor = "multi_thread")]
1922    async fn test_close() {
1923        let trader_id = TraderId::from("tester-001");
1924        let instance_id = UUID4::new();
1925        let config = MessageBusConfig {
1926            use_instance_id: true,
1927            ..Default::default()
1928        };
1929
1930        let mut db = RedisMessageBusBacking::new(
1931            trader_id,
1932            instance_id,
1933            config,
1934            RedisMessageBusConfig::default(),
1935        )
1936        .unwrap();
1937
1938        // Close the message bus backing (test should not hang)
1939        MessageBusBacking::close(&mut db);
1940    }
1941
1942    #[rstest]
1943    #[tokio::test(flavor = "multi_thread")]
1944    async fn test_heartbeat_task() {
1945        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<BusMessage>();
1946        let signal = Arc::new(AtomicBool::new(false));
1947
1948        // Start the heartbeat task with a short interval
1949        let handle = tokio::spawn(run_heartbeat(1, signal.clone(), tx));
1950
1951        let heartbeat = receive_unbounded_bus_message(&mut rx, Duration::from_secs(2)).await;
1952
1953        // Stop the heartbeat task
1954        signal.store(true, Ordering::Relaxed);
1955        handle.await.unwrap();
1956
1957        // Ensure heartbeats were sent
1958        assert_eq!(heartbeat.topic, HEARTBEAT_TOPIC);
1959    }
1960
1961    async fn receive_bus_message(
1962        rx: &mut tokio::sync::mpsc::Receiver<BusMessage>,
1963        timeout: Duration,
1964    ) -> BusMessage {
1965        let mut received = None;
1966
1967        wait_until_async(
1968            || {
1969                if received.is_none() {
1970                    received = rx.try_recv().ok();
1971                }
1972
1973                let has_received = received.is_some();
1974                async move { has_received }
1975            },
1976            timeout,
1977        )
1978        .await;
1979
1980        received.unwrap()
1981    }
1982
1983    async fn receive_unbounded_bus_message(
1984        rx: &mut tokio::sync::mpsc::UnboundedReceiver<BusMessage>,
1985        timeout: Duration,
1986    ) -> BusMessage {
1987        let mut received = None;
1988
1989        wait_until_async(
1990            || {
1991                if received.is_none() {
1992                    received = rx.try_recv().ok();
1993                }
1994
1995                let has_received = received.is_some();
1996                async move { has_received }
1997            },
1998            timeout,
1999        )
2000        .await;
2001
2002        received.unwrap()
2003    }
2004}
nautilus_infrastructure/redis/msgbus.rs

nautilus_infrastructure/redis/
msgbus.rs
